MPIClusterSvc.h

Go to the documentation of this file.

/*
  Copyright (C) 2002-2026 CERN for the benefit of the ATLAS collaboration
*/
#ifndef ATHENASERVICES_MPICLUSTERSVC_H_
#define ATHENASERVICES_MPICLUSTERSVC_H_
 
#include <memory>
#include <string>
 
#include "AthenaBaseComps/AthService.h"
#include "AthenaKernel/ClusterMessage.h"
#include "AthenaKernel/IMPIClusterSvc.h"
#include "GaudiKernel/IIncidentListener.h"
#include "SQLiteDBSvc/ISQLiteDBSvc.h"
#include "SQLiteDBSvc/Statement.h"
#include "mpi3/environment.hpp"
 
namespace mpi3 = boost::mpi3;

class MPIClusterSvc
    : public extends<AthService, IMPIClusterSvc, IIncidentListener> {
 public:
  MPIClusterSvc(const std::string& name, ISvcLocator* svcLoc)
      : extends(name, svcLoc) {}

  virtual StatusCode initialize() override final;

  virtual StatusCode finalize() override final;

  virtual void handle(const Incident& inc) override;

  virtual int numRanks() const override final;

  virtual int rank() const override final;

  virtual void barrier() override final;

  virtual void abort() override final;

  virtual void sendMessage(
      int destRank, ClusterMessage message,
      ClusterComm communicator = ClusterComm::Default) override final;

  virtual ClusterMessage waitReceiveMessage(
      ClusterComm communicator = ClusterComm::Default) override final;

  virtual mpi3::communicator& data_communicator() override final {
    return m_datacom;
  }

  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;
  virtual void log_completeEvent(int eventIdx, std::int64_t run_number,
                                 std::int64_t event_number,
                                 std::int64_t status) override final;
 
 private:
  std::unique_ptr<mpi3::environment> m_env;
  mpi3::communicator m_world;
  // Communicator for payload of event data messages
  mpi3::communicator m_datacom;
  int m_rank = -1;
 
  // MPI Log DB
  ServiceHandle<ISQLiteDBSvc> m_mpiLog{this, "LogDatabaseSvc", "",
                                       "SQLiteDBSvc for the MPI event log"};
  SQLite::Statement m_mpiLog_addEvent;
  SQLite::Statement m_mpiLog_completeEvent;
  SQLite::Statement m_mpiLog_addFile;
 
  // Hold current input filename hash for each slot
  std::int64_t m_lastInputFileHash{};
  std::map<std::size_t, std::int64_t> m_inputFileSlotMap{};
};
#endif  // ATHENASERVICES_MPICLUSTERSVC_H_