WorkerChannel.h

Go to the documentation of this file.

#ifndef FWCore_SharedMemory_WorkerChannel_h
#define FWCore_SharedMemory_WorkerChannel_h
// -*- C++ -*-
//
// Package:     FWCore/SharedMemory
// Class  :     WorkerChannel
//
//
// Original Author:  Chris Jones
//         Created:  21/01/2020
//

// system include files
#include <string>
#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/SharedMemory/interface/BufferInfo.h"

// forward declarations

namespace edm::shared_memory {
  class WorkerChannel {
  public:
    WorkerChannel(std::string const& iName, const std::string& iUniqueID);
    WorkerChannel(const WorkerChannel&) = delete;
    const WorkerChannel& operator=(const WorkerChannel&) = delete;
    WorkerChannel(WorkerChannel&&) = delete;
    const WorkerChannel& operator=(WorkerChannel&&) = delete;

    // ---------- member functions ---------------------------
    boost::interprocess::scoped_lock<boost::interprocess::named_mutex>* accessLock() { return &lock_; }

    BufferInfo* toWorkerBufferInfo() { return toWorkerBufferInfo_; }
    BufferInfo* fromWorkerBufferInfo() { return fromWorkerBufferInfo_; }

    void workerSetupDone() {
      //The controller is waiting for the worker to be setup
      notifyController();
    }

    template <typename F>
    void handleTransitions(F&& iF) {
      if (stopRequested()) {
        return;
      }
      while (true) {
        waitForController();
        if (stopRequested()) {
          break;
        }

        iF(transition(), transitionID());
        notifyController();
      }
    }

    void shouldKeepEvent(bool iChoice) { *keepEvent_ = iChoice; }

    void notifyController() {
      //change in transitionID_ used to signal worker finished
      *transitionID_ = ~(*transitionID_);
      cndToController_.notify_all();
    }
    void waitForController() { cndFromController_.wait(lock_); }

    // ---------- const member functions ---------------------------
    edm::Transition transition() const noexcept { return *transitionType_; }
    unsigned long long transitionID() const noexcept { return *transitionID_; }
    bool stopRequested() const noexcept { return *stop_; }

  private:
    // ---------- member data --------------------------------
    boost::interprocess::managed_shared_memory managed_shm_;

    boost::interprocess::named_mutex mutex_;
    boost::interprocess::named_condition cndFromController_;
    bool* stop_;
    edm::Transition* transitionType_;
    unsigned long long* transitionID_;
    BufferInfo* toWorkerBufferInfo_;
    BufferInfo* fromWorkerBufferInfo_;
    boost::interprocess::named_condition cndToController_;
    bool* keepEvent_;
    boost::interprocess::scoped_lock<boost::interprocess::named_mutex> lock_;
  };
}  // namespace edm::shared_memory

#endif