/rust/registry/src/index.crates.io-1949cf8c6b5b557f/tokio-1.48.0/src/runtime/driver.rs

Source
//! Abstracts out the entire chain of runtime sub-drivers into common types.

// Eventually, this file will see significant refactoring / cleanup. For now, we
// don't need to worry much about dead code with certain feature permutations.
#![cfg_attr(
    any(not(all(tokio_unstable, feature = "full")), target_family = "wasm"),
    allow(dead_code)
)]

use crate::runtime::park::{ParkThread, UnparkThread};

use std::io;
use std::time::Duration;

#[derive(Debug)]
pub(crate) struct Driver {
    inner: TimeDriver,
}

#[derive(Debug)]
pub(crate) struct Handle {
    /// IO driver handle
    pub(crate) io: IoHandle,

    /// Signal driver handle
    #[cfg_attr(any(not(unix), loom), allow(dead_code))]
    pub(crate) signal: SignalHandle,

    /// Time driver handle
    pub(crate) time: TimeHandle,

    /// Source of `Instant::now()`
    #[cfg_attr(not(all(feature = "time", feature = "test-util")), allow(dead_code))]
    pub(crate) clock: Clock,
}

pub(crate) struct Cfg {
    pub(crate) enable_io: bool,
    pub(crate) enable_time: bool,
    pub(crate) enable_pause_time: bool,
    pub(crate) start_paused: bool,
    pub(crate) nevents: usize,
}

impl Driver {
    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)?;

        let clock = create_clock(cfg.enable_pause_time, cfg.start_paused);

        let (time_driver, time_handle) = create_time_driver(cfg.enable_time, io_stack, &clock);

        Ok((
            Self { inner: time_driver },
            Handle {
                io: io_handle,
                signal: signal_handle,
                time: time_handle,
                clock,
            },
        ))
    }

    pub(crate) fn park(&mut self, handle: &Handle) {
        self.inner.park(handle);
    }

    pub(crate) fn park_timeout(&mut self, handle: &Handle, duration: Duration) {
        self.inner.park_timeout(handle, duration);
    }

    pub(crate) fn shutdown(&mut self, handle: &Handle) {
        self.inner.shutdown(handle);
    }
}

impl Handle {
    pub(crate) fn unpark(&self) {
        #[cfg(feature = "time")]
        if let Some(handle) = &self.time {
            handle.unpark();
        }

        self.io.unpark();
    }

    cfg_io_driver! {
        #[track_caller]
        pub(crate) fn io(&self) -> &crate::runtime::io::Handle {
            self.io
                .as_ref()
                .expect("A Tokio 1.x context was found, but IO is disabled. Call `enable_io` on the runtime builder to enable IO.")
        }
    }

    cfg_signal_internal_and_unix! {
        #[track_caller]
        pub(crate) fn signal(&self) -> &crate::runtime::signal::Handle {
            self.signal
                .as_ref()
                .expect("there is no signal driver running, must be called from the context of Tokio runtime")
        }
    }

    cfg_time! {
        /// Returns a reference to the time driver handle.
        ///
        /// Panics if no time driver is present.
        #[track_caller]
        pub(crate) fn time(&self) -> &crate::runtime::time::Handle {
            self.time
                .as_ref()
                .expect("A Tokio 1.x context was found, but timers are disabled. Call `enable_time` on the runtime builder to enable timers.")
        }

        pub(crate) fn clock(&self) -> &Clock {
            &self.clock
        }
    }
}

// ===== io driver =====

