Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
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
2 changes: 1 addition & 1 deletion .github/workflows/ci.yml
Original file line number Diff line number Diff line change
Expand Up @@ -55,7 +55,7 @@ jobs:
matrix:
# When updating this, the reminder to update the minimum supported
# Rust version in Cargo.toml.
rust: ['1.43']
rust: ['1.48']
steps:
- uses: actions/checkout@v3
- name: Install Rust
Expand Down
4 changes: 2 additions & 2 deletions Cargo.toml
Original file line number Diff line number Diff line change
Expand Up @@ -6,7 +6,7 @@ name = "async-lock"
version = "2.6.0"
authors = ["Stjepan Glavina <[email protected]>"]
edition = "2018"
rust-version = "1.43"
rust-version = "1.48"
description = "Async synchronization primitives"
license = "Apache-2.0 OR MIT"
repository = "https://github.com/smol-rs/async-lock"
Expand All @@ -15,12 +15,12 @@ categories = ["asynchronous", "concurrency"]
exclude = ["/.*"]

[dependencies]
futures-lite = "1.11.0"
event-listener = "2.5.1"

[dev-dependencies]
async-channel = "1.5.0"
fastrand = "1.4.0"
futures-lite = "1.12.0"

[target.'cfg(any(target_arch = "wasm32", target_arch = "wasm64"))'.dev-dependencies]
wasm-bindgen-test = "0.3"
1 change: 0 additions & 1 deletion src/barrier.rs
Original file line number Diff line number Diff line change
@@ -1,5 +1,4 @@
use event_listener::{Event, EventListener};
use futures_lite::ready;

use std::fmt;
use std::future::Future;
Expand Down
14 changes: 14 additions & 0 deletions src/lib.rs
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,20 @@

#![warn(missing_docs, missing_debug_implementations, rust_2018_idioms)]

/// Simple macro to extract the value of `Poll` or return `Pending`.
///
/// TODO: Drop in favor of `core::task::ready`, once MSRV is bumped to 1.64.
macro_rules! ready {
($e:expr) => {{
use ::core::task::Poll;

match $e {
Poll::Ready(v) => v,
Poll::Pending => return Poll::Pending,
}
}};
}

mod barrier;
mod mutex;
mod once_cell;
Expand Down
1 change: 0 additions & 1 deletion src/mutex.rs
Original file line number Diff line number Diff line change
Expand Up @@ -18,7 +18,6 @@ use std::time::{Duration, Instant};
use std::usize;

use event_listener::{Event, EventListener};
use futures_lite::ready;

/// An async mutex.
///
Expand Down
52 changes: 25 additions & 27 deletions src/once_cell.rs
Original file line number Diff line number Diff line change
Expand Up @@ -9,7 +9,6 @@ use std::sync::atomic::{AtomicUsize, Ordering};
use std::task::{Context, Poll, RawWaker, RawWakerVTable, Waker};

use event_listener::{Event, EventListener};
use futures_lite::future;

