diff --git a/src/libraries/System.Net.Http/src/System/Net/Http/Metrics/MetricsHandler.cs b/src/libraries/System.Net.Http/src/System/Net/Http/Metrics/MetricsHandler.cs index 4e6f6a05e69ce2..9b53197bb735d2 100644 --- a/src/libraries/System.Net.Http/src/System/Net/Http/Metrics/MetricsHandler.cs +++ b/src/libraries/System.Net.Http/src/System/Net/Http/Metrics/MetricsHandler.cs @@ -1,6 +1,7 @@ // Licensed to the .NET Foundation under one or more agreements. // The .NET Foundation licenses this file to you under the MIT license. +using System.Collections.Concurrent; using System.Collections.Generic; using System.Diagnostics; using System.Diagnostics.CodeAnalysis; @@ -10,10 +11,125 @@ namespace System.Net.Http.Metrics { + /// + /// Represents a unique combination of tags for tracking active requests. + /// + internal readonly struct ActiveRequestsTagKey : IEquatable + { + public readonly string? Scheme; + public readonly string? Host; + public readonly int Port; + public readonly string Method; + private readonly int _hashCode; + + public ActiveRequestsTagKey(string? scheme, string? host, int port, string method) + { + Scheme = scheme; + Host = host; + Port = port; + Method = method; + _hashCode = HashCode.Combine(scheme, host, port, method); + } + + public bool Equals(ActiveRequestsTagKey other) => + Scheme == other.Scheme && + Host == other.Host && + Port == other.Port && + Method == other.Method; + + public override bool Equals(object? obj) => obj is ActiveRequestsTagKey other && Equals(other); + + public override int GetHashCode() => _hashCode; + + public TagList ToTagList() + { + TagList tags = default; + if (Scheme is not null) + { + tags.Add("url.scheme", Scheme); + tags.Add("server.address", Host); + tags.Add("server.port", DiagnosticsHelper.GetBoxedInt32(Port)); + } + tags.Add("http.request.method", Method); + return tags; + } + + public override string ToString() => + $"{Method}{(Scheme is not null ? $" {Scheme}://{Host}:{Port}" : "")}"; + } + + /// + /// Thread-safe tracker for active request counts by tag combination. + /// + internal sealed class ActiveRequestsTracker + { + private readonly ConcurrentDictionary _counts = new(); + + /// + /// Increments the count for the specified tag combination. + /// + public void Increment(in ActiveRequestsTagKey key) + { + _counts.AddOrUpdate(key, 1, static (_, currentValue) => currentValue + 1); + } + + /// + /// Decrements the count for the specified tag combination. + /// Removes the entry if the count reaches zero. + /// + public void Decrement(in ActiveRequestsTagKey key) + { + // We need to atomically decrement and remove if zero. + // Use a spin loop with TryGetValue/TryUpdate/TryRemove to handle this safely. + while (true) + { + if (!_counts.TryGetValue(key, out long currentValue)) + { + // Key doesn't exist, nothing to decrement. + // This shouldn't happen in normal operation but we handle it gracefully. + Debug.Fail($"Decrement for non-existing request {key}"); + return; + } + + if (currentValue <= 1) + { + // Try to remove the entry since it will become zero. + // Use the overload that checks the current value to ensure atomicity. + if (_counts.TryRemove(new KeyValuePair(key, currentValue))) + { + return; + } + // Another thread modified the value, retry. + } + else + { + // Try to decrement the value. + if (_counts.TryUpdate(key, currentValue - 1, currentValue)) + { + return; + } + // Another thread modified the value, retry. + } + } + } + + /// + /// Returns measurements for all tag combinations with non-zero counts. + /// + public IEnumerable> GetMeasurements() + { + foreach (KeyValuePair entry in _counts) + { + yield return new Measurement(entry.Value, entry.Key.ToTagList()); + } + } + } + internal sealed class MetricsHandler : HttpMessageHandlerStage { private readonly HttpMessageHandler _innerHandler; - private readonly UpDownCounter _activeRequests; + private readonly ActiveRequestsTracker _activeRequestsTracker = new(); + private readonly ObservableUpDownCounter _activeRequests; private readonly Histogram _requestsDuration; private readonly IWebProxy? _proxy; @@ -27,8 +143,9 @@ public MetricsHandler(HttpMessageHandler innerHandler, IMeterFactory? meterFacto meter = meterFactory?.Create("System.Net.Http") ?? SharedMeter.Instance; // Meter has a cache for the instruments it owns - _activeRequests = meter.CreateUpDownCounter( + _activeRequests = meter.CreateObservableUpDownCounter( "http.client.active_requests", + observeValues: _activeRequestsTracker.GetMeasurements, unit: "{request}", description: "Number of outbound HTTP requests that are currently active on the client."); _requestsDuration = meter.CreateHistogram( @@ -56,7 +173,7 @@ private async ValueTask SendAsyncWithMetrics(HttpRequestMes { Debug.Assert(GlobalHttpSettings.MetricsHandler.IsGloballyEnabled); - (long startTimestamp, bool recordCurrentRequests) = RequestStart(request); + (long startTimestamp, bool recordCurrentRequests, ActiveRequestsTagKey requestTagKey) = RequestStart(request); HttpResponseMessage? response = null; Exception? exception = null; try @@ -73,7 +190,7 @@ await _innerHandler.SendAsync(request, cancellationToken).ConfigureAwait(false) } finally { - RequestStop(request, response, exception, startTimestamp, recordCurrentRequests); + RequestStop(request, response, exception, startTimestamp, recordCurrentRequests, requestTagKey); } } @@ -87,27 +204,25 @@ protected override void Dispose(bool disposing) base.Dispose(disposing); } - private (long StartTimestamp, bool RecordCurrentRequests) RequestStart(HttpRequestMessage request) + private (long StartTimestamp, bool RecordCurrentRequests, ActiveRequestsTagKey RequestTagKey) RequestStart(HttpRequestMessage request) { bool recordCurrentRequests = _activeRequests.Enabled; long startTimestamp = Stopwatch.GetTimestamp(); + ActiveRequestsTagKey requestTagKey = CreateActiveRequestsTagKey(request); if (recordCurrentRequests) { - TagList tags = InitializeCommonTags(request); - _activeRequests.Add(1, tags); + _activeRequestsTracker.Increment(requestTagKey); } - return (startTimestamp, recordCurrentRequests); + return (startTimestamp, recordCurrentRequests, requestTagKey); } - private void RequestStop(HttpRequestMessage request, HttpResponseMessage? response, Exception? exception, long startTimestamp, bool recordCurrentRequests) + private void RequestStop(HttpRequestMessage request, HttpResponseMessage? response, Exception? exception, long startTimestamp, bool recordCurrentRequests, ActiveRequestsTagKey requestTagKey) { - TagList tags = InitializeCommonTags(request); - if (recordCurrentRequests) { - _activeRequests.Add(-1, tags); + _activeRequestsTracker.Decrement(requestTagKey); } if (!_requestsDuration.Enabled) @@ -115,6 +230,7 @@ private void RequestStop(HttpRequestMessage request, HttpResponseMessage? respon return; } + TagList tags = requestTagKey.ToTagList(); if (response is not null) { tags.Add("http.response.status_code", DiagnosticsHelper.GetBoxedInt32((int)response.StatusCode)); @@ -139,19 +255,22 @@ private void RequestStop(HttpRequestMessage request, HttpResponseMessage? respon } } - private TagList InitializeCommonTags(HttpRequestMessage request) + private ActiveRequestsTagKey CreateActiveRequestsTagKey(HttpRequestMessage request) { - TagList tags = default; + string? scheme = null; + string? host = null; + int port = 0; if (request.RequestUri is Uri requestUri && requestUri.IsAbsoluteUri) { - tags.Add("url.scheme", requestUri.Scheme); - tags.Add("server.address", DiagnosticsHelper.GetServerAddress(request, _proxy)); - tags.Add("server.port", DiagnosticsHelper.GetBoxedInt32(requestUri.Port)); + scheme = requestUri.Scheme; + host = DiagnosticsHelper.GetServerAddress(request, _proxy); + port = requestUri.Port; } - tags.Add(DiagnosticsHelper.GetMethodTag(request.Method, out _)); - return tags; + string method = (string)DiagnosticsHelper.GetMethodTag(request.Method, out _).Value!; + + return new ActiveRequestsTagKey(scheme, host, port, method); } private sealed class SharedMeter : Meter diff --git a/src/libraries/System.Net.Http/src/System/Net/Http/SocketsHttpHandler/Http3Connection.cs b/src/libraries/System.Net.Http/src/System/Net/Http/SocketsHttpHandler/Http3Connection.cs index ebe52071a870bd..0610fa8a49b8d9 100644 --- a/src/libraries/System.Net.Http/src/System/Net/Http/SocketsHttpHandler/Http3Connection.cs +++ b/src/libraries/System.Net.Http/src/System/Net/Http/SocketsHttpHandler/Http3Connection.cs @@ -457,14 +457,14 @@ public void RemoveStream(QuicStream stream) { if (_activeRequests.Remove(stream)) { - if (ShuttingDown) + if (_activeRequests.Count == 0) { - CheckForShutdown(); + MarkConnectionAsIdle(); } - if (_activeRequests.Count == 0) + if (ShuttingDown) { - MarkConnectionAsIdle(); + CheckForShutdown(); } } } diff --git a/src/libraries/System.Net.Http/src/System/Net/Http/SocketsHttpHandler/Metrics/ConnectionMetrics.cs b/src/libraries/System.Net.Http/src/System/Net/Http/SocketsHttpHandler/Metrics/ConnectionMetrics.cs index 160592493caa85..d8022e621ae147 100644 --- a/src/libraries/System.Net.Http/src/System/Net/Http/SocketsHttpHandler/Metrics/ConnectionMetrics.cs +++ b/src/libraries/System.Net.Http/src/System/Net/Http/SocketsHttpHandler/Metrics/ConnectionMetrics.cs @@ -10,11 +10,11 @@ internal sealed class ConnectionMetrics { private readonly SocketsHttpHandlerMetrics _metrics; private readonly bool _openConnectionsEnabled; - private readonly object _protocolVersionTag; - private readonly object _schemeTag; - private readonly object _hostTag; - private readonly object _portTag; - private readonly object? _peerAddressTag; + private readonly string _protocolVersionTag; + private readonly string _schemeTag; + private readonly string _hostTag; + private readonly int _portTag; + private readonly string? _peerAddressTag; private bool _currentlyIdle; public ConnectionMetrics(SocketsHttpHandlerMetrics metrics, string protocolVersion, string scheme, string host, int port, string? peerAddress) @@ -24,7 +24,7 @@ public ConnectionMetrics(SocketsHttpHandlerMetrics metrics, string protocolVersi _protocolVersionTag = protocolVersion; _schemeTag = scheme; _hostTag = host; - _portTag = DiagnosticsHelper.GetBoxedInt32(port); + _portTag = port; _peerAddressTag = peerAddress; } @@ -36,7 +36,7 @@ private TagList GetTags() tags.Add("network.protocol.version", _protocolVersionTag); tags.Add("url.scheme", _schemeTag); tags.Add("server.address", _hostTag); - tags.Add("server.port", _portTag); + tags.Add("server.port", DiagnosticsHelper.GetBoxedInt32(_portTag)); if (_peerAddressTag is not null) { @@ -46,32 +46,34 @@ private TagList GetTags() return tags; } - private static KeyValuePair GetStateTag(bool idle) => new KeyValuePair("http.connection.state", idle ? "idle" : "active"); + private OpenConnectionsTagKey CreateTagKey(bool idle) => + new OpenConnectionsTagKey(_protocolVersionTag, _schemeTag, _hostTag, _portTag, idle, _peerAddressTag); public void ConnectionEstablished() { if (_openConnectionsEnabled) { - _currentlyIdle = true; - TagList tags = GetTags(); - tags.Add(GetStateTag(idle: true)); - _metrics.OpenConnections.Add(1, tags); + lock (this) + { + _currentlyIdle = true; + _metrics.OpenConnectionsTracker.Increment(CreateTagKey(idle: true)); + } } } public void ConnectionClosed(long durationMs) { - TagList tags = GetTags(); - if (_metrics.ConnectionDuration.Enabled) { - _metrics.ConnectionDuration.Record(durationMs / 1000d, tags); + _metrics.ConnectionDuration.Record(durationMs / 1000d, GetTags()); } if (_openConnectionsEnabled) { - tags.Add(GetStateTag(idle: _currentlyIdle)); - _metrics.OpenConnections.Add(-1, tags); + lock (this) + { + _metrics.OpenConnectionsTracker.Decrement(CreateTagKey(idle: _currentlyIdle)); + } } } @@ -79,12 +81,12 @@ public void IdleStateChanged(bool idle) { if (_openConnectionsEnabled && _currentlyIdle != idle) { - _currentlyIdle = idle; - TagList tags = GetTags(); - tags.Add(GetStateTag(idle: !idle)); - _metrics.OpenConnections.Add(-1, tags); - tags[tags.Count - 1] = GetStateTag(idle: idle); - _metrics.OpenConnections.Add(1, tags); + lock (this) + { + _currentlyIdle = idle; + _metrics.OpenConnectionsTracker.Decrement(CreateTagKey(idle: !idle)); + _metrics.OpenConnectionsTracker.Increment(CreateTagKey(idle: idle)); + } } } } diff --git a/src/libraries/System.Net.Http/src/System/Net/Http/SocketsHttpHandler/Metrics/SocketsHttpHandlerMetrics.cs b/src/libraries/System.Net.Http/src/System/Net/Http/SocketsHttpHandler/Metrics/SocketsHttpHandlerMetrics.cs index eabd5db88de69f..7d38f6fca17d49 100644 --- a/src/libraries/System.Net.Http/src/System/Net/Http/SocketsHttpHandler/Metrics/SocketsHttpHandlerMetrics.cs +++ b/src/libraries/System.Net.Http/src/System/Net/Http/SocketsHttpHandler/Metrics/SocketsHttpHandlerMetrics.cs @@ -1,33 +1,169 @@ // Licensed to the .NET Foundation under one or more agreements. // The .NET Foundation licenses this file to you under the MIT license. +using System.Collections.Concurrent; +using System.Collections.Generic; using System.Diagnostics; using System.Diagnostics.Metrics; namespace System.Net.Http.Metrics { - internal sealed class SocketsHttpHandlerMetrics(Meter meter) + /// + /// Represents a unique combination of tags for tracking open connections. + /// + internal readonly struct OpenConnectionsTagKey : IEquatable { - public readonly UpDownCounter OpenConnections = meter.CreateUpDownCounter( - name: "http.client.open_connections", - unit: "{connection}", - description: "Number of outbound HTTP connections that are currently active or idle on the client."); - - public readonly Histogram ConnectionDuration = meter.CreateHistogram( - name: "http.client.connection.duration", - unit: "s", - description: "The duration of successfully established outbound HTTP connections.", - advice: new InstrumentAdvice() + public readonly string ProtocolVersion; + public readonly string Scheme; + public readonly string Host; + public readonly int Port; + public readonly bool IsIdle; + public readonly string? PeerAddress; + private readonly int _hashCode; + + public OpenConnectionsTagKey(string protocolVersion, string scheme, string host, int port, bool isIdle, string? peerAddress) + { + ProtocolVersion = protocolVersion; + Scheme = scheme; + Host = host; + Port = port; + IsIdle = isIdle; + PeerAddress = peerAddress; + _hashCode = HashCode.Combine(protocolVersion, scheme, host, port, isIdle, peerAddress); + } + + public bool Equals(OpenConnectionsTagKey other) => + ProtocolVersion == other.ProtocolVersion && + Scheme == other.Scheme && + Host == other.Host && + Port == other.Port && + IsIdle == other.IsIdle && + PeerAddress == other.PeerAddress; + + public override bool Equals(object? obj) => obj is OpenConnectionsTagKey other && Equals(other); + + public override int GetHashCode() => _hashCode; + + public TagList ToTagList() + { + TagList tags = default; + tags.Add("network.protocol.version", ProtocolVersion); + tags.Add("url.scheme", Scheme); + tags.Add("server.address", Host); + tags.Add("server.port", DiagnosticsHelper.GetBoxedInt32(Port)); + tags.Add("http.connection.state", IsIdle ? "idle" : "active"); + if (PeerAddress is not null) + { + tags.Add("network.peer.address", PeerAddress); + } + return tags; + } + + public override string ToString() => + $"HTTP/{ProtocolVersion} {Scheme}://{Host}:{Port} {(IsIdle ? "idle" : "active")}{(PeerAddress is not null ? $" {PeerAddress}" : "")}"; + } + + /// + /// Thread-safe tracker for open connection counts by tag combination. + /// + internal sealed class OpenConnectionsTracker + { + private readonly ConcurrentDictionary _counts = new(); + + /// + /// Increments the count for the specified tag combination. + /// + public void Increment(in OpenConnectionsTagKey key) + { + _counts.AddOrUpdate(key, 1, static (_, currentValue) => currentValue + 1); + } + + /// + /// Decrements the count for the specified tag combination. + /// Removes the entry if the count reaches zero. + /// + public void Decrement(in OpenConnectionsTagKey key) + { + // We need to atomically decrement and remove if zero. + // Use a spin loop with TryGetValue/TryUpdate/TryRemove to handle this safely. + while (true) { - // These values are not based on a standard and may change in the future. - HistogramBucketBoundaries = [0.01, 0.02, 0.05, 0.1, 0.2, 0.5, 1, 2, 5, 10, 30, 60, 120, 300] - }); - - public readonly Histogram RequestsQueueDuration = meter.CreateHistogram( - name: "http.client.request.time_in_queue", - unit: "s", - description: "The amount of time requests spent on a queue waiting for an available connection.", - advice: DiagnosticsHelper.ShortHistogramAdvice); + if (!_counts.TryGetValue(key, out long currentValue)) + { + // Key doesn't exist, nothing to decrement. + // This shouldn't happen in normal operation but we handle it gracefully. + Debug.Fail($"Decrement for non-existing connection {key}"); + return; + } + + if (currentValue <= 1) + { + // Try to remove the entry since it will become zero. + // Use the overload that checks the current value to ensure atomicity. + if (_counts.TryRemove(new KeyValuePair(key, currentValue))) + { + return; + } + // Another thread modified the value, retry. + } + else + { + // Try to decrement the value. + if (_counts.TryUpdate(key, currentValue - 1, currentValue)) + { + return; + } + // Another thread modified the value, retry. + } + } + } + + /// + /// Returns measurements for all tag combinations with non-zero counts. + /// + public IEnumerable> GetMeasurements() + { + foreach (KeyValuePair entry in _counts) + { + yield return new Measurement(entry.Value, entry.Key.ToTagList()); + } + } + } + + internal sealed class SocketsHttpHandlerMetrics + { + public readonly OpenConnectionsTracker OpenConnectionsTracker = new(); + + public readonly ObservableUpDownCounter OpenConnections; + + public readonly Histogram ConnectionDuration; + + public readonly Histogram RequestsQueueDuration; + + public SocketsHttpHandlerMetrics(Meter meter) + { + OpenConnections = meter.CreateObservableUpDownCounter( + name: "http.client.open_connections", + observeValues: OpenConnectionsTracker.GetMeasurements, + unit: "{connection}", + description: "Number of outbound HTTP connections that are currently active or idle on the client."); + + ConnectionDuration = meter.CreateHistogram( + name: "http.client.connection.duration", + unit: "s", + description: "The duration of successfully established outbound HTTP connections.", + advice: new InstrumentAdvice() + { + // These values are not based on a standard and may change in the future. + HistogramBucketBoundaries = [0.01, 0.02, 0.05, 0.1, 0.2, 0.5, 1, 2, 5, 10, 30, 60, 120, 300] + }); + + RequestsQueueDuration = meter.CreateHistogram( + name: "http.client.request.time_in_queue", + unit: "s", + description: "The amount of time requests spent on a queue waiting for an available connection.", + advice: DiagnosticsHelper.ShortHistogramAdvice); + } public void RequestLeftQueue(HttpRequestMessage request, HttpConnectionPool pool, TimeSpan duration, int versionMajor) { diff --git a/src/libraries/System.Net.Http/tests/FunctionalTests/MetricsTest.cs b/src/libraries/System.Net.Http/tests/FunctionalTests/MetricsTest.cs index dd830324d6723a..5ba543931fcd45 100644 --- a/src/libraries/System.Net.Http/tests/FunctionalTests/MetricsTest.cs +++ b/src/libraries/System.Net.Http/tests/FunctionalTests/MetricsTest.cs @@ -221,6 +221,7 @@ private void OnMeasurementRecorded(Instrument instrument, T measurement, ReadOnl } } + public void RecordObservableInstruments() => _meterListener.RecordObservableInstruments(); public IReadOnlyList> GetMeasurements() => _values.ToArray(); public void Dispose() => _meterListener.Dispose(); } @@ -272,6 +273,7 @@ private MultiInstrumentRecorder(Meter? meter) _meterListener.Start(); } + public void RecordObservableInstruments() => _meterListener.RecordObservableInstruments(); public IReadOnlyList GetMeasurements() => _values.ToArray(); public void Dispose() => _meterListener.Dispose(); } @@ -292,21 +294,33 @@ public HttpMetricsTest(ITestOutputHelper output) : base(output) [ActiveIssue("https://github.com/dotnet/runtime/issues/129223", TestPlatforms.Wasi)] public Task ActiveRequests_Success_Recorded() { + var serverTcs = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var clientTcs = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + return LoopbackServerFactory.CreateClientAndServerAsync(async uri => { using HttpMessageInvoker client = CreateHttpMessageInvoker(); using InstrumentRecorder recorder = SetupInstrumentRecorder(InstrumentNames.ActiveRequests); using HttpRequestMessage request = new(HttpMethod.Get, uri) { Version = UseVersion }; - HttpResponseMessage response = await SendAsync(client, request); + var requestTask = SendAsync(client, request); + await serverTcs.Task.WaitAsync(TestHelper.PassingTestTimeout); + recorder.RecordObservableInstruments(); + clientTcs.SetResult(); + var response = await requestTask; response.Dispose(); // Make sure disposal doesn't interfere with recording by enforcing early disposal. Assert.Collection(recorder.GetMeasurements(), - m => VerifyActiveRequests(m, 1, uri), - m => VerifyActiveRequests(m, -1, uri)); + m => VerifyActiveRequests(m, 1, uri)); }, async server => { - await server.AcceptConnectionSendResponseAndCloseAsync(); + await server.AcceptConnectionAsync(async connection => + { + await connection.ReadRequestDataAsync(); + serverTcs.SetResult(); + await clientTcs.Task.WaitAsync(TestHelper.PassingTestTimeout); + await connection.SendResponseAsync(); + }); }); } @@ -393,9 +407,9 @@ public async Task ExternalServer_DurationMetrics_Recorded() using (HttpMessageInvoker client = CreateHttpMessageInvoker()) { using HttpRequestMessage request = new(HttpMethod.Get, uri) { Version = UseVersion }; - request.Headers.ConnectionClose = true; using HttpResponseMessage response = await SendAsync(client, request); await response.Content.LoadIntoBufferAsync(); + openConnectionsRecorder.RecordObservableInstruments(); await WaitForEnvironmentTicksToAdvance(); } @@ -673,6 +687,11 @@ public Task ActiveRequests_Redirect_RecordedForEachHttpSpan(int credentialsMode) Handler.Credentials = credentialsMode == 1 ? new CredentialCache() : new CustomCredentials(); } + var originalServerTcs = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var redirectServerTcs = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var clientTcs1 = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var clientTcs2 = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + return LoopbackServerFactory.CreateServerAsync((originalServer, originalUri) => { return LoopbackServerFactory.CreateServerAsync(async (redirectServer, redirectUri) => @@ -682,21 +701,36 @@ public Task ActiveRequests_Redirect_RecordedForEachHttpSpan(int credentialsMode) using HttpRequestMessage request = new(HttpMethod.Get, originalUri) { Version = UseVersion }; Task clientTask = SendAsync(client, request); - Task serverTask = originalServer.HandleRequestAsync(HttpStatusCode.Redirect, new[] { new HttpHeaderData("Location", redirectUri.AbsoluteUri) }); - + var serverTask = originalServer.AcceptConnectionAsync(async connection => + { + var requestData = await connection.ReadRequestDataAsync(); + originalServerTcs.SetResult(); + await clientTcs1.Task.WaitAsync(TestHelper.PassingTestTimeout); + await connection.SendResponseAsync(HttpStatusCode.Redirect, new[] { new HttpHeaderData("Location", redirectUri.AbsoluteUri) }); + }); + await originalServerTcs.Task.WaitAsync(TestHelper.PassingTestTimeout); + recorder.RecordObservableInstruments(); + clientTcs1.SetResult(); await Task.WhenAny(clientTask, serverTask); Assert.False(clientTask.IsCompleted, $"{clientTask.Status}: {clientTask.Exception}"); await serverTask; - serverTask = redirectServer.HandleRequestAsync(); + serverTask = redirectServer.AcceptConnectionAsync(async connection => + { + await connection.ReadRequestDataAsync(); + redirectServerTcs.SetResult(); + await clientTcs2.Task.WaitAsync(TestHelper.PassingTestTimeout); + await connection.SendResponseAsync(); + }); + await redirectServerTcs.Task.WaitAsync(TestHelper.PassingTestTimeout); + recorder.RecordObservableInstruments(); + clientTcs2.SetResult(); await TestHelper.WhenAllCompletedOrAnyFailed(clientTask, serverTask); await clientTask; Assert.Collection(recorder.GetMeasurements(), m => VerifyActiveRequests(m, 1, originalUri), - m => VerifyActiveRequests(m, -1, originalUri), - m => VerifyActiveRequests(m, 1, redirectUri), - m => VerifyActiveRequests(m, -1, redirectUri)); + m => VerifyActiveRequests(m, 1, redirectUri)); }); }); } @@ -751,7 +785,8 @@ await LoopbackServerFactory.CreateClientAndServerAsync(async uri => [ConditionalFact(typeof(SocketsHttpHandler), nameof(SocketsHttpHandler.IsSupported))] public async Task AllSocketsHttpHandlerCounters_Success_Recorded() { - TaskCompletionSource clientWaitingTcs = new(TaskCreationOptions.RunContinuationsAsynchronously); + TaskCompletionSource clientTcs = new(TaskCreationOptions.RunContinuationsAsynchronously); + TaskCompletionSource serverTcs = new(TaskCreationOptions.RunContinuationsAsynchronously); TaskCompletionSource clientDisposedTcs = new(TaskCreationOptions.RunContinuationsAsynchronously); await LoopbackServerFactory.CreateClientAndServerAsync(async uri => @@ -764,61 +799,33 @@ await LoopbackServerFactory.CreateClientAndServerAsync(async uri => using HttpRequestMessage request = new(HttpMethod.Get, uri) { Version = UseVersion }; Task sendAsyncTask = SendAsync(invoker, request); - clientWaitingTcs.SetResult(); - using HttpResponseMessage response = await sendAsyncTask; + await serverTcs.Task.WaitAsync(TestHelper.PassingTestTimeout); + recorder.RecordObservableInstruments(); + clientTcs.SetResult(); + HttpResponseMessage response = await sendAsyncTask; + response.Dispose(); await WaitForEnvironmentTicksToAdvance(); + recorder.RecordObservableInstruments(); } - clientDisposedTcs.SetResult(); - Action requestsQueueDuration = m => - VerifyTimeInQueue(m.InstrumentName, m.Value, m.Tags, uri, UseVersion); - Action connectionNoLongerIdle = m => - VerifyOpenConnections(m.InstrumentName, m.Value, m.Tags, -1, uri, UseVersion, "idle"); - Action connectionIsActive = m => - VerifyOpenConnections(m.InstrumentName, m.Value, m.Tags, 1, uri, UseVersion, "active"); - - Action check1 = requestsQueueDuration; - Action check2 = connectionNoLongerIdle; - Action check3 = connectionIsActive; - - if (UseVersion.Major > 2) - { - // With HTTP/3, the idle state change is emitted before RequestsQueueDuration. - check1 = connectionNoLongerIdle; - check2 = connectionIsActive; - check3 = requestsQueueDuration; - } - - IReadOnlyList measurements = recorder.GetMeasurements(); - foreach (RecordedCounter m in measurements) - { - _output.WriteLine(m.ToString()); - } - - Assert.Collection(measurements, + Assert.Collection(recorder.GetMeasurements(), + m => VerifyTimeInQueue(m.InstrumentName, m.Value, m.Tags, uri, UseVersion), m => VerifyActiveRequests(m.InstrumentName, (long)m.Value, m.Tags, 1, uri), - m => VerifyOpenConnections(m.InstrumentName, m.Value, m.Tags, 1, uri, UseVersion, "idle"), - check1, // requestsQueueDuration, connectionNoLongerIdle, connectionIsActive in the appropriate order. - check2, - check3, - m => VerifyOpenConnections(m.InstrumentName, m.Value, m.Tags, -1, uri, UseVersion, "active"), - m => VerifyOpenConnections(m.InstrumentName, m.Value, m.Tags, 1, uri, UseVersion, "idle"), - - m => VerifyActiveRequests(m.InstrumentName, (long)m.Value, m.Tags, -1, uri), + m => VerifyOpenConnections(m.InstrumentName, m.Value, m.Tags, 1, uri, UseVersion, "active"), m => VerifyRequestDuration(m.InstrumentName, (double)m.Value, m.Tags, uri, UseVersion, 200), - m => VerifyConnectionDuration(m.InstrumentName, m.Value, m.Tags, uri, UseVersion), - m => VerifyOpenConnections(m.InstrumentName, m.Value, m.Tags, -1, uri, UseVersion, "idle")); + m => VerifyOpenConnections(m.InstrumentName, m.Value, m.Tags, 1, uri, UseVersion, "idle"), + m => VerifyConnectionDuration(m.InstrumentName, m.Value, m.Tags, uri, UseVersion)); }, async server => { - await clientWaitingTcs.Task.WaitAsync(TestHelper.PassingTestTimeout); - await server.AcceptConnectionAsync(async connection => { await connection.ReadRequestDataAsync(); - await connection.SendResponseAsync(); + serverTcs.SetResult(); + await clientTcs.Task.WaitAsync(TestHelper.PassingTestTimeout); + await connection.SendResponseAsync(isFinal: true); await clientDisposedTcs.Task.WaitAsync(TestHelper.PassingTestTimeout); }); }); @@ -1500,6 +1507,9 @@ public async Task ActiveRequests_Success_Recorded() { await RemoteExecutor.Invoke(static async Task () => { + var serverTcs = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var clientTcs = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + using HttpMetricsTest_DefaultMeter test = new(null); await test.LoopbackServerFactory.CreateClientAndServerAsync(async uri => { @@ -1507,15 +1517,24 @@ await test.LoopbackServerFactory.CreateClientAndServerAsync(async uri => using InstrumentRecorder recorder = new InstrumentRecorder(InstrumentNames.ActiveRequests); using HttpRequestMessage request = new(HttpMethod.Get, uri) { Version = test.UseVersion }; - HttpResponseMessage response = await client.SendAsync(request); + var requestTask = client.SendAsync(request); + await serverTcs.Task.WaitAsync(TestHelper.PassingTestTimeout); + recorder.RecordObservableInstruments(); + clientTcs.SetResult(); + var response = await requestTask; response.Dispose(); // Make sure disposal doesn't interfere with recording by enforcing early disposal. Assert.Collection(recorder.GetMeasurements(), - m => VerifyActiveRequests(m, 1, uri), - m => VerifyActiveRequests(m, -1, uri)); + m => VerifyActiveRequests(m, 1, uri)); }, async server => { - await server.AcceptConnectionSendResponseAndCloseAsync(); + await server.AcceptConnectionAsync(async connection => + { + var requestData = await connection.ReadRequestDataAsync(); + serverTcs.SetResult(); + await clientTcs.Task.WaitAsync(TestHelper.PassingTestTimeout); + await connection.SendResponseAsync(); + }); }); }).DisposeAsync(); } @@ -1527,7 +1546,9 @@ public async Task AllSocketsHttpHandlerCounters_Success_Recorded() { await RemoteExecutor.Invoke(static async Task () => { - TaskCompletionSource clientWaitingTcs = new(TaskCreationOptions.RunContinuationsAsynchronously); + TaskCompletionSource clientTcs = new(TaskCreationOptions.RunContinuationsAsynchronously); + TaskCompletionSource serverTcs = new(TaskCreationOptions.RunContinuationsAsynchronously); + TaskCompletionSource clientDisposedTcs = new(TaskCreationOptions.RunContinuationsAsynchronously); using HttpMetricsTest_DefaultMeter test = new(null); await test.LoopbackServerFactory.CreateClientAndServerAsync(async uri => @@ -1538,37 +1559,35 @@ await test.LoopbackServerFactory.CreateClientAndServerAsync(async uri => { using HttpRequestMessage request = new(HttpMethod.Get, uri) { Version = test.UseVersion }; Task sendAsyncTask = client.SendAsync(request); - clientWaitingTcs.SetResult(); - using HttpResponseMessage response = await sendAsyncTask; + await serverTcs.Task.WaitAsync(TestHelper.PassingTestTimeout); + recorder.RecordObservableInstruments(); + clientTcs.SetResult(); + HttpResponseMessage response = await sendAsyncTask; + response.Dispose(); await WaitForEnvironmentTicksToAdvance(); + recorder.RecordObservableInstruments(); } + clientDisposedTcs.SetResult(); Version version = HttpVersion.Version11; Assert.Collection(recorder.GetMeasurements(), - m => VerifyActiveRequests(m.InstrumentName, (long)m.Value, m.Tags, 1, uri), - m => VerifyOpenConnections(m.InstrumentName, m.Value, m.Tags, 1, uri, version, "idle"), m => VerifyTimeInQueue(m.InstrumentName, m.Value, m.Tags, uri, version), - - m => VerifyOpenConnections(m.InstrumentName, m.Value, m.Tags, -1, uri, version, "idle"), + m => VerifyActiveRequests(m.InstrumentName, (long)m.Value, m.Tags, 1, uri), m => VerifyOpenConnections(m.InstrumentName, m.Value, m.Tags, 1, uri, version, "active"), - m => VerifyOpenConnections(m.InstrumentName, m.Value, m.Tags, -1, uri, version, "active"), - m => VerifyOpenConnections(m.InstrumentName, m.Value, m.Tags, 1, uri, version, "idle"), - - m => VerifyActiveRequests(m.InstrumentName, (long)m.Value, m.Tags, -1, uri), m => VerifyRequestDuration(m.InstrumentName, (double)m.Value, m.Tags, uri, version, 200), - m => VerifyConnectionDuration(m.InstrumentName, m.Value, m.Tags, uri, version), - m => VerifyOpenConnections(m.InstrumentName, m.Value, m.Tags, -1, uri, version, "idle")); + m => VerifyOpenConnections(m.InstrumentName, m.Value, m.Tags, 1, uri, version, "idle"), + m => VerifyConnectionDuration(m.InstrumentName, m.Value, m.Tags, uri, version)); }, async server => { - await clientWaitingTcs.Task.WaitAsync(TestHelper.PassingTestTimeout); - await server.AcceptConnectionAsync(async connection => { await connection.ReadRequestDataAsync(); + serverTcs.SetResult(); + await clientTcs.Task.WaitAsync(TestHelper.PassingTestTimeout); await connection.SendResponseAsync(isFinal: false); - await connection.WaitForCloseAsync(CancellationToken.None); + await clientDisposedTcs.Task.WaitAsync(TestHelper.PassingTestTimeout); }); }); }).DisposeAsync();