consumer.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/definitions.hpp>
#include <hicr/core/exceptions.hpp>
#include <hicr/frontends/channel/fixedSize/base.hpp>
#include <utility>
 
namespace HiCR::channel::fixedSize::SPSC
{
 
class Consumer final : public channel::fixedSize::Base
{
  private:
 
  const std::shared_ptr<HiCR::GlobalMemorySlot> _tokenBuffer;
 
  const std::shared_ptr<HiCR::GlobalMemorySlot> _producerCoordinationBuffer;
 
  public:
 
  Consumer(CommunicationManager                    &communicationManager,
           const std::shared_ptr<GlobalMemorySlot> &tokenBuffer,
           const std::shared_ptr<LocalMemorySlot>  &internalCoordinationBuffer,
           std::shared_ptr<GlobalMemorySlot>        producerCoordinationBuffer,
           const size_t                             tokenSize,
           const size_t                             capacity)
    : channel::fixedSize::Base(communicationManager, internalCoordinationBuffer, tokenSize, capacity),
      _tokenBuffer(tokenBuffer),
      _producerCoordinationBuffer(std::move(producerCoordinationBuffer))
 
  {
    // Checking whether the memory slot is local. This backend only supports local data transfers
    if (tokenBuffer->getSourceLocalMemorySlot() == nullptr)
      HICR_THROW_LOGIC("The passed token buffer slot was not created locally (it must be to be used internally by the channel implementation)\n");
 
    // Checking that the provided token exchange  buffer has the right size
    auto requiredTokenBufferSize = getTokenBufferSize(getTokenSize(), capacity);
    auto providedTokenBufferSize = tokenBuffer->getSourceLocalMemorySlot()->getSize();
    if (providedTokenBufferSize < requiredTokenBufferSize)
      HICR_THROW_LOGIC(
        "Attempting to create a channel with a token data buffer size (%lu) smaller than the required size (%lu).\n", providedTokenBufferSize, requiredTokenBufferSize);
  }
  ~Consumer() = default;
 
  __INLINE__ size_t peek(const size_t pos = 0)
  {
    // Check if the requested position exceeds the capacity of the channel
    if (pos >= getCircularBuffer()->getCapacity())
      HICR_THROW_LOGIC("Attempting to peek for a token with position (%lu), which is beyond than the channel capacity (%lu)", pos, getCircularBuffer()->getCapacity());
 
    // Make sure receiver queues are occasionally processed
    getCommunicationManager()->flushReceived();
 
    // Updating channel depth
    updateDepth();
 
    // Check if there are enough tokens in the buffer to satisfy the request
    if (pos >= getCircularBuffer()->getDepth())
      HICR_THROW_RUNTIME("Attempting to peek position (%lu) but not enough tokens (%lu) are in the buffer", pos, getCircularBuffer()->getDepth());
 
    // Calculating buffer position
    const size_t bufferPos = (getCircularBuffer()->getTailPosition() + pos) % getCircularBuffer()->getCapacity();
 
    // Succeeded in pushing the token(s)
    return bufferPos;
  }
 
  __INLINE__ void pop(const size_t n = 1)
  {
    if (n > getCircularBuffer()->getCapacity())
      HICR_THROW_LOGIC("Attempting to pop (%lu) tokens, which is larger than the channel capacity (%lu)", n, getCircularBuffer()->getCapacity());
 
    // Updating channel depth
    updateDepth();
 
    // If the exchange buffer does not have n tokens pushed, reject operation
    if (n > getCircularBuffer()->getDepth())
      HICR_THROW_RUNTIME("Attempting to pop (%lu) tokens, which is more than the number of current tokens in the channel (%lu)", n, getCircularBuffer()->getDepth());
 
    // Advancing tail (removes elements from the circular buffer)
    getCircularBuffer()->advanceTail(n);
 
    const auto coordBuffElemSize = sizeof(_HICR_CHANNEL_COORDINATION_BUFFER_ELEMENT_TYPE);
    // Notifying producer(s) of buffer liberation
    getCommunicationManager()->memcpy(_producerCoordinationBuffer,
                                      _HICR_CHANNEL_TAIL_ADVANCE_COUNT_IDX * coordBuffElemSize,
                                      getCoordinationBuffer(),
                                      _HICR_CHANNEL_TAIL_ADVANCE_COUNT_IDX * coordBuffElemSize,
                                      coordBuffElemSize);
    getCommunicationManager()->fence(getCoordinationBuffer(), 1, 0);
  }
 
  __INLINE__ void updateDepth() {}
 
  [[nodiscard]] __INLINE__ std::shared_ptr<GlobalMemorySlot> getTokenBuffer() const { return _tokenBuffer; }
};
 
} // namespace HiCR::channel::fixedSize::SPSC