16372702366

Committed 18 Jul 2025 02:11PM UTC coverage: 81.499% (-0.03%) from 81.529%

Build # 16372702366

Build Type

Pull #3899

github

Committed by

web-flow

Commit Message

Merge b2016e2e9 into e971e6c7f

Pull Request Pull Request #3899: Remove async API for scanning

Run Details

188 of 207 new or added lines in 9 files covered. (90.82%)

10 existing lines in 2 files now uncovered.

42034 of 51576 relevant lines covered (81.5%)

171519.79 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

51.02

/vortex-layout/src/executor.rs

// SPDX-License-Identifier: Apache-2.0
// SPDX-FileCopyrightText: Copyright the Vortex contributors

use std::sync::Arc;

use futures::FutureExt;
use futures::channel::oneshot;
use futures::executor::ThreadPool;
use futures::future::BoxFuture;
use futures::task::SpawnExt;
use vortex_error::{ResultExt, VortexResult, vortex_err};

pub trait TaskExecutor: 'static + Send + Sync {
    fn do_spawn(
        &self,
        fut: BoxFuture<'static, VortexResult<()>>,
    ) -> BoxFuture<'static, VortexResult<()>>;
}

impl<T: TaskExecutor> TaskExecutor for Arc<T> {
    fn do_spawn(
        &self,
        fut: BoxFuture<'static, VortexResult<()>>,
    ) -> BoxFuture<'static, VortexResult<()>> {
        self.as_ref().do_spawn(fut)
    }
}

pub trait TaskExecutorExt: TaskExecutor {
    fn spawn<T: 'static + Send>(
        &self,
        fut: BoxFuture<'static, VortexResult<T>>,
    ) -> BoxFuture<'static, VortexResult<T>>;
}

impl<E: TaskExecutor + ?Sized> TaskExecutorExt for E {
    fn spawn<T: 'static + Send>(
        &self,
        fut: BoxFuture<'static, VortexResult<T>>,
    ) -> BoxFuture<'static, VortexResult<T>> {
        let (send, recv) = oneshot::channel::<VortexResult<T>>();
        let fut = self.do_spawn(
            async move {
                let result = fut.await;
                send.send(result)
                    .map_err(|_| vortex_err!("Failed to send result"))
            }
            .boxed(),
        );

        Box::pin(async move {
            fut.await?;
            recv.await
                .map_err(|canceled| vortex_err!("Spawned task canceled {}", canceled))
                .unnest()
        })
    }
}

#[cfg(feature = "tokio")]
impl TaskExecutor for tokio::runtime::Handle {
    fn do_spawn(
        &self,
        f: BoxFuture<'static, VortexResult<()>>,
    ) -> BoxFuture<'static, VortexResult<()>> {
        use futures::TryFutureExt;

        tokio::runtime::Handle::spawn(self, f)
            .map_err(vortex_error::VortexError::from)
            .map(|result| result.unnest())
            .boxed()
    }
}

pub struct LocalExecutor;

impl TaskExecutor for LocalExecutor {
    fn do_spawn(
        &self,
        fut: BoxFuture<'static, VortexResult<()>>,
    ) -> BoxFuture<'static, VortexResult<()>> {
        fut
    }
}

impl TaskExecutor for ThreadPool {
    fn do_spawn(
        &self,
        fut: BoxFuture<'static, VortexResult<()>>,
    ) -> BoxFuture<'static, VortexResult<()>> {
        let fut = self
            .spawn_with_handle(fut)
            .map_err(|e| vortex_err!("Failed to spawn task: {}", e));
        async move { fut?.await }.boxed()
    }
}

1	// SPDX-License-Identifier: Apache-2.0
2	// SPDX-FileCopyrightText: Copyright the Vortex contributors
3
4	use std::sync::Arc;
5
6	use futures::FutureExt;
7	use futures::channel::oneshot;
8	use futures::executor::ThreadPool;
9	use futures::future::BoxFuture;
10	use futures::task::SpawnExt;
11	use vortex_error::{ResultExt, VortexResult, vortex_err};
12
13	pub trait TaskExecutor: 'static + Send + Sync {
14	fn do_spawn(
15	&self,
16	fut: BoxFuture<'static, VortexResult<()>>,
17	) -> BoxFuture<'static, VortexResult<()>>;
18	}
19
20	impl<T: TaskExecutor> TaskExecutor for Arc<T> {
21	fn do_spawn(	×
22	&self,	×
23	fut: BoxFuture<'static, VortexResult<()>>,	×
24	) -> BoxFuture<'static, VortexResult<()>> {	×
25	self.as_ref().do_spawn(fut)	×
26	}	×
27	}
28
29	pub trait TaskExecutorExt: TaskExecutor {
30	fn spawn<T: 'static + Send>(
31	&self,
32	fut: BoxFuture<'static, VortexResult<T>>,
33	) -> BoxFuture<'static, VortexResult<T>>;
34	}
35
36	impl<E: TaskExecutor + ?Sized> TaskExecutorExt for E {
37	fn spawn<T: 'static + Send>(	7,489✔
38	&self,	7,489✔
39	fut: BoxFuture<'static, VortexResult<T>>,	7,489✔
40	) -> BoxFuture<'static, VortexResult<T>> {	7,489✔
41	let (send, recv) = oneshot::channel::<VortexResult<T>>();	7,489✔
42	let fut = self.do_spawn(	7,489✔
43	async move {	7,489✔
44	let result = fut.await;	7,489✔
45	send.send(result)	7,489✔
46	.map_err(\|_\| vortex_err!("Failed to send result"))	7,489✔
47	}	7,489✔
48	.boxed(),	7,489✔
49	);
50
51	Box::pin(async move {	7,489✔
52	fut.await?;	7,489✔
53	recv.await	7,489✔
54	.map_err(\|canceled\| vortex_err!("Spawned task canceled {}", canceled))	7,489✔
55	.unnest()	7,489✔
56	})	7,489✔
57	}	7,489✔
58	}
59
60	#[cfg(feature = "tokio")]
61	impl TaskExecutor for tokio::runtime::Handle {
UNCOV 62	fn do_spawn(	×
UNCOV 63	&self,	×
UNCOV 64	f: BoxFuture<'static, VortexResult<()>>,	×
UNCOV 65	) -> BoxFuture<'static, VortexResult<()>> {	×
66	use futures::TryFutureExt;
67
UNCOV 68	tokio::runtime::Handle::spawn(self, f)	×
UNCOV 69	.map_err(vortex_error::VortexError::from)	×
UNCOV 70	.map(\|result\| result.unnest())	×
UNCOV 71	.boxed()	×
UNCOV 72	}	×
73	}
74
75	pub struct LocalExecutor;
76
77	impl TaskExecutor for LocalExecutor {
78	fn do_spawn(	7,489✔
79	&self,	7,489✔
80	fut: BoxFuture<'static, VortexResult<()>>,	7,489✔
81	) -> BoxFuture<'static, VortexResult<()>> {	7,489✔
82	fut	7,489✔
83	}	7,489✔
84	}
85
86	impl TaskExecutor for ThreadPool {
NEW 87	fn do_spawn(	×
NEW 88	&self,	×
NEW 89	fut: BoxFuture<'static, VortexResult<()>>,	×
NEW 90	) -> BoxFuture<'static, VortexResult<()>> {	×
NEW 91	let fut = self	×
NEW 92	.spawn_with_handle(fut)	×
NEW 93	.map_err(\|e\| vortex_err!("Failed to spawn task: {}", e));	×
NEW 94	async move { fut?.await }.boxed()	×
NEW 95	}	×
96	}

vortex-data / vortex / 16372702366

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous