Merge pull request #48 from Algebraic-Programming/feat/channelExcessBuffer

Feat: channel excess buffer

Variable size SPSC channel:
- the payload is not split into chunks. We require the user to provide a bigger buffer (ideally double the capacity of the channel), and we use part of it to store entire tokens. The channel logic remains unchanged and it allows you to push only MAX_TOKENS * TOKEN_SIZE
- channels take into account both how many tokens are in the channel and the available space
- extended the isFull function to take a size_t and compute whether a push with such size can be made.
- add MPI tests for channel testing the following:
         - a single push that fills the payload buffer
         - a series of push that fills the payload buffer
         - a series of push that fills the coordination buffer
         - push-pop-push with a buffer big enough to force the second push to use the excess buffer

Author: lterrac

examples/channels/variableSize/mpsc/nonlocking/include/producer.hpp

@@ -129,7 +129,7 @@ void producerFc(HiCR::MemoryManager               &coordinationMemoryManager,
     // is the message sizes buffer full?
     // Note: it might be necessary sometimes to also check the payload buffers
     // as they might overflow independently of the message size buffers
-    while (producer.isFull()) { producer.updateDepth(); }
+    while (producer.isFull(nextElemSize)) { producer.updateDepth(); }
 
     producer.push(sendSlot);
     Printer<ELEMENT_TYPE>::printBytes(prefix, elements[i].first, payloadCapacity, 0, nextElemSize);

examples/channels/variableSize/spsc/include/consumer.hpp

@@ -36,7 +36,7 @@ void consumerFc(HiCR::MemoryManager               &coordinationMemoryManager,
   auto sizesBufferSlot = coordinationMemoryManager.allocateLocalMemorySlot(coordinationMemorySpace, sizesBufferSize);
 
   // Allocating payload buffer as a local memory slot
-  auto payloadBufferSlot = payloadMemoryManager.allocateLocalMemorySlot(payloadMemorySpace, PAYLOAD_CAPACITY);
+  auto payloadBufferSlot = payloadMemoryManager.allocateLocalMemorySlot(payloadMemorySpace, PAYLOAD_CAPACITY * 2);
 
   // Getting required buffer size
   auto coordinationBufferSize = HiCR::channel::variableSize::Base::getCoordinationBufferSize();
@@ -86,7 +86,7 @@ void consumerFc(HiCR::MemoryManager               &coordinationMemoryManager,
                                                               channelCapacity);
 
   // Getting a single value from the channel
-  while (consumer.getDepth() != 1) consumer.updateDepth();
+  while (consumer.getCoordinationDepth() != 1) consumer.updateDepth();
 
   // Getting internal pointer of the token buffer slot
   auto payloadBufferPtr = (ELEMENT_TYPE *)payloadBufferSlot->getPointer();

include/hicr/frontends/channel/variableSize/mpsc/nonlocking/consumer.hpp

@@ -104,7 +104,7 @@ class Consumer
       * immediately upon creation of the SPSC channel. Therefore we do not reset
       * _depths to zero, and check for "early" received messages
       */
-      _depths.push_back(consumerPtr->getDepth());
+      _depths.push_back(consumerPtr->getCoordinationDepth());
       for (size_t j = 0; j < _depths.back(); j++) { _channelPushes.push(i); }
     }
   }
@@ -222,7 +222,7 @@ class Consumer
     for (size_t i = 0; i < _spscList.size(); i++)
     {
       _spscList[i]->updateDepth();
-      newDepths[i] = _spscList[i]->getDepth();
+      newDepths[i] = _spscList[i]->getCoordinationDepth();
     }
 
     for (size_t i = 0; i < _spscList.size(); i++)

include/hicr/frontends/channel/variableSize/spsc/consumer.hpp

@@ -49,9 +49,9 @@ class Consumer final : public variableSize::Base
    *
    * \param[in] coordinationCommunicationManager The backend's memory manager to facilitate communication between the producer and consumer coordination buffers
    * \param[in] payloadCommunicationManager The backend's memory manager to facilitate communication between the producer and consumer payload buffers
