TestableMode.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 "Logger.h"
#include "VariableNetworkNode.h"
 
#include <ChimeraTK/ControlSystemAdapter/BidirectionalProcessArray.h>
#include <ChimeraTK/TransferElement.h>
 
#include <boost/shared_ptr.hpp>
#include <boost/thread.hpp>
 
#include <atomic>
#include <cstddef>
#include <map>
#include <shared_mutex>
 
namespace ChimeraTK {
  class ConnectionMaker;
  class DeviceManager;
  class TriggerFanOut;
  class TestFacility;
} /* namespace ChimeraTK */
 
namespace ChimeraTK::detail {
  struct TestableMode {
    void lock(const std::string& name, bool shared);
 
    void unlock(const std::string& name);
 
    [[nodiscard]] bool testLock() const;
 
    [[nodiscard]] bool canStep() const { return _counter != 0; }
 
    void step(bool waitForDeviceInitialisation);
 
    void setThreadName(const std::string& name);
 
    void enable();
 
    [[nodiscard]] bool isEnabled() const { return _enabled; }
 
    static size_t getNextVariableId() {
      static size_t nextId{0};
      return ++nextId;
    }
 
    enum class DecoratorType { READ, WRITE };
 
    template<typename T>
    boost::shared_ptr<NDRegisterAccessor<T>> decorate(boost::shared_ptr<NDRegisterAccessor<T>> other,
        DecoratorType direction, const std::string& name = {}, size_t varId = 0);
 
    template<typename T>
    using AccessorPair = std::pair<boost::shared_ptr<NDRegisterAccessor<T>>, boost::shared_ptr<NDRegisterAccessor<T>>>;
 
    template<typename T>
    AccessorPair<T> decorate(
        AccessorPair<T> other, const VariableNetworkNode& producer, const VariableNetworkNode& consumer);
 
    template<typename UserType>
    class AccessorDecorator : public ChimeraTK::NDRegisterAccessorDecorator<UserType> {
     public:
      AccessorDecorator(detail::TestableMode& testableMode,
          boost::shared_ptr<ChimeraTK::NDRegisterAccessor<UserType>> accessor, bool handleRead, bool handleWrite,
          size_t variableIdRead, size_t variableIdWrite);
 
      // FIXME: https://redmine.msktools.desy.de/issues/12242
      // NOLINTNEXTLINE(google-default-arguments)
      bool doWriteTransfer(ChimeraTK::VersionNumber versionNumber = {}) override;
 
      // FIXME: https://redmine.msktools.desy.de/issues/12242
      // NOLINTNEXTLINE(google-default-arguments)
      bool doWriteTransferDestructively(ChimeraTK::VersionNumber versionNumber = {}) override;
 
      void doReadTransferSynchronously() override { _target->readTransfer(); }
 
      void releaseLock();
 
      void doPreRead(TransferType type) override;
 
      void obtainLockAndDecrementCounter(bool hasNewData);
 
      void decrementCounter();
 
      void doPostRead(TransferType type, bool hasNewData) override;
 
      [[nodiscard]] boost::shared_ptr<NDRegisterAccessor<UserType>> decorateDeepInside(
          [[maybe_unused]] std::function<boost::shared_ptr<NDRegisterAccessor<UserType>>(
              const boost::shared_ptr<NDRegisterAccessor<UserType>>&)> factory) override {
        // By returning nullptr, we forbid that DataConsistencyDecorator is put inside of this decorator.
        // It would mess up our data updates counting scheme.
        return {};
      }
 
     protected:
      using ChimeraTK::NDRegisterAccessor<UserType>::buffer_2D;
      using ChimeraTK::NDRegisterAccessorDecorator<UserType>::_target;
 
      bool _handleRead, _handleWrite;
      size_t _variableIdRead, _variableIdWrite;
      TestableMode& _testableMode;
 
      bool accountForWriteOperation(const std::function<bool(void)>& writeOperation);
    };
 
