MPIClusterSvc Class Reference

A service managing communications within a cluster using MPI. More...

#include <MPIClusterSvc.h>

Inheritance diagram for MPIClusterSvc:

Collaboration diagram for MPIClusterSvc:

Public Member Functions
	MPIClusterSvc (const std::string &name, ISvcLocator *svcLoc)
	Constructor.
virtual StatusCode	initialize () override final
	Initialize.
virtual StatusCode	finalize () override final
	Finalize.
virtual void	handle (const Incident &inc) override
	IIncidentListener handle.
virtual int	numRanks () const override final
	Return number of ranks.
virtual int	rank () const override final
	Return our rank.
virtual void	barrier () override final
	Insert a barrier No rank will continue until all ranks reach this point.
virtual void	abort () override final
	Abort the MPI run.
virtual void	sendMessage (int destRank, ClusterMessage message, ClusterComm communicator=ClusterComm::Default) override final
	Send an MPI message.
virtual ClusterMessage	waitReceiveMessage (ClusterComm communicator=ClusterComm::Default) override final
	Block until we receive an MPI message.
virtual mpi3::communicator &	data_communicator () override final
	Return the data communicator.
virtual void	log_addEvent (int eventIdx, std::int64_t run_number, std::int64_t event_number, std::int64_t request_time_ns, std::size_t slot) override final
	Add (begin) an event in the log.
virtual void	log_completeEvent (std::int64_t run_number, std::int64_t event_number, std::int64_t status) override final
	Complete an event in the log.

Private Attributes
std::unique_ptr< mpi3::environment >	m_env
mpi3::communicator	m_world
mpi3::communicator	m_datacom
int	m_rank = -1
ServiceHandle< ISQLiteDBSvc >	m_mpiLog
SQLite::Statement	m_mpiLog_addEvent
SQLite::Statement	m_mpiLog_completeEvent
SQLite::Statement	m_mpiLog_addFile
std::int64_t	m_lastInputFileHash {}
std::map< std::size_t, std::int64_t >	m_inputFileSlotMap {}

Detailed Description

A service managing communications within a cluster using MPI.

Definition at line 26 of file MPIClusterSvc.h.

Constructor & Destructor Documentation

MPIClusterSvc()

MPIClusterSvc::MPIClusterSvc ( const std::string & name, ISvcLocator * svcLoc )

inline

Constructor.

Definition at line 29 of file MPIClusterSvc.h.

      : extends(name, svcLoc) {}

Member Function Documentation

abort()

void MPIClusterSvc::abort ( )

finaloverridevirtual

Abort the MPI run.

Definition at line 119 of file MPIClusterSvc.cxx.

                          {
  m_world.abort();
}

barrier()

void MPIClusterSvc::barrier ( )

finaloverridevirtual

Insert a barrier No rank will continue until all ranks reach this point.

Definition at line 114 of file MPIClusterSvc.cxx.

                            {
  ATH_MSG_DEBUG("Barrier on rank " << rank() << " of " << numRanks());
  m_world.barrier();
}

data_communicator()

virtual mpi3::communicator & MPIClusterSvc::data_communicator ( )

inlinefinaloverridevirtual

Return the data communicator.

Definition at line 64 of file MPIClusterSvc.h.

                                                               {
    return m_datacom;
  }

finalize()

StatusCode MPIClusterSvc::finalize ( )

finaloverridevirtual

Finalize.

Definition at line 72 of file MPIClusterSvc.cxx.

                                   {
  m_mpiLog
      ->createStatement(
          "UPDATE ranks SET end_time = julianday('now') WHERE rank = ?1")
      .run(m_rank);
  m_env.reset(nullptr);
  return StatusCode::SUCCESS;
}

handle()

void MPIClusterSvc::handle ( const Incident & inc )

overridevirtual

IIncidentListener handle.

Handles BeginInputFile to keep track of which input file an event came from.

Definition at line 82 of file MPIClusterSvc.cxx.

                                              {
  // Fill in slot map at start of every event
  if (inc.type() == IncidentType::BeginProcessing) {
    const std::size_t slot = Gaudi::Hive::currentContext().slot();
    m_inputFileSlotMap[slot] = m_lastInputFileHash;
  }
 
  // Cache new input filename on start of every file
  if (inc.type() == IncidentType::BeginInputFile) {
    const FileIncident* fileInc = dynamic_cast<const FileIncident*>(&inc);
    if (fileInc == nullptr) {
      ATH_MSG_ERROR("BeginInputFile does not have a file name attached");
      return;
    }
 
    const std::string fileName = fileInc->fileName();
    // Convert the hash into a signed int64. Just a hash so this doesn't matter.
    m_lastInputFileHash = static_cast<std::int64_t>(xxh3::hash64(fileName));
    m_mpiLog_addFile.run(m_lastInputFileHash, std::move(fileName));
  }
  return;
}

initialize()

StatusCode MPIClusterSvc::initialize ( )

finaloverridevirtual

Initialize.

