BufferRP2040 blocking read

Author: pschatzmann

examples/tests/rp2040/buffer2040_1/buffer2040_1.ino

@@ -0,0 +1,36 @@
+#include "AudioTools.h"
+#include "AudioTools/Concurrency/RP2040.h"
+
+BufferRP2040T<int> buffer(10, 20);  //20 blocks with 10 ints->100 ints
+
+void setup() {
+  Serial.begin(115200);
+  AudioToolsLogger.begin(Serial, AudioToolsLogLevel::Info);
+  while (!Serial)
+    ;
+  delay(1000);
+  Serial.println("testing...");
+
+  assert(buffer.size() == 200);
+  assert(buffer.available() == 0);
+  assert(buffer.availableForWrite() == 200);
+  for (int j = 0; j < 10; j++) {
+    assert(buffer.write(j));
+  }
+  assert(buffer.size() == 200);
+  assert(buffer.available() == 10);
+
+  int tmp[20];
+  assert(buffer.readArray(tmp, 20) == 10);
+
+  for (int j = 0; j < 10; j++) {
+    assert(tmp[j] == j);
+  }
+
+  assert(buffer.writeArray(tmp, 10) == 10);
+
+  Serial.println("success!");
+}
+
+void loop() {
+}

examples/tests/rp2040/buffer2040/buffer2040.ino renamed to examples/tests/rp2040/buffer2040_2/buffer2040_2.ino

