Skip to content

Commit

Permalink
Make MPI version a subclass and reduce reduce code duplication
Browse files Browse the repository at this point in the history
  • Loading branch information
@hellkite500
hellkite500 committed Nov 2, 2023
1 parent 55c7007 commit 5e4a9bc
Show file tree
Hide file tree
Showing 2 changed files with 29 additions and 114 deletions.
63 changes: 4 additions & 59 deletions include/core/HY_Features_MPI.hpp
View file
Original file line number Diff line number Diff line change
Expand Up @@ -11,84 +11,29 @@
#include <network.hpp>
#include <Formulation_Manager.hpp>
#include <Partition_Parser.hpp>
#include <HY_Features.hpp>

namespace hy_features {

class HY_Features_MPI {
class HY_Features_MPI: public HY_Features {
public:

using Formulation_Manager = realization::Formulation_Manager;

HY_Features_MPI(PartitionData partition_data, geojson::GeoJSON linked_hydro_fabric,
std::shared_ptr<Formulation_Manager> formulations, int mpi_rank, int mpi_num_procs);

std::shared_ptr<HY_CatchmentRealization> catchment_at(std::string id) {
return (_catchments.find(id) != _catchments.end()) ? _catchments[id]->realization : nullptr;
}

inline auto catchments() {
return network.filter("cat");
}
HY_Features_MPI() = delete; //Explicitly delete the default constructor

inline bool is_remote_sender_nexus(const std::string& id) {
return _nexuses.find(id) != _nexuses.end() && _nexuses[id]->is_remote_sender();
}

inline auto catchments(long lyr) {
return network.filter("cat",lyr);
}

/**
* @brief Return a set of layers that contain a catchment
*/

inline const auto& layers() { return hf_layers; }

inline std::vector<std::shared_ptr<HY_HydroNexus>> destination_nexuses(const std::string& id) {
std::vector<std::shared_ptr<HY_HydroNexus>> downstream;
if (_catchments.find(id) != _catchments.end()) {
for(const auto& nex_id : _catchments[id]->get_outflow_nexuses()) {
downstream.push_back(_nexuses[nex_id]);
}
}
return downstream;
}

std::shared_ptr<HY_HydroNexus> nexus_at(const std::string& id) {
return (_nexuses.find(id) != _nexuses.end()) ? _nexuses[id] : nullptr;
}

inline auto nexuses() {
return network.filter("nex");
}

void validate_dendritic() {
for(const auto& id : catchments()) {
auto downstream = network.get_destination_ids(id);
if(downstream.size() > 1) {
std::cerr << "Catchment " << id << " has more than one downstream connection." << std::endl;
std::cerr << "Downstreams are: ";
for(const auto& id : downstream){
std::cerr <<id<<" ";
}
std::cerr << std::endl;
assert( false );
}
else if (downstream.size() == 0) {
std::cerr << "Catchment " << id << " has 0 downstream connections, must have 1." << std::endl;
assert( false );
}
}
std::cout<<"Catchment topology is dendritic."<<std::endl;
}

private:

std::unordered_map<std::string, std::shared_ptr<HY_Catchment>> _catchments;
std::unordered_map<std::string, std::shared_ptr<HY_PointHydroNexusRemote>> _nexuses;
network::Network network;
std::shared_ptr<Formulation_Manager> formulations;
std::set<long> hf_layers;

int mpi_rank;
int mpi_num_procs;

Expand Down
80 changes: 25 additions & 55 deletions src/core/HY_Features_MPI.cpp
View file
Original file line number Diff line number Diff line change
@@ -1,12 +1,13 @@
#include <HY_Features_MPI.hpp>
#include <HY_PointHydroNexusRemote.hpp>
#include <network.hpp>

#ifdef NGEN_MPI_ACTIVE

using namespace hy_features;

HY_Features_MPI::HY_Features_MPI( PartitionData partition_data, geojson::GeoJSON linked_hydro_fabric, std::shared_ptr<Formulation_Manager> formulations, int mpi_rank, int mpi_num_procs) :
network(linked_hydro_fabric), formulations(formulations), mpi_rank(mpi_rank), mpi_num_procs(mpi_num_procs)
HY_Features(network::Network(linked_hydro_fabric), formulations, linked_hydro_fabric), mpi_rank(mpi_rank), mpi_num_procs(mpi_num_procs)
{
std::string feat_id;
std::string feat_type;
Expand All @@ -27,62 +28,31 @@ HY_Features_MPI::HY_Features_MPI( PartitionData partition_data, geojson::GeoJSON
remote_connection_direction[remote_nexi][remote_catchments] = std::get<3>(remote_tuple);
}

for(const auto& feat_idx : network){
feat_id = network.get_id(feat_idx);//feature->get_id();
feat_type = feat_id.substr(0, 3);

destinations = network.get_destination_ids(feat_id);
//Find upstream ids
origins = network.get_origination_ids(feat_id);
if(feat_type == "cat" || feat_type == "agg")
{
//Find and prepare formulation
auto formulation = formulations->get_formulation(feat_id);
formulation->set_output_stream(formulations->get_output_root() + feat_id + ".csv");
// TODO: add command line or config option to have this be omitted
//FIXME why isn't default param working here??? get_output_header_line() fails.
formulation->write_output("Time Step,""Time,"+formulation->get_output_header_line(",")+"\n");

// get the catchment layer from the hydro fabric
const auto& cat_json_node = linked_hydro_fabric->get_feature(feat_id);
long lyr = cat_json_node->has_key("layer") ? cat_json_node->get_property("layer").as_natural_number() : 0;

// add this layer to the set of layers if needed
if (hf_layers.find(lyr) == hf_layers.end() )
//catchment and nexus objects already exist for the partitioned data
//we simply need to replace the nexuses which require remote interactions
//with the approriate remote nexus
//origins only contains LOCAL origin features (catchments) as read from
//the geojson/partition subset. We need to make sure `origins` passed to remote nexus
//contain IDS of ALL upstream features, including those in remote partitions
//The same applies to destinations as well.
//Find all remote catchments related to each remote feature
for(const auto& kv : remote_connection_direction)
{
auto feat_id = kv.first;
for(const auto& catchment_direction : kv.second ){
destinations = network.get_destination_ids(feat_id);
origins = network.get_origination_ids(feat_id);
//Determine how it is related to the nexus. This helps determine which side is a "sender"
//and which side is "receiver"
if(catchment_direction.second == "nex-to-dest_cat"){
destinations.push_back(catchment_direction.first);
}else if( catchment_direction.second == "orig_cat-to-nex" )
{
hf_layers.insert(lyr);
origins.push_back(catchment_direction.first);
}

//Create the HY_Catchment with the formulation realization
std::shared_ptr<HY_Catchment> c = std::make_shared<HY_Catchment>(
HY_Catchment(feat_id, origins, destinations, formulation, lyr)
);

_catchments.emplace(feat_id, c);
_nexuses[feat_id] = std::make_unique<HY_PointHydroNexusRemote>(feat_id, destinations, origins, remote_connections[feat_id]) ;
}
else if(feat_type == "nex" || feat_type == "tnx")
{ //origins only contains LOCAL origin features (catchments) as read from
//the geojson/partition subset. We need to make sure `origins` passed to remote nexus
//contain IDS of ALL upstream features, including those in remote partitions
//The same applies to destinations as well.
//Find all remote catchments related to this feature
for(auto& catchment_direction : remote_connection_direction[feat_id])
{ //Determine how it is related to the nexus. This helps determine which side is a "sender"
//and which side is "receiver"
if(catchment_direction.second == "nex-to-dest_cat"){
destinations.push_back(catchment_direction.first);
}else if( catchment_direction.second == "orig_cat-to-nex" )
{
origins.push_back(catchment_direction.first);
}
}
_nexuses.emplace(feat_id, std::make_unique<HY_PointHydroNexusRemote>(feat_id, destinations, origins, remote_connections[feat_id]) );
}
else
{
std::cerr<<"HY_Features::HY_Features unknown feature identifier type "<<feat_type<<" for feature id."<<feat_id
<<" Skipping feature"<<std::endl;
}
}
}

}
#endif //NGEN_MPI_ACTIVE

0 comments on commit 5e4a9bc

Please sign in to comment.