Public Member Functions
void	asyncWork (edm::StreamID id, edm::WaitingTaskWithArenaHolder iTask, long initTime, long workTime, long finishTime)

	SleepingServer (edm::ParameterSet const &iPS, edm::ActivityRegistry &iAR)

Private Member Functions
bool	readyToDoSomething ()

void	threadWork ()

Private Attributes
std::atomic< unsigned int >	activeStreams_ {0}

std::condition_variable	condition_

std::mutex	mutex_

const unsigned int	nWaitingEvents_

std::unique_ptr< std::thread >	serverThread_

std::atomic< bool >	stopProcessing_ {false}

std::vector< int >	waitingStreams_

std::vector< edm::WaitingTaskWithArenaHolder >	waitingTaskPerStream_

std::vector< std::array< long, 3 > >	waitTimesPerStream_

Detailed Description

Definition at line 185 of file TimeStudyModules.cc.

Constructor & Destructor Documentation

◆ SleepingServer()

timestudy::SleepingServer::SleepingServer	(	edm::ParameterSet const &	iPS,
		edm::ActivityRegistry &	iAR
	)

inline

Definition at line 187 of file TimeStudyModules.cc.

         : nWaitingEvents_(iPS.getUntrackedParameter<unsigned int>("nWaitingEvents")) {
       iAR.watchPreallocate([this](edm::service::SystemBounds const& iBounds) {
         auto const nStreams = iBounds.maxNumberOfStreams();
         waitingStreams_.reserve(nStreams);
         waitTimesPerStream_.resize(nStreams);
         waitingTaskPerStream_.resize(nStreams);
       });
  
       iAR.watchPreEndJob([this]() {
         stopProcessing_ = true;
         condition_.notify_one();
         serverThread_->join();
       });
       iAR.watchPreStreamBeginLumi([this](edm::StreamContext const&) { ++activeStreams_; });
       iAR.watchPreStreamEndLumi([this](edm::StreamContext const&) {
         --activeStreams_;
         condition_.notify_one();
       });
  
       serverThread_ = std::make_unique<std::thread>([this]() { threadWork(); });
     }

References activeStreams_, condition_, edm::service::SystemBounds::maxNumberOfStreams(), runTheMatrix::nStreams, serverThread_, stopProcessing_, threadWork(), waitingStreams_, waitingTaskPerStream_, waitTimesPerStream_, edm::ActivityRegistry::watchPreallocate(), edm::ActivityRegistry::watchPreEndJob(), edm::ActivityRegistry::watchPreStreamBeginLumi(), and edm::ActivityRegistry::watchPreStreamEndLumi().

Member Function Documentation

◆ asyncWork()

void timestudy::SleepingServer::asyncWork	(	edm::StreamID	id,
		edm::WaitingTaskWithArenaHolder	iTask,
		long	initTime,
		long	workTime,
		long	finishTime
	)

inline

Definition at line 210 of file TimeStudyModules.cc.

                                                                                                             {
       waitTimesPerStream_[id.value()] = {{initTime, workTime, finishTime}};
       waitingTaskPerStream_[id.value()] = std::move(iTask);
       {
         std::lock_guard<std::mutex> lk{mutex_};
         waitingStreams_.push_back(id.value());
       }
       condition_.notify_one();
     }

References condition_, eostools::move(), mutex_, relativeConstraints::value, waitingStreams_, waitingTaskPerStream_, and waitTimesPerStream_.

◆ readyToDoSomething()

bool timestudy::SleepingServer::readyToDoSomething ( )

inlineprivate

Definition at line 222 of file TimeStudyModules.cc.

                               {
       if (stopProcessing_) {
         return true;
       }
       if (waitingStreams_.size() >= nWaitingEvents_) {
         return true;
       }
       //every running stream is now waiting
       return waitingStreams_.size() == activeStreams_;
     }

References activeStreams_, nWaitingEvents_, stopProcessing_, and waitingStreams_.

Referenced by threadWork().

◆ threadWork()

void timestudy::SleepingServer::threadWork ( )

inlineprivate

Definition at line 233 of file TimeStudyModules.cc.

                       {
       while (not stopProcessing_.load()) {
         std::vector<int> streamsToProcess;
         {
           std::unique_lock<std::mutex> lk(mutex_);
           condition_.wait(lk, [this]() { return readyToDoSomething(); });
           swap(streamsToProcess, waitingStreams_);
         }
         if (stopProcessing_) {
           break;
         }
         long longestTime = 0;
         //simulate filling the external device
         for (auto i : streamsToProcess) {
           auto const& v = waitTimesPerStream_[i];
           if (v[1] > longestTime) {
             longestTime = v[1];
           }
           usleep(v[0]);
         }
         //simulate running external device
         usleep(longestTime);
  
         //simulate copying data back
         for (auto i : streamsToProcess) {
           auto const& v = waitTimesPerStream_[i];
           usleep(v[2]);
           waitingTaskPerStream_[i].doneWaiting(std::exception_ptr());
         }
       }
       waitingTaskPerStream_.clear();
     }