Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

im: Relax Send + Sync bounds #63

Merged
merged 2 commits into from
Nov 11, 2024
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
14 changes: 7 additions & 7 deletions eyeball-im-util/src/vector/traits.rs
Original file line number Diff line number Diff line change
Expand Up @@ -22,18 +22,18 @@ pub trait VectorDiffContainer:
{
/// The element type of the [`Vector`][imbl::Vector] that diffs are being
/// handled for.
type Element: Clone + Send + Sync + 'static;
type Element: Clone + 'static;

#[doc(hidden)]
type Family: VectorDiffContainerFamily<Member<Self::Element> = Self>;
}

impl<T: Clone + Send + Sync + 'static> VectorDiffContainer for VectorDiff<T> {
impl<T: Clone + 'static> VectorDiffContainer for VectorDiff<T> {
type Element = T;
type Family = VectorDiffFamily;
}

impl<T: Clone + Send + Sync + 'static> VectorDiffContainer for Vec<VectorDiff<T>> {
impl<T: Clone + 'static> VectorDiffContainer for Vec<VectorDiff<T>> {
type Element = T;
type Family = VecVectorDiffFamily;
}
Expand Down Expand Up @@ -69,15 +69,15 @@ pub trait VectorObserver<T>: Sized {
fn into_parts(self) -> (Vector<T>, Self::Stream);
}

impl<T: Clone + Send + Sync + 'static> VectorObserver<T> for VectorSubscriber<T> {
impl<T: Clone + 'static> VectorObserver<T> for VectorSubscriber<T> {
type Stream = VectorSubscriberStream<T>;

fn into_parts(self) -> (Vector<T>, Self::Stream) {
self.into_values_and_stream()
}
}

impl<T: Clone + Send + Sync + 'static> VectorObserver<T> for BatchedVectorSubscriber<T> {
impl<T: Clone + 'static> VectorObserver<T> for BatchedVectorSubscriber<T> {
type Stream = VectorSubscriberBatchedStream<T>;