Definition at line 11 of file MPIClusterSvc.cxx.

                                     {
  ATH_MSG_DEBUG("Initializing MPI");
  m_env = std::make_unique<mpi3::environment>(mpi3::thread_level::single);
  ATH_MSG_DEBUG("Created MPI environment");
  m_world = m_env->world();
  m_datacom =
      m_world.duplicate();  // make a duplicate communicator for event data
  ATH_MSG_DEBUG("Got MPI_COMM_WORLD");
  m_rank = m_world.rank();
  ATH_MSG_INFO("On MPI rank " << m_rank);
 
  ATH_CHECK(m_mpiLog.retrieve());
  m_mpiLog->createStatement("PRAGMA foreign_keys = ON").run();
 
  m_mpiLog
      ->createStatement(
          "CREATE TABLE ranks (rank INTEGER PRIMARY KEY, "
          "node TEXT, start_time FLOAT, end_time FLOAT)")
      .run();
  m_mpiLog
      ->createStatement(
          "INSERT INTO ranks (rank, node, start_time) "
          "VALUES(?1, ?2, julianday('now'))")
      .run(m_rank, m_env->processor_name());
  m_mpiLog->createStatement(
      "CREATE TABLE files (fileId INTEGER PRIMARY KEY, fileName TEXT)")
      .run();
  m_mpiLog
      ->createStatement(
          "CREATE TABLE event_log (rank INTEGER, id INTEGER UNIQUE,"
          "inputFileId INTEGER,"
          "runNumber INTEGER, eventNumber INTEGER, complete INTEGER,"
          "status INTEGER, request_time_ns INTEGER, start_time FLOAT,"
          "end_time FLOAT, PRIMARY KEY (runNumber, eventNumber), "
          "FOREIGN KEY (rank) REFERENCES ranks(rank),"
          "FOREIGN KEY (inputFileId) REFERENCES files(fileId))")
      .run();
  m_mpiLog_addEvent = m_mpiLog->createStatement(
      "INSERT INTO event_log(id, rank, inputFileId, runNumber, eventNumber, complete, "
      "start_time, request_time_ns) "
      "VALUES(?1, ?4, ?6, ?2, ?3, 0, julianday('now'), ?5)");
  m_mpiLog_completeEvent = m_mpiLog->createStatement(
      "UPDATE event_log SET complete = 1, status = ?3, end_time = "
      "julianday('now') WHERE runNumber = ?1 "
      "AND "
      "eventNumber = ?2");
  m_mpiLog_addFile = m_mpiLog->createStatement(
      "INSERT INTO files (fileId, fileName) VALUES(?1, ?2)");
 
  // Set up incident listener
  ServiceHandle<IIncidentSvc> incsvc("IncidentSvc", this->name());
  if (!incsvc.retrieve().isSuccess()) {
    ATH_MSG_FATAL("Cannot get IncidentSvc.");
    return(StatusCode::FAILURE);
  }
  incsvc->addListener(this, IncidentType::BeginInputFile, 100);
  incsvc->addListener(this, IncidentType::BeginProcessing, 100);
 
  return StatusCode::SUCCESS;
}

log_addEvent()

void MPIClusterSvc::log_addEvent ( int eventIdx, std::int64_t run_number, std::int64_t event_number, std::int64_t request_time_ns, std::size_t slot )

finaloverridevirtual

Add (begin) an event in the log.

Definition at line 202 of file MPIClusterSvc.cxx.

                                                 {
  m_mpiLog_addEvent.run(eventIdx, run_number, event_number, m_rank,
                        request_time_ns,
                        m_inputFileSlotMap[slot]);
}

log_completeEvent()

void MPIClusterSvc::log_completeEvent ( std::int64_t run_number, std::int64_t event_number, std::int64_t status )

finaloverridevirtual

Complete an event in the log.

Definition at line 211 of file MPIClusterSvc.cxx.

                                                         {
  m_mpiLog_completeEvent.run(run_number, event_number, status);
}

numRanks()

int MPIClusterSvc::numRanks ( ) const

finaloverridevirtual

Return number of ranks.

Definition at line 106 of file MPIClusterSvc.cxx.

                                  {
  return m_world.size();
}

rank()

int MPIClusterSvc::rank ( ) const

finaloverridevirtual

Return our rank.

Definition at line 110 of file MPIClusterSvc.cxx.

                              {
  return m_rank;
}

sendMessage()

void MPIClusterSvc::sendMessage ( int destRank, ClusterMessage message, ClusterComm communicator = ClusterComm::Default )

finaloverridevirtual

Send an MPI message.

