O2SimDevice.cxx

Go to the documentation of this file.

// Copyright 2019-2020 CERN and copyright holders of ALICE O2.
// See https://alice-o2.web.cern.ch/copyright for details of the copyright holders.
// All rights not expressly granted are reserved.
//
// This software is distributed under the terms of the GNU General Public
// License v3 (GPL Version 3), copied verbatim in the file "COPYING".
//
// In applying this license CERN does not waive the privileges and immunities
// granted to it by virtue of its status as an Intergovernmental Organization
// or submit itself to any jurisdiction.
 
 
#include "O2SimDevice.h"
#include "Steer/O2MCApplication.h"
#include "../macro/o2sim.C"
#include "TVirtualMC.h"
#include <fairmq/Message.h>
#include <fairmq/Parts.h>
#include <fairlogger/Logger.h>
#include <DetectorsBase/Stack.h>
#include <DetectorsBase/VMCSeederService.h>
#include <TRandom.h>
#include <SimConfig/SimConfig.h>
#include <cstring>
#include <atomic>
#include "DetectorsBase/SimRandom.h"
 
void doLogInfo(int workerID, std::string const& message)
{
  LOG(info) << "[W" << workerID << "] " << message;
}
 
namespace o2
{
namespace devices
{
 
O2SimDevice::~O2SimDevice()
{
  FairSystemInfo sysinfo;
  o2::utils::ShmManager::Instance().release();
  LOG(info) << "Shutting down O2SimDevice";
  LOG(info) << "TIME-STAMP " << mTimer.RealTime() << "\t";
  LOG(info) << "MEM-STAMP " << sysinfo.GetCurrentMemory() / (1024. * 1024) << " " << sysinfo.GetMaxMemory() << " MB\n";
}
 
void O2SimDevice::InitTask()
{
  // in the initialization phase we will init the simulation
  // NOTE: In a fair::mq::Device this is better done here (instead of outside) since
  // we have to setup simulation + worker in the same thread (due to many threadlocal variables
  // in the simulation) ... at least as long fair::mq::Device is not spawning workers on the master thread
  initSim(GetChannels().at("o2sim-primserv-info").at(0), mSimRun);
 
  // set the vmc and app pointers
  mVMC = TVirtualMC::GetMC();
  mVMCApp = static_cast<o2::steer::O2MCApplication*>(TVirtualMCApplication::Instance());
  lateInit();
}
 
void O2SimDevice::lateInit()
{
  // late init
  mVMCApp->initLate();
}
 
bool O2SimDevice::initSim(fair::mq::Channel& channel, std::unique_ptr<FairRunSim>& simptr)
{
  if (!o2::querySimConfig(channel)) {
    return false;
  }
 
  LOG(info) << "Setting up the simulation ...";
  simptr = std::move(std::unique_ptr<FairRunSim>(o2sim_init(true)));
  FairSystemInfo sysinfo;
 
  // to finish initialization (trigger further cross section table building etc) -- which especially
  // G4 is doing at the first ProcessRun
  // The goal is to have everything setup before we fork
  TVirtualMC::GetMC()->ProcessRun(0);
 
  LOG(info) << "MEM-STAMP END OF SIM INIT" << sysinfo.GetCurrentMemory() / (1024. * 1024) << " "
            << sysinfo.GetMaxMemory() << " MB\n";
 
  return true;
}
 
bool O2SimDevice::isWorkAvailable(fair::mq::Channel& statuschannel, int workerID)
{
  std::stringstream str;
  str << "[W" << workerID << "]";
  auto workerStr = str.str();
 
  int timeoutinMS = 2000; // wait for 2s max
  bool reprobe = true;
  while (reprobe) {
    reprobe = false;
    int i = -1;
    fair::mq::MessagePtr request(statuschannel.NewSimpleMessage((int)O2PrimaryServerInfoRequest::Status));
    fair::mq::MessagePtr reply(statuschannel.NewSimpleMessage(i));
    auto sendcode = statuschannel.Send(request, timeoutinMS);
    if (sendcode > 0) {
      LOG(info) << workerStr << " Waiting for status answer ";
      auto code = statuschannel.Receive(reply, timeoutinMS);
      if (code > 0) {
        int state(*((int*)(reply->GetData())));
        if (state == (int)o2::O2PrimaryServerState::ReadyToServe) {
          LOG(info) << workerStr << " SERVER IS SERVING";
          return true;
        } else if (state == (int)o2::O2PrimaryServerState::Initializing) {
          LOG(info) << workerStr << " SERVER IS STILL INITIALIZING";
          reprobe = true;
          sleep(1);
        } else if (state == (int)o2::O2PrimaryServerState::WaitingEvent) {
          LOG(info) << workerStr << " SERVER IS WAITING FOR EVENT";
          reprobe = true;
          sleep(1);
        } else if (state == (int)o2::O2PrimaryServerState::Idle) {
          LOG(info) << workerStr << " SERVER IS IDLE";
          return false;
        } else {
          LOG(info) << workerStr << " SERVER STATE UNKNOWN OR STOPPED";
        }
      } else {
        LOG(error) << workerStr << " STATUS REQUEST UNSUCCESSFUL";
      }
    }
  }
  return false;
}
 
bool O2SimDevice::Kernel(int workerID, fair::mq::Channel& requestchannel, fair::mq::Channel& dataoutchannel, fair::mq::Channel* statuschannel)
{
  bool reproducibleSim = true;
  if (getenv("O2_DISABLE_REPRODUCIBLE_SIM")) {
    reproducibleSim = false;
  }
 
  mVMCApp->setSimDataChannel(&dataoutchannel);
 
  // we log info with workerID prepended
  auto workerStr = [workerID]() {
    std::stringstream str;
    str << "[W" << workerID << "]";
    return str.str();
  };
 
  auto fetched = fetchChunk(workerID, requestchannel);
  if (fetched.status == ChunkStatus::Stop) {
    return false;
  }
  if (fetched.status != ChunkStatus::Work) {
    // EndOfWork deliberately keeps asking: the server answers the next request as idle, which is the Stop
    return true;
  }
  auto& chunk = fetched.chunk;
 
  auto info = chunk->mSubEventInfo;
  LOG(info) << workerStr() << " Processing " << chunk->mParticles.size() << " primary particles "
            << "for event " << info.eventID << "/" << info.maxEvents << " "
            << "part " << info.part << "/" << info.nparts;
 
  // for debugging, O2SIM_RESTRICT_EVENTPART="event:part" transports only that sub-event
  if (o2::steer::O2MCApplication::isSelectedSubEvent(info.eventID, info.part)) {
    mVMCApp->setPrimaries(chunk->mParticles);
  } else {
    // nothing to transport here
    mVMCApp->setPrimaries(std::vector<TParticle>{});
    LOG(info) << workerStr() << " This chunk will be skipped";
  }
 
  mVMCApp->setSubEventInfo(&info);
 
  if (reproducibleSim) {
    LOG(info) << workerStr() << " Setting seed for this sub-event to " << chunk->mSubEventInfo.seed;
    o2::base::simRandom()->SetSeed(chunk->mSubEventInfo.seed);
    o2::base::VMCSeederService::instance().setSeed();
  }
 
  // Process one event
  auto& conf = o2::conf::SimConfig::Instance();
  if (strcmp(conf.getMCEngine().c_str(), "TGeant4") == 0 || strcmp(conf.getMCEngine().c_str(), "O2TrivialMCEngine") == 0) {
    // this is preferred and necessary for Geant4
    // since repeated "ProcessRun" might have significant overheads
    mVMC->ProcessEvent();
  } else {
    // for Geant3 calling ProcessEvent is not enough
    // as some hooks are not called
    mVMC->ProcessRun(1);
  }
 
  FairSystemInfo sysinfo;
  LOG(info) << workerStr() << " TIME-STAMP " << mTimer.RealTime() << "\t";
  mTimer.Continue();
  LOG(info) << workerStr() << " MEM-STAMP " << sysinfo.GetCurrentMemory() / (1024. * 1024) << " "
            << sysinfo.GetMaxMemory() << " MB\n";
  return true;
}
 
O2SimDevice::FetchedChunk O2SimDevice::fetchChunk(int workerID, fair::mq::Channel& requestchannel)
{
  static std::atomic<int> counter{0};                                                                          // shared by the transport threads in multithreaded mode
  fair::mq::MessagePtr request(requestchannel.NewSimpleMessage(PrimaryChunkRequest{workerID, -1, counter++})); // <-- don't need content; channel means -> give primaries
  fair::mq::Parts reply;
 
  auto workerStr = [workerID]() {
    std::stringstream str;
    str << "[W" << workerID << "]";
    return str.str();
  };
 
  doLogInfo(workerID, "Requesting work chunk");
  int timeoutinMS = 2000;
  auto sendcode = requestchannel.Send(request, timeoutinMS);
  if (sendcode <= 0) {
    LOG(info) << workerStr() << " Requesting work from server not possible. Return code " << sendcode;
    return {ChunkStatus::Stop, nullptr};
  }
  doLogInfo(workerID, "Waiting for answer");
  // asking for primary generation
  auto code = requestchannel.Receive(reply);
  if (code <= 0) {
    LOG(info) << workerStr() << " No primary answer received from server (within timeout). Return code " << code;
    return {ChunkStatus::Retry, nullptr};
  }
  doLogInfo(workerID, "Primary chunk received");
  auto rawmessage = std::move(reply.At(0));
  auto header = *(o2::PrimaryChunkAnswer*)(rawmessage->GetData());
  if (!header.payload_attached) {
    doLogInfo(workerID, "No payload; Server in stage " + std::string(PrimStateToString[(int)header.serverstate]));
    // if no payload attached we inspect the server state, to see what to do
    if (header.serverstate == O2PrimaryServerState::Initializing || header.serverstate == O2PrimaryServerState::WaitingEvent) {
      sleep(1); // back-off and retry
      return {ChunkStatus::Retry, nullptr};
    }
    // we need to decide what to do when the server is idle ---> if this happens immediately after a new batch request it means that the server might just lag a bit behind
    return {ChunkStatus::Stop, nullptr};
  }
  auto payload = std::move(reply.At(1));
  // wrap incoming bytes as a TMessageWrapper which offers "adoption" of a buffer
  TMessageWrapper message(payload->GetData(), payload->GetSize());
  std::unique_ptr<o2::data::PrimaryChunk> chunk(static_cast<o2::data::PrimaryChunk*>(message.ReadObjectAny(message.GetClass())));
 
  // no particles and eventID == -1 --> indication for no more work
  if (chunk->mParticles.size() == 0 && chunk->mSubEventInfo.eventID == -1) {
    doLogInfo(workerID, "No particles in reply : quitting kernel");
    return {ChunkStatus::EndOfWork, nullptr};
  }
  return {ChunkStatus::Work, std::move(chunk)};
}
 
bool O2SimDevice::ConditionalRun()
{
  return Kernel(-1, GetChannels().at("primary-get").at(0), GetChannels().at("simdata").at(0));
}
 
void O2SimDevice::PostRun() { LOG(info) << "Shutting down "; }
 
} // namespace devices
} // namespace o2