Corrections for Buffer

examples/tests/basic/test-buffer/test-buffer.ino

@@ -0,0 +1,40 @@
+#include "AudioTools.h"
+
+void test(BaseBuffer<int16_t> &b, const char* title){
+ Serial.println(title);
+ assert(b.isEmpty());
+ for (int j=0;j<200;j++){
+ assert(b.write(j));
+ }
+ assert(b.isFull());
+ for (int j=0;j<200;j++){
+ int16_t v = b.read();
+ assert(v==j);
+ }
+ assert(b.isEmpty());
+ int16_t array[200];
+ for (int j=0;j<200;j++){
+ array[j] = j;
+ }
+ assert(b.writeArray(array, 200)==200);
+ assert(b.readArray(array,200)==200);
+ for (int j=0;j<200;j++){
+ assert(array[j]==j);
+ }
+
+ Serial.println("Test OK");
+}
+
+void setup(){
+ SingleBuffer<int16_t> b1(200);
+ RingBuffer<int16_t> b2(200);
+ NBuffer<int16_t> b3(50,4);
+
+ test(b1,"SingleBuffer");
+ test(b2,"RingBuffer");
+ test(b3,"NBuffer");
+
+ Serial.println("Tests OK");
+}
+
+void loop(){}

examples/tests/basic/test-queue/test-queue.ino

@@ -2,12 +2,13 @@
 #include "Concurrency/QueueLockFree.h"
 
 // select the queue implementation:
-//Queue<int> q;
-//QueueFromVector<int> q{10, 0};
-QueueLockFree<int> q(10);
+Queue<int> q1;
+QueueFromVector<int> q2{10, 0};
+QueueLockFree<int> q3(10);
 
