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::MPSC::locking
{
 
class Consumer final : public channel::fixedSize::Base
{
  private:
 
  const std::shared_ptr<HiCR::GlobalMemorySlot> _tokenBuffer;
 
  /*
   * Global Memory slot pointing to the consumer's coordination buffer for acquiring a lock and updating
   */
  const std::shared_ptr<HiCR::GlobalMemorySlot> _consumerCoordinationBuffer;
 
  public:
 
  Consumer(CommunicationManager                   &coordinationCommunicationManager,
           CommunicationManager                   &payloadCommunicationManager,
           std::shared_ptr<GlobalMemorySlot>       tokenBuffer,
           const std::shared_ptr<LocalMemorySlot> &internalCoordinationBuffer,
           std::shared_ptr<GlobalMemorySlot>       consumerCoordinationBuffer,
           const size_t                            tokenSize,
           const size_t                            capacity)
    : channel::fixedSize::Base(coordinationCommunicationManager, payloadCommunicationManager, internalCoordinationBuffer, tokenSize, capacity),
      _tokenBuffer(std::move(tokenBuffer)),
      _consumerCoordinationBuffer(std::move(consumerCoordinationBuffer))
  {}
  ~Consumer() = default;
 
  __INLINE__ ssize_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 (token number %lu when starting from zero), which is beyond than the channel capacity (%lu)",
                       pos,
                       pos + 1,
                       getCircularBuffer()->getCapacity());
 
    // Value to return, initially set as -1 as default (not able to find the requested value)
    ssize_t bufferPos = -1;
 
    auto coordinationCommunicationManager = getCoordinationCommunicationManager();
 
    // Obtaining coordination buffer slot lock
    if (coordinationCommunicationManager->acquireGlobalLock(_consumerCoordinationBuffer) == false) return bufferPos;
 
    coordinationCommunicationManager->flushReceived();
    getPayloadCommunicationManager()->flushReceived();
    // Calculating current channel depth
    const auto curDepth = getDepth();
 
    // Calculating buffer position, if there are enough tokens in the buffer to satisfy the request
    if (pos < curDepth) bufferPos = (ssize_t)((getCircularBuffer()->getTailPosition() + pos) % getCircularBuffer()->getCapacity());
 
    // Releasing coordination buffer slot lock
    coordinationCommunicationManager->releaseGlobalLock(_consumerCoordinationBuffer);
 
    // Succeeded in pushing the token(s)
    return bufferPos;
  }
 
  __INLINE__ bool 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());
 
    // Flag to indicate whether the operaton was successful
    bool successFlag = false;
 
    auto coordinationCommunicationManager = getCoordinationCommunicationManager();
 
    // Obtaining coordination buffer slot lock
    if (coordinationCommunicationManager->acquireGlobalLock(_consumerCoordinationBuffer) == false) return successFlag;
 
    // If the exchange buffer does not have n tokens pushed, reject operation, otherwise succeed
    if (n <= getDepth())
    {
      // Advancing tail (removes elements from the circular buffer)
      getCircularBuffer()->advanceTail(n);
 
      // Setting success flag
      successFlag = true;
    }
 
    // Releasing coordination buffer slot lock
    coordinationCommunicationManager->releaseGlobalLock(_consumerCoordinationBuffer);
 
    // Operation was successful
    return successFlag;
  }
 
  [[nodiscard]] __INLINE__ std::shared_ptr<GlobalMemorySlot> getTokenBuffer() const { return _tokenBuffer; }
};
 
} // namespace HiCR::channel::fixedSize::MPSC::locking