RPCEngine.hpp

Go to the documentation of this file.

/*
 *   Copyright 2025 Huawei Technologies Co., Ltd.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
 
#pragma once
 
#include <hicr/core/instanceManager.hpp>
#include <hicr/core/communicationManager.hpp>
#include <hicr/core/memoryManager.hpp>
#include <hicr/core/topologyManager.hpp>
#include <hicr/frontends/channel/variableSize/mpsc/locking/consumer.hpp>
#include <hicr/frontends/channel/variableSize/mpsc/locking/producer.hpp>
#include <hicr/frontends/channel/fixedSize/mpsc/nonlocking/consumer.hpp>
#include <hicr/frontends/channel/fixedSize/mpsc/nonlocking/producer.hpp>
 
#define _HICR_RPC_ENGINE_CHANNEL_PAYLOAD_CAPACITY 1048576
 
#define _HICR_RPC_ENGINE_CHANNEL_COUNT_CAPACITY 1024
 
#define _HICR_RPC_ENGINE_CHANNEL_BASE_TAG 0xF0000000
 
namespace HiCR::frontend
{
 
class RPCEngine
{
  public:
 
  using RPCTargetIndex_t = uint64_t;
 
  using RPCArgument_t = uint64_t;
 
  struct RPCPayload_t
  {
    RPCTargetIndex_t index;
 
    RPCArgument_t argument;
  };
 
  struct RPCTarget_t
  {
    std::string name;
 
    std::shared_ptr<HiCR::ExecutionUnit> executionUnit;
  };
 
  RPCEngine(CommunicationManager            &communicationManager,
            InstanceManager                 &instanceManager,
            MemoryManager                   &memoryManager,
            ComputeManager                  &computeManager,
            std::shared_ptr<MemorySpace>     bufferMemorySpace,
            std::shared_ptr<ComputeResource> computeResource,
            const uint64_t                   baseTag = _HICR_RPC_ENGINE_CHANNEL_BASE_TAG)
    : _communicationManager(communicationManager),
      _instanceManager(instanceManager),
      _memoryManager(memoryManager),
      _computeManager(computeManager),
      _bufferMemorySpace(bufferMemorySpace),
      _computeResource(computeResource),
      _baseTag(baseTag)
  {}
 
  __INLINE__ void initialize()
  {
    // Creating MPSC channels to receive RPC requests
    initializeRPCChannels();
    initializeReturnValueChannels();
  }
 
  ~RPCEngine() = default;
 
  __INLINE__ void addRPCTarget(const std::string &RPCName, const std::shared_ptr<HiCR::ExecutionUnit> e)
  {
    // Obtaining hash from the RPC name
    const auto idx = getRPCTargetIndexFromString(RPCName);
 
    // Inserting the new entry
    _RPCTargetMap[idx] = {RPCName, e};
  }
 
  __INLINE__ bool hasPendingRPCs()
  {
    // Updating depth
    _RPCConsumerChannel->updateDepth();
 
    // Checking if empty
    return _RPCConsumerChannel->isEmpty() == false;
  }
 
  __INLINE__ void listen()
  {
    // Keep querying the channel until a new message arrives
    while (_RPCConsumerChannel->getDepth() == 0) _RPCConsumerChannel->updateDepth();
 
    // Execute the RPC
    parseAndExecuteRPC();
  }
 
  __INLINE__ bool tryListen()
  {
    // If there are messages to process early return
    if (_RPCConsumerChannel->getDepth() > 0) return true;
 
    // If the channel is empty, check if new messages arrived since last check
    _RPCConsumerChannel->updateDepth();
 
    // Return whether there are new messages
    return _RPCConsumerChannel->getDepth() > 0;
  }
 
  __INLINE__ void parseAndExecuteRPC()
  {
    // Once a request has arrived, gather its value from the channel
    auto          request   = _RPCConsumerChannel->peek();
    auto          requester = request[0];
    RPCPayload_t *buffer    = (RPCPayload_t *)(_RPCConsumerChannel->getTokenBuffers()[requester]->getSourceLocalMemorySlot()->getPointer());
    RPCPayload_t  payload   = buffer[request[1]];
    _RPCConsumerChannel->pop();
 
    // Setting requester instance index
    _requesterInstanceIdx = requester;
 
    // Storing rpc argument
    _currentRPCArgument = payload.argument;
 
    // Execute RPC
    executeRPC(payload.index);
  }
 
  __INLINE__ std::shared_ptr<HiCR::Instance> getRPCRequester() { return _instanceManager.getInstances()[_requesterInstanceIdx]; }
 
  [[nodiscard]] __INLINE__ const RPCArgument_t getRPCArgument() { return _currentRPCArgument; }
 
  virtual void requestRPC(HiCR::Instance::instanceId_t targetInstanceId, const std::string &RPCName, const HiCR::frontend::RPCEngine::RPCArgument_t argument = 0)
  {
    // Creating message payload
    RPCPayload_t RPCPayload;
    RPCPayload.index    = getRPCTargetIndexFromString(RPCName);
    RPCPayload.argument = argument;
 
    // Registering source buffer
    auto tempBufferSlot = _memoryManager.registerLocalMemorySlot(_bufferMemorySpace, (void *)&RPCPayload, sizeof(RPCPayload_t));
 
    // Sending source buffer
    _RPCProducerChannels.at(targetInstanceId)->push(tempBufferSlot);
  }
 
  __INLINE__ void submitReturnValue(void *pointer, const size_t size)
  {
    // Getting source buffers
    auto tempBufferSlot   = _memoryManager.allocateLocalMemorySlot(_bufferMemorySpace, size);
    auto sourceBufferSlot = _memoryManager.registerLocalMemorySlot(_bufferMemorySpace, pointer, size);
 
    // Copying data
    _communicationManager.memcpy(tempBufferSlot, 0, sourceBufferSlot, 0, size);
 
    // Waiting for communication to end
    _communicationManager.fence(tempBufferSlot, 1, 0);
 
    // Sending return value data
    _returnValueProducerChannels.at(_requesterInstanceIdx)->push(tempBufferSlot);
 
    // Freeing up local memory slot
    _memoryManager.freeLocalMemorySlot(tempBufferSlot);
  }
 
  __INLINE__ std::shared_ptr<HiCR::LocalMemorySlot> getReturnValue() const
  {
    // Calling the backend-specific implementation of the listen function
    while (_returnValueConsumerChannel->isEmpty());
 
    // Calling backend-specific implementation of this function
    auto returnValue = _returnValueConsumerChannel->peek();
 
    // Getting message info
    auto msgOffset = returnValue[0];
    auto msgSize   = returnValue[1];
 
    // Creating local buffer
    auto tempBufferSlot = _memoryManager.allocateLocalMemorySlot(_bufferMemorySpace, msgSize);
 
    // Copying data
    _communicationManager.memcpy(tempBufferSlot, 0, _returnValueConsumerChannel->getPayloadBufferMemorySlot(), msgOffset, msgSize);
 
    // Waiting for communication to end
    _communicationManager.fence(tempBufferSlot, 1, 0);
 
    // Freeing up channel
    _returnValueConsumerChannel->pop();
 
    // Returning internal buffer
    return tempBufferSlot;
  }
 
  [[nodiscard]] __INLINE__ HiCR::CommunicationManager *getCommunicationManager() const { return &_communicationManager; }
 
  [[nodiscard]] __INLINE__ HiCR::InstanceManager *getInstanceManager() const { return &_instanceManager; }
 
  [[nodiscard]] __INLINE__ HiCR::MemoryManager *getMemoryManager() const { return &_memoryManager; }
 
  [[nodiscard]] __INLINE__ HiCR::ComputeManager *getComputeManager() const { return &_computeManager; }
 
  private:
 
  static RPCTargetIndex_t getRPCTargetIndexFromString(const std::string &name) { return std::hash<std::string>()(name); }
 
  __INLINE__ void executeRPC(const RPCTargetIndex_t rpcIdx) const
  {
    // Getting RPC target from the index
    if (_RPCTargetMap.contains(rpcIdx) == false) HICR_THROW_RUNTIME("Attempting to run an RPC target (Hash: %lu) that was not defined in this instance (0x%lX).\n", rpcIdx, this);
    auto &target = _RPCTargetMap.at(rpcIdx);
    // auto &targetName          = target.name;
    auto &targetExecutionUnit = target.executionUnit;
 
    // printf("Running: %s\n", targetName.c_str());
 
    // Creating new processing unit to execute the RPC
    auto p = _computeManager.createProcessingUnit(_computeResource);
    _computeManager.initialize(p);
 
    // Creating execution state
    auto s = _computeManager.createExecutionState(targetExecutionUnit);
 
    // Executing RPC
    _computeManager.start(p, s);
 
    // Waiting for execution to finish
    _computeManager.await(p);
  }
 
  __INLINE__ void initializeRPCChannels()
  {
    // Defining tag values
    const uint64_t _HICR_RPC_ENGINE_CHANNEL_CONSUMER_TOKEN_BUFFER_TAG        = _baseTag + 4;
    const uint64_t _HICR_RPC_ENGINE_CHANNEL_PRODUCER_COORDINATION_BUFFER_TAG = _baseTag + 5;
    const uint64_t _HICR_RPC_ENGINE_CHANNEL_CONSUMER_COORDINATION_BUFFER_TAG = _baseTag + 6;
 
    // Getting required buffer sizes
    auto tokenSize = sizeof(RPCPayload_t);
 
    // Getting required buffer sizes
    auto tokenBufferSize = HiCR::channel::fixedSize::Base::getTokenBufferSize(tokenSize, _HICR_RPC_ENGINE_CHANNEL_COUNT_CAPACITY);
 
    // Getting my current instance
    const auto currentInstanceId = _instanceManager.getCurrentInstance()->getId();
 
    // Getting total instance count
    const auto instanceCount = _instanceManager.getInstances().size();
 
    // Creating and exchanging buffers
 
    std::vector<HiCR::CommunicationManager::globalKeyMemorySlotPair_t> tokenBuffers;
    std::vector<HiCR::CommunicationManager::globalKeyMemorySlotPair_t> consumerCoordinationBuffers;
    std::vector<HiCR::CommunicationManager::globalKeyMemorySlotPair_t> producerCoordinationBuffers;
    std::vector<std::shared_ptr<HiCR::LocalMemorySlot>>                localConsumerCoordinationBuffers;
    std::vector<std::shared_ptr<HiCR::LocalMemorySlot>>                localProducerCoordinationBuffers;
 
    auto coordinationBufferSize = HiCR::channel::fixedSize::Base::getCoordinationBufferSize();
 
    for (size_t i = 0; i < instanceCount; i++)
    {
      // Calculating these particular slots' key
      const HiCR::GlobalMemorySlot::globalKey_t slotkey = currentInstanceId * instanceCount + i;
 
      // consumer needs to allocate #producers token buffers for #producers SPSCs
      auto tokenBufferSlot = _memoryManager.allocateLocalMemorySlot(_bufferMemorySpace, tokenBufferSize);
 
      // consumer needs to allocate #producers consumer side coordination buffers for #producers SPSCs
      auto consumerCoordinationBuffer = _memoryManager.allocateLocalMemorySlot(_bufferMemorySpace, coordinationBufferSize);
      HiCR::channel::fixedSize::Base::initializeCoordinationBuffer(consumerCoordinationBuffer);
 
      // producer needs to allocate #consumers producer-side coordination buffers for #consumer SPSCs
      auto producerCoordinationBuffer = _memoryManager.allocateLocalMemorySlot(_bufferMemorySpace, coordinationBufferSize);
      HiCR::channel::fixedSize::Base::initializeCoordinationBuffer(producerCoordinationBuffer);
 
      // Adding to collections
      localConsumerCoordinationBuffers.push_back(consumerCoordinationBuffer);
      localProducerCoordinationBuffers.push_back(producerCoordinationBuffer);
      tokenBuffers.push_back(std::make_pair(slotkey, tokenBufferSlot));
      consumerCoordinationBuffers.push_back(std::make_pair(slotkey, consumerCoordinationBuffer));
      producerCoordinationBuffers.push_back(std::make_pair(slotkey, producerCoordinationBuffer));
    }
 
    // communicate to producers the token buffer references
    _communicationManager.exchangeGlobalMemorySlots(_HICR_RPC_ENGINE_CHANNEL_CONSUMER_TOKEN_BUFFER_TAG, tokenBuffers);
    _communicationManager.fence(_HICR_RPC_ENGINE_CHANNEL_CONSUMER_TOKEN_BUFFER_TAG);
 
    // get from producers their coordination buffer references
    _communicationManager.exchangeGlobalMemorySlots(_HICR_RPC_ENGINE_CHANNEL_PRODUCER_COORDINATION_BUFFER_TAG, producerCoordinationBuffers);
    _communicationManager.fence(_HICR_RPC_ENGINE_CHANNEL_PRODUCER_COORDINATION_BUFFER_TAG);
 
    // communicate to producers the consumer buffers
    _communicationManager.exchangeGlobalMemorySlots(_HICR_RPC_ENGINE_CHANNEL_CONSUMER_COORDINATION_BUFFER_TAG, consumerCoordinationBuffers);
    _communicationManager.fence(_HICR_RPC_ENGINE_CHANNEL_CONSUMER_COORDINATION_BUFFER_TAG);
 
    {
      std::vector<std::shared_ptr<HiCR::GlobalMemorySlot>> globalConsumerTokenBuffers;
      std::vector<std::shared_ptr<HiCR::GlobalMemorySlot>> globalConsumerCoordinationBuffers;
      std::vector<std::shared_ptr<HiCR::GlobalMemorySlot>> globalProducerCoordinationBuffers;
 
      for (size_t i = 0; i < instanceCount; i++)
      {
        // Calculating these particular slots' key
        const HiCR::GlobalMemorySlot::globalKey_t localSlotKey  = currentInstanceId * instanceCount + i;
        const HiCR::GlobalMemorySlot::globalKey_t remoteSlotKey = i * instanceCount + currentInstanceId;
 
        auto globalConsumerTokenBufferSlot = _communicationManager.getGlobalMemorySlot(_HICR_RPC_ENGINE_CHANNEL_CONSUMER_TOKEN_BUFFER_TAG, localSlotKey);
        globalConsumerTokenBuffers.push_back(globalConsumerTokenBufferSlot);
 
        auto producerCoordinationBuffer = _communicationManager.getGlobalMemorySlot(_HICR_RPC_ENGINE_CHANNEL_PRODUCER_COORDINATION_BUFFER_TAG, remoteSlotKey);
        globalProducerCoordinationBuffers.push_back(producerCoordinationBuffer);
 
        auto consumerCoordinationBuffer = _communicationManager.getGlobalMemorySlot(_HICR_RPC_ENGINE_CHANNEL_CONSUMER_COORDINATION_BUFFER_TAG, localSlotKey);
        globalConsumerCoordinationBuffers.push_back(consumerCoordinationBuffer);
      }
 
      _RPCConsumerChannel = std::make_shared<HiCR::channel::fixedSize::MPSC::nonlocking::Consumer>(_communicationManager,
                                                                                                   _communicationManager,
                                                                                                   globalConsumerTokenBuffers,
                                                                                                   localConsumerCoordinationBuffers,
                                                                                                   globalProducerCoordinationBuffers,
                                                                                                   tokenSize,
                                                                                                   _HICR_RPC_ENGINE_CHANNEL_COUNT_CAPACITY);
    }
 
 
    {
      std::vector<std::shared_ptr<HiCR::GlobalMemorySlot>> globalConsumerTokenBuffers;
      std::vector<std::shared_ptr<HiCR::GlobalMemorySlot>> globalConsumerCoordinationBuffers;
      std::vector<std::shared_ptr<HiCR::GlobalMemorySlot>> globalProducerCoordinationBuffers;
 
      for (size_t i = 0; i < instanceCount; i++)
      {
        // Calculating these particular slots' key
        const HiCR::GlobalMemorySlot::globalKey_t localSlotKey  = currentInstanceId * instanceCount + i;
        const HiCR::GlobalMemorySlot::globalKey_t remoteSlotKey = i * instanceCount + currentInstanceId;
 
        auto globalConsumerTokenBufferSlot = _communicationManager.getGlobalMemorySlot(_HICR_RPC_ENGINE_CHANNEL_CONSUMER_TOKEN_BUFFER_TAG, remoteSlotKey);
        globalConsumerTokenBuffers.push_back(globalConsumerTokenBufferSlot);
 
        auto producerCoordinationBuffer = _communicationManager.getGlobalMemorySlot(_HICR_RPC_ENGINE_CHANNEL_PRODUCER_COORDINATION_BUFFER_TAG, localSlotKey);
        globalProducerCoordinationBuffers.push_back(producerCoordinationBuffer);
 
        auto consumerCoordinationBuffer = _communicationManager.getGlobalMemorySlot(_HICR_RPC_ENGINE_CHANNEL_CONSUMER_COORDINATION_BUFFER_TAG, remoteSlotKey);
        globalConsumerCoordinationBuffers.push_back(consumerCoordinationBuffer);
      }
 
      for (size_t i = 0; i < instanceCount; i++)
      {
        // Getting consumer instance id
        const auto consumerInstanceId = _instanceManager.getInstances()[i]->getId();
 
        // Creating producer channel
        // This call does the same as the SPSC Producer constructor, as
        // the producer of MPSC::nonlocking has the same view
        _RPCProducerChannels[consumerInstanceId] =
          std::make_shared<HiCR::channel::fixedSize::MPSC::nonlocking::Producer>(_communicationManager,
                                                                                 _communicationManager,
                                                                                 globalConsumerTokenBuffers[i],
                                                                                 globalProducerCoordinationBuffers[i]->getSourceLocalMemorySlot(),
                                                                                 globalConsumerCoordinationBuffers[i],
                                                                                 tokenSize,
                                                                                 _HICR_RPC_ENGINE_CHANNEL_COUNT_CAPACITY);
      }
    }
  }
 
  __INLINE__ void initializeReturnValueChannels()
  {
    // Defining tag values
    const uint64_t _HICR_RPC_ENGINE_CHANNEL_CONSUMER_SIZES_BUFFER_TAG                 = _baseTag + 0;
    const uint64_t _HICR_RPC_ENGINE_CHANNEL_CONSUMER_PAYLOAD_BUFFER_TAG               = _baseTag + 1;
    const uint64_t _HICR_RPC_ENGINE_CHANNEL_CONSUMER_COORDINATION_BUFFER_SIZES_TAG    = _baseTag + 2;
    const uint64_t _HICR_RPC_ENGINE_CHANNEL_CONSUMER_COORDINATION_BUFFER_PAYLOADS_TAG = _baseTag + 3;
 
    // Getting my current instance
    const auto currentInstanceId = _instanceManager.getCurrentInstance()->getId();
 
 
    // Getting required buffer sizes
    auto tokenSizeBufferSize = HiCR::channel::variableSize::Base::getTokenBufferSize(sizeof(size_t), _HICR_RPC_ENGINE_CHANNEL_COUNT_CAPACITY);
 
    // Allocating token size buffer as a local memory slot
    auto tokenSizeBufferSlot = _memoryManager.allocateLocalMemorySlot(_bufferMemorySpace, tokenSizeBufferSize);
 
    // Allocating token size buffer as a local memory slot
    auto payloadBufferSlot = _memoryManager.allocateLocalMemorySlot(_bufferMemorySpace, _HICR_RPC_ENGINE_CHANNEL_PAYLOAD_CAPACITY);
 
    // Getting required buffer size for coordination buffers
    auto coordinationBufferSize = HiCR::channel::variableSize::Base::getCoordinationBufferSize();
 
    // Allocating coordination buffers
    auto localConsumerCoordinationBufferMessageSizes    = _memoryManager.allocateLocalMemorySlot(_bufferMemorySpace, coordinationBufferSize);
    auto localConsumerCoordinationBufferMessagePayloads = _memoryManager.allocateLocalMemorySlot(_bufferMemorySpace, coordinationBufferSize);
 
    // Initializing coordination buffers
    HiCR::channel::variableSize::Base::initializeCoordinationBuffer(localConsumerCoordinationBufferMessageSizes);
    HiCR::channel::variableSize::Base::initializeCoordinationBuffer(localConsumerCoordinationBufferMessagePayloads);
 
    // Exchanging local memory slots to become global for them to be used by the remote end
    _communicationManager.exchangeGlobalMemorySlots(_HICR_RPC_ENGINE_CHANNEL_CONSUMER_SIZES_BUFFER_TAG, {{currentInstanceId, tokenSizeBufferSlot}});
    _communicationManager.fence(_HICR_RPC_ENGINE_CHANNEL_CONSUMER_SIZES_BUFFER_TAG);
 
    _communicationManager.exchangeGlobalMemorySlots(_HICR_RPC_ENGINE_CHANNEL_CONSUMER_PAYLOAD_BUFFER_TAG, {{currentInstanceId, payloadBufferSlot}});
    _communicationManager.fence(_HICR_RPC_ENGINE_CHANNEL_CONSUMER_PAYLOAD_BUFFER_TAG);
 
    _communicationManager.exchangeGlobalMemorySlots(_HICR_RPC_ENGINE_CHANNEL_CONSUMER_COORDINATION_BUFFER_SIZES_TAG,
                                                    {{currentInstanceId, localConsumerCoordinationBufferMessageSizes}});
    _communicationManager.fence(_HICR_RPC_ENGINE_CHANNEL_CONSUMER_COORDINATION_BUFFER_SIZES_TAG);
 
    _communicationManager.exchangeGlobalMemorySlots(_HICR_RPC_ENGINE_CHANNEL_CONSUMER_COORDINATION_BUFFER_PAYLOADS_TAG,
                                                    {{currentInstanceId, localConsumerCoordinationBufferMessagePayloads}});
    _communicationManager.fence(_HICR_RPC_ENGINE_CHANNEL_CONSUMER_COORDINATION_BUFFER_PAYLOADS_TAG);
 
    // Obtaining the globally exchanged memory slots
    auto consumerMessagePayloadBuffer      = _communicationManager.getGlobalMemorySlot(_HICR_RPC_ENGINE_CHANNEL_CONSUMER_PAYLOAD_BUFFER_TAG, currentInstanceId);
    auto consumerMessageSizesBuffer        = _communicationManager.getGlobalMemorySlot(_HICR_RPC_ENGINE_CHANNEL_CONSUMER_SIZES_BUFFER_TAG, currentInstanceId);
    auto consumerCoodinationPayloadsBuffer = _communicationManager.getGlobalMemorySlot(_HICR_RPC_ENGINE_CHANNEL_CONSUMER_COORDINATION_BUFFER_PAYLOADS_TAG, currentInstanceId);
    auto consumerCoordinationSizesBuffer   = _communicationManager.getGlobalMemorySlot(_HICR_RPC_ENGINE_CHANNEL_CONSUMER_COORDINATION_BUFFER_SIZES_TAG, currentInstanceId);
 
    // Creating channel
    _returnValueConsumerChannel = std::make_shared<HiCR::channel::variableSize::MPSC::locking::Consumer>(_communicationManager,
                                                                                                         _communicationManager,
                                                                                                         consumerMessagePayloadBuffer,
                                                                                                         consumerMessageSizesBuffer,
                                                                                                         localConsumerCoordinationBufferMessageSizes,
                                                                                                         localConsumerCoordinationBufferMessagePayloads,
                                                                                                         consumerCoordinationSizesBuffer,
                                                                                                         consumerCoodinationPayloadsBuffer,
                                                                                                         _HICR_RPC_ENGINE_CHANNEL_PAYLOAD_CAPACITY,
                                                                                                         _HICR_RPC_ENGINE_CHANNEL_COUNT_CAPACITY);
 
    // Creating producer channels
    for (const auto &instance : _instanceManager.getInstances())
    {
      // Getting consumer instance id
      const auto consumerInstanceId = instance->getId();
 
      // Allocating coordination buffers
      auto localProducerSizeInfoBuffer                    = _memoryManager.allocateLocalMemorySlot(_bufferMemorySpace, sizeof(size_t));
      auto localProducerCoordinationBufferMessageSizes    = _memoryManager.allocateLocalMemorySlot(_bufferMemorySpace, coordinationBufferSize);
      auto localProducerCoordinationBufferMessagePayloads = _memoryManager.allocateLocalMemorySlot(_bufferMemorySpace, coordinationBufferSize);
 
      // Initializing coordination buffers
      HiCR::channel::variableSize::Base::initializeCoordinationBuffer(localProducerCoordinationBufferMessageSizes);
      HiCR::channel::variableSize::Base::initializeCoordinationBuffer(localProducerCoordinationBufferMessagePayloads);
 
      // Obtaining the globally exchanged memory slots
      auto consumerMessagePayloadBuffer  = _communicationManager.getGlobalMemorySlot(_HICR_RPC_ENGINE_CHANNEL_CONSUMER_PAYLOAD_BUFFER_TAG, consumerInstanceId);
      auto consumerMessageSizesBuffer    = _communicationManager.getGlobalMemorySlot(_HICR_RPC_ENGINE_CHANNEL_CONSUMER_SIZES_BUFFER_TAG, consumerInstanceId);
      auto consumerPayloadConsumerBuffer = _communicationManager.getGlobalMemorySlot(_HICR_RPC_ENGINE_CHANNEL_CONSUMER_COORDINATION_BUFFER_PAYLOADS_TAG, consumerInstanceId);
      auto consumerSizesConsumerBuffer   = _communicationManager.getGlobalMemorySlot(_HICR_RPC_ENGINE_CHANNEL_CONSUMER_COORDINATION_BUFFER_SIZES_TAG, consumerInstanceId);
 
      // Creating channel
      _returnValueProducerChannels[consumerInstanceId] = std::make_shared<HiCR::channel::variableSize::MPSC::locking::Producer>(_communicationManager,
                                                                                                                                _communicationManager,
                                                                                                                                localProducerSizeInfoBuffer,
                                                                                                                                consumerMessagePayloadBuffer,
                                                                                                                                consumerMessageSizesBuffer,
                                                                                                                                localProducerCoordinationBufferMessageSizes,
                                                                                                                                localProducerCoordinationBufferMessagePayloads,
                                                                                                                                consumerSizesConsumerBuffer,
                                                                                                                                consumerPayloadConsumerBuffer,
                                                                                                                                _HICR_RPC_ENGINE_CHANNEL_PAYLOAD_CAPACITY,
                                                                                                                                sizeof(uint8_t),
                                                                                                                                _HICR_RPC_ENGINE_CHANNEL_COUNT_CAPACITY);
    }
  }
 
  CommunicationManager &_communicationManager;
 
  InstanceManager &_instanceManager;
 
  MemoryManager &_memoryManager;
 
  ComputeManager &_computeManager;
 
  const std::shared_ptr<HiCR::MemorySpace> _bufferMemorySpace;
 
  const std::shared_ptr<HiCR::ComputeResource> _computeResource;
 
  const uint64_t _baseTag;
 
  size_t        _requesterInstanceIdx;
  RPCArgument_t _currentRPCArgument;
 
  std::shared_ptr<HiCR::channel::variableSize::MPSC::locking::Consumer> _returnValueConsumerChannel;
 
  std::map<HiCR::Instance::instanceId_t, std::shared_ptr<HiCR::channel::variableSize::MPSC::locking::Producer>> _returnValueProducerChannels;
 
  std::shared_ptr<HiCR::channel::fixedSize::MPSC::nonlocking::Consumer> _RPCConsumerChannel;
 
  std::map<HiCR::Instance::instanceId_t, std::shared_ptr<HiCR::channel::fixedSize::MPSC::nonlocking::Producer>> _RPCProducerChannels;
 
  std::map<RPCTargetIndex_t, RPCTarget_t> _RPCTargetMap;
};
 
} // namespace HiCR::frontend