DomainsContainer.h

Go to the documentation of this file.

// SPDX-FileCopyrightText: Deutsches Elektronen-Synchrotron DESY, MSK, ChimeraTK Project <chimeratk-support@desy.de>
// SPDX-License-Identifier: LGPL-3.0-or-later
#pragma once
 
#include "DomainImpl.h"
 
#include <ChimeraTK/cppext/future_queue.hpp>
 
#include <iostream>
#include <map>
#include <sstream>
#include <thread>
 
namespace ChimeraTK::async {
 
  class DomainsContainer {
   public:
    ~DomainsContainer();
 
    void sendExceptions(const std::string& exceptionMessage);
 
    bool isSendingExceptions() { return _isSendingExceptions; }
 
    template<typename BackendType, typename BackendSpecificDataType, typename UserDataType>
    boost::shared_ptr<AsyncNDRegisterAccessor<UserDataType>> subscribe(boost::shared_ptr<BackendType> backend,
        size_t domainId, bool activate, RegisterPath name, size_t numberOfWords, size_t wordOffsetInRegister,
        AccessModeFlags flags);
 
    boost::shared_ptr<Domain> getDomain(size_t key);
 
    void forEach(const std::function<void(size_t, boost::shared_ptr<Domain>&)>& executeMe);
 
   protected:
    std::atomic_bool _isSendingExceptions{false};
 
    void distributeExceptions();
 
    cppext::future_queue<std::string> _startExceptionDistribution{2};
    std::thread _distributorThread;
    std::mutex _threadCreationMutex;
    std::atomic_bool _threadIsRunning{false}; // Cache whether the thread is running so we don't have to lock the mutex
                                              // each time to be able to ask the thread itself.
    class StopThread : public std::exception {};
 
    std::mutex _domainsMutex;
    std::map<size_t, boost::weak_ptr<Domain>> _domains;
  };
 
  /********************************************************************************************************************/
 
  template<typename BackendType, typename BackendSpecificDataType, typename UserDataType>
  boost::shared_ptr<AsyncNDRegisterAccessor<UserDataType>> DomainsContainer::subscribe(
      boost::shared_ptr<BackendType> backend, size_t domainId, bool activate, RegisterPath name, size_t numberOfWords,
      size_t wordOffsetInRegister, AccessModeFlags flags) {
    std::lock_guard<std::mutex> domainsLock(_domainsMutex);
 
    auto domain = _domains[domainId].lock();
    boost::shared_ptr<DomainImpl<BackendSpecificDataType>> domainImpl;
 
    bool domainCreated{false};
    if(domain) {
      domainImpl = boost::dynamic_pointer_cast<DomainImpl<BackendSpecificDataType>>(domain);
      assert(domainImpl);
    }
    else {
      // The domain does not exist, create it
      domainCreated = true;
      domainImpl = boost::make_shared<DomainImpl<BackendSpecificDataType>>(backend, domainId);
      // start the thread if not already running
      { // thread lock scope
        auto threadCreationLock = std::lock_guard(_threadCreationMutex);
 
        if(!_distributorThread.joinable()) {
          _distributorThread = std::thread([&] { distributeExceptions(); });
          _threadIsRunning = true;
        }
      } // end of thread lock scope
      _domains[domainId] = domainImpl;
    }
 
    auto newSubscriber = domainImpl->template subscribe<UserDataType>(name, numberOfWords, wordOffsetInRegister, flags);
 
    // Only activate a newly created domain after the subscription is done. This allows to distribute the intial value
    // from here, and not from inside the subscription.
    if(domainCreated && activate) {
      auto subscriptionDone = backend->activateSubscription(domainId, domainImpl);
      // wait until the backend reports that the subscription is ready before proceeding with the creation of the
      // accessor, which will poll the initial value.
      subscriptionDone.wait();
      auto [value, version] = backend->template getAsyncDomainInitialValue<BackendSpecificDataType>(domainId);
      domainImpl->activate(value, version);
    }
 
    return newSubscriber;
  }
 
} // namespace ChimeraTK::async