-void setup(){
- Serial.begin(115200);
+template <typename T>
+void test(T &q, const char* cls){
+ Serial.println(cls);
  assert(q.empty());
 
  for (int j=0;j<10;j++){
@@ -22,6 +23,15 @@ void setup(){
  assert(q.size()==10-(j+1));
  }
  assert(q.empty());
+ Serial.println("-> ok");
+}
+
+void setup() {
+ Serial.begin(115200);
+ test<Queue<int>>(q1,"Queue");
+ test<QueueFromVector<int>>(q2,"QueueFromVector");
+ test<QueueLockFree<int>>(q3,"QueueLockFree");
+ Serial.println("all tests passed");
 }
 
 void loop(){}

src/AudioBasic/Collections/QueueFromVector.h

@@ -21,56 +21,57 @@ class QueueFromVector {
  bool enqueue(T& data){
  if (is_full()) 
  return false;
- vector[_size++] = data;
+ vector[_end_pos++] = data;
  return true;
  }
 
  bool peek(T& data){
- if (_size <= 0 ) {
+ if (_end_pos <= 0 ) {
  data = null_value;
+ _end_pos = 0;
  return false;
  }
  data = vector[0];
  return true;
  }
 
  bool dequeue(T& data){
- if (_size <= 0 ) {
+ if (_end_pos <= 0 ) {
  data = null_value;
+ _end_pos = 0;
  return false;
  }
  data = vector[0];
- memmove(&vector[0], &vector[1], _size*sizeof(T));
- vector[_size--] = null_value;
+ memmove(&vector[0], &vector[1], (_end_pos-1)*sizeof(T));
+ vector[_end_pos--] = null_value;
  return true;
  }
 
  size_t size() {
- return _size;
+ return _end_pos < 0 ? 0 : _end_pos;
  }
 
  bool resize(size_t size) {
  if (!vector.resize(size)){
  return false;
  } 
- _size = size;
  return clear();
  }
 
  bool clear() {
- for (int j=0;j<_size;j++){
+ for (int j=0;j<vector.size();j++){
  vector[j] = null_value;
  }
- _size = 0;
+ _end_pos = 0;
  return true;
  }
 
  bool empty() {
- return _size == 0;
+ return _end_pos == 0;
  }
 
  bool is_full() {
- return _size >= vector.size(); 
+ return _end_pos >= vector.size(); 
  }
 
  size_t capacity() { return vector.capacity(); }
@@ -81,7 +82,7 @@ class QueueFromVector {
 
  protected:
  Vector<T> vector;
- size_t _size = 0;
+ int32_t _end_pos = 0;
  int empty_pos = 0;
  T null_value;
 };

src/AudioTools/Buffers.h

@@ -167,8 +167,7 @@ class SingleBuffer : public BaseBuffer<T> {
  * @param size
  */
  SingleBuffer(int size) {
- this->max_size = size;
- buffer.resize(max_size);
+ buffer.resize(size);
  reset();
  }
 
@@ -187,7 +186,7 @@ class SingleBuffer : public BaseBuffer<T> {
 
  bool write(T sample) override {
  bool result = false;
- if (current_write_pos < max_size) {
+ if (current_write_pos < buffer.size()) {
  buffer[current_write_pos++] = sample;
  result = true;
  }
@@ -215,7 +214,7 @@ class SingleBuffer : public BaseBuffer<T> {
  return max(result, 0);
  }
 
- int availableForWrite() override { return max_size - current_write_pos; }
+ int availableForWrite() override { return buffer.size() - current_write_pos; }
 
  bool isFull() override { return availableForWrite() <= 0; }
 
@@ -256,20 +255,19 @@ class SingleBuffer : public BaseBuffer<T> {
  /// If we load values directly into the address we need to set the avialeble
  /// size
  size_t setAvailable(size_t available_size) {
- size_t result = min(available_size, (size_t) max_size);
+ size_t result = min(available_size, (size_t) buffer.size());
  current_read_pos = 0;
  current_write_pos = result;
  return result;
  }
 
 
- size_t size() { return max_size; }
+ size_t size() { return buffer.size(); }
 
  void resize(int size) {
  if (buffer.size() != size) {
  TRACED();
  buffer.resize(size);
- max_size = size;
  }
  }
 
@@ -279,7 +277,6 @@ class SingleBuffer : public BaseBuffer<T> {
  }
 
  protected:
- int max_size = 0;
  int current_read_pos = 0;
  int current_write_pos = 0;
  bool owns_buffer = true;
@@ -607,7 +604,8 @@ class NBuffer : public BaseBuffer<T> {
  if (start_time == 0l) {
  start_time = millis();
  }
- sample_count++;
+ if (result)
+ sample_count++;
 
  return result;
  }
@@ -624,8 +622,7 @@ class NBuffer : public BaseBuffer<T> {
  if (result == 0) {
  // make current read buffer available again
  resetCurrent();
- result =
- actual_read_buffer == nullptr ? 0 : actual_read_buffer->available();
+ result = (actual_read_buffer == nullptr) ? 0 : actual_read_buffer->available();
  }
  return result;
  }
@@ -702,13 +699,16 @@ class NBuffer : public BaseBuffer<T> {
  if (buffer_size==size && buffer_count == count)
  return;
  freeMemory();
- filled_buffers.resize(count);
- available_buffers.resize(count);
+ //filled_buffers.resize(count);
+ //available_buffers.resize(count);
+ filled_buffers.clear();
+ available_buffers.clear();
 
  buffer_count = count;
  buffer_size = size;
  for (int j = 0; j < count; j++) {
  BaseBuffer<T>* buffer = new SingleBuffer<T>(size);
+ LOGD("new buffer %p", buffer);
  available_buffers.enqueue(buffer);
  }
  }
@@ -720,28 +720,38 @@ class NBuffer : public BaseBuffer<T> {
  uint16_t buffer_count = 0;
  BaseBuffer<T> *actual_read_buffer = nullptr;
  BaseBuffer<T> *actual_write_buffer = nullptr;
- QueueFromVector<BaseBuffer<T> *> available_buffers{0, nullptr};
- QueueFromVector<BaseBuffer<T> *> filled_buffers{0, nullptr};
+ //QueueFromVector<BaseBuffer<T> *> available_buffers{0, nullptr};
+ //QueueFromVector<BaseBuffer<T> *> filled_buffers{0, nullptr};
+ Queue<BaseBuffer<T> *> available_buffers;
+ Queue<BaseBuffer<T> *> filled_buffers;
  unsigned long start_time = 0;
  unsigned long sample_count = 0;
 
  // empty constructor only allowed by subclass
  NBuffer() = default;
 
  void freeMemory() {
- delete actual_write_buffer;
- actual_write_buffer = nullptr;
- delete actual_read_buffer;
- actual_read_buffer = nullptr;
-
+ if (actual_write_buffer){
+ LOGD("deleting %p", actual_write_buffer);
+ delete actual_write_buffer;
+ actual_write_buffer = nullptr;
+ }
+ if (actual_read_buffer){
+ LOGD("deleting %p", actual_read_buffer);
+ delete actual_read_buffer;
+ actual_read_buffer = nullptr;
+ }
+
  BaseBuffer<T> *ptr = getNextAvailableBuffer();
  while (ptr != nullptr) {
+ LOGD("deleting %p", ptr);
  delete ptr;
  ptr = getNextAvailableBuffer();
  }
 
  ptr = getNextFilledBuffer();
  while (ptr != nullptr) {
+ LOGD("deleting %p", ptr);
  delete ptr;
  ptr = getNextFilledBuffer();
  }
@@ -757,6 +767,7 @@ class NBuffer : public BaseBuffer<T> {
  }
 
  virtual BaseBuffer<T> *getNextAvailableBuffer() {
+ if (available_buffers.empty()) return nullptr;
  BaseBuffer<T> *result = nullptr;
  available_buffers.dequeue(result);
  return result;
@@ -767,6 +778,7 @@ class NBuffer : public BaseBuffer<T> {
  }
 
  virtual BaseBuffer<T> *getNextFilledBuffer() {
+ if (filled_buffers.empty()) return nullptr;
  BaseBuffer<T> *result = nullptr;
  filled_buffers.dequeue(result);
  return result;

src/Concurrency/QueueLockFree.h

@@ -22,7 +22,7 @@ class QueueLockFree {
  for (size_t i = head_pos; i != tail_pos; ++i)
  (&p_node[i & capacity_mask].data)->~T();
 
- delete[] (char*)p_node;
+ //delete[] (char*)p_node;
  }
 
  void setAllocator(Allocator& allocator) { vector.setAllocator(allocator); }
@@ -101,7 +101,7 @@ class QueueLockFree {
  std::atomic<size_t> head;
  };
 
- Node* p_node;
+ Node* p_node = nullptr;
  size_t capacity_mask;
  size_t capacity_value;
  std::atomic<size_t> tail_pos;

src/Concurrency/SynchronizedBuffers.h

@@ -201,6 +201,7 @@ class SynchronizedNBuffer : public NBuffer<T> {
 
  /// Removes all allocated buffers
  void cleanup(){
+ TRACED();
  BaseBuffer<T>* buffer = nullptr;;
  while (available_buffers.dequeue(buffer)){
  delete buffer;