perf: Replace Dictionary+lock with ConcurrentDictionary in ModuleResultRegistry (#1726)

* perf: Replace Dictionary+lock with ConcurrentDictionary in ModuleResultRegistry (#1540)

Use ConcurrentDictionary for lock-free reads and thread-safe writes,
reducing lock contention during parallel module execution.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

* fix: Use GetOrAdd in all RegisterResult/SetException methods to prevent race condition

Ensures TCS is created atomically when registering results/exceptions,
preventing race conditions where TryGetValue might miss a not-yet-registered
module entry.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

* fix: Reorder operations to prevent race condition in GetCompletionTask

Create TCS before storing result to ensure GetCompletionTask never
returns null when a result already exists. This prevents a race where:
1. Thread A stores result
2. Thread B calls GetCompletionTask, gets null (TCS doesn't exist)
3. Thread A creates TCS

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

* fix: Add memory barrier to ensure result visibility before task completion

Adds Thread.MemoryBarrier() between writing to _results and calling
TrySetResult. This ensures that threads waking up from task completion
will see the _results write, preventing a potential memory visibility
race condition on weak memory model architectures.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

* refactor: Simplify memory ordering and fix GetCompletionTask consistency

- Remove explicit Thread.MemoryBarrier() - TrySetResult already provides
  release semantics ensuring _results write is visible to awaiters
- Change GetCompletionTask to use GetOrAdd for consistency with SetException
  (both now create TCS on-demand)
- Update interface: GetCompletionTask now returns Task instead of Task?
  since it always returns a valid task

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

* fix: Preserve original behavioral semantics for SetException and GetCompletionTask

Revert behavioral changes to maintain backward compatibility:
- SetException: Only sets exception if module was previously registered
  (does nothing for unregistered modules, matching original behavior)
- GetCompletionTask: Returns null for unregistered modules
  (matching original behavior)

RegisterResult methods still use GetOrAdd since registering a result
implies the module should be tracked.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 4.5 <noreply@anthropic.com>

Author: thomhurst

src/ModularPipelines/Engine/IModuleResultRegistry.cs

@@ -1,3 +1,4 @@
+using System.Collections.Concurrent;
 using ModularPipelines.Models;
 
 namespace ModularPipelines.Engine;
@@ -61,92 +62,67 @@ internal interface IModuleResultRegistry
 
 /// <summary>
 /// Default implementation of the module result registry.
+/// Uses ConcurrentDictionary for lock-free reads and thread-safe writes.
 /// </summary>
 internal class ModuleResultRegistry : IModuleResultRegistry
 {
-    private readonly Dictionary<Type, object> _results = new();
-    private readonly Dictionary<Type, TaskCompletionSource<object?>> _completionSources = new();
-    private readonly object _lock = new();
+    private readonly ConcurrentDictionary<Type, object> _results = new();
+    private readonly ConcurrentDictionary<Type, TaskCompletionSource<object?>> _completionSources = new();
 
     public void RegisterModule(Type moduleType)
     {
-        lock (_lock)
-        {
-            if (!_completionSources.ContainsKey(moduleType))
-            {
-                _completionSources[moduleType] = new TaskCompletionSource<object?>();
-            }
-        }
+        _completionSources.GetOrAdd(moduleType, _ => new TaskCompletionSource<object?>());
     }
 
     public void RegisterResult<T>(Type moduleType, ModuleResult<T> result)
     {
-        lock (_lock)
-        {
-            _results[moduleType] = result;
-
-            if (_completionSources.TryGetValue(moduleType, out var tcs))
-            {
-                tcs.TrySetResult(result);
-            }
-        }
+        // Store result first, then signal completion
+        // TrySetResult provides release semantics, ensuring _results write is visible to awaiters
+        _results[moduleType] = result;
+        var tcs = _completionSources.GetOrAdd(moduleType, _ => new TaskCompletionSource<object?>());
+        tcs.TrySetResult(result);
     }
 
     public ModuleResult<T>? GetResult<T>(Type moduleType)
     {
-        lock (_lock)
+        if (_results.TryGetValue(moduleType, out var result) && result is ModuleResult<T> typedResult)
         {
-            if (_results.TryGetValue(moduleType, out var result) && result is ModuleResult<T> typedResult)
-            {
-                return typedResult;
-            }
-
-            return null;
+            return typedResult;
         }
+
+        return null;
     }
 
     public IModuleResult? GetResult(Type moduleType)
     {
-        lock (_lock)
+        if (_results.TryGetValue(moduleType, out var result))
         {
-            if (_results.TryGetValue(moduleType, out var result))
-            {
-                return result as IModuleResult;
-            }
-
-            return null;
+            return result as IModuleResult;
         }
+
+        return null;
     }
 
     public Task? GetCompletionTask(Type moduleType)
     {
-        lock (_lock)
-        {
-            return _completionSources.TryGetValue(moduleType, out var tcs) ? tcs.Task : null;
-        }
+        return _completionSources.TryGetValue(moduleType, out var tcs) ? tcs.Task : null;
     }
 
     public void SetException(Type moduleType, Exception exception)
     {
-        lock (_lock)
+        // Only set exception if module was previously registered (preserves original behavior)
+        if (_completionSources.TryGetValue(moduleType, out var tcs))
         {
-            if (_completionSources.TryGetValue(moduleType, out var tcs))
-            {
-                tcs.TrySetException(exception);
-            }
+            tcs.TrySetException(exception);
         }
     }
 
     public void RegisterResult(Type moduleType, IModuleResult result)
     {
-        lock (_lock)
-        {
-            _results[moduleType] = result;
-
-            if (_completionSources.TryGetValue(moduleType, out var tcs))
-            {
-                tcs.TrySetResult(result);
-            }
-        }
+        // Store result first, then signal completion
+        // TrySetResult provides release semantics, ensuring _results write is visible to awaiters
+        _results[moduleType] = result;
+        var tcs = _completionSources.GetOrAdd(moduleType, _ => new TaskCompletionSource<object?>());
+        tcs.TrySetResult(result);
     }
 }