TransferElement.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 "AccessMode.h"
#include "DeviceBackend.h"
#include "Exception.h"
#include "TransferElementID.h"
#include "VersionNumber.h"
 
#include <ChimeraTK/cppext/future_queue.hpp>
 
#include <boost/bind/bind.hpp>
#include <boost/enable_shared_from_this.hpp>
#include <boost/numeric/conversion/cast.hpp>
#include <boost/shared_ptr.hpp>
#include <boost/thread.hpp>
#include <boost/thread/future.hpp>
 
#include <functional>
#include <iostream>
#include <list>
#include <string>
#include <typeinfo>
#include <utility>
#include <vector>
 
namespace ChimeraTK {
  class PersistentDataStorage;
  class TransferGroup;
  class ReadAnyGroup;
 
  enum class DataValidity {
    ok,    
    faulty 
  };
 
  std::ostream& operator<<(std::ostream& os, const DataValidity& validity);
 
  enum class TransferType { read, readNonBlocking, readLatest, write, writeDestructively };
 
  namespace detail {
    class DiscardValueException {};
 
  } /* namespace detail */
 
  /********************************************************************************************************************/
 
  class TransferElement : public boost::enable_shared_from_this<TransferElement> {
   public:
    TransferElement(std::string name, AccessModeFlags accessModeFlags, std::string unit = std::string(unitNotSet),
        std::string description = std::string())
    : _name(std::move(name)), _unit(std::move(unit)), _description(std::move(description)),
      _accessModeFlags(std::move(accessModeFlags)) {}
 
    TransferElement(const TransferElement& other) = delete;
    TransferElement(TransferElement&& other) = delete;
    TransferElement& operator=(const TransferElement& other) = delete;
    TransferElement& operator=(TransferElement&& other) = delete;
 
    virtual ~TransferElement() = default;
 
    using SharedPtr = boost::shared_ptr<TransferElement>;
 
    const std::string& getName() const { return _name; }
 
    const std::string& getUnit() const { return _unit; }
 
    const std::string& getDescription() const { return _description; }
 
    virtual const std::type_info& getValueType() const = 0;
 
    AccessModeFlags getAccessModeFlags() const { return _accessModeFlags; }
 
    void setDataValidity(DataValidity validity = DataValidity::ok) { _dataValidity = validity; }
 
    DataValidity dataValidity() const { return _dataValidity; }
 
    void read() {
      if(TransferElement::_isInTransferGroup) {
        throw ChimeraTK::logic_error("Calling read() or write() on the TransferElement '" + _name +
            "' which is part of a TransferGroup is not allowed.");
      }
      if(TransferElement::_inReadAnyGroup) {
        throw ChimeraTK::logic_error("Directly calling read() on the TransferElement '" + _name +
            "' which is part of a ReadAnyGroup is not allowed.");
      }
      this->readTransactionInProgress = false;
 
      preReadAndHandleExceptions(TransferType::read);
      if(!_activeException) {
        handleTransferException([&] { readTransfer(); });
      }
 
      postReadAndHandleExceptions(TransferType::read, !_activeException);
    }
 
    bool readNonBlocking() {
      if(TransferElement::_isInTransferGroup) {
        throw ChimeraTK::logic_error("Calling read() or write() on the TransferElement '" + _name +
            "' which is part of a TransferGroup is not allowed.");
      }
      if(TransferElement::_inReadAnyGroup) {
        throw ChimeraTK::logic_error("Directly calling readNonBlocking() on the TransferElement '" + _name +
            "' which is part of a ReadAnyGroup is not allowed.");
      }
      this->readTransactionInProgress = false;
      preReadAndHandleExceptions(TransferType::readNonBlocking);
      bool updateDataBuffer = false;
      if(!_activeException) {
        handleTransferException([&] { updateDataBuffer = readTransferNonBlocking(); });
      }
 
      bool retVal = updateDataBuffer;
      if(_activeException) {
        auto previousVersionNumber = _versionNumber;
        auto previousDataValidity = _dataValidity;
        // always call postRead with updateDataBuffer = false in case of an exception
        postReadAndHandleExceptions(TransferType::readNonBlocking, false);
        // Usually we do not reach this point because postRead() is re-throwing the _activeException.
        // If we reach this point the exception has been suppressed. We have to calculate a
        // new return value because the dataBuffer has not changed, but the meta data
        // could have, in which case we have to return true.
        retVal = (previousVersionNumber != _versionNumber) || (previousDataValidity != _dataValidity);
      }
      else {
        // call postRead with updateDataBuffer as returned by readTransferNonBlocking
        postReadAndHandleExceptions(TransferType::readNonBlocking, updateDataBuffer);
      }
      return retVal;
    }
 
