Documented swimos_trigger.

Made promises more flexible.

Author: horned-sphere

client/runtime/src/runtime.rs

@@ -48,7 +48,7 @@ use swimos_utilities::trigger::promise;
 pub type BoxedDownlink = Box<dyn Downlink + Send + Sync + 'static>;
 type ByteChannel = (ByteWriter, ByteReader);
 type CallbackResult =
-    Result<promise::Receiver<Result<(), DownlinkRuntimeError>>, Arc<DownlinkRuntimeError>>;
+    Result<promise::Receiver<Result<(), Arc<DownlinkRuntimeError>>>, Arc<DownlinkRuntimeError>>;
 pub type DownlinkCallback = oneshot::Sender<CallbackResult>;
 
 impl Debug for DownlinkRegistrationRequest {
@@ -172,7 +172,7 @@ enum RuntimeEvent {
         kind: DownlinkKind,
         address: RemotePath,
         result: Result<(), DownlinkRuntimeError>,
-        tx: promise::Sender<Result<(), DownlinkRuntimeError>>,
+        tx: promise::Sender<Result<(), Arc<DownlinkRuntimeError>>>,
     },
     /// A downlink runtime has completed.
     DownlinkRuntimeComplete {
@@ -662,7 +662,7 @@ async fn runtime_task<Net, Ws, Provider>(
                     }
                 }
 
-                let _r = tx.provide(result);
+                let _r = tx.provide(result.map_err(Arc::new));
             }
             RuntimeEvent::Shutdown => {
                 for peer in peers.values_mut() {

client/runtime/src/tests.rs

@@ -387,7 +387,7 @@ struct ValueDownlinkContext {
     stopped: Arc<Notify>,
     handle_tx: mpsc::Sender<ValueDownlinkSet<i32>>,
     server: Server,
-    promise: promise::Receiver<Result<(), DownlinkRuntimeError>>,
+    promise: promise::Receiver<Result<(), Arc<DownlinkRuntimeError>>>,
     stop_tx: trigger::Sender,
 }
 
@@ -610,14 +610,14 @@ struct TrackingValueContext {
     spawned: Arc<Notify>,
     stopped: Arc<Notify>,
     handle_tx: mpsc::Sender<ValueDownlinkSet<i32>>,
-    promise: promise::Receiver<Result<(), DownlinkRuntimeError>>,
+    promise: promise::Receiver<Result<(), Arc<DownlinkRuntimeError>>>,
 }
 
 struct TrackingMapContext {
     spawned: Arc<Notify>,
     stopped: Arc<Notify>,
     tx: MapDownlinkHandle<i32, i32>,
-    promise: promise::Receiver<Result<(), DownlinkRuntimeError>>,
+    promise: promise::Receiver<Result<(), Arc<DownlinkRuntimeError>>>,
 }
 
 #[tokio::test]

client/swimos_client/src/lib.rs

@@ -379,7 +379,7 @@ impl From<NotYetSyncedError> for ValueDownlinkOperationError {
 #[derive(Debug, Clone)]
 pub struct ValueDownlinkView<T> {
     tx: mpsc::Sender<ValueDownlinkSet<T>>,
-    stop_rx: promise::Receiver<Result<(), DownlinkRuntimeError>>,
+    stop_rx: promise::Receiver<Result<(), Arc<DownlinkRuntimeError>>>,
 }
 
 impl<T> ValueDownlinkView<T> {
@@ -390,7 +390,7 @@ impl<T> ValueDownlinkView<T> {
     }
 
     /// Returns a receiver that completes with the result of downlink's internal task.
-    pub fn stop_notification(&self) -> promise::Receiver<Result<(), DownlinkRuntimeError>> {
+    pub fn stop_notification(&self) -> promise::Receiver<Result<(), Arc<DownlinkRuntimeError>>> {
         self.stop_rx.clone()
     }
 }
@@ -482,7 +482,7 @@ impl<'h, L> MapDownlinkBuilder<'h, L> {
 #[derive(Debug, Clone)]
 pub struct MapDownlinkView<K, V> {
     inner: MapDownlinkHandle<K, V>,
-    stop_rx: promise::Receiver<Result<(), DownlinkRuntimeError>>,
+    stop_rx: promise::Receiver<Result<(), Arc<DownlinkRuntimeError>>>,
 }
 
 impl<K, V> MapDownlinkView<K, V> {
@@ -512,7 +512,7 @@ impl<K, V> MapDownlinkView<K, V> {
     }
 
     /// Returns a receiver that completes with the result of downlink's internal task.
-    pub fn stop_notification(&self) -> promise::Receiver<Result<(), DownlinkRuntimeError>> {
+    pub fn stop_notification(&self) -> promise::Receiver<Result<(), Arc<DownlinkRuntimeError>>> {
         self.stop_rx.clone()
     }
 }
@@ -598,12 +598,12 @@ impl<'h, L> EventDownlinkBuilder<'h, L> {
 #[derive(Debug, Clone)]
 pub struct EventDownlinkView<T> {
     _type: PhantomData<T>,
-    stop_rx: promise::Receiver<Result<(), DownlinkRuntimeError>>,
+    stop_rx: promise::Receiver<Result<(), Arc<DownlinkRuntimeError>>>,
 }
 
 impl<T> EventDownlinkView<T> {
     /// Returns a receiver that completes with the result of downlink's internal task.
-    pub fn stop_notification(&self) -> promise::Receiver<Result<(), DownlinkRuntimeError>> {
+    pub fn stop_notification(&self) -> promise::Receiver<Result<(), Arc<DownlinkRuntimeError>>> {
         self.stop_rx.clone()
     }
 }

runtime/swimos_runtime/src/agent/task/remotes/tests.rs

@@ -239,5 +239,5 @@ async fn remove_remote() {
         .expect("Timed out.")
         .expect("Reason promised dropped.");
 
-    assert_eq!(*result, DisconnectionReason::RemoteTimedOut);
+    assert_eq!(result, DisconnectionReason::RemoteTimedOut);
 }

runtime/swimos_runtime/src/agent/task/tests/mod.rs

@@ -600,10 +600,7 @@ impl RemoteReceiver {
         if !lanes.is_empty() {
             panic!("Some lanes were not unlinked: {:?}", lanes);
         }
-        let reason = completion_rx
-            .await
-            .map(|arc| *arc)
-            .unwrap_or(DisconnectionReason::Failed);
+        let reason = completion_rx.await.unwrap_or(DisconnectionReason::Failed);
 
         assert_eq!(
             reason,

server/swimos_server_app/src/server/runtime/mod.rs

@@ -1037,7 +1037,7 @@ async fn attach_agent(
         Ok(true) => {
             if provider.send(Ok((in_tx, out_rx))).is_ok() {
                 if let Ok(reason) = disconnect_rx.await {
-                    *reason
+                    reason
                 } else {
                     DisconnectionReason::Failed
                 }
@@ -1077,7 +1077,7 @@ async fn attach_link_remote(
                 info!("Downlink request dropped before satisfied.");
                 DisconnectionReason::Failed
             } else if let Ok(reason) = disconnect_rx.await {
-                *reason
+                reason
             } else {
                 DisconnectionReason::Failed
             }

swimos_utilities/swimos_trigger/src/lib.rs

@@ -12,6 +12,14 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+//! # Asynchronous Trigger
+//!
+//! This crate provides a [SPMC barrier](trigger::trigger) where any number of parties can wait for
+//! a single other party to complete an action.
+//!
+//! Additionally, a [promise](promise::promise) implementation is provided, implemented in terms of
+//! the trigger, that allows any number of parties to wait for a single producer to generate a value.
+
 pub mod promise;
 mod trigger;
 

swimos_utilities/swimos_trigger/src/promise/mod.rs

@@ -26,7 +26,7 @@ use std::sync::Arc;
 mod tests;
 
 #[derive(Debug)]
-struct PromiseInner<T>(Arc<UnsafeCell<Option<Arc<T>>>>);
+struct PromiseInner<T>(Arc<UnsafeCell<Option<T>>>);
 
 unsafe impl<T: Send> Send for PromiseInner<T> {}
 unsafe impl<T: Sync> Sync for PromiseInner<T> {}
@@ -55,7 +55,7 @@ impl<T> Clone for Receiver<T> {
 }
 
 /// A promise allows a value to be provided, exactly once, at some time in the future.
-pub fn promise<T: Send + Sync>() -> (Sender<T>, Receiver<T>) {
+pub fn promise<T: Send + Sync + Clone>() -> (Sender<T>, Receiver<T>) {
     let data = Arc::new(UnsafeCell::new(None));
     let (tx, rx) = crate::trigger();
     (
@@ -79,10 +79,10 @@ impl<T: Send + Sync> Sender<T> {
             data: PromiseInner(data),
         } = self;
         unsafe {
-            *data.get() = Some(Arc::new(value));
+            *data.get() = Some(value);
             if trigger.trigger() {
                 Ok(())
-            } else if let Ok(value) = Arc::try_unwrap((*data.get()).take().unwrap()) {
+            } else if let Some(value) = (*data.get()).take() {
                 Err(value)
             } else {
                 unreachable!()
@@ -91,6 +91,7 @@ impl<T: Send + Sync> Sender<T> {
     }
 }
 
+/// An error generated if the [`Sender`] was dropped before it provided a value.
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
 pub struct PromiseError;
 
@@ -101,15 +102,16 @@ impl Display for PromiseError {
 }
 
 impl<T> Receiver<T> {
+    /// Determine if two [`Receiver`]s are connected to the same [`Sender`].
     pub fn same_promise(left: &Self, right: &Self) -> bool {
         crate::Receiver::same_receiver(&left.trigger, &right.trigger)
     }
 }
 
 impl Error for PromiseError {}
 
-impl<T: Send + Sync> Future for Receiver<T> {
-    type Output = Result<Arc<T>, PromiseError>;
+impl<T: Send + Sync + Clone> Future for Receiver<T> {
+    type Output = Result<T, PromiseError>;
 
     fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
         if ready!(self.trigger.poll_unpin(cx)).is_ok() {

swimos_utilities/swimos_trigger/src/promise/tests.rs

@@ -32,7 +32,7 @@ async fn await_promise() {
 
     let receive_task = async move {
         let result = rx.await;
-        assert!(matches!(result, Ok(v) if *v == 4));
+        assert!(matches!(result, Ok(v) if v == 4));
     };
 
     join(send_task, receive_task).await;
@@ -49,7 +49,7 @@ async fn await_promise_threaded() {
 
         let receive_task = async move {
             let result = rx.await;
-            assert!(matches!(result, Ok(v) if *v == 4));
+            assert!(matches!(result, Ok(v) if v == 4));
         };
 
         let (r1, r2) = join(tokio::spawn(send_task), tokio::spawn(receive_task)).await;

swimos_utilities/swimos_trigger/src/trigger/mod.rs

@@ -34,21 +34,23 @@ struct TriggerInner {
     waiters: Mutex<Slab<Waker>>,
 }
 
+/// The sending party of a trigger.
 #[derive(Debug)]
 pub struct Sender {
     inner: Option<Weak<TriggerInner>>,
 }
 
 impl Error for TriggerError {}
 
+/// A receiving party of a trigger.
 #[derive(Clone, Debug)]
 pub struct Receiver {
     inner: Arc<TriggerInner>,
     slot: Option<usize>,
 }
 
 impl Receiver {
-    /// Determine if two triggers are the same.
+    /// Determine if two [`Receiver`]s are connected to the same [`Sender`].
     pub fn same_receiver(this: &Self, other: &Self) -> bool {
         Arc::ptr_eq(&this.inner, &other.inner)
     }
@@ -99,6 +101,7 @@ impl Drop for Sender {
     }
 }
 
+/// An error generated if the [`Sender`] was dropped before it was triggered.
 #[derive(Debug, PartialEq, Eq)]
 pub struct TriggerError;
 
@@ -139,6 +142,7 @@ impl Future for Receiver {
 }
 
 impl Receiver {
+    /// Check if the [`Sender`] has been dropped (regardless of whether it has been triggered).
     pub fn check_state(&self) -> Option<Result<(), TriggerError>> {
         let Receiver { inner, .. } = self;
         let flag = inner.flag.load(Ordering::Acquire);