tweak

Author: mox692

tokio/src/process/unix/orphan.rs

@@ -295,7 +295,7 @@ pub(crate) mod test {
     #[cfg_attr(miri, ignore)] // Miri does not support epoll.
     #[test]
     fn does_not_register_signal_if_queue_empty() {
-        let (io_driver, io_handle) = IoDriver::new(1024).unwrap();
+        let (io_driver, io_handle) = IoDriver::new(1024, 1).unwrap();
         let signal_driver = SignalDriver::new(io_driver, &io_handle).unwrap();
         let handle = signal_driver.handle();
 

tokio/src/runtime/builder.rs

@@ -940,6 +940,13 @@ impl Builder {
             enable_time: self.enable_time,
             start_paused: self.start_paused,
             nevents: self.nevents,
+            num_workers: match self.kind {
+                Kind::CurrentThread => 1,
+                #[cfg(feature = "rt-multi-thread")]
+                Kind::MultiThread => self
+                    .worker_threads
+                    .unwrap_or_else(crate::loom::sys::num_cpus),
+            },
         }
     }
 
@@ -1457,7 +1464,7 @@ impl Builder {
         use crate::runtime::scheduler;
         use crate::runtime::Config;
 
-        let (driver, driver_handle) = driver::Driver::new(self.get_cfg(), 1)?;
+        let (driver, driver_handle) = driver::Driver::new(self.get_cfg())?;
 
         // Blocking pool
         let blocking_pool = blocking::create_blocking_pool(self, self.max_blocking_threads);
@@ -1612,7 +1619,7 @@ cfg_rt_multi_thread! {
 
             let worker_threads = self.worker_threads.unwrap_or_else(num_cpus);
 
-            let (driver, driver_handle) = driver::Driver::new(self.get_cfg(), worker_threads)?;
+            let (driver, driver_handle) = driver::Driver::new(self.get_cfg())?;
 
             // Create the blocking pool
             let blocking_pool =

tokio/src/runtime/driver.rs

@@ -40,12 +40,13 @@ pub(crate) struct Cfg {
     pub(crate) enable_pause_time: bool,
     pub(crate) start_paused: bool,
     pub(crate) nevents: usize,
+    pub(crate) num_workers: usize,
 }
 
 impl Driver {
-    pub(crate) fn new(cfg: Cfg, num_workers: usize) -> io::Result<(Self, Handle)> {
+    pub(crate) fn new(cfg: Cfg) -> io::Result<(Self, Handle)> {
         let (io_stack, io_handle, signal_handle) =
-            create_io_stack(cfg.enable_io, cfg.nevents, num_workers)?;
+            create_io_stack(cfg.enable_io, cfg.nevents, cfg.num_workers)?;
 
         let clock = create_clock(cfg.enable_pause_time, cfg.start_paused);
 
@@ -142,7 +143,7 @@ cfg_io_driver! {
         assert!(!enabled);
 
         let ret = if enabled {
-            let (io_driver, io_handle) = crate::runtime::io::Driver::new(nevents,num_workers)?;
+            let (io_driver, io_handle) = crate::runtime::io::Driver::new(nevents, num_workers)?;
 
             let (signal_driver, signal_handle) = create_signal_driver(io_driver, &io_handle)?;
             let process_driver = create_process_driver(signal_driver);
@@ -203,7 +204,7 @@ cfg_not_io_driver! {
     #[derive(Debug)]
     pub(crate) struct IoStack(ParkThread);
 
-    fn create_io_stack(_enabled: bool, _nevents: usize,  _num_worker: usize) -> io::Result<(IoStack, IoHandle, SignalHandle)> {
+    fn create_io_stack(_enabled: bool, _nevents: usize, _num_worker: usize) -> io::Result<(IoStack, IoHandle, SignalHandle)> {
         let park_thread = ParkThread::new();
         let unpark_thread = park_thread.unpark();
         Ok((IoStack(park_thread), unpark_thread, Default::default()))

tokio/src/runtime/driver/op.rs

@@ -37,7 +37,7 @@ pub(crate) enum State {
 
 pub(crate) struct Op<T: Cancellable> {
     // Uring shard id this operation belongs to
-    pub(crate) shard_id: usize,
+    shard_id: usize,
     // Handle to the runtime
     handle: Handle,
     // State of this Op
@@ -55,8 +55,9 @@ impl<T: Cancellable> Op<T> {
     pub(crate) unsafe fn new(entry: Entry, data: T) -> Self {
         let handle = Handle::current();
         let shard_size = handle.inner.driver().io().uring_context.len();
+        let shard_id = OpId::next().as_u64() as usize % shard_size;
         Self {
-            shard_id: OpId::next().as_u64() as usize % shard_size,
+            shard_id,
             handle,
             data: Some(data),
             state: State::Initialize(Some(entry)),