`progress-reactive`: Re-worked internal sync-to-async-signal barrier API to the much more simple and complete `reactive::Crosslink{,Sender,Receiver}`

Fortune for leanify-many's current commit: Half curse − 半凶

Cargo.toml

@@ -19,7 +19,7 @@ codegen-units = 1
 default = ["splash", "progress-reactive", "colour", "collect_err", "shutdown"]
 
 # Enable progress bar
-progress = ["termprogress", "pin-project", "terminal_size"]
+progress = ["termprogress", "pin-project"]
 
 # Enable threaded scheduler
 #
@@ -49,7 +49,7 @@ collect_err = []
 shutdown = ["libc"]
 
 # TODO: Implement this to: Capture `SIGWINCH` events and re-size + re-render the progress bar to the new terminal width. (XXX: Use a background thread (outside the thread-pool, as it's blocking) listening on `signal_hooks::Signals.forever()` for this that sends events through a shared Tokio `CondVar<TerminalWidth>` notify_all() call.)
-progress-reactive = ["progress", "tokio/signal", "signal-hook"]
+progress-reactive = ["progress", "tokio/signal", "signal-hook", "terminal_size"]
 
 [dependencies]
 lazy_static = "1.4"

src/progress.rs

@@ -43,6 +43,9 @@ use termprogress::{
     ProgressBar,
 };
 
