From c80aa7d22791c85f43e4f4fb166de434c2b63828 Mon Sep 17 00:00:00 2001 From: "ADOMAS BEKERAS (from Dev Box)" Date: Mon, 13 Jul 2026 11:53:35 +0200 Subject: [PATCH 1/4] Implement ThreadAware for more more types --- Cargo.lock | 1 + crates/thread_aware/Cargo.toml | 5 ++ crates/thread_aware/README.md | 4 +- crates/thread_aware/src/impls.rs | 23 ++++++++ crates/thread_aware/src/lib.rs | 1 + crates/thread_aware/src/third_party/mod.rs | 8 ++- crates/thread_aware/src/third_party/tokio.rs | 57 ++++++++++++++++++++ 7 files changed, 96 insertions(+), 3 deletions(-) create mode 100644 crates/thread_aware/src/third_party/tokio.rs diff --git a/Cargo.lock b/Cargo.lock index d9033289c..7bff95ad2 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3764,6 +3764,7 @@ dependencies = [ "mutants", "static_assertions", "thread_aware_macros", + "tokio", "uuid", ] diff --git a/crates/thread_aware/Cargo.toml b/crates/thread_aware/Cargo.toml index 11b87bf0c..8d127a620 100644 --- a/crates/thread_aware/Cargo.toml +++ b/crates/thread_aware/Cargo.toml @@ -47,6 +47,8 @@ allowed_external_types = [ "jiff::signed_duration::SignedDuration", "jiff::span::Span", "jiff::timestamp::Timestamp", + # `tokio` feature + "tokio::sync::mpsc::unbounded::UnboundedSender", # `uuid` feature "uuid::Uuid", ] @@ -80,6 +82,7 @@ threads = ["dep:many_cpus"] bytes = ["dep:bytes"] http = ["dep:http"] jiff02 = ["dep:jiff"] +tokio = ["dep:tokio"] uuid = ["dep:uuid"] [dependencies] @@ -90,6 +93,7 @@ thread_aware_macros = { workspace = true, optional = true } bytes = { version = "1.11.1", default-features = false, optional = true } http = { version = "1.4.1", default-features = false, features = ["std"], optional = true } jiff = { version = "0.2.21", default-features = false, optional = true } +tokio = { version = "1.48.0", default-features = false, features = ["sync"], optional = true } uuid = { version = "1.21.0", default-features = false, optional = true } [dev-dependencies] @@ -102,6 +106,7 @@ many_cpus = { workspace = true, features = ["test-util"] } mutants = { workspace = true } static_assertions = { workspace = true } thread_aware_macros = { path = "../thread_aware_macros" } +tokio = { version = "1.48.0", default-features = false, features = ["sync"] } uuid = { version = "1.21.0", default-features = false } # Workspace dep declares `default-features = false`; we re-enable `default` here diff --git a/crates/thread_aware/README.md b/crates/thread_aware/README.md index d0eb7b88d..ef81a1f11 100644 --- a/crates/thread_aware/README.md +++ b/crates/thread_aware/README.md @@ -127,6 +127,8 @@ the wrapped crate can be supported additively: * **`jiff02`**: Impls for `jiff::Timestamp`, `jiff::civil::DateTime`, etc. +* **`tokio`**: Impl for `tokio::sync::mpsc::UnboundedSender`. + * **`uuid`**: Impl for `uuid::Uuid`. ## Examples @@ -176,7 +178,7 @@ impl Service { This crate was developed as part of The Oxidizer Project. Browse this crate's source code. - [__cargo_doc2readme_dependencies_info]: ggGmYW0CYXZlMC43LjJhdIQbLiTyV0MU86EbZU15e0PmecoboQ9jo59bnAEbyDXw04U13GlhYvRhcoQbN-qa6ScfeXYbpUaqiTirDJMb-9jGY2W0shYbe4CUzbbLbn5hZIKCbHRocmVhZF9hd2FyZWUwLjguMIJzdGhyZWFkX2F3YXJlX21hY3Jvc2UwLjcuNQ + [__cargo_doc2readme_dependencies_info]: ggGmYW0CYXZlMC43LjJhdIQbLiTyV0MU86EbZU15e0PmecoboQ9jo59bnAEbyDXw04U13GlhYvRhcoQbYGr2MB-HcFMbhXOmoFr06X4bqEfiaABNbkkbelhYtCG4ukZhZIKCbHRocmVhZF9hd2FyZWUwLjguMIJzdGhyZWFkX2F3YXJlX21hY3Jvc2UwLjcuNQ [__link0]: https://docs.rs/thread_aware_macros/0.7.5/thread_aware_macros/?search=ThreadAware [__link1]: https://doc.rust-lang.org/stable/std/clone/trait.Clone.html [__link10]: https://docs.rs/thread_aware_macros/0.7.5/thread_aware_macros/?search=ThreadAware diff --git a/crates/thread_aware/src/impls.rs b/crates/thread_aware/src/impls.rs index bd8e35f32..c263a6ed2 100644 --- a/crates/thread_aware/src/impls.rs +++ b/crates/thread_aware/src/impls.rs @@ -2,6 +2,7 @@ // Licensed under the MIT License. use std::collections::HashMap; +use std::net::SocketAddr; use std::path::{Path, PathBuf}; use std::time::Duration; @@ -36,6 +37,7 @@ impl_transfer!(String); impl_transfer!(PathBuf); impl_transfer!(Duration); impl_transfer!(&Path); +impl_transfer!(SocketAddr); impl_transfer!(&'static str); @@ -332,6 +334,27 @@ mod tests { assert_eq!(err_string, Err("error".to_string())); } + #[test] + fn test_socket_addr() { + use std::net::{Ipv4Addr, Ipv6Addr, SocketAddr, SocketAddrV4, SocketAddrV6}; + + use crate::ThreadAware; + + let affinities = pinned_affinities(&[2]); + let source = Some(affinities[0]); + let destination = affinities[1]; + + let mut v4 = SocketAddr::V4(SocketAddrV4::new(Ipv4Addr::LOCALHOST, 8080)); + let expected_v4 = v4; + v4.relocate(source, destination); + assert_eq!(v4, expected_v4); + + let mut v6 = SocketAddr::V6(SocketAddrV6::new(Ipv6Addr::LOCALHOST, 9090, 0, 0)); + let expected_v6 = v6; + v6.relocate(source, destination); + assert_eq!(v6, expected_v6); + } + // std::sync::Arc a type that introduces sharing across threads and thus is very likely to introduce // contention. The main point of ThreadAware is to prevent contention where possible, so it should not be // implemented for Arc. If a user depends on Arc, they need to take special steps to decide how diff --git a/crates/thread_aware/src/lib.rs b/crates/thread_aware/src/lib.rs index e4e368120..5062f585c 100644 --- a/crates/thread_aware/src/lib.rs +++ b/crates/thread_aware/src/lib.rs @@ -114,6 +114,7 @@ //! `http::uri::PathAndQuery`, `http::uri::Port`, `http::Error`, //! `http::uri::InvalidUri`, `http::Request`, `http::Response`. //! * **`jiff02`**: Impls for `jiff::Timestamp`, `jiff::civil::DateTime`, etc. +//! * **`tokio`**: Impl for `tokio::sync::mpsc::UnboundedSender`. //! * **`uuid`**: Impl for `uuid::Uuid`. //! //! # Examples diff --git a/crates/thread_aware/src/third_party/mod.rs b/crates/thread_aware/src/third_party/mod.rs index bc48e1ba6..affce3a14 100644 --- a/crates/thread_aware/src/third_party/mod.rs +++ b/crates/thread_aware/src/third_party/mod.rs @@ -5,8 +5,9 @@ //! //! Each submodule is gated behind a Cargo feature named after the wrapped //! crate (and its major / 0.x minor where applicable): `bytes`, `http`, -//! `jiff02`, `uuid`. Enabling a feature pulls in that crate as a dependency -//! and exposes `ThreadAware` impls for inert, self-contained types from it. +//! `jiff02`, `tokio`, `uuid`. Enabling a feature pulls in that crate as a +//! dependency and exposes `ThreadAware` impls for inert, self-contained +//! types from it. //! By default no such features are enabled, so this crate does not pull in //! any of these wrapped crates as additional dependencies. //! @@ -53,5 +54,8 @@ mod http; #[cfg(any(test, feature = "jiff02"))] mod jiff02; +#[cfg(any(test, feature = "tokio"))] +mod tokio; + #[cfg(any(test, feature = "uuid"))] mod uuid; diff --git a/crates/thread_aware/src/third_party/tokio.rs b/crates/thread_aware/src/third_party/tokio.rs new file mode 100644 index 000000000..2846a67e7 --- /dev/null +++ b/crates/thread_aware/src/third_party/tokio.rs @@ -0,0 +1,57 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT License. + +//! `ThreadAware` impl for [`tokio::sync::mpsc::UnboundedSender`](UnboundedSender) (tokio 1.x). +//! +//! Enable with the `tokio` Cargo feature. +//! +//! `UnboundedSender` is a cheaply-cloneable, thread-safe handle onto a +//! channel's shared queue: sending through it never touches thread-local +//! state, so relocating the handle itself is a no-op. This is unrelated to +//! the messages flowing through the channel — those are ordinary values of +//! `T` moving via `send`/`recv`, not something `relocate` reaches into. + +use ::tokio::sync::mpsc::UnboundedSender; + +use crate::ThreadAware; +use crate::affinity::Affinity; + +impl ThreadAware for UnboundedSender { + fn relocate(&mut self, _source: Option, _destination: Affinity) {} +} + +#[cfg(test)] +mod tests { + use std::rc::Rc; + + use ::tokio::sync::mpsc::{UnboundedSender, unbounded_channel}; + use static_assertions::{assert_impl_all, assert_not_impl_any}; + + use crate::ThreadAware; + use crate::affinity::pinned_affinities; + + assert_impl_all!(UnboundedSender: ThreadAware, Send, Sync, Clone); + + assert_not_impl_any!(UnboundedSender>: ThreadAware, Send); + + #[test] + fn unbounded_sender_relocate_is_noop_and_stays_usable() { + let affinities = pinned_affinities(&[2]); + let (mut tx, mut rx) = unbounded_channel::(); + + tx.send(1).expect("receiver is still alive"); + + tx.relocate(Some(affinities[0]), affinities[1]); + + let tx_clone = tx.clone(); + tx_clone.send(2).expect("receiver is still alive"); + tx.send(3).expect("receiver is still alive"); + drop(tx); + drop(tx_clone); + + assert_eq!(rx.try_recv().expect("first message"), 1); + assert_eq!(rx.try_recv().expect("second message"), 2); + assert_eq!(rx.try_recv().expect("third message"), 3); + assert!(rx.try_recv().is_err(), "channel must be drained and closed"); + } +} From de465ddc0a0e3d8a6abb2790c237d361cb9e70e6 Mon Sep 17 00:00:00 2001 From: "ADOMAS BEKERAS (from Dev Box)" Date: Mon, 13 Jul 2026 12:26:34 +0200 Subject: [PATCH 2/4] spelling --- .spelling | 1 + 1 file changed, 1 insertion(+) diff --git a/.spelling b/.spelling index bd7694ad0..7c82cf814 100644 --- a/.spelling +++ b/.spelling @@ -168,6 +168,7 @@ TTR Templated TinyLFU Tokio +tokio Tokio's Treiber Tunable From f86e19309129817de6532663f9788d978c69e86b Mon Sep 17 00:00:00 2001 From: "ADOMAS BEKERAS (from Dev Box)" Date: Mon, 13 Jul 2026 13:33:59 +0200 Subject: [PATCH 3/4] fix(thread_aware): remove tokio implementation Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: 6503296c-52d4-4236-82bb-958903e3e519 --- .spelling | 1 - Cargo.lock | 1 - crates/thread_aware/Cargo.toml | 5 -- crates/thread_aware/README.md | 4 +- crates/thread_aware/src/lib.rs | 1 - crates/thread_aware/src/third_party/mod.rs | 8 +-- crates/thread_aware/src/third_party/tokio.rs | 57 -------------------- 7 files changed, 3 insertions(+), 74 deletions(-) delete mode 100644 crates/thread_aware/src/third_party/tokio.rs diff --git a/.spelling b/.spelling index 7c82cf814..bd7694ad0 100644 --- a/.spelling +++ b/.spelling @@ -168,7 +168,6 @@ TTR Templated TinyLFU Tokio -tokio Tokio's Treiber Tunable diff --git a/Cargo.lock b/Cargo.lock index 7bff95ad2..d9033289c 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3764,7 +3764,6 @@ dependencies = [ "mutants", "static_assertions", "thread_aware_macros", - "tokio", "uuid", ] diff --git a/crates/thread_aware/Cargo.toml b/crates/thread_aware/Cargo.toml index 8d127a620..11b87bf0c 100644 --- a/crates/thread_aware/Cargo.toml +++ b/crates/thread_aware/Cargo.toml @@ -47,8 +47,6 @@ allowed_external_types = [ "jiff::signed_duration::SignedDuration", "jiff::span::Span", "jiff::timestamp::Timestamp", - # `tokio` feature - "tokio::sync::mpsc::unbounded::UnboundedSender", # `uuid` feature "uuid::Uuid", ] @@ -82,7 +80,6 @@ threads = ["dep:many_cpus"] bytes = ["dep:bytes"] http = ["dep:http"] jiff02 = ["dep:jiff"] -tokio = ["dep:tokio"] uuid = ["dep:uuid"] [dependencies] @@ -93,7 +90,6 @@ thread_aware_macros = { workspace = true, optional = true } bytes = { version = "1.11.1", default-features = false, optional = true } http = { version = "1.4.1", default-features = false, features = ["std"], optional = true } jiff = { version = "0.2.21", default-features = false, optional = true } -tokio = { version = "1.48.0", default-features = false, features = ["sync"], optional = true } uuid = { version = "1.21.0", default-features = false, optional = true } [dev-dependencies] @@ -106,7 +102,6 @@ many_cpus = { workspace = true, features = ["test-util"] } mutants = { workspace = true } static_assertions = { workspace = true } thread_aware_macros = { path = "../thread_aware_macros" } -tokio = { version = "1.48.0", default-features = false, features = ["sync"] } uuid = { version = "1.21.0", default-features = false } # Workspace dep declares `default-features = false`; we re-enable `default` here diff --git a/crates/thread_aware/README.md b/crates/thread_aware/README.md index ef81a1f11..d0eb7b88d 100644 --- a/crates/thread_aware/README.md +++ b/crates/thread_aware/README.md @@ -127,8 +127,6 @@ the wrapped crate can be supported additively: * **`jiff02`**: Impls for `jiff::Timestamp`, `jiff::civil::DateTime`, etc. -* **`tokio`**: Impl for `tokio::sync::mpsc::UnboundedSender`. - * **`uuid`**: Impl for `uuid::Uuid`. ## Examples @@ -178,7 +176,7 @@ impl Service { This crate was developed as part of The Oxidizer Project. Browse this crate's source code. - [__cargo_doc2readme_dependencies_info]: ggGmYW0CYXZlMC43LjJhdIQbLiTyV0MU86EbZU15e0PmecoboQ9jo59bnAEbyDXw04U13GlhYvRhcoQbYGr2MB-HcFMbhXOmoFr06X4bqEfiaABNbkkbelhYtCG4ukZhZIKCbHRocmVhZF9hd2FyZWUwLjguMIJzdGhyZWFkX2F3YXJlX21hY3Jvc2UwLjcuNQ + [__cargo_doc2readme_dependencies_info]: ggGmYW0CYXZlMC43LjJhdIQbLiTyV0MU86EbZU15e0PmecoboQ9jo59bnAEbyDXw04U13GlhYvRhcoQbN-qa6ScfeXYbpUaqiTirDJMb-9jGY2W0shYbe4CUzbbLbn5hZIKCbHRocmVhZF9hd2FyZWUwLjguMIJzdGhyZWFkX2F3YXJlX21hY3Jvc2UwLjcuNQ [__link0]: https://docs.rs/thread_aware_macros/0.7.5/thread_aware_macros/?search=ThreadAware [__link1]: https://doc.rust-lang.org/stable/std/clone/trait.Clone.html [__link10]: https://docs.rs/thread_aware_macros/0.7.5/thread_aware_macros/?search=ThreadAware diff --git a/crates/thread_aware/src/lib.rs b/crates/thread_aware/src/lib.rs index 5062f585c..e4e368120 100644 --- a/crates/thread_aware/src/lib.rs +++ b/crates/thread_aware/src/lib.rs @@ -114,7 +114,6 @@ //! `http::uri::PathAndQuery`, `http::uri::Port`, `http::Error`, //! `http::uri::InvalidUri`, `http::Request`, `http::Response`. //! * **`jiff02`**: Impls for `jiff::Timestamp`, `jiff::civil::DateTime`, etc. -//! * **`tokio`**: Impl for `tokio::sync::mpsc::UnboundedSender`. //! * **`uuid`**: Impl for `uuid::Uuid`. //! //! # Examples diff --git a/crates/thread_aware/src/third_party/mod.rs b/crates/thread_aware/src/third_party/mod.rs index affce3a14..bc48e1ba6 100644 --- a/crates/thread_aware/src/third_party/mod.rs +++ b/crates/thread_aware/src/third_party/mod.rs @@ -5,9 +5,8 @@ //! //! Each submodule is gated behind a Cargo feature named after the wrapped //! crate (and its major / 0.x minor where applicable): `bytes`, `http`, -//! `jiff02`, `tokio`, `uuid`. Enabling a feature pulls in that crate as a -//! dependency and exposes `ThreadAware` impls for inert, self-contained -//! types from it. +//! `jiff02`, `uuid`. Enabling a feature pulls in that crate as a dependency +//! and exposes `ThreadAware` impls for inert, self-contained types from it. //! By default no such features are enabled, so this crate does not pull in //! any of these wrapped crates as additional dependencies. //! @@ -54,8 +53,5 @@ mod http; #[cfg(any(test, feature = "jiff02"))] mod jiff02; -#[cfg(any(test, feature = "tokio"))] -mod tokio; - #[cfg(any(test, feature = "uuid"))] mod uuid; diff --git a/crates/thread_aware/src/third_party/tokio.rs b/crates/thread_aware/src/third_party/tokio.rs deleted file mode 100644 index 2846a67e7..000000000 --- a/crates/thread_aware/src/third_party/tokio.rs +++ /dev/null @@ -1,57 +0,0 @@ -// Copyright (c) Microsoft Corporation. -// Licensed under the MIT License. - -//! `ThreadAware` impl for [`tokio::sync::mpsc::UnboundedSender`](UnboundedSender) (tokio 1.x). -//! -//! Enable with the `tokio` Cargo feature. -//! -//! `UnboundedSender` is a cheaply-cloneable, thread-safe handle onto a -//! channel's shared queue: sending through it never touches thread-local -//! state, so relocating the handle itself is a no-op. This is unrelated to -//! the messages flowing through the channel — those are ordinary values of -//! `T` moving via `send`/`recv`, not something `relocate` reaches into. - -use ::tokio::sync::mpsc::UnboundedSender; - -use crate::ThreadAware; -use crate::affinity::Affinity; - -impl ThreadAware for UnboundedSender { - fn relocate(&mut self, _source: Option, _destination: Affinity) {} -} - -#[cfg(test)] -mod tests { - use std::rc::Rc; - - use ::tokio::sync::mpsc::{UnboundedSender, unbounded_channel}; - use static_assertions::{assert_impl_all, assert_not_impl_any}; - - use crate::ThreadAware; - use crate::affinity::pinned_affinities; - - assert_impl_all!(UnboundedSender: ThreadAware, Send, Sync, Clone); - - assert_not_impl_any!(UnboundedSender>: ThreadAware, Send); - - #[test] - fn unbounded_sender_relocate_is_noop_and_stays_usable() { - let affinities = pinned_affinities(&[2]); - let (mut tx, mut rx) = unbounded_channel::(); - - tx.send(1).expect("receiver is still alive"); - - tx.relocate(Some(affinities[0]), affinities[1]); - - let tx_clone = tx.clone(); - tx_clone.send(2).expect("receiver is still alive"); - tx.send(3).expect("receiver is still alive"); - drop(tx); - drop(tx_clone); - - assert_eq!(rx.try_recv().expect("first message"), 1); - assert_eq!(rx.try_recv().expect("second message"), 2); - assert_eq!(rx.try_recv().expect("third message"), 3); - assert!(rx.try_recv().is_err(), "channel must be drained and closed"); - } -} From 7f3c9cf479eb4a66eeaf05fd44e0c12a7e253f17 Mon Sep 17 00:00:00 2001 From: "ADOMAS BEKERAS (from Dev Box)" Date: Tue, 14 Jul 2026 16:11:36 +0200 Subject: [PATCH 4/4] test(thread_aware): remove redundant socket address test Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: 6503296c-52d4-4236-82bb-958903e3e519 --- crates/thread_aware/src/impls.rs | 21 --------------------- 1 file changed, 21 deletions(-) diff --git a/crates/thread_aware/src/impls.rs b/crates/thread_aware/src/impls.rs index c263a6ed2..297844da6 100644 --- a/crates/thread_aware/src/impls.rs +++ b/crates/thread_aware/src/impls.rs @@ -334,27 +334,6 @@ mod tests { assert_eq!(err_string, Err("error".to_string())); } - #[test] - fn test_socket_addr() { - use std::net::{Ipv4Addr, Ipv6Addr, SocketAddr, SocketAddrV4, SocketAddrV6}; - - use crate::ThreadAware; - - let affinities = pinned_affinities(&[2]); - let source = Some(affinities[0]); - let destination = affinities[1]; - - let mut v4 = SocketAddr::V4(SocketAddrV4::new(Ipv4Addr::LOCALHOST, 8080)); - let expected_v4 = v4; - v4.relocate(source, destination); - assert_eq!(v4, expected_v4); - - let mut v6 = SocketAddr::V6(SocketAddrV6::new(Ipv6Addr::LOCALHOST, 9090, 0, 0)); - let expected_v6 = v6; - v6.relocate(source, destination); - assert_eq!(v6, expected_v6); - } - // std::sync::Arc a type that introduces sharing across threads and thus is very likely to introduce // contention. The main point of ThreadAware is to prevent contention where possible, so it should not be // implemented for Arc. If a user depends on Arc, they need to take special steps to decide how