    bool readLatest() {
      if(_accessModeFlags.has(AccessMode::wait_for_new_data)) {
        bool updateDataBuffer = false;
        // Call readNonBlocking until there is no new data to be read any more
        while(readNonBlocking()) {
          // remember whether we have new data
          updateDataBuffer = true;
        }
        return updateDataBuffer;
      }
      // Without wait_for_new_data readNonBlocking always returns true, and the while loop above would never end.
      // Hence we just call the (synchronous) read and return true;
      read();
      return true;
    }
 
    bool write(ChimeraTK::VersionNumber versionNumber = {}) {
      if(TransferElement::_isInTransferGroup) {
        throw ChimeraTK::logic_error("Calling read() or write() on the TransferElement '" + _name +
            "' which is part of a TransferGroup is not allowed.");
      }
      this->writeTransactionInProgress = false;
      bool previousDataLost = true; // the value here does not matter. If there was an exception, it will be re-thrown
                                    // in postWrite, so it is never returned
 
      preWriteAndHandleExceptions(TransferType::write, versionNumber);
      if(!_activeException) {
        handleTransferException([&] { previousDataLost = writeTransfer(versionNumber); });
      }
 
      postWriteAndHandleExceptions(TransferType::write, versionNumber);
      return previousDataLost;
    }
 
    bool writeDestructively(ChimeraTK::VersionNumber versionNumber = {}) {
      if(TransferElement::_isInTransferGroup) {
        throw ChimeraTK::logic_error("Calling read() or write() on the TransferElement '" + _name +
            "' which is part of a TransferGroup is not allowed.");
      }
      this->writeTransactionInProgress = false;
 
      preWriteAndHandleExceptions(TransferType::writeDestructively, versionNumber);
      bool previousDataLost = true; // the value here does not matter. If there was an exception, it will be re-thrown
                                    // in postWrite, so it is never returned
      if(!_activeException) {
        handleTransferException([&] { previousDataLost = writeTransferDestructively(versionNumber); });
      }
 
      postWriteAndHandleExceptions(TransferType::writeDestructively, versionNumber);
      return previousDataLost;
    }
 
    ChimeraTK::VersionNumber getVersionNumber() const { return _versionNumber; }
 
    virtual bool isReadOnly() const = 0;
 
    virtual bool isReadable() const = 0;
 
    virtual bool isWriteable() const = 0;
 
    void setActiveException(std::exception_ptr& setThisException) {
      if(setThisException) {
        _activeException = setThisException;
        setThisException = nullptr;
      }
    }
 
    virtual void setExceptionBackend(boost::shared_ptr<DeviceBackend> exceptionBackend) {
      _exceptionBackend = std::move(exceptionBackend);
    }
 
    boost::shared_ptr<DeviceBackend> getExceptionBackend() { return _exceptionBackend; }
 
    cppext::future_queue<void> getReadQueue() { return _readQueue; }
 
   protected:
    boost::shared_ptr<DeviceBackend> _exceptionBackend;
 
   private:
    template<typename Callable>
    void handleTransferException(Callable function) {
      try {
        function();
      }
      catch(ChimeraTK::runtime_error&) {
        _activeException = std::current_exception();
      }
      catch(boost::thread_interrupted&) {
        _activeException = std::current_exception();
      }
    }
 
    // helper function that just gets rid of the DiscardValueException and otherwise does a pop_wait on the _readQueue.
    // It does not deal with other exceptions. This is done in handleTransferException.
    void readTransferAsyncWaitingImpl() {
    retry:
      try {
        _readQueue.pop_wait();
      }
      catch(detail::DiscardValueException&) {
        goto retry;
      }
    }
 
