Fixes #2033, shared queue push bug (#2158)

core/shared_queue.js

@@ -1,4 +1,20 @@
 // Copyright 2018-2019 the Deno authors. All rights reserved. MIT license.
+/*
+SharedQueue Binary Layout
++-------------------------------+-------------------------------+
+|                        NUM_RECORDS (32)                       |
++---------------------------------------------------------------+
+|                        NUM_SHIFTED_OFF (32)                   |
++---------------------------------------------------------------+
+|                        HEAD (32)                              |
++---------------------------------------------------------------+
+|                        OFFSETS (32)                           |
++---------------------------------------------------------------+
+|                        RECORD_ENDS (*MAX_RECORDS)           ...
++---------------------------------------------------------------+
+|                        RECORDS (*MAX_RECORDS)               ...
++---------------------------------------------------------------+
+ */
 
 (window => {
   const GLOBAL_NAMESPACE = "Deno";
@@ -69,7 +85,7 @@
     let off = head();
     let end = off + buf.byteLength;
     let index = numRecords();
-    if (end > shared32.byteLength) {
+    if (end > shared32.byteLength || index >= MAX_RECORDS) {
       console.log("shared_queue.ts push fail");
       return false;
     }
@@ -141,6 +157,7 @@
     setAsyncHandler,
     dispatch,
     sharedQueue: {
+      MAX_RECORDS,
       head,
       numRecords,
       size,

core/shared_queue.rs

@@ -1,4 +1,21 @@
 // Copyright 2018 the Deno authors. All rights reserved. MIT license.
+/*
+SharedQueue Binary Layout
++-------------------------------+-------------------------------+
+|                        NUM_RECORDS (32)                       |
++---------------------------------------------------------------+
+|                        NUM_SHIFTED_OFF (32)                   |
++---------------------------------------------------------------+
+|                        HEAD (32)                              |
++---------------------------------------------------------------+
+|                        OFFSETS (32)                           |
++---------------------------------------------------------------+
+|                        RECORD_ENDS (*MAX_RECORDS)           ...
++---------------------------------------------------------------+
+|                        RECORDS (*MAX_RECORDS)               ...
++---------------------------------------------------------------+
+ */
+
 use crate::libdeno::deno_buf;
 
 const MAX_RECORDS: usize = 100;
@@ -36,6 +53,7 @@ impl SharedQueue {
   }
 
   fn reset(&mut self) {
+    debug!("rust:shared_queue:reset");
     let s: &mut [u32] = self.as_u32_slice_mut();
     s[INDEX_NUM_RECORDS] = 0;
     s[INDEX_NUM_SHIFTED_OFF] = 0;
@@ -75,6 +93,11 @@ impl SharedQueue {
     s[INDEX_HEAD] as usize
   }
 
+  fn num_shifted_off(&self) -> usize {
+    let s = self.as_u32_slice();
+    return s[INDEX_NUM_SHIFTED_OFF] as usize;
+  }
+
   fn set_end(&mut self, index: usize, end: usize) {
     let s = self.as_u32_slice_mut();
     s[INDEX_OFFSETS + index] = end as u32;
@@ -120,7 +143,12 @@ impl SharedQueue {
     } else {
       self.reset();
     }
-
+    debug!(
+      "rust:shared_queue:shift: num_records={}, num_shifted_off={}, head={}",
+      self.num_records(),
+      self.num_shifted_off(),
+      self.head()
+    );
     Some(&self.bytes[off..end])
   }
 
@@ -128,7 +156,7 @@ impl SharedQueue {
     let off = self.head();
     let end = off + record.len();
     let index = self.num_records();
-    if end > self.bytes.len() {
+    if end > self.bytes.len() || index >= MAX_RECORDS {
       debug!("WARNING the sharedQueue overflowed");
       return false;
     }
@@ -138,6 +166,12 @@ impl SharedQueue {
     let u32_slice = self.as_u32_slice_mut();
     u32_slice[INDEX_NUM_RECORDS] += 1;
     u32_slice[INDEX_HEAD] = end as u32;
+    debug!(
+      "rust:shared_queue:push: num_records={}, num_shifted_off={}, head={}",
+      self.num_records(),
+      self.num_shifted_off(),
+      self.head()
+    );
     true
   }
 }
@@ -213,4 +247,16 @@ mod tests {
     assert_eq!(q.shift().unwrap().len(), 1);
     assert_eq!(q.size(), 0);
   }
+
+  #[test]
+  fn full_records() {
+    let mut q = SharedQueue::new(RECOMMENDED_SIZE);
+    for _ in 0..MAX_RECORDS {
+      assert!(q.push(&alloc_buf(1)))
+    }
+    assert_eq!(q.push(&alloc_buf(1)), false);
+    // Even if we shift one off, we still cannot push a new record.
+    assert_eq!(q.shift().unwrap().len(), 1);
+    assert_eq!(q.push(&alloc_buf(1)), false);
+  }
 }

core/shared_queue_test.js

@@ -6,6 +6,21 @@ function assert(cond) {
   }
 }
 
+// Check overflow (corresponds to full_records test in rust)
+function fullRecords(q) {
+  q.reset();
+  const oneByte = new Uint8Array([42]);
+  for (let i = 0; i < q.MAX_RECORDS; i++) {
+    assert(q.push(oneByte));
+  }
+  assert(!q.push(oneByte));
+  r = q.shift();
+  assert(r.byteLength == 1);
+  assert(r[0] == 42);
+  // Even if we shift one off, we still cannot push a new record.
+  assert(!q.push(oneByte));
+}
+
 function main() {
   const q = Deno.core.sharedQueue;
 
@@ -56,7 +71,10 @@ function main() {
   assert(q.numRecords() == 0);
   assert(q.size() == 0);
 
+  fullRecords(q);
+
   Deno.core.print("shared_queue_test.js ok\n");
+  q.reset();
 }
 
 main();