[Experiment] Reduces use of locks and improves thread-safe logic by juliomenendez · Pull Request #542 · microsoft/Agents-for-net

Julio Carlos Menendez (juliomenendez) · 2025-12-18T21:43:42Z

Removes direct use of lock and timers in StreamingReponse for an approach using reactive programming.

…otential deadlock code.

…hout-lock

Copilot

Pull request overview

This PR refactors the StreamingResponse class to use reactive programming patterns instead of explicit locks and timers. The goal is to reduce lock contention and improve thread-safe logic by introducing a new DynamicIntervalTextChunkBatcher component based on System.Reactive.

Key Changes:

Replaces timer-based batching with a reactive observable pipeline using DynamicIntervalTextChunkBatcher
Removes explicit lock statements and synchronization primitives (Timer, AutoResetEvent, queue)
Introduces reactive subjects for managing activity emissions and text batching

Reviewed changes

Copilot reviewed 5 out of 5 changed files in this pull request and generated 23 comments.

Show a summary per file

File	Description
`DynamicIntervalTextChunkBatcher.cs`	New reactive component that batches text chunks using Rx observables with configurable intervals
`DynamicIntervalTextChunkBatcherTests.cs`	Comprehensive test suite for the new batcher component
`StreamingResponse.cs`	Major refactoring to use reactive pipeline instead of locks and timers
`Microsoft.Agents.Builder.csproj`	Adds System.Reactive package reference
`Directory.Packages.props`	Adds System.Reactive version specification

Comments suppressed due to low confidence (1)

src/libraries/Builder/Microsoft.Agents.Builder/StreamingResponse.cs:415

Condition is always false because of access to field _canceled.

                    if (_canceled)

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

Copilot · 2025-12-23T21:50:09Z

+        public async Task TestOnCompletedFlushesRemainingText()
+        {
+            var batcher = new DynamicIntervalTextChunkBatcher(5000); // Long interval
+            string text = string.Empty;
+            var subscription = batcher.Subscribe(value => text = value);
+
+            batcher.OnNext("quick");
+            batcher.OnNext(" flush");
+
+            // Complete immediately without waiting for interval
+            batcher.OnCompleted();
+
+            // Give a small delay for the completion to process
+            await Task.Delay(50);
+
+            Assert.Equal("quick flush", text);
+
+            subscription.Dispose();
+            batcher.Dispose();
+        }


The test name "TestOnCompletedFlushesRemainingText" implies that OnCompleted should flush remaining text, but given the current implementation's flaw (issue #8), this test might not actually verify the correct behavior. The 50ms delay is a race condition - if the interval hasn't fired, the text won't be flushed. Consider testing this behavior more reliably or updating the test name if the current behavior is intentional.

Copilot · 2025-12-23T21:50:10Z

-            Func<IActivity> queueFunc;
-
-            lock (this)
+            if (_ended)


Race condition: The check of _ended and subsequent operations are not atomic. Without synchronization, another thread could set _ended to true after the check but before setting _streamStarted, leading to inconsistent state. The old implementation protected these operations with a lock. Consider using Volatile.Read for the check or adding appropriate synchronization.

Suggested change

if (_ended)

if (Volatile.Read(ref _ended))

Copilot · 2025-12-23T21:50:10Z

+            _streamStarted = true;
+            _accumulatedMessage += text;
+            Message = CitationUtils.FormatCitationsResponse(_accumulatedMessage);


Race condition: Multiple threads could simultaneously update _accumulatedMessage and Message, causing data loss or corruption. The string concatenation operations are not atomic. The old implementation protected these operations with a lock. Consider using a thread-safe approach or ensuring only one thread writes to these fields.

Copilot · 2025-12-23T21:50:10Z

