runtime: Instrument with tracing (#6981)

## Summary

This is so that tracing subscribers, can correctly attribute information
to eg. opentelemetry distributed tracing IDs.

## API Changes

None

## Testing

Wrote a test, but didn't check it in because it would require the `std`
feature of tracing and make tracing-subscriber a dependency. Happy to
add the test if those things are ok.

```
    #[tokio::test]
    async fn test_spawn_propagates_tracing_span() {
        install_global_subscriber();
        let handle = TokioRuntime::current();

        let span = tracing::info_span!("parent_span");
        let parent_id = span.id();
        assert!(parent_id.is_some(), "subscriber must be active");
        let _enter = span.enter();

        // spawn: async future is instrumented with the parent span
        let spawned_id = handle.spawn(async { tracing::Span::current().id() }).await;
        assert_eq!(spawned_id, parent_id, "spawn should propagate span");

        // spawn_cpu: closure enters the parent span
        let cpu_id = handle.spawn_cpu(|| tracing::Span::current().id()).await;
        assert_eq!(cpu_id, parent_id, "spawn_cpu should propagate span");

        // spawn_blocking: closure enters the parent span on the blocking thread
        let blocking_id = handle
            .spawn_blocking(|| tracing::Span::current().id())
            .await;
        assert_eq!(
            blocking_id, parent_id,
            "spawn_blocking should propagate span"
        );
    }
```

Signed-off-by: Frederic Branczyk <fbranczyk@gmail.com>

Author: brancz

vortex-io/src/runtime/handle.rs

@@ -9,6 +9,7 @@ use std::task::Poll;
 use std::task::ready;
 
 use futures::FutureExt;
+use tracing::Instrument;
 use vortex_error::vortex_panic;
 
 use crate::runtime::AbortHandleRef;
@@ -65,11 +66,13 @@ impl Handle {
         R: Send + 'static,
     {
         let (send, recv) = oneshot::channel();
+        let span = tracing::Span::current();
         let abort_handle = self.runtime().spawn(
             async move {
                 // Task::detach allows the receiver to be dropped, so we ignore send errors.
                 drop(send.send(f.await));
             }
+            .instrument(span)
             .boxed(),
         );
         Task {
@@ -103,7 +106,9 @@ impl Handle {
         R: Send + 'static,
     {
         let (send, recv) = oneshot::channel();
+        let span = tracing::Span::current();
         let abort_handle = self.runtime().spawn_cpu(Box::new(move || {
+            let _guard = span.enter();
             // Optimistically avoid the work if the result won't be used.
             if !send.is_closed() {
                 // Task::detach allows the receiver to be dropped, so we ignore send errors.
@@ -123,7 +128,9 @@ impl Handle {
         R: Send + 'static,
     {
         let (send, recv) = oneshot::channel();
+        let span = tracing::Span::current();
         let abort_handle = self.runtime().spawn_blocking_io(Box::new(move || {
+            let _guard = span.enter();
             // Optimistically avoid the work if the result won't be used.
             if !send.is_closed() {
                 // Task::detach allows the receiver to be dropped, so we ignore send errors.