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

Constructor & Destructor Documentation

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

inline

Definition at line 187 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 210 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 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_;
    }

void timestudy::SleepingServer::threadWork ( )

inlineprivate

Definition at line 233 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 272 of file TimeStudyModules.cc.

std::condition_variable timestudy::SleepingServer::condition_

private

Definition at line 271 of file TimeStudyModules.cc.

Referenced by upgradeWorkflowComponents.UpgradeWorkflowTracking::condition().

std::mutex timestudy::SleepingServer::mutex_

private

Definition at line 270 of file TimeStudyModules.cc.

const unsigned int timestudy::SleepingServer::nWaitingEvents_

private

Definition at line 265 of file TimeStudyModules.cc.

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

private

Definition at line 266 of file TimeStudyModules.cc.

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

private

Definition at line 273 of file TimeStudyModules.cc.

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

private

Definition at line 267 of file TimeStudyModules.cc.

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

private

Definition at line 269 of file TimeStudyModules.cc.

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

private

Definition at line 268 of file TimeStudyModules.cc.