Fix write reordering bug (#1448)

If we have _any_ block operations in the deferred queue (i.e. because
we're encrypting a large write in the thread pool), then every block
operation must go through the queue to preserve ordering.

Unfortunately, we were checking the **wrong** `DeferredQueue` when
checking whether to defer writes. This means that a large (deferred)
write followed by a short (non-deferred) write could be reordered so
that the short write happens first.

This is probably the root cause of CI failures that we saw in #1445:
fast-ack means that the Guest is able to send the short write with less
delay, making it more likely to hit this race condition.

Author: mkeeter

upstairs/src/deferred.rs

@@ -93,6 +93,12 @@ impl<T> DeferredQueue<T> {
     pub fn is_empty(&self) -> bool {
         self.empty
     }
+
+    /// Returns the number of futures in the queue
+    #[allow(dead_code)] // only used in unit tests
+    pub fn len(&self) -> usize {
+        self.stream.len()
+    }
 }
 
 ////////////////////////////////////////////////////////////////////////////////

upstairs/src/upstairs.rs

@@ -1391,7 +1391,7 @@ impl Upstairs {
         if let Some(w) =
             self.compute_deferred_write(offset, data, res, is_write_unwritten)
         {
-            let should_defer = !self.deferred_msgs.is_empty()
+            let should_defer = !self.deferred_ops.is_empty()
                 || w.data.len() > MIN_DEFER_SIZE_BYTES as usize;
             if should_defer {
                 let tx = self.deferred_ops.push_oneshot();
@@ -4205,4 +4205,41 @@ pub(crate) mod test {
         assert!(!r.contains("HashMismatch"));
         assert!(r.contains("read hash mismatch"));
     }
+
+    #[test]
+    fn write_defer() {
+        let mut up = make_upstairs();
+        up.force_active().unwrap();
+        set_all_active(&mut up.downstairs);
+
+        const NODEFER_SIZE: usize = MIN_DEFER_SIZE_BYTES as usize - 512;
+        const DEFER_SIZE: usize = MIN_DEFER_SIZE_BYTES as usize * 2;
+
+        // Submit a short write, which should not be deferred
+        let mut data = BytesMut::new();
+        data.extend_from_slice(vec![1; NODEFER_SIZE].as_slice());
+        let offset = BlockIndex(7);
+        let (_res, done) = BlockOpWaiter::pair();
+        up.apply(UpstairsAction::Guest(BlockOp::Write { offset, data, done }));
+        assert_eq!(up.deferred_ops.len(), 0);
+
+        // Submit a long write, which should be deferred
+        let mut data = BytesMut::new();
+        data.extend_from_slice(vec![2; DEFER_SIZE].as_slice());
+        let offset = BlockIndex(7);
+        let (_res, done) = BlockOpWaiter::pair();
+        up.apply(UpstairsAction::Guest(BlockOp::Write { offset, data, done }));
+        assert_eq!(up.deferred_ops.len(), 1);
+        assert_eq!(up.deferred_msgs.len(), 0);
+
+        // Submit a short write, which would normally not be deferred, but
+        // there's already a deferred job in the queue
+        let mut data = BytesMut::new();
+        data.extend_from_slice(vec![3; NODEFER_SIZE].as_slice());
+        let offset = BlockIndex(7);
+        let (_res, done) = BlockOpWaiter::pair();
+        up.apply(UpstairsAction::Guest(BlockOp::Write { offset, data, done }));
+        assert_eq!(up.deferred_ops.len(), 2);
+        assert_eq!(up.deferred_msgs.len(), 0);
+    }
 }