Merge branch 'main' into semaphore

.rustfmt.toml

@@ -1,4 +1,4 @@
 max_width = 80
 tab_spaces = 2
 edition = "2021"
-imports_granularity = "Item"
+imports_granularity = "Item"

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "deno_unsync"
-version = "0.3.3"
+version = "0.3.4"
 edition = "2021"
 authors = ["the Deno authors"]
 license = "MIT"

LICENSE

@@ -0,0 +1,21 @@
+MIT License
+
+Copyright (c) 2018-2024 the Deno authors
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+SOFTWARE.

rust-toolchain.toml

@@ -0,0 +1,3 @@
+[toolchain]
+channel = "1.78.0"
+components = [ "clippy", "rustfmt" ]

src/flag.rs

@@ -1,3 +1,5 @@
+// Copyright 2018-2024 the Deno authors. MIT license.
+
 use std::cell::Cell;
 
 /// A flag with interior mutability that can be raised or lowered.

src/joinset.rs

@@ -1,4 +1,4 @@
-// Copyright 2018-2023 the Deno authors. All rights reserved. MIT license.
+// Copyright 2018-2024 the Deno authors. MIT license.
 // Some code and comments under MIT license where adapted from Tokio code
 // Copyright (c) 2023 Tokio Contributors
 

src/lib.rs

@@ -2,6 +2,7 @@ mod flag;
 mod joinset;
 mod notify;
 mod semaphore;
+pub mod mpsc;
 mod split;
 mod task;
 mod task_queue;

src/mpsc/chunked_queue.rs

@@ -0,0 +1,82 @@
+// Copyright 2018-2024 the Deno authors. MIT license.
+
+use std::collections::LinkedList;
+use std::collections::VecDeque;
+
+const CHUNK_SIZE: usize = 1024;
+
+/// A queue that stores elements in chunks in a linked list
+/// to reduce allocations.
+pub(crate) struct ChunkedQueue<T> {
+  chunks: LinkedList<VecDeque<T>>,
+}
+
+impl<T> Default for ChunkedQueue<T> {
+  fn default() -> Self {
+    Self {
+      chunks: Default::default(),
+    }
+  }
+}
+
+impl<T> ChunkedQueue<T> {
+  pub fn len(&self) -> usize {
+    match self.chunks.len() {
+      0 => 0,
+      1 => self.chunks.front().unwrap().len(),
+      2 => {
+        self.chunks.front().unwrap().len() + self.chunks.back().unwrap().len()
+      }
+      _ => {
+        self.chunks.front().unwrap().len()
+          + CHUNK_SIZE * (self.chunks.len() - 2)
+          + self.chunks.back().unwrap().len()
+      }
+    }
+  }
+
+  pub fn push_back(&mut self, value: T) {
+    if let Some(tail) = self.chunks.back_mut() {
+      if tail.len() < CHUNK_SIZE {
+        tail.push_back(value);
+        return;
+      }
+    }
+    let mut new_buffer = VecDeque::with_capacity(CHUNK_SIZE);
+    new_buffer.push_back(value);
+    self.chunks.push_back(new_buffer);
+  }
+
+  pub fn pop_front(&mut self) -> Option<T> {
+    if let Some(head) = self.chunks.front_mut() {
+      let value = head.pop_front();
+      if value.is_some() && head.is_empty() && self.chunks.len() > 1 {
+        self.chunks.pop_front().unwrap();
+      }
+      value
+    } else {
+      None
+    }
+  }
+}
+
+#[cfg(test)]
+mod test {
+  use super::CHUNK_SIZE;
+
+  #[test]
+  fn ensure_len_correct() {
+    let mut queue = super::ChunkedQueue::default();
+    for _ in 0..2 {
+      for i in 0..CHUNK_SIZE * 20 {
+        queue.push_back(i);
+        assert_eq!(queue.len(), i + 1);
+      }
+      for i in (0..CHUNK_SIZE * 20).rev() {
+        queue.pop_front();
+        assert_eq!(queue.len(), i);
+      }
+      assert_eq!(queue.len(), 0);
+    }
+  }
+}

