AsyncAccessorManager.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 "../VersionNumber.h"
#include "AsyncNDRegisterAccessor.h"
 
#include <utility>
 
namespace ChimeraTK::async {
 
  struct AsyncVariable {
    virtual ~AsyncVariable() = default;
 
    virtual void send() = 0;
 
    virtual void sendException(std::exception_ptr e) = 0;
 
    virtual unsigned int getNumberOfChannels() = 0;
    virtual unsigned int getNumberOfSamples() = 0;
    virtual const std::string& getUnit() = 0;
    virtual const std::string& getDescription() = 0;
 
    virtual void fillSendBuffer() = 0;
  };
 
  struct AccessorInstanceDescriptor {
    RegisterPath name;
    std::type_index type;
    size_t numberOfWords;
    size_t wordOffsetInRegister;
    AccessModeFlags flags;
  };
 
  class AsyncAccessorManager : public boost::enable_shared_from_this<AsyncAccessorManager> {
   public:
    explicit AsyncAccessorManager(boost::shared_ptr<DeviceBackend> backend, boost::shared_ptr<Domain> asyncDomain)
    : _backend(std::move(backend)), _asyncDomain(std::move(asyncDomain)) {}
    virtual ~AsyncAccessorManager() = default;
 
    template<typename UserType>
    boost::shared_ptr<AsyncNDRegisterAccessor<UserType>> subscribe(
        RegisterPath name, size_t numberOfWords, size_t wordOffsetInRegister, AccessModeFlags flags);
 
    void unsubscribe(TransferElementID id);
 
    void sendException(const std::exception_ptr& e);
 
   protected:
    /*** Each implementation must provide a function to create specific AsyncVariables.
     *   When the variable is returned, async accessor is not set yet. This would leave the whole creation
     *   of the async accessor to the backend, would mean a lot of code duplication. It also cannot be
     *   retrieved from the AsyncVariable as this only contains a weak pointer.
     *   If the isActive flag is set, the _sendBuffer must already contain the initial value. The variable itself
     *   is not activated yet as the async accessor is still not set.
     */
    DEFINE_VIRTUAL_FUNCTION_TEMPLATE_VTABLE(
        createAsyncVariable, std::unique_ptr<AsyncVariable>(AccessorInstanceDescriptor const&));
 
    std::map<TransferElementID, std::unique_ptr<AsyncVariable>> _asyncVariables;
 
    boost::shared_ptr<DeviceBackend> _backend;
    boost::shared_ptr<Domain> _asyncDomain;
 
    virtual void asyncVariableMapChanged(TransferElementID) {}
 
    thread_local static AsyncAccessorManager* _isHoldingDomainLock;
    std::list<TransferElementID> _delayedUnsubscriptions;
 
    void unsubscribeImpl(TransferElementID id);
  };
 
  template<typename SourceType>
  class SourceTypedAsyncAccessorManager : public AsyncAccessorManager {
   public:
    using AsyncAccessorManager::AsyncAccessorManager;
 
    VersionNumber distribute(SourceType, VersionNumber version);
 
    virtual bool prepareIntermediateBuffers() { return true; };
 
   protected:
    SourceType _sourceBuffer;
    VersionNumber _version{nullptr};
  };
 
  template<typename UserType>
  struct AsyncVariableImpl : public AsyncVariable {
    AsyncVariableImpl(size_t nChannels, size_t nElements);
 
    void send() final;
    void sendException(std::exception_ptr e) final;
    unsigned int getNumberOfChannels() override;
    unsigned int getNumberOfSamples() override;
 
    typename NDRegisterAccessor<UserType>::Buffer _sendBuffer;
 
   private:
    // This weak pointer is private so the user cannot bypass correct locking and nullptr-checking.
    boost::weak_ptr<AsyncNDRegisterAccessor<UserType>> _asyncAccessor;
    friend AsyncAccessorManager; // is allowed to set the _asyncAccessor
 
    VersionNumber _lastSentVersion{nullptr};
  };
 
  /********************************************************************************************************************/
  // Implementations
  /********************************************************************************************************************/
 
