#include <ControllerChannel.h>

Classes
struct	CheckWorkerStatus

Public Member Functions
	ControllerChannel (std::string const &iName, int iID, unsigned int iMaxWaitInSeconds)

	ControllerChannel (const ControllerChannel &)=delete

	ControllerChannel (ControllerChannel &&)=delete

template<typename F >
bool	doTransition (F &&iF, edm::Transition iTrans, unsigned long long iTransitionID)

template<typename F , typename FRETRY >
bool	doTransitionWithRetry (F &&iF, FRETRY &&iRetry, edm::Transition iTrans, unsigned long long iTransitionID)

BufferInfo *	fromWorkerBufferInfo ()
	This can be used with ReadBuffer to keep Controller and Worker in sync. More...

unsigned int	maxWaitInSeconds () const

const ControllerChannel &	operator= (const ControllerChannel &)=delete

const ControllerChannel &	operator= (ControllerChannel &&)=delete

template<typename F >
void	setupWorker (F &&iF)

template<typename F , typename FRETRY >
void	setupWorkerWithRetry (F &&iF, FRETRY &&iRetry)

std::string const &	sharedMemoryName () const

bool	shouldKeepEvent () const

void	stopWorker ()

BufferInfo *	toWorkerBufferInfo ()
	This can be used with WriteBuffer to keep Controller and Worker in sync. More...

std::string	uniqueID () const

	~ControllerChannel ()

Private Member Functions
bool	continueWait (boost::interprocess::scoped_lock< boost::interprocess::named_mutex > &lock)

CheckWorkerStatus	initCheckWorkerStatus (unsigned long long *iPtr) const

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)

bool	wait (boost::interprocess::scoped_lock< boost::interprocess::named_mutex > &lock)

Static Private Member Functions
static BufferInfo *	bufferInfo (const char *iWhich, boost::interprocess::managed_shared_memory &mem)

Private Attributes
boost::interprocess::named_condition	cndFromMain_

boost::interprocess::named_condition	cndToMain_

BufferInfo *	fromWorkerBufferInfo_

int	id_

bool *	keepEvent_

boost::interprocess::managed_shared_memory	managed_sm_

unsigned int	maxWaitInSeconds_

boost::interprocess::named_mutex	mutex_

std::string	smName_

bool *	stop_

BufferInfo *	toWorkerBufferInfo_

unsigned long long *	transitionID_

edm::Transition *	transitionType_

Detailed Description

Definition at line 37 of file ControllerChannel.h.

Constructor & Destructor Documentation

ControllerChannel::ControllerChannel	(	std::string const &	iName,
		int	iID,
		unsigned int	iMaxWaitInSeconds
	)

iName is used as the base for the shared memory name. The full name uses iID as well as getpid() to create the value sharedMemoryName(). iID allows multiple ControllChannels to use the same base name iName.

Definition at line 34 of file ControllerChannel.cc.

     : id_{id},
       maxWaitInSeconds_{iMaxWaitInSeconds},
       smName_{uniqueName(iName)},
       managed_sm_{open_or_create, smName_.c_str(), 1024},
       toWorkerBufferInfo_{bufferInfo(channel_names::kToWorkerBufferInfo, managed_sm_)},
       fromWorkerBufferInfo_{bufferInfo(channel_names::kFromWorkerBufferInfo, managed_sm_)},
       mutex_{open_or_create, uniqueName(channel_names::kMutex).c_str()},
       cndFromMain_{open_or_create, uniqueName(channel_names::kConditionFromMain).c_str()},
       cndToMain_{open_or_create, uniqueName(channel_names::kConditionToMain).c_str()} {
   managed_sm_.destroy<bool>(channel_names::kStop);
   stop_ = managed_sm_.construct<bool>(channel_names::kStop)(false);
   assert(stop_);
   keepEvent_ = managed_sm_.construct<bool>(channel_names::kKeepEvent)(true);
   assert(keepEvent_);
 
   managed_sm_.destroy<edm::Transition>(channel_names::kTransitionType);
   transitionType_ =
       managed_sm_.construct<edm::Transition>(channel_names::kTransitionType)(edm::Transition::NumberOfTransitions);
   assert(transitionType_);
 
   managed_sm_.destroy<unsigned long long>(channel_names::kTransitionID);
   transitionID_ = managed_sm_.construct<unsigned long long>(channel_names::kTransitionID)(0);
   assert(transitionID_);
 }

ControllerChannel::~ControllerChannel ( )

