added configuration to run mpi in local and remote mode

Author: Annnnya

HeterogeneousCore/MPICore/plugins/MPIController.cc

@@ -63,6 +63,7 @@ class MPIController : public edm::one::EDProducer<edm::one::WatchRuns, edm::one:
   MPI_Comm comm_ = MPI_COMM_NULL;
   MPIChannel channel_;
   edm::EDPutTokenT<MPIToken> token_;
+  bool run_local_;
 };
 
 MPIController::MPIController(edm::ParameterSet const& config)
@@ -71,30 +72,45 @@ MPIController::MPIController(edm::ParameterSet const& config)
   // make sure that MPI is initialised
   MPIService::required();
 
+  // get from parameter set whether remote mpi is used
+  run_local_ = config.getUntrackedParameter<bool>("run_local", false);
+
+  edm::LogAbsolute("MPI") << "Running controller in " << (run_local_ ? "local" : "remote") << " mode.";
+
   // FIXME move into the MPIService ?
   // make sure the EDM MPI types are available
   EDM_MPI_build_types();
 
-  // look up the "server" port
-  char port[MPI_MAX_PORT_NAME];
-  MPI_Lookup_name("server", MPI_INFO_NULL, port);
-  edm::LogAbsolute("MPI") << "Trying to connect to the MPI server on port " << port;
-
-  // connect to the server
-  int size;
-  MPI_Comm_connect(port, MPI_INFO_NULL, 0, MPI_COMM_SELF, &comm_);
-  MPI_Comm_remote_size(comm_, &size);
-  edm::LogAbsolute("MPI") << "Client connected to " << size << (size == 1 ? " server" : " servers");
-  if (size > 1) {
-    throw cms::Exception("UnsupportedFeature")
-        << "MPIController supports only a single follower, but it was connected to " << size << " followers";
+  if (!run_local_) {
+    // look up the "server" port
+    char port[MPI_MAX_PORT_NAME];
+    MPI_Lookup_name("server", MPI_INFO_NULL, port);
+    edm::LogAbsolute("MPI") << "Trying to connect to the MPI server on port " << port;
+
+    // connect to the server
+    int size;
+    MPI_Comm_connect(port, MPI_INFO_NULL, 0, MPI_COMM_SELF, &comm_);
+    MPI_Comm_remote_size(comm_, &size);
+    edm::LogAbsolute("MPI") << "Client connected to " << size << (size == 1 ? " server" : " servers");
+    if (size > 1) {
+      throw cms::Exception("UnsupportedFeature")
+          << "MPIController supports only a single follower, but it was connected to " << size << " followers";
+    }
+    channel_ = MPIChannel(comm_, 0);
+  } else {
+    comm_ = MPI_COMM_WORLD;
+    channel_ = MPIChannel(comm_, 0);
+    int world_rank;
+    MPI_Comm_rank(MPI_COMM_WORLD, &world_rank);
+    std::cout << "MPI Controller Rank: " << world_rank << " in MPI_COMM_WORLD" << std::endl;
   }
-  channel_ = MPIChannel(comm_, 0);
 }
 
 MPIController::~MPIController() {
   // close the intercommunicator
-  MPI_Comm_disconnect(&comm_);
+  if (!run_local_) {
+    MPI_Comm_disconnect(&comm_);
+  }
 }
 
 void MPIController::beginJob() {
@@ -223,6 +239,7 @@ void MPIController::fillDescriptions(edm::ConfigurationDescriptions& description
       "Runs, LuminosityBlocks and Events from the current process to the remote one.");
 
   edm::ParameterSetDescription desc;
+  desc.addOptionalUntracked<bool>("run_local", false);
   descriptions.addWithDefaultLabel(desc);
 }
 

HeterogeneousCore/MPICore/plugins/MPIReceiver.cc