   public:
    void readTransfer() {
      if(_accessModeFlags.has(AccessMode::wait_for_new_data)) {
        readTransferAsyncWaitingImpl();
      }
      else {
        doReadTransferSynchronously();
      }
    }
 
   protected:
    virtual void doReadTransferSynchronously() = 0;
 
   private:
    // helper function that just gets rid of the DiscardValueException and otherwise does a pop on the _readQueue.
    // It does not deal with other exceptions. This is done in handleTransferException.
    bool readTransferAsyncNonWaitingImpl() {
    retry:
      try {
        return _readQueue.pop();
      }
      catch(detail::DiscardValueException&) {
        goto retry;
      }
    }
 
   public:
    bool readTransferNonBlocking() {
      if(_accessModeFlags.has(AccessMode::wait_for_new_data)) {
        return readTransferAsyncNonWaitingImpl();
      }
      doReadTransferSynchronously();
      return true;
    }
 
   private:
    void preReadAndHandleExceptions(TransferType type) noexcept {
      try {
        preRead(type);
      }
      catch(ChimeraTK::logic_error&) {
        _activeException = std::current_exception();
      }
      catch(ChimeraTK::runtime_error&) {
        _activeException = std::current_exception();
      }
      catch(boost::thread_interrupted&) {
        _activeException = std::current_exception();
      }
    }
 
   public:
    void preRead(TransferType type) {
      if(readTransactionInProgress) return;
      _activeException = {nullptr};
 
      readTransactionInProgress = true; // remember that doPreRead has been called. It might throw, so we remember
                                        // before we call it
      doPreRead(type);
    }
 
   protected:
    virtual void doPreRead(TransferType) {}
 
   private:
    void postReadAndHandleExceptions(TransferType type, bool updateDataBuffer) {
      try {
        postRead(type, updateDataBuffer);
      }
      catch(ChimeraTK::runtime_error& ex) {
        if(_exceptionBackend) {
          _exceptionBackend->setException(ex.what());
        }
        throw;
      }
    }
 
   public:
    void postRead(TransferType type, bool updateDataBuffer) {
      // only delegate to doPostRead() the first time postRead() is called in a row.
      if(readTransactionInProgress) {
        readTransactionInProgress = false;
        doPostRead(type, updateDataBuffer);
      }
 
      // Throw on each call of postRead(). All high-level elements for a shared low-level transfer element must see the
      // exception. Note: doPostRead can throw an exception, but in that case _activeException must be false (we can
      // only have one exception at a time). In case other code is added here later which needs to be executed after
      // doPostRead() always, a try-catch block may be necessary.
      if(_activeException) {
        // don't clear the active connection. This is done in preRead().
        std::rethrow_exception(_activeException);
      }
    }
 
   protected:
    virtual void doPostRead(TransferType, bool /*updateDataBuffer*/) {}
 
   private:
    void preWriteAndHandleExceptions(TransferType type, ChimeraTK::VersionNumber versionNumber) noexcept {
      try {
        preWrite(type, versionNumber);
      }
      catch(ChimeraTK::logic_error&) {
        _activeException = std::current_exception();
      }
      catch(ChimeraTK::runtime_error&) {
        _activeException = std::current_exception();
      }
      catch(boost::thread_interrupted&) {
        _activeException = std::current_exception();
      }
    }
 
   public:
    void preWrite(TransferType type, ChimeraTK::VersionNumber versionNumber) {
      if(writeTransactionInProgress) return;
 
      _activeException = {};
      if(versionNumber < getVersionNumber()) {
        throw ChimeraTK::logic_error("The version number " + std::string(versionNumber) +
            " passed to write() of TransferElement '" + _name + "' is less than the last version number used " +
            std::string(getVersionNumber()) + ".");
      }
      writeTransactionInProgress = true; // must not be set, if the logic_error is thrown above due to the old version
      doPreWrite(type, versionNumber);
    }
 
   protected:
    virtual void doPreWrite(TransferType, VersionNumber) {}
 
   private:
    void postWriteAndHandleExceptions(TransferType type, VersionNumber versionNumber) {
      try {
        postWrite(type, versionNumber);
      }
      catch(ChimeraTK::runtime_error& ex) {
        if(_exceptionBackend) {
          _exceptionBackend->setException(ex.what());
        }
        throw;
      }
    }
 
