Executor

operators/Cargo.toml

@@ -42,6 +42,7 @@ uuid = { version = "0.8", features = ["serde", "v4", "v5"] }
 
 [dev-dependencies]
 geo-rand = { git = "https://github.com/lelongg/geo-rand", tag = "v0.3.0" }
+tokio-util = "0.6"
 rand = "0.8"
 tempfile = "3.1"
 

operators/src/error.rs

@@ -294,6 +294,11 @@ pub enum Error {
     Statistics {
         source: crate::util::statistics::StatisticsError,
     },
+
+    #[snafu(display("Executor error: {}", source))]
+    Executor {
+        source: crate::executor::error::ExecutorError,
+    },
 }
 
 impl From<geoengine_datatypes::error::Error> for Error {
@@ -353,3 +358,9 @@ impl From<crate::util::statistics::StatisticsError> for Error {
         Error::Statistics { source }
     }
 }
+
+impl From<crate::executor::error::ExecutorError> for Error {
+    fn from(source: crate::executor::error::ExecutorError) -> Self {
+        Self::Executor { source }
+    }
+}

operators/src/executor/error.rs

@@ -0,0 +1,39 @@
+use snafu::Snafu;
+use tokio::sync::mpsc::error::SendError;
+use tokio::sync::oneshot::error::RecvError;
+use tokio::task::JoinError;
+
+pub type Result<T> = std::result::Result<T, ExecutorError>;
+
+#[derive(Debug, Clone, Snafu)]
+pub enum ExecutorError {
+    Submission { message: String },
+    Panic,
+    Cancelled,
+}
+
+impl From<JoinError> for ExecutorError {
+    fn from(src: JoinError) -> Self {
+        if src.is_cancelled() {
+            ExecutorError::Cancelled
+        } else {
+            ExecutorError::Panic
+        }
+    }
+}
+
+impl<T> From<SendError<T>> for ExecutorError {
+    fn from(e: SendError<T>) -> Self {
+        Self::Submission {
+            message: e.to_string(),
+        }
+    }
+}
+
+impl From<RecvError> for ExecutorError {
+    fn from(e: RecvError) -> Self {
+        Self::Submission {
+            message: e.to_string(),
+        }
+    }
+}

operators/src/executor/mod.rs

@@ -0,0 +1,106 @@
+use error::Result;
+use futures::{Future, Stream, StreamExt};
+use std::hash::Hash;
+use std::marker::PhantomData;
+use std::sync::Arc;
+
+pub mod error;
+
+pub struct Executor<Key, T>
+where
+    Key: Hash + Clone + Eq + Send + 'static,
+    T: Sync + Send + 'static,
+{
+    _pk: PhantomData<Key>,
+    _pv: PhantomData<T>,
+}
+
+impl<Key, T> Executor<Key, T>
+where
+    Key: Hash + Clone + Eq + Send + 'static,
+    T: Sync + Send + 'static,
+{
+    /// Creates a new `Executor` instance, ready to serve computations. The buffer
+    /// size determines how much elements are at most kept  in memory per computation.
+    pub fn new() -> Executor<Key, T> {
+        Executor {
+            _pk: Default::default(),
+            _pv: Default::default(),
+        }
+    }
+
+    /// Submits a streaming computation to this executor. In contrast
+    /// to `Executor.submit_stream`, this method returns a Stream of
+    /// `Arc<T>` that allows to use the executor with non-cloneable
+    /// results.
+    ///
+    /// #Errors
+    /// This call fails, if the `Executor` was already closed.
+    pub async fn submit_stream_ref<F>(
+        &self,
+        _key: &Key,
+        stream: F,
+    ) -> Result<impl Stream<Item = Arc<T>>>
+    where
+        F: Stream<Item = T> + Send + 'static,
+    {
+        Ok(stream.map(|x| Arc::new(x)))
+    }
+
+    /// Submits a single-result computation to this executor. In contrast
+    /// to `Executor.submit`, this method returns an
+    /// `Arc<T>` that allows to use the executor with non-cloneable
+    /// results.
+    ///
+    /// #Errors
+    /// This call fails, if the `Executor` was already closed.
+    pub async fn submit_ref<F>(&self, _key: &Key, f: F) -> Result<Arc<T>>
+    where
+        F: Future<Output = T> + Send + 'static,
+    {
+        Ok(Arc::new(f.await))
+    }
+
+    pub async fn close(self) -> Result<()> {
+        Ok(())
+    }
+}
+
+impl<Key, T> Executor<Key, T>
+where
+    Key: Hash + Clone + Eq + Send + 'static,
+    T: Clone + Sync + Send + 'static,
+{
+    /// Submits a streaming computation to this executor. This method
+    /// returns a stream providing the results of the original (given) stream.
+    ///
+    /// #Errors
+    /// This call fails, if the `Executor` was already closed.
+    pub async fn submit_stream<F>(&self, _key: &Key, stream: F) -> Result<impl Stream<Item = T>>
+    where
+        F: Stream<Item = T> + Send + 'static,
+    {
+        Ok(stream)
+    }
+
+    /// Submits a single-result computation to this executor.
+    ///
+    /// #Errors
+    /// This call fails, if the `Executor` was already closed.
+    pub async fn submit<F>(&self, _key: &Key, f: F) -> Result<T>
+    where
+        F: Future<Output = T> + Send + 'static,
+    {
+        Ok(f.await)
+    }
+}
+
+impl<Key, T> Default for Executor<Key, T>
+where
+    Key: Hash + Clone + Eq + Send + 'static,
+    T: Sync + Send + 'static,
+{
+    fn default() -> Self {
+        Self::new()
+    }
+}

operators/src/lib.rs

@@ -25,6 +25,7 @@ pub mod adapters;
 #[macro_use]
 pub mod engine;
 pub mod error;
+pub mod executor;
 pub mod mock;
 pub mod opencl;
 pub mod plot;