ControllerChannel.h

Go to the documentation of this file.

#ifndef FWCore_SharedMemory_ControllerChannel_h
#define FWCore_SharedMemory_ControllerChannel_h
// -*- C++ -*-
//
// Package:     FWCore/SharedMemory
// Class  :     ControllerChannel
//
//
// Original Author:  Chris Jones
//         Created:  21/01/2020
//
 
// system include files
#include <string>
#include <iostream>
#include "boost/interprocess/managed_shared_memory.hpp"
#include "boost/interprocess/sync/named_mutex.hpp"
#include "boost/interprocess/sync/named_condition.hpp"
#include "boost/interprocess/sync/scoped_lock.hpp"
 
// user include files
#include "FWCore/Utilities/interface/Transition.h"
#include "FWCore/Utilities/interface/Exception.h"
#include "FWCore/SharedMemory/interface/BufferInfo.h"
 
// forward declarations
 
namespace edm::shared_memory {
  class ControllerChannel {
  public:
    ControllerChannel(std::string const& iName, int iID, unsigned int iMaxWaitInSeconds);
    ~ControllerChannel();
    ControllerChannel(const ControllerChannel&) = delete;
    const ControllerChannel& operator=(const ControllerChannel&) = delete;
    ControllerChannel(ControllerChannel&&) = delete;
    const ControllerChannel& operator=(ControllerChannel&&) = delete;
 
    // ---------- member functions ---------------------------
 
    template <typename F>
    void setupWorker(F&& iF) {
      using namespace boost::interprocess;
      scoped_lock<named_mutex> lock(mutex_);
      iF();
      using namespace boost::posix_time;
      //std::cout << id_ << " waiting for external process" << std::endl;
 
      if (not cndToMain_.timed_wait(lock, microsec_clock::universal_time() + seconds(maxWaitInSeconds_))) {
        //std::cout << id_ << " FAILED waiting for external process" << std::endl;
        throw cms::Exception("ExternalFailed");
      } else {
        //std::cout << id_ << " done waiting for external process" << std::endl;
      }
    }
 
    template <typename F>
    bool doTransition(F&& iF, edm::Transition iTrans, unsigned long long iTransitionID) {
      using namespace boost::interprocess;
 
      //std::cout << id_ << " taking from lock" << std::endl;
      scoped_lock<named_mutex> lock(mutex_);
 
      if (not wait(lock, iTrans, iTransitionID)) {
        return false;
      }
      //std::cout <<id_<<"running doTranstion command"<<std::endl;
      iF();
      return true;
    }
 
    BufferInfo* toWorkerBufferInfo() { return toWorkerBufferInfo_; }
    BufferInfo* fromWorkerBufferInfo() { return fromWorkerBufferInfo_; }
 
    void stopWorker() {
      //std::cout <<"stopWorker"<<std::endl;
      using namespace boost::interprocess;
      scoped_lock<named_mutex> lock(mutex_);
      *stop_ = true;
      //std::cout <<"stopWorker sending notification"<<std::endl;
      cndFromMain_.notify_all();
    }
 
    // ---------- const member functions ---------------------------
    std::string const& sharedMemoryName() const { return smName_; }
    std::string uniqueID() const { return uniqueName(""); }
 
    //should only be called after calling `doTransition`
    bool shouldKeepEvent() const { return *keepEvent_; }
 
  private:
    static BufferInfo* bufferInfo(const char* iWhich, boost::interprocess::managed_shared_memory& mem);
 
    std::string uniqueName(std::string iBase) const;
 
    bool wait(boost::interprocess::scoped_lock<boost::interprocess::named_mutex>& lock,
              edm::Transition iTrans,
              unsigned long long iTransID);
 
    // ---------- member data --------------------------------
    int id_;
    unsigned int maxWaitInSeconds_;
    std::string smName_;
    boost::interprocess::managed_shared_memory managed_sm_;
    BufferInfo* toWorkerBufferInfo_;
    BufferInfo* fromWorkerBufferInfo_;
 
    boost::interprocess::named_mutex mutex_;
    boost::interprocess::named_condition cndFromMain_;
 
    boost::interprocess::named_condition cndToMain_;
 
    edm::Transition* transitionType_;
    unsigned long long* transitionID_;
    bool* stop_;
    bool* keepEvent_;
  };
}  // namespace edm::shared_memory
 
#endif