feat: minor tokio parity improvements

madsim-tokio/src/sim/runtime.rs

@@ -166,6 +166,29 @@ impl Handle {
     }
 }
 
+/// Dummy tokio runtime metrics.
+///
+/// This is only provided to provide API compatibility. Metric values are always zero.
+#[derive(Clone)]
+pub struct RuntimeMetrics;
+
+impl RuntimeMetrics {
+    /// Returns the number of worker threads used by the runtime.
+    pub fn num_workers(&self) -> usize {
+        0
+    }
+
+    /// Returns the current number of alive tasks in the runtime.
+    pub fn num_alive_tasks(&self) -> usize {
+        0
+    }
+
+    /// Returns the number of tasks currently scheduled in the runtime’s global queue.
+    pub fn global_queue_depth(&self) -> usize {
+        0
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;

madsim/src/sim/task/join.rs

@@ -1,3 +1,5 @@
+use std::any::Any;
+
 use super::*;
 
 /// An owned permission to join on a task (await its termination).
@@ -100,6 +102,23 @@ impl JoinError {
     pub fn is_panic(&self) -> bool {
         self.is_panic
     }
+
+    /// Consumes the join error, returning the object with which the task panicked.
+    pub fn into_panic(self) -> Box<dyn Any + Send + 'static> {
+        self.try_into_panic()
+            .expect("`JoinError` reason is not a panic.")
+    }
+
+    /// Consumes the join error, returning the object with which the task
+    /// panicked if the task terminated due to a panic. Otherwise, `self` is
+    /// returned.
+    pub fn try_into_panic(self) -> Result<Box<dyn Any + Send + 'static>, JoinError> {
+        if self.is_panic {
+            Ok(Box::new("task panicked"))
+        } else {
+            Err(self)
+        }
+    }
 }
 
 impl fmt::Display for JoinError {

madsim/src/sim/time/mod.rs

@@ -139,6 +139,22 @@ impl TimeHandle {
         }
     }
 
+    /// Require a `Future` to complete before the specified instant in time.
+    // TODO: make it Send
+    pub fn timeout_at<T: Future>(
+        &self,
+        deadline: Instant,
+        future: T,
+    ) -> impl Future<Output = Result<T::Output, error::Elapsed>> {
+        let timeout = self.sleep_until(deadline);
+        async move {
+            select_biased! {
+                res = future.fuse() => Ok(res),
+                _ = timeout.fuse() => Err(error::Elapsed),
+            }
+        }
+    }
+
     pub(crate) fn add_timer_at(
         &self,
         deadline: Instant,
@@ -162,6 +178,15 @@ pub fn timeout<T: Future>(
     handle.timeout(duration, future)
 }
 
+/// Require a `Future` to complete before the specified instant in time.
+pub fn timeout_at<T: Future>(
+    deadline: Instant,
+    future: T,
+) -> impl Future<Output = Result<T::Output, error::Elapsed>> {
+    let handle = TimeHandle::current();
+    handle.timeout_at(deadline, future)
+}
+
 /// Advances time.
 ///
 /// Increments the saved `Instant::now()` value by `duration`.