   private:
    friend class ChimeraTK::DeviceManager;
    friend class ChimeraTK::TriggerFanOut;
    friend class ChimeraTK::TestFacility;
 
    std::atomic<size_t> _counter{0};
 
    bool _enabled{false};
 
    std::atomic<size_t> _deviceInitialisationCounter{0};
 
    struct VariableDescriptor {
      std::string name;
 
      boost::shared_ptr<TransferElement> processVariable;
 
      std::atomic<size_t> counter{0};
    };
 
    static std::shared_timed_mutex _mutex;
 
    static std::shared_mutex _mutex2;
 
    std::map<size_t, VariableDescriptor> _variables;
 
    class LastMutexOwner {
     public:
      LastMutexOwner& operator=(const boost::thread::id& id);
      // NOLINTNEXTLINE(google-explicit-constructor)
      operator boost::thread::id();
 
     private:
      boost::thread::id _lastMutexOwner;
      std::mutex _mxLastMutexOwner;
    } _lastMutexOwner;
 
    // forward declaration
    class Lock;
 
    static Lock& getLockObject();
 
    class Lock {
     public:
      [[nodiscard]] bool tryLockFor(std::chrono::seconds timeout, bool shared);
      void unlock();
      [[nodiscard]] bool ownsLock() const { return _ownsLock; }
      ~Lock();
 
     private:
      Lock() = default;
 
      bool _ownsLock{false};
      bool _isShared{false}; // only meaningful when _ownsLock = true
 
      friend Lock& TestableMode::getLockObject();
    };
 
    std::map<boost::thread::id, std::string> _threadNames;
 
    std::map<boost::thread::id, pid_t> _threadPThreadId;
 
    std::mutex _threadNamesMutex;
 
    std::string threadName(const boost::thread::id& threadId = boost::this_thread::get_id());
 
    pid_t pthreadId(const boost::thread::id& threadId = boost::this_thread::get_id());
 
    friend class ChimeraTK::ConnectionMaker;
  };
 
  /********************************************************************************************************************/
  /********************************************************************************************************************/
  /***** Inline implementations for TestableMode **/
  /********************************************************************************************************************/
  /********************************************************************************************************************/
 
  template<typename T>
  TestableMode::AccessorPair<T> TestableMode::decorate(
      AccessorPair<T> other, const VariableNetworkNode& producer, const VariableNetworkNode& consumer) {
    if(not _enabled) {
      return other;
    }
 
    logger(Logger::Severity::info, "TestableMode")
        << "      Decorating pair " << producer.getQualifiedName() << "[" << other.first->getId() << "] -> "
        << consumer.getQualifiedName() << "[" << other.second->getId() << "]";
 
    // create variable IDs
    size_t varId = detail::TestableMode::getNextVariableId();
    size_t varIdReturn;
    AccessorPair<T> result;
 
    if(producer.getDirection().withReturn) {
      varIdReturn = detail::TestableMode::getNextVariableId();
    }
 
    // decorate the process variable if testable mode is enabled and mode is push-type
    if(!producer.getDirection().withReturn) {
      result.first = boost::make_shared<AccessorDecorator<T>>(*this, other.first, false, true, varId, varId);
      result.second = boost::make_shared<AccessorDecorator<T>>(*this, other.second, true, false, varId, varId);
    }
    else {
      result.first = boost::make_shared<AccessorDecorator<T>>(*this, other.first, true, true, varIdReturn, varId);
      result.second = boost::make_shared<AccessorDecorator<T>>(*this, other.second, true, true, varId, varIdReturn);
    }
 
    // put the decorators into the list
    auto& variable = _variables.at(varId);
    variable.name = "Internal:" + producer.getQualifiedName();
    if(consumer.getType() != NodeType::invalid) {
      variable.name += "->" + consumer.getQualifiedName();
    }
    if(producer.getDirection().withReturn) {
      auto& returnVariable = _variables.at(varIdReturn);
      returnVariable.name = variable.name + " (return)";
    }
 
    return result;
  }
 