src/mpsc/mod.rs

@@ -0,0 +1,246 @@
+// Copyright 2018-2024 the Deno authors. MIT license.
+
+use std::cell::RefCell;
+use std::fmt::Formatter;
+use std::future::Future;
+use std::pin::Pin;
+use std::rc::Rc;
+use std::task::Context;
+use std::task::Poll;
+
+mod chunked_queue;
+
+use crate::UnsyncWaker;
+use chunked_queue::ChunkedQueue;
+
+pub struct SendError<T>(pub T);
+
+impl<T> std::fmt::Debug for SendError<T>
+where
+  T: std::fmt::Debug,
+{
+  fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+    f.debug_tuple("SendError").field(&self.0).finish()
+  }
+}
+
+pub struct Sender<T> {
+  shared: Rc<RefCell<Shared<T>>>,
+}
+
+impl<T> Sender<T> {
+  pub fn send(&self, value: T) -> Result<(), SendError<T>> {
+    let mut shared = self.shared.borrow_mut();
+    if shared.closed {
+      return Err(SendError(value));
+    }
+    shared.queue.push_back(value);
+    shared.waker.wake();
+    Ok(())
+  }
+}
+
+impl<T> Drop for Sender<T> {
+  fn drop(&mut self) {
+    let mut shared = self.shared.borrow_mut();
+    shared.closed = true;
+    shared.waker.wake();
+  }
+}
+
+pub struct Receiver<T> {
+  shared: Rc<RefCell<Shared<T>>>,
+}
+
+impl<T> Drop for Receiver<T> {
+  fn drop(&mut self) {
+    let mut shared = self.shared.borrow_mut();
+    shared.closed = true;
+  }
+}
+
+impl<T> Receiver<T> {
+  /// Receives a value from the channel, returning `None` if there
+  /// are no more items and the channel is closed.
+  pub async fn recv(&mut self) -> Option<T> {
+    // note: this is `&mut self` so that it can't be polled
+    // concurrently. DO NOT change this to `&self` because
+    // then futures will lose their wakers.
+    RecvFuture {
+      shared: &self.shared,
+    }
+    .await
+  }
+
+  /// Number of pending unread items.
+  pub fn len(&self) -> usize {
+    self.shared.borrow().queue.len()
+  }
+
+  /// If the receiver has no pending items.
+  pub fn is_empty(&self) -> bool {
+    self.len() == 0
+  }
+}
+
+struct RecvFuture<'a, T> {
+  shared: &'a RefCell<Shared<T>>,
+}
+
+impl<'a, T> Future for RecvFuture<'a, T> {
+  type Output = Option<T>;
+
+  fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+    let mut shared = self.shared.borrow_mut();
+    if let Some(value) = shared.queue.pop_front() {
+      Poll::Ready(Some(value))
+    } else if shared.closed {
+      Poll::Ready(None)
+    } else {
+      shared.waker.register(cx.waker());
+      Poll::Pending
+    }
+  }
+}
+
+struct Shared<T> {
+  queue: ChunkedQueue<T>,
+  waker: UnsyncWaker,
+  closed: bool,
+}
+
+/// A ![`Sync`] and ![`Sync`] equivalent to `tokio::sync::unbounded_channel`.
+pub fn unbounded_channel<T>() -> (Sender<T>, Receiver<T>) {
+  let shared = Rc::new(RefCell::new(Shared {
+    queue: ChunkedQueue::default(),
+    waker: UnsyncWaker::default(),
+    closed: false,
+  }));
+  (
+    Sender {
+      shared: shared.clone(),
+    },
+    Receiver { shared },
+  )
+}
+
+#[cfg(test)]
+mod test {
+  use std::time::Duration;
+
+  use tokio::join;
+
+  use super::*;
+
+  #[tokio::test(flavor = "current_thread")]
+  async fn sends_receives_exits() {
+    let (sender, mut receiver) = unbounded_channel::<usize>();
+    sender.send(1).unwrap();
+    assert_eq!(receiver.recv().await, Some(1));
+    sender.send(2).unwrap();
+    assert_eq!(receiver.recv().await, Some(2));
+    drop(sender);
+    assert_eq!(receiver.recv().await, None);
+  }
+
+  #[tokio::test(flavor = "current_thread")]
+  async fn sends_multiple_then_drop() {
+    let (sender, mut receiver) = unbounded_channel::<usize>();
+    sender.send(1).unwrap();
+    sender.send(2).unwrap();
+    drop(sender);
+    assert_eq!(receiver.len(), 2);
+    assert!(!receiver.is_empty());
+    assert_eq!(receiver.recv().await, Some(1));
+    assert_eq!(receiver.recv().await, Some(2));
+    assert_eq!(receiver.recv().await, None);
+    assert_eq!(receiver.len(), 0);
+    assert!(receiver.is_empty());
+  }
+
+  #[tokio::test(flavor = "current_thread")]
+  async fn receiver_dropped_sending() {
+    let (sender, receiver) = unbounded_channel::<usize>();
+    drop(receiver);
+    let err = sender.send(1).unwrap_err();
+    assert_eq!(err.0, 1);
+  }
+
+  #[tokio::test(flavor = "current_thread")]
+  async fn receiver_recv_then_drop_sender() {
+    let (sender, mut receiver) = unbounded_channel::<usize>();
+    let future = crate::task::spawn(async move {
+      let value = receiver.recv().await;
+      value.is_none()
+    });
+    let future2 = crate::task::spawn(async move {
+      drop(sender);
+      true
+    });
+    let (first, second) = join!(future, future2);
+    assert!(first.unwrap());
+    assert!(second.unwrap());
+  }
+
+  #[tokio::test(flavor = "current_thread")]
+  async fn multiple_senders_divided_work() {
+    for receiver_sleep in [None, Some(1)] {
+      for sender_sleep in [None, Some(1)] {
+        for sender_count in [1000, 100, 10, 2, 1] {
+          let (sender, mut receiver) = unbounded_channel::<usize>();
+          let future = crate::task::spawn(async move {
+            let mut values = Vec::with_capacity(1000);
+            for _ in 0..1000 {
+              if let Some(micros) = receiver_sleep {
+                if cfg!(windows) {
+                  // windows min sleep resolution is too slow
+                  tokio::task::yield_now().await;
+                } else {
+                  tokio::time::sleep(Duration::from_micros(micros)).await;
+                }
+              }
+              let value = receiver.recv().await;
+              values.push(value.unwrap());
+            }
+            // both senders should be dropped at this point
+            let value = receiver.recv().await;
+            assert!(value.is_none());
+
+            values.sort();
+            // ensure we received these values
+            for i in 0..1000 {
+              assert_eq!(values[i], i);
+            }
+          });
+
+          let mut futures = Vec::with_capacity(1 + sender_count);
+          futures.push(future);
+          let sender = Rc::new(sender);
+          for sender_index in 0..sender_count {
+            let sender = sender.clone();
+            let batch_count = 1000 / sender_count;
+            futures.push(crate::task::spawn(async move {
+              for i in 0..batch_count {
+                if let Some(micros) = sender_sleep {
+                  if cfg!(windows) {
+                    // windows min sleep resolution is too slow
+                    tokio::task::yield_now().await;
+                  } else {
+                    tokio::time::sleep(Duration::from_micros(micros)).await;
+                  }
+                }
+                sender.send(batch_count * sender_index + i).unwrap();
+              }
+            }));
+          }
+          drop(sender);
+
+          // wait all futures
+          for future in futures {
+            future.await.unwrap();
+          }
+        }
+      }
+    }
+  }
+}

src/split.rs

@@ -1,3 +1,5 @@
+// Copyright 2018-2024 the Deno authors. MIT license.
+
 use crate::UnsendMarker;
 use std::cell::Cell;
 use std::cell::UnsafeCell;