cfg_io_driver! {
    pub(crate) type IoDriver = crate::runtime::io::Driver;

    #[derive(Debug)]
    pub(crate) enum IoStack {
        Enabled(ProcessDriver),
        Disabled(ParkThread),
    }

    #[derive(Debug)]
    pub(crate) enum IoHandle {
        Enabled(crate::runtime::io::Handle),
        Disabled(UnparkThread),
    }

    fn create_io_stack(enabled: bool, nevents: usize) -> io::Result<(IoStack, IoHandle, SignalHandle)> {
        #[cfg(loom)]
        assert!(!enabled);

        let ret = if enabled {
            let (io_driver, io_handle) = crate::runtime::io::Driver::new(nevents)?;

            let (signal_driver, signal_handle) = create_signal_driver(io_driver, &io_handle)?;
            let process_driver = create_process_driver(signal_driver);

            (IoStack::Enabled(process_driver), IoHandle::Enabled(io_handle), signal_handle)
        } else {
            let park_thread = ParkThread::new();
            let unpark_thread = park_thread.unpark();
            (IoStack::Disabled(park_thread), IoHandle::Disabled(unpark_thread), Default::default())
        };

        Ok(ret)
    }

    impl IoStack {
        pub(crate) fn park(&mut self, handle: &Handle) {
            match self {
                IoStack::Enabled(v) => v.park(handle),
                IoStack::Disabled(v) => v.park(),
            }
        }

        pub(crate) fn park_timeout(&mut self, handle: &Handle, duration: Duration) {
            match self {
                IoStack::Enabled(v) => v.park_timeout(handle, duration),
                IoStack::Disabled(v) => v.park_timeout(duration),
            }
        }

        pub(crate) fn shutdown(&mut self, handle: &Handle) {
            match self {
                IoStack::Enabled(v) => v.shutdown(handle),
                IoStack::Disabled(v) => v.shutdown(),
            }
        }
    }

    impl IoHandle {
        pub(crate) fn unpark(&self) {
            match self {
                IoHandle::Enabled(handle) => handle.unpark(),
                IoHandle::Disabled(handle) => handle.unpark(),
            }
        }

        pub(crate) fn as_ref(&self) -> Option<&crate::runtime::io::Handle> {
            match self {
                IoHandle::Enabled(v) => Some(v),
                IoHandle::Disabled(..) => None,
            }
        }
    }
}

cfg_not_io_driver! {
    pub(crate) type IoHandle = UnparkThread;

    #[derive(Debug)]
    pub(crate) struct IoStack(ParkThread);

    fn create_io_stack(_enabled: bool, _nevents: 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()))
    }

    impl IoStack {
        pub(crate) fn park(&mut self, _handle: &Handle) {
            self.0.park();
        }

        pub(crate) fn park_timeout(&mut self, _handle: &Handle, duration: Duration) {
            self.0.park_timeout(duration);
        }

        pub(crate) fn shutdown(&mut self, _handle: &Handle) {
            self.0.shutdown();
        }

        /// This is not a "real" driver, so it is not considered enabled.
        pub(crate) fn is_enabled(&self) -> bool {
            false
        }
    }
}

// ===== signal driver =====

cfg_signal_internal_and_unix! {
    type SignalDriver = crate::runtime::signal::Driver;
    pub(crate) type SignalHandle = Option<crate::runtime::signal::Handle>;

    fn create_signal_driver(io_driver: IoDriver, io_handle: &crate::runtime::io::Handle) -> io::Result<(SignalDriver, SignalHandle)> {
        let driver = crate::runtime::signal::Driver::new(io_driver, io_handle)?;
        let handle = driver.handle();
        Ok((driver, Some(handle)))
    }
}

cfg_not_signal_internal! {
    pub(crate) type SignalHandle = ();

    cfg_io_driver! {
        type SignalDriver = IoDriver;

        fn create_signal_driver(io_driver: IoDriver, _io_handle: &crate::runtime::io::Handle) -> io::Result<(SignalDriver, SignalHandle)> {
            Ok((io_driver, ()))
        }
    }
}

// ===== process driver =====

cfg_process_driver! {
    type ProcessDriver = crate::runtime::process::Driver;

    fn create_process_driver(signal_driver: SignalDriver) -> ProcessDriver {
        ProcessDriver::new(signal_driver)
    }
}