  /********************************************************************************************************************/
 
  template<typename T>
  boost::shared_ptr<NDRegisterAccessor<T>> TestableMode::decorate(
      boost::shared_ptr<NDRegisterAccessor<T>> other, DecoratorType direction, const std::string& name, size_t varId) {
    if(not _enabled) {
      return other;
    }
 
    logger(Logger::Severity::info, "TestableMode")
        << "      Decorating single " << (direction == DecoratorType::READ ? "consumer " : "feeder ") << name << "["
        << other->getId() << "]";
 
    if(varId == 0) {
      varId = detail::TestableMode::getNextVariableId();
    }
 
    _variables[varId].processVariable = other;
    if(not name.empty()) {
      _variables.at(varId).name = name;
    }
 
    auto pvarDec = boost::make_shared<AccessorDecorator<T>>(
        *this, other, direction == DecoratorType::READ, direction == DecoratorType::WRITE, varId, varId);
 
    return pvarDec;
  }
 
  /********************************************************************************************************************/
  /********************************************************************************************************************/
  /***** Implementations for TestableMode::AccessorDecorator **/
  /********************************************************************************************************************/
  /********************************************************************************************************************/
 
  template<typename UserType>
  TestableMode::AccessorDecorator<UserType>::AccessorDecorator(detail::TestableMode& testableMode,
      boost::shared_ptr<ChimeraTK::NDRegisterAccessor<UserType>> accessor, bool handleRead, bool handleWrite,
      size_t variableIdRead, size_t variableIdWrite)
  : ChimeraTK::NDRegisterAccessorDecorator<UserType>(accessor), _handleRead(handleRead), _handleWrite(handleWrite),
    _variableIdRead(variableIdRead), _variableIdWrite(variableIdWrite), _testableMode(testableMode) {
    assert(_variableIdRead != 0);
    assert(_variableIdWrite != 0);
 
    // if receiving end, register for testable mode (stall detection)
    if(this->isReadable() && handleRead) {
      _testableMode._variables[_variableIdRead].processVariable = accessor;
      assert(accessor->getAccessModeFlags().has(AccessMode::wait_for_new_data));
    }
 
    // if this decorating a bidirectional process variable, set the
    // valueRejectCallback
    auto bidir = boost::dynamic_pointer_cast<BidirectionalProcessArray<UserType>>(accessor);
    if(bidir) {
      bidir->setValueRejectCallback([this] { decrementCounter(); });
    }
    else {
      assert(!(handleRead && handleWrite));
    }
  }
 
  /********************************************************************************************************************/
 
  template<typename UserType>
  // FIXME: https://redmine.msktools.desy.de/issues/12242
  // NOLINTNEXTLINE(google-default-arguments)
  bool TestableMode::AccessorDecorator<UserType>::doWriteTransfer(ChimeraTK::VersionNumber versionNumber) {
    if(!_handleWrite) {
      return _target->writeTransfer(versionNumber);
    }
    return accountForWriteOperation([this, versionNumber]() { return _target->writeTransfer(versionNumber); });
  }
 
  /********************************************************************************************************************/
 
  template<typename UserType>
  // FIXME: https://redmine.msktools.desy.de/issues/12242
  // NOLINTNEXTLINE(google-default-arguments)
  bool TestableMode::AccessorDecorator<UserType>::doWriteTransferDestructively(ChimeraTK::VersionNumber versionNumber) {
    if(!_handleWrite) {
      return _target->writeTransferDestructively(versionNumber);
    }
 
    return accountForWriteOperation(
        [this, versionNumber]() { return _target->writeTransferDestructively(versionNumber); });
  }
 
  /********************************************************************************************************************/
 