Definition at line 60 of file ControllerChannel.cc.

                                       {
   managed_sm_.destroy<bool>(channel_names::kKeepEvent);
   managed_sm_.destroy<bool>(channel_names::kStop);
   managed_sm_.destroy<unsigned int>(channel_names::kTransitionType);
   managed_sm_.destroy<unsigned long long>(channel_names::kTransitionID);
   managed_sm_.destroy<BufferInfo>(channel_names::kToWorkerBufferInfo);
   managed_sm_.destroy<BufferInfo>(channel_names::kFromWorkerBufferInfo);
 
   named_mutex::remove(uniqueName(channel_names::kMutex).c_str());
   named_condition::remove(uniqueName(channel_names::kConditionFromMain).c_str());
   named_condition::remove(uniqueName(channel_names::kConditionToMain).c_str());
 }

edm::shared_memory::ControllerChannel::ControllerChannel ( const ControllerChannel & )

delete

edm::shared_memory::ControllerChannel::ControllerChannel ( ControllerChannel && )

delete

Member Function Documentation

BufferInfo * ControllerChannel::bufferInfo	(	const char *	iWhich,
		boost::interprocess::managed_shared_memory &	mem
	)

staticprivate

Definition at line 137 of file ControllerChannel.cc.

References findQualityFiles::v.

Referenced by ControllerChannel(), and initCheckWorkerStatus().

                                                                                         {
   mem.destroy<BufferInfo>(iWhich);
   BufferInfo* v = mem.construct<BufferInfo>(iWhich)();
   return v;
 }

bool ControllerChannel::continueWait ( boost::interprocess::scoped_lock< boost::interprocess::named_mutex > & lock )

private

Definition at line 116 of file ControllerChannel.cc.

References cndToMain_, initCheckWorkerStatus(), maxWaitInSeconds_, seconds(), and transitionID_.

Referenced by doTransitionWithRetry(), and initCheckWorkerStatus().

                                                                    {
   //std::cout << id_ << " waiting" << std::endl;
   using namespace boost::posix_time;
   //NOTE: value of *transitionID_ can not have been changed by the worker since call to wait()
   //  as we've had the lock since the end of that call.
   auto workerStatus = initCheckWorkerStatus(transitionID_);
   if (not cndToMain_.timed_wait(lock, microsec_clock::universal_time() + seconds(maxWaitInSeconds_)) and
       not workerStatus.workerFinished()) {
     //std::cout << id_ << " waiting FAILED" << std::endl;
     return false;
   }
   return true;
 }

template<typename F >

bool edm::shared_memory::ControllerChannel::doTransition	(	F &&	iF,
		edm::Transition	iTrans,
		unsigned long long	iTransitionID
	)

inline

Definition at line 102 of file ControllerChannel.h.

References CommonMethods::lock(), mutex_, and wait().

                                                                                       {
       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;
     }

template<typename F , typename FRETRY >

bool edm::shared_memory::ControllerChannel::doTransitionWithRetry	(	F &&	iF,
		FRETRY &&	iRetry,
		edm::Transition	iTrans,
		unsigned long long	iTransitionID
	)

inline

Definition at line 116 of file ControllerChannel.h.

References continueWait(), CommonMethods::lock(), mutex_, and wait().

                                                                                                                 {
       using namespace boost::interprocess;
 
       //std::cout << id_ << " taking from lock" << std::endl;
       scoped_lock<named_mutex> lock(mutex_);
       if (not wait(lock, iTrans, iTransitionID)) {
         if (not iRetry()) {
           return false;
         }
         bool shouldContinue = true;
         do {
           using namespace boost::posix_time;
           if (not continueWait(lock)) {
             if (not iRetry()) {
               return false;
             }
           } else {
             shouldContinue = false;
           }
         } while (shouldContinue);
       }
       //std::cout <<id_<<"running doTranstion command"<<std::endl;
       iF();
       return true;
     }

BufferInfo* edm::shared_memory::ControllerChannel::fromWorkerBufferInfo ( )

inline

This can be used with ReadBuffer to keep Controller and Worker in sync.

Definition at line 145 of file ControllerChannel.h.

References fromWorkerBufferInfo_.

145 { return fromWorkerBufferInfo_; }

edm::shared_memory::ControllerChannel::fromWorkerBufferInfo_

BufferInfo * fromWorkerBufferInfo_

Definition: ControllerChannel.h:193

CheckWorkerStatus edm::shared_memory::ControllerChannel::initCheckWorkerStatus ( unsigned long long * iPtr ) const

inlineprivate

Definition at line 173 of file ControllerChannel.h.

References bufferInfo(), continueWait(), CommonMethods::lock(), mem, AlCaHLTBitMon_QueryRunRegistry::string, uniqueName(), and wait().

Referenced by continueWait(), and uniqueName().

                                                                                                    {
       return {*iPtr, iPtr};
     }

unsigned int edm::shared_memory::ControllerChannel::maxWaitInSeconds ( ) const

inline

Definition at line 163 of file ControllerChannel.h.

References maxWaitInSeconds_.

163 { return maxWaitInSeconds_; }

edm::shared_memory::ControllerChannel::maxWaitInSeconds_

