Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions tokio/src/sync/barrier.rs
Original file line number Diff line number Diff line change
Expand Up @@ -74,6 +74,7 @@ impl Barrier {
let resource_span = {
let location = std::panic::Location::caller();
let resource_span = tracing::trace_span!(
parent: None,
"runtime.resource",
concrete_type = "Barrier",
kind = "Sync",
Expand Down
1 change: 1 addition & 0 deletions tokio/src/sync/mutex.rs
Original file line number Diff line number Diff line change
Expand Up @@ -340,6 +340,7 @@ impl<T: ?Sized> Mutex<T> {
let location = std::panic::Location::caller();

tracing::trace_span!(
parent: None,
"runtime.resource",
concrete_type = "Mutex",
kind = "Sync",
Expand Down
1 change: 1 addition & 0 deletions tokio/src/sync/oneshot.rs
Original file line number Diff line number Diff line change
Expand Up @@ -473,6 +473,7 @@ pub fn channel<T>() -> (Sender<T>, Receiver<T>) {
let location = std::panic::Location::caller();

let resource_span = tracing::trace_span!(
parent: None,
"runtime.resource",
concrete_type = "Sender|Receiver",
kind = "Sync",
Expand Down
2 changes: 2 additions & 0 deletions tokio/src/sync/rwlock.rs
Original file line number Diff line number Diff line change
Expand Up @@ -209,6 +209,7 @@ impl<T: ?Sized> RwLock<T> {
let resource_span = {
let location = std::panic::Location::caller();
let resource_span = tracing::trace_span!(
parent: None,
"runtime.resource",
concrete_type = "RwLock",
kind = "Sync",
Expand Down Expand Up @@ -282,6 +283,7 @@ impl<T: ?Sized> RwLock<T> {
let location = std::panic::Location::caller();

let resource_span = tracing::trace_span!(
parent: None,
"runtime.resource",
concrete_type = "RwLock",
kind = "Sync",
Expand Down
1 change: 1 addition & 0 deletions tokio/src/sync/semaphore.rs
Original file line number Diff line number Diff line change
Expand Up @@ -400,6 +400,7 @@ impl Semaphore {
let location = std::panic::Location::caller();

tracing::trace_span!(
parent: None,
"runtime.resource",
concrete_type = "Semaphore",
kind = "Sync",
Expand Down
1 change: 1 addition & 0 deletions tokio/src/time/interval.rs
Original file line number Diff line number Diff line change
Expand Up @@ -122,6 +122,7 @@ fn internal_interval_at(
let location = location.expect("should have location if tracing");

tracing::trace_span!(
parent: None,
"runtime.resource",
concrete_type = "Interval",
kind = "timer",
Expand Down
248 changes: 248 additions & 0 deletions tokio/tests/tracing-instrumentation/tests/sync.rs
Original file line number Diff line number Diff line change
@@ -0,0 +1,248 @@
//! Tests for sync instrumentation.
//!
//! These tests ensure that the instrumentation for tokio
//! synchronization primitives is correct.

use tokio::sync;
use tracing_mock::{expect, subscriber};

#[tokio::test]
async fn test_barrier_creates_span() {
let barrier_span = expect::span()
.named("runtime.resource")
.with_target("tokio::sync::barrier");

let size_event = expect::event()
.with_target("runtime::resource::state_update")
.with_fields(expect::field("size").with_value(&1u64));

let arrived_event = expect::event()
.with_target("runtime::resource::state_update")
.with_fields(expect::field("arrived").with_value(&0));

let (subscriber, handle) = subscriber::mock()
.new_span(barrier_span.clone().with_explicit_parent(None))
.enter(barrier_span.clone())
.event(size_event)
.event(arrived_event)
.exit(barrier_span.clone())
.drop_span(barrier_span)
.run_with_handle();

{
let _guard = tracing::subscriber::set_default(subscriber);
let _ = sync::Barrier::new(1);
}

handle.assert_finished();
}

#[tokio::test]
async fn test_mutex_creates_span() {
let mutex_span = expect::span()
.named("runtime.resource")
.with_target("tokio::sync::mutex");

let locked_event = expect::event()
.with_target("runtime::resource::state_update")
.with_fields(expect::field("locked").with_value(&false));

let batch_semaphore_span = expect::span()
.named("runtime.resource")
.with_target("tokio::sync::batch_semaphore");

let batch_semaphore_permits_event = expect::event()
.with_target("runtime::resource::state_update")
.with_fields(expect::field("permits").with_value(&1u64))
.with_fields(expect::field("permits.op").with_value(&"override"));

let (subscriber, handle) = subscriber::mock()
.new_span(mutex_span.clone().with_explicit_parent(None))
.enter(mutex_span.clone())
.event(locked_event)
.new_span(batch_semaphore_span.clone())
.enter(batch_semaphore_span.clone())
.event(batch_semaphore_permits_event)
.exit(batch_semaphore_span.clone())
.exit(mutex_span.clone())
.drop_span(mutex_span)
.drop_span(batch_semaphore_span)
.run_with_handle();

{
let _guard = tracing::subscriber::set_default(subscriber);
let _ = sync::Mutex::new(true);
}

handle.assert_finished();
}

#[tokio::test]
async fn test_oneshot_creates_span() {
let oneshot_span = expect::span()
.named("runtime.resource")
.with_target("tokio::sync::oneshot");

let initial_tx_dropped_event = expect::event()
.with_target("runtime::resource::state_update")
.with_fields(expect::field("tx_dropped").with_value(&false))
.with_fields(expect::field("tx_dropped.op").with_value(&"override"));

let final_tx_dropped_event = expect::event()
.with_target("runtime::resource::state_update")
.with_fields(expect::field("tx_dropped").with_value(&true))
.with_fields(expect::field("tx_dropped.op").with_value(&"override"));

let initial_rx_dropped_event = expect::event()
.with_target("runtime::resource::state_update")
.with_fields(expect::field("rx_dropped").with_value(&false))
.with_fields(expect::field("rx_dropped.op").with_value(&"override"));

let final_rx_dropped_event = expect::event()
.with_target("runtime::resource::state_update")
.with_fields(expect::field("rx_dropped").with_value(&true))
.with_fields(expect::field("rx_dropped.op").with_value(&"override"));

let value_sent_event = expect::event()
.with_target("runtime::resource::state_update")
.with_fields(expect::field("value_sent").with_value(&false))
.with_fields(expect::field("value_sent.op").with_value(&"override"));

let value_received_event = expect::event()
.with_target("runtime::resource::state_update")
.with_fields(expect::field("value_received").with_value(&false))
.with_fields(expect::field("value_received.op").with_value(&"override"));

let async_op_span = expect::span()
.named("runtime.resource.async_op")
.with_target("tokio::sync::oneshot");

let async_op_poll_span = expect::span()
.named("runtime.resource.async_op.poll")
.with_target("tokio::sync::oneshot");

let (subscriber, handle) = subscriber::mock()
.new_span(oneshot_span.clone().with_explicit_parent(None))
.enter(oneshot_span.clone())
.event(initial_tx_dropped_event)
.exit(oneshot_span.clone())
.enter(oneshot_span.clone())
.event(initial_rx_dropped_event)
.exit(oneshot_span.clone())
.enter(oneshot_span.clone())
.event(value_sent_event)
.exit(oneshot_span.clone())
.enter(oneshot_span.clone())
.event(value_received_event)
.exit(oneshot_span.clone())
.enter(oneshot_span.clone())
.new_span(async_op_span.clone())
.exit(oneshot_span.clone())
.enter(async_op_span.clone())
.new_span(async_op_poll_span.clone())
.exit(async_op_span.clone())
.enter(oneshot_span.clone())
.event(final_tx_dropped_event)
.exit(oneshot_span.clone())
.enter(oneshot_span.clone())
.event(final_rx_dropped_event)
.exit(oneshot_span.clone())
.drop_span(oneshot_span)
.drop_span(async_op_span)
.drop_span(async_op_poll_span)
.run_with_handle();

{
let _guard = tracing::subscriber::set_default(subscriber);
let _ = sync::oneshot::channel::<bool>();
}

handle.assert_finished();
}

#[tokio::test]
async fn test_rwlock_creates_span() {
let rwlock_span = expect::span()
.named("runtime.resource")
.with_target("tokio::sync::rwlock");

let max_readers_event = expect::event()
.with_target("runtime::resource::state_update")
.with_fields(expect::field("max_readers").with_value(&0x1FFFFFFF_u64));

let write_locked_event = expect::event()
.with_target("runtime::resource::state_update")
.with_fields(expect::field("write_locked").with_value(&false));

let current_readers_event = expect::event()
.with_target("runtime::resource::state_update")
.with_fields(expect::field("current_readers").with_value(&0));

let batch_semaphore_span = expect::span()
.named("runtime.resource")
.with_target("tokio::sync::batch_semaphore");

let batch_semaphore_permits_event = expect::event()
.with_target("runtime::resource::state_update")
.with_fields(expect::field("permits").with_value(&1u64))
.with_fields(expect::field("permits.op").with_value(&"override"));

let (subscriber, handle) = subscriber::mock()
.new_span(rwlock_span.clone().with_explicit_parent(None))
.enter(rwlock_span.clone())
.event(max_readers_event)
.event(write_locked_event)
.event(current_readers_event)
.exit(rwlock_span.clone())
.enter(rwlock_span.clone())
.new_span(batch_semaphore_span.clone())
.enter(batch_semaphore_span.clone())
.event(batch_semaphore_permits_event)
.exit(batch_semaphore_span.clone())
.exit(rwlock_span.clone())
.drop_span(rwlock_span)
.drop_span(batch_semaphore_span)
.run_with_handle();

{
let _guard = tracing::subscriber::set_default(subscriber);
let _ = sync::RwLock::new(true);
}

handle.assert_finished();
}

#[tokio::test]
async fn test_semaphore_creates_span() {
let semaphore_span = expect::span()
.named("runtime.resource")
.with_target("tokio::sync::semaphore");

let batch_semaphore_span = expect::span()
.named("runtime.resource")
.with_target("tokio::sync::batch_semaphore");

let batch_semaphore_permits_event = expect::event()
.with_target("runtime::resource::state_update")
.with_fields(expect::field("permits").with_value(&1u64))
.with_fields(expect::field("permits.op").with_value(&"override"));

let (subscriber, handle) = subscriber::mock()
.new_span(semaphore_span.clone().with_explicit_parent(None))
.enter(semaphore_span.clone())
.new_span(batch_semaphore_span.clone())
.enter(batch_semaphore_span.clone())
.event(batch_semaphore_permits_event)
.exit(batch_semaphore_span.clone())
.exit(semaphore_span.clone())
.drop_span(semaphore_span)
.drop_span(batch_semaphore_span)
.run_with_handle();

{
let _guard = tracing::subscriber::set_default(subscriber);
let _ = sync::Semaphore::new(1);
}

handle.assert_finished();
}