Skip to content

Fix a race in thread wakeup #459

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
carllerche merged 1 commit into from Jul 3, 2018
Merged

Fix a race in thread wakeup #459

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
16 changes: 3 additions & 13 deletions tokio-threadpool/src/builder.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -41,7 +41,6 @@ use futures2;
/// use std::time::Duration;
///
/// # pub fn main() {
/// // Create a thread pool with default configuration values
Copy link
Author

@ghost ghost Jun 29, 2018

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I found this comment confusing. It seems to imply that the example shows how the builder is configured by default. However, this comment applies only to Builder::new(), while the chained methods do custom configuration (which is different from default).

Perhaps a different sentence would work better, but due to lack of ideas I've decided to simply remove it. :)

Copy link
Member

@carllerche carllerche Jul 3, 2018

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

👍

/// let thread_pool = Builder::new()
/// .pool_size(4)
/// .keep_alive(Some(Duration::from_secs(30)))
Expand Down Expand Up @@ -86,7 +85,6 @@ impl Builder {
/// use std::time::Duration;
///
/// # pub fn main() {
/// // Create a thread pool with default configuration values
/// let thread_pool = Builder::new()
/// .pool_size(4)
/// .keep_alive(Some(Duration::from_secs(30)))
Expand Down Expand Up @@ -131,7 +129,6 @@ impl Builder {
/// # use tokio_threadpool::Builder;
///
/// # pub fn main() {
/// // Create a thread pool with default configuration values
/// let thread_pool = Builder::new()
/// .pool_size(4)
/// .build();
Expand Down Expand Up @@ -164,7 +161,6 @@ impl Builder {
/// # use tokio_threadpool::Builder;
///
/// # pub fn main() {
/// // Create a thread pool with default configuration values
/// let thread_pool = Builder::new()
/// .max_blocking(200)
/// .build();
Expand Down Expand Up @@ -196,7 +192,6 @@ impl Builder {
/// use std::time::Duration;
///
/// # pub fn main() {
/// // Create a thread pool with default configuration values
/// let thread_pool = Builder::new()
/// .keep_alive(Some(Duration::from_secs(30)))
/// .build();
Expand Down Expand Up @@ -224,7 +219,6 @@ impl Builder {
/// # use tokio_threadpool::Builder;
///
/// # pub fn main() {
/// // Create a thread pool with default configuration values
/// let thread_pool = Builder::new()
/// .name_prefix("my-pool-")
/// .build();
Expand All @@ -251,7 +245,6 @@ impl Builder {
/// # use tokio_threadpool::Builder;
///
/// # pub fn main() {
/// // Create a thread pool with default configuration values
/// let thread_pool = Builder::new()
/// .stack_size(32 * 1024)
/// .build();
Expand All @@ -265,7 +258,7 @@ impl Builder {
/// Execute function `f` on each worker thread.
///
/// This function is provided a handle to the worker and is expected to call
/// `Worker::run`, otherwise the worker thread will shutdown without doing
/// [`Worker::run`], otherwise the worker thread will shutdown without doing
/// any work.
///
/// # Examples
Expand All @@ -276,7 +269,6 @@ impl Builder {
/// # use tokio_threadpool::Builder;
///
/// # pub fn main() {
/// // Create a thread pool with default configuration values
/// let thread_pool = Builder::new()
/// .around_worker(|worker, _| {
/// println!("worker is starting up");
Expand All @@ -286,6 +278,8 @@ impl Builder {
/// .build();
/// # }
/// ```
///
/// [`Worker::run`]: struct.Worker.html#method.run
pub fn around_worker<F>(&mut self, f: F) -> &mut Self
where F: Fn(&Worker, &mut Enter) + Send + Sync + 'static
{
Expand All @@ -306,7 +300,6 @@ impl Builder {
/// # use tokio_threadpool::Builder;
///
/// # pub fn main() {
/// // Create a thread pool with default configuration values
/// let thread_pool = Builder::new()
/// .after_start(|| {
/// println!("thread started");
Expand All @@ -333,7 +326,6 @@ impl Builder {
/// # use tokio_threadpool::Builder;
///
/// # pub fn main() {
/// // Create a thread pool with default configuration values
/// let thread_pool = Builder::new()
/// .before_stop(|| {
/// println!("thread stopping");
Expand Down Expand Up @@ -362,7 +354,6 @@ impl Builder {
/// # fn decorate<F>(f: F) -> F { f }
///
/// # pub fn main() {
/// // Create a thread pool with default configuration values
/// let thread_pool = Builder::new()
/// .custom_park(|_| {
/// use tokio_threadpool::park::DefaultPark;
Expand Down Expand Up @@ -402,7 +393,6 @@ impl Builder {
/// # use tokio_threadpool::Builder;
///
/// # pub fn main() {
/// // Create a thread pool with default configuration values
/// let thread_pool = Builder::new()
/// .build();
/// # }
Expand Down
12 changes: 7 additions & 5 deletions tokio-threadpool/src/park/default_park.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -133,8 +133,8 @@ impl Inner {
None => self.condvar.wait(m).unwrap(),
};

// Transition back to idle. If the state has transitions dto `NOTIFY`,
// this will consume that notification
// Transition back to idle. If the state has transitioned to `NOTIFY`,
// this will consume that notification.
self.state.store(IDLE, SeqCst);

// Explicitly drop the mutex guard. There is no real point in doing it
Expand All @@ -155,10 +155,12 @@ impl Inner {
// The other half is sleeping, this requires a lock
let _m = self.mutex.lock().unwrap();

// Transition from SLEEP -> NOTIFY
match self.state.compare_and_swap(SLEEP, NOTIFY, SeqCst) {
// Transition to NOTIFY
match self.state.swap(NOTIFY, SeqCst) {
SLEEP => {}
_ => return,
NOTIFY => return,
IDLE => return,
_ => unreachable!(),
}

// Wakeup the sleeper
Expand Down
8 changes: 6 additions & 2 deletions tokio-threadpool/src/worker/mod.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -29,8 +29,12 @@ use std::time::{Duration, Instant};

/// Thread worker
///
/// This is passed to the `around_worker` callback set on `Builder`. This
/// callback is only expected to call `run` on it.
/// This is passed to the [`around_worker`] callback set on [`Builder`]. This
/// callback is only expected to call [`run`] on it.
///
/// [`Builder`]: struct.Builder.html
/// [`around_worker`]: struct.Builder.html#method.around_worker
/// [`run`]: struct.Worker.html#method.run
#[derive(Debug)]
pub struct Worker {
// Shared scheduler data
Expand Down