src/limited/OutputModuleBase.cc

0001 // -*- C++ -*-
0002 //
0003 // Package:     FWCore/Framework
0004 // Class  :     OutputModuleBase
0005 //
0006 // Implementation:
0007 //     [Notes on implementation]
0008 //
0009 //
0010
0011 // system include files
0012 #include <cassert>
0013
0014 // user include files
0015 #include "FWCore/Framework/interface/limited/OutputModuleBase.h"
0016 #include "FWCore/Framework/interface/PreallocationConfiguration.h"
0017 #include "FWCore/ParameterSet/interface/ParameterSetDescription.h"
0018
0019 namespace edm {
0020   namespace limited {
0021
0022     // -------------------------------------------------------
0023     OutputModuleBase::OutputModuleBase(ParameterSet const& pset)
0024         : core::OutputModuleCore(pset), queue_(pset.getUntrackedParameter<unsigned int>("concurrencyLimit")) {}
0025
0026     void OutputModuleBase::doPreallocate(PreallocationConfiguration const& iPC) {
0027       auto nstreams = iPC.numberOfStreams();
0028
0029       preallocStreams(nstreams);
0030       core::OutputModuleCore::doPreallocate_(iPC);
0031       preallocate(iPC);
0032     }
0033
0034     void OutputModuleBase::doBeginJob() { core::OutputModuleCore::doBeginJob_(); }
0035
0036     bool OutputModuleBase::doEvent(EventTransitionInfo const& info,
0037                                    ActivityRegistry* act,
0038                                    ModuleCallingContext const* mcc) {
0039       { core::OutputModuleCore::doEvent_(info, act, mcc); }
0040
0041       auto remainingEvents = remainingEvents_.load();
0042       bool keepTrying = remainingEvents > 0;
0043       while (keepTrying) {
0044         auto newValue = remainingEvents - 1;
0045         keepTrying = !remainingEvents_.compare_exchange_strong(remainingEvents, newValue);
0046         if (keepTrying) {
0047           // the exchange failed because the value was changed by another thread.
0048           // remainingEvents was changed to be the new value of remainingEvents_;
0049           keepTrying = remainingEvents > 0;
0050         }
0051       }
0052       return true;
0053     }
0054
0055     void OutputModuleBase::fillDescription(ParameterSetDescription& desc,
0056                                            std::vector<std::string> const& defaultOutputCommands) {
0057       core::OutputModuleCore::fillDescription(desc, defaultOutputCommands);
0058       desc.addUntracked<unsigned int>("concurrencyLimit", 1);
0059     }
0060   }  // namespace limited
0061 }  // namespace edm