ReadAnyGroup.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 "TransferElement.h"
#include "TransferElementAbstractor.h"
 
#include <ChimeraTK/cppext/future_queue.hpp>
 
#include <functional>
#include <initializer_list>
 
namespace ChimeraTK {
 
  namespace DataConsistencyGroupDetail {
    class HistorizedMatcher;
  } // namespace DataConsistencyGroupDetail
 
  class ReadAnyGroup {
    // friend because it needs to decorate our push_elements
    friend class ChimeraTK::DataConsistencyGroupDetail::HistorizedMatcher;
 
   public:
    class Notification {
     public:
      friend class ReadAnyGroup;
 
      Notification();
 
      Notification(Notification&& other) noexcept;
 
      ~Notification();
 
      Notification& operator=(Notification&& other) noexcept;
 
      bool accept();
 
      [[nodiscard]] TransferElementID getId();
 
      [[nodiscard]] std::size_t getIndex() const;
 
      [[nodiscard]] TransferElementAbstractor getTransferElement();
 
      [[nodiscard]] bool isReady() const;
 
      /*
       * Notifications cannot be copied because each notification can only be accepted once.
       */
      Notification(const Notification&) = delete;
      Notification& operator=(const Notification&) = delete;
 
     private:
      Notification(std::size_t index, ReadAnyGroup* owner);
 
      bool accepted{false};
 
      std::size_t index{};
 
      bool valid{false};
 
      ReadAnyGroup* _owner{nullptr};
    };
 
    ReadAnyGroup();
 
    ReadAnyGroup(std::initializer_list<std::reference_wrapper<TransferElementAbstractor>> list);
 
    ReadAnyGroup(std::initializer_list<boost::shared_ptr<TransferElement>> list);
 
    template<typename ITERATOR>
    ReadAnyGroup(ITERATOR first, ITERATOR last);
 
    void add(TransferElementAbstractor& element);
 
    ReadAnyGroup(const ReadAnyGroup&) = delete;
    ReadAnyGroup(ReadAnyGroup&& other) noexcept { operator=(std::move(other)); }
    ReadAnyGroup& operator=(ReadAnyGroup&& other) noexcept;
 
    void add(boost::shared_ptr<TransferElement> element);
 
    void finalise();
 
    TransferElementID readAny();
 
    TransferElementID readAnyNonBlocking();
 
    void readUntil(const TransferElementID& id);
 
    void readUntil(const TransferElementAbstractor& element);
 
    void readUntilAll(const std::vector<TransferElementID>& ids);
 
    void readUntilAll(const std::vector<TransferElementAbstractor>& elements);
 
    Notification waitAny();
 
    Notification waitAnyNonBlocking();
 
    void processPolled();
 
    void interrupt() { push_elements.front().getHighLevelImplElement()->interrupt(); }
 
   private:
    void handlePreRead();
 
    bool isFinalised{false};
 
    std::vector<TransferElementAbstractor> push_elements;
 
    std::vector<TransferElementAbstractor> poll_elements;
 
    cppext::future_queue<size_t> notification_queue;
 
    size_t _lastOperationIndex{std::numeric_limits<size_t>::max()};
  };
 
  /********************************************************************************************************************/
  /* Implementations of ReadAnyGroup::Notification */
  /********************************************************************************************************************/
 
  inline ReadAnyGroup::Notification::Notification() = default;
 
  /********************************************************************************************************************/
 
  inline ReadAnyGroup::Notification::Notification(Notification&& other) noexcept
  : accepted(other.accepted), index(other.index), valid(other.valid), _owner(other._owner) {
    other.valid = false;
  }
 
  /********************************************************************************************************************/
 
  inline ReadAnyGroup::Notification::~Notification() {
    // It is important that each received notification is consumed. This means that we have to accept a notification
    // before we can destroy it.
    if(this->valid && !this->accepted) {
      try {
        this->accept();
      }
      catch(...) {
        // Do not let exceptions escape the destructor, rather terminate the application instead. This will still print
        // the exception message.
        std::terminate();
      }
    }
  }
 
  /********************************************************************************************************************/
 
