diskio: Add buffer_used() to Executor and verify it in the test

In rust-lang#2774 we have seen hangs during installation for some low tier
platforms for whom the 32M default if sysinfo doesn't work causes
problems.  This change *exposes* that problem in a failing test.

Signed-off-by: Daniel Silverstone <[email protected]>

src/diskio/immediate.rs

@@ -136,6 +136,11 @@ impl Executor for ImmediateUnpacker {
     fn buffer_available(&self, _len: usize) -> bool {
         true
     }
+
+    #[cfg(test)]
+    fn buffer_used(&self) -> usize {
+        0
+    }
 }
 
 /// The non-shared state for writing a file incrementally

src/diskio/mod.rs

@@ -328,6 +328,10 @@ pub(crate) trait Executor {
 
     /// Query the memory budget to see if a particular size buffer is available
     fn buffer_available(&self, len: usize) -> bool;
+
+    #[cfg(test)]
+    /// Query the memory budget to see how much of the buffer pool is in use
+    fn buffer_used(&self) -> usize;
 }
 
 /// Trivial single threaded IO to be used from executors.

src/diskio/test.rs

@@ -48,13 +48,13 @@ fn test_incremental_file(io_threads: &str) -> Result<()> {
             }
         }
         // sending a zero length chunk closes the file
-        let mut chunk = io_executor.get_buffer(0);
+        let mut chunk = io_executor.get_buffer(super::IO_CHUNK_SIZE);
         chunk = chunk.finished();
         sender(chunk);
         loop {
             for work in io_executor.completed().collect::<Vec<_>>() {
                 match work {
-                    super::CompletedIo::Chunk(_) => unreachable!(),
+                    super::CompletedIo::Chunk(_) => {}
                     super::CompletedIo::Item(_) => {
                         file_finished = true;
                     }
@@ -69,6 +69,8 @@ fn test_incremental_file(io_threads: &str) -> Result<()> {
             // no more work should be outstanding
             unreachable!();
         }
+
+        assert_eq!(io_executor.buffer_used(), 0);
         Ok(())
     })?;
     // We should be able to read back the file

src/diskio/threaded.rs

@@ -350,6 +350,11 @@ impl<'a> Executor for Threaded<'a> {
         let total_used = Threaded::ram_highwater(&self.vec_pools);
         total_used + size < self.ram_budget
     }
+
+    #[cfg(test)]
+    fn buffer_used(&self) -> usize {
+        self.vec_pools.iter().map(|(_, p)| *p.in_use.borrow()).sum()
+    }
 }
 
 impl<'a> Drop for Threaded<'a> {