fn into_parts(self) -> (Vector<T>, Self::Stream) {
Expand All @@ -102,7 +102,7 @@ where
/// See that trait for which types implement this.
pub trait VectorObserverExt<T>: VectorObserver<T>
where
T: Clone + Send + Sync + 'static,
T: Clone + 'static,
<Self::Stream as Stream>::Item: VectorDiffContainer<Element = T>,
{
/// Filter the vector's values with the given function.
Expand Down Expand Up @@ -197,7 +197,7 @@ where

impl<T, O> VectorObserverExt<T> for O
where
T: Clone + Send + Sync + 'static,
T: Clone + 'static,
O: VectorObserver<T>,
<Self::Stream as Stream>::Item: VectorDiffContainer<Element = T>,
{
Expand Down
1 change: 0 additions & 1 deletion eyeball-im/Cargo.toml
Original file line number Diff line number Diff line change
Expand Up @@ -17,7 +17,6 @@ futures-core.workspace = true
imbl = "3.0.0"
serde = { version = "1.0", optional = true }
tokio.workspace = true
tokio-util.workspace = true
tracing = { workspace = true, optional = true }

[dev-dependencies]
Expand Down
1 change: 1 addition & 0 deletions eyeball-im/src/lib.rs
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,7 @@
//!
//! - `tracing`: Emit [tracing] events when updates are sent out

mod reusable_box;
mod vector;

pub use vector::{
Expand Down
161 changes: 161 additions & 0 deletions eyeball-im/src/reusable_box.rs
Original file line number Diff line number Diff line change
@@ -0,0 +1,161 @@
// Copy-pasted from https://docs.rs/tokio-util/latest/src/tokio_util/sync/reusable_box.rs.html
Copy link
Collaborator

Choose a reason for hiding this comment

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

Should we add the license, or create an ATTRIBUTION file?

Copy link
Owner Author

Choose a reason for hiding this comment

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

Is that some sort of standard? I didn't find anything immediately.

I felt when writing this that the comment at the top should be sufficient attribution for copying a small file out of a permissively-licensed project. I suppose it wouldn't hurt to note the original license. I don't think adding the license anywhere is necessary though, since MIT allows redistributing stuff under a different license IIRC (I am not a lawyer, of course).

Also, I plan to get rid of this module again in the future anyways, once tokio-rs/tokio#6908 is merged and another release is made it should be possible to write a self-referential future by hand that can be stored in the VectorSubscriberStream without going through dyn Future. It's a bit tricky though, and I haven't had any focus time for that sort of stuff lately. Let me know if you're interested in helping out 😉

// Removed all `+ Send`s.

use std::{
alloc::Layout,
fmt,
future::{self, Future},
mem::{self, ManuallyDrop},
pin::Pin,
ptr,
task::{Context, Poll},
};

/// A reusable `Pin<Box<dyn Future<Output = T> + 'a>>`.
///
/// This type lets you replace the future stored in the box without
/// reallocating when the size and alignment permits this.
pub(crate) struct ReusableBoxFuture<'a, T> {
boxed: Pin<Box<dyn Future<Output = T> + 'a>>,
}

impl<'a, T> ReusableBoxFuture<'a, T> {
/// Create a new `ReusableBoxFuture<T>` containing the provided future.
pub(crate) fn new<F>(future: F) -> Self
where
F: Future<Output = T> + 'a,
{
Self { boxed: Box::pin(future) }
}

/// Replace the future currently stored in this box.
///
/// This reallocates if and only if the layout of the provided future is
/// different from the layout of the currently stored future.
pub(crate) fn set<F>(&mut self, future: F)
where
F: Future<Output = T> + 'a,
{
if let Err(future) = self.try_set(future) {
*self = Self::new(future);
}
}

/// Replace the future currently stored in this box.
///
/// This function never reallocates, but returns an error if the provided
/// future has a different size or alignment from the currently stored
/// future.
pub(crate) fn try_set<F>(&mut self, future: F) -> Result<(), F>
where
F: Future<Output = T> + 'a,
{
// If we try to inline the contents of this function, the type checker complains
// because the bound `T: 'a` is not satisfied in the call to
// `pending()`. But by putting it in an inner function that doesn't have
// `T` as a generic parameter, we implicitly get the bound `F::Output:
// 'a` transitively through `F: 'a`, allowing us to call `pending()`.
#[inline(always)]
fn real_try_set<'a, F>(
this: &mut ReusableBoxFuture<'a, F::Output>,
future: F,
) -> Result<(), F>
where
F: Future + 'a,
{
// future::Pending<T> is a ZST so this never allocates.
let boxed = mem::replace(&mut this.boxed, Box::pin(future::pending()));
reuse_pin_box(boxed, future, |boxed| this.boxed = Pin::from(boxed))
}

real_try_set(self, future)
}

/// Get a pinned reference to the underlying future.
pub(crate) fn get_pin(&mut self) -> Pin<&mut (dyn Future<Output = T>)> {
self.boxed.as_mut()
}

/// Poll the future stored inside this box.
pub(crate) fn poll(&mut self, cx: &mut Context<'_>) -> Poll<T> {
self.get_pin().poll(cx)
}
}

impl<T> Future for ReusableBoxFuture<'_, T> {
type Output = T;

/// Poll the future stored inside this box.
fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<T> {
Pin::into_inner(self).get_pin().poll(cx)
}
}

// The only method called on self.boxed is poll, which takes &mut self, so this
// struct being Sync does not permit any invalid access to the Future, even if
// the future is not Sync.
unsafe impl<T> Sync for ReusableBoxFuture<'_, T> {}

impl<T> fmt::Debug for ReusableBoxFuture<'_, T> {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.debug_struct("ReusableBoxFuture").finish()
}
}