  inline ReadAnyGroup::Notification& ReadAnyGroup::Notification::operator=(Notification&& other) noexcept {
    // It is important that each received notification is consumed. This means that we have to accept this notification
    // before we can overwrite it with another one.
    if(this->valid && !this->accepted) {
      try {
        this->accept();
      }
      catch(...) {
        // Do not let exceptions escape the move operation, rather terminate the application instead. This will still
        // print the exception message.
        std::terminate();
      }
    }
    this->accepted = other.accepted;
    this->index = other.index;
    this->valid = other.valid;
    this->_owner = other._owner;
    other.valid = false;
    return *this;
  }
 
  /********************************************************************************************************************/
 
  inline bool ReadAnyGroup::Notification::accept() {
    if(!this->valid) {
      throw ChimeraTK::logic_error("This notification object is invalid.");
    }
    if(this->accepted) {
      throw ChimeraTK::logic_error("This notification has already been accepted.");
    }
    this->accepted = true;
    bool hasSeenException = false;
    try {
      _owner->push_elements[index].getHighLevelImplElement()->_readQueue.pop_wait();
    }
    catch(ChimeraTK::runtime_error&) {
      _owner->push_elements[index].getHighLevelImplElement()->_activeException = std::current_exception();
      hasSeenException = true;
    }
    catch(boost::thread_interrupted&) {
      _owner->push_elements[index].getHighLevelImplElement()->_activeException = std::current_exception();
      hasSeenException = true;
    }
    catch(detail::DiscardValueException&) {
      // we must not call postRead() in this case, hence we do not call preRead()
      _owner->_lastOperationIndex = std::numeric_limits<size_t>::max() - 1;
      return false;
    }
    _owner->_lastOperationIndex = index;
    _owner->push_elements[index].getHighLevelImplElement()->postRead(TransferType::read, !hasSeenException);
    return true;
  }
 
  /********************************************************************************************************************/
 
  inline TransferElementID ReadAnyGroup::Notification::getId() {
    if(!this->valid) {
      throw ChimeraTK::logic_error("This notification object is invalid.");
    }
    return _owner->push_elements[index].getId();
  }
 
  /********************************************************************************************************************/
 
  inline std::size_t ReadAnyGroup::Notification::getIndex() const {
    if(!this->valid) {
      throw ChimeraTK::logic_error("This notification object is invalid.");
    }
    return this->index;
  }
 
  /********************************************************************************************************************/
 
  inline TransferElementAbstractor ReadAnyGroup::Notification::getTransferElement() {
    if(!this->valid) {
      throw ChimeraTK::logic_error("This notification object is invalid.");
    }
    return _owner->push_elements[index];
  }
 
  /********************************************************************************************************************/
 
  inline bool ReadAnyGroup::Notification::isReady() const {
    return this->valid && !this->accepted;
  }
 
  /********************************************************************************************************************/
 
  inline ReadAnyGroup::Notification::Notification(std::size_t index_, ReadAnyGroup* owner)
  : index(index_), valid(true), _owner(owner) {}
 
  /********************************************************************************************************************/
  /* Implementations of ReadAnyGroup */
  /********************************************************************************************************************/
 
  inline ReadAnyGroup::ReadAnyGroup() = default;
 
  /********************************************************************************************************************/
 
  inline ReadAnyGroup::ReadAnyGroup(std::initializer_list<std::reference_wrapper<TransferElementAbstractor>> list) {
    for(const auto& element : list) add(element);
    finalise();
  }
 
  /********************************************************************************************************************/
 
  inline ReadAnyGroup::ReadAnyGroup(std::initializer_list<boost::shared_ptr<TransferElement>> list) {
    for(const auto& element : list) add(element);
    finalise();
  }
 
  /********************************************************************************************************************/
 
  template<typename ITERATOR>
  ReadAnyGroup::ReadAnyGroup(ITERATOR first, ITERATOR last) {
    for(ITERATOR it = first; it != last; ++it) add(*it);
    finalise();
  }
 
  /********************************************************************************************************************/
 
  inline ReadAnyGroup& ReadAnyGroup::operator=(ReadAnyGroup&& other) noexcept {
    // we need non-default implementation because we have to move pointers to ReadAnyGroup
    this->isFinalised = other.isFinalised;
    this->push_elements = std::move(other.push_elements);
    this->poll_elements = std::move(other.poll_elements);
    this->_lastOperationIndex = other._lastOperationIndex;
    this->notification_queue = std::move(other.notification_queue);
    for(auto& e : push_elements) {
      e.getHighLevelImplElement()->setInReadAnyGroup(this);
    }
    return *this;
  }
 
  /********************************************************************************************************************/
 