   public:
    void postWrite(TransferType type, VersionNumber versionNumber) {
      if(writeTransactionInProgress) {
        writeTransactionInProgress = false;
        doPostWrite(type, versionNumber);
      }
 
      // Note: doPostWrite can throw an exception, but in that case hasSeenException must be false (we can only have one
      // exception at a time). In case other code is added here later which needs to be executed after doPostWrite()
      // always, a try-catch block may be necessary.
      // Another note: If writeTransactionInProgress == false, there can still be an exception, if the version number
      // used in a write was too old (see preWrite).
      if(_activeException) {
        std::rethrow_exception(_activeException);
      }
 
      // only after a successful write the version number is updated
      _versionNumber = versionNumber;
    }
 
   protected:
    virtual void doPostWrite(TransferType, VersionNumber) {}
 
   public:
    bool writeTransfer(ChimeraTK::VersionNumber versionNumber) { return doWriteTransfer(versionNumber); }
 
   protected:
    virtual bool doWriteTransfer(ChimeraTK::VersionNumber versionNumber) = 0;
 
   public:
    bool writeTransferDestructively(ChimeraTK::VersionNumber versionNumber) {
      return doWriteTransferDestructively(versionNumber);
    }
 
   protected:
    virtual bool doWriteTransferDestructively(ChimeraTK::VersionNumber versionNumber) {
      return doWriteTransfer(versionNumber);
    }
 
   public:
    virtual bool mayReplaceOther(const boost::shared_ptr<TransferElement const>& other) const {
      (void)other; // prevent warning
      return false;
    }
 
    virtual std::vector<boost::shared_ptr<TransferElement>> getHardwareAccessingElements() = 0;
 
    virtual std::list<boost::shared_ptr<TransferElement>> getInternalElements() = 0;
 
    virtual boost::shared_ptr<TransferElement> getHighLevelImplElement() { return shared_from_this(); }
 
    // FIXME #11279 Implement API breaking changes from linter warnings
    // NOLINTNEXTLINE(performance-unnecessary-value-param)
    virtual void replaceTransferElement([[maybe_unused]] boost::shared_ptr<TransferElement> newElement) {}
 
    virtual boost::shared_ptr<TransferElement> makeCopyRegisterDecorator() = 0;
 
    static constexpr char unitNotSet[] = "n./a.";
 
    // FIXME #11279 Implement API breaking changes from linter warnings
    // NOLINTNEXTLINE(performance-unnecessary-value-param)
    virtual void setPersistentDataStorage(boost::shared_ptr<ChimeraTK::PersistentDataStorage>) {}
 
    TransferElementID getId() const { return _id; }
 
    virtual void interrupt() {}
 
    template<typename QUEUE_TYPE>
    void interrupt_impl(QUEUE_TYPE& dataTransportQueue) {
      dataTransportQueue.push_overwrite_exception(std::make_exception_ptr(boost::thread_interrupted()));
    }
 
    bool isReadTransactionInProgress() const { return readTransactionInProgress; }
 
    bool isWriteTransactionInProgress() const { return writeTransactionInProgress; }
 
    [[nodiscard]] ReadAnyGroup* getReadAnyGroup() const { return _inReadAnyGroup; }
 
    virtual void setInReadAnyGroup(ReadAnyGroup* rag) { _inReadAnyGroup = rag; }
 
   protected:
    std::string _name;
 
    std::string _unit;
 
    std::string _description;
 
    TransferElementID _id;
 
    void makeUniqueId() { _id.makeUnique(); }
 
    bool _isInTransferGroup{false};
 
    ReadAnyGroup* _inReadAnyGroup{nullptr};
 
    AccessModeFlags _accessModeFlags;
 
    friend class TransferGroup;
    friend class ReadAnyGroup;
 
   private:
    bool readTransactionInProgress{false};
 
    bool writeTransactionInProgress{false};
 
   protected:
    cppext::future_queue<void> _readQueue;
 
    VersionNumber _versionNumber{nullptr};
 
    DataValidity _dataValidity{DataValidity::ok};
 
    std::exception_ptr _activeException{nullptr};
  }; // namespace ChimeraTK
 
} /* namespace ChimeraTK */