@@ -16,6 +16,11 @@ namespace audio_tools {
  * @brief Buffer implementation which is based on a RP2040 queue. This
  * class is intended to be used to exchange data between the 2 different
  * cores. Multi-core and IRQ safe queue implementation!
+ * 
+ * In order to increase the efficiency we to not enqueue individual items
+ * but write them into a temporary buffer of bufferSize and write this
+ * array to the queue when it is full.
+ * 
  * @ingroup buffers
  * @ingroup concurrency
  * @author Phil Schatzmann
@@ -25,34 +30,40 @@ namespace audio_tools {
 template <typename T>
 class BufferRP2040T : public BaseBuffer<T> {
  public:
-  BufferRP2040T(size_t bufferSize, int bufferCount = 2) : BaseBuffer<T>() {
-    buffer_size = bufferSize * sizeof(T);
-    buffer_size_req_bytes = buffer_size * bufferCount;
+  BufferRP2040T(int bufferCount) : BaseBuffer<T>() {
+    buffer_size = 1;
+    buffer_size_bytes = sizeof(T);
+    buffer_size_req_bytes = buffer_size_bytes * bufferCount;
+  }
+
+  BufferRP2040T(size_t bufferSize, int bufferCount) : BaseBuffer<T>() {
+    buffer_size = bufferSize;
+    buffer_size_bytes = bufferSize * sizeof(T);
+    buffer_size_req_bytes = buffer_size_bytes * bufferCount;
   }
 
   ~BufferRP2040T() { reset(); }
 
-  /// Re-Allocats the memory and the queue
+  /// Re-Allocats the memory and the queue (size is in entries)
   bool resize(size_t size) {
-    buffer_size_req_bytes = size;
-    if (buffer_size_total_bytes != buffer_size_req_bytes) {
-      LOGI("resize %d -> %d", buffer_size_total_bytes, buffer_size_req_bytes);
-      assert(buffer_size > 0);
-      if (is_blocking_write){
-        write_buffer.resize(buffer_size);
-        read_buffer.resize(buffer_size * 2);
-      }
-      // release existing queue
-      if (buffer_size_total_bytes > 0) {
-        queue_free(&queue);
-      }
+    int req_bytes = size * sizeof(T);
+    if (buffer_size_alloc_bytes < req_bytes) {
+      LOGI("resize %d -> %d", buffer_size_alloc_bytes / sizeof(T), size);
+      assert(buffer_size_bytes > 0);
+      write_buffer.resize(buffer_size);
+      read_buffer.resize(buffer_size * 2);
       // create new queu
-      if (buffer_size_req_bytes > 0) {
-        int count = buffer_size_req_bytes / buffer_size;
-        LOGI("queue_init(size:%d, count:%d)", buffer_size, count);
-        queue_init(&queue, buffer_size, count);
+      if (req_bytes > buffer_size_alloc_bytes) {
+        // release existing queue
+        if (buffer_size_alloc_bytes > 0) {
+          queue_free(&queue);
+        }
+
+        int count = req_bytes / buffer_size_bytes;
+        LOGI("queue_init(size:%d, count:%d)", buffer_size_bytes, count);
+        queue_init(&queue, buffer_size_bytes, count);
+        buffer_size_alloc_bytes = req_bytes;
       }
-      buffer_size_total_bytes = buffer_size_req_bytes;
     }
     return true;
   }
@@ -73,13 +84,13 @@ class BufferRP2040T : public BaseBuffer<T> {
   // reads multiple values
   int readArray(T data[], int len) override {
     LOGD("readArray: %d", len);
-    if (!is_blocking_write && read_buffer.size()==0){
-      // make sure that the read buffer is big enough
-      read_buffer.resize(len + buffer_size);
-    }
     // handle unalloc;ated queue
-    if (buffer_size_total_bytes == 0) return 0;
-    if (isEmpty() && read_buffer.isEmpty()) return 0;
+    if (buffer_size_alloc_bytes == 0) return 0;
+
+    // blocking read
+    while (is_blocking_read && read_buffer.available() + available() < len) 
+      delay(1);
+
     // fill read buffer if necessary
     while (read_buffer.availableForWrite() >= buffer_size) {
       LOGD("reading %d %d ", buffer_size, read_buffer.availableForWrite());
@@ -102,7 +113,7 @@ class BufferRP2040T : public BaseBuffer<T> {
     LOGD("writeArray: %d", len);
     int result = 0;
     // make sure that we have the data allocated
-    resize(buffer_size_req_bytes);
+    resize(buffer_size_req_bytes / sizeof(T));
 
     if (is_blocking_write) {
       result = writeBlocking(data, len);
@@ -115,12 +126,12 @@ class BufferRP2040T : public BaseBuffer<T> {
 
   // checks if the buffer is full
   bool isFull() override {
-    if (buffer_size_total_bytes == 0) return false;
+    if (buffer_size_alloc_bytes == 0) return false;
     return queue_is_full(&queue);
   }
 
   bool isEmpty() {
-    if (buffer_size_total_bytes == 0) return true;
+    if (buffer_size_alloc_bytes == 0) return true;
     return queue_is_empty(&queue);
   }
 
@@ -130,44 +141,55 @@ class BufferRP2040T : public BaseBuffer<T> {
   // clears the buffer
   void reset() override {
     queue_free(&queue);
-    buffer_size_total_bytes = 0;
+    buffer_size_alloc_bytes = 0;
   }
 
   // provides the number of entries that are available to read
   int available() override {
-    return buffer_size / sizeof(T);
+    if (buffer_size_alloc_bytes == 0) return 0;
+    return (queue_get_level(&queue) * buffer_size);
   }
 
   // provides the number of entries that are available to write
-  int availableForWrite() override { return size() - available(); }
+  int availableForWrite() override { 
+    if (buffer_size_alloc_bytes == 0) return size();
+    return size() - available(); }
 
   // returns the address of the start of the physical read buffer
   T *address() override {
     LOGE("address() not implemented");
     return nullptr;
   }
 
-  size_t size() { return buffer_size_req_bytes / sizeof(T); }
+  size_t size() { return buffer_size_alloc_bytes / sizeof(T); }
 
   /// When we use a non blocking write, the write size must be identical with the buffer size
   void setBlockingWrite(bool flag){
     is_blocking_write = flag;
   }
 
+  /// When we use a blockingread, the we wait for the data to be available
+  void setBlockingRead(bool flag){
+    is_blocking_read = flag;
+  }
+
+
  protected:
   queue_t queue;
-  int buffer_size_total_bytes = 0;
+  int buffer_size_alloc_bytes = 0;
   int buffer_size_req_bytes = 0;
+  int buffer_size_bytes = 0;
   int buffer_size = 0;
   SingleBuffer<T> write_buffer{0};
   audio_tools::RingBuffer<T> read_buffer{0};
   bool is_blocking_write = true;
+  bool is_blocking_read = false;
 
   int writeBlocking(const T data[], int len) {
     LOGD("writeArray: %d", len);
 
-    if (len > buffer_size){
-      LOGE("write %d too big for buffer_size: %d", len, buffer_size);
+    if (len > buffer_size_bytes){
+      LOGE("write %d too big for buffer_size: %d", len, buffer_size_bytes);
       return 0;
     }
 
@@ -184,8 +206,8 @@ class BufferRP2040T : public BaseBuffer<T> {
   }
 
   int writeNonBlocking(const T data[], int len) {
-    if (len != buffer_size){
-      LOGE("write %d must be buffer_size: %d", len, buffer_size);
+    if (len != buffer_size_bytes){
+      LOGE("write %d must be buffer_size: %d", len, buffer_size_bytes);
       return 0;
     }