producer.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 Producer final : public fixedSize::Base
{
  private:
 
  const std::shared_ptr<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:
 
  Producer(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)
    : fixedSize::Base(coordinationCommunicationManager, payloadCommunicationManager, internalCoordinationBuffer, tokenSize, capacity),
      _tokenBuffer(std::move(tokenBuffer)),
      _consumerCoordinationBuffer(std::move(consumerCoordinationBuffer))
  {}
  ~Producer() = default;
 
  __INLINE__ bool push(const std::shared_ptr<LocalMemorySlot> &sourceSlot, const size_t n = 1)
  {
    // Make sure source slot is big enough to satisfy the operation
    auto requiredBufferSize = getTokenSize() * n;
    auto providedBufferSize = sourceSlot->getSize();
    if (providedBufferSize < requiredBufferSize)
      HICR_THROW_LOGIC("Attempting to push with a source buffer size (%lu) smaller than the required size (Token Size (%lu) x n (%lu) = %lu).\n",
                       providedBufferSize,
                       getTokenSize(),
                       n,
                       requiredBufferSize);
 
    // Flag to record whether the operation was successful or not (it simplifies code by releasing locks only once)
    bool successFlag = false;
 
    auto coordinationCommunicationManager = getCoordinationCommunicationManager();
 
    // Locking remote token and coordination buffer slots
    if (coordinationCommunicationManager->acquireGlobalLock(_consumerCoordinationBuffer) == false) return successFlag;
 
    // Updating local coordination buffer
    coordinationCommunicationManager->memcpy(getCoordinationBuffer(), 0, _consumerCoordinationBuffer, 0, getCoordinationBufferSize());
 
    // Adding fence operation to ensure buffers are ready for re-use
    coordinationCommunicationManager->fence(getCoordinationBuffer(), 0, 1);
 
    // Calculating current channel depth
    const auto depth = getDepth();
 
    auto payloadCommunicationManager = getPayloadCommunicationManager();
 
    // If the exchange buffer does not have n free slots, reject the operation
    if (depth + n <= getCircularBuffer()->getCapacity())
    {
      // Copying with source increasing offset per token
      for (size_t i = 0; i < n; i++)
      {
        payloadCommunicationManager->memcpy(_tokenBuffer,                                            /* destination */
                                            getTokenSize() * getCircularBuffer()->getHeadPosition(), /* dst_offset */
                                            sourceSlot,                                              /* source */
                                            i * getTokenSize(),                                      /* src_offset */
                                            getTokenSize());                                         /* size*/
        // Advance head here, since the memcpy relies on the up-to-date head position
        getCircularBuffer()->advanceHead(1);
      }
 
      payloadCommunicationManager->fence(sourceSlot, n, 0);
 
      // Updating global coordination buffer
      coordinationCommunicationManager->memcpy(_consumerCoordinationBuffer, 0, getCoordinationBuffer(), 0, getCoordinationBufferSize());
      // Adding fence operation to ensure buffers are ready for re-use
      coordinationCommunicationManager->fence(getCoordinationBuffer(), 1, 0);
 
      // Mark operation as successful
      successFlag = true;
    }
 
    // Releasing remote token and coordination buffer slots
    coordinationCommunicationManager->releaseGlobalLock(_consumerCoordinationBuffer);
 
    // Succeeded
    return successFlag;
  }
};
 
} // namespace HiCR::channel::fixedSize::MPSC::locking