fix(esp_ringbuf): Fixed no split buffer in case of wrap around with dummy data

Author: upolc-lg

components/esp_ringbuf/ringbuf.c

@@ -437,6 +437,7 @@ static void prvCopyItemAllowSplit(Ringbuffer_t *pxRingbuffer, const uint8_t *puc
             xItemSize -= xRemLen;
             xAlignedItemSize -= xRemLen;
             pxFirstHeader->uxItemFlags |= rbITEM_SPLIT_FLAG;        //There must be more data
+            pxFirstHeader->uxItemFlags |= rbITEM_WRITTEN_FLAG;
         } else {
             //Remaining length was only large enough to fit header
             pxFirstHeader->uxItemFlags |= rbITEM_DUMMY_DATA_FLAG;   //Item will completely be stored in 2nd part
@@ -450,6 +451,7 @@ static void prvCopyItemAllowSplit(Ringbuffer_t *pxRingbuffer, const uint8_t *puc
     pxSecondHeader->uxItemFlags = 0;
     pxRingbuffer->pucAcquire += rbHEADER_SIZE;     //Advance acquire pointer past header
     memcpy(pxRingbuffer->pucAcquire, pucItem, xItemSize);
+    pxSecondHeader->uxItemFlags |= rbITEM_WRITTEN_FLAG;
     pxRingbuffer->xItemsWaiting++;
     pxRingbuffer->pucAcquire += xAlignedItemSize;  //Advance pucAcquire past item to next aligned address
 
@@ -506,13 +508,6 @@ static BaseType_t prvCheckItemAvail(Ringbuffer_t *pxRingbuffer)
         return pdFALSE;     //Byte buffers do not allow multiple retrievals before return
     }
     if ((pxRingbuffer->xItemsWaiting > 0) && ((pxRingbuffer->pucRead != pxRingbuffer->pucWrite) || (pxRingbuffer->uxRingbufferFlags & rbBUFFER_FULL_FLAG))) {
-        // If the ring buffer is a no-split buffer, the read pointer must point to an item that has been written to.
-        if ((pxRingbuffer->uxRingbufferFlags & (rbBYTE_BUFFER_FLAG | rbALLOW_SPLIT_FLAG)) == 0) {
-            ItemHeader_t *pxHeader = (ItemHeader_t *)pxRingbuffer->pucRead;
-            if ((pxHeader->uxItemFlags & rbITEM_WRITTEN_FLAG) == 0) {
-                return pdFALSE;
-            }
-        }
         return pdTRUE;      //Items/data available for retrieval
     } else {
         return pdFALSE;     //No items/data available for retrieval
@@ -540,22 +535,30 @@ static void *prvGetItemDefault(Ringbuffer_t *pxRingbuffer,
         pxHeader = (ItemHeader_t *)pxRingbuffer->pucRead;
         configASSERT(pxHeader->xItemLen <= pxRingbuffer->xMaxItemSize);
     }
-    pcReturn = pxRingbuffer->pucRead + rbHEADER_SIZE;    //Get pointer to part of item containing data (point past the header)
-    if (pxHeader->xItemLen == 0) {
-        //Inclusive of pucTail for special case where item of zero length just fits at the end of the buffer
-        configASSERT(pcReturn >= pxRingbuffer->pucHead && pcReturn <= pxRingbuffer->pucTail);
-    } else {
-        //Exclusive of pucTail if length is larger than zero, pcReturn should never point to pucTail
-        configASSERT(pcReturn >= pxRingbuffer->pucHead && pcReturn < pxRingbuffer->pucTail);
-    }
-    *pxItemSize = pxHeader->xItemLen;   //Get length of item
-    pxRingbuffer->xItemsWaiting --;     //Update item count
-    *pxIsSplit = (pxHeader->uxItemFlags & rbITEM_SPLIT_FLAG) ? pdTRUE : pdFALSE;
 
-    pxRingbuffer->pucRead += rbHEADER_SIZE + rbALIGN_SIZE(pxHeader->xItemLen);   //Update pucRead
-    //Check if pucRead requires wrap around
-    if ((pxRingbuffer->pucTail - pxRingbuffer->pucRead) < rbHEADER_SIZE) {
-        pxRingbuffer->pucRead = pxRingbuffer->pucHead;
+    //In case of wrap around data might not be ready
+    if ((pxHeader->uxItemFlags & rbITEM_WRITTEN_FLAG) == 0) {
+        *pxIsSplit = pdFALSE;
+        *pxItemSize = 0;
+        pcReturn = NULL;
+    } else {
+        pcReturn = pxRingbuffer->pucRead + rbHEADER_SIZE;    //Get pointer to part of item containing data (point past the header)
+        if (pxHeader->xItemLen == 0) {
+            //Inclusive of pucTail for special case where item of zero length just fits at the end of the buffer
+            configASSERT(pcReturn >= pxRingbuffer->pucHead && pcReturn <= pxRingbuffer->pucTail);
+        } else {
+            //Exclusive of pucTail if length is larger than zero, pcReturn should never point to pucTail
+            configASSERT(pcReturn >= pxRingbuffer->pucHead && pcReturn < pxRingbuffer->pucTail);
+        }
+        *pxItemSize = pxHeader->xItemLen;   //Get length of item
+        pxRingbuffer->xItemsWaiting --;     //Update item count
+        *pxIsSplit = (pxHeader->uxItemFlags & rbITEM_SPLIT_FLAG) ? pdTRUE : pdFALSE;
+
+        pxRingbuffer->pucRead += rbHEADER_SIZE + rbALIGN_SIZE(pxHeader->xItemLen);   //Update pucRead
+        //Check if pucRead requires wrap around
+        if ((pxRingbuffer->pucTail - pxRingbuffer->pucRead) < rbHEADER_SIZE) {
+            pxRingbuffer->pucRead = pxRingbuffer->pucHead;
+        }
     }
     return (void *)pcReturn;
 }
@@ -827,13 +830,14 @@ static BaseType_t prvReceiveGeneric(Ringbuffer_t *pxRingbuffer,
     BaseType_t xReturn = pdFALSE;
     BaseType_t xExitLoop = pdFALSE;
     BaseType_t xEntryTimeSet = pdFALSE;
+    BaseType_t xSkipCheckAvail = pdFALSE;
     TimeOut_t xTimeOut;
 
     ESP_STATIC_ANALYZER_CHECK(!pvItem1 || !pvItem2 || !xItemSize1 || !xItemSize2, pdFALSE);
 
     while (xExitLoop == pdFALSE) {
         portENTER_CRITICAL(&pxRingbuffer->mux);
-        if (prvCheckItemAvail(pxRingbuffer) == pdTRUE) {
+        if ((!xSkipCheckAvail) && prvCheckItemAvail(pxRingbuffer) == pdTRUE) {
             //Item/data is available for retrieval
             BaseType_t xIsSplit = pdFALSE;
             if (pxRingbuffer->uxRingbufferFlags & rbBYTE_BUFFER_FLAG) {
@@ -853,9 +857,15 @@ static BaseType_t prvReceiveGeneric(Ringbuffer_t *pxRingbuffer,
                     *pvItem2 = NULL;
                 }
             }
-            xReturn = pdTRUE;
-            xExitLoop = pdTRUE;
-            goto loop_end;
+
+            if (*pvItem1 == NULL) {
+                xSkipCheckAvail = pdTRUE;
+                goto loop_end;
+            } else {
+                xReturn = pdTRUE;
+                xExitLoop = pdTRUE;
+                goto loop_end;
+            }
         } else if (xTicksToWait == (TickType_t) 0) {
             //No block time. Return immediately.
             xExitLoop = pdTRUE;
@@ -868,6 +878,7 @@ static BaseType_t prvReceiveGeneric(Ringbuffer_t *pxRingbuffer,
 
         if (xTaskCheckForTimeOut(&xTimeOut, &xTicksToWait) == pdFALSE) {
             //Not timed out yet. Block the current task
+            xSkipCheckAvail = pdFALSE;
             vTaskPlaceOnEventList(&pxRingbuffer->xTasksWaitingToReceive, xTicksToWait);
             portYIELD_WITHIN_API();
         } else {

components/esp_ringbuf/test_apps/main/test_ringbuf_common.c

@@ -1042,3 +1042,84 @@ TEST_CASE("Test no-split buffers always receive items in order", "[esp_ringbuf][
     // Cleanup
     vRingbufferDelete(buffer_handle);
 }
+
+/* ---------------------------- Test no-split ring buffer SendAquire and SendComplete with dummy ---------------------------
+ * The following test case tests the SendAquire and SendComplete functions of the no-split ring buffer.
+ *
+ * The test case will do the following...
+ * 1) Create a no-split ring buffer.
+ * 2) Acquire space on the buffer to send an item.
+ * 3) Send the item to the buffer.
+ * 4) Verify that the item is received correctly.
+ * 5) Acquire space on the buffer until the buffer can no longer receive a full item without wrap around.
+ * 6) Send the items out-of-order to the buffer.
+ * 7) Verify that the items are not received until the first item is sent.
+ * 8) Send the first item.
+ * 9) Verify that the items are received in the correct order.
+ */
+TEST_CASE("Test no-split buffers always receive items in order (with dummy)", "[esp_ringbuf][linux]")
+{
+    const uint8_t C_BUFFER_SIZE = 64;
+    const uint8_t C_ITEM_SIZE = 15;
+
+    // Create buffer
+    RingbufHandle_t buffer_handle = xRingbufferCreate(C_BUFFER_SIZE, RINGBUF_TYPE_NOSPLIT);
+    TEST_ASSERT_MESSAGE(buffer_handle != NULL, "Failed to create ring buffer");
+
+    // Acquire space on the buffer to send an item and write to the item
+    void *item1;
+    TEST_ASSERT_EQUAL(pdTRUE, xRingbufferSendAcquire(buffer_handle, &item1, C_ITEM_SIZE, TIMEOUT_TICKS));
+    *(uint32_t *)item1 = 0x123;
+
+    // Send the item to the buffer
+    TEST_ASSERT_EQUAL(pdTRUE, xRingbufferSendComplete(buffer_handle, item1));
+
+    // Verify that the item is received correctly
+    size_t item_size;
+    uint32_t *received_item = xRingbufferReceive(buffer_handle, &item_size, TIMEOUT_TICKS);
+    TEST_ASSERT_NOT_NULL(received_item);
+    TEST_ASSERT_EQUAL(item_size, C_ITEM_SIZE);
+    TEST_ASSERT_EQUAL(*(uint32_t *)received_item, 0x123);
+
+    // Return the space to the buffer after receiving the item
+    vRingbufferReturnItem(buffer_handle, received_item);
+
+    // At this point, the buffer should be empty
+    UBaseType_t items_waiting;
+    vRingbufferGetInfo(buffer_handle, NULL, NULL, NULL, NULL, &items_waiting);
+    TEST_ASSERT_MESSAGE(items_waiting == 0, "Incorrect items waiting");
+
+    // Acquire space on the buffer until the buffer is full
+#define MAX_NUM_ITEMS_DUMMY ( C_BUFFER_SIZE / ( C_ITEM_SIZE + ITEM_HDR_SIZE ) )
+    void *items[MAX_NUM_ITEMS_DUMMY];
+    for (int i = 0; i < MAX_NUM_ITEMS_DUMMY; i++) {
+        TEST_ASSERT_EQUAL(pdTRUE, xRingbufferSendAcquire(buffer_handle, &items[i], C_ITEM_SIZE, TIMEOUT_TICKS));
+        TEST_ASSERT_NOT_NULL(items[i]);
+        *(uint32_t *)items[i] = (0x100 + i);
+    }
+
+    // Verify that the buffer is full by attempting to acquire space for another item
+    void *another_item;
+    TEST_ASSERT_EQUAL(pdFALSE, xRingbufferSendAcquire(buffer_handle, &another_item, C_ITEM_SIZE, TIMEOUT_TICKS));
+
+    // Send the items out-of-order to the buffer. Verify that the items are not received until the first item is sent.
+    // In this case, we send the items in the reverse order until the first item is sent.
+    for (int i = MAX_NUM_ITEMS_DUMMY - 1; i > 0; i--) {
+        TEST_ASSERT_EQUAL(pdTRUE, xRingbufferSendComplete(buffer_handle, items[i]));
+        TEST_ASSERT_NULL(xRingbufferReceive(buffer_handle, &item_size, 0));
+    }
+
+    // Send the first item
+    TEST_ASSERT_EQUAL(pdTRUE, xRingbufferSendComplete(buffer_handle, items[0]));
+
+    // Verify that the items are received in the correct order
+    for (int i = 0; i < MAX_NUM_ITEMS_DUMMY; i++) {
+        received_item = xRingbufferReceive(buffer_handle, &item_size, TIMEOUT_TICKS);
+        TEST_ASSERT_NOT_NULL(received_item);
+        TEST_ASSERT_EQUAL(*(uint32_t *)received_item, (0x100 + i));
+        vRingbufferReturnItem(buffer_handle, received_item);
+    }
+
+    // Cleanup
+    vRingbufferDelete(buffer_handle);
+}