cfg_not_process_driver! {
    cfg_io_driver! {
        type ProcessDriver = SignalDriver;

        fn create_process_driver(signal_driver: SignalDriver) -> ProcessDriver {
            signal_driver
        }
    }
}

// ===== time driver =====

cfg_time! {
    #[derive(Debug)]
    pub(crate) enum TimeDriver {
        Enabled {
            driver: crate::runtime::time::Driver,
        },
        Disabled(IoStack),
    }

    pub(crate) type Clock = crate::time::Clock;
    pub(crate) type TimeHandle = Option<crate::runtime::time::Handle>;

    fn create_clock(enable_pausing: bool, start_paused: bool) -> Clock {
        crate::time::Clock::new(enable_pausing, start_paused)
    }

    fn create_time_driver(
        enable: bool,
        io_stack: IoStack,
        clock: &Clock,
    ) -> (TimeDriver, TimeHandle) {
        if enable {
            let (driver, handle) = crate::runtime::time::Driver::new(io_stack, clock);

            (TimeDriver::Enabled { driver }, Some(handle))
        } else {
            (TimeDriver::Disabled(io_stack), None)
        }
    }

    impl TimeDriver {
        pub(crate) fn park(&mut self, handle: &Handle) {
            match self {
                TimeDriver::Enabled { driver, .. } => driver.park(handle),
                TimeDriver::Disabled(v) => v.park(handle),
            }
        }

        pub(crate) fn park_timeout(&mut self, handle: &Handle, duration: Duration) {
            match self {
                TimeDriver::Enabled { driver } => driver.park_timeout(handle, duration),
                TimeDriver::Disabled(v) => v.park_timeout(handle, duration),
            }
        }

        pub(crate) fn shutdown(&mut self, handle: &Handle) {
            match self {
                TimeDriver::Enabled { driver } => driver.shutdown(handle),
                TimeDriver::Disabled(v) => v.shutdown(handle),
            }
        }
    }
}

cfg_not_time! {
    type TimeDriver = IoStack;

    pub(crate) type Clock = ();
    pub(crate) type TimeHandle = ();

    fn create_clock(_enable_pausing: bool, _start_paused: bool) -> Clock {
        ()
    }

    fn create_time_driver(
        _enable: bool,
        io_stack: IoStack,
        _clock: &Clock,
    ) -> (TimeDriver, TimeHandle) {
        (io_stack, ())
    }
}

cfg_io_uring! {
    pub(crate) mod op;
}

Coverage Report

Created: 2025-11-16 06:13

