im: Relax Send + Sync bounds

eyeball-im-util/src/vector/traits.rs

@@ -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;
 }
@@ -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) {
@@ -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.
@@ -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>,
 {

eyeball-im/Cargo.toml

@@ -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]

eyeball-im/src/lib.rs

@@ -4,6 +4,7 @@
 //!
 //! - `tracing`: Emit [tracing] events when updates are sent out
 
+mod reusable_box;
 mod vector;
 
 pub use vector::{

eyeball-im/src/reusable_box.rs

@@ -0,0 +1,161 @@
+// Copy-pasted from https://docs.rs/tokio-util/latest/src/tokio_util/sync/reusable_box.rs.html
+// 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();
+    }
+}

eyeball-im/src/vector.rs

@@ -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
@@ -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()
     }
@@ -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);

eyeball-im/src/vector/entry.rs

@@ -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) }
@@ -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 }