Introduce a mode where all processes are in MPI_COMM_WORLD

Co-authored-by: Anna Polova <[email protected]>

Author: fwyzard

HeterogeneousCore/MPICore/plugins/MPIController.cc

@@ -16,6 +16,7 @@
 #include "FWCore/Framework/interface/one/EDProducer.h"
 #include "FWCore/MessageLogger/interface/MessageLogger.h"
 #include "FWCore/ParameterSet/interface/ConfigurationDescriptions.h"
+#include "FWCore/ParameterSet/interface/EmptyGroupDescription.h"
 #include "FWCore/ParameterSet/interface/ParameterDescriptionNode.h"
 #include "FWCore/ParameterSet/interface/ParameterSet.h"
 #include "FWCore/ParameterSet/interface/ParameterSetDescription.h"
@@ -60,41 +61,93 @@ class MPIController : public edm::one::EDProducer<edm::one::WatchRuns, edm::one:
   static void fillDescriptions(edm::ConfigurationDescriptions& descriptions);
 
 private:
+  enum Mode { kInvalid = 0, kCommWorld, kIntercommunicator };
+  static constexpr const char* ModeDescription[] = {"Invalid", "CommWorld", "Intercommunicator"};
+  Mode parseMode(std::string const& label) {
+    if (label == ModeDescription[kCommWorld])
+      return kCommWorld;
+    else if (label == ModeDescription[kIntercommunicator])
+      return kIntercommunicator;
+    else
+      return kInvalid;
+  }
+
   MPI_Comm comm_ = MPI_COMM_NULL;
   MPIChannel channel_;
   edm::EDPutTokenT<MPIToken> token_;
+  Mode mode_;
 };
 
 MPIController::MPIController(edm::ParameterSet const& config)
-    : token_(produces<MPIToken>())  //
+    : token_(produces<MPIToken>()),
+      mode_(parseMode(config.getUntrackedParameter<std::string>("mode")))  //
 {
   // make sure that MPI is initialised
   MPIService::required();
 
-  // 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 (mode_ == kCommWorld) {
+    // All processes are in MPI_COMM_WORLD.
+    // The current implementation supports only two processes: one controller and one source.
+    edm::LogAbsolute("MPI") << "MPIController in " << ModeDescription[mode_] << " mode.";
+
+    // Check how many processes are there in MPI_COMM_WORLD
+    int size;
+    MPI_Comm_size(MPI_COMM_WORLD, &size);
+    if (size != 2) {
+      throw edm::Exception(edm::errors::Configuration)
+          << "The current implementation supports only two processes: one controller and one source.";
+    }
+
+    // Check the rank of this process, and determine the rank of the other process.
+    int rank;
+    MPI_Comm_rank(MPI_COMM_WORLD, &rank);
+    edm::LogAbsolute("MPI") << "MPIController has rank " << rank << " in MPI_COMM_WORLD.";
+    int other_rank = 1 - rank;
+    comm_ = MPI_COMM_WORLD;
+    channel_ = MPIChannel(comm_, other_rank);
+  } else if (mode_ == kIntercommunicator) {
+    // Use an intercommunicator to let two groups of processes communicate with each other.
+    // The current implementation supports only two processes: one controller and one source.
+    edm::LogAbsolute("MPI") << "MPISource in " << ModeDescription[mode_] << " mode.";
+
+    // Check how many processes are there in MPI_COMM_WORLD
+    int size;
+    MPI_Comm_size(MPI_COMM_WORLD, &size);
+    if (size != 1) {
+      throw edm::Exception(edm::errors::Configuration)
+          << "The current implementation supports only two processes: one controller and one source.";
+    }
+
+    // Look for the port under the name indicated by the parameter "server".
+    std::string name = config.getUntrackedParameter<std::string>("name", "server");
+    char port[MPI_MAX_PORT_NAME];
+    MPI_Lookup_name(name.c_str(), MPI_INFO_NULL, port);
+    edm::LogAbsolute("MPI") << "Trying to connect to the MPI server on port " << port;
+
+    // Create an intercommunicator and connect to the server.
+    MPI_Comm_connect(port, MPI_INFO_NULL, 0, MPI_COMM_SELF, &comm_);
+    MPI_Comm_remote_size(comm_, &size);
+    if (size != 1) {
+      throw edm::Exception(edm::errors::Configuration)
+          << "The current implementation supports only two processes: one controller and one source.";
+    }
+    edm::LogAbsolute("MPI") << "Client connected to " << size << (size == 1 ? " server" : " servers");
+    channel_ = MPIChannel(comm_, 0);
+  } else {
+    // Invalid mode.
+    throw edm::Exception(edm::errors::Configuration)
+        << "Invalid mode \"" << config.getUntrackedParameter<std::string>("mode") << "\"";
   }