unsigned int maxWaitInSeconds_

Definition: ControllerChannel.h:189

const ControllerChannel& edm::shared_memory::ControllerChannel::operator= ( const ControllerChannel & )

delete

const ControllerChannel& edm::shared_memory::ControllerChannel::operator= ( ControllerChannel && )

delete

template<typename F >

void edm::shared_memory::ControllerChannel::setupWorker ( F && iF )

inline

setupWorker must be called only once and done before any calls to doTransition. The functor iF should setup values associated with shared memory use, such as manipulating the value from toWorkerBufferInfo(). The call to setupWorker proper synchronizes the Controller and Worker processes.

Definition at line 56 of file ControllerChannel.h.

References Exception, CommonMethods::lock(), maxWaitInSeconds_, mutex_, stop_, and wait().

                              {
       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 wait(lock)) {
         //std::cout << id_ << " FAILED waiting for external process" << std::endl;
         *stop_ = true;
         throw cms::Exception("ExternalFailed")
             << "Failed waiting for external process while setting up the process. Timed out after " << maxWaitInSeconds_
             << " seconds.";
       } else {
         //std::cout << id_ << " done waiting for external process" << std::endl;
       }
     }

template<typename F , typename FRETRY >

void edm::shared_memory::ControllerChannel::setupWorkerWithRetry	(	F &&	iF,
		FRETRY &&	iRetry
	)

inline

setupWorkerWithRetry works just like setupWorker except it gives a way to continue waiting. The functor iRetry should return true if, after a timeout, the code should continue to wait.

Definition at line 77 of file ControllerChannel.h.

References Exception, CommonMethods::lock(), maxWaitInSeconds_, mutex_, stop_, and wait().

                                                        {
       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;
       bool shouldContinue = true;
       long long int retryCount = 0;
       do {
         if (not wait(lock)) {
           if (not iRetry()) {
             *stop_ = true;
             throw cms::Exception("ExternalFailed")
                 << "Failed waiting for external process while setting up the process. Timed out after "
                 << maxWaitInSeconds_ << " seconds with " << retryCount << " retries.";
           }
           //std::cerr<<"retrying\n";
           ++retryCount;
         } else {
           shouldContinue = false;
         }
       } while (shouldContinue);
     }

std::string const& edm::shared_memory::ControllerChannel::sharedMemoryName ( ) const

inline

Definition at line 157 of file ControllerChannel.h.

References smName_.

157 { return smName_; }

edm::shared_memory::ControllerChannel::smName_

std::string smName_

Definition: ControllerChannel.h:190

bool edm::shared_memory::ControllerChannel::shouldKeepEvent ( ) const

inline

Definition at line 161 of file ControllerChannel.h.

References keepEvent_.

161 { return *keepEvent_; }

edm::shared_memory::ControllerChannel::keepEvent_

bool * keepEvent_

Definition: ControllerChannel.h:203

void edm::shared_memory::ControllerChannel::stopWorker ( )

inline

Definition at line 147 of file ControllerChannel.h.

References cndFromMain_, CommonMethods::lock(), mutex_, and stop_.

                       {
       //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();
     }

BufferInfo* edm::shared_memory::ControllerChannel::toWorkerBufferInfo ( )

inline

This can be used with WriteBuffer to keep Controller and Worker in sync.

Definition at line 143 of file ControllerChannel.h.

References toWorkerBufferInfo_.

143 { return toWorkerBufferInfo_; }

edm::shared_memory::ControllerChannel::toWorkerBufferInfo_

BufferInfo * toWorkerBufferInfo_

Definition: ControllerChannel.h:192

std::string edm::shared_memory::ControllerChannel::uniqueID ( ) const

inline

Definition at line 158 of file ControllerChannel.h.

References uniqueName().

158 { return uniqueName(""); }

edm::shared_memory::ControllerChannel::uniqueName

std::string uniqueName(std::string iBase) const

Definition: ControllerChannel.cc:76

std::string ControllerChannel::uniqueName ( std::string iBase ) const

private

Definition at line 76 of file ControllerChannel.cc.

References cndFromMain_, cndToMain_, id_, initCheckWorkerStatus(), CommonMethods::lock(), maxWaitInSeconds_, sysUtil::pid, seconds(), transitionID_, transitionType_, and wait().

Referenced by ControllerChannel(), initCheckWorkerStatus(), uniqueID(), and ~ControllerChannel().

                                                              {
   auto pid = getpid();
   iBase += std::to_string(pid);
   iBase += "_";
   iBase += std::to_string(id_);
 
   return iBase;
 }

bool edm::shared_memory::ControllerChannel::wait	(	boost::interprocess::scoped_lock< boost::interprocess::named_mutex > &	lock,
		edm::Transition	iTrans,
		unsigned long long	iTransID
	)