during untar, read ahead, causing decompression to occur in the bg (#18205)

* shared buffered decompressor for untar

* spell check in comment

runtime/src/waitable_condvar.rs

@@ -0,0 +1,76 @@
+use std::{
+    sync::{Condvar, Mutex},
+    time::Duration,
+};
+
+// encapsulate complications of unneeded mutex and Condvar to give us event behavior of wait and notify
+// this will likely be wrapped in an arc somehow
+#[derive(Default, Debug)]
+pub struct WaitableCondvar {
+    pub mutex: Mutex<u8>,
+    pub event: Condvar,
+}
+
+impl WaitableCondvar {
+    pub fn notify_all(&self) {
+        self.event.notify_all();
+    }
+    pub fn wait_timeout(&self, timeout: Duration) -> bool {
+        let lock = self.mutex.lock().unwrap();
+        let res = self.event.wait_timeout(lock, timeout).unwrap();
+        if res.1.timed_out() {
+            return true;
+        }
+        false
+    }
+}
+
+#[cfg(test)]
+pub mod tests {
+    use super::*;
+    use std::{
+        sync::{
+            atomic::{AtomicBool, Ordering},
+            Arc,
+        },
+        thread::Builder,
+    };
+    #[test]
+    fn test_waitable_condvar() {
+        let data = Arc::new(AtomicBool::new(false));
+        let data_ = data.clone();
+        let cv = Arc::new(WaitableCondvar::default());
+        let cv2 = Arc::new(WaitableCondvar::default());
+        let cv_ = cv.clone();
+        let cv2_ = cv2.clone();
+        let cv2__ = cv2.clone();
+        // several passes to check re-notification and drop one of the
+        let passes = 3;
+        let handle = Builder::new().spawn(move || {
+            for _pass in 0..passes {
+                let mut notified = false;
+                while cv2_.wait_timeout(Duration::from_millis(1)) {
+                    if !notified && data_.load(Ordering::Relaxed) {
+                        notified = true;
+                        cv_.notify_all();
+                    }
+                }
+                assert!(data_.swap(false, Ordering::Relaxed));
+            }
+        });
+        // just wait, but 1 less pass - verifies that notify_all works with multiple and with 1
+        let handle2 = Builder::new().spawn(move || {
+            for _pass in 0..(passes - 1) {
+                assert!(!cv2__.wait_timeout(Duration::from_millis(10000))); // long enough to not be intermittent, short enough to fail if we really don't get notified
+            }
+        });
+        for _pass in 0..passes {
+            assert!(cv.wait_timeout(Duration::from_millis(1)));
+            assert!(!data.swap(true, Ordering::Relaxed));
+            assert!(!cv.wait_timeout(Duration::from_millis(10000))); // should barely wait, but don't want intermittent
+            cv2.notify_all();
+        }
+        assert!(handle.unwrap().join().is_ok());
+        assert!(handle2.unwrap().join().is_ok());
+    }
+}