fn reuse_pin_box<T: ?Sized, U, O, F>(boxed: Pin<Box<T>>, new_value: U, callback: F) -> Result<O, U>
where
F: FnOnce(Box<U>) -> O,
{
let layout = Layout::for_value::<T>(&*boxed);
if layout != Layout::new::<U>() {
return Err(new_value);
}

// SAFETY: We don't ever construct a non-pinned reference to the old `T` from
// now on, and we always drop the `T`.
let raw: *mut T = Box::into_raw(unsafe { Pin::into_inner_unchecked(boxed) });

// When dropping the old value panics, we still want to call `callback` — so
// move the rest of the code into a guard type.
let guard = CallOnDrop::new(|| {
let raw: *mut U = raw.cast::<U>();
unsafe { raw.write(new_value) };

// SAFETY:
// - `T` and `U` have the same layout.
// - `raw` comes from a `Box` that uses the same allocator as this one.
// - `raw` points to a valid instance of `U` (we just wrote it in).
let boxed = unsafe { Box::from_raw(raw) };

callback(boxed)
});

// Drop the old value.
unsafe { ptr::drop_in_place(raw) };

// Run the rest of the code.
Ok(guard.call())
}

struct CallOnDrop<O, F: FnOnce() -> O> {
f: ManuallyDrop<F>,
}

impl<O, F: FnOnce() -> O> CallOnDrop<O, F> {
fn new(f: F) -> Self {
let f = ManuallyDrop::new(f);
Self { f }
}
fn call(self) -> O {
let mut this = ManuallyDrop::new(self);
let f = unsafe { ManuallyDrop::take(&mut this.f) };
f()
}
}

impl<O, F: FnOnce() -> O> Drop for CallOnDrop<O, F> {
fn drop(&mut self) {
let f = unsafe { ManuallyDrop::take(&mut self.f) };
f();
}
}
6 changes: 3 additions & 3 deletions eyeball-im/src/vector.rs
Original file line number Diff line number Diff line change
Expand Up @@ -22,7 +22,7 @@ pub struct ObservableVector<T> {
sender: Sender<BroadcastMessage<T>>,
}

impl<T: Clone + Send + Sync + 'static> ObservableVector<T> {
impl<T: Clone + 'static> ObservableVector<T> {
/// Create a new `ObservableVector`.
///
/// As of the time of writing, this is equivalent to
Expand Down Expand Up @@ -290,7 +290,7 @@ impl<T: Clone + Send + Sync + 'static> ObservableVector<T> {
}
}

impl<T: Clone + Send + Sync + 'static> Default for ObservableVector<T> {
impl<T: Clone + 'static> Default for ObservableVector<T> {
fn default() -> Self {
Self::new()
}
Expand All @@ -315,7 +315,7 @@ impl<T> ops::Deref for ObservableVector<T> {
}
}

impl<T: Clone + Send + Sync + 'static> From<Vector<T>> for ObservableVector<T> {
impl<T: Clone + 'static> From<Vector<T>> for ObservableVector<T> {
fn from(values: Vector<T>) -> Self {
let mut this = Self::new();
this.append(values);
Expand Down
4 changes: 2 additions & 2 deletions eyeball-im/src/vector/entry.rs
Original file line number Diff line number Diff line change
Expand Up @@ -10,7 +10,7 @@ pub struct ObservableVectorEntry<'a, T> {

impl<'a, T> ObservableVectorEntry<'a, T>
where
T: Clone + Send + Sync + 'static,
T: Clone + 'static,
{
pub(super) fn new(inner: &'a mut ObservableVector<T>, index: usize) -> Self {
Self { inner, index: EntryIndex::Owned(index) }
Expand Down Expand Up @@ -115,7 +115,7 @@ pub struct ObservableVectorEntries<'a, T> {

impl<'a, T> ObservableVectorEntries<'a, T>
where
T: Clone + Send + Sync + 'static,
T: Clone + 'static,
{
pub(super) fn new(inner: &'a mut ObservableVector<T>) -> Self {
Self { inner, index: 0 }
Expand Down
Loading