-   * \param[in] payloadBuffer The memory slot pertaining to the payload buffer. The producer will push new tokens
-   *            into this buffer, while there is enough space (in bytes). This buffer should be big enough to hold at least the
-   *            largest message of the variable-sized messages to be pushed.
+   * \param[in] payloadBuffer The memory slot pertaining to the payload buffer. The producer will push messages into this
+   *            buffer, while there is enough space. This buffer should be large enough to hold twice the capacity specified by payloadCapacity argument.
+   *            Half of the buffer is used as excess buffer to avoid internal fragmentation of messages
    * \param[in] tokenBuffer The memory slot pertaining to the token buffer. This buffer is only used to exchange internal metadata
    *            about the sizes of the individual messages being sent.
    * \param[in] internalCoordinationBufferForCounts This is a small buffer to hold the internal (local) state of the
@@ -62,7 +62,7 @@ class Consumer final : public variableSize::Base
    *            buffer for message counts, used for remote updates on pop()
    * \param[in] producerCoordinationBufferForPayloads A global reference to the producer channel's internal coordination
    *            buffer for payload sizes (in bytes), used for remote updates on pop()
-   * \param[in] payloadCapacity The capacity (in bytes) of the buffer for variable-sized messages
+   * \param[in] payloadCapacity The capacity (in bytes) of the buffer for variable-sized messages.
    * \param[in] capacity The maximum number of tokens that will be held by this channel
    * @note: The token size in var-size channels is used only internally, and is passed as having a type size_t (with size sizeof(size_t))
    */
@@ -243,23 +243,28 @@ class Consumer final : public variableSize::Base
    * receiving the message counts (phase 2), returning this depth should guarantee
    * we already have received the payloads
    *
-   * \note This is not a thread-safe call
    *
    * This is a getter function that should complete in \f$ \Theta(1) \f$ time.
    *
    * @return The number of elements in variable-size consumer channel
+   * 
+   * \note This is not a thread-safe call
+   * \note Even though there might be space for additional tokens in the coordination buffer, it is not guaranteed that 
+   *       the push() will succeed due to insufficient space in the payload buffer
    */
-  size_t getDepth() { return getCircularBufferForCounts()->getDepth(); }
+  size_t getCoordinationDepth() { return getCircularBufferForCounts()->getDepth(); }
 
   /**
    * Returns the current depth of the channel holding the payloads
    *
-   * \note This is not a thread-safe call
    *
    * This is a getter function that should complete in \f$ \Theta(1) \f$ time.
    *
    * @returns The number of total bytes in the payloads channel
    *
+   * \note This is not a thread-safe call
+   * \note Even though there might be space for additional tokens in the payload buffer, it is not guaranteed that 
+   *       the push() will succeed due to insufficient space in the coordination buffer
    */
   size_t getPayloadDepth() { return getCircularBufferForPayloads()->getDepth(); }
 
@@ -271,7 +276,28 @@ class Consumer final : public variableSize::Base
    * \returns true, if both message count and payload buffers are empty
    * \returns false, if one of the buffers is not empty
    */
-  bool isEmpty() { return (getDepth() == 0); }
+  bool isEmpty() { return getCoordinationDepth() == 0; }
+
+  /**
+   * This funciton can be used to quickly check whether the channel is becoming full when trying 
+   * to push an element of a given size. First thing, we are checking if we can still 
+   * push tokens (i.e., if the coordination buffer has space). Second thing, we are checking the
+   * payload buffer. If the current depth of the payload and the requiredBufferSize to push 
+   * exceed the channel capacity, the channel is considered full.
+   * 
+   * \param[in] requiredBufferSize size of the token to push into the channel
+   * 
+   * \return true if there is enough space to push the token, false otherwise
+  */
+  bool isFull(size_t requiredBufferSize)
+  {
+    auto coordinationCircularBuffer = getCircularBufferForCounts();
+    if (coordinationCircularBuffer->getDepth() == coordinationCircularBuffer->getCapacity()) return true;
+    auto payloadCircularBuffer = getCircularBufferForPayloads();
+    if (payloadCircularBuffer->getDepth() + requiredBufferSize > payloadCircularBuffer->getCapacity()) return true;
+
+    return false;
+  }
 
   /**
    * Retrieves the pointer to the channel's payload buffer