Framework/src/SharedResourcesRegistry.cc

0001 // -*- C++ -*-
0002 //
0003 // Package:     FWCore/Framework
0004 // Class  :     SharedResourcesRegistry
0005 //
0006 // Implementation:
0007 //     [Notes on implementation]
0008 //
0009 // Original Author:  Chris Jones
0010 //         Created:  Sun, 06 Oct 2013 15:48:50 GMT
0011 //
0012
0013 #include <algorithm>
0014 #include <cassert>
0015
0016 #include "FWCore/Framework/interface/SharedResourcesRegistry.h"
0017 #include "FWCore/Framework/interface/SharedResourcesAcquirer.h"
0018
0019 namespace edm {
0020
0021   SharedResourcesRegistry* SharedResourcesRegistry::instance() {
0022     static SharedResourcesRegistry s_instance;
0023     return &s_instance;
0024   }
0025
0026   SharedResourcesRegistry::SharedResourcesRegistry() {}
0027
0028   void SharedResourcesRegistry::registerSharedResource(const std::string& resourceName) {
0029     auto& queueAndCounter = resourceMap_[resourceName];
0030
0031     // count the number of times the resource was registered
0032     ++queueAndCounter.second;
0033
0034     // If registering a resource the second time
0035     // we know we will need the queue so go ahead and create it.
0036     if (queueAndCounter.second == 2) {
0037       queueAndCounter.first = std::make_shared<SerialTaskQueue>();
0038     }
0039   }
0040
0041   std::pair<SharedResourcesAcquirer, std::shared_ptr<std::recursive_mutex>>
0042   SharedResourcesRegistry::createAcquirerForSourceDelayedReader() {
0043     if (not resourceForDelayedReader_) {
0044       resourceForDelayedReader_ =
0045           std::make_shared<std::recursive_mutex>();  // propagate_const<T> has no reset() function
0046       queueForDelayedReader_ = std::make_shared<SerialTaskQueue>();
0047     }
0048
0049     std::vector<std::shared_ptr<SerialTaskQueue>> queues = {get_underlying(queueForDelayedReader_)};
0050     return std::make_pair(SharedResourcesAcquirer(std::move(queues)), get_underlying(resourceForDelayedReader_));
0051   }
0052
0053   SharedResourcesAcquirer SharedResourcesRegistry::createAcquirer(std::vector<std::string> const& resourceNames) const {
0054     // The acquirer will acquire the shared resources declared by a module
0055     // so that only it can use those resources while it runs. The other
0056     // modules using the same resource will not be run until the module
0057     // that acquired the resources completes its task.
0058
0059     // Sort by how often used and then by name
0060     // Consistent sorting avoids deadlocks and this particular order optimizes performance
0061     std::map<std::pair<unsigned int, std::string>, std::shared_ptr<SerialTaskQueue>> sortedResources;
0062
0063     for (auto const& name : resourceNames) {
0064       auto resource = resourceMap_.find(name);
0065       assert(resource != resourceMap_.end());
0066       // If only one module wants it, it really isn't shared
0067       if (resource->second.second > 1) {
0068         sortedResources.insert(
0069             std::make_pair(std::make_pair(resource->second.second, resource->first), resource->second.first));
0070       }
0071     }
0072
0073     std::vector<std::shared_ptr<SerialTaskQueue>> queues;
0074     queues.reserve(sortedResources.size());
0075     for (auto const& resource : sortedResources) {
0076       queues.push_back(resource.second);
0077     }
0078     if (queues.empty()) {
0079       //Calling code is depending on there being at least one shared queue
0080       queues.reserve(1);
0081       queues.push_back(std::make_shared<SerialTaskQueue>());
0082     }
0083
0084     return SharedResourcesAcquirer(std::move(queues));
0085   }
0086 }  // namespace edm