+#[cfg(feature="progress-reactive")]
+mod reactive;
+
 #[derive(Debug)]
 pub struct Task
 {
@@ -457,12 +460,14 @@ pub fn create_progress<P: ProgressBar + WithTitle + Send + 'static,
 		    let mut has_blanked = false;
 		    while let Some(command) = commands.next().await {
 			match command {
+                            #[cfg(feature="progress-reactive")]
 			    CommandKind::Resize { to: None } => {
 				use std::os::fd::{
 				    AsFd,
 				    AsRawFd,
 				};
 				
+                                
 				if let Some((w, _)) = terminal_size::terminal_size()
 				    .or_else(|| terminal_size::terminal_size_using_fd(std::io::stderr().as_fd().as_raw_fd()))
 				{
@@ -480,6 +485,8 @@ pub fn create_progress<P: ProgressBar + WithTitle + Send + 'static,
 				
 				progress.update_dimensions(to.get());
 			    },
+                            #[allow(unreachable_patterns)]
+                            CommandKind::Resize { .. } => unimplemented!("`progress-reactive` feature not enabled! Cannot resize to terminal size."),
 			    CommandKind::BumpHigh(high) => {
 				let stat = stat.to_mut();
 				stat.high+=high;

src/progress/reactive.rs

@@ -0,0 +1,291 @@
+//! Implementation of `progress-reactive` signal hooking & passing features
+use super::*;
+
+use std::{
+    ops,
+    thread,
+    sync::{
+	Arc,
+	Weak,
+    },
+};
+use tokio::{
+    sync::{
+	Notify,
+	Mutex,
+	RwLock,
+    },
+};
+use futures::{
+    prelude::*,
+    stream::{
+	self,
+	Stream,
+    },
+    future::{
+	Aborted, Abortable, abortable, AbortHandle,
+	
+	Shared,
+	WeakShared,
+
+	Remote, RemoteHandle,
+    },
+};
+
+/// Inner type for sending pulse signal from sync backing thread to async task.
+#[derive(Debug, Default)]
+pub(super) struct Crosslink
+{
+    pub notification: Notify,
+}
+
+/// Sends pulse signals synchonously to an async
+#[derive(Debug, Clone)]
+pub struct CrosslinkSender(Weak<Crosslink>);
+
+/// Receives pulse signals in a way that can be `await`ed for.
+#[derive(Debug)] // NOTE: Not `Clone`, dealing with multiple receivers is too much of a headache with no `notify_all()`.
+pub struct CrosslinkReceiver(Arc<Crosslink>);
+
+impl Crosslink
+{
+    /// Consumes this owned reference into a `Clone`able future that can be waited on by multiple tasks at once.
+    #[inline] 
+    pub fn waiter_shared(self: Arc<Self>) -> impl Future<Output = ()> + Clone + Send + Sync + Unpin + Sized + 'static
+    {
+	async move {
+	    self.notification.notified().await
+	}.shared()
+    }
+
+    /// Create a `Clone`able future that can be waited on by multiple tasks at once, **but** is still lifetime-bound by-ref to the instance.
+    ///
+    /// # Outliving the owner
+    /// If a `'static` lifetime bound is required (e.g. due to spawning on a non-local task-set,) use `waiter_shared()` on `Arc<Self>`.
+    #[inline] 
+    pub fn create_waiter_shared(&self) -> impl Future<Output = ()> + Clone + Send + Sync + Unpin + Sized + use<'_>
+    {
+	self.notification.notified().shared()
+    }
+
+    /// Split the link into `(tx, rx)` pair.
+    ///
+    /// # Example usage
+    /**```
+    # use std::sync::Arc;
+    # use leanify_many::progress::reactive::*;
+    let (tx, rx) = Arc::new(Crosslink::default()).into_split();
+    let rx = tokio::spawn!(async move {
+    let rx = rx.into_stream();
+    let mut n = 0usize;
+    while let Some(_) = rx.next().await {
+    n+=1;
+    println!("Received notification {n} time(s)!");
+}
+    println!("Sender(s) all gone!");
+});
+    
+    // Notify the backing task twice.
+    tx.notify();
+    tx.notify();
+
+    // Drop the sender, and wait for the backing task to exit.
+    drop(tx);
+    rx.await.unwrap();
+    ```*/
+    #[inline] 
+    pub(super) fn into_split(self: Arc<Self>) -> (CrosslinkSender, CrosslinkReceiver)
+    {
+	let tx = Arc::downgrade(&self);
+	(CrosslinkSender(tx), CrosslinkReceiver(self))
+    }
+}
+
+impl CrosslinkReceiver
+{
+    fn has_senders(&self) -> bool
+    {
+	Arc::weak_count(&self.0) != 0
+    }
+    /// Consume receiver into a `Stream` that yields each notification on `.next()`.
+    ///
+    /// The stream ends when it is determined that there can be no more signals sent.
+    pub fn into_stream(self) -> impl Stream<Item = ()> + Send + Sync + 'static
+    {
+	stream::unfold(self, move |state| async move {
+	    if state.has_senders() { // If there are more than 0 senders (weak references.)
+		state.0.notification.notified().await; // Wait for a notification. (XXX: This may not complete if all senders drop *while* it's being waited on.)
+	    } else {
+		return None;
+	    }
+
+	    // If there are no senders left (i.e. we received a notification from the final sender `Drop`ing) we do not want to yield an element but end the stream.
+	    state.has_senders().then(move || ((), state))
+	})
+    }
+
+    /// Wait for a notification or for there to be no senders left.
+    ///
+    /// Note that this *will* complete spuriously if it is the final receiver and the final sender is dropped, however it **also** *may* complete spuriously before that.
+    ///
+    /// (This future is cancel-safe.)
+    #[inline]
+    #[must_use]
+    pub async fn try_wait(&self) -> bool
+    {
+	if self.has_senders() {
+	    return false;
+	}
+	
+	self.0.notification.notified().await;
+	self.has_senders()
+    }
+
+    /// Wait for a notification to be sent.
+    ///
+    /// # Panics
+    /// If a signal is not received before the last sender is dropped.
+    pub fn wait(&self) -> impl Future<Output = ()> + Send + Sync +  '_
+    {
+	#[inline(never)]
+	#[cold]
+	fn _panic_no_senders() -> !
+	{
+	    panic!("no senders left that can signal")
+	}
+	self.try_wait().map(|r| if !r {
+	    
+	})
+    }
+
+    /// Wait for a notification to be sent or a final sender to be dropped without monitoring the number of senders.
+    ///
+    /// # Safety
+    /// This function will return a non-completable future if there are already no senders when it is called.
+    /// It may be preferable to use `try_wait_unsafe()` instead, (as that returns `ready()` if there are none instead of `pending()`.)
+    #[inline(always)] 
+    fn wait_unsafe(&self) -> impl Future + Send + Sync + '_
+    {
+	self.0.notification.notified()
+    }
+
+    
+    /// Wait for a notification to be sent or a final sender to be dropped without monitoring the number of senders.
+    ///
+    /// # Safety
+    /// This function will return an immediately completed function if there are no senders when it is called.
+    /// But when the returned future completes it cannot be differentiated between an actual intentional `Sender::notify()` call, or if it's from the final sender being dropped.
+    #[inline] 
+    fn try_wait_unsafe(&self) -> impl Future + Send + Sync + '_
+    {
+	if ! self.has_senders() {
+	    future::ready(()).left_future()
+	} else {
+	    self.0.notification.notified().right_future()
+	}
+    }
+}
+
+impl ops::Drop for CrosslinkSender
+{
+    fn drop(&mut self)
+    {
+	// This is the last sender, dropping now.
+	if self.is_last_sender() {
+	    let n = {
+		// Remove the last sender from the receiver's view.
+		let this = std::mem::replace(&mut self.0, Weak::new());
+		
+		// So we will tell the receiver to wake up.
+		let Some(n) = this.upgrade() else {
+		    // If there are any...
+		    return
+		};
+		
+		// Ensure there are no living senders it can see, before...
+		drop(this); 
+		n
+	    };
+	    // ...we wake it up, so it knows to die.
+	    n.notification.notify();
+	}
+    }
+}
+
+impl CrosslinkSender
+{
+    #[inline(always)] 
+    fn has_receivers(&self) -> bool
+    {
+	Weak::strong_count(&self.0) > 0
+    }
+
+    #[inline(always)] 
+    pub fn is_last_sender(&self) -> bool
+    {
+	Weak::weak_count(&self.0) == 1
+    }
+
+    /// If there are receivers that can be notified.
+    ///
+    /// # **Non-atomic** operations
+    /// Note that it is still possible for `notify()` to panic if called after checking this, due to the lack of atomicity.
+    /// If atomicity is needed, either use `try_map_notify()` (or, if atomicity isn't needed, just ignore the result of `try_notify()` failing instead.)
+    #[inline] 
+    pub fn can_notify(&self) -> bool
+    {
+	self.has_receivers()
+    }
+
+    /// If there are any receivers, returns a thunk that when called will notify one.
+    ///
+    /// # Usage
+    /// It is **not** intended for the returned function be kept around long, it is entirely possible that by the time the function is invoked, there are no receivers left.
+    /// The function will attempt to notify, and if there was no receiver to notify, this will be ignored.
+    ///
+    /// The intended use is that if there is some work that needs to be done before sending the signal, but that can be skipped if there is no signal to send, the check can be made via a call to this method, and the signal can be sent by calling the returned thunk.
+    /** ```
+    # use std::sync::Arc;
+    # use leanify_many::progress::reactive::*;
+    # let (tx, _rx) = Arc::new(Crosslink::default()).into_split();
+    if let Some(notify) = tx.try_map_notify() {
+    /* ...do some work before sending the signal... */
+    notify();
+}
+    ```*/
+    #[inline]
+    #[must_use]
+    fn try_map_notify(&self) -> Option<impl FnOnce() + Unpin + Send + '_>
+    {
+	self.0.upgrade().map(|s| move || s.notification.notify())
+    }
+
+    /// Send a notification signal if possible.
+    ///
+    /// # Return value
+    /// If there was a receiver to notify.
+    #[must_use]
+    pub fn try_notify(&self) -> bool
+    {
+	self.0.upgrade().map(|s| s.notification.notify())
+	    .map(|_| true).unwrap_or(false)
+    }
+
+    /// Send a notification signal
+    ///
+    /// # Panics
+    /// If there are no receivers to notify (See [try_notify()](try_notify).)
+    pub fn notify(&self)
+    {
+	#[inline(never)]
+	#[cold]
+	fn _panic_no_waiters() -> !
+	{
+	    panic!("attempted to notify no waiters")
+	}
+	if !self.try_notify() {
+	    _panic_no_waiters()
+	}
+    }
+}
+

src/work.rs

@@ -147,7 +147,7 @@ where I: IntoIterator<Item=T>,
     #[cfg(feature="progress-reactive")]
     let _size_handle = {
 	let mut progress = progress.clone();
-	todo!("progress-reactive: Spawn the `signal_hooks` thread and set up the Tokio `CondVar` to wait on in a background task, sending `Resize{None} to `progress` when an event comes in.");
+	todo!("progress-reactive: Spawn the `signal_hooks` thread and set up the Tokio `CondVar` to wait on in a background task, sending `Resize{{to: None}} to `progress` when an event comes in.");
     };
     
     let display = {