From baa2502ec64587710fd8db460c5751b838138a63 Mon Sep 17 00:00:00 2001 From: Carl Lerche Date: Fri, 30 Mar 2018 11:50:02 -0700 Subject: [PATCH] Integrate timers with runtime. (#266) This patch integrate the new timer implementation with the runtime by initializing a timer per worker thread. This allows minimizing the amount of synchronization needed for using timers. --- Cargo.toml | 6 +- src/lib.rs | 12 +++- src/runtime/builder.rs | 32 +++++++-- src/runtime/mod.rs | 111 +++++++++++++++++++++++++++++++- src/timer.rs | 85 ++++++++++++++++++++++++ src/util/future.rs | 61 ++++++++++++++++++ src/util/mod.rs | 9 +++ tests/timer.rs | 94 +++++++++++++++++++++++++++ tokio-threadpool/src/builder.rs | 20 +++--- tokio-threadpool/src/inner.rs | 8 +-- tokio-threadpool/src/worker.rs | 53 ++++++++++----- 11 files changed, 455 insertions(+), 36 deletions(-) create mode 100644 src/timer.rs create mode 100644 src/util/future.rs create mode 100644 src/util/mod.rs create mode 100644 tests/timer.rs diff --git a/Cargo.toml b/Cargo.toml index 6f6d30fc3..ab2871d25 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -44,9 +44,13 @@ tokio-reactor = { version = "0.1.1", path = "tokio-reactor" } tokio-threadpool = { version = "0.1.1", path = "tokio-threadpool" } tokio-tcp = { version = "0.1.0", path = "tokio-tcp" } tokio-udp = { version = "0.1.0", path = "tokio-udp" } -mio = "0.6.14" +tokio-timer = { version = "0.2.0", path = "tokio-timer" } + futures = "0.1.19" +# Needed until `reactor` is removed from `tokio`. +mio = "0.6.14" + # Futures 0.2 integration futures2 = { version = "0.1.0", path = "futures2", optional = true } diff --git a/src/lib.rs b/src/lib.rs index cbe9b32ff..42fd8ca55 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -8,9 +8,10 @@ //! * A [reactor][reactor] backed by the operating system's event queue (epoll, kqueue, //! IOCP, etc...). //! * Asynchronous [TCP and UDP][net] sockets. +//! * [Timer][timer] API for scheduling work in the future. //! -//! Tokio is built using futures (provided by the [futures] crate) as the -//! abstraction for managing the complexity of asynchronous programming. +//! Tokio is built using [futures] as the abstraction for managing the +//! complexity of asynchronous programming. //! //! Guide level documentation is found on the [website]. //! @@ -72,6 +73,7 @@ extern crate tokio_io; extern crate tokio_executor; extern crate tokio_reactor; extern crate tokio_threadpool; +extern crate tokio_timer; extern crate tokio_tcp; extern crate tokio_udp; @@ -82,6 +84,8 @@ pub mod executor; pub mod net; pub mod reactor; pub mod runtime; +pub mod timer; +pub mod util; pub use executor::spawn; #[cfg(feature = "unstable-futures")] @@ -171,6 +175,10 @@ pub mod prelude { AsyncWrite, }; + pub use util::{ + FutureExt, + }; + pub use ::std::io::{ Read, Write, diff --git a/src/runtime/builder.rs b/src/runtime/builder.rs index 0b9e759fa..e65fa7634 100644 --- a/src/runtime/builder.rs +++ b/src/runtime/builder.rs @@ -4,9 +4,10 @@ use reactor::Reactor; use std::io; +use tokio_reactor; use tokio_threadpool::Builder as ThreadPoolBuilder; - - +use tokio_threadpool::park::DefaultPark; +use tokio_timer::timer::{self, Timer}; /// Builds Tokio Runtime with custom configuration values. /// @@ -83,18 +84,39 @@ impl Builder { /// # } /// ``` pub fn build(&mut self) -> io::Result { + use std::collections::HashMap; + use std::sync::{Arc, Mutex}; + + let timers = Arc::new(Mutex::new(HashMap::<_, timer::Handle>::new())); + let t1 = timers.clone(); + // Spawn a reactor on a background thread. let reactor = Reactor::new()?.background()?; // Get a handle to the reactor. - let handle = reactor.handle().clone(); + let reactor_handle = reactor.handle().clone(); let pool = self.threadpool_builder .around_worker(move |w, enter| { - ::tokio_reactor::with_default(&handle, enter, |_| { - w.run(); + let timer_handle = t1.lock().unwrap() + .get(w.id()).unwrap() + .clone(); + + tokio_reactor::with_default(&reactor_handle, enter, |enter| { + timer::with_default(&timer_handle, enter, |_| { + w.run(); + }); }); }) + .custom_park(move |worker_id| { + // Create a new timer + let timer = Timer::new(DefaultPark::new()); + + timers.lock().unwrap() + .insert(worker_id.clone(), timer.handle()); + + timer + }) .build(); Ok(Runtime { diff --git a/src/runtime/mod.rs b/src/runtime/mod.rs index 037473579..27024a5a4 100644 --- a/src/runtime/mod.rs +++ b/src/runtime/mod.rs @@ -4,6 +4,7 @@ //! //! * A [reactor] to drive I/O resources. //! * An [executor] to execute tasks that use these I/O resources. +//! * A [timer] for scheduling work to run after a set period of time. //! //! While it is possible to setup each component manually, this involves a bunch //! of boilerplate. @@ -19,11 +20,15 @@ //! //! * Spawn a background thread running a [`Reactor`] instance. //! * Start a [`ThreadPool`] for executing futures. +//! * Run an instance of [`Timer`] **per** thread pool worker thread. //! //! The thread pool uses a work-stealing strategy and is configured to start a //! worker thread for each CPU core available on the system. This tends to be //! the ideal setup for Tokio applications. //! +//! A timer per thread pool worker thread is used to minimize the amount of +//! synchronization that is required for working with the timer. +//! //! # Usage //! //! Most applications will use the [`run`] function. This takes a future to @@ -98,11 +103,14 @@ //! //! [reactor]: ../reactor/struct.Reactor.html //! [executor]: https://tokio.rs/docs/getting-started/runtime-model/#executors +//! [timer]: ../timer/index.html //! [`Runtime`]: struct.Runtime.html +//! [`Reactor`]: ../reactor/struct.Reactor.html //! [`ThreadPool`]: ../executor/thread_pool/struct.ThreadPool.html //! [`run`]: fn.run.html //! [idle]: struct.Runtime.html#method.shutdown_on_idle //! [`tokio::spawn`]: ../executor/fn.spawn.html +//! [`Timer`]: https://docs.rs/tokio-timer/0.2/tokio_timer/timer/struct.Timer.html mod builder; mod shutdown; @@ -127,9 +135,15 @@ use futures2; /// The Tokio runtime includes a reactor as well as an executor for running /// tasks. /// +/// Instances of `Runtime` can be created using [`new`] or [`Builder`]. However, +/// most users will use [`tokio::run`], which uses a `Runtime` internally. +/// /// See [module level][mod] documentation for more details. /// /// [mod]: index.html +/// [`new`]: #method.new +/// [`Builder`]: struct.Builder.html +/// [`tokio::run`]: fn.run.html #[derive(Debug)] pub struct Runtime { inner: Option, @@ -214,19 +228,82 @@ pub fn run2(future: F) impl Runtime { /// Create a new runtime instance with default configuration values. /// + /// This results in a reactor, thread pool, and timer being initialized. The + /// thread pool will not spawn any worker threads until it needs to, i.e. + /// tasks are scheduled to run. + /// + /// Most users will not need to call this function directly, instead they + /// will use [`tokio::run`][fn.run.html]. + /// /// See [module level][mod] documentation for more details. /// + /// # Examples + /// + /// Creating a new `Runtime` with default configuration values. + /// + /// ``` + /// use tokio::runtime::Runtime; + /// use tokio::prelude::*; + /// + /// let rt = Runtime::new() + /// .unwrap(); + /// + /// // Use the runtime... + /// + /// // Shutdown the runtime + /// rt.shutdown_now() + /// .wait().unwrap(); + /// ``` + /// /// [mod]: index.html pub fn new() -> io::Result { Builder::new().build() } - /// Return a reference to the reactor handle for this runtime instance. + #[deprecated(since = "0.1.5", note = "use `reactor` instead")] + #[doc(hidden)] pub fn handle(&self) -> &Handle { + self.reactor() + } + + /// Return a reference to the reactor handle for this runtime instance. + /// + /// The returned handle reference can be cloned in order to get an owned + /// value of the handle. This handle can be used to initialize I/O resources + /// (like TCP or UDP sockets) that will not be used on the runtime. + /// + /// # Examples + /// + /// ``` + /// use tokio::runtime::Runtime; + /// + /// let rt = Runtime::new() + /// .unwrap(); + /// + /// let reactor_handle = rt.reactor().clone(); + /// + /// // use `reactor_handle` + /// ``` + pub fn reactor(&self) -> &Handle { self.inner().reactor.handle() } /// Return a handle to the runtime's executor. + /// + /// The returned handle can be used to spawn tasks that run on this runtime. + /// + /// # Examples + /// + /// ``` + /// use tokio::runtime::Runtime; + /// + /// let rt = Runtime::new() + /// .unwrap(); + /// + /// let executor_handle = rt.executor(); + /// + /// // use `executor_handle` + /// ``` pub fn executor(&self) -> TaskExecutor { let inner = self.inner().pool.sender().clone(); TaskExecutor { inner } @@ -302,6 +379,22 @@ impl Runtime { /// /// See [module level][mod] documentation for more details. /// + /// # Examples + /// + /// ``` + /// use tokio::runtime::Runtime; + /// use tokio::prelude::*; + /// + /// let rt = Runtime::new() + /// .unwrap(); + /// + /// // Use the runtime... + /// + /// // Shutdown the runtime + /// rt.shutdown_on_idle() + /// .wait().unwrap(); + /// ``` + /// /// [mod]: index.html pub fn shutdown_on_idle(mut self) -> Shutdown { let inner = self.inner.take().unwrap(); @@ -336,6 +429,22 @@ impl Runtime { /// /// See [module level][mod] documentation for more details. /// + /// # Examples + /// + /// ``` + /// use tokio::runtime::Runtime; + /// use tokio::prelude::*; + /// + /// let rt = Runtime::new() + /// .unwrap(); + /// + /// // Use the runtime... + /// + /// // Shutdown the runtime + /// rt.shutdown_now() + /// .wait().unwrap(); + /// ``` + /// /// [mod]: index.html pub fn shutdown_now(mut self) -> Shutdown { let inner = self.inner.take().unwrap(); diff --git a/src/timer.rs b/src/timer.rs new file mode 100644 index 000000000..8339a03a3 --- /dev/null +++ b/src/timer.rs @@ -0,0 +1,85 @@ +//! Utilities for tracking time. +//! +//! This module provides a number of types for executing code after a set period +//! of time. +//! +//! * [`Sleep`][Sleep] is a future that does no work and completes at a specific `Instant` +//! in time. +//! +//! * [`Interval`][Interval] is a stream yielding a value at a fixed period. It +//! is initialized with a `Duration` and repeatedly yields each time the +//! duration elapses. +//! +//! * [`Deadline`][Deadline] wraps a future, requiring that it completes before +//! a specified `Instant` in time. If the future does not complete in time, +//! then it is canceled and an error is returned. +//! +//! These types are sufficient for handling a large number of scenarios +//! involving time. +//! +//! These types must be used from within the context of the +//! [`Runtime`][runtime] or a timer context must be setup explicitly. See the +//! [`tokio-timer`][tokio-timer] crate for more details on how to setup a timer +//! context. +//! +//! # Examples +//! +//! Wait 100ms and print "Hello World!" +//! +//! ``` +//! use tokio::prelude::*; +//! use tokio::timer::Sleep; +//! +//! use std::time::{Duration, Instant}; +//! +//! let when = Instant::now() + Duration::from_millis(100); +//! +//! tokio::run({ +//! Sleep::new(when) +//! .map_err(|e| panic!("timer failed; err={:?}", e)) +//! .and_then(|_| { +//! println!("Hello world!"); +//! Ok(()) +//! }) +//! }) +//! ``` +//! +//! Require that an operation takes no more than 300ms. Note that this uses the +//! [`deadline`][ext] function on the [`FutureExt`][ext] trait. This trait is +//! included in the prelude. +//! +//! ``` +//! # extern crate futures; +//! # extern crate tokio; +//! use tokio::prelude::*; +//! +//! use std::time::{Duration, Instant}; +//! +//! fn long_op() -> Box + Send> { +//! // ... +//! # Box::new(futures::future::ok(())) +//! } +//! +//! # fn main() { +//! let when = Instant::now() + Duration::from_millis(300); +//! +//! tokio::run({ +//! long_op() +//! .deadline(when) +//! .map_err(|e| { +//! println!("operation timed out"); +//! }) +//! }) +//! # } +//! ``` +//! +//! [runtime]: ../runtime/struct.Runtime.html +//! [tokio-timer]: https://docs.rs/tokio-timer +//! [ext]: ../util/trait.FutureExt.html#method.deadline + +pub use tokio_timer::{ + Deadline, + DeadlineError, + Interval, + Sleep, +}; diff --git a/src/util/future.rs b/src/util/future.rs new file mode 100644 index 000000000..9e0fd6f04 --- /dev/null +++ b/src/util/future.rs @@ -0,0 +1,61 @@ +use tokio_timer::Deadline; + +use futures::Future; + +use std::time::Instant; + + +/// An extension trait for `Future` that provides a variety of convenient +/// combinator functions. +/// +/// Currently, there only is a [`deadline`] function, but this will increase +/// over time. +/// +/// Users are not expected to implement this trait. All types that implement +/// `Future` already implement `FutureExt`. +/// +/// This trait can be imported directly or via the Tokio prelude: `use +/// tokio::prelude::*`. +/// +/// [`deadline`]: #method.deadline +pub trait FutureExt: Future { + + /// Creates a new future which allows `self` until `deadline`. + /// + /// This combinator creates a new future which wraps the receiving future + /// with a deadline. The returned future is allowed to execute until it + /// completes or `deadline` is reached, whicheever happens first. + /// + /// If the future completes before `deadline` then the future will resolve + /// with that item. Otherwise the future will resolve to an error once + /// `deadline` is reached. + /// + /// # Examples + /// + /// ``` + /// # extern crate tokio; + /// # extern crate futures; + /// use tokio::prelude::*; + /// use std::time::{Duration, Instant}; + /// # use futures::future::{self, FutureResult}; + /// + /// # fn long_future() -> FutureResult<(), ()> { + /// # future::ok(()) + /// # } + /// # + /// # fn main() { + /// let future = long_future() + /// .deadline(Instant::now() + Duration::from_secs(1)) + /// .map_err(|e| println!("error = {:?}", e)); + /// + /// tokio::run(future); + /// # } + /// ``` + fn deadline(self, deadline: Instant) -> Deadline + where Self: Sized, + { + Deadline::new(self, deadline) + } +} + +impl FutureExt for T where T: Future {} diff --git a/src/util/mod.rs b/src/util/mod.rs new file mode 100644 index 000000000..490d0cf6e --- /dev/null +++ b/src/util/mod.rs @@ -0,0 +1,9 @@ +//! Utilities for working with Tokio. +//! +//! This module contains utilities that are useful for working with Tokio. +//! Currently, this only includes [`FutureExt`][FutureExt]. However, this will +//! include over time. + +mod future; + +pub use self::future::FutureExt; diff --git a/tests/timer.rs b/tests/timer.rs new file mode 100644 index 000000000..4cd3ec2e9 --- /dev/null +++ b/tests/timer.rs @@ -0,0 +1,94 @@ +extern crate futures; +extern crate tokio; +extern crate tokio_io; +extern crate env_logger; + +use tokio::prelude::*; +use tokio::timer::*; + +use std::sync::mpsc; +use std::time::{Duration, Instant}; + +#[test] +fn timer_with_runtime() { + let _ = env_logger::init(); + + let when = Instant::now() + Duration::from_millis(100); + let (tx, rx) = mpsc::channel(); + + tokio::run({ + Sleep::new(when) + .map_err(|e| panic!("unexpected error; err={:?}", e)) + .and_then(move |_| { + assert!(Instant::now() >= when); + tx.send(()).unwrap(); + Ok(()) + }) + }); + + rx.recv().unwrap(); +} + +#[test] +fn starving() { + use futures::{task, Poll, Async}; + + let _ = env_logger::init(); + + struct Starve(Sleep, u64); + + impl Future for Starve { + type Item = u64; + type Error = (); + + fn poll(&mut self) -> Poll { + if self.0.poll().unwrap().is_ready() { + return Ok(self.1.into()); + } + + self.1 += 1; + + task::current().notify(); + + Ok(Async::NotReady) + } + } + + let when = Instant::now() + Duration::from_millis(20); + let starve = Starve(Sleep::new(when), 0); + + let (tx, rx) = mpsc::channel(); + + tokio::run({ + starve + .and_then(move |_ticks| { + assert!(Instant::now() >= when); + tx.send(()).unwrap(); + Ok(()) + }) + }); + + rx.recv().unwrap(); +} + +#[test] +fn deadline() { + use futures::future; + + let _ = env_logger::init(); + + let when = Instant::now() + Duration::from_millis(20); + let (tx, rx) = mpsc::channel(); + + tokio::run({ + future::empty::<(), ()>() + .deadline(when) + .then(move |res| { + assert!(res.is_err()); + tx.send(()).unwrap(); + Ok(()) + }) + }); + + rx.recv().unwrap(); +} diff --git a/tokio-threadpool/src/builder.rs b/tokio-threadpool/src/builder.rs index b2788e83f..a5c719a15 100644 --- a/tokio-threadpool/src/builder.rs +++ b/tokio-threadpool/src/builder.rs @@ -7,7 +7,7 @@ use sleep_stack::SleepStack; use state::State; use thread_pool::ThreadPool; use inner::Inner; -use worker::Worker; +use worker::{Worker, WorkerId}; use worker_entry::WorkerEntry; use std::error::Error; @@ -70,7 +70,7 @@ pub struct Builder { pool_size: usize, /// Generates the `Park` instances - new_park: Box BoxPark>, + new_park: Box BoxPark>, } impl Builder { @@ -98,7 +98,7 @@ impl Builder { pub fn new() -> Builder { let num_cpus = num_cpus::get(); - let new_park = Box::new(|| { + let new_park = Box::new(|_: &WorkerId| { Box::new(BoxedPark::new(DefaultPark::new())) as BoxPark }); @@ -277,7 +277,7 @@ impl Builder { /// # pub fn main() { /// // Create a thread pool with default configuration values /// let thread_pool = Builder::new() - /// .custom_park(|| { + /// .custom_park(|_| { /// use tokio_threadpool::park::DefaultPark; /// /// // This is the default park type that the worker would use if we @@ -292,11 +292,14 @@ impl Builder { /// # } /// ``` pub fn custom_park(&mut self, f: F) -> &mut Self - where F: Fn() -> P + 'static, + where F: Fn(&WorkerId) -> P + 'static, P: Park + Send + 'static, P::Error: Error, { - self.new_park = Box::new(move || Box::new(BoxedPark::new(f()))); + self.new_park = Box::new(move |id| { + Box::new(BoxedPark::new(f(id))) + }); + self } @@ -322,8 +325,9 @@ impl Builder { trace!("build; num-workers={}", self.pool_size); - for _ in 0..self.pool_size { - let park = (self.new_park)(); + for i in 0..self.pool_size { + let id = WorkerId::new(i); + let park = (self.new_park)(&id); let unpark = park.unpark(); workers.push(WorkerEntry::new(park, unpark)); diff --git a/tokio-threadpool/src/inner.rs b/tokio-threadpool/src/inner.rs index 5fc55d195..cd6ba5b5a 100644 --- a/tokio-threadpool/src/inner.rs +++ b/tokio-threadpool/src/inner.rs @@ -7,7 +7,7 @@ use sleep_stack::{ use shutdown_task::ShutdownTask; use state::{State, SHUTDOWN_ON_IDLE, SHUTDOWN_NOW}; use task::Task; -use worker::Worker; +use worker::{Worker, WorkerId}; use worker_entry::WorkerEntry; use worker_state::{ WorkerState, @@ -189,7 +189,7 @@ impl Inner { Worker::with_current(|worker| { match worker { Some(worker) => { - let idx = worker.idx; + let idx = worker.id.idx; trace!(" -> submit internal; idx={}", idx); @@ -236,7 +236,7 @@ impl Inner { let entry = &self.workers[idx]; if !entry.submit_external(task, state) { - Worker::spawn(idx, inner); + Worker::spawn(WorkerId::new(idx), inner); } } @@ -273,7 +273,7 @@ impl Inner { } WORKER_SHUTDOWN => { trace!("signal_work -- spawn; idx={}", idx); - Worker::spawn(idx, inner); + Worker::spawn(WorkerId::new(idx), inner); } _ => {} } diff --git a/tokio-threadpool/src/worker.rs b/tokio-threadpool/src/worker.rs index a1439cdc6..3fb18f6f9 100644 --- a/tokio-threadpool/src/worker.rs +++ b/tokio-threadpool/src/worker.rs @@ -33,7 +33,7 @@ pub struct Worker { pub(crate) inner: Arc, // WorkerEntry index - pub(crate) idx: usize, + pub(crate) id: WorkerId, // Set when the worker should finalize on drop should_finalize: Cell, @@ -42,17 +42,26 @@ pub struct Worker { _p: PhantomData>, } +/// Identifiers a thread pool worker. +/// +/// This identifier is unique scoped by the thread pool. It is possible that +/// different thread pool instances share worker identifier values. +#[derive(Debug, Clone, Hash, Eq, PartialEq)] +pub struct WorkerId { + pub(crate) idx: usize, +} + // Pointer to the current worker info thread_local!(static CURRENT_WORKER: Cell<*const Worker> = Cell::new(0 as *const _)); impl Worker { - pub(crate) fn spawn(idx: usize, inner: &Arc) { - trace!("spawning new worker thread; idx={}", idx); + pub(crate) fn spawn(id: WorkerId, inner: &Arc) { + trace!("spawning new worker thread; id={}", id.idx); let mut th = thread::Builder::new(); if let Some(ref prefix) = inner.config.name_prefix { - th = th.name(format!("{}{}", prefix, idx)); + th = th.name(format!("{}{}", prefix, id.idx)); } if let Some(stack) = inner.config.stack_size { @@ -63,8 +72,8 @@ impl Worker { th.spawn(move || { let worker = Worker { - inner: inner, - idx: idx, + inner, + id, should_finalize: Cell::new(false), _p: PhantomData, }; @@ -106,6 +115,14 @@ impl Worker { }) } + /// Returns a reference to the worker's identifier. + /// + /// This identifier is unique scoped by the thread pool. It is possible that + /// different thread pool instances share worker identifier values. + pub fn id(&self) -> &WorkerId { + &self.id + } + /// Run the worker /// /// This function blocks until the worker is shutting down. @@ -261,7 +278,7 @@ impl Worker { self.run_task(task, notify, sender); trace!("try_steal_task -- signal_work; self={}; from={}", - self.idx, idx); + self.id.idx, idx); // Signal other workers that work is available self.inner.signal_work(&self.inner); @@ -370,7 +387,7 @@ impl Worker { /// /// Returns `true` if woken up due to new work arriving. fn sleep(&self) -> bool { - trace!("Worker::sleep; idx={}", self.idx); + trace!("Worker::sleep; idx={}", self.id.idx); let mut state: WorkerState = self.entry().state.load(Acquire).into(); @@ -409,12 +426,12 @@ impl Worker { if !state.is_pushed() { debug_assert!(next.is_pushed()); - trace!(" sleeping -- push to stack; idx={}", self.idx); + trace!(" sleeping -- push to stack; idx={}", self.id.idx); // We obtained permission to push the worker into the // sleeper queue. - if let Err(_) = self.inner.push_sleeper(self.idx) { - trace!(" sleeping -- push to stack failed; idx={}", self.idx); + if let Err(_) = self.inner.push_sleeper(self.id.idx) { + trace!(" sleeping -- push to stack failed; idx={}", self.id.idx); // The push failed due to the pool being terminated. // // This is true because the "work" being woken up for is @@ -429,7 +446,7 @@ impl Worker { state = actual; } - trace!(" -> starting to sleep; idx={}", self.idx); + trace!(" -> starting to sleep; idx={}", self.id.idx); let sleep_until = self.inner.config.keep_alive .map(|dur| Instant::now() + dur); @@ -465,7 +482,7 @@ impl Worker { } } - trace!(" -> wakeup; idx={}", self.idx); + trace!(" -> wakeup; idx={}", self.id.idx); // Reload the state state = self.entry().state.load(Acquire).into(); @@ -527,13 +544,13 @@ impl Worker { } fn entry(&self) -> &WorkerEntry { - &self.inner.workers[self.idx] + &self.inner.workers[self.id.idx] } } impl Drop for Worker { fn drop(&mut self) { - trace!("shutting down thread; idx={}", self.idx); + trace!("shutting down thread; idx={}", self.id.idx); if self.should_finalize.get() { // Drain all work @@ -547,3 +564,9 @@ impl Drop for Worker { } } } + +impl WorkerId { + pub(crate) fn new(idx: usize) -> WorkerId { + WorkerId { idx } + } +}