diff --git a/Cargo.lock b/Cargo.lock index 1d6e8ce04..e6f1c5a57 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2754,7 +2754,6 @@ version = "0.5.0-alpha.0" dependencies = [ "dioxus", "dioxus-core", - "flume", "futures-channel", "futures-util", "generational-box", diff --git a/examples/memo_chain.rs b/examples/memo_chain.rs index bbcca9954..745c5d01e 100644 --- a/examples/memo_chain.rs +++ b/examples/memo_chain.rs @@ -26,11 +26,7 @@ fn app() -> Element { } #[component] -fn Child( - state: ReadOnlySignal, - items: ReadOnlySignal>, - depth: ReadOnlySignal, -) -> Element { +fn Child(state: Memo, items: Memo>, depth: ReadOnlySignal) -> Element { if depth() == 0 { return None; } diff --git a/packages/hooks/src/use_effect.rs b/packages/hooks/src/use_effect.rs index 33e65309b..069b58e3d 100644 --- a/packages/hooks/src/use_effect.rs +++ b/packages/hooks/src/use_effect.rs @@ -1,5 +1,6 @@ use dioxus_core::prelude::*; use dioxus_signals::ReactiveContext; +use futures_util::StreamExt; /// `use_effect` will subscribe to any changes in the signal values it captures /// effects will always run after first mount and then whenever the signal values change @@ -26,13 +27,13 @@ pub fn use_effect(mut callback: impl FnMut() + 'static) { use_hook(|| { spawn(async move { - let rc = ReactiveContext::new_with_origin(location); + let (rc, mut changed) = ReactiveContext::new_with_origin(location); loop { // Run the effect rc.run_in(&mut callback); // Wait for context to change - rc.changed().await; + let _ = changed.next().await; // Wait for the dom the be finished with sync work wait_for_next_render().await; diff --git a/packages/hooks/src/use_memo.rs b/packages/hooks/src/use_memo.rs index adbea360f..36a65f83a 100644 --- a/packages/hooks/src/use_memo.rs +++ b/packages/hooks/src/use_memo.rs @@ -1,8 +1,10 @@ use crate::dependency::Dependency; -use crate::use_signal; +use crate::{use_callback, use_signal}; use dioxus_core::prelude::*; +use dioxus_signals::Memo; use dioxus_signals::{ReactiveContext, ReadOnlySignal, Readable, Signal, SignalData}; use dioxus_signals::{Storage, Writable}; +use futures_util::StreamExt; /// Creates a new unsync Selector. The selector will be run immediately and whenever any signal it reads changes. /// @@ -22,51 +24,9 @@ use dioxus_signals::{Storage, Writable}; /// } /// ``` #[track_caller] -pub fn use_memo(f: impl FnMut() -> R + 'static) -> ReadOnlySignal { - use_maybe_sync_memo(f) -} - -/// Creates a new Selector that may be sync. The selector will be run immediately and whenever any signal it reads changes. -/// -/// Selectors can be used to efficiently compute derived data from signals. -/// -/// ```rust -/// use dioxus::prelude::*; -/// use dioxus_signals::*; -/// -/// fn App() -> Element { -/// let mut count = use_signal(|| 0); -/// let double = use_memo(move || count * 2); -/// count += 1; -/// assert_eq!(double(), count * 2); -/// -/// rsx! { "{double}" } -/// } -/// ``` -#[track_caller] -pub fn use_maybe_sync_memo>>( - mut f: impl FnMut() -> R + 'static, -) -> ReadOnlySignal { - use_hook(|| { - // Create a new reactive context for the memo - let rc = ReactiveContext::new(); - - // Create a new signal in that context, wiring up its dependencies and subscribers - let mut state: Signal = rc.run_in(|| Signal::new_maybe_sync(f())); - - spawn(async move { - loop { - rc.changed().await; - let new = rc.run_in(&mut f); - if new != *state.peek() { - *state.write() = new; - } - } - }); - - // And just return the readonly variant of that signal - ReadOnlySignal::new_maybe_sync(state) - }) +pub fn use_memo(f: impl FnMut() -> R + 'static) -> Memo { + let mut callback = use_callback(f); + use_hook(|| Signal::memo(move || callback.call())) } /// Creates a new unsync Selector with some local dependencies. The selector will be run immediately and whenever any signal it reads or any dependencies it tracks changes @@ -127,7 +87,7 @@ where let selector = use_hook(|| { // Get the current reactive context - let rc = ReactiveContext::new(); + let (rc, mut changed) = ReactiveContext::new(); // Create a new signal in that context, wiring up its dependencies and subscribers let mut state: Signal = @@ -135,7 +95,8 @@ where spawn(async move { loop { - rc.changed().await; + // Wait for context to change + let _ = changed.next().await; let new = rc.run_in(|| f(dependencies_signal.read().clone())); if new != *state.peek() { diff --git a/packages/hooks/src/use_resource.rs b/packages/hooks/src/use_resource.rs index 389697d74..4fa39c222 100644 --- a/packages/hooks/src/use_resource.rs +++ b/packages/hooks/src/use_resource.rs @@ -6,8 +6,8 @@ use dioxus_core::{ Task, }; use dioxus_signals::*; -use futures_util::{future, pin_mut, FutureExt}; -use std::future::Future; +use futures_util::{future, pin_mut, FutureExt, StreamExt}; +use std::{cell::Cell, future::Future, rc::Rc}; /// A memo that resolve to a value asynchronously. /// Unlike `use_future`, `use_resource` runs on the **server** @@ -44,7 +44,10 @@ where { let mut value = use_signal(|| None); let mut state = use_signal(|| UseResourceState::Pending); - let rc = use_hook(ReactiveContext::new); + let (rc, changed) = use_hook(|| { + let (rc, changed) = ReactiveContext::new(); + (rc, Rc::new(Cell::new(Some(changed)))) + }); let mut cb = use_callback(move || { // Create the user's task @@ -70,10 +73,11 @@ where let mut task = use_hook(|| Signal::new(cb.call())); use_hook(|| { + let mut changed = changed.take().unwrap(); spawn(async move { loop { // Wait for the dependencies to change - rc.changed().await; + let _ = changed.next().await; // Stop the old task task.write().cancel(); diff --git a/packages/signals/Cargo.toml b/packages/signals/Cargo.toml index b8b29c2c5..a4a94393f 100644 --- a/packages/signals/Cargo.toml +++ b/packages/signals/Cargo.toml @@ -22,7 +22,6 @@ once_cell = "1.18.0" rustc-hash = { workspace = true } futures-channel = { workspace = true } futures-util = { workspace = true } -flume = { version = "0.11.0", default-features = false, features = ["async"] } [dev-dependencies] dioxus = { workspace = true } diff --git a/packages/signals/src/copy_value.rs b/packages/signals/src/copy_value.rs index ca318bf2c..fb984eb6f 100644 --- a/packages/signals/src/copy_value.rs +++ b/packages/signals/src/copy_value.rs @@ -237,7 +237,7 @@ impl> Writable for CopyValue { S::try_map_mut(mut_, f) } - fn try_write(&self) -> Result, generational_box::BorrowMutError> { + fn try_write(&mut self) -> Result, generational_box::BorrowMutError> { self.value.try_write() } diff --git a/packages/signals/src/global/memo.rs b/packages/signals/src/global/memo.rs index 4a94d29f6..61d1c10f4 100644 --- a/packages/signals/src/global/memo.rs +++ b/packages/signals/src/global/memo.rs @@ -1,9 +1,9 @@ -use crate::{read::Readable, ReadableRef}; +use crate::{read::Readable, Memo, ReadableRef}; use dioxus_core::prelude::{IntoAttributeValue, ScopeId}; use generational_box::UnsyncStorage; use std::{mem::MaybeUninit, ops::Deref}; -use crate::{ReadOnlySignal, Signal}; +use crate::Signal; use super::get_global_context; @@ -22,14 +22,14 @@ impl GlobalMemo { } /// Get the signal that backs this global. - pub fn signal(&self) -> ReadOnlySignal { + pub fn memo(&self) -> Memo { let key = self as *const _ as *const (); let context = get_global_context(); let read = context.signal.borrow(); match read.get(&key) { - Some(signal) => *signal.downcast_ref::>().unwrap(), + Some(signal) => *signal.downcast_ref::>().unwrap(), None => { drop(read); // Constructors are always run in the root scope @@ -47,7 +47,7 @@ impl GlobalMemo { /// Get the generational id of the signal. pub fn id(&self) -> generational_box::GenerationalBoxId { - self.signal().id() + self.memo().id() } } @@ -57,12 +57,12 @@ impl Readable for GlobalMemo { #[track_caller] fn try_read(&self) -> Result, generational_box::BorrowError> { - self.signal().try_read() + self.memo().try_read() } #[track_caller] fn peek(&self) -> ReadableRef { - self.signal().peek() + self.memo().peek() } } @@ -71,7 +71,7 @@ where T: Clone + IntoAttributeValue, { fn into_value(self) -> dioxus_core::AttributeValue { - self.signal().into_value() + self.memo().into_value() } } @@ -81,7 +81,7 @@ impl PartialEq for GlobalMemo { } } -/// Allow calling a signal with signal() syntax +/// Allow calling a signal with memo() syntax /// /// Currently only limited to copy types, though could probably specialize for string/arc/rc impl Deref for GlobalMemo { diff --git a/packages/signals/src/global/signal.rs b/packages/signals/src/global/signal.rs index dda337232..3dd57f484 100644 --- a/packages/signals/src/global/signal.rs +++ b/packages/signals/src/global/signal.rs @@ -103,7 +103,7 @@ impl Writable for GlobalSignal { } #[track_caller] - fn try_write(&self) -> Result, generational_box::BorrowMutError> { + fn try_write(&mut self) -> Result, generational_box::BorrowMutError> { self.signal().try_write() } } diff --git a/packages/signals/src/impls.rs b/packages/signals/src/impls.rs index 3d120b54e..806a1fe1a 100644 --- a/packages/signals/src/impls.rs +++ b/packages/signals/src/impls.rs @@ -1,4 +1,5 @@ use crate::copy_value::CopyValue; +use crate::memo::Memo; use crate::read::Readable; use crate::signal::Signal; use crate::write::Writable; @@ -159,6 +160,16 @@ impl>> Clone for ReadOnlySignal { impl>> Copy for ReadOnlySignal {} +read_impls!(Memo: PartialEq); + +impl Clone for Memo { + fn clone(&self) -> Self { + *self + } +} + +impl Copy for Memo {} + read_impls!(GlobalSignal); default_impl!(GlobalSignal); diff --git a/packages/signals/src/lib.rs b/packages/signals/src/lib.rs index 46953ec5a..1492fe205 100644 --- a/packages/signals/src/lib.rs +++ b/packages/signals/src/lib.rs @@ -19,6 +19,9 @@ pub use map::*; // mod comparer; // pub use comparer::*; +mod memo; +pub use memo::*; + mod global; pub use global::*; diff --git a/packages/signals/src/memo.rs b/packages/signals/src/memo.rs new file mode 100644 index 000000000..e8ad0ec44 --- /dev/null +++ b/packages/signals/src/memo.rs @@ -0,0 +1,173 @@ +use crate::write::Writable; +use crate::{read::Readable, ReactiveContext, ReadableRef, Signal}; +use crate::{CopyValue, ReadOnlySignal}; +use std::{ + cell::RefCell, + ops::Deref, + panic::Location, + sync::{atomic::AtomicBool, Arc}, +}; + +use dioxus_core::prelude::*; +use futures_util::StreamExt; +use generational_box::UnsyncStorage; +struct UpdateInformation { + dirty: Arc, + callback: RefCell T>>, +} + +/// A value that is memoized. This is useful for caching the result of a computation. +pub struct Memo { + inner: Signal, + update: CopyValue>, +} + +impl From> for ReadOnlySignal +where + T: PartialEq, +{ + fn from(val: Memo) -> Self { + ReadOnlySignal::new(val.inner) + } +} + +impl Memo { + /// Create a new memo + #[track_caller] + pub fn new(mut f: impl FnMut() -> T + 'static) -> Self + where + T: PartialEq, + { + let dirty = Arc::new(AtomicBool::new(true)); + let (tx, mut rx) = futures_channel::mpsc::unbounded(); + + let callback = { + let dirty = dirty.clone(); + move || { + dirty.store(true, std::sync::atomic::Ordering::Relaxed); + tx.unbounded_send(()).unwrap(); + } + }; + let rc = ReactiveContext::new_with_callback( + callback, + current_scope_id().unwrap(), + Location::caller(), + ); + + // Create a new signal in that context, wiring up its dependencies and subscribers + let value = rc.run_in(&mut f); + let recompute = RefCell::new(Box::new(f) as Box T>); + let update = CopyValue::new(UpdateInformation { + dirty, + callback: recompute, + }); + let state: Signal = Signal::new(value); + + let memo = Memo { + inner: state, + update, + }; + + spawn(async move { + while rx.next().await.is_some() { + memo.recompute(); + } + }); + + memo + } + + /// Rerun the computation and update the value of the memo if the result has changed. + fn recompute(&self) + where + T: PartialEq, + { + let mut update_copy = self.update; + let update_write = update_copy.write(); + let peak = self.inner.peek(); + let new_value = (update_write.callback.borrow_mut())(); + if new_value != *peak { + drop(peak); + let mut copy = self.inner; + let mut write = copy.write(); + *write = new_value; + update_write + .dirty + .store(false, std::sync::atomic::Ordering::Relaxed); + } + } + + /// Get the scope that the signal was created in. + pub fn origin_scope(&self) -> ScopeId { + self.inner.origin_scope() + } + + /// Get the id of the signal. + pub fn id(&self) -> generational_box::GenerationalBoxId { + self.inner.id() + } +} + +impl Readable for Memo +where + T: PartialEq, +{ + type Target = T; + type Storage = UnsyncStorage; + + #[track_caller] + fn try_read(&self) -> Result, generational_box::BorrowError> { + let read = self.inner.try_read(); + match read { + Ok(r) => { + let needs_update = self + .update + .read() + .dirty + .swap(false, std::sync::atomic::Ordering::Relaxed); + if needs_update { + drop(r); + self.recompute(); + self.inner.try_read() + } else { + Ok(r) + } + } + Err(e) => Err(e), + } + } + + /// Get the current value of the signal. **Unlike read, this will not subscribe the current scope to the signal which can cause parts of your UI to not update.** + /// + /// If the signal has been dropped, this will panic. + #[track_caller] + fn peek(&self) -> ReadableRef { + self.inner.peek() + } +} + +impl IntoAttributeValue for Memo +where + T: Clone + IntoAttributeValue + PartialEq, +{ + fn into_value(self) -> dioxus_core::AttributeValue { + self.with(|f| f.clone().into_value()) + } +} + +impl PartialEq for Memo { + fn eq(&self, other: &Self) -> bool { + self.inner == other.inner + } +} + +impl Deref for Memo +where + T: PartialEq, +{ + type Target = dyn Fn() -> T; + + fn deref(&self) -> &Self::Target { + Readable::deref_impl(self) + } +} diff --git a/packages/signals/src/reactive_context.rs b/packages/signals/src/reactive_context.rs index 98b611bb1..86eebb687 100644 --- a/packages/signals/src/reactive_context.rs +++ b/packages/signals/src/reactive_context.rs @@ -1,9 +1,9 @@ use dioxus_core::prelude::{ current_scope_id, has_context, provide_context, schedule_update_any, ScopeId, }; +use futures_channel::mpsc::UnboundedReceiver; use generational_box::SyncStorage; -use rustc_hash::FxHashSet; -use std::{cell::RefCell, hash::Hash, sync::Arc}; +use std::{cell::RefCell, hash::Hash}; use crate::{CopyValue, Readable, Writable}; @@ -25,66 +25,48 @@ thread_local! { impl std::fmt::Display for ReactiveContext { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { let read = self.inner.read(); - match read.scope_subscriber { - Some(scope) => write!(f, "ReactiveContext for scope {:?}", scope), - None => { - #[cfg(debug_assertions)] - return write!(f, "ReactiveContext created at {}", read.origin); - #[cfg(not(debug_assertions))] - write!(f, "ReactiveContext") - } - } - } -} - -impl Default for ReactiveContext { - #[track_caller] - fn default() -> Self { - Self::new_for_scope(None, std::panic::Location::caller()) + #[cfg(debug_assertions)] + return write!(f, "ReactiveContext created at {}", read.origin); + #[cfg(not(debug_assertions))] + write!(f, "ReactiveContext") } } impl ReactiveContext { /// Create a new reactive context #[track_caller] - pub fn new() -> Self { - Self::default() + pub fn new() -> (Self, UnboundedReceiver<()>) { + Self::new_with_origin(std::panic::Location::caller()) } /// Create a new reactive context with a location for debugging purposes /// This is useful for reactive contexts created within closures - pub fn new_with_origin(origin: &'static std::panic::Location<'static>) -> Self { - Self::new_for_scope(None, origin) + pub fn new_with_origin( + origin: &'static std::panic::Location<'static>, + ) -> (Self, UnboundedReceiver<()>) { + let (tx, rx) = futures_channel::mpsc::unbounded(); + let callback = move || { + let _ = tx.unbounded_send(()); + }; + let _self = Self::new_with_callback(callback, current_scope_id().unwrap(), origin); + (_self, rx) } - /// Create a new reactive context that may update a scope - #[allow(unused)] - pub(crate) fn new_for_scope( - scope: Option, + /// Create a new reactive context that may update a scope. When any signal that this context subscribes to changes, the callback will be run + pub fn new_with_callback( + callback: impl FnMut() + Send + Sync + 'static, + scope: ScopeId, origin: &'static std::panic::Location<'static>, ) -> Self { - let (tx, rx) = flume::unbounded(); - - let mut scope_subscribers = FxHashSet::default(); - if let Some(scope) = scope { - scope_subscribers.insert(scope); - } - let inner = Inner { - scope_subscriber: scope, - sender: tx, self_: None, - update_any: schedule_update_any(), - receiver: rx, + update: Box::new(callback), #[cfg(debug_assertions)] origin, }; let mut self_ = Self { - inner: CopyValue::new_maybe_sync_in_scope( - inner, - scope.or_else(current_scope_id).unwrap(), - ), + inner: CopyValue::new_maybe_sync_in_scope(inner, scope), }; self_.inner.write().self_ = Some(self_); @@ -112,10 +94,17 @@ impl ReactiveContext { if let Some(cx) = has_context() { return Some(cx); } + let update_any = schedule_update_any(); + let scope_id = current_scope_id().unwrap(); + let update_scope = move || { + tracing::trace!("Marking scope {:?} as dirty", scope_id); + update_any(scope_id) + }; // Otherwise, create a new context at the current scope - Some(provide_context(ReactiveContext::new_for_scope( - current_scope_id(), + Some(provide_context(ReactiveContext::new_with_callback( + update_scope, + scope_id, std::panic::Location::caller(), ))) } @@ -137,25 +126,18 @@ impl ReactiveContext { /// /// Returns true if the context was marked as dirty, or false if the context has been dropped pub fn mark_dirty(&self) -> bool { - if let Ok(self_read) = self.inner.try_read() { + let mut copy = self.inner; + if let Ok(mut self_write) = copy.try_write() { #[cfg(debug_assertions)] { - if let Some(scope) = self_read.scope_subscriber { - tracing::trace!("Marking reactive context for scope {:?} as dirty", scope); - } else { - tracing::trace!( - "Marking reactive context created at {} as dirty", - self_read.origin - ); - } - } - if let Some(scope) = self_read.scope_subscriber { - (self_read.update_any)(scope); + tracing::trace!( + "Marking reactive context created at {} as dirty", + self_write.origin + ); } - // mark the listeners as dirty - // If the channel is full it means that the receivers have already been marked as dirty - _ = self_read.sender.try_send(()); + (self_write.update)(); + true } else { false @@ -166,12 +148,6 @@ impl ReactiveContext { pub fn origin_scope(&self) -> ScopeId { self.inner.origin_scope() } - - /// Wait for this reactive context to change - pub async fn changed(&self) { - let rx = self.inner.read().receiver.clone(); - _ = rx.recv_async().await; - } } impl Hash for ReactiveContext { @@ -181,14 +157,10 @@ impl Hash for ReactiveContext { } struct Inner { - // A scope we mark as dirty when this context is written to - scope_subscriber: Option, self_: Option, - update_any: Arc, // Futures will call .changed().await - sender: flume::Sender<()>, - receiver: flume::Receiver<()>, + update: Box, // Debug information for signal subscriptions #[cfg(debug_assertions)] diff --git a/packages/signals/src/signal.rs b/packages/signals/src/signal.rs index 68e851c4d..cdf8595bf 100644 --- a/packages/signals/src/signal.rs +++ b/packages/signals/src/signal.rs @@ -1,11 +1,9 @@ +use crate::Memo; use crate::{ read::Readable, write::Writable, CopyValue, GlobalMemo, GlobalSignal, ReactiveContext, - ReadOnlySignal, ReadableRef, -}; -use dioxus_core::{ - prelude::{spawn, IntoAttributeValue}, - ScopeId, + ReadableRef, }; +use dioxus_core::{prelude::IntoAttributeValue, ScopeId}; use generational_box::{AnyStorage, Storage, SyncStorage, UnsyncStorage}; use std::{ any::Any, @@ -88,35 +86,8 @@ impl Signal { /// /// Selectors can be used to efficiently compute derived data from signals. #[track_caller] - pub fn memo(f: impl FnMut() -> T + 'static) -> ReadOnlySignal { - Self::use_maybe_sync_memo(f) - } - - /// Creates a new Selector that may be Sync + Send. The selector will be run immediately and whenever any signal it reads changes. - /// - /// Selectors can be used to efficiently compute derived data from signals. - #[track_caller] - pub fn use_maybe_sync_memo>>( - mut f: impl FnMut() -> T + 'static, - ) -> ReadOnlySignal { - // Get the current reactive context - let rc = ReactiveContext::new(); - - // Create a new signal in that context, wiring up its dependencies and subscribers - let mut state: Signal = rc.run_in(|| Signal::new_maybe_sync(f())); - - spawn(async move { - loop { - rc.changed().await; - let new = f(); - if new != *state.peek() { - *state.write() = new; - } - } - }); - - // And just return the readonly variant of that signal - ReadOnlySignal::new_maybe_sync(state) + pub fn memo(f: impl FnMut() -> T + 'static) -> Memo { + Memo::new(f) } } @@ -237,7 +208,7 @@ impl>> Writable for Signal { } #[track_caller] - fn try_write(&self) -> Result, generational_box::BorrowMutError> { + fn try_write(&mut self) -> Result, generational_box::BorrowMutError> { self.inner.try_write().map(|inner| { let borrow = S::map_mut(inner, |v| &mut v.value); Write { diff --git a/packages/signals/src/write.rs b/packages/signals/src/write.rs index 4cba48c62..8abc54b44 100644 --- a/packages/signals/src/write.rs +++ b/packages/signals/src/write.rs @@ -27,7 +27,7 @@ pub trait Writable: Readable { } /// Try to get a mutable reference to the value. If the value has been dropped, this will panic. - fn try_write(&self) -> Result, generational_box::BorrowMutError>; + fn try_write(&mut self) -> Result, generational_box::BorrowMutError>; /// Run a function with a mutable reference to the value. If the value has been dropped, this will panic. #[track_caller]