fix rustfmt issues

tokio-rs · Sep 15, 2023 · 537b964 · 537b964
1 parent 7baf04a
commit 537b964
Show file tree

Hide file tree

Showing 3 changed files with 23 additions and 26 deletions.
diff --git a/benches/sync_mpsc_recv_many.rs b/benches/sync_mpsc_recv_many.rs
@@ -10,7 +10,7 @@ fn rt() -> tokio::runtime::Runtime {
 
 // Simulate a use case of an actor that must update
 // a resource, but resource only needs last value
-fn publish_last_value(last_value:usize) -> usize {
+fn publish_last_value(last_value: usize) -> usize {
     std::thread::sleep(std::time::Duration::from_nanos(1));
     last_value
 }
@@ -64,15 +64,14 @@ fn contention_bounded_updater_recv_many(b: &mut Bencher) {
                 let count = rx.recv_many(&mut buffer).await;
                 total += count;
                 if count > 0 {
-                    last_value = buffer[buffer.len()-1]
+                    last_value = buffer[buffer.len() - 1]
                 }
             }
             last_value
         })
     });
 }
 
-
 fn contention_bounded_updater_publish_recv(b: &mut Bencher) {
     let rt = rt();
 
@@ -89,7 +88,7 @@ fn contention_bounded_updater_publish_recv(b: &mut Bencher) {
                 });
             }
 
-            for _ in 0..1_000  {
+            for _ in 0..1_000 {
                 let Some(v) = rx.recv().await else {continue};
                 let _ = publish_last_value(v);
             }
@@ -119,7 +118,7 @@ fn contention_bounded_updater_publish_recv_many(b: &mut Bencher) {
                 let count = rx.recv_many(&mut buffer).await;
                 total += count;
                 if count > 0 {
-                    publish_last_value(buffer[buffer.len()-1]);
+                    publish_last_value(buffer[buffer.len() - 1]);
                 }
             }
         })
@@ -175,7 +174,7 @@ fn contention_bounded_full_updater_recv_many(b: &mut Bencher) {
                 let count = rx.recv_many(&mut buffer).await;
                 total += count;
                 if count > 0 {
-                    last_value = buffer[buffer.len()-1]
+                    last_value = buffer[buffer.len() - 1]
                 }
             }
             last_value
@@ -232,7 +231,7 @@ fn contention_unbounded_updater_recv_many(b: &mut Bencher) {
                 let count = rx.recv_many(&mut buffer).await;
                 total += count;
                 if count > 0 {
-                    last_value = buffer[buffer.len()-1]
+                    last_value = buffer[buffer.len() - 1]
                 }
             }
             last_value
@@ -279,15 +278,14 @@ fn uncontented_bounded_updater_recv_many(b: &mut Bencher) {
                 let count = rx.recv_many(&mut buffer).await;
                 total += count;
                 if count > 0 {
-                    last_value = buffer[buffer.len()-1]
+                    last_value = buffer[buffer.len() - 1]
                 }
             }
             last_value
         })
     });
 }
 
-
 fn uncontented_unbounded_updater_recv(b: &mut Bencher) {
     let rt = rt();
 
@@ -327,7 +325,7 @@ fn uncontented_unbounded_updater_recv_many(b: &mut Bencher) {
                 let count = rx.recv_many(&mut buffer).await;
                 total += count;
                 if count > 0 {
-                    last_value = buffer[buffer.len()-1]
+                    last_value = buffer[buffer.len() - 1]
                 }
             }
             last_value
@@ -347,7 +345,6 @@ bencher::benchmark_group!(
     contention_bounded_updater_publish_recv_many
 );
 
-
 bencher::benchmark_group!(
     contention_bounded_full_updater,
     contention_bounded_full_updater_recv,

diff --git a/tokio/src/sync/mpsc/chan.rs b/tokio/src/sync/mpsc/chan.rs
@@ -41,7 +41,7 @@ pub(crate) trait Semaphore {
 
     fn add_permit(&self);
 
-    fn add_permits(&self, n:usize);
+    fn add_permits(&self, n: usize);
 
     fn close(&self);
 
@@ -321,14 +321,17 @@ impl<T, S: Semaphore> Rx<T, S> {
                             }
                             buffer.push(value);
                             let mut next = rx_fields.list.peek();
-                            while (buffer.len() < capacity && match next {
-                                Some(Value(value)) => {
-                                    rx_fields.list.pop(&self.inner.tx);
-                                    buffer.push(value);
-                                    next = rx_fields.list.peek();
-                                    true }
-                                _ => false
-                            }) {}
+                            while (buffer.len() < capacity
+                                && match next {
+                                    Some(Value(value)) => {
+                                        rx_fields.list.pop(&self.inner.tx);
+                                        buffer.push(value);
+                                        next = rx_fields.list.peek();
+                                        true
+                                    }
+                                    _ => false,
+                                })
+                            {}
                             self.inner.semaphore.add_permits(buffer.len());
                             coop.made_progress();
                             return Ready(buffer.len());
@@ -367,7 +370,6 @@ impl<T, S: Semaphore> Rx<T, S> {
         })
     }
 
-
     /// Try to receive the next value.
     pub(crate) fn try_recv(&mut self) -> Result<T, TryRecvError> {
         use super::list::TryPopResult;
@@ -464,7 +466,7 @@ impl Semaphore for bounded::Semaphore {
         self.semaphore.release(1)
     }
 
-    fn add_permits(&self, n:usize) {
+    fn add_permits(&self, n: usize) {
         self.semaphore.release(n)
     }
 
@@ -497,7 +499,7 @@ impl Semaphore for unbounded::Semaphore {
         }
     }
 
-    fn add_permits(&self, n:usize) {
+    fn add_permits(&self, n: usize) {
         let prev = self.0.fetch_sub(n << 1, Release);
 
         if prev >> 1 == 0 {
@@ -521,5 +523,4 @@ impl Semaphore for unbounded::Semaphore {
     fn num_acquired(&self) -> usize {
         self.0.load(Acquire) >> 1
     }
-
 }
diff --git a/tokio/tests/sync_mpsc.rs b/tokio/tests/sync_mpsc.rs
@@ -151,7 +151,6 @@ async fn async_send_recv_many_with_buffer() {
     assert_eq!(None, rx.recv().await);
 }
 
-
 #[tokio::test]
 #[cfg(feature = "full")]
 async fn start_send_past_cap() {
@@ -220,7 +219,7 @@ async fn send_recv_many_unbounded() {
 
     let mut buffer = vec![0; 0];
     assert_eq!(rx.recv_many(&mut buffer).await, 4);
-    assert_eq!(vec![7,13,100,1002], buffer);
+    assert_eq!(vec![7, 13, 100, 1002], buffer);
     assert!(buffer.capacity() >= 4);
 
     drop(tx);