InterfaceQueued.h

/* Author: Niklas Eiling <niklas.eiling@eonerc.rwth-aachen.de>
 * SPDX-FileCopyrightText: 2023-2024 Niklas Eiling <niklas.eiling@eonerc.rwth-aachen.de>
 * SPDX-License-Identifier: MPL-2.0
 */
 
#pragma once
 
#include <thread>
 
#include <dpsim-models/Attribute.h>
#include <dpsim-models/Logger.h>
#include <dpsim-models/Task.h>
#include <dpsim/Config.h>
#include <dpsim/Definitions.h>
#include <dpsim/Interface.h>
#include <dpsim/Scheduler.h>
 
#include <readerwriterqueue.h>
 
namespace DPsim {
 
class InterfaceWorker;
 
class InterfaceQueued : public Interface,
                        public SharedFactory<InterfaceQueued> {
 
public:
  typedef std::shared_ptr<InterfaceQueued> Ptr;
 
  using AttributePacket = struct AttributePacket {
    CPS::AttributeBase::Ptr value;
    UInt
        attributeId; // Used to identify the attribute. Defined by the position in the `mExportAttrsDpsim` and `mImportAttrsDpsim` lists
    UInt
        sequenceId; // Increasing ID used to discern multiple consecutive updates of a single attribute
    unsigned char flags; // Bit 0 set: Close interface
  };
 
  enum AttributePacketFlags {
    PACKET_NO_FLAGS = 0,
    PACKET_CLOSE_INTERFACE = 1,
  };
 
  InterfaceQueued(std::shared_ptr<InterfaceWorker> intf,
                  const String &name = "", UInt downsampling = 1)
      : Interface(name), mInterfaceWorker(intf), mDownsampling(downsampling) {
    mQueueDpsimToInterface = std::make_shared<
        moodycamel::BlockingReaderWriterQueue<AttributePacket>>();
    mQueueInterfaceToDpsim = std::make_shared<
        moodycamel::BlockingReaderWriterQueue<AttributePacket>>();
  };
 
  virtual void open() override;
  virtual void close() override;
 
  // Function used in the interface's simulation task to read all imported attributes from the queue
  // Called once before every simulation timestep
  virtual void pushDpsimAttrsToQueue();
  // Function used in the interface's simulation task to write all exported attributes to the queue
  // Called once after every simulation timestep
  virtual void popDpsimAttrsFromQueue(bool isSync = false);
 
  // Function called by the Simulation to perform interface synchronization
  virtual void syncExports() override;
  virtual void syncImports() override;
 
  virtual CPS::Task::List getTasks() override;
 
  virtual void setLogger(CPS::Logger::Log log) override;
 
  virtual ~InterfaceQueued() {
    if (mOpened)
      close();
  }
 
protected:
  std::shared_ptr<InterfaceWorker> mInterfaceWorker;
  UInt mDownsampling;
  std::thread mInterfaceWriterThread;
  std::thread mInterfaceReaderThread;
 
  std::shared_ptr<moodycamel::BlockingReaderWriterQueue<AttributePacket>>
      mQueueDpsimToInterface;
  std::shared_ptr<moodycamel::BlockingReaderWriterQueue<AttributePacket>>
      mQueueInterfaceToDpsim;
 
public:
  class WriterThread {
  private:
    std::shared_ptr<moodycamel::BlockingReaderWriterQueue<AttributePacket>>
        mQueueDpsimToInterface;
    std::shared_ptr<InterfaceWorker> mInterfaceWorker;
 
  public:
    WriterThread(
        std::shared_ptr<moodycamel::BlockingReaderWriterQueue<AttributePacket>>
            queueDpsimToInterface,
        std::shared_ptr<InterfaceWorker> intf)
        : mQueueDpsimToInterface(queueDpsimToInterface),
          mInterfaceWorker(intf){};
    void operator()() const;
  };
 
  class ReaderThread {
  private:
    std::shared_ptr<moodycamel::BlockingReaderWriterQueue<AttributePacket>>
        mQueueInterfaceToDpsim;
    std::shared_ptr<InterfaceWorker> mInterfaceWorker;
    std::atomic<bool> &mOpened;
 
  public:
    ReaderThread(
        std::shared_ptr<moodycamel::BlockingReaderWriterQueue<AttributePacket>>
            queueInterfaceToDpsim,
        std::shared_ptr<InterfaceWorker> intf, std::atomic<bool> &opened)
        : mQueueInterfaceToDpsim(queueInterfaceToDpsim), mInterfaceWorker(intf),
          mOpened(opened){};
    void operator()() const;
  };
 
  class PreStep : public CPS::Task {
  public:
    explicit PreStep(InterfaceQueued &intf)
        : Task(intf.getName() + ".Read"), mIntf(intf) {
      for (const auto &[attr, _seqId, _blockOnRead, _syncOnStart] :
           intf.mImportAttrsDpsim) {
        mModifiedAttributes.push_back(attr);
      }
    }
 
    void execute(Real time, Int timeStepCount) override;
 
  private:
    InterfaceQueued &mIntf;
  };
 
  class PostStep : public CPS::Task {
  public:
    explicit PostStep(InterfaceQueued &intf)
        : Task(intf.getName() + ".Write"), mIntf(intf) {
      for (const auto &[attr, _seqId] : intf.mExportAttrsDpsim) {
        mAttributeDependencies.push_back(attr);
      }
      mModifiedAttributes.push_back(Scheduler::external);
    }
 
    void execute(Real time, Int timeStepCount) override;
 
  private:
    InterfaceQueued &mIntf;
  };
};
} // namespace DPsim