Dispatch each I/O request (#4464)

An attempt at short-term fix for
#4400

The coalescing driver was spawned as a single task, and so landed on a
single thread of the I/O dispatcher regardless of how many threads you
gave it. This PR spawns each read request to allow other threads a
chance to play ball.

The reason the Polars patch doesn't work is that we need to continue to
dispatch work in order to provide a Tokio runtime within non-Tokio
clients, such as DuckDB. This is a bigger change as part of #4406

Author: gatesn

vortex-file/src/driver.rs

@@ -421,7 +421,7 @@ impl CoalescedSegmentRequest {
     }
 
     /// Launch the request, reading the byte range from the provided reader.
-    pub async fn launch<R: VortexReadAt>(self, read: &R) {
+    pub async fn launch<R: VortexReadAt>(self, read: R) {
         let alignment = self.segment_map[*self.requests[0].id() as usize].alignment;
         let byte_range = self.byte_range.clone();
         let buffer = read

vortex-file/src/generic.rs

@@ -120,14 +120,21 @@ impl VortexOpenOptions<GenericVortexFile> {
 
         // Spawn an I/O driver onto the dispatcher.
         let io_concurrency = self.options.io_concurrency;
+        let io_dispatcher = self.options.io_dispatcher.clone();
         self.options
             .io_dispatcher
             .dispatch(move || {
                 async move {
                     // Drive the segment event stream.
                     let stream = driver
-                        .map(|coalesced_req| coalesced_req.launch(&read))
-                        .buffer_unordered(io_concurrency);
+                        .map(|coalesced_req| {
+                            let read = read.clone();
+                            io_dispatcher
+                                .dispatch(move || coalesced_req.launch(read))
+                                .vortex_expect("Failed to dispatch I/O request")
+                        })
+                        .buffer_unordered(io_concurrency)
+                        .map(|result| result.vortex_expect("infallible"));
                     pin_mut!(stream);
 
                     // Drive the stream to completion.

vortex-io/src/dispatcher/mod.rs

@@ -81,9 +81,9 @@ impl IoDispatcher {
             if #[cfg(target_arch = "wasm32")] {
                 Self(Arc::new(Inner::Wasm(WasmDispatcher::new())))
             } else if #[cfg(not(feature = "compio"))] {
-                Self(Arc::new(Inner::Tokio(TokioDispatcher::new(1))))
+                Self(Arc::new(Inner::Tokio(TokioDispatcher::new(4))))
             } else {
-                Self(Arc::new(Inner::Compio(CompioDispatcher::new(1))))
+                Self(Arc::new(Inner::Compio(CompioDispatcher::new(4))))
             }
         }
     }