You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@opendal.apache.org by "NKID00 (via GitHub)" <gi...@apache.org> on 2024/07/12 08:44:24 UTC
[PR] feat(services/monoiofs): monoio wrapper [opendal]
NKID00 opened a new pull request, #4885:
URL: https://github.com/apache/opendal/pull/4885
Part of #4552.
Since monoio runtime can't cooperate with tokio runtime on the same thread, this PR introduces a wrapper in `monoiofs` that wraps monoio runtime in a thread pool and provides a method that may be called in context of other runtimes to dispatch futures to the thread pool and execute it in context of monoio.
This is similar to the implementation of `compfs`, however since monoio doesn't provide a crate like `compio-dispatcher` to do these, I've implemented a minimal usable dispatcher here. This dispatcher implementation sends futures to threads at random (same as `compio-dispatcher`) without scheduling or load balancing, but it should serve for implementing rest of the service on top of it and we can improve its performance later if necessary.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(services/monoiofs): monoio wrapper [opendal]
Posted by "NKID00 (via GitHub)" <gi...@apache.org>.
NKID00 commented on PR #4885:
URL: https://github.com/apache/opendal/pull/4885#issuecomment-2227807697
> That sounds great. I recall ihciah mentioning that they've done some work internally to make monoio compatible with the tokio runtime.
Yes, there is already a compatibility layer crate `monoio-compat`. But AFAIK, this crate focuses only on api-level compatibility with tokio. It mainly facilitates porting existing tokio-based code (especially network related) to monoio and still requires user to run the code in context of monoio. However the goal of monoiofs is to operate even in context of other runtime such as tokio, and we don't need to port existing code either.
> Do you think this feature should be integrated into the upstream monoio?
My code here only focuses on our use case inside monoiofs. There are a few issue comments in monoio requesting similar feature but most of them concern other scenarios. Maybe we can later generalize it and contribute that to monoio.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(services/monoiofs): monoio wrapper [opendal]
Posted by "NKID00 (via GitHub)" <gi...@apache.org>.
NKID00 commented on code in PR #4885:
URL: https://github.com/apache/opendal/pull/4885#discussion_r1678740934
##########
core/src/services/monoiofs/core.rs:
##########
@@ -0,0 +1,230 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use std::{mem, path::PathBuf, sync::Mutex, time::Duration};
+
+use flume::{Receiver, Sender};
+use futures::{channel::oneshot, Future};
+use monoio::{FusionDriver, RuntimeBuilder};
+
+/// a boxed function that spawns task in current monoio runtime
+type TaskSpawner = Box<dyn FnOnce() + Send>;
+
+#[derive(Debug)]
+pub struct MonoiofsCore {
+ root: PathBuf,
+ #[allow(dead_code)]
+ /// sender that sends [`TaskSpawner`] to worker threads
+ tx: Sender<TaskSpawner>,
+ #[allow(dead_code)]
+ /// join handles of worker threads
+ threads: Mutex<Vec<std::thread::JoinHandle<()>>>,
+}
+
+impl MonoiofsCore {
+ pub fn new(root: PathBuf, worker_threads: usize, io_uring_entries: u32) -> Self {
+ // Since users use monoiofs in a context of tokio, all monoio
+ // operations need to be dispatched to a dedicated thread pool
+ // where a monoio runtime runs on each thread. Here we spawn
+ // these worker threads.
+ let (tx, rx) = flume::unbounded();
+ let threads = (0..worker_threads)
+ .map(move |i| {
+ let rx = rx.clone();
+ std::thread::Builder::new()
+ .name(format!("monoiofs-worker-{i}"))
+ .spawn(move || Self::worker_entrypoint(rx, io_uring_entries))
+ .expect("spawn worker thread should success")
+ })
+ .collect();
+ let threads = Mutex::new(threads);
+
+ Self { root, tx, threads }
+ }
+
+ pub fn root(&self) -> &PathBuf {
+ &self.root
+ }
+
+ /// entrypoint of each worker thread, sets up monoio runtimes and channels
+ fn worker_entrypoint(rx: Receiver<TaskSpawner>, io_uring_entries: u32) {
+ let mut rt = RuntimeBuilder::<FusionDriver>::new()
+ .enable_all()
+ .with_entries(io_uring_entries)
+ .build()
+ .expect("monoio runtime initialize should success");
+ // run a infinite loop that receives TaskSpawner and calls
+ // them in a context of monoio
+ rt.block_on(async {
+ while let Ok(spawner) = rx.recv_async().await {
+ spawner();
+ }
+ })
+ }
+
+ #[allow(dead_code)]
+ /// create a TaskSpawner, send it to the thread pool and wait
+ /// for its result
+ pub async fn dispatch<F, Fut, T>(&self, f: F) -> T
+ where
+ F: FnOnce() -> Fut + 'static + Send,
+ Fut: Future<Output = T>,
+ T: 'static + Send,
+ {
+ // oneshot channel to send result back
+ let (tx, rx) = oneshot::channel();
+ self.tx
+ .send_async(Box::new(move || {
+ monoio::spawn(async move {
+ tx.send(f().await)
+ // discard result because it may be non-Debug and
+ // we don't need it to appear in the panic message
+ .map_err(|_| ())
+ .expect("send result from worker thread should success");
+ });
+ }))
+ .await
+ .expect("send new TaskSpawner to worker thread should success");
+ match rx.await {
+ Ok(result) => result,
+ // tx is dropped without sending result, probably the worker
+ // thread has panicked.
+ Err(_) => self.propagate_worker_panic(),
+ }
+ }
+
+ /// This method always panics. It is called only when at least a
+ /// worker thread has panicked or meet a broken rx, which is
+ /// unrecoverable. It propagates worker thread's panic if there
+ /// is any and panics on normally exited thread.
+ fn propagate_worker_panic(&self) -> ! {
Review Comment:
Return position never type is stable. See [Rust Reference](https://doc.rust-lang.org/1.75.0/reference/types/never.html).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(services/monoiofs): monoio wrapper [opendal]
Posted by "Xuanwo (via GitHub)" <gi...@apache.org>.
Xuanwo commented on code in PR #4885:
URL: https://github.com/apache/opendal/pull/4885#discussion_r1678729280
##########
core/src/services/monoiofs/core.rs:
##########
@@ -0,0 +1,230 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use std::{mem, path::PathBuf, sync::Mutex, time::Duration};
+
+use flume::{Receiver, Sender};
+use futures::{channel::oneshot, Future};
+use monoio::{FusionDriver, RuntimeBuilder};
+
+/// a boxed function that spawns task in current monoio runtime
+type TaskSpawner = Box<dyn FnOnce() + Send>;
+
+#[derive(Debug)]
+pub struct MonoiofsCore {
+ root: PathBuf,
+ #[allow(dead_code)]
+ /// sender that sends [`TaskSpawner`] to worker threads
+ tx: Sender<TaskSpawner>,
+ #[allow(dead_code)]
+ /// join handles of worker threads
+ threads: Mutex<Vec<std::thread::JoinHandle<()>>>,
+}
+
+impl MonoiofsCore {
+ pub fn new(root: PathBuf, worker_threads: usize, io_uring_entries: u32) -> Self {
+ // Since users use monoiofs in a context of tokio, all monoio
+ // operations need to be dispatched to a dedicated thread pool
+ // where a monoio runtime runs on each thread. Here we spawn
+ // these worker threads.
+ let (tx, rx) = flume::unbounded();
+ let threads = (0..worker_threads)
+ .map(move |i| {
+ let rx = rx.clone();
+ std::thread::Builder::new()
+ .name(format!("monoiofs-worker-{i}"))
+ .spawn(move || Self::worker_entrypoint(rx, io_uring_entries))
+ .expect("spawn worker thread should success")
+ })
+ .collect();
+ let threads = Mutex::new(threads);
+
+ Self { root, tx, threads }
+ }
+
+ pub fn root(&self) -> &PathBuf {
+ &self.root
+ }
+
+ /// entrypoint of each worker thread, sets up monoio runtimes and channels
+ fn worker_entrypoint(rx: Receiver<TaskSpawner>, io_uring_entries: u32) {
+ let mut rt = RuntimeBuilder::<FusionDriver>::new()
+ .enable_all()
+ .with_entries(io_uring_entries)
+ .build()
+ .expect("monoio runtime initialize should success");
+ // run a infinite loop that receives TaskSpawner and calls
+ // them in a context of monoio
+ rt.block_on(async {
+ while let Ok(spawner) = rx.recv_async().await {
+ spawner();
+ }
+ })
+ }
+
+ #[allow(dead_code)]
+ /// create a TaskSpawner, send it to the thread pool and wait
+ /// for its result
+ pub async fn dispatch<F, Fut, T>(&self, f: F) -> T
+ where
+ F: FnOnce() -> Fut + 'static + Send,
+ Fut: Future<Output = T>,
+ T: 'static + Send,
+ {
+ // oneshot channel to send result back
+ let (tx, rx) = oneshot::channel();
+ self.tx
+ .send_async(Box::new(move || {
+ monoio::spawn(async move {
+ tx.send(f().await)
+ // discard result because it may be non-Debug and
+ // we don't need it to appear in the panic message
+ .map_err(|_| ())
+ .expect("send result from worker thread should success");
+ });
+ }))
+ .await
+ .expect("send new TaskSpawner to worker thread should success");
+ match rx.await {
+ Ok(result) => result,
+ // tx is dropped without sending result, probably the worker
+ // thread has panicked.
+ Err(_) => self.propagate_worker_panic(),
+ }
+ }
+
+ /// This method always panics. It is called only when at least a
+ /// worker thread has panicked or meet a broken rx, which is
+ /// unrecoverable. It propagates worker thread's panic if there
+ /// is any and panics on normally exited thread.
+ fn propagate_worker_panic(&self) -> ! {
Review Comment:
I'm guessing `!` is not stablized yet?
##########
core/src/services/monoiofs/backend.rs:
##########
@@ -31,8 +36,16 @@ pub struct MonoiofsConfig {
///
/// All operations will happen under this root.
///
- /// Default to `/` if not set.
+ /// Builder::build will return error if not set.
pub root: Option<String>,
+ /// Count of worker threads that each runs a monoio runtime.
+ ///
+ /// Default to 1.
+ pub worker_threads: Option<NonZeroUsize>,
Review Comment:
It's a bit odd for OpenDAL to expose such low-level configurations. Let's set a hardcoded number for these and remove those configurations. We can add them back if there is user demand.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(services/monoiofs): monoio wrapper [opendal]
Posted by "NKID00 (via GitHub)" <gi...@apache.org>.
NKID00 commented on code in PR #4885:
URL: https://github.com/apache/opendal/pull/4885#discussion_r1678761309
##########
core/src/services/monoiofs/backend.rs:
##########
@@ -31,8 +36,16 @@ pub struct MonoiofsConfig {
///
/// All operations will happen under this root.
///
- /// Default to `/` if not set.
+ /// Builder::build will return error if not set.
pub root: Option<String>,
+ /// Count of worker threads that each runs a monoio runtime.
+ ///
+ /// Default to 1.
+ pub worker_threads: Option<NonZeroUsize>,
Review Comment:
These low-level configurations are removed in the latest commit. Please review it again. Thanks.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(services/monoiofs): monoio wrapper [opendal]
Posted by "Xuanwo (via GitHub)" <gi...@apache.org>.
Xuanwo merged PR #4885:
URL: https://github.com/apache/opendal/pull/4885
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(services/monoiofs): monoio wrapper [opendal]
Posted by "Xuanwo (via GitHub)" <gi...@apache.org>.
Xuanwo commented on PR #4885:
URL: https://github.com/apache/opendal/pull/4885#issuecomment-2225176474
> Since monoio runtime can't cooperate with tokio runtime on the same thread, this PR introduces a wrapper in `monoiofs` that wraps monoio runtime in a thread pool and provides a method that may be called in context of other runtimes to dispatch futures to the thread pool and execute it in context of monoio.
>
> This is similar to the implementation of `compfs`, however since monoio doesn't provide a crate like `compio-dispatcher` to do these, I've implemented a minimal usable dispatcher here. This dispatcher implementation sends futures to threads at random (same as `compio-dispatcher`) without scheduling or load balancing, but it should serve for implementing rest of the service on top of it and we can improve its performance later if necessary.
That sounds great. I recall @ihciah mentioning that they've done some work internally to make monoio compatible with the tokio runtime. Do you think this feature should be integrated into the upstream monoio?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(services/monoiofs): monoio wrapper [opendal]
Posted by "NKID00 (via GitHub)" <gi...@apache.org>.
NKID00 commented on PR #4885:
URL: https://github.com/apache/opendal/pull/4885#issuecomment-2236001491
> Hi, @NKID00, thanks for your update. This PR is mostly ready to merge, can you help resolving the conflicts?
Sure.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(services/monoiofs): monoio wrapper [opendal]
Posted by "Xuanwo (via GitHub)" <gi...@apache.org>.
Xuanwo commented on PR #4885:
URL: https://github.com/apache/opendal/pull/4885#issuecomment-2235999424
Hi, @NKID00, thanks for your update. This PR is mostly ready to merge, can you help resolving the conflicts?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org