Fixed circular buffer resizing bug.

src/Makefile

@@ -374,7 +374,6 @@ TEST_XFAILS_X86 :=  $(MUT_BOX_XFAILS) \
                     test/run-pass/vec-alloc-append.rs \
                     test/run-pass/vec-slice.rs \
 					test/run-pass/task-comm-3.rs \
-					test/run-pass/task-comm-4.rs \
                     test/compile-fail/bad-recv.rs \
                     test/compile-fail/bad-send.rs \
                     test/compile-fail/infinite-tag-type-recursion.rs \

src/rt/circular_buffer.cpp

@@ -7,7 +7,7 @@
 circular_buffer::circular_buffer(rust_dom *dom, size_t unit_sz) :
     dom(dom),
     _buffer_sz(INITIAL_CIRCULAR_BUFFFER_SIZE_IN_UNITS * unit_sz),
-    unit_sz(unit_sz),
+    _unit_sz(unit_sz),
     _next(0),
     _unread(0),
     _buffer((uint8_t *)dom->calloc(_buffer_sz)) {
@@ -27,7 +27,7 @@ circular_buffer::~circular_buffer() {
              "~circular_buffer 0x%" PRIxPTR,
              this);
     I(dom, _buffer);
-    // I(dom, unread == 0);
+    // I(dom, _unread == 0);
     dom->free(_buffer);
 }
 
@@ -38,8 +38,8 @@ void
 circular_buffer::transfer(void *dst) {
     I(dom, dst);
     uint8_t *ptr = (uint8_t *) dst;
-    for (size_t i = 0; i < _unread; i += unit_sz) {
+    for (size_t i = 0; i < _unread; i += _unit_sz) {
-        memcpy(&ptr[i], &_buffer[_next + i % _buffer_sz], unit_sz);
+        memcpy(&ptr[i], &_buffer[(_next + i) % _buffer_sz], _unit_sz);
     }
 }
 
@@ -65,15 +65,15 @@ circular_buffer::enqueue(void *src) {
     dom->log(rust_log::MEM | rust_log::COMM,
              "circular_buffer enqueue "
              "unread: %d, buffer_sz: %d, unit_sz: %d",
-             _unread, _buffer_sz, unit_sz);
+             _unread, _buffer_sz, _unit_sz);
 
     I(dom, _unread < _buffer_sz);
-    I(dom, _unread + unit_sz <= _buffer_sz);
+    I(dom, _unread + _unit_sz <= _buffer_sz);
 
     // Copy data
     size_t i = (_next + _unread) % _buffer_sz;
-    memcpy(&_buffer[i], src, unit_sz);
+    memcpy(&_buffer[i], src, _unit_sz);
-    _unread += unit_sz;
+    _unread += _unit_sz;
 
     dom->log(rust_log::MEM | rust_log::COMM,
              "circular_buffer pushed data at index: %d", i);
@@ -86,30 +86,34 @@ circular_buffer::enqueue(void *src) {
 void
 circular_buffer::dequeue(void *dst) {
     I(dom, dst);
-    I(dom, unit_sz > 0);
+    I(dom, _unit_sz > 0);
-    I(dom, _unread >= unit_sz);
+    I(dom, _unread >= _unit_sz);
     I(dom, _unread <= _buffer_sz);
     I(dom, _buffer);
-    size_t i = _next;
+
-    memcpy(dst, &_buffer[i], unit_sz);
+    memcpy(dst, &_buffer[_next], _unit_sz);
     dom->log(rust_log::MEM | rust_log::COMM,
-             "shifted data from index %d", i);
+             "shifted data from index %d", _next);
-    _unread -= unit_sz;
+    _unread -= _unit_sz;
-    _next += unit_sz;
+    _next += _unit_sz;
     I(dom, _next <= _buffer_sz);
     if (_next == _buffer_sz) {
         _next = 0;
     }
 
     // Shrink if possible.
-    if (_buffer_sz >= INITIAL_CIRCULAR_BUFFFER_SIZE_IN_UNITS * unit_sz &&
+    if (_buffer_sz >= INITIAL_CIRCULAR_BUFFFER_SIZE_IN_UNITS * _unit_sz &&
         _unread <= _buffer_sz / 4) {
+        dom->log(rust_log::MEM | rust_log::COMM,
+                 "circular_buffer is shrinking to %d bytes", _buffer_sz / 2);
         void *tmp = dom->malloc(_buffer_sz / 2);
         transfer(tmp);
         _buffer_sz >>= 1;
         dom->free(_buffer);
         _buffer = (uint8_t *)tmp;
+        _next = 0;
     }
+
 }
 
 bool

src/rt/circular_buffer.h

@@ -20,10 +20,19 @@ public:
     bool is_empty();
 
 private:
+    // Size of the buffer in bytes.
     size_t _buffer_sz;
-    size_t unit_sz;
+
+    // Size of the data unit in bytes.
+    size_t _unit_sz;
+
+    // Byte offset within the buffer where to read the next unit of data.
     size_t _next;
+
+    // Number of bytes that have not been read from the buffer.
     size_t _unread;
+
+    // The buffer itself.
     uint8_t *_buffer;
 };
 

src/test/run-pass/task-comm-4.rs

@@ -3,8 +3,30 @@ io fn main() -> () {
 }
 
 io fn test00() {
+    let int r = 0;    
+    let int sum = 0;
     let port[int] p = port();
     let chan[int] c = chan(p);
-    c <| 42;
+
-    let int r <- p;
+    c <| 1;
+    c <| 2;
+    c <| 3;
+    c <| 4;
+
+    r <- p; sum += r; log (r);
+    r <- p; sum += r; log (r);
+    r <- p; sum += r; log (r);
+    r <- p; sum += r; log (r);
+    
+    c <| 5;
+    c <| 6;
+    c <| 7;
+    c <| 8;
+
+    r <- p; sum += r; log (r);
+    r <- p; sum += r; log (r);
+    r <- p; sum += r; log (r);
+    r <- p; sum += r; log (r);
+    
+    check (sum == 1 + 2 + 3 + 4 + 5 + 6 + 7 + 8);
 }

src/test/run-pass/task-comm-5.rs

@@ -0,0 +1,25 @@
+io fn main() -> () {
+   test00();
+}
+
+io fn test00() {
+    let int r = 0;    
+    let int sum = 0;
+    let port[int] p = port();
+    let chan[int] c = chan(p);
+    let int number_of_messages = 1000;
+
+    let int i = 0;
+    while (i < number_of_messages) {
+        c <| i;
+        i += 1;
+    }
+
+    i = 0;
+    while (i < number_of_messages) {
+        r <- p; sum += r;
+        i += 1;
+    }
+    
+    check (sum == ((number_of_messages * (number_of_messages - 1)) / 2));
+}