Line	Count	Source
1		//! Abstracts out the entire chain of runtime sub-drivers into common types.
2
3		// Eventually, this file will see significant refactoring / cleanup. For now, we
4		// don't need to worry much about dead code with certain feature permutations.
5		#![cfg_attr(
6		any(not(all(tokio_unstable, feature = "full")), target_family = "wasm"),
7		allow(dead_code)
8		)]
9
10		use crate::runtime::park::{ParkThread, UnparkThread};
11
12		use std::io;
13		use std::time::Duration;
14
15		#[derive(Debug)]
16		pub(crate) struct Driver {
17		inner: TimeDriver,
18		}
19
20		#[derive(Debug)]
21		pub(crate) struct Handle {
22		/// IO driver handle
23		pub(crate) io: IoHandle,
24
25		/// Signal driver handle
26		#[cfg_attr(any(not(unix), loom), allow(dead_code))]
27		pub(crate) signal: SignalHandle,
28
29		/// Time driver handle
30		pub(crate) time: TimeHandle,
31
32		/// Source of `Instant::now()`
33		#[cfg_attr(not(all(feature = "time", feature = "test-util")), allow(dead_code))]
34		pub(crate) clock: Clock,
35		}
36
37		pub(crate) struct Cfg {
38		pub(crate) enable_io: bool,
39		pub(crate) enable_time: bool,
40		pub(crate) enable_pause_time: bool,
41		pub(crate) start_paused: bool,
42		pub(crate) nevents: usize,
43		}
44
45		impl Driver {
46	17.1k	pub(crate) fn new(cfg: Cfg) -> io::Result<(Self, Handle)> {
47	17.1k	let (io_stack, io_handle, signal_handle) = create_io_stack(cfg.enable_io, cfg.nevents)?;
48
49	17.1k	let clock = create_clock(cfg.enable_pause_time, cfg.start_paused);
50
51	17.1k	let (time_driver, time_handle) = create_time_driver(cfg.enable_time, io_stack, &clock);
52
53	17.1k	Ok((
54	17.1k	Self { inner: time_driver },
55	17.1k	Handle {
56	17.1k	io: io_handle,
57	17.1k	signal: signal_handle,
58	17.1k	time: time_handle,
59	17.1k	clock,
60	17.1k	},
61	17.1k	))
62	17.1k	}
63
64	20.9k	pub(crate) fn park(&mut self, handle: &Handle) {
65	20.9k	self.inner.park(handle);
66	20.9k	}
67
68	0	pub(crate) fn park_timeout(&mut self, handle: &Handle, duration: Duration) {
69	0	self.inner.park_timeout(handle, duration);
70	0	}
71
72	547k	pub(crate) fn shutdown(&mut self, handle: &Handle) {
73	547k	self.inner.shutdown(handle);
74	547k	}
75		}
76
77		impl Handle {
78	20.9k	pub(crate) fn unpark(&self) {
79		#[cfg(feature = "time")]
80	20.9k	if let Some(handle) = &self.time {
81	20.9k	handle.unpark();
82	20.9k	}
83
84	20.9k	self.io.unpark();
85	20.9k	}
86
87		cfg_io_driver! {
88		#[track_caller]
89	38.0k	pub(crate) fn io(&self) -> &crate::runtime::io::Handle {
90	38.0k	self.io
91	38.0k	.as_ref()
92	38.0k	.expect("A Tokio 1.x context was found, but IO is disabled. Call `enable_io` on the runtime builder to enable IO.")
93	38.0k	}
94		}
95
96		cfg_signal_internal_and_unix! {
97		#[track_caller]
98	0	pub(crate) fn signal(&self) -> &crate::runtime::signal::Handle {
99	0	self.signal
100	0	.as_ref()
101	0	.expect("there is no signal driver running, must be called from the context of Tokio runtime")
102	0	}
103		}
104
105		cfg_time! {
106		/// Returns a reference to the time driver handle.
107		///
108		/// Panics if no time driver is present.
109		#[track_caller]
110	568k	pub(crate) fn time(&self) -> &crate::runtime::time::Handle {
111	568k	self.time
112	568k	.as_ref()
113	568k	.expect("A Tokio 1.x context was found, but timers are disabled. Call `enable_time` on the runtime builder to enable timers.")
114	568k	}
115
116	20.9k	pub(crate) fn clock(&self) -> &Clock {
117	20.9k	&self.clock
118	20.9k	}
119		}
120		}
121
122		// ===== io driver =====
123
124		cfg_io_driver! {
125		pub(crate) type IoDriver = crate::runtime::io::Driver;
126
127		#[derive(Debug)]
128		pub(crate) enum IoStack {
129		Enabled(ProcessDriver),
130		Disabled(ParkThread),
131		}
132
133		#[derive(Debug)]
134		pub(crate) enum IoHandle {
135		Enabled(crate::runtime::io::Handle),
136		Disabled(UnparkThread),
137		}
138
139	17.1k	fn create_io_stack(enabled: bool, nevents: usize) -> io::Result<(IoStack, IoHandle, SignalHandle)> {
140		#[cfg(loom)]
141		assert!(!enabled);
142
143	17.1k	let ret = if enabled {
144	17.1k	let (io_driver, io_handle) = crate::runtime::io::Driver::new(nevents)?;
145
146	17.1k	let (signal_driver, signal_handle) = create_signal_driver(io_driver, &io_handle)?;
147	17.1k	let process_driver = create_process_driver(signal_driver);
148
149	17.1k	(IoStack::Enabled(process_driver), IoHandle::Enabled(io_handle), signal_handle)
150		} else {
151	0	let park_thread = ParkThread::new();
152	0	let unpark_thread = park_thread.unpark();
153	0	(IoStack::Disabled(park_thread), IoHandle::Disabled(unpark_thread), Default::default())
154		};
155
156	17.1k	Ok(ret)
157	17.1k	}
158
159		impl IoStack {
160	20.9k	pub(crate) fn park(&mut self, handle: &Handle) {
161	20.9k	match self {
162	20.9k	IoStack::Enabled(v) => v.park(handle),
163	0	IoStack::Disabled(v) => v.park(),
164		}
165	20.9k	}
166
167	0	pub(crate) fn park_timeout(&mut self, handle: &Handle, duration: Duration) {
168	0	match self {
169	0	IoStack::Enabled(v) => v.park_timeout(handle, duration),
170	0	IoStack::Disabled(v) => v.park_timeout(duration),
171		}
172	0	}
173
174	17.1k	pub(crate) fn shutdown(&mut self, handle: &Handle) {
175	17.1k	match self {
176	17.1k	IoStack::Enabled(v) => v.shutdown(handle),
177	0	IoStack::Disabled(v) => v.shutdown(),
178		}
179	17.1k	}
180		}
181
182		impl IoHandle {
183	20.9k	pub(crate) fn unpark(&self) {
184	20.9k	match self {
185	20.9k	IoHandle::Enabled(handle) => handle.unpark(),
186	0	IoHandle::Disabled(handle) => handle.unpark(),
187		}
188	20.9k	}
189
190	38.0k	pub(crate) fn as_ref(&self) -> Option<&crate::runtime::io::Handle> {
191	38.0k	match self {
192	38.0k	IoHandle::Enabled(v) => Some(v),
193	0	IoHandle::Disabled(..) => None,
194		}
195	38.0k	}
196		}
197		}
198
199		cfg_not_io_driver! {
200		pub(crate) type IoHandle = UnparkThread;
201
202		#[derive(Debug)]
203		pub(crate) struct IoStack(ParkThread);
204
205		fn create_io_stack(_enabled: bool, _nevents: usize) -> io::Result<(IoStack, IoHandle, SignalHandle)> {
206		let park_thread = ParkThread::new();
207		let unpark_thread = park_thread.unpark();
208		Ok((IoStack(park_thread), unpark_thread, Default::default()))
209		}
210
211		impl IoStack {
212		pub(crate) fn park(&mut self, _handle: &Handle) {
213		self.0.park();
214		}
215
216		pub(crate) fn park_timeout(&mut self, _handle: &Handle, duration: Duration) {
217		self.0.park_timeout(duration);
218		}
219
220		pub(crate) fn shutdown(&mut self, _handle: &Handle) {
221		self.0.shutdown();
222		}
223
224		/// This is not a "real" driver, so it is not considered enabled.
225		pub(crate) fn is_enabled(&self) -> bool {
226		false
227		}
228		}
229		}
230
231		// ===== signal driver =====
232
233		cfg_signal_internal_and_unix! {
234		type SignalDriver = crate::runtime::signal::Driver;
235		pub(crate) type SignalHandle = Option<crate::runtime::signal::Handle>;
236
237	17.1k	fn create_signal_driver(io_driver: IoDriver, io_handle: &crate::runtime::io::Handle) -> io::Result<(SignalDriver, SignalHandle)> {
238	17.1k	let driver = crate::runtime::signal::Driver::new(io_driver, io_handle)?;
239	17.1k	let handle = driver.handle();
240	17.1k	Ok((driver, Some(handle)))
241	17.1k	}
242		}
243
244		cfg_not_signal_internal! {
245		pub(crate) type SignalHandle = ();
246
247		cfg_io_driver! {
248		type SignalDriver = IoDriver;
249
250		fn create_signal_driver(io_driver: IoDriver, _io_handle: &crate::runtime::io::Handle) -> io::Result<(SignalDriver, SignalHandle)> {
251		Ok((io_driver, ()))
252		}
253		}
254		}
255
256		// ===== process driver =====
257
258		cfg_process_driver! {
259		type ProcessDriver = crate::runtime::process::Driver;
260
261	17.1k	fn create_process_driver(signal_driver: SignalDriver) -> ProcessDriver {
262	17.1k	ProcessDriver::new(signal_driver)
263	17.1k	}
264		}
265
266		cfg_not_process_driver! {
267		cfg_io_driver! {
268		type ProcessDriver = SignalDriver;
269
270		fn create_process_driver(signal_driver: SignalDriver) -> ProcessDriver {
271		signal_driver
272		}
273		}
274		}
275
276		// ===== time driver =====
277
278		cfg_time! {
279		#[derive(Debug)]
280		pub(crate) enum TimeDriver {
281		Enabled {
282		driver: crate::runtime::time::Driver,
283		},
284		Disabled(IoStack),
285		}
286
287		pub(crate) type Clock = crate::time::Clock;
288		pub(crate) type TimeHandle = Option<crate::runtime::time::Handle>;
289
290	17.1k	fn create_clock(enable_pausing: bool, start_paused: bool) -> Clock {
291	17.1k	crate::time::Clock::new(enable_pausing, start_paused)
292	17.1k	}
293
294	17.1k	fn create_time_driver(
295	17.1k	enable: bool,
296	17.1k	io_stack: IoStack,
297	17.1k	clock: &Clock,
298	17.1k	) -> (TimeDriver, TimeHandle) {
299	17.1k	if enable {
300	17.1k	let (driver, handle) = crate::runtime::time::Driver::new(io_stack, clock);
301
302	17.1k	(TimeDriver::Enabled { driver }, Some(handle))
303		} else {
304	0	(TimeDriver::Disabled(io_stack), None)
305		}
306	17.1k	}
307
308		impl TimeDriver {
309	20.9k	pub(crate) fn park(&mut self, handle: &Handle) {
310	20.9k	match self {
311	20.9k	TimeDriver::Enabled { driver, .. } => driver.park(handle),
312	0	TimeDriver::Disabled(v) => v.park(handle),
313		}
314	20.9k	}
315
316	0	pub(crate) fn park_timeout(&mut self, handle: &Handle, duration: Duration) {
317	0	match self {
318	0	TimeDriver::Enabled { driver } => driver.park_timeout(handle, duration),
319	0	TimeDriver::Disabled(v) => v.park_timeout(handle, duration),
320		}
321	0	}
322
323	547k	pub(crate) fn shutdown(&mut self, handle: &Handle) {
324	547k	match self {
325	547k	TimeDriver::Enabled { driver } => driver.shutdown(handle),
326	0	TimeDriver::Disabled(v) => v.shutdown(handle),
327		}
328	547k	}
329		}
330		}
331
332		cfg_not_time! {
333		type TimeDriver = IoStack;
334
335		pub(crate) type Clock = ();
336		pub(crate) type TimeHandle = ();
337
338		fn create_clock(_enable_pausing: bool, _start_paused: bool) -> Clock {
339		()
340		}
341
342		fn create_time_driver(
343		_enable: bool,
344		io_stack: IoStack,
345		_clock: &Clock,
346		) -> (TimeDriver, TimeHandle) {
347		(io_stack, ())
348		}
349		}
350
351		cfg_io_uring! {
352		pub(crate) mod op;
353		}