Refine runtime trait

Author: martintmk

opentelemetry-sdk/src/logs/log_processor_with_async_runtime.rs

@@ -17,7 +17,7 @@ use std::{
 };
 
 use super::{BatchConfig, LogProcessor};
-use crate::runtime::{RuntimeChannel, TrySend};
+use crate::runtime::{to_interval_stream, RuntimeChannel, TrySend};
 use futures_channel::oneshot;
 use futures_util::{
     future::{self, Either},
@@ -129,10 +129,10 @@ impl<R: RuntimeChannel> BatchLogProcessor<R> {
         runtime.spawn(Box::pin(async move {
             // Timer will take a reference to the current runtime, so its important we do this within the
             // runtime.spawn()
-            let ticker = inner_runtime
-                .interval(config.scheduled_delay)
+            let ticker = to_interval_stream(inner_runtime.clone(), config.scheduled_delay)
                 .skip(1) // The ticker is fired immediately, so we should skip the first one to align with the interval.
                 .map(|_| BatchMessage::Flush(None));
+
             let timeout_runtime = inner_runtime.clone();
             let mut logs = Vec::new();
             let mut messages = Box::pin(stream::select(message_receiver, ticker));

opentelemetry-sdk/src/metrics/periodic_reader_with_async_runtime.rs

@@ -13,7 +13,7 @@ use futures_util::{
 };
 use opentelemetry::{otel_debug, otel_error};
 
-use crate::runtime::Runtime;
+use crate::runtime::{to_interval_stream, Runtime};
 use crate::{
     error::{OTelSdkError, OTelSdkResult},
     metrics::{exporter::PushMetricExporter, reader::SdkProducer, MetricError, MetricResult},
@@ -110,8 +110,7 @@ where
             let runtime = self.runtime.clone();
             let reader = reader.clone();
             self.runtime.spawn(Box::pin(async move {
-                let ticker = runtime
-                    .interval(self.interval)
+                let ticker = to_interval_stream(runtime.clone(), self.interval)
                     .skip(1) // The ticker is fired immediately, so we should skip the first one to align with the interval.
                     .map(|_| Message::Export);
                 let messages = Box::pin(stream::select(message_receiver, ticker));

opentelemetry-sdk/src/runtime.rs

@@ -8,40 +8,48 @@
 
 use futures_util::{future::BoxFuture, stream::Stream};
 use std::{fmt::Debug, future::Future, time::Duration};
+use futures_util::stream::unfold;
 use thiserror::Error;
 
 /// A runtime is an abstraction of an async runtime like [Tokio] or [async-std]. It allows
-/// OpenTelemetry to work with any current and hopefully future runtime implementation.
+/// OpenTelemetry to work with any current and hopefully future runtime implementations.
 ///
 /// [Tokio]: https://crates.io/crates/tokio
 /// [async-std]: https://crates.io/crates/async-std
+///
+/// # Note
+///
+/// OpenTelemetry expects a *multi-threaded* runtime because its types can move across threads.
+/// For this reason, this trait requires the `Send` and `Sync` bounds. Single-threaded runtimes
+/// can implement this trait in a way that spawns the tasks on the same thread as the calling code.
 #[cfg(feature = "experimental_async_runtime")]
 pub trait Runtime: Clone + Send + Sync + 'static {
-    /// A future stream, which returns items in a previously specified interval. The item type is
-    /// not important.
-    type Interval: Stream + Send;
-
-    /// A future, which resolves after a previously specified amount of time. The output type is
-    /// not important.
-    type Delay: Future + Send + Unpin;
-
-    /// Create a [futures_util::stream::Stream], which returns a new item every
-    /// [std::time::Duration].
-    fn interval(&self, duration: Duration) -> Self::Interval;
-
     /// Spawn a new task or thread, which executes the given future.
     ///
     /// # Note
     ///
     /// This is mainly used to run batch span processing in the background. Note, that the function
     /// does not return a handle. OpenTelemetry will use a different way to wait for the future to
-    /// finish when TracerProvider gets shutdown. At the moment this happens by blocking the
+    /// finish when `TracerProvider` gets shutdown. At the moment this happens by blocking the
     /// current thread. This means runtime implementations need to make sure they can still execute
     /// the given future even if the main thread is blocked.
     fn spawn(&self, future: BoxFuture<'static, ()>);
 
     /// Return a new future, which resolves after the specified [std::time::Duration].
-    fn delay(&self, duration: Duration) -> Self::Delay;
+    fn delay(&self, duration: Duration) -> impl Future<Output = ()> + Send + Sync + 'static;
+}
+
+/// Uses the given runtime to produce an interval stream.
+#[cfg(feature = "experimental_async_runtime")]
+pub(crate) fn to_interval_stream<T: Runtime>(runtime: T, interval: Duration) -> impl Stream<Item = ()> {
+    unfold((), move |_| {
+        let runtime_cloned = runtime.clone();
+
+        async move {
+            runtime_cloned.delay(interval).await;
+            Some(((), ()))
+        }
+    })
 }
 
 /// Runtime implementation, which works with Tokio's multi thread runtime.
@@ -59,21 +67,14 @@ pub struct Tokio;
     doc(cfg(all(feature = "experimental_async_runtime", feature = "rt-tokio")))
 )]
 impl Runtime for Tokio {
-    type Interval = tokio_stream::wrappers::IntervalStream;
-    type Delay = ::std::pin::Pin<Box<tokio::time::Sleep>>;
-
-    fn interval(&self, duration: Duration) -> Self::Interval {
-        crate::util::tokio_interval_stream(duration)
-    }
-
     fn spawn(&self, future: BoxFuture<'static, ()>) {
         #[allow(clippy::let_underscore_future)]
         // we don't have to await on the returned future to execute
         let _ = tokio::spawn(future);
     }
 
-    fn delay(&self, duration: Duration) -> Self::Delay {
-        Box::pin(tokio::time::sleep(duration))
+    fn delay(&self, duration: Duration) -> impl Future<Output=()> + Send + Sync + 'static {
+        tokio::time::sleep(duration)
     }
 }
 
@@ -104,13 +105,6 @@ pub struct TokioCurrentThread;
     )))
 )]
 impl Runtime for TokioCurrentThread {
-    type Interval = tokio_stream::wrappers::IntervalStream;
-    type Delay = ::std::pin::Pin<Box<tokio::time::Sleep>>;
-
-    fn interval(&self, duration: Duration) -> Self::Interval {
-        crate::util::tokio_interval_stream(duration)
-    }
-
     fn spawn(&self, future: BoxFuture<'static, ()>) {
         // We cannot force push tracing in current thread tokio scheduler because we rely on
         // BatchSpanProcessor to export spans in a background task, meanwhile we need to block the
@@ -127,8 +121,8 @@ impl Runtime for TokioCurrentThread {
         });
     }
 
-    fn delay(&self, duration: Duration) -> Self::Delay {
-        Box::pin(tokio::time::sleep(duration))
+    fn delay(&self, duration: Duration) -> impl Future<Output=()> + Send + Sync + 'static {
+        tokio::time::sleep(duration)
     }
 }
 
@@ -147,20 +141,13 @@ pub struct AsyncStd;
     doc(cfg(all(feature = "experimental_async_runtime", feature = "rt-async-std")))
 )]
 impl Runtime for AsyncStd {
-    type Interval = async_std::stream::Interval;
-    type Delay = BoxFuture<'static, ()>;
-
-    fn interval(&self, duration: Duration) -> Self::Interval {
-        async_std::stream::interval(duration)
-    }
-
     fn spawn(&self, future: BoxFuture<'static, ()>) {
         #[allow(clippy::let_underscore_future)]
         let _ = async_std::task::spawn(future);
     }
 
-    fn delay(&self, duration: Duration) -> Self::Delay {
-        Box::pin(async_std::task::sleep(duration))
+    fn delay(&self, duration: Duration) -> impl Future<Output=()> + Send + Sync + 'static {
+        async_std::task::sleep(duration)
     }
 }
 