  inline void ReadAnyGroup::add(TransferElementAbstractor& element) {
    if(isFinalised) {
      throw ChimeraTK::logic_error("ReadAnyGroup has already been finalised, calling "
                                   "add() is no longer allowed.");
    }
    if(!element.isReadable()) {
      throw ChimeraTK::logic_error(
          "Cannot add non-readable accessor for register " + element.getName() + " to ReadAnyGroup.");
    }
    if(element.getReadAnyGroup() == this) {
      return;
    }
    if(element.getReadAnyGroup() != nullptr) {
      throw ChimeraTK::logic_error(element.getName() + " is already in a different ReadAnyGroup");
    }
    if(element.getAccessModeFlags().has(AccessMode::wait_for_new_data)) {
      push_elements.push_back(element);
    }
    else {
      poll_elements.push_back(element);
    }
    // set flag on the accessor that it is now in a ReadAnyGroup:
    // We do this for push-types only, since poll-types technically still allow calling read() without the ReadAnyGroup,
    // although its documentation states that would not be allowed.
    if(element.getAccessModeFlags().has(AccessMode::wait_for_new_data)) {
      element.getHighLevelImplElement()->setInReadAnyGroup(this);
    }
  }
 
  /********************************************************************************************************************/
 
  inline void ReadAnyGroup::add(boost::shared_ptr<TransferElement> element) {
    TransferElementAbstractor a(std::move(element));
    add(a);
  }
 
  /********************************************************************************************************************/
 
  inline void ReadAnyGroup::finalise() {
    if(isFinalised) {
      throw ChimeraTK::logic_error("ReadAnyGroup has already been finalised, calling "
                                   "finalise() is no longer allowed.");
    }
    std::vector<cppext::future_queue<void>> queueList;
    bool groupEmpty = true;
    for(auto& e : push_elements) {
      queueList.push_back(e.getHighLevelImplElement()->_readQueue);
      groupEmpty = false;
    }
    if(groupEmpty) {
      throw ChimeraTK::logic_error("ReadAnyGroup has no element with AccessMode::wait_for_new_data.");
    }
    notification_queue = cppext::when_any(queueList.begin(), queueList.end());
    isFinalised = true;
  }
 
  /********************************************************************************************************************/
 
  inline TransferElementID ReadAnyGroup::readAny() {
    Notification notification;
    do {
      notification = this->waitAny();
    } while(!notification.accept());
 
    this->processPolled();
 
    return notification.getId();
  }
 
  /********************************************************************************************************************/
 
  inline TransferElementID ReadAnyGroup::readAnyNonBlocking() {
    Notification notification;
    do {
      notification = this->waitAnyNonBlocking();
      if(!notification.isReady()) {
        this->processPolled();
        return {};
      }
    } while(!notification.accept());
 
    this->processPolled();
 
    return notification.getId();
  }
  /********************************************************************************************************************/
 
  inline void ReadAnyGroup::handlePreRead() {
    // preRead() and postRead() must be called in pairs. Hence we call all preReads here before waiting for transfers to
    // finish. postRead() will be called when accepting the notification. We can call preRead() repeatedly on the same
    // element, even if no transfer and call to postRead() have happened. It is just ignored (see Transfer element spec
    // B.5.2). Since this has a performance impact which might be significant on big applications, we try to avoid
    // unnecessary calls anyway.
 
    // Notice : This has the side effect that decorators can block here, for instance for the setup phase. This is used
    // by ApplicationCore in testable mode.
    if(_lastOperationIndex == std::numeric_limits<size_t>::max()) {
      for(auto& elem : push_elements) {
        elem.getHighLevelImplElement()->preRead(TransferType::read);
      }
    }
    else if(_lastOperationIndex != std::numeric_limits<size_t>::max() - 1) {
      // Note: _lastOperationIndex is set to std::numeric_limits<size_t>::max() - 1 in case a DiscardValueException
      // has been seen, in which case no postRead() is called.
      push_elements[_lastOperationIndex].getHighLevelImplElement()->preRead(TransferType::read);
    }
  }
 
  /********************************************************************************************************************/
 
  inline ReadAnyGroup::Notification ReadAnyGroup::waitAny() {
    handlePreRead();
 
    // Wait for notification
    std::size_t index;
    notification_queue.pop_wait(index);
    // clazy has a false positive warning about index being uninitialised.
    // NOLINTNEXTLINE(clang-analyzer-core.CallAndMessage)
    return {index, this};
  }
 