/// The current state of the `OnceCell`.
#[derive(Copy, Clone, PartialEq, Eq)]
Expand Down Expand Up @@ -427,7 +426,9 @@ impl<T> OnceCell<T> {

// Slow path: initialize the value.
// The futures provided should never block, so we can use `now_or_never`.
now_or_never(self.initialize_or_wait(move || future::ready(closure()), &mut Blocking))?;
now_or_never(
self.initialize_or_wait(move || std::future::ready(closure()), &mut Blocking),
)?;
debug_assert!(self.is_initialized());

// SAFETY: We know that the value is initialized, so it is safe to
Expand Down Expand Up @@ -594,22 +595,13 @@ impl<T> OnceCell<T> {
// but we do not have the ability to initialize it.
//
// We need to wait the initialization to complete.
future::poll_fn(|cx| {
match event_listener.take() {
None => {
event_listener = Some(self.active_initializers.listen());
}
Some(evl) => {
if let Err(evl) = strategy.poll(evl, cx) {
event_listener = Some(evl);
return Poll::Pending;
}
}
match event_listener.take() {
None => {
event_listener = Some(self.active_initializers.listen());
}

Poll::Ready(())
})
.await;
Some(evl) => strategy.poll(evl).await,
}
}
State::Uninitialized => {
// Try to move the cell into the initializing state.
Expand Down Expand Up @@ -758,7 +750,7 @@ impl<T> Drop for OnceCell<T> {
}

/// Either return the result of a future now, or panic.
fn now_or_never<T>(f: impl Future<Output = T>) -> T {
fn now_or_never<T>(mut f: impl Future<Output = T>) -> T {
const NOOP_WAKER: RawWakerVTable = RawWakerVTable::new(clone, wake, wake_by_ref, drop);

unsafe fn wake(_: *const ()) {}
Expand All @@ -768,42 +760,48 @@ fn now_or_never<T>(f: impl Future<Output = T>) -> T {
}
unsafe fn drop(_: *const ()) {}

futures_lite::pin!(f);
// SAFETY: We don't move the future after we pin it here.
let future = unsafe { Pin::new_unchecked(&mut f) };
Copy link
Collaborator

@taiki-e taiki-e Apr 30, 2023

Choose a reason for hiding this comment

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

I strongly recommend that you use the same variable names here. Otherwise, it is very easy to access the owned value incorrectly.

Simple shadowing is not sufficient to prevent misuse, but in this case it is fine because it is on the outermost scope. (This is a common oversight, I sometimes see crates make the false claim that simple shadowing is "doing the exact same thing as pin_mut".)

Well, it would be better to duplicate the pin macro as you have done for the ready macro.


let waker = unsafe { Waker::from_raw(RawWaker::new(ptr::null(), &NOOP_WAKER)) };

// Poll the future exactly once.
let mut cx = Context::from_waker(&waker);

match f.poll(&mut cx) {
match future.poll(&mut cx) {
Poll::Ready(value) => value,
Poll::Pending => unreachable!("future not ready"),
}
}

/// The strategy for polling an `event_listener::EventListener`.
trait Strategy {
/// The future that can be polled to wait on the listener.
type Fut: Future<Output = ()>;

/// Poll the event listener.
fn poll(&mut self, evl: EventListener, ctx: &mut Context<'_>) -> Result<(), EventListener>;
fn poll(&mut self, evl: EventListener) -> Self::Fut;
}

/// The strategy for blocking the current thread on an `EventListener`.
struct Blocking;

impl Strategy for Blocking {
fn poll(&mut self, evl: EventListener, _: &mut Context<'_>) -> Result<(), EventListener> {
type Fut = std::future::Ready<()>;

fn poll(&mut self, evl: EventListener) -> Self::Fut {
evl.wait();
Ok(())
std::future::ready(())
}
}

/// The strategy for polling an `EventListener` in an async context.
struct NonBlocking;

impl Strategy for NonBlocking {
fn poll(&mut self, mut evl: EventListener, ctx: &mut Context<'_>) -> Result<(), EventListener> {
match Pin::new(&mut evl).poll(ctx) {
Poll::Pending => Err(evl),
Poll::Ready(()) => Ok(()),
}
type Fut = EventListener;

fn poll(&mut self, evl: EventListener) -> Self::Fut {
evl
}
}
1 change: 0 additions & 1 deletion src/rwlock.rs
Original file line number Diff line number Diff line change
Expand Up @@ -9,7 +9,6 @@ use std::sync::atomic::{AtomicUsize, Ordering};
use std::task::{Context, Poll};

use event_listener::{Event, EventListener};
use futures_lite::ready;

use crate::futures::Lock;
use crate::{Mutex, MutexGuard};
Expand Down
1 change: 0 additions & 1 deletion src/semaphore.rs
Original file line number Diff line number Diff line change
Expand Up @@ -6,7 +6,6 @@ use std::sync::Arc;
use std::task::{Context, Poll};

use event_listener::{Event, EventListener};
use futures_lite::ready;

/// A counter for limiting the number of concurrent operations.
#[derive(Debug)]
Expand Down