@@ -50,9 +50,9 @@ class MPIReceiver : public edm::global::EDProducer<> {
     MPIToken token = event.get(upstream_);
 
     // Receive the number of products
-    int numProducts;
-    token.channel()->receiveProduct(instance_, numProducts);
-    edm::LogVerbatim("MPIReceiver") << "Received number of products: " << numProducts;
+    // int numProducts;
+    // token.channel()->receiveProduct(instance_, numProducts);
+    // edm::LogVerbatim("MPIReceiver") << "Received number of products: " << numProducts;
 
     for (auto const& entry : products_) {
       std::unique_ptr<edm::WrapperBase> wrapper(

HeterogeneousCore/MPICore/plugins/MPISender.cc

@@ -83,8 +83,8 @@ class MPISender : public edm::global::EDProducer<> {
     // read the MPIToken used to establish the communication channel
     MPIToken token = event.get(upstream_);
 
-    int numProducts = static_cast<int>(products_.size());
-    token.channel()->sendProduct(instance_, numProducts);
+    // int numProducts = static_cast<int>(products_.size());
+    // token.channel()->sendProduct(instance_, numProducts);
 
     for (auto const& entry : products_) {
       // read the products to be sent over the MPI channel

HeterogeneousCore/MPICore/plugins/MPISource.cc

@@ -58,6 +58,7 @@ class MPISource : public edm::ProducerSourceBase {
   MPI_Comm comm_ = MPI_COMM_NULL;
   MPIChannel channel_;
   edm::EDPutTokenT<MPIToken> token_;
+  bool run_local_;
 
   edm::ProcessHistory history_;
 };
@@ -76,22 +77,37 @@ MPISource::MPISource(edm::ParameterSet const& config, edm::InputSourceDescriptio
   // make sure the EDM MPI types are available
   EDM_MPI_build_types();
 
-  // open a server-side port
-  MPI_Open_port(MPI_INFO_NULL, port_);
-
-  // publish the port under the name "server"
-  MPI_Info port_info;
-  MPI_Info_create(&port_info);
-  MPI_Info_set(port_info, "ompi_global_scope", "true");
-  MPI_Info_set(port_info, "ompi_unique", "true");
-  MPI_Publish_name("server", port_info, port_);
-
-  // create an intercommunicator and accept a client connection
-  edm::LogAbsolute("MPI") << "waiting for a connection to the MPI server at port " << port_;
-  MPI_Comm_accept(port_, MPI_INFO_NULL, 0, MPI_COMM_WORLD, &comm_);
-  channel_ = MPIChannel(comm_, 0);
+  // get from parameter set whether remote mpi is used
+  run_local_ = config.getUntrackedParameter<bool>("run_local", false);
+
+  edm::LogAbsolute("MPI") << "Running source in " << (run_local_ ? "local" : "remote") << " mode.";
+
+  if (run_local_) {
+    int world_rank;
+    MPI_Comm_rank(MPI_COMM_WORLD, &world_rank);
+    std::cout << "MPI Source Rank: " << world_rank << " in MPI_COMM_WORLD" << std::endl;
+    comm_ = MPI_COMM_WORLD;
+    channel_ = MPIChannel(comm_, 1);
+  } else {
+    // open a server-side port
+    MPI_Open_port(MPI_INFO_NULL, port_);
+
+    // publish the port under the name "server"
+    MPI_Info port_info;
+    MPI_Info_create(&port_info);
+    MPI_Info_set(port_info, "ompi_global_scope", "true");
+    MPI_Info_set(port_info, "ompi_unique", "true");
+    MPI_Publish_name("server", port_info, port_);
+
+    // create an intercommunicator and accept a client connection
+    edm::LogAbsolute("MPI") << "waiting for a connection to the MPI server at port " << port_;
+
+    MPI_Comm_accept(port_, MPI_INFO_NULL, 0, MPI_COMM_WORLD, &comm_);
+    edm::LogAbsolute("MPI") << "connection accept ";
+    channel_ = MPIChannel(comm_, 0);
+  }
 
-  // wait for a client to connect
+  // // wait for a client to connect
   MPI_Status status;
   EDM_MPI_Empty_t buffer;
   MPI_Recv(&buffer, 1, EDM_MPI_Empty, MPI_ANY_SOURCE, EDM_MPI_Connect, comm_, &status);
@@ -100,15 +116,17 @@ MPISource::MPISource(edm::ParameterSet const& config, edm::InputSourceDescriptio
 
 MPISource::~MPISource() {
   // close the intercommunicator
-  MPI_Comm_disconnect(&comm_);
-
-  // unpublish and close the port
-  MPI_Info port_info;
-  MPI_Info_create(&port_info);
-  MPI_Info_set(port_info, "ompi_global_scope", "true");
-  MPI_Info_set(port_info, "ompi_unique", "true");
-  MPI_Unpublish_name("server", port_info, port_);
-  MPI_Close_port(port_);
+  if (!run_local_) {
+    MPI_Comm_disconnect(&comm_);
+
+    // unpublish and close the port
+    MPI_Info port_info;
+    MPI_Info_create(&port_info);
+    MPI_Info_set(port_info, "ompi_global_scope", "true");
+    MPI_Info_set(port_info, "ompi_unique", "true");
+    MPI_Unpublish_name("server", port_info, port_);
+    MPI_Close_port(port_);
+  }
 }
 
 //MPISource::ItemTypeInfo MPISource::getNextItemType() {
@@ -259,6 +277,8 @@ void MPISource::fillDescriptions(edm::ConfigurationDescriptions& descriptions) {
   desc.setComment("Comunicate with another cmsRun process over MPI.");
   edm::ProducerSourceBase::fillDescription(desc);
 
+  desc.addOptionalUntracked<bool>("run_local");
+
   descriptions.add("source", desc);
 }