timestudy::SleepingServer Class Reference

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 200 of file TimeStudyModules.cc.

Constructor & Destructor Documentation

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

inline

Definition at line 202 of file TimeStudyModules.cc.

References edm::service::SystemBounds::maxNumberOfStreams(), edm::ActivityRegistry::watchPreallocate(), edm::ActivityRegistry::watchPreEndJob(), edm::ActivityRegistry::watchPreStreamBeginLumi(), and edm::ActivityRegistry::watchPreStreamEndLumi().

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

Member Function Documentation

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

inline

Definition at line 228 of file TimeStudyModules.cc.

References eostools::move(), and relativeConstraints::value.

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

bool timestudy::SleepingServer::readyToDoSomething ( )

inlineprivate

Definition at line 239 of file TimeStudyModules.cc.

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

void timestudy::SleepingServer::threadWork ( )

inlineprivate

Definition at line 250 of file TimeStudyModules.cc.

References mps_fire::i, edm::swap(), and findQualityFiles::v.

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

Member Data Documentation

std::atomic<unsigned int> timestudy::SleepingServer::activeStreams_ {0}

private

Definition at line 291 of file TimeStudyModules.cc.

std::condition_variable timestudy::SleepingServer::condition_

private

Definition at line 290 of file TimeStudyModules.cc.

std::mutex timestudy::SleepingServer::mutex_

private

Definition at line 289 of file TimeStudyModules.cc.

const unsigned int timestudy::SleepingServer::nWaitingEvents_

private

Definition at line 284 of file TimeStudyModules.cc.

std::unique_ptr<std::thread> timestudy::SleepingServer::serverThread_

private

Definition at line 285 of file TimeStudyModules.cc.

std::atomic<bool> timestudy::SleepingServer::stopProcessing_ {false}

private

Definition at line 292 of file TimeStudyModules.cc.

std::vector<int> timestudy::SleepingServer::waitingStreams_

private

Definition at line 286 of file TimeStudyModules.cc.

std::vector<edm::WaitingTaskWithArenaHolder> timestudy::SleepingServer::waitingTaskPerStream_

private

Definition at line 288 of file TimeStudyModules.cc.

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

private

Definition at line 287 of file TimeStudyModules.cc.