Correct Concurrency Tests

Author: pschatzmann

examples/tests/concurrency/synchBufferRTOS/synchBufferRTOS.ino renamed to examples/tests/concurrency/BufferRTOS/BufferRTOS.ino

@@ -11,21 +11,24 @@
 #include "AudioTools.h"
 #include "AudioLibs/Concurrency.h"
 
-BufferRTOS<int16_t> buffer(1024, 512);
+BufferRTOS<int16_t> buffer(512, 10);
 
 Task writeTask("write", 3000, 10, 0);
+
 Task readTask("read", 3000, 10, 1);
 
 void setup() {
   Serial.begin(115200);
-  AudioLogger::instance().begin(Serial, AudioLogger::Info);
 
   writeTask.begin([]() {
     int16_t data[512];
     for (int j = 0; j < 512; j++) {
       data[j] = j;
     }
-    buffer.writeArray(data, 512);
+    
+    size_t len = buffer.writeArray(data, 512);
+    delay(1);
+
   });
 
   readTask.begin([]() {
@@ -35,7 +38,9 @@ void setup() {
     static int16_t data[512];
 
     // read data
-    buffer.readArray(data, 512);
+    size_t read = buffer.readArray(data, 512);
+    delay(1);
+    assert(read==512);
 
     // check data
     for (int j = 0; j < 512; j++) {

examples/tests/concurrency/NBuffer/NBuffer.ino

@@ -25,7 +25,9 @@ void setup() {
     for (int j = 0; j < 512; j++) {
       data[j] = j;
     }
-    buffer.writeArray(data, 512);
+    assert(buffer.writeArray(data, 512)==512);
+    delay(1);
+
   });
 
   readTask.begin([]() {
@@ -35,7 +37,10 @@ void setup() {
     static int16_t data[512];
 
     // read data
-    buffer.readArray(data, 512);
+    size_t read = buffer.readArray(data, 512);
+    delay(1);
+    if (read==0) return;
+    assert(read==512);
 
     // check data
     for (int j = 0; j < 512; j++) {

examples/tests/concurrency/SynchronizedNBuffer/SynchronizedNBuffer.ino

@@ -0,0 +1,67 @@
+/**
+ * @file SynchonizedNBuffer.ino
+ * @author Phil Schatzmann
+ * @brief Multitask with SynchronizedBuffer using NBuffer
+ * @version 0.1
+ * @date 2022-11-25
+ *
+ * @copyright Copyright (c) 2022
+ *
+ */
+#include "AudioTools.h"
+#include "AudioLibs/Concurrency.h"
+
+SynchronizedNBuffer<int16_t> buffer(1024, 10);
+
+Task writeTask("write", 3000, 10, 0);
+
+Task readTask("read", 3000, 10, 1);
+
+void setup() {
+  Serial.begin(115200);
+
+  writeTask.begin([]() {
+    int16_t data[512];
+    for (int j = 0; j < 512; j++) {
+      data[j] = j;
+    }
+    assert(buffer.writeArray(data, 512)==512);
+    delay(1);
+
+  });
+
+  readTask.begin([]() {
+    static uint64_t start = millis();
+    static size_t total_bytes = 0;
+    static size_t errors = 0;
+    static int16_t data[512];
+
+    // read data
+    assert(buffer.readArray(data, 512)==512);
+    delay(1);
+
+    // check data
+    for (int j = 0; j < 512; j++) {
+      if (data[j] != j) errors++;
+    }
+    // calculate bytes per second
+    total_bytes += sizeof(int16_t) * 512;
+    if (total_bytes >= 1024000) {
+      uint64_t duration = millis() - start;
+      float mbps = static_cast<float>(total_bytes) / duration / 1000.0;
+
+      // print result
+      Serial.print("Mbytes per second: ");
+      Serial.print(mbps);
+      Serial.print(" with ");
+      Serial.print(errors);
+      Serial.println(" errors");
+
+      start = millis();
+      errors = 0;
+      total_bytes = 0;
+    }
+  });
+}
+
+void loop() { delay(1000); }

examples/tests/concurrency/synchNBuffer/synchNBuffer.ino

@@ -12,7 +12,7 @@
 #include "AudioLibs/Concurrency.h"
 
 audio_tools::Mutex mutex;
-NBuffer<int16_t> nbuffer(512, 8);
+NBuffer<int16_t> nbuffer(512, 10);
 SynchronizedBuffer<int16_t> buffer(nbuffer, mutex);
 
 Task writeTask("write", 3000, 10, 0);
@@ -27,7 +27,9 @@ void setup() {
     for (int j = 0; j < 512; j++) {
       data[j] = j;
     }
-    buffer.writeArray(data, 512);
+    assert(buffer.writeArray(data, 512)==512);
+    delay(1);
+
   });
 
   readTask.begin([]() {
@@ -37,7 +39,8 @@ void setup() {
     static int16_t data[512];
 
     // read data
-    buffer.readArray(data, 512);
+    assert(buffer.readArray(data, 512)==512);
+    delay(1);
 
     // check data
     for (int j = 0; j < 512; j++) {

examples/tests/concurrency/synchRingBuffer/synchRingBuffer.ino

@@ -21,21 +21,26 @@ Task readTask("read", 3000, 10, 1);
 
 void setup() {
   Serial.begin(115200);
+
   writeTask.begin([]() {
     int16_t data[512];
     for (int j = 0; j < 512; j++) {
       data[j] = j;
     }
-    buffer.writeArray(data, 512);
+    assert(buffer.writeArray(data, 512)==512);
+    delay(1);
+
   });
+
   readTask.begin([]() {
     static uint64_t start = millis();
     static size_t total_bytes = 0;
     static size_t errors = 0;
     static int16_t data[512];
 
     // read data
-    buffer.readArray(data, 512);
+    assert(buffer.readArray(data, 512)==512);
+    delay(1);
 
     // check data
     for (int j = 0; j < 512; j++) {

src/Concurrency/BufferRTOS.h

@@ -28,18 +28,18 @@ namespace audio_tools {
 template <typename T>
 class BufferRTOS : public BaseBuffer<T> {
  public:
-  BufferRTOS(size_t xStreamBufferSizeBytes, size_t xTriggerLevel = 1,
+  BufferRTOS(size_t streamBufferSize, size_t xTriggerLevel = 1,
              TickType_t writeMaxWait = portMAX_DELAY,
              TickType_t readMaxWait = portMAX_DELAY,
              Allocator &allocator = DefaultAllocator)
       : BaseBuffer<T>() {
     readWait = readMaxWait;
     writeWait = writeMaxWait;
-    current_size = xStreamBufferSizeBytes;
+    current_size_bytes = (streamBufferSize+1)  * sizeof(T);
     trigger_level = xTriggerLevel;
     p_allocator = &allocator;
 
-    if (current_size > 0) {
+    if (streamBufferSize > 0) {
       setup();
     }
   }
@@ -49,9 +49,10 @@ class BufferRTOS : public BaseBuffer<T> {
   /// Re-Allocats the memory and the queue
   bool resize(size_t size) {
     bool result = true;
-    if (current_size != size) {
+    int req_size_bytes = (size + 1)*sizeof(T);
+    if (current_size_bytes != req_size_bytes) {
       end();
-      current_size = size;
+      current_size_bytes = req_size_bytes;
       result = setup();
     }
     return result;
@@ -84,10 +85,10 @@ class BufferRTOS : public BaseBuffer<T> {
 #else
       portYIELD_FROM_ISR(xHigherPriorityTaskWoken);
 #endif
-      return result;
+      return result / sizeof(T);
     } else {
       return xStreamBufferReceive(xStreamBuffer, (void *)data, sizeof(T) * len,
-                                  readWait);
+                                  readWait) / sizeof(T);
     }
   }
 
@@ -103,10 +104,10 @@ class BufferRTOS : public BaseBuffer<T> {
 #else
       portYIELD_FROM_ISR(xHigherPriorityTaskWoken);
 #endif
-      return result;
+      return result / sizeof(T);
     } else {
       return xStreamBufferSend(xStreamBuffer, (void *)data, sizeof(T) * len,
-                               writeWait);
+                               writeWait) / sizeof(T);
     }
   }
 
@@ -134,12 +135,12 @@ class BufferRTOS : public BaseBuffer<T> {
 
   // provides the number of entries that are available to read
   int available() override {
-    return xStreamBufferBytesAvailable(xStreamBuffer);
+    return xStreamBufferBytesAvailable(xStreamBuffer) / sizeof(T);
   }
 
   // provides the number of entries that are available to write
   int availableForWrite() override {
-    return xStreamBufferSpacesAvailable(xStreamBuffer);
+    return xStreamBufferSpacesAvailable(xStreamBuffer) / sizeof(T);
   }
 
   // returns the address of the start of the physical read buffer
@@ -148,7 +149,7 @@ class BufferRTOS : public BaseBuffer<T> {
     return nullptr;
   }
 
-  size_t size() { return current_size; }
+  size_t size() { return current_size_bytes / sizeof(T); }
 
  protected:
   StreamBufferHandle_t xStreamBuffer = nullptr;
@@ -160,16 +161,16 @@ class BufferRTOS : public BaseBuffer<T> {
   int writeWait = portMAX_DELAY;
   bool read_from_isr = false;
   bool write_from_isr = false;
-  size_t current_size = 0;
+  size_t current_size_bytes = 0;
   size_t trigger_level = 0;
 
   /// The allocation has been postponed to be done here, so that we can e.g. use
   /// psram
   bool setup() {
-    if (current_size == 0) return true;
+    if (current_size_bytes == 0) return true;
 
     // allocate data if necessary
-    int size = (current_size + 1) * sizeof(T);
+    int size = (current_size_bytes + 1) * sizeof(T);
     if (p_data == nullptr) {
       p_data = (uint8_t *)p_allocator->allocate(size);
       // check allocation
@@ -182,7 +183,7 @@ class BufferRTOS : public BaseBuffer<T> {
 
     // create stream buffer if necessary
     if (xStreamBuffer == nullptr) {
-      xStreamBuffer = xStreamBufferCreateStatic(current_size, trigger_level,
+      xStreamBuffer = xStreamBufferCreateStatic(current_size_bytes, trigger_level,
                                                 p_data, &static_stream_buffer);
     }
     if (xStreamBuffer == nullptr) {
@@ -198,7 +199,7 @@ class BufferRTOS : public BaseBuffer<T> {
   void end() {
     if (xStreamBuffer != nullptr) vStreamBufferDelete(xStreamBuffer);
     p_allocator->free(p_data);
-    current_size = 0;
+    current_size_bytes = 0;
     p_data = nullptr;
     xStreamBuffer = nullptr;
   }

src/Concurrency/SynchronizedBuffers.h

@@ -31,9 +31,10 @@ namespace audio_tools {
 template <typename T> 
 class SynchronizedBuffer : public BaseBuffer<T> {
 public:
-  SynchronizedBuffer(BaseBuffer<T> &buffer, Mutex &mutex) {
+  SynchronizedBuffer(BaseBuffer<T> &buffer, Mutex &mutex, bool syncAvailable=false) {
     p_buffer = &buffer;
     p_mutex = &mutex;
+    is_sync_available = syncAvailable;
   }
 
   // reads a single value
@@ -96,14 +97,14 @@ class SynchronizedBuffer : public BaseBuffer<T> {
   // provides the number of entries that are available to read
   int available() override {
     TRACED();
-    LockGuard guard(p_mutex);
+    if (is_sync_available) LockGuard guard(p_mutex);
     return p_buffer->available();
   }
 
   // provides the number of entries that are available to write
   int availableForWrite() override {
     TRACED();
-    LockGuard guard(p_mutex);
+    if (is_sync_available) LockGuard guard(p_mutex);
     return p_buffer->availableForWrite();
   }
 
@@ -120,6 +121,7 @@ class SynchronizedBuffer : public BaseBuffer<T> {
 protected:
   BaseBuffer<T> *p_buffer = nullptr;
   Mutex *p_mutex = nullptr;
+  bool is_sync_available = false;
 };
 
 /**