  /********************************************************************************************************************/
  template<typename UserType>
  void AsyncVariableImpl<UserType>::sendException(std::exception_ptr e) {
    auto subscriber = _asyncAccessor.lock();
    if(subscriber.get() != nullptr) { // Solves possible race condition: The subscriber is being destructed.
      subscriber->sendException(e);
    }
  }
 
  /********************************************************************************************************************/
  template<typename UserType>
  unsigned int AsyncVariableImpl<UserType>::getNumberOfChannels() {
    return _sendBuffer.value.size();
  }
 
  /********************************************************************************************************************/
  template<typename UserType>
  unsigned int AsyncVariableImpl<UserType>::getNumberOfSamples() {
    if(_sendBuffer.value.size() > 0) {
      return _sendBuffer.value[0].size();
    }
    return 0;
  }
 
  /********************************************************************************************************************/
  template<typename UserType>
  AsyncVariableImpl<UserType>::AsyncVariableImpl(size_t nChannels, size_t nElements)
  : _sendBuffer(nChannels, nElements) {}
 
  /********************************************************************************************************************/
  template<typename UserType>
  void AsyncVariableImpl<UserType>::send() {
    auto subscriber = _asyncAccessor.lock();
 
    if(_lastSentVersion > _sendBuffer.versionNumber) {
      throw ChimeraTK::logic_error(std::format("Trying to send decreased version {} < {} on AsyncVariable {}.",
          _sendBuffer.versionNumber, _lastSentVersion, subscriber->getName()));
    }
    _lastSentVersion = _sendBuffer.versionNumber;
 
    if(subscriber.get() != nullptr) { // Solves possible race condition: The subscriber is being destructed.
      subscriber->sendDestructively(_sendBuffer);
    }
  }
 
  /********************************************************************************************************************/
  template<typename UserType>
  boost::shared_ptr<AsyncNDRegisterAccessor<UserType>> AsyncAccessorManager::subscribe(
      RegisterPath name, size_t numberOfWords, size_t wordOffsetInRegister, AccessModeFlags flags) {
    AccessorInstanceDescriptor descriptor{name, typeid(UserType), numberOfWords, wordOffsetInRegister, flags};
    auto untypedAsyncVariable = CALL_VIRTUAL_FUNCTION_TEMPLATE(createAsyncVariable, UserType, descriptor);
 
    auto asyncVariable = dynamic_cast<AsyncVariableImpl<UserType>*>(untypedAsyncVariable.get());
    // we take all the information we need for the async accessor from AsyncVariable because we cannot use the catalogue
    // here
    auto newSubscriber = boost::make_shared<AsyncNDRegisterAccessor<UserType>>(_backend, shared_from_this(),
        _asyncDomain, name, asyncVariable->getNumberOfChannels(), asyncVariable->getNumberOfSamples(), flags,
        asyncVariable->getUnit(), asyncVariable->getDescription());
    // Set the exception backend here. It might be that the accessor is already activated during subscription, and the
    // backend should be set at that point
    newSubscriber->setExceptionBackend(_backend);
 
    asyncVariable->_asyncAccessor = newSubscriber;
    // Now that the AsyncVariable is complete we can finally activate it.
    if(_asyncDomain->unsafeGetIsActive()) {
      asyncVariable->fillSendBuffer();
      asyncVariable->send();
    }
 
    _asyncVariables[newSubscriber->getId()] = std::move(untypedAsyncVariable);
    asyncVariableMapChanged(newSubscriber->getId());
 
    return newSubscriber;
  }
  /********************************************************************************************************************/
  template<typename SourceType>
  VersionNumber SourceTypedAsyncAccessorManager<SourceType>::distribute(SourceType data, VersionNumber version) {
    if(!_asyncDomain->unsafeGetIsActive()) {
      return version;
    }
 
    _sourceBuffer = data;
    _version = version;
 
    if(prepareIntermediateBuffers()) {
      assert(_delayedUnsubscriptions.empty());
      for(auto& var : _asyncVariables) {
        var.second->fillSendBuffer();
        _isHoldingDomainLock = this;
        var.second->send(); // function from  the AsyncVariable base class
        _isHoldingDomainLock = nullptr;
      }
      for(auto id : _delayedUnsubscriptions) {
        unsubscribeImpl(id);
      }
      _delayedUnsubscriptions.clear();
    }
 
    return _version;
  }
 
} // namespace ChimeraTK::async