Compare commits
6 Commits
927d6b9888
...
d8f217d7cb
| Author | SHA1 | Date |
|---|---|---|
 Avril
|
d8f217d7cb
|
5 days ago |
|
Avril
|
aaaddd66ec
|
5 days ago |
|
Avril
|
ad03ce86e8
|
5 days ago |
|
Avril
|
9412636a0f
|
6 days ago |
|
Avril
|
a4f288fd33
|
6 days ago |
|
Avril
|
8d8f932081
|
1 week ago |
@ -0,0 +1,422 @@
|
||||
//! Implementation of `progress-reactive` signal hooking & passing features
|
||||
use super::*;
|
||||
|
||||
use std::{
|
||||
ops,
|
||||
thread,
|
||||
io,
|
||||
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()
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Create a new shared `Crosslink` that can be split into a `(CrosslinkSender, CrosslinkReceiver)` pair.
|
||||
///
|
||||
/// This convenience function is identical to `Arc::new(Crosslink::default())`.
|
||||
#[inline(always)]
|
||||
pub(super) fn crosslink_unsplit() -> Arc<Crosslink>
|
||||
{
|
||||
Arc::new(Default::default())
|
||||
}
|
||||
|
||||
/// Create a new shared `(CrosslinkSender, CrosslinkReceiver)` pair.
|
||||
///
|
||||
/// This convenience function is identical to `Arc::new(Crosslink::default()).into_split()`.
|
||||
#[inline(always)]
|
||||
pub(crate) fn crosslink() -> (CrosslinkSender, CrosslinkReceiver)
|
||||
{
|
||||
Arc::new(Crosslink::default()).into_split()
|
||||
}
|
||||
|
||||
/// Handle to a backing signal-watcher thread (see `spawn_signal_watcher()`.)
|
||||
#[derive(Debug)]
|
||||
pub struct Handle
|
||||
{
|
||||
joiner: thread::JoinHandle<io::Result<bool>>,
|
||||
handle: signal_hook::iterator::Handle,
|
||||
}
|
||||
|
||||
impl Handle {
|
||||
/// Check if the backing thread is running.
|
||||
pub fn is_running(&self) -> bool
|
||||
{
|
||||
! self.joiner.is_finished()
|
||||
}
|
||||
/// Attempt to close the handle without waiting for it to respond *at all*.
|
||||
#[inline]
|
||||
pub fn signal_close(&self)
|
||||
{
|
||||
self.handle.close();
|
||||
}
|
||||
/// Close the handle without joining the backing thread
|
||||
///
|
||||
/// If the backing thread completes without having to `join()`, that result is returned; otherwise, it is ignored.
|
||||
#[inline]
|
||||
pub fn close(self) -> io::Result<bool>
|
||||
{
|
||||
self.handle.close();
|
||||
if self.joiner.is_finished() {
|
||||
return self.joiner.join().unwrap();
|
||||
}
|
||||
Ok(true)
|
||||
}
|
||||
/// Close the handle and wait for the backing thread to complete.
|
||||
#[inline]
|
||||
pub fn close_sync(self) -> io::Result<bool>
|
||||
{
|
||||
self.handle.close();
|
||||
self.joiner.join().expect("background signal_hook thread panicked!")
|
||||
}
|
||||
|
||||
/// Check if the signal watcher has been requested to close.
|
||||
#[inline]
|
||||
pub fn is_closed(&self) -> bool
|
||||
{
|
||||
self.handle.is_closed()
|
||||
}
|
||||
|
||||
/// Get a handle to the background signal watcher
|
||||
#[inline(always)]
|
||||
pub fn signal_handle(&self) -> &signal_hook::iterator::Handle
|
||||
{
|
||||
&self.handle
|
||||
}
|
||||
}
|
||||
|
||||
fn spawn_signal_watcher_with_callback<I, S, F: Send + 'static>(signals: I, mut callback: F) -> io::Result<(CrosslinkReceiver, Handle)>
|
||||
where I: IntoIterator<Item = S>,
|
||||
S: std::borrow::Borrow<std::ffi::c_int>,
|
||||
F: FnMut(std::ffi::c_int, &CrosslinkSender) -> io::Result<bool>,
|
||||
{
|
||||
use signal_hook::{
|
||||
consts::signal,
|
||||
iterator::Signals,
|
||||
};
|
||||
|
||||
let (tx, rx) = crosslink();
|
||||
let mut signals = Signals::new(signals)?;
|
||||
let handle = signals.handle();
|
||||
let joiner = thread::spawn(move || {
|
||||
let handle = signals.handle();
|
||||
let _exit = defer::Defer::new(move || handle.close());
|
||||
|
||||
for signal in signals.forever() {
|
||||
match callback(signal, &tx) {
|
||||
res @ Ok(false) | res @ Err(_) => return res,
|
||||
_ => (),
|
||||
};
|
||||
}
|
||||
Ok(true)
|
||||
});
|
||||
Ok((rx, Handle {
|
||||
joiner,
|
||||
handle,
|
||||
}))
|
||||
}
|
||||
|
||||
//TODO: The API for spawning the `signal_hook` thread & returning a `CrosslinkReceiver` that it notifies on `SIGWINCH`s (with its join-handle) (that will break out of the loop and cleanly exit the thread if there are no receivers left.)
|
||||
|
||||
/// Spawn a background thread to intercept specified `signals` to this process.
|
||||
///
|
||||
/// When any of the specified signals are intercepted, a notification is sent to the returned [CrosslinkReceiver], which can be waited on in an async context.
|
||||
///
|
||||
/// # Closing
|
||||
/// The background thread can be explicitly communicated with through `Handle`, but will automatically close if a signal arrives *after* the returned receiver has been dropped.
|
||||
/// However, as the dropping of the receiver will not auto-shutdown the background thread *until* another signal comes in and it notices it cannot send the notification, it is desireable that one of the explicit `close()` methods on `Handle` be registered to be called when either the receiver becomes unavailable to the caller's task-set or the task is over.
|
||||
///
|
||||
/// Dropping the handle will not communicate a close, and since the lifetimes of the receiver and handle are seperated, they do not automatically interface with eachother.
|
||||
pub fn spawn_signal_watcher<I, S>(signals: I) -> io::Result<(CrosslinkReceiver, Handle)>
|
||||
where I: IntoIterator<Item = S>,
|
||||
S: std::borrow::Borrow<std::ffi::c_int>,
|
||||
{
|
||||
use std::collections::BTreeSet;
|
||||
let signals: BTreeSet<_> = signals.into_iter().map(|x| *x.borrow()).collect();
|
||||
|
||||
spawn_signal_watcher_with_callback(signals.clone(), move |sig, sender| {
|
||||
Ok(if signals.contains(&sig) {
|
||||
sender.try_notify()
|
||||
} else {
|
||||
return Err(io::Error::new(io::ErrorKind::NotFound, format!("Signal {:?} was not found in the original signal list {:?}", sig, signals.iter().copied().collect::<Box<[_]>>())));
|
||||
})
|
||||
})
|
||||
}
|
||||
Loading…
Reference in new issue