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. More...

virtual StatusCode	initialize () override final
	Initialize. More...

virtual StatusCode	finalize () override final
	Finalize. More...

virtual int	numRanks () const override final
	Return number of ranks. More...

virtual int	rank () const override final
	Return our rank. More...

virtual void	barrier () override final
	Insert a barrier No rank will continue until all ranks reach this point. More...

virtual void	abort () override final
	Abort the MPI run. More...

virtual void	sendMessage (int destRank, ClusterMessage message, ClusterComm communicator=ClusterComm::Default) override final
	Send an MPI message. More...

virtual ClusterMessage	waitReceiveMessage (ClusterComm communicator=ClusterComm::Default) override final
	Block until we receive an MPI message. More...

virtual mpi3::communicator &	data_communicator () override final
	Return the data communicator. More...

virtual void	log_addEvent (int eventIdx, std::int64_t run_number, std::int64_t event_number, std::int64_t request_time_ns) override final
	Add (begin) an event in the log. More...

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. More...

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

Detailed Description

A service managing communications within a cluster using MPI.

Definition at line 23 of file MPIClusterSvc.h.

Constructor & Destructor Documentation

◆ MPIClusterSvc()

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

inline

Constructor.

Definition at line 26 of file MPIClusterSvc.h.

27 : extends(name, svcLoc) {}

Member Function Documentation

◆ abort()

void MPIClusterSvc::abort ( )

finaloverridevirtual

Abort the MPI run.

Definition at line 74 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 69 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 58 of file MPIClusterSvc.h.

                                                                {
     return m_datacom;
   }

◆ finalize()

StatusCode MPIClusterSvc::finalize ( )

finaloverridevirtual

Finalize.

Definition at line 52 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;
 }

◆ initialize()

StatusCode MPIClusterSvc::initialize ( )

finaloverridevirtual

Initialize.

Definition at line 8 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 event_log (rank INTEGER, id INTEGER UNIQUE,"
           "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))")
       .run();
   m_mpiLog_addEvent = m_mpiLog->createStatement(
       "INSERT INTO event_log(id, rank, runNumber, eventNumber, complete, "
       "start_time, request_time_ns) "
       "VALUES(?1, ?4, ?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");
   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
	)

finaloverridevirtual

Add (begin) an event in the log.

Definition at line 157 of file MPIClusterSvc.cxx.

                                                              {
   m_mpiLog_addEvent.run(eventIdx, run_number, event_number, m_rank,
                         request_time_ns);
 }

◆ 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 164 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 61 of file MPIClusterSvc.cxx.

                                   {
   return m_world.size();
 }

◆ rank()

int MPIClusterSvc::rank ( ) const

finaloverridevirtual

Return our rank.

Definition at line 65 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 78 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 120 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 75 of file MPIClusterSvc.h.

◆ m_env

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

private

Definition at line 72 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 79 of file MPIClusterSvc.h.

◆ m_mpiLog_addEvent

SQLite::Statement MPIClusterSvc::m_mpiLog_addEvent

private

Definition at line 81 of file MPIClusterSvc.h.

◆ m_mpiLog_completeEvent

SQLite::Statement MPIClusterSvc::m_mpiLog_completeEvent

private

Definition at line 82 of file MPIClusterSvc.h.

◆ m_rank

int MPIClusterSvc::m_rank = -1

private

Definition at line 76 of file MPIClusterSvc.h.

◆ m_world

mpi3::communicator MPIClusterSvc::m_world

private

Definition at line 73 of file MPIClusterSvc.h.

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

Public Member Functions

Private Attributes

Detailed Description

Constructor & Destructor Documentation

◆ MPIClusterSvc()

Member Function Documentation

◆ abort()

◆ barrier()

◆ data_communicator()

◆ finalize()

◆ initialize()

◆ log_addEvent()

◆ log_completeEvent()

◆ numRanks()

◆ rank()

◆ sendMessage()

◆ waitReceiveMessage()

Member Data Documentation

◆ m_datacom

◆ m_env

◆ m_mpiLog

◆ m_mpiLog_addEvent

◆ m_mpiLog_completeEvent

◆ m_rank

◆ m_world