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: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
