Re-enable HTTP connection pooling (#2927)

Re-enabled HTTP connection pooling. In benchmarks using `criterion`,
this results in a 3x performance improvement on repeated HTTP operations
against the same server, which is not quite the 10x performance
reduction reported in #2901, but is a significant improvement.

Author: LarryOsterman

Cargo.lock

@@ -44,6 +44,7 @@ azure_identity.workspace = true
 azure_security_keyvault_certificates.path = "../../keyvault/azure_security_keyvault_certificates"
 azure_security_keyvault_secrets.path = "../../keyvault/azure_security_keyvault_secrets"
 criterion.workspace = true
+reqwest.workspace = true
 thiserror.workspace = true
 tokio.workspace = true
 tracing-subscriber.workspace = true
@@ -87,3 +88,7 @@ features = [
 [[bench]]
 name = "benchmarks"
 harness = false
+
+[[bench]]
+name = "http_transport_benchmarks"
+harness = false

sdk/core/azure_core/Cargo.toml

@@ -435,6 +435,46 @@ Though not recommended for production, you can enable normal `core::fmt::Debug`
 cargo add azure_core -F debug
 ```
 
+### Specific issues
+
+#### Hang when invoking multiple HTTP operations using the default HTTP transport.
+
+Some customers have reported hangs when using the default `reqwest` based transport, the issue is tracked in [this GitHub issue](https://github.com/hyperium/hyper/issues/2312). The indicated workaround
+for the problem is to disable connection pooling in the `reqwest` transport.
+
+If you are encountering this issue, you can disable connection pooling by using the following function to construct an HTTP client which disables HTTP connection pooling:
+
+```rust
+# use std::sync::Arc;
+# use azure_core::http::HttpClient;
+pub fn create_reqwest_client_without_connection_pooling() -> Arc<dyn HttpClient> {
+    let client = ::reqwest::ClientBuilder::new()
+        .pool_max_idle_per_host(0)
+        .build()
+        .expect("failed to build `reqwest` client");
+
+    Arc::new(client)
+}
+```
+
+You can then set this transport in the `ClientOptions` used to configure your Azure SDK client:
+
+```rust
+# use std::sync::Arc;
+# use azure_core::http::{HttpClient, ClientOptions, TransportOptions};
+# #[derive(Default)]struct MyServiceClientOptions { client_options: ClientOptions};
+# let transport = Arc::new(reqwest::Client::new());
+let options = MyServiceClientOptions {
+    client_options: ClientOptions {
+        transport: Some(TransportOptions::new(transport)),
+        ..Default::default()
+    },
+    ..Default::default()
+};
+```
+
+Note that implementing this workaround can result in a significant (3x) performance slowdown depending on your use case.
+
 ## Contributing
 
 See the [CONTRIBUTING.md] for details on building, testing, and contributing to these libraries.

sdk/core/azure_core/README.md

@@ -60,10 +60,12 @@ fn http_transport_test(c: &mut Criterion) {
         });
     });
 }
+
 // Main benchmark configuration
 criterion_group! {
     name = benchmarks;
     config = Criterion::default();
     targets = url_parsing_benchmark, http_transport_test
 }
+
 criterion_main!(benchmarks);

sdk/core/azure_core/benches/benchmarks.rs

@@ -0,0 +1,226 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License.
+#[cfg_attr(target_os = "macos", allow(unused_imports))]
+use azure_core::{
+    credentials::TokenCredential,
+    fmt::SafeDebug,
+    http::{
+        ClientMethodOptions, ClientOptions, HttpClient, Method, Pipeline, RawResponse, Request,
+        TransportOptions, Url,
+    },
+    Result,
+};
+#[cfg_attr(target_os = "macos", allow(unused_imports))]
+use azure_identity::DeveloperToolsCredential;
+use criterion::{criterion_group, criterion_main, Criterion};
+use std::sync::Arc;
+
+#[derive(Clone, SafeDebug)]
+pub struct TestServiceClientOptions {
+    pub client_options: ClientOptions,
+    pub api_version: Option<String>,
+}
+
+impl Default for TestServiceClientOptions {
+    fn default() -> Self {
+        Self {
+            client_options: ClientOptions::default(),
+            api_version: Some("2023-10-01".to_string()),
+        }
+    }
+}
+
+pub struct TestServiceClient {
+    endpoint: Url,
+    api_version: String,
+    pipeline: Pipeline,
+}
+
+#[derive(Default, SafeDebug)]
+pub struct TestServiceClientGetMethodOptions<'a> {
+    pub method_options: ClientMethodOptions<'a>,
+}
+
+impl TestServiceClient {
+    pub fn new(
+        endpoint: &str,
+        _credential: Arc<dyn TokenCredential>,
+        options: Option<TestServiceClientOptions>,
+    ) -> Result<Self> {
+        let options = options.unwrap_or_default();
+        let mut endpoint = Url::parse(endpoint)?;
+        if !endpoint.scheme().starts_with("http") {
+            return Err(azure_core::Error::message(
+                azure_core::error::ErrorKind::Other,
+                format!("{endpoint} must use http(s)"),
+            ));
+        }
+        endpoint.set_query(None);
+
+        Ok(Self {
+            endpoint,
+            api_version: options.api_version.unwrap_or_default(),
+            pipeline: Pipeline::new(
+                option_env!("CARGO_PKG_NAME"),
+                option_env!("CARGO_PKG_VERSION"),
+                options.client_options,
+                Vec::default(),
+                Vec::default(),
+            ),
+        })
+    }
+
+    /// Returns the Url associated with this client.
+    pub fn endpoint(&self) -> &Url {
+        &self.endpoint
+    }
+
+    /// Returns the result of a Get verb against the configured endpoint with the specified path.
+    ///
+    /// This method demonstrates a service client which does not have per-method spans but which will create
+    /// HTTP client spans if the `InstrumentationOptions` are configured in the client options.
+    ///
+    pub async fn get(
+        &self,
+        path: &str,
+        options: Option<TestServiceClientGetMethodOptions<'_>>,
+    ) -> Result<RawResponse> {
+        let options = options.unwrap_or_default();
+        let mut url = self.endpoint.clone();
+        url.set_path(path);
+        url.query_pairs_mut()
+            .append_pair("api-version", &self.api_version);
+
+        let mut request = Request::new(url, azure_core::http::Method::Get);
+
+        let response = self
+            .pipeline
+            .send(&options.method_options.context, &mut request)
+            .await?;
+        if !response.status().is_success() {
+            return Err(azure_core::Error::message(
+                azure_core::error::ErrorKind::HttpResponse {
+                    status: response.status(),
+                    error_code: None,
+                },
+                format!("Failed to GET {}: {}", request.url(), response.status()),
+            ));
+        }
+        Ok(response)
+    }
+}
+
+pub fn new_reqwest_client_disable_connection_pool() -> Arc<dyn HttpClient> {
+    let client = ::reqwest::ClientBuilder::new()
+        .pool_max_idle_per_host(0)
+        .build()
+        .expect("failed to build `reqwest` client");
+
+    Arc::new(client)
+}
+
+pub fn new_default_reqwest_client() -> Arc<dyn HttpClient> {
+    let client = ::reqwest::Client::new();
+
+    Arc::new(client)
+}
+
+#[cfg_attr(target_os = "macos", allow(unused_variables))]
+pub fn simple_http_transport_test(c: &mut Criterion) {
+    #[cfg(target_os = "macos")]
+    return;
+
+    #[cfg(not(target_os = "macos"))]
+    {
+        let rt = tokio::runtime::Runtime::new().unwrap();
+
+        let endpoint = "https://azuresdkforcpp.azurewebsites.net";
+        let credential = DeveloperToolsCredential::new(None).unwrap();
+        let options = TestServiceClientOptions::default();
+
+        let client = TestServiceClient::new(endpoint, credential, Some(options)).unwrap();
+
+        // Benchmark GET and POST requests
+        c.bench_function("default_http_pipeline_test", |b| {
+            b.to_async(&rt).iter(|| async {
+                let response = client.get("get", None).await;
+                assert!(response.is_ok());
+                let response = response.unwrap();
+                assert_eq!(response.status(), azure_core::http::StatusCode::Ok);
+            });
+        });
+    }
+}
+
+#[cfg_attr(target_os = "macos", allow(unused_variables))]
+pub fn disable_pooling_http_transport_test(c: &mut Criterion) {
+    #[cfg(target_os = "macos")]
+    return;
+
+    #[cfg(not(target_os = "macos"))]
+    {
+        let rt = tokio::runtime::Runtime::new().unwrap();
+
+        let endpoint = "https://azuresdkforcpp.azurewebsites.net";
+        let credential = DeveloperToolsCredential::new(None).unwrap();
+        let transport = new_reqwest_client_disable_connection_pool();
+        let options = TestServiceClientOptions {
+            client_options: ClientOptions {
+                transport: Some(TransportOptions::new(transport)),
+                ..Default::default()
+            },
+            ..Default::default()
+        };
+
+        let client = TestServiceClient::new(endpoint, credential, Some(options)).unwrap();
+
+        // Benchmark GET and POST requests
+        c.bench_function("disable_pooling_http_pipeline_test", |b| {
+            b.to_async(&rt).iter(|| async {
+                let response = client.get("get", None).await;
+                assert!(response.is_ok());
+                let response = response.unwrap();
+                assert_eq!(response.status(), azure_core::http::StatusCode::Ok);
+            });
+        });
+    }
+}
+
+#[cfg_attr(target_os = "macos", allow(unused_variables))]
+pub fn baseline_http_transport_test(c: &mut Criterion) {
+    #[cfg(target_os = "macos")]
+    return;
+
+    #[cfg(not(target_os = "macos"))]
+    {
+        let rt = tokio::runtime::Runtime::new().unwrap();
+        let endpoint = "https://azuresdkforcpp.azurewebsites.net";
+
+        let http_client = new_default_reqwest_client();
+
+        // Benchmark GET and POST requests
+        c.bench_function("baseline_http_pipeline_test", |b| {
+            b.to_async(&rt).iter(|| async {
+                let request = Request::new(
+                    Url::parse(&format!("{}/get", endpoint)).unwrap(),
+                    Method::Get,
+                );
+                let response = http_client.execute_request(&request).await;
+                assert!(response.is_ok());
+                let response = response.unwrap();
+                assert_eq!(response.status(), azure_core::http::StatusCode::Ok);
+            });
+        });
+    }
+}
+
+// Main benchmark configuration
+criterion_group!(name=http_transport_benchmarks;
+    config=Criterion::default()
+        .sample_size(100)
+        .warm_up_time(std::time::Duration::new(10, 0))
+        .measurement_time(std::time::Duration::new(50, 0));
+    targets=simple_http_transport_test, disable_pooling_http_transport_test, baseline_http_transport_test
+);
+
+criterion_main!(http_transport_benchmarks);

sdk/core/azure_core/benches/http_transport_benchmarks.rs

@@ -29,6 +29,7 @@ azure_identity.workspace = true
 azure_security_keyvault_test = { path = "../azure_security_keyvault_test" }
 criterion.workspace = true
 rand.workspace = true
+reqwest.workspace = true
 sha2.workspace = true
 tokio.workspace = true