tokio-rs/tokio
1
0
Fork 0
mirror of https://github.com/tokio-rs/tokio.git synced 2026-08-29 00:00:11 +02:00
Code Issues Packages Projects Releases Wiki Activity

chore: apply rustfmt to all crates (#917)

Browse Source
This commit is contained in:
Carl Lerche
2019-02-21 11:56:15 -08:00
committed by GitHub
parent ab595d0825
commit 80162306e7
253 changed files with 3710 additions and 3407 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tokio-sync/src/lib.rs
+1 -1
View File
@@ -23,7 +23,7 @@ macro_rules! if_fuzz {
}
mod loom;
pub mod oneshot;
pub mod mpsc;
pub mod oneshot;
pub mod semaphore;
pub mod task;
tokio-sync/src/loom.rs
+1 -1
View File
@@ -1,6 +1,6 @@
pub(crate) mod futures {
pub(crate) use futures::task;
pub(crate) use ::task::AtomicTask;
pub(crate) use task::AtomicTask;
}
pub(crate) mod sync {
tokio-sync/src/mpsc/block.rs
+24 -30
View File
@@ -1,16 +1,13 @@
use loom::{
self,
sync::atomic::{AtomicPtr, AtomicUsize},
sync::CausalCell,
sync::atomic::{
AtomicPtr,
AtomicUsize,
},
};
use std::mem::{self, ManuallyDrop};
use std::ops;
use std::ptr::{self, NonNull};
use std::sync::atomic::Ordering::{self, Acquire, Release, AcqRel};
use std::sync::atomic::Ordering::{self, AcqRel, Acquire, Release};
/// A block in a linked list.
///
@@ -133,9 +130,7 @@ impl<T> Block<T> {
}
// Get the value
let value = self.values[offset].with(|ptr| {
ptr::read(ptr)
});
let value = self.values[offset].with(|ptr| ptr::read(ptr));
Some(Read::Value(ManuallyDrop::into_inner(value)))
}
@@ -195,7 +190,8 @@ impl<T> Block<T> {
pub(crate) unsafe fn tx_release(&self, tail_position: usize) {
// Track the observed tail_position. Any sender targetting a greater
// tail_position is guaranteed to not access this block.
self.observed_tail_position.with_mut(|ptr| *ptr = tail_position);
self.observed_tail_position
.with_mut(|ptr| *ptr = tail_position);
// Set the released bit, signalling to the receiver that it is safe to
// free the block's memory as soon as all slots **prior** to
@@ -238,9 +234,8 @@ impl<T> Block<T> {
let ret = NonNull::new(self.next.load(ordering));
debug_assert!(unsafe {
ret.map(|block| {
block.as_ref().start_index == self.start_index.wrapping_add(BLOCK_CAP)
}).unwrap_or(true)
ret.map(|block| block.as_ref().start_index == self.start_index.wrapping_add(BLOCK_CAP))
.unwrap_or(true)
});
ret
@@ -262,14 +257,16 @@ impl<T> Block<T> {
/// To maintain safety, the caller must ensure:
///
/// * `block` is not freed until it has been removed from the list.
pub(crate) unsafe fn try_push(&self, block: &mut NonNull<Block<T>>, ordering: Ordering)
-> Result<(), NonNull<Block<T>>>
{
block.as_mut().start_index =
self.start_index.wrapping_add(BLOCK_CAP);
pub(crate) unsafe fn try_push(
&self,
block: &mut NonNull<Block<T>>,
ordering: Ordering,
) -> Result<(), NonNull<Block<T>>> {
block.as_mut().start_index = self.start_index.wrapping_add(BLOCK_CAP);
let next_ptr = self.next.compare_and_swap(
ptr::null_mut(), block.as_ptr(), ordering);
let next_ptr = self
.next
.compare_and_swap(ptr::null_mut(), block.as_ptr(), ordering);
match NonNull::new(next_ptr) {
Some(next_ptr) => Err(next_ptr),
@@ -295,12 +292,9 @@ impl<T> Block<T> {
// Create the new block. It is assumed that the block will become the
// next one after `&self`. If this turns out to not be the case,
// `start_index` is updated accordingly.
let new_block = Box::new(
Block::new(self.start_index + BLOCK_CAP));
let new_block = Box::new(Block::new(self.start_index + BLOCK_CAP));
let mut new_block = unsafe {
NonNull::new_unchecked(Box::into_raw(new_block))
};
let mut new_block = unsafe { NonNull::new_unchecked(Box::into_raw(new_block)) };
// Attempt to store the block. The first compare-and-swap attempt is
// "unrolled" due to minor differences in logic
@@ -314,9 +308,11 @@ impl<T> Block<T> {
//
// `Release` ensures that the newly allocated block is available to
// other threads acquiring the next pointer.
let next = NonNull::new(
self.next.compare_and_swap(
ptr::null_mut(), new_block.as_ptr(), AcqRel));
let next = NonNull::new(self.next.compare_and_swap(
ptr::null_mut(),
new_block.as_ptr(),
AcqRel,
));
let next = match next {
Some(next) => next,
@@ -339,9 +335,7 @@ impl<T> Block<T> {
// TODO: Should this iteration be capped?
loop {
let actual = unsafe {
curr.as_ref().try_push(&mut new_block, AcqRel)
};
let actual = unsafe { curr.as_ref().try_push(&mut new_block, AcqRel) };
curr = match actual {
Ok(_) => {
tokio-sync/src/mpsc/bounded.rs
+8 -11
View File
@@ -13,7 +13,9 @@ pub struct Sender<T> {
impl<T> Clone for Sender<T> {
fn clone(&self) -> Self {
Sender { chan: self.chan.clone() }
Sender {
chan: self.chan.clone(),
}
}
}
@@ -144,12 +146,10 @@ impl<T> Stream for Receiver<T> {
type Error = RecvError;
fn poll(&mut self) -> Poll<Option<T>, Self::Error> {
self.chan.recv()
.map_err(|_| RecvError(()))
self.chan.recv().map_err(|_| RecvError(()))
}
}
impl<T> Sender<T> {
pub(crate) fn new(chan: chan::Tx<T, Semaphore>) -> Sender<T> {
Sender { chan }
@@ -176,8 +176,7 @@ impl<T> Sender<T> {
/// capacity is available;
/// - `Err(SendError)` if the receiver has been dropped.
pub fn poll_ready(&mut self) -> Poll<(), SendError> {
self.chan.poll_ready()
.map_err(|_| SendError(()))
self.chan.poll_ready().map_err(|_| SendError(()))
}
/// Attempts to send a message on this `Sender`, returning the message
@@ -193,17 +192,15 @@ impl<T> Sink for Sender<T> {
type SinkError = SendError;
fn start_send(&mut self, msg: T) -> StartSend<T, Self::SinkError> {
use futures::AsyncSink;
use futures::Async::*;
use futures::AsyncSink;
match self.poll_ready()? {
Ready(_) => {
self.try_send(msg).map_err(|_| SendError(()))?;
Ok(AsyncSink::Ready)
}
NotReady => {
Ok(AsyncSink::NotReady(msg))
}
NotReady => Ok(AsyncSink::NotReady(msg)),
}
}
@@ -283,7 +280,7 @@ impl<T> From<(T, chan::TrySendError)> for TrySendError<T> {
kind: match err {
chan::TrySendError::Closed => ErrorKind::Closed,
chan::TrySendError::NoPermits => ErrorKind::NoCapacity,
}
},
}
}
}
tokio-sync/src/mpsc/chan.rs
+25 -28
View File
@@ -1,14 +1,14 @@
use super::list;
use futures::Poll;
use ::loom::{
use loom::{
futures::AtomicTask,
sync::{Arc, CausalCell},
sync::atomic::AtomicUsize,
sync::{Arc, CausalCell},
};
use std::process;
use std::fmt;
use std::process;
use std::sync::atomic::Ordering::{AcqRel, Relaxed};
/// Channel sender
@@ -18,8 +18,9 @@ pub(crate) struct Tx<T, S: Semaphore> {
}
impl<T, S: Semaphore> fmt::Debug for Tx<T, S>
where S::Permit: fmt::Debug,
S: fmt::Debug
where
S::Permit: fmt::Debug,
S: fmt::Debug,
{
fn fmt(&self, fmt: &mut fmt::Formatter) -> fmt::Result {
fmt.debug_struct("Tx")
@@ -35,12 +36,11 @@ pub(crate) struct Rx<T, S: Semaphore> {
}
impl<T, S: Semaphore> fmt::Debug for Rx<T, S>
where S: fmt::Debug
where
S: fmt::Debug,
{
fn fmt(&self, fmt: &mut fmt::Formatter) -> fmt::Result {
fmt.debug_struct("Rx")
.field("inner", &self.inner)
.finish()
fmt.debug_struct("Rx").field("inner", &self.inner).finish()
}
}
@@ -95,7 +95,9 @@ struct Chan<T, S> {
rx_fields: CausalCell<RxFields<T>>,
}
impl<T, S> fmt::Debug for Chan<T, S> where S: fmt::Debug
impl<T, S> fmt::Debug for Chan<T, S>
where
S: fmt::Debug,
{
fn fmt(&self, fmt: &mut fmt::Formatter) -> fmt::Result {
fmt.debug_struct("Chan")
@@ -117,8 +119,7 @@ struct RxFields<T> {
rx_closed: bool,
}
impl<T> fmt::Debug for RxFields<T>
{
impl<T> fmt::Debug for RxFields<T> {
fn fmt(&self, fmt: &mut fmt::Formatter) -> fmt::Result {
fmt.debug_struct("RxFields")
.field("list", &self.list)
@@ -273,7 +274,7 @@ where
}
None => {} // fall through
}
}
};
}
try_recv!();
@@ -285,8 +286,11 @@ where
// second time here.
try_recv!();
debug!("recv; rx_closed = {:?}; is_idle = {:?}",
rx_fields.rx_closed, self.inner.semaphore.is_idle());
debug!(
"recv; rx_closed = {:?}; is_idle = {:?}",
rx_fields.rx_closed,
self.inner.semaphore.is_idle()
);
if rx_fields.rx_closed && self.inner.semaphore.is_idle() {
Ok(Ready(None))
@@ -325,8 +329,7 @@ impl<T, S> Drop for Chan<T, S> {
self.rx_fields.with_mut(|rx_fields_ptr| {
let rx_fields = unsafe { &mut *rx_fields_ptr };
while let Some(Value(_)) = rx_fields.list.pop(&self.tx) {
}
while let Some(Value(_)) = rx_fields.list.pop(&self.tx) {}
});
}
}
@@ -369,8 +372,7 @@ impl Semaphore for (::semaphore::Semaphore, usize) {
}
fn poll_acquire(&self, permit: &mut Permit) -> Poll<(), ()> {
permit.poll_acquire(&self.0)
.map_err(|_| ())
permit.poll_acquire(&self.0).map_err(|_| ())
}
fn try_acquire(&self, permit: &mut Permit) -> Result<(), TrySendError> {
@@ -395,11 +397,9 @@ use std::usize;
impl Semaphore for AtomicUsize {
type Permit = ();
fn new_permit() {
}
fn new_permit() {}
fn drop_permit(&self, _permit: &mut ()) {
}
fn drop_permit(&self, _permit: &mut ()) {}
fn add_permit(&self) {
let prev = self.fetch_sub(2, Release);
@@ -416,9 +416,7 @@ impl Semaphore for AtomicUsize {
fn poll_acquire(&self, permit: &mut ()) -> Poll<(), ()> {
use futures::Async::Ready;
self.try_acquire(permit)
.map(Ready)
.map_err(|_| ())
self.try_acquire(permit).map(Ready).map_err(|_| ())
}
fn try_acquire(&self, _permit: &mut ()) -> Result<(), TrySendError> {
@@ -444,8 +442,7 @@ impl Semaphore for AtomicUsize {
}
}
fn forget(&self, _permit: &mut ()) {
}
fn forget(&self, _permit: &mut ()) {}
fn close(&self) {
self.fetch_or(1, Release);
tokio-sync/src/mpsc/list.rs
+17 -20
View File
@@ -4,12 +4,12 @@ use super::block::{self, Block};
use loom::{
self,
sync::atomic::{AtomicUsize, AtomicPtr},
sync::atomic::{AtomicPtr, AtomicUsize},
};
use std::fmt;
use std::ptr::NonNull;
use std::sync::atomic::Ordering::{Acquire, Release, AcqRel, Relaxed};
use std::sync::atomic::Ordering::{AcqRel, Acquire, Relaxed, Release};
/// List queue transmit handle
pub(crate) struct Tx<T> {
@@ -59,8 +59,7 @@ impl<T> Tx<T> {
pub(crate) fn push(&self, value: T) {
// First, claim a slot for the value. `Acquire` is used here to
// synchronize with the `fetch_add` in `free_blocks`.
let slot_index = self.tail_position
.fetch_add(1, Acquire);
let slot_index = self.tail_position.fetch_add(1, Acquire);
// Load the current block and write the value
let block = self.find_block(slot_index);
@@ -78,14 +77,11 @@ impl<T> Tx<T> {
pub(crate) fn close(&self) {
// First, claim a slot for the value. This is the last slot that will be
// claimed.
let slot_index = self.tail_position
.fetch_add(1, Acquire);
let slot_index = self.tail_position.fetch_add(1, Acquire);
let block = self.find_block(slot_index);
unsafe {
block.as_ref().tx_close()
}
unsafe { block.as_ref().tx_close() }
}
fn find_block(&self, slot_index: usize) -> NonNull<Block<T>> {
@@ -123,7 +119,8 @@ impl<T> Tx<T> {
return unsafe { NonNull::new_unchecked(block_ptr) };
}
let next_block = block.load_next(Acquire)
let next_block = block
.load_next(Acquire)
// There is no allocated next block, grow the linked list.
.unwrap_or_else(|| block.grow());
@@ -146,15 +143,17 @@ impl<T> Tx<T> {
//
// Acquire is not needed as any "actual" value is not accessed.
// At this point, the linked list is walked to acquire blocks.
let actual = self.block_tail.compare_and_swap(
block_ptr, next_block.as_ptr(), Release);
let actual =
self.block_tail
.compare_and_swap(block_ptr, next_block.as_ptr(), Release);
if actual == block_ptr {
// Synchronize with any senders
let tail_position =
self.tail_position.fetch_add(0, Release);
let tail_position = self.tail_position.fetch_add(0, Release);
unsafe { block.tx_release(tail_position); }
unsafe {
block.tx_release(tail_position);
}
} else {
// A concurrent sender is also working on advancing
// `block_tail` and this thread is falling behind.
@@ -207,7 +206,7 @@ impl<T> Tx<T> {
}
if !reused {
let _ = Box::from_raw(block.as_ptr());
let _ = Box::from_raw(block.as_ptr());
}
}
}
@@ -286,8 +285,7 @@ impl<T> Rx<T> {
// `free_head` to point to the next block.
let block = self.free_head;
let observed_tail_position =
block.as_ref().observed_tail_position();
let observed_tail_position = block.as_ref().observed_tail_position();
let required_index = match observed_tail_position {
Some(i) => i,
@@ -302,8 +300,7 @@ impl<T> Rx<T> {
// guaranteed that the `free_blocks` routine trails the `recv`
// routine. Any memory accessed by `free_blocks` has already
// been acquired by `recv`.
let next_block =
block.as_ref().load_next(Relaxed);
let next_block = block.as_ref().load_next(Relaxed);
// Update the free list head
self.free_head = next_block.unwrap();
tokio-sync/src/mpsc/mod.rs
+4 -20
View File
@@ -40,32 +40,16 @@ mod chan;
mod list;
mod unbounded;
pub use self::bounded::{
channel,
Receiver,
Sender
};
pub use self::bounded::{channel, Receiver, Sender};
pub use self::unbounded::{
unbounded_channel,
UnboundedReceiver,
UnboundedSender,
};
pub use self::unbounded::{unbounded_channel, UnboundedReceiver, UnboundedSender};
pub mod error {
//! Channel error types
pub use super::bounded::{
SendError,
TrySendError,
RecvError,
};
pub use super::bounded::{RecvError, SendError, TrySendError};
pub use super::unbounded::{
UnboundedSendError,
UnboundedTrySendError,
UnboundedRecvError,
};
pub use super::unbounded::{UnboundedRecvError, UnboundedSendError, UnboundedTrySendError};
}
/// The number of values a block can contain.
tokio-sync/src/mpsc/unbounded.rs
+6 -8
View File
@@ -1,7 +1,7 @@
use super::chan;
use loom::sync::atomic::AtomicUsize;
use futures::{Poll, Sink, StartSend, Stream};
use loom::sync::atomic::AtomicUsize;
use std::fmt;
@@ -15,7 +15,9 @@ pub struct UnboundedSender<T> {
impl<T> Clone for UnboundedSender<T> {
fn clone(&self) -> Self {
UnboundedSender { chan: self.chan.clone() }
UnboundedSender {
chan: self.chan.clone(),
}
}
}
@@ -97,21 +99,17 @@ impl<T> Stream for UnboundedReceiver<T> {
type Error = UnboundedRecvError;
fn poll(&mut self) -> Poll<Option<T>, Self::Error> {
self.chan.recv()
.map_err(|_| UnboundedRecvError(()))
self.chan.recv().map_err(|_| UnboundedRecvError(()))
}
}
impl<T> UnboundedSender<T> {
pub(crate) fn new(chan: chan::Tx<T, Semaphore>) -> UnboundedSender<T> {
UnboundedSender { chan }
}
/// Attempts to send a message on this `UnboundedSender` without blocking.
pub fn try_send(&mut self, message: T)
-> Result<(), UnboundedTrySendError<T>>
{
pub fn try_send(&mut self, message: T) -> Result<(), UnboundedTrySendError<T>> {
self.chan.try_send(message)?;
Ok(())
}
tokio-sync/src/oneshot.rs
+39 -50
View File
@@ -2,16 +2,16 @@
use loom::{
futures::task::{self, Task},
sync::CausalCell,
sync::atomic::AtomicUsize,
sync::CausalCell,
};
use futures::{Async, Future, Poll};
use std::fmt;
use std::mem::{self, ManuallyDrop};
use std::sync::atomic::Ordering::{self, AcqRel, Acquire};
use std::sync::Arc;
use std::sync::atomic::Ordering::{self, Acquire, AcqRel};
/// Sends a value to the associated `Receiver`.
///
@@ -102,7 +102,9 @@ pub fn channel<T>() -> (Sender<T>, Receiver<T>) {
rx_task: CausalCell::new(ManuallyDrop::new(unsafe { mem::uninitialized() })),
});
let tx = Sender { inner: Some(inner.clone()) };
let tx = Sender {
inner: Some(inner.clone()),
};
let rx = Receiver { inner: Some(inner) };
(tx, rx)
@@ -121,14 +123,14 @@ impl<T> Sender<T> {
pub fn send(mut self, t: T) -> Result<(), T> {
let inner = self.inner.take().unwrap();
inner.value.with_mut(|ptr| {
unsafe { *ptr = Some(t); }
inner.value.with_mut(|ptr| unsafe {
*ptr = Some(t);
});
if !inner.complete() {
return Err(inner.value.with_mut(|ptr| {
unsafe { (*ptr).take() }.unwrap()
}));
return Err(inner
.value
.with_mut(|ptr| unsafe { (*ptr).take() }.unwrap()));
}
Ok(())
@@ -156,9 +158,9 @@ impl<T> Sender<T> {
}
if state.is_tx_task_set() {
let will_notify = inner.tx_task.with(|ptr| unsafe {
(&*ptr).will_notify_current()
});
let will_notify = inner
.tx_task
.with(|ptr| unsafe { (&*ptr).will_notify_current() });
if !will_notify {
state = State::unset_tx_task(&inner.state);
@@ -173,7 +175,9 @@ impl<T> Sender<T> {
if !state.is_tx_task_set() {
// Attempt to set the task
unsafe { inner.set_tx_task(); }
unsafe {
inner.set_tx_task();
}
// Update the state
state = State::set_tx_task(&inner.state);
@@ -186,7 +190,6 @@ impl<T> Sender<T> {
Ok(Async::NotReady)
}
/// Check if the associated [`Receiver`] handle has been dropped.
///
/// Unlike [`poll_close`], this function does not register a task for
@@ -271,7 +274,7 @@ impl<T> Future for Receiver<T> {
type Error = RecvError;
fn poll(&mut self) -> Poll<T, RecvError> {
use futures::Async::{Ready, NotReady};
use futures::Async::{NotReady, Ready};
// If `inner` is `None`, then `poll()` has already completed.
let ret = if let Some(inner) = self.inner.as_ref() {
@@ -298,16 +301,14 @@ impl<T> Inner<T> {
}
if prev.is_rx_task_set() {
self.rx_task.with(|ptr| unsafe {
(&*ptr).notify()
});
self.rx_task.with(|ptr| unsafe { (&*ptr).notify() });
}
true
}
fn poll_recv(&self) -> Poll<T, RecvError> {
use futures::Async::{Ready, NotReady};
use futures::Async::{NotReady, Ready};
// Load the state
let mut state = State::load(&self.state, Acquire);
@@ -321,9 +322,9 @@ impl<T> Inner<T> {
Err(RecvError(()))
} else {
if state.is_rx_task_set() {
let will_notify = self.rx_task.with(|ptr| unsafe {
(&*ptr).will_notify_current()
});
let will_notify = self
.rx_task
.with(|ptr| unsafe { (&*ptr).will_notify_current() });
// Check if the task is still the same
if !will_notify {
@@ -342,7 +343,9 @@ impl<T> Inner<T> {
if !state.is_rx_task_set() {
// Attempt to set the task
unsafe { self.set_rx_task(); }
unsafe {
self.set_rx_task();
}
// Update the state
state = State::set_rx_task(&self.state);
@@ -366,41 +369,31 @@ impl<T> Inner<T> {
let prev = State::set_closed(&self.state);
if prev.is_tx_task_set() && !prev.is_complete() {
self.tx_task.with(|ptr| unsafe {
(&*ptr).notify()
});
self.tx_task.with(|ptr| unsafe { (&*ptr).notify() });
}
}
/// Consume the value. This function does not check `state`.
unsafe fn consume_value(&self) -> Option<T> {
self.value.with_mut(|ptr| {
(*ptr).take()
})
self.value.with_mut(|ptr| (*ptr).take())
}
unsafe fn drop_rx_task(&self) {
self.rx_task.with_mut(|ptr| {
ManuallyDrop::drop(&mut *ptr)
})
self.rx_task.with_mut(|ptr| ManuallyDrop::drop(&mut *ptr))
}
unsafe fn drop_tx_task(&self) {
self.tx_task.with_mut(|ptr| {
ManuallyDrop::drop(&mut *ptr)
})
self.tx_task.with_mut(|ptr| ManuallyDrop::drop(&mut *ptr))
}
unsafe fn set_rx_task(&self) {
self.rx_task.with_mut(|ptr| {
*ptr = ManuallyDrop::new(task::current())
});
self.rx_task
.with_mut(|ptr| *ptr = ManuallyDrop::new(task::current()));
}
unsafe fn set_tx_task(&self) {
self.tx_task.with_mut(|ptr| {
*ptr = ManuallyDrop::new(task::current())
});
self.tx_task
.with_mut(|ptr| *ptr = ManuallyDrop::new(task::current()));
}
}
@@ -412,18 +405,14 @@ impl<T> Drop for Inner<T> {
let state = State(*self.state.get_mut());
if state.is_rx_task_set() {
self.rx_task.with_mut(|ptr| {
unsafe {
ManuallyDrop::drop(&mut *ptr);
}
self.rx_task.with_mut(|ptr| unsafe {
ManuallyDrop::drop(&mut *ptr);
});
}
if state.is_tx_task_set() {
self.tx_task.with_mut(|ptr| {
unsafe {
ManuallyDrop::drop(&mut *ptr);
}
self.tx_task.with_mut(|ptr| unsafe {
ManuallyDrop::drop(&mut *ptr);
});
}
}
@@ -440,8 +429,8 @@ impl<T: fmt::Debug> fmt::Debug for Inner<T> {
}
const RX_TASK_SET: usize = 0b00001;
const VALUE_SENT: usize = 0b00010;
const CLOSED: usize = 0b00100;
const VALUE_SENT: usize = 0b00010;
const CLOSED: usize = 0b00100;
const TX_TASK_SET: usize = 0b01000;
impl State {
tokio-sync/src/semaphore.rs
+65 -71
View File
@@ -11,8 +11,8 @@
use loom::{
futures::AtomicTask,
sync::{
atomic::{AtomicPtr, AtomicUsize},
CausalCell,
atomic::{AtomicUsize, AtomicPtr},
},
yield_now,
};
@@ -21,8 +21,8 @@ use futures::Poll;
use std::fmt;
use std::ptr::{self, NonNull};
use std::sync::atomic::Ordering::{self, AcqRel, Acquire, Relaxed, Release};
use std::sync::Arc;
use std::sync::atomic::Ordering::{self, Acquire, Release, AcqRel, Relaxed};
use std::usize;
/// Futures-aware semaphore.
@@ -176,9 +176,7 @@ impl Semaphore {
}
/// Poll for a permit
fn poll_permit(&self, mut permit: Option<&mut Permit>)
-> Poll<(), AcquireError>
{
fn poll_permit(&self, mut permit: Option<&mut Permit>) -> Poll<(), AcquireError> {
use futures::Async::*;
// Load the current state
@@ -201,7 +199,7 @@ impl Semaphore {
let waiter = unsafe { Arc::from_raw(waiter.as_ptr()) };
waiter.revert_to_idle();
}
}
};
}
loop {
@@ -220,7 +218,8 @@ impl Semaphore {
if maybe_strong.is_none() {
if let Some(ref mut permit) = permit {
// Get the Sender's waiter node, or initialize one
let waiter = permit.waiter
let waiter = permit
.waiter
.get_or_insert_with(|| Arc::new(WaiterNode::new()));
waiter.register();
@@ -259,8 +258,7 @@ impl Semaphore {
// Finish pushing
unsafe {
prev_waiter.as_ref()
.next.store(waiter.as_ptr(), Release);
prev_waiter.as_ref().next.store(waiter.as_ptr(), Release);
}
debug!(" + poll_permit -- waiter pushed");
@@ -327,8 +325,10 @@ impl Semaphore {
fn add_permits_locked(&self, mut rem: usize, mut closed: bool) {
while rem > 0 || closed {
debug!(" + add_permits_locked -- iter; rem = {}; closed = {:?}",
rem, closed);
debug!(
" + add_permits_locked -- iter; rem = {}; closed = {:?}",
rem, closed
);
if closed {
SemState::fetch_set_closed(&self.state, AcqRel);
@@ -341,13 +341,19 @@ impl Semaphore {
let actual = if closed {
let actual = self.rx_lock.fetch_sub(n | 1, AcqRel);
debug!(" + add_permits_locked; rx_lock.fetch_sub(n | 1); n = {}; actual={}", n, actual);
debug!(
" + add_permits_locked; rx_lock.fetch_sub(n | 1); n = {}; actual={}",
n, actual
);
closed = false;
actual
} else {
let actual = self.rx_lock.fetch_sub(n, AcqRel);
debug!(" + add_permits_locked; rx_lock.fetch_sub(n); n = {}; actual={}", n, actual);
debug!(
" + add_permits_locked; rx_lock.fetch_sub(n); n = {}; actual={}",
n, actual
);
closed = actual & 1 == 1;
actual
@@ -389,8 +395,7 @@ impl Semaphore {
fn pop(&self, rem: usize, closed: bool) -> Option<Arc<WaiterNode>> {
debug!(" + pop; rem = {}", rem);
'outer:
loop {
'outer: loop {
unsafe {
let mut head = self.head.with(|head| *head);
let mut next_ptr = head.as_ref().next.load(Acquire);
@@ -502,12 +507,10 @@ impl Semaphore {
// operation
stub.as_ref().next.store(ptr::null_mut(), Relaxed);
// Update the tail to point to the new node. We need to see the previous
// node in order to update the next pointer as well as release `task`
// to any other threads calling `push`.
let prev = SemState::new_ptr(stub, closed)
.swap(&self.state, AcqRel);
let prev = SemState::new_ptr(stub, closed).swap(&self.state, AcqRel);
debug_assert_eq!(closed, prev.is_closed());
@@ -523,9 +526,7 @@ impl Semaphore {
}
fn stub(&self) -> NonNull<WaiterNode> {
unsafe {
NonNull::new_unchecked(&*self.stub as *const _ as *mut _)
}
unsafe { NonNull::new_unchecked(&*self.stub as *const _ as *mut _) }
}
}
@@ -574,9 +575,7 @@ impl Permit {
/// Try to acquire the permit. If no permits are available, the current task
/// is notified once a new permit becomes available.
pub fn poll_acquire(&mut self, semaphore: &Semaphore)
-> Poll<(), AcquireError>
{
pub fn poll_acquire(&mut self, semaphore: &Semaphore) -> Poll<(), AcquireError> {
use futures::Async::*;
match self.state {
@@ -609,9 +608,7 @@ impl Permit {
}
/// Try to acquire the permit.
pub fn try_acquire(&mut self, semaphore: &Semaphore)
-> Result<(), TryAcquireError>
{
pub fn try_acquire(&mut self, semaphore: &Semaphore) -> Result<(), TryAcquireError> {
use futures::Async::*;
match self.state {
@@ -636,9 +633,7 @@ impl Permit {
self.state = PermitState::Acquired;
Ok(())
}
NotReady => {
Err(TryAcquireError::no_permits())
}
NotReady => Err(TryAcquireError::no_permits()),
}
}
@@ -665,10 +660,7 @@ impl Permit {
match self.state {
PermitState::Idle => false,
PermitState::Waiting => {
let ret = self.waiter
.as_ref()
.unwrap()
.cancel_interest();
let ret = self.waiter.as_ref().unwrap().cancel_interest();
self.state = PermitState::Idle;
ret
}
@@ -709,11 +701,15 @@ impl ::std::error::Error for AcquireError {
impl TryAcquireError {
fn closed() -> TryAcquireError {
TryAcquireError { kind: ErrorKind::Closed }
TryAcquireError {
kind: ErrorKind::Closed,
}
}
fn no_permits() -> TryAcquireError {
TryAcquireError { kind: ErrorKind::NoPermits }
TryAcquireError {
kind: ErrorKind::NoPermits,
}
}
/// Returns true if the error was caused by a closed semaphore.
@@ -865,22 +861,18 @@ impl WaiterNode {
};
match next.compare_exchange(&self.state, curr, AcqRel, Acquire) {
Ok(_) => {
match curr {
QueuedWaiting => {
debug!(" + notify -- task notified");
self.task.notify();
return true;
}
other => {
debug!(" + notify -- not notified; state = {:?}", other);
return false;
}
Ok(_) => match curr {
QueuedWaiting => {
debug!(" + notify -- task notified");
self.task.notify();
return true;
}
}
Err(actual) => {
curr = actual
}
other => {
debug!(" + notify -- not notified; state = {:?}", other);
return false;
}
},
Err(actual) => curr = actual,
}
}
}
@@ -1003,8 +995,7 @@ impl SemState {
/// Returns the waiter, if one is set.
fn waiter(&self) -> Option<NonNull<WaiterNode>> {
if self.is_waiter() {
let waiter = NonNull::new(self.as_ptr())
.expect("null pointer stored");
let waiter = NonNull::new(self.as_ptr()).expect("null pointer stored");
Some(waiter)
} else {
@@ -1047,22 +1038,25 @@ impl SemState {
}
/// Compare and exchange the current value into the provided cell
fn compare_exchange(&self,
cell: &AtomicUsize,
prev: SemState,
success: Ordering,
failure: Ordering)
-> Result<SemState, SemState>
{
fn compare_exchange(
&self,
cell: &AtomicUsize,
prev: SemState,
success: Ordering,
failure: Ordering,
) -> Result<SemState, SemState> {
debug_assert_eq!(prev.is_closed(), self.is_closed());
let res = cell.compare_exchange(prev.to_usize(), self.to_usize(), success, failure);
debug!(" + SemState::compare_exchange; prev = {}; next = {}; result = {:?}",
prev.to_usize(), self.to_usize(), res);
debug!(
" + SemState::compare_exchange; prev = {}; next = {}; result = {:?}",
prev.to_usize(),
self.to_usize(),
res
);
res.map(SemState)
.map_err(SemState)
res.map(SemState).map_err(SemState)
}
fn fetch_set_closed(cell: &AtomicUsize, ordering: Ordering) -> SemState {
@@ -1123,13 +1117,13 @@ impl NodeState {
cell.store(value.to_usize(), ordering);
}
fn compare_exchange(&self,
cell: &AtomicUsize,
prev: NodeState,
success: Ordering,
failure: Ordering)
-> Result<NodeState, NodeState>
{
fn compare_exchange(
&self,
cell: &AtomicUsize,
prev: NodeState,
success: Ordering,
failure: Ordering,
) -> Result<NodeState, NodeState> {
cell.compare_exchange(prev.to_usize(), self.to_usize(), success, failure)
.map(NodeState::from_usize)
.map_err(NodeState::from_usize)
tokio-sync/src/task/atomic_task.rs
+11 -12
View File
@@ -1,11 +1,11 @@
use ::loom::{
use loom::{
futures::task::{self, Task},
sync::CausalCell,
sync::atomic::AtomicUsize,
sync::CausalCell,
};
use std::fmt;
use std::sync::atomic::Ordering::{Acquire, Release, AcqRel};
use std::sync::atomic::Ordering::{AcqRel, Acquire, Release};
/// A synchronization primitive for task notification.
///
@@ -189,8 +189,9 @@ impl AtomicTask {
//
// Start by assuming that the state is `REGISTERING` as this
// is what we jut set it to.
let res = self.state.compare_exchange(
REGISTERING, WAITING, AcqRel, Acquire);
let res = self
.state
.compare_exchange(REGISTERING, WAITING, AcqRel, Acquire);
match res {
Ok(_) => {}
@@ -230,9 +231,7 @@ impl AtomicTask {
//
// We just want to maintain memory safety. It is ok to drop the
// call to `register`.
debug_assert!(
state == REGISTERING ||
state == REGISTERING | NOTIFYING);
debug_assert!(state == REGISTERING || state == REGISTERING | NOTIFYING);
}
}
}
@@ -276,9 +275,8 @@ impl AtomicTask {
// not.
//
debug_assert!(
state == REGISTERING ||
state == REGISTERING | NOTIFYING ||
state == NOTIFYING);
state == REGISTERING || state == REGISTERING | NOTIFYING || state == NOTIFYING
);
None
}
}
@@ -309,7 +307,8 @@ struct CurrentTask;
impl Register for CurrentTask {
fn register(self, slot: &mut Option<Task>) {
let should_update = (&*slot).as_ref()
let should_update = (&*slot)
.as_ref()
.map(|prev| !prev.will_notify_current())
.unwrap_or(true);
if should_update {