From b9b22d7c6f827edc7b517c9a85d3e655bb7adf20 Mon Sep 17 00:00:00 2001 From: Nikolay Borisenko <22616990+nvborisenko@users.noreply.github.com> Date: Thu, 12 Mar 2026 20:13:07 +0300 Subject: [PATCH 1/2] [dotnet] [bidi] Keep events deserialization at transport layer --- dotnet/src/webdriver/BiDi/Broker.cs | 28 +++++++++---- dotnet/src/webdriver/BiDi/EventDispatcher.cs | 44 ++++++++++---------- 2 files changed, 40 insertions(+), 32 deletions(-) diff --git a/dotnet/src/webdriver/BiDi/Broker.cs b/dotnet/src/webdriver/BiDi/Broker.cs index f9e34da0a15ff..a121dde38732d 100644 --- a/dotnet/src/webdriver/BiDi/Broker.cs +++ b/dotnet/src/webdriver/BiDi/Broker.cs @@ -128,8 +128,7 @@ private void ProcessReceivedMessage(byte[] data) string? error = default; string? message = default; Utf8JsonReader resultReader = default; - int paramsStartIndex = 0; - int paramsEndIndex = 0; + Utf8JsonReader paramsReader = default; Utf8JsonReader reader = new(data); reader.Read(); // "{" @@ -138,8 +137,6 @@ private void ProcessReceivedMessage(byte[] data) while (reader.TokenType == JsonTokenType.PropertyName) { - bool isParams = false; - if (reader.ValueTextEquals("id"u8)) { reader.Read(); @@ -165,8 +162,7 @@ private void ProcessReceivedMessage(byte[] data) else if (reader.ValueTextEquals("params"u8)) { reader.Read(); - paramsStartIndex = (int)reader.TokenStartIndex; - isParams = true; + paramsReader = reader; // snapshot } else if (reader.ValueTextEquals("error"u8)) { @@ -184,7 +180,6 @@ private void ProcessReceivedMessage(byte[] data) } reader.Skip(); - if (isParams) paramsEndIndex = (int)reader.BytesConsumed; reader.Read(); } @@ -223,8 +218,23 @@ private void ProcessReceivedMessage(byte[] data) case TypeEvent: if (method is null) throw new BiDiException($"The remote end responded with 'event' message type, but missed required 'method' property. Message content: {System.Text.Encoding.UTF8.GetString(data)}"); - var paramsJsonData = new ReadOnlyMemory(data, paramsStartIndex, paramsEndIndex - paramsStartIndex); - _eventDispatcher.EnqueueEvent(method, paramsJsonData, _bidi); + + if (!_eventDispatcher.TryGetJsonTypeInfo(method, out var jsonTypeInfo)) + { + if (_logger.IsEnabled(LogEventLevel.Warn)) + { + _logger.Warn($"Received BiDi event with method '{method}', but no event type mapping was found. Event will be ignored. Message content: {System.Text.Encoding.UTF8.GetString(data)}"); + } + + break; + } + + var eventArgs = JsonSerializer.Deserialize(ref paramsReader, jsonTypeInfo) as EventArgs + ?? throw new BiDiException("Remote end returned null event args in the 'params' property."); + + eventArgs.BiDi = _bidi; + + _eventDispatcher.EnqueueEvent(method, eventArgs); break; case TypeError: diff --git a/dotnet/src/webdriver/BiDi/EventDispatcher.cs b/dotnet/src/webdriver/BiDi/EventDispatcher.cs index a1ab136b8c84d..39c9d82c7c0e7 100644 --- a/dotnet/src/webdriver/BiDi/EventDispatcher.cs +++ b/dotnet/src/webdriver/BiDi/EventDispatcher.cs @@ -18,7 +18,7 @@ // using System.Collections.Concurrent; -using System.Text.Json; +using System.Diagnostics.CodeAnalysis; using System.Text.Json.Serialization.Metadata; using System.Threading.Channels; using OpenQA.Selenium.BiDi.Session; @@ -32,7 +32,7 @@ internal sealed class EventDispatcher : IAsyncDisposable private readonly Func _sessionProvider; - private readonly ConcurrentDictionary _events = new(); + private readonly ConcurrentDictionary _eventRegistrations = new(); private readonly Channel _pendingEvents = Channel.CreateUnbounded(new() { @@ -51,7 +51,7 @@ public EventDispatcher(Func sessionProvider) public async Task SubscribeAsync(string eventName, EventHandler eventHandler, SubscriptionOptions? options, JsonTypeInfo jsonTypeInfo, CancellationToken cancellationToken) where TEventArgs : EventArgs { - var registration = _events.GetOrAdd(eventName, _ => new EventRegistration(jsonTypeInfo)); + var registration = _eventRegistrations.GetOrAdd(eventName, _ => new EventRegistration(jsonTypeInfo)); var subscribeResult = await _sessionProvider().SubscribeAsync([eventName], new() { Contexts = options?.Contexts, UserContexts = options?.UserContexts }, cancellationToken).ConfigureAwait(false); @@ -62,26 +62,16 @@ public async Task SubscribeAsync(string eventName, Eve public async ValueTask UnsubscribeAsync(Subscription subscription, CancellationToken cancellationToken) { - if (_events.TryGetValue(subscription.EventHandler.EventName, out var registration)) + if (_eventRegistrations.TryGetValue(subscription.EventHandler.EventName, out var registration)) { await _sessionProvider().UnsubscribeAsync([subscription.SubscriptionId], null, cancellationToken).ConfigureAwait(false); registration.RemoveHandler(subscription.EventHandler); } } - public void EnqueueEvent(string method, ReadOnlyMemory jsonUtf8Bytes, IBiDi bidi) + public void EnqueueEvent(string method, EventArgs eventArgs) { - if (_events.TryGetValue(method, out var registration) && registration.TypeInfo is not null) - { - _pendingEvents.Writer.TryWrite(new PendingEvent(method, jsonUtf8Bytes, bidi, registration.TypeInfo)); - } - else - { - if (_logger.IsEnabled(LogEventLevel.Warn)) - { - _logger.Warn($"Received BiDi event with method '{method}', but no event type mapping was found. Event will be ignored."); - } - } + _pendingEvents.Writer.TryWrite(new PendingEvent(method, eventArgs)); } private async Task ProcessEventsAwaiterAsync() @@ -93,15 +83,11 @@ private async Task ProcessEventsAwaiterAsync() { try { - if (_events.TryGetValue(result.Method, out var registration)) + if (_eventRegistrations.TryGetValue(result.Method, out var registration)) { - // Deserialize on background thread instead of network thread (single parse) - var eventArgs = (EventArgs)JsonSerializer.Deserialize(result.JsonUtf8Bytes.Span, result.TypeInfo)!; - eventArgs.BiDi = result.BiDi; - foreach (var handler in registration.GetHandlers()) // copy-on-write array, safe to iterate { - await handler.InvokeAsync(eventArgs).ConfigureAwait(false); + await handler.InvokeAsync(result.EventArgs).ConfigureAwait(false); } } } @@ -125,7 +111,19 @@ public async ValueTask DisposeAsync() GC.SuppressFinalize(this); } - private readonly record struct PendingEvent(string Method, ReadOnlyMemory JsonUtf8Bytes, IBiDi BiDi, JsonTypeInfo TypeInfo); + public bool TryGetJsonTypeInfo(string eventName, [NotNullWhen(true)] out JsonTypeInfo? jsonTypeInfo) + { + if (_eventRegistrations.TryGetValue(eventName, out var registration)) + { + jsonTypeInfo = registration.TypeInfo; + return true; + } + + jsonTypeInfo = null; + return false; + } + + private readonly record struct PendingEvent(string Method, EventArgs EventArgs); private sealed class EventRegistration(JsonTypeInfo typeInfo) { From f1e8710651018f20433b2d70ff8b01ebda5f7dc0 Mon Sep 17 00:00:00 2001 From: Nikolay Borisenko <22616990+nvborisenko@users.noreply.github.com> Date: Thu, 12 Mar 2026 21:21:15 +0300 Subject: [PATCH 2/2] [dotnet] [bidi] Parallel event handlers --- dotnet/src/webdriver/BiDi/EventDispatcher.cs | 37 ++++++++++++++------ 1 file changed, 26 insertions(+), 11 deletions(-) diff --git a/dotnet/src/webdriver/BiDi/EventDispatcher.cs b/dotnet/src/webdriver/BiDi/EventDispatcher.cs index 39c9d82c7c0e7..16753b106ff18 100644 --- a/dotnet/src/webdriver/BiDi/EventDispatcher.cs +++ b/dotnet/src/webdriver/BiDi/EventDispatcher.cs @@ -34,6 +34,8 @@ internal sealed class EventDispatcher : IAsyncDisposable private readonly ConcurrentDictionary _eventRegistrations = new(); + private readonly ConcurrentDictionary _runningHandlers = new(); + private readonly Channel _pendingEvents = Channel.CreateUnbounded(new() { SingleReader = true, @@ -81,23 +83,34 @@ private async Task ProcessEventsAwaiterAsync() { while (reader.TryRead(out var result)) { - try + if (_eventRegistrations.TryGetValue(result.Method, out var registration)) { - if (_eventRegistrations.TryGetValue(result.Method, out var registration)) + foreach (var handler in registration.GetHandlers()) // copy-on-write array, safe to iterate { - foreach (var handler in registration.GetHandlers()) // copy-on-write array, safe to iterate + var runningHandlerTask = InvokeHandlerAsync(handler, result.EventArgs); + if (!runningHandlerTask.IsCompleted) { - await handler.InvokeAsync(result.EventArgs).ConfigureAwait(false); + _runningHandlers.TryAdd(runningHandlerTask, 0); + _ = runningHandlerTask.ContinueWith(static (t, state) => ((ConcurrentDictionary)state!).TryRemove(t, out _), + _runningHandlers, TaskContinuationOptions.ExecuteSynchronously); } } } - catch (Exception ex) - { - if (_logger.IsEnabled(LogEventLevel.Error)) - { - _logger.Error($"Unhandled error processing BiDi event handler: {ex}"); - } - } + } + } + } + + private async Task InvokeHandlerAsync(EventHandler handler, EventArgs eventArgs) + { + try + { + await handler.InvokeAsync(eventArgs).ConfigureAwait(false); + } + catch (Exception ex) + { + if (_logger.IsEnabled(LogEventLevel.Error)) + { + _logger.Error($"Unhandled error processing BiDi event handler: {ex}"); } } } @@ -108,6 +121,8 @@ public async ValueTask DisposeAsync() await _eventEmitterTask.ConfigureAwait(false); + await Task.WhenAll(_runningHandlers.Keys).ConfigureAwait(false); + GC.SuppressFinalize(this); }