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

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

Author: asubiotto

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 + Send + Sync>(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

@@ -7,7 +7,7 @@ use std::sync::Arc;
 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 vortex_utils::aliases::dash_map::DashMap;
 
@@ -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
     }
 
@@ -120,28 +115,18 @@ 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| {
-                            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.
-                    stream.collect::<()>().await
-                }
-            })
-            .vortex_expect("Failed to spawn I/O driver");
+        tokio::task::spawn(async move {
+            // Drive the segment event stream.
+            // TODO(asubiotto): Experiment with dispatching coalesced req as
+            //  tokio task.
+            let stream = driver
+                .map(|coalesced_req| coalesced_req.launch(read.clone()))
+                .buffer_unordered(io_concurrency);
+            pin_mut!(stream);
+
+            // Drive the stream to completion.
+            stream.collect::<()>().await
+        });
 
         Ok(VortexFile {
             footer,
@@ -246,11 +231,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.
@@ -259,11 +245,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.
@@ -296,17 +283,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.
@@ -333,8 +317,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 {
@@ -344,7 +326,6 @@ impl Default for GenericFileOptions {
             initial_read_size: 0,
             initial_read_segments: Default::default(),
             io_concurrency: 8,
-            io_dispatcher: IoDispatcher::shared(),
         }
     }
 }

vortex-io/src/dispatcher/mod.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<()> {

vortex-io/src/read.rs

@@ -12,25 +12,24 @@ use vortex_metrics::{Histogram, Timer, VortexMetrics};
 
 /// A trait for types that support asynchronous reads.
 ///
-/// References to the type must be safe to [share across threads][Send], but spawned
-/// futures may be `!Send` to support thread-per-core implementations.
+/// References to the type must be safe to [share across threads][Send], and spawned
+/// futures must also be [`Send`] to support spawning on multi-threaded executors.
 ///
 /// Readers must be cheaply cloneable to allow for easy sharing across tasks or threads.
-pub trait VortexReadAt: 'static {
+pub trait VortexReadAt: 'static + Send + Sync {
     /// Request an asynchronous positional read. Results will be returned as a [`ByteBuffer`].
     ///
     /// If the reader does not have the requested number of bytes, the returned Future will complete
     /// with an [`UnexpectedEof`][std::io::ErrorKind::UnexpectedEof].
     ///
     /// ## Thread Safety
     ///
-    /// The resultant Future need not be [`Send`], allowing implementations that use thread-per-core
-    /// executors.
+    /// The resultant Future must be [`Send`] to allow spawning on multi-threaded executors.
     fn read_byte_range(
         &self,
         range: Range<u64>,
         alignment: Alignment,
-    ) -> impl Future<Output = io::Result<ByteBuffer>>;
+    ) -> impl Future<Output = io::Result<ByteBuffer>> + Send;
 
     // TODO(ngates): the read implementation should be able to hint at its latency/throughput
     //  allowing the caller to make better decisions about how to coalesce reads.
@@ -42,7 +41,7 @@ pub trait VortexReadAt: 'static {
     ///
     /// For a file it will be the size in bytes, for an object in an
     /// `ObjectStore` it will be the `ObjectMeta::size`.
-    fn size(&self) -> impl Future<Output = io::Result<u64>>;
+    fn size(&self) -> impl Future<Output = io::Result<u64>> + Send;
 }
 
 #[derive(Debug, Clone)]
@@ -83,7 +82,7 @@ impl PerformanceHint {
     }
 }
 
-impl<T: VortexReadAt> VortexReadAt for Arc<T> {
+impl<T: VortexReadAt + Send + Sync> VortexReadAt for Arc<T> {
     async fn read_byte_range(
         &self,
         range: Range<u64>,
@@ -134,7 +133,7 @@ pub struct InstrumentedReadAt<T: VortexReadAt> {
     durations: Arc<Timer>,
 }
 
-impl<T: VortexReadAt> InstrumentedReadAt<T> {
+impl<T: VortexReadAt + Send + Sync> InstrumentedReadAt<T> {
     pub fn new(read: T, metrics: &VortexMetrics) -> Self {
         Self {
             read,
@@ -144,7 +143,7 @@ impl<T: VortexReadAt> InstrumentedReadAt<T> {
     }
 }
 
-impl<T: VortexReadAt> VortexReadAt for InstrumentedReadAt<T> {
+impl<T: VortexReadAt + Send + Sync> VortexReadAt for InstrumentedReadAt<T> {
     async fn read_byte_range(
         &self,
         range: Range<u64>,