+                catch
                {


The catch block catches all exceptions without logging or rethrowing, which could hide bugs and make debugging difficult. The old implementation caught specific exception types and had detailed error handling. Consider at minimum catching specific exceptions or adding logging before swallowing exceptions.

Suggested change

catch

{

catch (Exception ex)

{

_logger?.LogError(ex, "Error while completing streaming response.");

Copilot · 2025-12-23T21:50:11Z

+        public void TestOnNextAfterCompleted()
+        {
+            var batcher = new DynamicIntervalTextChunkBatcher(200);
+            var emissions = new List<string>();
+            var subscription = batcher.Subscribe(emissions.Add);
+
+            batcher.OnNext("before");
+            batcher.OnCompleted();
+
+            // This should be ignored
+            batcher.OnNext("after");
+
+            Assert.DoesNotContain("after", string.Join("", emissions));


Race condition in test: After calling OnCompleted, the test immediately checks if "before" is in emissions without waiting for the completion to propagate. Due to timing, the emissions list might still be empty. Consider adding a small delay or using a more reliable synchronization mechanism to ensure the completion has been processed before asserting.

Copilot · 2025-12-23T21:50:14Z

+                .Subscribe(
+                    async activity =>
+                    {
+                        await SendActivityAsync(activity, CancellationToken.None).ConfigureAwait(false);


The async lambda in the Subscribe method can lead to unobserved exceptions and race conditions. Reactive Extensions' Subscribe doesn't handle async delegates correctly - it returns immediately without awaiting the task, which can cause exceptions to be swallowed and operations to complete out of order. Consider using Observable.SelectMany or a synchronous subscription that properly handles the async work.

Suggested change

.Subscribe(

async activity =>

{

await SendActivityAsync(activity, CancellationToken.None).ConfigureAwait(false);

.SelectMany(activity => Observable.FromAsync(() => SendActivityAsync(activity, CancellationToken.None)))

.Subscribe(

_ =>

{

// No-op: all work is done in SendActivityAsync; this handler exists to observe completion.

Copilot · 2025-12-23T21:50:14Z

+                    buffer.Append(chunk);
+                    return buffer;


The Scan operator accumulates text in a single StringBuilder instance that is mutated and reused across emissions. This can cause race conditions where the same StringBuilder reference is emitted multiple times, leading to downstream observers seeing the same (changing) buffer. Each emission of ToString() will reflect the current state of the buffer, not the state at the time of emission. Consider creating a new string or StringBuilder for each emission.

Suggested change

buffer.Append(chunk);

return buffer;

var newBuffer = new StringBuilder(buffer.ToString());

newBuffer.Append(chunk);

return newBuffer;

Copilot · 2025-12-23T21:50:15Z

+                if (_ended)
                {
-                    if (_ended)
-                    {
-                        return StreamingResponseResult.AlreadyEnded;
-                    }
+                    return StreamingResponseResult.AlreadyEnded;
+                }

-                    _ended = true;
+                _ended = true;


Race condition: Multiple threads could read and write _streamStarted, _ended, and other boolean fields without synchronization. While boolean reads/writes are atomic on most platforms, memory visibility is not guaranteed across threads without proper synchronization. Consider using Volatile.Read/Write or Interlocked operations for all shared boolean fields to ensure thread-safe access.

Copilot · 2025-12-23T21:50:15Z

        /// <exception cref="System.InvalidOperationException">Throws if the stream has already ended.</exception>
-        public async Task QueueInformativeUpdateAsync(string text, CancellationToken cancellationToken = default)
+        public Task QueueInformativeUpdateAsync(string text, CancellationToken cancellationToken = default)
        {


The cancellationToken parameter is ignored. The old implementation used CancellationToken.None explicitly, but the new implementation doesn't use the provided token at all. If the intention is to not support cancellation, this should be documented, otherwise the provided cancellationToken should be used or passed through to async operations.

Suggested change

{

{

if (cancellationToken.IsCancellationRequested)

{

return Task.FromCanceled(cancellationToken);

}

Copilot · 2025-12-23T21:50:15Z

+            _disposed = true;
+            _subscription?.Dispose();
+            _inputSubject?.Dispose();
+            _intervalSubject?.Dispose();
+            _outputSubject?.Dispose();


Setting _disposed = true before disposing resources creates a time window where the object appears disposed but resources are still active. If another thread checks _disposed during disposal, it might assume cleanup is complete when it's not. Consider setting the flag after all resources are disposed, or use a lock to make the entire disposal atomic.

Suggested change

_disposed = true;

_subscription?.Dispose();

_inputSubject?.Dispose();

_intervalSubject?.Dispose();

_outputSubject?.Dispose();

_subscription?.Dispose();

_inputSubject?.Dispose();

_intervalSubject?.Dispose();

_outputSubject?.Dispose();

_disposed = true;

tracyboehrer (tracyboehrer) · 2026-01-19T18:05:17Z

Julio Carlos Menendez (@juliomenendez) I see this and will review soon. Some of the Copilot suggestions seem relevant.

Julio Carlos Menendez (juliomenendez) · 2026-01-19T18:42:09Z

Still have work on-going for this, trying to align more to observables architecture. Let me mark it as draft until I get time to wrap it up. Thanks tracyboehrer (@tracyboehrer) !

Julio Carlos Menendez (juliomenendez) · 2026-01-21T18:53:34Z

Superseeded by #655

Introduces thread safe operations and use of obsersavbles to remove p…

12f536e

…otential deadlock code.

github-actions Bot added ML: Core Tags changes to core libraries ML: Packages labels Dec 18, 2025

Uses a newer version of System.Reactive.

036c0e2

Julio Carlos Menendez (juliomenendez) force-pushed the juliome/streaming-without-lock branch from 4ca84e0 to 036c0e2 Compare December 18, 2025 22:40

Julio Carlos Menendez (juliomenendez) added 2 commits December 19, 2025 15:08

Replaces all state with IObservables. WIP.

e7ba7a9

Refactors text chunking to use better reactive logic.

7b0e19b

github-actions Bot added the ML: Tests Tags changes to tests label Dec 23, 2025

Julio Carlos Menendez (juliomenendez) added 2 commits December 23, 2025 12:58

Uses StreamTypes enum. Fixes non async task return.

34af4e2

Merge remote-tracking branch 'origin/main' into juliome/streaming-wit…

10929b0

…hout-lock

Julio Carlos Menendez (juliomenendez) marked this pull request as ready for review December 23, 2025 21:43

Julio Carlos Menendez (juliomenendez) requested a review from a team as a code owner December 23, 2025 21:43

Copilot AI review requested due to automatic review settings December 23, 2025 21:43

Copilot started reviewing on behalf of Julio Carlos Menendez (juliomenendez) December 23, 2025 21:44 View session

Copilot AI reviewed Dec 23, 2025

View reviewed changes

Julio Carlos Menendez (juliomenendez) marked this pull request as draft January 19, 2026 18:42

Julio Carlos Menendez (juliomenendez) closed this Jan 21, 2026

Julio Carlos Menendez (juliomenendez) mentioned this pull request Jan 22, 2026

Migrate from lock-based synchronization to Observable/Rx patterns #655

Open

3 tasks

-                catch
-                {
+                catch (Exception ex)
+                {
+                    _logger?.LogError(ex, "Error while completing streaming response.");

-                    buffer.Append(chunk);
-                    return buffer;
+                    var newBuffer = new StringBuilder(buffer.ToString());
+                    newBuffer.Append(chunk);
+                    return newBuffer;

-        {
+        {
+            if (cancellationToken.IsCancellationRequested)
+            {
+                return Task.FromCanceled(cancellationToken);
+            }

Uh oh!

Conversation

Julio Carlos Menendez (juliomenendez) commented Dec 18, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull request overview

Reviewed changes

Uh oh!

Copilot AI Dec 23, 2025

Choose a reason for hiding this comment

Uh oh!

Copilot AI Dec 23, 2025

Choose a reason for hiding this comment

Uh oh!

Copilot AI Dec 23, 2025

Choose a reason for hiding this comment

Uh oh!

Copilot AI Dec 23, 2025

Choose a reason for hiding this comment

Uh oh!

Copilot AI Dec 23, 2025

Choose a reason for hiding this comment

Uh oh!

Copilot AI Dec 23, 2025

Choose a reason for hiding this comment

Uh oh!

Copilot AI Dec 23, 2025

Choose a reason for hiding this comment

Uh oh!

Copilot AI Dec 23, 2025

Choose a reason for hiding this comment

Uh oh!

Copilot AI Dec 23, 2025

Choose a reason for hiding this comment

Uh oh!

Copilot AI Dec 23, 2025

Choose a reason for hiding this comment

Uh oh!

tracyboehrer (tracyboehrer) commented Jan 19, 2026

Uh oh!

Julio Carlos Menendez (juliomenendez) commented Jan 19, 2026

Uh oh!

Julio Carlos Menendez (juliomenendez) commented Jan 21, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

Julio Carlos Menendez (juliomenendez) commented Dec 18, 2025 •

edited

Loading