  template<typename UserType>
  void TestableMode::AccessorDecorator<UserType>::releaseLock() {
    if(_testableMode.testLock()) {
      _testableMode.unlock("doReadTransfer " + this->getName());
    }
  }
 
  /********************************************************************************************************************/
 
  template<typename UserType>
  void TestableMode::AccessorDecorator<UserType>::doPreRead(TransferType type) {
    _target->preRead(type);
 
    // Blocking reads have to release the lock so the data transport can happen
    if(_handleRead && type == TransferType::read &&
        TransferElement::_accessModeFlags.has(AccessMode::wait_for_new_data)) {
      releaseLock();
    }
  }
 
  /********************************************************************************************************************/
 
  template<typename UserType>
  void TestableMode::AccessorDecorator<UserType>::obtainLockAndDecrementCounter(bool hasNewData) {
    if(!_testableMode.testLock()) {
      _testableMode.lock("doReadTransfer " + this->getName(), true);
    }
    if(!hasNewData) {
      return;
    }
    auto& variable = _testableMode._variables.at(_variableIdRead);
    if(variable.counter > 0) {
      assert(_testableMode._counter > 0);
      --_testableMode._counter;
      --variable.counter;
      logger(Logger::Severity::trace, "TestableMode")
          << "TestableModeAccessorDecorator[name='" << this->getName() << "', id=" << _variableIdRead
          << "]: testableMode.counter decreased, now at value " << _testableMode._counter << " / " << variable.counter;
    }
    else {
      logger(Logger::Severity::trace, "TestableMode")
          << "TestableModeAccessorDecorator[name='" << this->getName() << "', id=" << _variableIdRead
          << "]: testableMode.counter NOT decreased, was already at value " << _testableMode._counter << " / "
          << variable.counter << "\n"
          << variable.name;
    }
  }
 
  /********************************************************************************************************************/
 
  template<typename UserType>
  void TestableMode::AccessorDecorator<UserType>::decrementCounter() {
    obtainLockAndDecrementCounter(true);
    releaseLock();
  }
 
  /********************************************************************************************************************/
 
  template<typename UserType>
  void TestableMode::AccessorDecorator<UserType>::doPostRead(TransferType type, bool hasNewData) {
    if(_handleRead) {
      obtainLockAndDecrementCounter(hasNewData);
    }
    ChimeraTK::NDRegisterAccessorDecorator<UserType>::doPostRead(type, hasNewData);
  }
 
  template<typename UserType>
  bool TestableMode::AccessorDecorator<UserType>::accountForWriteOperation(
      const std::function<bool(void)>& writeOperation) {
    bool dataLost = false;
    if(!_testableMode.testLock()) {
      // may happen if first write in thread is done before first blocking read
      _testableMode.lock("write " + this->getName(), true);
    }
 
    // Increment counters before write(), since another thread might react to the value on the queue already and
    // try to do something with the counter( e.g. decrement it conditionally, see obtainLockAndDecrementCounter()).
    _testableMode._variables.at(_variableIdWrite).counter++;
    _testableMode._counter++;
 
    dataLost = writeOperation();
 
    if(dataLost) {
      // if data has been lost, decrement counter again since we never actually put data onto the queue.
      _testableMode._variables.at(_variableIdWrite).counter--;
      _testableMode._counter--;
    }
 
    if(!dataLost) {
      logger(Logger::Severity::trace, "TestableMode")
          << "TestableModeAccessorDecorator::write[name='" << this->getName() << "', id=" << _variableIdWrite
          << "]: testableMode.counter increased, now at value " << _testableMode._counter;
    }
    else {
      logger(Logger::Severity::trace, "TestableMode")
          << "TestableModeAccessorDecorator::write[name='" << this->getName() << "', id=" << _variableIdWrite
          << "]: testableMode.counter not increased due to lost data";
    }
    return dataLost;
  }
 
  /********************************************************************************************************************/
 
} // namespace ChimeraTK::detail