[WIP] some io stuff

vortex-file/src/read/driver.rs

@@ -136,7 +136,7 @@ impl State {
         tracing::debug!(?event, "Received ReadEvent");
         match event {
             ReadEvent::Request(req) => {
-                if req.callback.is_closed() {
+                if req.is_closed() {
                     tracing::debug!(?req, "ReadRequest dropped before registration");
                     return;
                 }
@@ -145,7 +145,7 @@ impl State {
             }
             ReadEvent::Polled(req_id) => {
                 if let Some(req) = self.requests.remove(&req_id) {
-                    if req.callback.is_closed() {
+                    if req.is_closed() {
                         self.requests_by_offset.remove(&(req.offset, req_id));
                         tracing::debug!(?req, "ReadRequest dropped before poll");
                     } else {
@@ -192,7 +192,7 @@ impl State {
     fn next_uncoalesced(&mut self) -> Option<ReadRequest> {
         while let Some((req_id, req)) = self.polled_requests.pop_first() {
             self.requests_by_offset.remove(&(req.offset, req_id));
-            if req.callback.is_closed() {
+            if req.is_closed() {
                 tracing::debug!("Dropping canceled request");
                 continue;
             }
@@ -250,7 +250,7 @@ impl State {
                     .vortex_expect("Missing request in requests_by_offset");
 
                 // Skip any cancelled requests
-                if req.callback.is_closed() {
+                if req.is_closed() {
                     if ids_to_remove.insert(req_id) {
                         keys_to_remove.push((req_offset, req_id));
                     }

vortex-file/src/read/mod.rs

@@ -5,5 +5,8 @@ mod driver;
 mod request;
 
 pub(crate) use driver::IoRequestStream;
+#[cfg(test)]
+pub(crate) use request::CoalescedRequest;
+pub(crate) use request::IoRequest;
 pub(crate) use request::ReadRequest;
 pub(crate) use request::RequestId;

vortex-file/src/read/request.rs

@@ -57,6 +57,14 @@ impl IoRequest {
             IoRequestInner::Coalesced(req) => req.resolve(result),
         }
     }
+
+    /// Returns true if no callback remains to receive the read result.
+    pub(crate) fn is_cancelled(&self) -> bool {
+        match &self.0 {
+            IoRequestInner::Single(req) => req.is_closed(),
+            IoRequestInner::Coalesced(req) => req.is_cancelled(),
+        }
+    }
 }
 
 // Testing functionality
@@ -100,12 +108,16 @@ impl Debug for ReadRequest {
             .field("offset", &self.offset)
             .field("length", &self.length)
             .field("alignment", &self.alignment)
-            .field("is_closed", &self.callback.is_closed())
+            .field("is_closed", &self.is_closed())
             .finish()
     }
 }
 
 impl ReadRequest {
+    pub(crate) fn is_closed(&self) -> bool {
+        self.callback.is_closed()
+    }
+
     pub(crate) fn resolve(self, result: VortexResult<BufferHandle>) {
         if let Err(e) = self.callback.send(result) {
             tracing::debug!("ReadRequest {} dropped before resolving: {e}", self.id);
@@ -132,6 +144,10 @@ impl Debug for CoalescedRequest {
 }
 
 impl CoalescedRequest {
+    pub(crate) fn is_cancelled(&self) -> bool {
+        self.requests.iter().all(ReadRequest::is_closed)
+    }
+
     pub fn resolve(self, result: VortexResult<BufferHandle>) {
         match result {
             Ok(buffer) => {