diff --git a/crates/bytesbuf/src/mem/opaque_memory.rs b/crates/bytesbuf/src/mem/opaque_memory.rs index 8c374c466..4f699d6ba 100644 --- a/crates/bytesbuf/src/mem/opaque_memory.rs +++ b/crates/bytesbuf/src/mem/opaque_memory.rs @@ -3,6 +3,7 @@ #[cfg(not(test))] use alloc::boxed::Box; +use core::any::{Any, TypeId}; use thread_aware::ThreadAware; @@ -24,9 +25,21 @@ pub struct OpaqueMemory { impl OpaqueMemory { /// Creates a new instance of the adapter. + /// + /// # Panics + /// + /// Panics only if runtime type identification reports [`OpaqueMemory`] but the downcast of + /// the same value to [`OpaqueMemory`] fails, which would indicate a standard library defect. #[must_use] - pub fn new(inner: impl MemoryShared) -> Self { - Self { inner: Box::new(inner) } + pub fn new(inner: M) -> Self { + if TypeId::of::() == TypeId::of::() { + let inner: Box = Box::new(inner); + *inner + .downcast::() + .expect("the concrete type was verified as OpaqueMemory above") + } else { + Self { inner: Box::new(inner) } + } } /// Reserves at least `min_bytes` bytes of memory capacity. @@ -91,6 +104,16 @@ mod tests { assert!(builder.capacity() >= 1024); } + #[test] + fn accepts_existing_opaque_memory() { + let memory = OpaqueMemory::new(GlobalPool::new()); + let memory = OpaqueMemory::new(memory); + + let builder = memory.reserve(1024); + + assert!(builder.capacity() >= 1024); + } + #[test] fn memory_trait() { let provider = GlobalPool::new(); diff --git a/crates/fetch/Cargo.toml b/crates/fetch/Cargo.toml index 72ea9c287..2743edf96 100644 --- a/crates/fetch/Cargo.toml +++ b/crates/fetch/Cargo.toml @@ -19,9 +19,9 @@ repository = "https://github.com/microsoft/oxidizer/tree/main/crates/fetch" [package.metadata.cargo_check_external_types] allowed_external_types = [ - "bytesbuf::mem::global::GlobalPool", "bytesbuf::mem::has_memory::HasMemory", "bytesbuf::mem::memory::Memory", + "bytesbuf::mem::opaque_memory::OpaqueMemory", "data_privacy::redaction_engine::RedactionEngine", "fetch_options::*", "fetch_tls::*", diff --git a/crates/fetch/examples/http_client_app.rs b/crates/fetch/examples/http_client_app.rs index 9d7846a6c..90c301d92 100644 --- a/crates/fetch/examples/http_client_app.rs +++ b/crates/fetch/examples/http_client_app.rs @@ -6,7 +6,7 @@ use std::sync::Arc; -use bytesbuf::mem::GlobalPool; +use bytesbuf::mem::{GlobalPool, OpaqueMemory}; use fetch::HttpClient; use fetch::tls::TlsOptions; use ohno::ErrorExt; @@ -18,7 +18,7 @@ use tick::Clock; #[fundle::bundle] struct App { clock: Clock, - global_pool: GlobalPool, + global_pool: OpaqueMemory, client: HttpClient, } @@ -42,7 +42,7 @@ async fn main() -> Result<(), ohno::AppError> { // Initialize and set up the App instance; fundle ensures all fields are properly constructed. let app = App::builder() .clock(|_| Clock::new_tokio()) - .global_pool(|_| GlobalPool::new()) + .global_pool(|_| OpaqueMemory::new(GlobalPool::new())) .client({ let meter_provider = meter_provider.clone(); move |x| { diff --git a/crates/fetch/examples/http_client_custom.rs b/crates/fetch/examples/http_client_custom.rs index dea4f0dc8..1889cd426 100644 --- a/crates/fetch/examples/http_client_custom.rs +++ b/crates/fetch/examples/http_client_custom.rs @@ -4,7 +4,7 @@ //! Plugs a custom `EchoHandler` into [`fetch::custom::create_builder`] as the transport //! handler. Every request's body is returned verbatim in the response. -use bytesbuf::mem::GlobalPool; +use bytesbuf::mem::{GlobalPool, OpaqueMemory}; use fetch::custom::{CustomDeps, Isolation, create_builder}; use fetch::{HttpRequest, HttpResponse, HttpResponseBuilder}; use http::StatusCode; @@ -16,7 +16,7 @@ use tick::Clock; async fn main() -> Result<(), ohno::AppError> { let deps = CustomDeps { clock: Clock::new_tokio(), - global_pool: GlobalPool::new(), + memory: OpaqueMemory::new(GlobalPool::new()), extras: (), }; diff --git a/crates/fetch/src/custom.rs b/crates/fetch/src/custom.rs index e1bf7fff7..c850a4019 100644 --- a/crates/fetch/src/custom.rs +++ b/crates/fetch/src/custom.rs @@ -17,7 +17,7 @@ use std::borrow::Cow; use std::fmt::Debug; use std::sync::Arc; -use bytesbuf::mem::GlobalPool; +use bytesbuf::mem::OpaqueMemory; use http_extensions::{HttpBodyBuilder, RequestHandler}; use opentelemetry::metrics::Meter; use thread_aware::{PerCore, ThreadAware, unaware}; @@ -53,7 +53,7 @@ where /// Clock for timing operations and timeouts. pub clock: Clock, /// Memory pool for usage-neutral memory allocations. - pub global_pool: GlobalPool, + pub memory: OpaqueMemory, /// Extra dependencies forwarded verbatim to [`CustomContext::extras`]. pub extras: Extras, } @@ -210,12 +210,12 @@ impl HttpClient { runtime_name: runtime.into(), name: transport.into(), clock: deps.clock.clone(), - global_pool: deps.global_pool.clone(), + memory: deps.memory.clone(), isolation, inner: thread_aware::Arc::new_with((deps, unaware(factory)), |(deps, factory)| { Arc::new(move |options, meter, pool_index| { let context = CustomContext { - body_builder: create_body_builder(&deps.global_pool, &deps.clock, &options), + body_builder: create_body_builder(&deps.memory, &deps.clock, &options), clock: deps.clock.clone(), pool_index, extras: deps.extras.clone(), @@ -242,7 +242,7 @@ pub(crate) struct Transport { name: Cow<'static, str>, inner: thread_aware::Arc, clock: Clock, - global_pool: GlobalPool, + memory: OpaqueMemory, isolation: Isolation, } @@ -268,7 +268,7 @@ impl Transport { } pub(crate) fn create_body_builder(&self, options: &ClientOptions) -> HttpBodyBuilder { - create_body_builder(&self.global_pool, &self.clock, options) + create_body_builder(&self.memory, &self.clock, options) } } @@ -278,7 +278,7 @@ impl Debug for Transport { } } -pub(crate) fn create_body_builder(pool: &GlobalPool, clock: &Clock, options: &ClientOptions) -> HttpBodyBuilder { +pub(crate) fn create_body_builder(pool: &OpaqueMemory, clock: &Clock, options: &ClientOptions) -> HttpBodyBuilder { HttpBodyBuilder::new(pool.clone(), clock).with_options(options.response_body_options) } @@ -288,9 +288,11 @@ mod tests { use std::sync::Arc; use std::sync::atomic::{AtomicUsize, Ordering}; + use bytesbuf::BytesBuf; + use bytesbuf::mem::{GlobalPool, Memory, OpaqueMemory}; use http::StatusCode; use http_extensions::FakeHandler; - use thread_aware::unaware; + use thread_aware::{ThreadAware, unaware}; use super::{CustomContext, CustomDeps, Isolation, create_builder}; use crate::HttpResponseBuilder; @@ -301,7 +303,7 @@ mod tests { fn custom_deps() -> CustomDeps { CustomDeps { clock: FakeDeps::default().clock, - global_pool: bytesbuf::mem::GlobalPool::new(), + memory: bytesbuf::mem::OpaqueMemory::new(bytesbuf::mem::GlobalPool::new()), extras: (), } } @@ -311,6 +313,36 @@ mod tests { FakeHandler::from_fn(|_req| HttpResponseBuilder::new_fake().status(StatusCode::OK).build()) } + #[derive(Clone, Debug, ThreadAware)] + struct CustomMemory { + inner: GlobalPool, + } + + impl Memory for CustomMemory { + fn reserve(&self, min_bytes: usize) -> BytesBuf { + self.inner.reserve(min_bytes) + } + } + + #[cfg_attr(miri, ignore)] + #[tokio::test] + async fn custom_deps_accept_custom_opaque_memory() { + let deps = CustomDeps { + clock: FakeDeps::default().clock, + memory: OpaqueMemory::new(CustomMemory { inner: GlobalPool::new() }), + extras: (), + }; + + let client = create_builder("test-runtime", "test", ok_factory, Isolation::Shared, deps) + .insecure_allow_http() + .minimal_pipeline() + .build(); + + let response = client.post("http://example.com").text("custom pool").fetch().await.unwrap(); + + assert_eq!(response.status(), StatusCode::OK); + } + #[cfg_attr(miri, ignore)] #[tokio::test] async fn create_builder_serves_requests_through_custom_pipeline() { @@ -348,7 +380,7 @@ mod tests { let counter = Arc::new(AtomicUsize::new(0)); let deps = CustomDeps { clock: FakeDeps::default().clock, - global_pool: bytesbuf::mem::GlobalPool::new(), + memory: bytesbuf::mem::OpaqueMemory::new(bytesbuf::mem::GlobalPool::new()), extras: unaware(Arc::clone(&counter)), }; diff --git a/crates/fetch/src/fake.rs b/crates/fetch/src/fake.rs index 2669a4c13..6aecf5023 100644 --- a/crates/fetch/src/fake.rs +++ b/crates/fetch/src/fake.rs @@ -84,7 +84,7 @@ impl HttpClient { Isolation::Shared, CustomDeps { clock: deps.clock, - global_pool: bytesbuf::mem::GlobalPool::new(), + memory: bytesbuf::mem::OpaqueMemory::new(bytesbuf::mem::GlobalPool::new()), extras: handler, }, ) diff --git a/crates/fetch/src/tokio.rs b/crates/fetch/src/tokio.rs index 71a00d6f4..74bb0f6a4 100644 --- a/crates/fetch/src/tokio.rs +++ b/crates/fetch/src/tokio.rs @@ -33,7 +33,7 @@ pub struct TokioDeps { /// Clock for timing operations and timeouts. pub clock: Clock, /// Memory pool for usage-neutral memory allocations. - pub global_pool: bytesbuf::mem::GlobalPool, + pub global_pool: bytesbuf::mem::OpaqueMemory, } impl Default for TokioDeps { @@ -47,7 +47,7 @@ impl TokioDeps { #[must_use] pub fn with_clock(clock: &Clock) -> Self { Self { - global_pool: bytesbuf::mem::GlobalPool::new(), + global_pool: bytesbuf::mem::OpaqueMemory::new(bytesbuf::mem::GlobalPool::new()), clock: clock.clone(), } } @@ -64,7 +64,7 @@ impl HttpClient { pub fn builder_tokio(deps: impl Into) -> HttpClientBuilder { let deps = deps.into(); let clock = deps.clock.clone(); - let global_pool = deps.global_pool.clone(); + let memory = deps.global_pool.clone(); // Re-layer on top of the in-crate `builder_custom_internal` path: the // full `TokioDeps` rides through `CustomDeps::extras` so that the @@ -77,7 +77,7 @@ impl HttpClient { Isolation::Shared, CustomDeps { clock, - global_pool, + memory, extras: deps, }, ) diff --git a/crates/fetch/tests/telemetry_scope.rs b/crates/fetch/tests/telemetry_scope.rs index b3e09999e..685f988cc 100644 --- a/crates/fetch/tests/telemetry_scope.rs +++ b/crates/fetch/tests/telemetry_scope.rs @@ -94,7 +94,7 @@ async fn custom_transport_scope_attribute() { let deps = CustomDeps { clock: Clock::new_frozen(), - global_pool: bytesbuf::mem::GlobalPool::new(), + memory: bytesbuf::mem::OpaqueMemory::new(bytesbuf::mem::GlobalPool::new()), extras: (), }; @@ -128,7 +128,7 @@ async fn custom_transport_instrument_inherits_scope() { let deps = CustomDeps { clock: Clock::new_frozen(), - global_pool: bytesbuf::mem::GlobalPool::new(), + memory: bytesbuf::mem::OpaqueMemory::new(bytesbuf::mem::GlobalPool::new()), extras: (), }; diff --git a/crates/http_extensions/Cargo.toml b/crates/http_extensions/Cargo.toml index 5434d2dc2..19b9536e1 100644 --- a/crates/http_extensions/Cargo.toml +++ b/crates/http_extensions/Cargo.toml @@ -25,10 +25,10 @@ ignored = ["uuid"] allowed_external_types = [ "futures_core::stream::Stream", "bytes::bytes::Bytes", - "bytesbuf::mem::global::GlobalPool", "bytesbuf::mem::has_memory::HasMemory", "bytesbuf::mem::memory::Memory", "bytesbuf::mem::memory_shared::MemoryShared", + "bytesbuf::mem::opaque_memory::OpaqueMemory", "bytesbuf::view::BytesView", "http::*", "http_body::Body", diff --git a/crates/http_extensions/benches/http_request_builder.rs b/crates/http_extensions/benches/http_request_builder.rs index cbf6d28b5..6be54e5a5 100644 --- a/crates/http_extensions/benches/http_request_builder.rs +++ b/crates/http_extensions/benches/http_request_builder.rs @@ -153,7 +153,7 @@ fn entry(c: &mut Criterion) { // Use TransparentMemory instead of GlobalPool so that every reserve() call from the // serde_json writer results in a real heap allocation. This makes alloc_tracker report // the true number of memory reservations, which GlobalPool would otherwise absorb. - let transparent_body_builder = HttpBodyBuilder::with_custom_memory(TransparentMemory::new(), &tick::Clock::new_frozen()); + let transparent_body_builder = HttpBodyBuilder::new(TransparentMemory::new(), &tick::Clock::new_frozen()); let operation = session.operation("json_body_large_transparent"); group.bench_function("json_body_large_transparent", |b| { b.iter_custom(|iters| { diff --git a/crates/http_extensions/src/body/builder.rs b/crates/http_extensions/src/body/builder.rs index 9f8f38ffe..bcb6f8b03 100644 --- a/crates/http_extensions/src/body/builder.rs +++ b/crates/http_extensions/src/body/builder.rs @@ -1,7 +1,9 @@ // Copyright (c) Microsoft Corporation. // Licensed under the MIT License. -use bytesbuf::mem::{GlobalPool, HasMemory, Memory, MemoryShared, OpaqueMemory}; +#[cfg(any(feature = "test-util", test))] +use bytesbuf::mem::GlobalPool; +use bytesbuf::mem::{HasMemory, Memory, MemoryShared, OpaqueMemory}; use bytesbuf::{BytesBuf, BytesView}; use futures::{Stream, TryStreamExt}; use http_body::{Body, Frame}; @@ -44,7 +46,7 @@ use crate::{HttpError, Result}; /// With the `test-util` feature enabled, you can create a test instance using `HttpBodyBuilder::new_fake()`. #[derive(Debug, Clone, ThreadAware)] pub struct HttpBodyBuilder { - memory: MemoryWrapper, + memory: OpaqueMemory, clock: Clock, pub(super) options: HttpBodyOptions, } @@ -76,26 +78,11 @@ impl HttpBodyBuilder { /// Creates a new instance of [`HttpBodyBuilder`]. /// - /// This method uses a per-thread memory pool from [`GlobalPool`]. + /// The provided memory is type-erased unless it is already an [`OpaqueMemory`]. #[must_use] - pub fn new(memory: GlobalPool, clock: &Clock) -> Self { + pub fn new(memory: impl MemoryShared, clock: &Clock) -> Self { Self { - memory: MemoryWrapper::Global(memory), - clock: clock.clone(), - options: HttpBodyOptions::default(), - } - } - - /// Creates a new instance of [`HttpBodyBuilder`] with custom memory. - /// - /// The provided memory provider is type-erased and used in place of the global per-thread - /// memory used by [`HttpBodyBuilder::new`]. It remains thread-aware: when the builder is moved - /// between threads via a thread-aware runtime mechanism, the provider's thread-affine state is - /// relocated along with it. - #[must_use] - pub fn with_custom_memory(memory: impl MemoryShared, clock: &Clock) -> Self { - Self { - memory: MemoryWrapper::Opaque(OpaqueMemory::new(memory)), + memory: OpaqueMemory::new(memory), clock: clock.clone(), options: HttpBodyOptions::default(), } @@ -357,21 +344,6 @@ impl HasMemory for HttpBodyBuilder { } } -#[derive(Debug, Clone, ThreadAware)] -enum MemoryWrapper { - Global(GlobalPool), - Opaque(OpaqueMemory), -} - -impl Memory for MemoryWrapper { - fn reserve(&self, min_bytes: usize) -> BytesBuf { - match self { - Self::Global(pool) => pool.reserve(min_bytes), - Self::Opaque(memory) => memory.reserve(min_bytes), - } - } -} - impl AsRef for HttpBodyBuilder { fn as_ref(&self) -> &Clock { &self.clock @@ -400,24 +372,25 @@ mod tests { } #[test] - fn new_with_global_memory() { + fn new_accepts_opaque_memory_with_custom_provider() { let clock = Clock::new_frozen(); - let memory = GlobalPool::new(); + let memory = OpaqueMemory::new(TransparentMemory::new()); + let builder = HttpBodyBuilder::new(memory, &clock); - let body = builder.text("test"); - assert_eq!(body.content_length(), Some(4)); + let body = builder.text("custom pool"); - // access the clock - let _clock: &Clock = builder.as_ref(); + assert_eq!(body.content_length(), Some(11)); } #[test] - fn with_custom_memory() { + fn new_with_global_memory() { let clock = Clock::new_frozen(); - let builder = HttpBodyBuilder::with_custom_memory(TransparentMemory::new(), &clock); - let body = builder.text("hello"); - let data = BytesView::try_from(body).unwrap(); - assert_eq!(data.len(), 5); + let builder = HttpBodyBuilder::new(GlobalPool::new(), &clock); + let body = builder.text("test"); + assert_eq!(body.content_length(), Some(4)); + + // access the clock + let _clock: &Clock = builder.as_ref(); } #[test] @@ -602,7 +575,7 @@ mod tests { ); let clock = Clock::new_frozen(); - let builder = HttpBodyBuilder::with_custom_memory(TransparentMemory::new(), &clock); + let builder = HttpBodyBuilder::new(TransparentMemory::new(), &clock); let body = builder.json(&payload).unwrap(); let bytes_view = body.into_bytes_no_buffering().unwrap();