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