From 4436c9d35498e7ae3da261f6141d6d73b915e1e8 Mon Sep 17 00:00:00 2001 From: David Tolnay Date: Wed, 27 Nov 2019 10:29:00 -0800 Subject: Format libstd with rustfmt This commit applies rustfmt with rust-lang/rust's default settings to files in src/libstd *that are not involved in any currently open PR* to minimize merge conflicts. THe list of files involved in open PRs was determined by querying GitHub's GraphQL API with this script: https://gist.github.com/dtolnay/aa9c34993dc051a4f344d1b10e4487e8 With the list of files from the script in outstanding_files, the relevant commands were: $ find src/libstd -name '*.rs' \ | xargs rustfmt --edition=2018 --unstable-features --skip-children $ rg libstd outstanding_files | xargs git checkout -- Repeating this process several months apart should get us coverage of most of the rest of libstd. To confirm no funny business: $ git checkout $THIS_COMMIT^ $ git show --pretty= --name-only $THIS_COMMIT \ | xargs rustfmt --edition=2018 --unstable-features --skip-children $ git diff $THIS_COMMIT # there should be no difference --- src/libstd/sync/mpsc/sync.rs | 121 +++++++++++++++++++++++-------------------- 1 file changed, 66 insertions(+), 55 deletions(-) (limited to 'src/libstd/sync/mpsc/sync.rs') diff --git a/src/libstd/sync/mpsc/sync.rs b/src/libstd/sync/mpsc/sync.rs index 58a4b716afb..79e86817154 100644 --- a/src/libstd/sync/mpsc/sync.rs +++ b/src/libstd/sync/mpsc/sync.rs @@ -1,3 +1,4 @@ +use self::Blocker::*; /// Synchronous channels/ports /// /// This channel implementation differs significantly from the asynchronous @@ -22,17 +23,15 @@ /// implementation shares almost all code for the buffered and unbuffered cases /// of a synchronous channel. There are a few branches for the unbuffered case, /// but they're mostly just relevant to blocking senders. - pub use self::Failure::*; -use self::Blocker::*; use core::intrinsics::abort; use core::isize; use core::mem; use core::ptr; -use crate::sync::atomic::{Ordering, AtomicUsize}; -use crate::sync::mpsc::blocking::{self, WaitToken, SignalToken}; +use crate::sync::atomic::{AtomicUsize, Ordering}; +use crate::sync::mpsc::blocking::{self, SignalToken, WaitToken}; use crate::sync::{Mutex, MutexGuard}; use crate::time::Instant; @@ -46,9 +45,9 @@ pub struct Packet { lock: Mutex>, } -unsafe impl Send for Packet { } +unsafe impl Send for Packet {} -unsafe impl Sync for Packet { } +unsafe impl Sync for Packet {} struct State { disconnected: bool, // Is the channel disconnected yet? @@ -72,7 +71,7 @@ unsafe impl Send for State {} enum Blocker { BlockedSender(SignalToken), BlockedReceiver(SignalToken), - NoneBlocked + NoneBlocked, } /// Simple queue for threading threads together. Nodes are stack-allocated, so @@ -104,35 +103,35 @@ pub enum Failure { /// Atomically blocks the current thread, placing it into `slot`, unlocking `lock` /// in the meantime. This re-locks the mutex upon returning. -fn wait<'a, 'b, T>(lock: &'a Mutex>, - mut guard: MutexGuard<'b, State>, - f: fn(SignalToken) -> Blocker) - -> MutexGuard<'a, State> -{ +fn wait<'a, 'b, T>( + lock: &'a Mutex>, + mut guard: MutexGuard<'b, State>, + f: fn(SignalToken) -> Blocker, +) -> MutexGuard<'a, State> { let (wait_token, signal_token) = blocking::tokens(); match mem::replace(&mut guard.blocker, f(signal_token)) { NoneBlocked => {} _ => unreachable!(), } - drop(guard); // unlock - wait_token.wait(); // block + drop(guard); // unlock + wait_token.wait(); // block lock.lock().unwrap() // relock } /// Same as wait, but waiting at most until `deadline`. -fn wait_timeout_receiver<'a, 'b, T>(lock: &'a Mutex>, - deadline: Instant, - mut guard: MutexGuard<'b, State>, - success: &mut bool) - -> MutexGuard<'a, State> -{ +fn wait_timeout_receiver<'a, 'b, T>( + lock: &'a Mutex>, + deadline: Instant, + mut guard: MutexGuard<'b, State>, + success: &mut bool, +) -> MutexGuard<'a, State> { let (wait_token, signal_token) = blocking::tokens(); match mem::replace(&mut guard.blocker, BlockedReceiver(signal_token)) { NoneBlocked => {} _ => unreachable!(), } - drop(guard); // unlock - *success = wait_token.wait_max_until(deadline); // block + drop(guard); // unlock + *success = wait_token.wait_max_until(deadline); // block let mut new_guard = lock.lock().unwrap(); // relock if !*success { abort_selection(&mut new_guard); @@ -147,7 +146,10 @@ fn abort_selection(guard: &mut MutexGuard<'_, State>) -> bool { guard.blocker = BlockedSender(token); true } - BlockedReceiver(token) => { drop(token); false } + BlockedReceiver(token) => { + drop(token); + false + } } } @@ -168,12 +170,9 @@ impl Packet { blocker: NoneBlocked, cap: capacity, canceled: None, - queue: Queue { - head: ptr::null_mut(), - tail: ptr::null_mut(), - }, + queue: Queue { head: ptr::null_mut(), tail: ptr::null_mut() }, buf: Buffer { - buf: (0..capacity + if capacity == 0 {1} else {0}).map(|_| None).collect(), + buf: (0..capacity + if capacity == 0 { 1 } else { 0 }).map(|_| None).collect(), start: 0, size: 0, }, @@ -200,7 +199,9 @@ impl Packet { pub fn send(&self, t: T) -> Result<(), T> { let mut guard = self.acquire_send_slot(); - if guard.disconnected { return Err(t) } + if guard.disconnected { + return Err(t); + } guard.buf.enqueue(t); match mem::replace(&mut guard.blocker, NoneBlocked) { @@ -213,14 +214,17 @@ impl Packet { assert!(guard.canceled.is_none()); guard.canceled = Some(unsafe { mem::transmute(&mut canceled) }); let mut guard = wait(&self.lock, guard, BlockedSender); - if canceled {Err(guard.buf.dequeue())} else {Ok(())} + if canceled { Err(guard.buf.dequeue()) } else { Ok(()) } } // success, we buffered some data NoneBlocked => Ok(()), // success, someone's about to receive our buffered data. - BlockedReceiver(token) => { wakeup(token, guard); Ok(()) } + BlockedReceiver(token) => { + wakeup(token, guard); + Ok(()) + } BlockedSender(..) => panic!("lolwut"), } @@ -271,10 +275,8 @@ impl Packet { // while loop because we're the only receiver. if !guard.disconnected && guard.buf.size() == 0 { if let Some(deadline) = deadline { - guard = wait_timeout_receiver(&self.lock, - deadline, - guard, - &mut woke_up_after_waiting); + guard = + wait_timeout_receiver(&self.lock, deadline, guard, &mut woke_up_after_waiting); } else { guard = wait(&self.lock, guard, BlockedReceiver); woke_up_after_waiting = true; @@ -290,7 +292,9 @@ impl Packet { // Pick up the data, wake up our neighbors, and carry on assert!(guard.buf.size() > 0 || (deadline.is_some() && !woke_up_after_waiting)); - if guard.buf.size() == 0 { return Err(Empty); } + if guard.buf.size() == 0 { + return Err(Empty); + } let ret = guard.buf.dequeue(); self.wakeup_senders(woke_up_after_waiting, guard); @@ -301,8 +305,12 @@ impl Packet { let mut guard = self.lock.lock().unwrap(); // Easy cases first - if guard.disconnected && guard.buf.size() == 0 { return Err(Disconnected) } - if guard.buf.size() == 0 { return Err(Empty) } + if guard.disconnected && guard.buf.size() == 0 { + return Err(Disconnected); + } + if guard.buf.size() == 0 { + return Err(Empty); + } // Be sure to wake up neighbors let ret = Ok(guard.buf.dequeue()); @@ -357,12 +365,14 @@ impl Packet { // Only flag the channel as disconnected if we're the last channel match self.channels.fetch_sub(1, Ordering::SeqCst) { 1 => {} - _ => return + _ => return, } // Not much to do other than wake up a receiver if one's there let mut guard = self.lock.lock().unwrap(); - if guard.disconnected { return } + if guard.disconnected { + return; + } guard.disconnected = true; match mem::replace(&mut guard.blocker, NoneBlocked) { NoneBlocked => {} @@ -374,7 +384,9 @@ impl Packet { pub fn drop_port(&self) { let mut guard = self.lock.lock().unwrap(); - if guard.disconnected { return } + if guard.disconnected { + return; + } guard.disconnected = true; // If the capacity is 0, then the sender may want its data back after @@ -382,15 +394,9 @@ impl Packet { // the buffered data. As with many other portions of this code, this // needs to be careful to destroy the data *outside* of the lock to // prevent deadlock. - let _data = if guard.cap != 0 { - mem::take(&mut guard.buf.buf) - } else { - Vec::new() - }; - let mut queue = mem::replace(&mut guard.queue, Queue { - head: ptr::null_mut(), - tail: ptr::null_mut(), - }); + let _data = if guard.cap != 0 { mem::take(&mut guard.buf.buf) } else { Vec::new() }; + let mut queue = + mem::replace(&mut guard.queue, Queue { head: ptr::null_mut(), tail: ptr::null_mut() }); let waiter = match mem::replace(&mut guard.blocker, NoneBlocked) { NoneBlocked => None, @@ -402,7 +408,9 @@ impl Packet { }; mem::drop(guard); - while let Some(token) = queue.dequeue() { token.signal(); } + while let Some(token) = queue.dequeue() { + token.signal(); + } waiter.map(|t| t.signal()); } } @@ -416,7 +424,6 @@ impl Drop for Packet { } } - //////////////////////////////////////////////////////////////////////////////// // Buffer, a simple ring buffer backed by Vec //////////////////////////////////////////////////////////////////////////////// @@ -437,8 +444,12 @@ impl Buffer { result.take().unwrap() } - fn size(&self) -> usize { self.size } - fn capacity(&self) -> usize { self.buf.len() } + fn size(&self) -> usize { + self.size + } + fn capacity(&self) -> usize { + self.buf.len() + } } //////////////////////////////////////////////////////////////////////////////// @@ -466,7 +477,7 @@ impl Queue { fn dequeue(&mut self) -> Option { if self.head.is_null() { - return None + return None; } let node = self.head; self.head = unsafe { (*node).next }; -- cgit 1.4.1-3-g733a5