@@ -193,7 +180,7 @@ pub enum TrySendError {
     /// Send failed due to the channel being closed.
     #[error("cannot send message to batch processor as the channel is closed")]
     ChannelClosed,
-    /// Any other send error that isnt covered above.
+    /// Any other send error that isn't covered above.
     #[error(transparent)]
     Other(#[from] Box<dyn std::error::Error + Send + Sync + 'static>),
 }

opentelemetry-sdk/src/trace/sampler/jaeger_remote/sampler.rs

@@ -1,4 +1,4 @@
-use crate::runtime::RuntimeChannel;
+use crate::runtime::{to_interval_stream, RuntimeChannel};
 use crate::trace::sampler::jaeger_remote::remote::SamplingStrategyResponse;
 use crate::trace::sampler::jaeger_remote::sampling_strategy::Inner;
 use crate::trace::{Sampler, ShouldSample};
@@ -189,7 +189,8 @@ impl JaegerRemoteSampler {
         C: HttpClient + 'static,
     {
         // todo: review if we need 'static here
-        let interval = runtime.interval(update_timeout);
+        let interval = to_interval_stream(runtime.clone(), update_timeout);
+
         runtime.spawn(Box::pin(async move {
             // either update or shutdown
             let mut update = Box::pin(stream::select(

opentelemetry-sdk/src/trace/span_processor_with_async_runtime.rs

@@ -1,6 +1,6 @@
 use crate::error::{OTelSdkError, OTelSdkResult};
 use crate::resource::Resource;
-use crate::runtime::{RuntimeChannel, TrySend};
+use crate::runtime::{to_interval_stream, RuntimeChannel, TrySend};
 use crate::trace::BatchConfig;
 use crate::trace::Span;
 use crate::trace::SpanProcessor;
@@ -308,7 +308,7 @@ impl<R: RuntimeChannel> BatchSpanProcessorInternal<R> {
         }
 
         let export = self.exporter.export(self.spans.split_off(0));
-        let timeout = self.runtime.delay(self.config.max_export_timeout);
+        let timeout = Box::pin(self.runtime.delay(self.config.max_export_timeout));
         let time_out = self.config.max_export_timeout;
 
         Box::pin(async move {
@@ -354,8 +354,7 @@ impl<R: RuntimeChannel> BatchSpanProcessor<R> {
         runtime.spawn(Box::pin(async move {
             // Timer will take a reference to the current runtime, so its important we do this within the
             // runtime.spawn()
-            let ticker = inner_runtime
-                .interval(config.scheduled_delay)
+            let ticker = to_interval_stream(inner_runtime.clone(), config.scheduled_delay)
                 .skip(1) // The ticker is fired immediately, so we should skip the first one to align with the interval.
                 .map(|_| BatchMessage::Flush(None));
             let timeout_runtime = inner_runtime.clone();