-  channel_ = MPIChannel(comm_, 0);
 }
 
 MPIController::~MPIController() {
-  // close the intercommunicator
-  MPI_Comm_disconnect(&comm_);
+  // Close the intercommunicator.
+  if (mode_ == kIntercommunicator) {
+    MPI_Comm_disconnect(&comm_);
+  }
 }
 
 void MPIController::beginJob() {
@@ -219,10 +272,18 @@ void MPIController::produce(edm::Event& event, edm::EventSetup const& setup) {
 
 void MPIController::fillDescriptions(edm::ConfigurationDescriptions& descriptions) {
   descriptions.setComment(
-      "This module connects to an \"MPISource\" in a separate CMSSW job, and transmits all "
-      "Runs, LuminosityBlocks and Events from the current process to the remote one.");
+      "This module connects to an \"MPISource\" in a separate CMSSW job, and transmits all Runs, LuminosityBlocks and "
+      "Events from the current process to the remote one.");
 
   edm::ParameterSetDescription desc;
+  desc.ifValue(
+          edm::ParameterDescription<std::string>("mode", "CommWorld", false),
+          ModeDescription[kCommWorld] >> edm::EmptyGroupDescription() or
+              ModeDescription[kIntercommunicator] >> edm::ParameterDescription<std::string>("name", "server", false))
+      ->setComment(
+          "Valid modes are CommWorld (use MPI_COMM_WORLD) and Intercommunicator (use an MPI name server to setup an "
+          "intercommunicator).");
+
   descriptions.addWithDefaultLabel(desc);
 }
 

HeterogeneousCore/MPICore/plugins/MPISource.cc

@@ -29,10 +29,12 @@
 #include "FWCore/MessageLogger/interface/ErrorObj.h"
 #include "FWCore/MessageLogger/interface/MessageLogger.h"
 #include "FWCore/ParameterSet/interface/ConfigurationDescriptions.h"
+#include "FWCore/ParameterSet/interface/EmptyGroupDescription.h"
+#include "FWCore/ParameterSet/interface/ParameterSet.h"
 #include "FWCore/ParameterSet/interface/ParameterSetDescription.h"
 #include "FWCore/ParameterSet/interface/ParameterSetDescriptionFiller.h"
-#include "FWCore/ParameterSet/interface/ParameterSet.h"
 #include "FWCore/Sources/interface/ProducerSourceBase.h"
+#include "FWCore/Utilities/interface/EDMException.h"
 #include "HeterogeneousCore/MPICore/interface/MPIToken.h"
 #include "HeterogeneousCore/MPIServices/interface/MPIService.h"
 
@@ -54,10 +56,22 @@ class MPISource : public edm::ProducerSourceBase {
   bool setRunAndEventInfo(edm::EventID& id, edm::TimeValue_t& time, edm::EventAuxiliary::ExperimentType&) override;
   void produce(edm::Event&) override;
 
+  enum Mode { kInvalid = 0, kCommWorld, kIntercommunicator };
+  static constexpr const char* ModeDescription[] = {"Invalid", "CommWorld", "Intercommunicator"};
+  Mode parseMode(std::string const& label) {
+    if (label == ModeDescription[kCommWorld])
+      return kCommWorld;
+    else if (label == ModeDescription[kIntercommunicator])
+      return kIntercommunicator;
+    else
+      return kInvalid;
+  }
+
   char port_[MPI_MAX_PORT_NAME];
   MPI_Comm comm_ = MPI_COMM_NULL;
   MPIChannel channel_;
   edm::EDPutTokenT<MPIToken> token_;
+  Mode mode_;
 
   edm::ProcessHistory history_;
 };
@@ -67,48 +81,91 @@ MPISource::MPISource(edm::ParameterSet const& config, edm::InputSourceDescriptio
        // effectively be ignored, because this ConfigurableSource will explicitly set the run, lumi, and event
        // numbers, the timestamp, and the event type
       edm::ProducerSourceBase(config, desc, false),
-      token_(produces<MPIToken>())  //
+      token_(produces<MPIToken>()),
+      mode_(parseMode(config.getUntrackedParameter<std::string>("mode")))  //
 {
   // make sure that MPI is initialised
   MPIService::required();
 
-  // FIXME move into the MPIService ?
-  // make sure the EDM MPI types are available
+  // Make sure the EDM MPI types are available.
   EDM_MPI_build_types();
 
-  // open a server-side port
-  MPI_Open_port(MPI_INFO_NULL, port_);
+  if (mode_ == kCommWorld) {
+    // All processes are in MPI_COMM_WORLD.
+    // The current implementation supports only two processes: one controller and one source.
+    edm::LogAbsolute("MPI") << "MPISource in " << ModeDescription[mode_] << " mode.";
+
+    // Check how many processes are there in MPI_COMM_WORLD
+    int size;
+    MPI_Comm_size(MPI_COMM_WORLD, &size);
+    if (size != 2) {
+      throw edm::Exception(edm::errors::Configuration)
+          << "The current implementation supports only two processes: one controller and one source.";
+    }
 
-  // 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_);
+    // Check the rank of this process, and determine the rank of the other process.
+    int rank;
+    MPI_Comm_rank(MPI_COMM_WORLD, &rank);
+    edm::LogAbsolute("MPI") << "MPISource has rank " << rank << " in MPI_COMM_WORLD.";
+    int other_rank = 1 - rank;
+    comm_ = MPI_COMM_WORLD;
+    channel_ = MPIChannel(comm_, other_rank);
+  } else if (mode_ == kIntercommunicator) {
+    // Use an intercommunicator to let two groups of processes communicate with each other.
+    // The current implementation supports only two processes: one controller and one source.
+    edm::LogAbsolute("MPI") << "MPISource in " << ModeDescription[mode_] << " mode.";
+
+    // Check how many processes are there in MPI_COMM_WORLD
+    int size;
+    MPI_Comm_size(MPI_COMM_WORLD, &size);
+    if (size != 1) {
+      throw edm::Exception(edm::errors::Configuration)
+          << "The current implementation supports only two processes: one controller and one source.";
+    }
 
-  // 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);
+    // Open a server-side port.
+    MPI_Open_port(MPI_INFO_NULL, port_);
+
+    // Publish the port under the name indicated by the parameter "server".
+    std::string name = config.getUntrackedParameter<std::string>("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(name.c_str(), 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_SELF, &comm_);
+    edm::LogAbsolute("MPI") << "Connection accepted.";
+    channel_ = MPIChannel(comm_, 0);
+  } else {
+    // Invalid mode.
+    throw edm::Exception(edm::errors::Configuration)
+        << "Invalid mode \"" << config.getUntrackedParameter<std::string>("mode") << "\"";
+  }
 
-  // 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);
   edm::LogAbsolute("MPI") << "connected from " << status.MPI_SOURCE;
 }
 
 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 (mode_ == kIntercommunicator) {
+    // 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_);
+  }
 }
 
 //MPISource::ItemTypeInfo MPISource::getNextItemType() {
@@ -255,9 +312,19 @@ void MPISource::produce(edm::Event& event) {
 }
 
 void MPISource::fillDescriptions(edm::ConfigurationDescriptions& descriptions) {
+  descriptions.setComment(
+      "This module connects to an \"MPIController\" in a separate CMSSW job, receives all Runs, LuminosityBlocks and "
+      "Events from the remote process and reproduces them in the local one.");
+
   edm::ParameterSetDescription desc;
-  desc.setComment("Comunicate with another cmsRun process over MPI.");
   edm::ProducerSourceBase::fillDescription(desc);
+  desc.ifValue(
+          edm::ParameterDescription<std::string>("mode", "CommWorld", false),
+          ModeDescription[kCommWorld] >> edm::EmptyGroupDescription() or
+              ModeDescription[kIntercommunicator] >> edm::ParameterDescription<std::string>("name", "server", false))
+      ->setComment(
+          "Valid modes are CommWorld (use MPI_COMM_WORLD) and Intercommunicator (use an MPI name server to setup an "
+          "intercommunicator).");
 
   descriptions.add("source", desc);
 }