roc-streaming#770: Added resize() for RingQueue

src/internal_modules/roc_core/ring_queue.h

@@ -170,6 +170,29 @@ template <class T, size_t EmbeddedCapacity = 0> class RingQueue : public NonCopy
         end_ = (end_ - 1 + buff_len_) % buff_len_;
     }
 
+    //! Set ring queue size
+    //! @remarks
+    //! Calls grow() to ensure that there is enough space in ring queue
+    //! @returns
+    //! false if the allocation failed
+    ROC_ATTR_NODISCARD bool resize(size_t new_size) {
+        if (!grow(new_size)) {
+            return false;
+        }
+
+        for (size_t n = size(); n < new_size; ++n) {
+            new (&buff_[end_]) T();
+            end_ = (end_ + 1) % buff_len_;
+        }
+
+        for (size_t n = size(); n > new_size; n--) {
+            end_ = (end_ - 1 + buff_len_) % buff_len_;
+            buff_[end_].~T();
+        }
+
+        return true;
+    }
+
 private:
     T* allocate_(size_t n_buff_elems) {
         T* data = NULL;
@@ -200,6 +223,41 @@ template <class T, size_t EmbeddedCapacity = 0> class RingQueue : public NonCopy
         }
     }
 
+    bool grow(size_t min_len) {
+        if (min_len <= capacity()) {
+            return true;
+        }
+
+        T* new_buff = allocate_(min_len + 1);
+        if (!new_buff) {
+            return false;
+        }
+
+        if (new_buff != buff_) {
+            // Copy old objects to the beginning of the new memory.
+            size_t end_index = (end_ - 1 + buff_len_) % buff_len_;
+            for (size_t i = begin_, j = 0; i != end_index; i = (i + 1) % buff_len_, ++j) {
+                new (new_buff + j) T(buff_[i]);
+            }
+
+            for (size_t i = end_index; i != begin_; i = (i - 1 + buff_len_) % buff_len_) {
+                buff_[i].~T();
+            }
+
+            // Free old memory
+            if (buff_) {
+                deallocate_(buff_);
+            }
+
+            buff_ = new_buff;
+            end_ = size();
+            begin_ = 0;
+        }
+
+        buff_len_ = min_len + 1;
+        return true;
+    }
+
     T* buff_;
     size_t buff_len_;
     size_t begin_;

src/tests/roc_core/test_ring_queue.cpp

@@ -307,5 +307,31 @@ TEST(ring_queue, constructor_destructor) {
     LONGS_EQUAL(0, Object::n_objects);
 }
 
+TEST(ring_queue, resize) {
+    size_t half_NumObjects = NumObjects >> 1;
+    RingQueue<Object, EmbeddedCap> queue(arena, half_NumObjects);
+    for (size_t n = 0; n < half_NumObjects; ++n) {
+        queue.push_back(Object(42));
+    }
+    queue.pop_front();
+    queue.push_back(Object(42));
+
+    LONGS_EQUAL(half_NumObjects, queue.capacity());
+    LONGS_EQUAL(half_NumObjects, queue.size());
+
+    CHECK(queue.resize(NumObjects));
+
+    LONGS_EQUAL(1, arena.num_allocations());
+    LONGS_EQUAL(NumObjects, queue.capacity());
+    LONGS_EQUAL(NumObjects, queue.size());
+    LONGS_EQUAL(NumObjects, Object::n_objects);
+
+    CHECK(queue.resize(half_NumObjects - 2));
+
+    LONGS_EQUAL(NumObjects, queue.capacity());
+    LONGS_EQUAL(half_NumObjects - 2, queue.size());
+    LONGS_EQUAL(half_NumObjects - 2, Object::n_objects);
+}
+
 } // namespace core
 } // namespace roc