  /********************************************************************************************************************/
 
  inline ReadAnyGroup::Notification ReadAnyGroup::waitAnyNonBlocking() {
  restart_after_discard_value:
    // check if update is available
    if(notification_queue.empty()) {
      // If no notification is present, do not even execute preRead. This is necessary for two reasons:
      // - We always used TransferType::read to avoid mixing TransferType::read and TransferType::readNonBlocking
      //   in the same transfer of the same variable. We can do this even in this non-blocking case, if we already know
      //   that there will be an update read, since there will not be any difference beyond this point.
      // - In ApplicationCore testable mode, the testable mode lock must be released in a preRead before any blocking
      //   read operation. If preRead is called here when no update is available, no preRead will be called in a
      //   possible subsequend blocking readAny(), hence there would be no way to release the testable mode lock in the
      //   right place.
      return {};
    }
 
    // if update is available, peek into the queue to check whether a DiscardValueException will be read
    auto id = notification_queue.front();
    try {
      // call to empty() necessary before call to front(), to gain ownership of front element
      if(push_elements[id].getHighLevelImplElement()->_readQueue.empty()) {
        // cannot place the call to empty() into the assert, since it must be executed also in release builds
        assert(false);
      }
      push_elements[id].getHighLevelImplElement()->_readQueue.front();
    }
    catch(detail::DiscardValueException&) {
      // Remove discarded transfer from the queues and go back to square one
      notification_queue.pop();
      try {
        push_elements[id].getHighLevelImplElement()->_readQueue.pop();
      }
      catch(detail::DiscardValueException&) {
        goto restart_after_discard_value;
      }
      assert(false); // we must never end up at this point
    }
    catch(ChimeraTK::runtime_error&) {
      // While peeking we found another runtime which is stored in the queue.
      // Don't let it through, but leave it on the queue and continue with the waitAny().
      // It will be handled later.
    }
    catch(boost::thread_interrupted&) {
      // Also suppress the thread_interrupted exception here.
      // It will stay on the queue, hence it's not lost and will be handled later.
    }
    catch(...) {
      std::cout << "Fatal ERROR in ReadAnyGroup: Found unexpected exception on the read queue. Terminating!"
                << std::endl;
      std::terminate();
    }
 
    // now that we know that an update is available, we can defer to waitAny()
    return waitAny();
  }
 
  /********************************************************************************************************************/
 
  inline void ReadAnyGroup::processPolled() {
    // update all poll-type elements in the group
    for(auto& e : poll_elements) {
      if(!e.getAccessModeFlags().has(AccessMode::wait_for_new_data)) e.readLatest();
    }
  }
 
  /********************************************************************************************************************/
 
  inline void ReadAnyGroup::readUntil(const TransferElementID& id) {
    while(true) {
      auto read = readAny();
      if(read == id) return;
    }
  }
 
  /********************************************************************************************************************/
 
  inline void ReadAnyGroup::readUntil(const TransferElementAbstractor& element) {
    readUntil(element.getId());
  }
 
  /********************************************************************************************************************/
 
  inline void ReadAnyGroup::readUntilAll(const std::vector<TransferElementID>& ids) {
    std::map<TransferElementID, bool> found;
    for(const auto& id : ids) {
      found[id] = false; // initialise map so we can tell from the map if a
                         // variable is in the vector
    }
    size_t leftToFind = ids.size();
    while(true) {
      auto read = readAny();
      if(found.count(read) == 0) continue; // variable is not in the vector ids
      if(found[read]) continue;            // variable has been read already
      found[read] = true;
      --leftToFind;
      if(leftToFind == 0) return;
    }
  }
 
  /********************************************************************************************************************/
 
  inline void ReadAnyGroup::readUntilAll(const std::vector<TransferElementAbstractor>& elements) {
    std::map<TransferElementID, bool> found;
    for(const auto& elem : elements) {
      found[elem.getId()] = false; // initialise map so we can tell from the map
                                   // if a variable is in the vector
    }
    size_t leftToFind = elements.size();
    while(true) {
      auto read = readAny();
      if(found.count(read) == 0) continue; // variable is not in the vector elements
      if(found[read]) continue;            // variable has been read already
      found[read] = true;
      --leftToFind;
      if(leftToFind == 0) return;
    }
  }
 
  /********************************************************************************************************************/
 
} /* namespace ChimeraTK */