vortex-file: don't run separate IO dispatch pool

Signed-off-by: Alfonso Subiotto Marques <[email protected]>

Author: asubiotto

Cargo.lock

@@ -8,7 +8,7 @@ use dashmap::DashMap;
 use futures::{StreamExt, pin_mut};
 use vortex_buffer::{Alignment, ByteBuffer, ByteBufferMut};
 use vortex_error::{VortexExpect, VortexResult, vortex_err};
-use vortex_io::{Dispatch, InstrumentedReadAt, IoDispatcher, VortexReadAt};
+use vortex_io::{InstrumentedReadAt, VortexReadAt};
 use vortex_layout::segments::{SegmentEvents, SegmentId};
 
 use crate::driver::CoalescedDriver;
@@ -18,10 +18,6 @@ use crate::segments::{
 };
 use crate::{EOF_SIZE, FileType, Footer, MAX_FOOTER_SIZE, VortexFile, VortexOpenOptions};
 
-#[cfg(feature = "tokio")]
-static TOKIO_DISPATCHER: std::sync::LazyLock<IoDispatcher> =
-    std::sync::LazyLock::new(|| IoDispatcher::new_tokio(1));
-
 /// A type of Vortex file that supports any [`VortexReadAt`] implementation.
 ///
 /// This is a reasonable choice for files backed by a network since it performs I/O coalescing.
@@ -76,8 +72,7 @@ impl VortexOpenOptions<GenericVortexFile> {
 
     /// Open a Vortex file using the provided [`std::path::Path`].
     #[cfg(feature = "tokio")]
-    pub async fn open(mut self, read: impl AsRef<std::path::Path>) -> VortexResult<VortexFile> {
-        self.options.io_dispatcher = TOKIO_DISPATCHER.clone();
+    pub async fn open(self, read: impl AsRef<std::path::Path>) -> VortexResult<VortexFile> {
         self.open_read_at(vortex_io::TokioFile::open(read)?).await
     }
 
@@ -119,7 +114,7 @@ impl VortexOpenOptions<GenericVortexFile> {
         );
 
         // Spawn an I/O driver onto the dispatcher.
-        let io_concurrency = self.options.io_concurrency;
+        /*let io_concurrency = self.options.io_concurrency;
         self.options
             .io_dispatcher
             .dispatch(move || {
@@ -134,7 +129,19 @@ impl VortexOpenOptions<GenericVortexFile> {
                     stream.collect::<()>().await
                 }
             })
-            .vortex_expect("Failed to spawn I/O driver");
+            .vortex_expect("Failed to spawn I/O driver");*/
+
+        let io_concurrency = self.options.io_concurrency;
+        tokio::task::spawn_local(async move {
+            // Drive the segment event stream.
+            let stream = driver
+                .map(|coalesced_req| coalesced_req.launch(&read))
+                .buffer_unordered(io_concurrency);
+            pin_mut!(stream);
+
+            // Drive the stream to completion.
+            stream.collect::<()>().await;
+        });
 
         Ok(VortexFile {
             footer,
@@ -239,11 +246,12 @@ impl VortexOpenOptions<GenericVortexFile> {
         &self,
         read: Arc<R>,
     ) -> VortexResult<u64> {
-        Ok(self
-            .options
-            .io_dispatcher
-            .dispatch(move || async move { read.size().await })?
-            .await??)
+        Ok(read.size().await?)
+        /*Ok(self
+        .options
+        .io_dispatcher
+        .dispatch(move || async move { read.size().await })?
+        .await??)*/
     }
 
     /// Dispatch a read onto the configured I/O dispatcher.
@@ -252,11 +260,12 @@ impl VortexOpenOptions<GenericVortexFile> {
         read: Arc<R>,
         range: Range<u64>,
     ) -> VortexResult<ByteBuffer> {
-        Ok(self
-            .options
-            .io_dispatcher
-            .dispatch(move || async move { read.read_byte_range(range, Alignment::none()).await })?
-            .await??)
+        Ok(read.read_byte_range(range, Alignment::none()).await?)
+        /*Ok(self
+        .options
+        .io_dispatcher
+        .dispatch(move || async move { read.read_byte_range(range, Alignment::none()).await })?
+        .await??)*/
     }
 
     /// Populate segments in the cache that were covered by the initial read.
@@ -289,17 +298,14 @@ impl VortexOpenOptions<GenericVortexFile> {
 #[cfg(feature = "object_store")]
 impl VortexOpenOptions<GenericVortexFile> {
     pub async fn open_object_store(
-        mut self,
+        self,
         object_store: &Arc<dyn object_store::ObjectStore>,
         path: &str,
     ) -> VortexResult<VortexFile> {
         use std::path::Path;
 
         use vortex_io::ObjectStoreReadAt;
 
-        // Object store _must_ use tokio for I/O.
-        self.options.io_dispatcher = TOKIO_DISPATCHER.clone();
-
         // If the file is local, we much prefer to use TokioFile since object store re-opens the
         // file on every read. This check is a little naive... but we hope that ObjectStore will
         // soon expose the scheme in a way that we can check more thoroughly.
@@ -326,8 +332,6 @@ pub struct GenericFileOptions {
     /// The number of concurrent I/O requests to spawn.
     /// This should be smaller than execution concurrency for coalescing to occur.
     io_concurrency: usize,
-    /// The dispatcher to use for I/O requests.
-    io_dispatcher: IoDispatcher,
 }
 
 impl Default for GenericFileOptions {
@@ -337,7 +341,6 @@ impl Default for GenericFileOptions {
             initial_read_size: 0,
             initial_read_segments: Default::default(),
             io_concurrency: 8,
-            io_dispatcher: IoDispatcher::shared(),
         }
     }
 }

vortex-file/src/generic.rs

@@ -15,7 +15,7 @@ use std::task::Poll;
 use cfg_if::cfg_if;
 use futures::FutureExt;
 use futures::channel::oneshot;
-use vortex_error::{VortexResult, vortex_err};
+use vortex_error::{VortexResult, vortex_err, vortex_panic};
 
 static SHARED: LazyLock<IoDispatcher> = LazyLock::new(IoDispatcher::new);
 
@@ -157,14 +157,17 @@ impl Dispatch for IoDispatcher {
         Fut: Future<Output = R> + 'static,
         R: Send + 'static,
     {
-        match self.0.as_ref() {
+        vortex_panic!(
+            "IoDispatcher::dispatch is disabled by Polar Signals due to performance reasons."
+        );
+        /*match self.0.as_ref() {
             #[cfg(not(target_arch = "wasm32"))]
             Inner::Tokio(tokio_dispatch) => tokio_dispatch.dispatch(task),
             #[cfg(feature = "compio")]
             Inner::Compio(compio_dispatch) => compio_dispatch.dispatch(task),
             #[cfg(target_arch = "wasm32")]
             Inner::Wasm(wasm_dispatch) => wasm_dispatch.dispatch(task),
-        }
+        }*/
     }
 
     fn shutdown(self) -> VortexResult<()> {