Definition at line 123 of file MPIClusterSvc.cxx.

                                                          {
  ATH_MSG_DEBUG("Sending message from rank " << rank() << " to " << destRank);
  // Don't send event request message if we're not the master *and* we have a
  // message waiting.
  // Probably an emergency stop message
  if (m_rank != 0 && message.messageType == ClusterMessageType::RequestEvent &&
      m_world.iprobe().has_value()) {
    return;
  }
 
  // Select correct communicator
  mpi3::communicator& comm =
      (communicator == ClusterComm::EventData) ? m_datacom : m_world;
  if (message.messageType == ClusterMessageType::Data &&
      communicator != ClusterComm::EventData) {
    ATH_MSG_WARNING(
        "Event data should be sent with EventData communicator. "
        "Dropping message");
    return;
  }
 
  message.source = m_rank;
  const auto& [header, body] = message.wire_msg();
  comm.send_n(header.begin(), header.size(), destRank, 0);
  if (body.has_value()) {
    comm.send_n(body->begin(), body->size(), destRank, header[2]);
    if (message.messageType == ClusterMessageType::Data) {
      const ClusterMessage::WireMsgBody& bdy = *body;
      // Decode the body to figure out what to send
      char* ptr = reinterpret_cast<char*>((std::uint64_t(bdy[0]) << 32) +
                                          std::uint64_t(bdy[1]));
      std::size_t len = (std::uint64_t(bdy[2]) << 32) + std::uint64_t(bdy[3]);
 
      // Offset the tag by 16384 to minimize chance of conflict
      // (max tag in MPI spec is 32767)
      constexpr int tag_offset = 16384;
      comm.send_n(ptr, len, destRank, header[2] + tag_offset);
    }
  }
}

waitReceiveMessage()

ClusterMessage MPIClusterSvc::waitReceiveMessage ( ClusterComm communicator = ClusterComm::Default )

finaloverridevirtual

Block until we receive an MPI message.

Definition at line 165 of file MPIClusterSvc.cxx.

                                                                         {
  // Same offset as line 114
  constexpr int tag_offset = 16384;
  constexpr std::uint64_t thirtytwo_ones = 0xFFFFFFFF;
 
  // Select correct communicator
  mpi3::communicator& comm =
      (communicator == ClusterComm::EventData) ? m_datacom : m_world;
  ClusterMessage::WireMsg msg{};
  auto&& [head, body] = msg;
  comm.receive_n(head.begin(), head.size());
  // Only time we need to figure out ourselves whether there's a body
  if (head[0] == int(ClusterMessageType::FinalWorkerStatus) ||
      head[0] == int(ClusterMessageType::WorkerError) ||
      head[0] == int(ClusterMessageType::Data)) {
    body = ClusterMessage::WireMsgBody{};
    comm.receive_n(body->begin(), body->size(), head[1], head[2]);
    if (head[0] == int(ClusterMessageType::Data)) {
      ClusterMessage::WireMsgBody& bdy = *body;
      // Decode the body to figure out what to recieve
      std::size_t len = (std::uint64_t(bdy[2]) << 32) + std::uint64_t(bdy[3]);
      std::size_t align = (std::uint64_t(bdy[4]) << 32) + std::uint64_t(bdy[5]);
 
      char* ptr = static_cast<char*>(std::aligned_alloc(align, len));
      comm.receive_n(ptr, len, head[1], head[2] + tag_offset);
 
      // update the pointer in the WireMsgBody
      bdy[0] = int(std::uint64_t(ptr) >> 32);
      bdy[1] = int(std::uint64_t(ptr) & thirtytwo_ones);
    }
  }
  ClusterMessage message(msg);
  ATH_MSG_DEBUG("Rank " << rank() << " received message from "
                        << message.source);
  return message;
}

Member Data Documentation

m_datacom

mpi3::communicator MPIClusterSvc::m_datacom

private

Definition at line 82 of file MPIClusterSvc.h.

m_env

std::unique_ptr<mpi3::environment> MPIClusterSvc::m_env

private

Definition at line 79 of file MPIClusterSvc.h.

m_inputFileSlotMap

std::map<std::size_t, std::int64_t> MPIClusterSvc::m_inputFileSlotMap {}

private

Definition at line 94 of file MPIClusterSvc.h.

{};

m_lastInputFileHash

std::int64_t MPIClusterSvc::m_lastInputFileHash {}

private

Definition at line 93 of file MPIClusterSvc.h.

{};

m_mpiLog

ServiceHandle<ISQLiteDBSvc> MPIClusterSvc::m_mpiLog

private

Initial value:

{this, "LogDatabaseSvc", "",
                                       "SQLiteDBSvc for the MPI event log"}

Definition at line 86 of file MPIClusterSvc.h.

                                      {this, "LogDatabaseSvc", "",
                                       "SQLiteDBSvc for the MPI event log"};

m_mpiLog_addEvent

SQLite::Statement MPIClusterSvc::m_mpiLog_addEvent

private

Definition at line 88 of file MPIClusterSvc.h.

m_mpiLog_addFile

SQLite::Statement MPIClusterSvc::m_mpiLog_addFile

private

Definition at line 90 of file MPIClusterSvc.h.

m_mpiLog_completeEvent

SQLite::Statement MPIClusterSvc::m_mpiLog_completeEvent

private

Definition at line 89 of file MPIClusterSvc.h.

m_rank

int MPIClusterSvc::m_rank = -1

private

Definition at line 83 of file MPIClusterSvc.h.

m_world

mpi3::communicator MPIClusterSvc::m_world

private

Definition at line 80 of file MPIClusterSvc.h.

---

The documentation for this class was generated from the following files:

• MPIClusterSvc.h
• MPIClusterSvc.cxx