using ModelContextProtocol.Protocol; using System.Diagnostics; namespace ModelContextProtocol; /// Provides thread-safe storage for notification handlers. internal sealed class NotificationHandlers { /// A dictionary of linked lists of registrations, indexed by the notification method. private readonly Dictionary _handlers = []; /// Gets the object to be used for all synchronization. private object SyncObj => _handlers; /// /// Registers a collection of notification handlers at once. /// /// /// A collection of notification method names paired with their corresponding handler functions. /// Each key in the collection is a notification method name, and each value is a handler function /// that will be invoked when a notification with that method name is received. /// /// /// /// This method is typically used during client or server initialization to register /// all notification handlers provided in capabilities. /// /// /// Registrations completed with this method are permanent and non-removable. /// This differs from handlers registered with which can be temporary. /// /// /// When multiple handlers are registered for the same method, all handlers will be invoked /// in reverse order of registration (newest first) when a notification is received. /// /// /// The registered handlers will be invoked by when a notification /// with the corresponding method name is received. /// /// public void RegisterRange(IEnumerable>> handlers) { foreach (var entry in handlers) { _ = Register(entry.Key, entry.Value, temporary: false); } } /// /// Adds a notification handler as part of configuring the endpoint. /// /// The notification method for which the handler is being registered. /// The handler being registered. /// /// if the registration can be removed later; if it cannot. /// If , the registration will be permanent: calling /// on the returned instance will not unregister the handler. /// /// /// An that when disposed will unregister the handler if is . /// /// /// Multiple handlers can be registered for the same method. When a notification for that method is received, /// all registered handlers will be invoked in reverse order of registration (newest first). /// public IAsyncDisposable Register( string method, Func handler, bool temporary = true) { // Create the new registration instance. Registration reg = new(this, method, handler, temporary); // Store the registration into the dictionary. If there's not currently a registration for the method, // then this registration instance just becomes the single value. If there is currently a registration, // then this new registration becomes the new head of the linked list, and the old head becomes the next // item in the list. lock (SyncObj) { if (_handlers.TryGetValue(method, out var existingHandlerHead)) { reg.Next = existingHandlerHead; existingHandlerHead.Prev = reg; } _handlers[method] = reg; } // Return the new registration. It must be disposed of when no longer used, or it will end up being // leaked into the list. This is the same as with CancellationToken.Register. return reg; } /// /// Invokes all registered handlers for the specified notification method. /// /// The notification method name to invoke handlers for. /// The notification object to pass to each handler. /// The to monitor for cancellation requests. The default is . /// /// Handlers are invoked in reverse order of registration (newest first). /// If any handler throws an exception, all handlers will still be invoked, and an /// containing all exceptions will be thrown after all handlers have been invoked. /// public async Task InvokeHandlers(string method, JsonRpcNotification notification, CancellationToken cancellationToken) { // If there are no handlers registered for this method, we're done. Registration? reg; lock (SyncObj) { if (!_handlers.TryGetValue(method, out reg)) { return; } } // Invoke each handler in the list. We guarantee that we'll try to invoke // any handlers that were in the list when the list was fetched from the dictionary, // which is why DisposeAsync doesn't modify the Prev/Next of the registration being // disposed; if those were nulled out, we'd be unable to walk around it in the list // if we happened to be on that item when it was disposed. List? exceptions = null; while (reg is not null) { try { await reg.InvokeAsync(notification, cancellationToken).ConfigureAwait(false); } catch (Exception e) { (exceptions ??= []).Add(e); } lock (SyncObj) { reg = reg.Next; } } if (exceptions is not null) { throw new AggregateException(exceptions); } } /// Provides storage for a handler registration. private sealed class Registration( NotificationHandlers handlers, string method, Func handler, bool unregisterable) : IAsyncDisposable { /// Used to prevent deadlocks during disposal. /// /// The task returned from does not complete until all invocations of the handler /// have completed and no more will be performed, so that the consumer can then trust that any resources accessed /// by that handler are no longer in use and may be cleaned up. If were to be invoked /// and its task awaited from within the invocation of the handler, however, that would result in deadlock, since /// the task wouldn't complete until the invocation completed, and the invocation wouldn't complete until the task /// completed. To circumvent that, we track via an in-flight invocations. If /// detects it's being invoked from within an invocation, it will avoid waiting. For /// simplicity, we don't require that it's the same handler. /// private static readonly AsyncLocal s_invokingAncestor = new(); /// The parent to which this registration belongs. private readonly NotificationHandlers _handlers = handlers; /// The method with which this registration is associated. private readonly string _method = method; /// The handler this registration represents. private readonly Func _handler = handler; /// true if this instance is temporary; false if it's permanent private readonly bool _temporary = unregisterable; /// Provides a task that can await to know when all in-flight invocations have completed. /// /// This will only be initialized if sees in-flight invocations, in which case it'll initialize /// this and then await its task. The task will be completed when the last /// in-flight notification completes. /// private TaskCompletionSource? _disposeTcs; /// The number of remaining references to this registration. /// /// The ref count starts life at 1 to represent the whole registration; that ref count will be subtracted when /// the instance is disposed. Every invocation then temporarily increases the ref count before invocation and /// decrements it after. When is called, it decrements the ref count. In the common /// case, that'll bring the count down to 0, in which case the instance will never be subsequently invoked. /// If, however, after that decrement the count is still positive, then there are in-flight invocations; the last /// one of those to complete will end up decrementing the ref count to 0. /// private int _refCount = 1; /// Tracks whether has ever been invoked. /// /// It's rare but possible is called multiple times. Only the first /// should decrement the initial ref count, but they all must wait until all invocations have quiesced. /// private bool _disposedCalled = false; /// The next registration in the linked list. public Registration? Next; /// /// The previous registration in the linked list of handlers for a specific notification method. /// Used to maintain the bidirectional linked list when handlers are added or removed. /// public Registration? Prev; /// Removes the registration. public async ValueTask DisposeAsync() { if (!_temporary) { return; } lock (_handlers.SyncObj) { // If DisposeAsync was previously called, we don't want to do all of the work again // to remove the registration from the list, and we must not do the work again to // decrement the ref count and possibly initialize the _disposeTcs. if (!_disposedCalled) { _disposedCalled = true; // If this handler is the head of the list for this method, we need to update // the dictionary, either to point to a different head, or if this is the only // item in the list, to remove the entry from the dictionary entirely. if (_handlers._handlers.TryGetValue(_method, out var handlers) && handlers == this) { if (Next is not null) { _handlers._handlers[_method] = Next; } else { _handlers._handlers.Remove(_method); } } // Remove the registration from the linked list by routing the nodes around it // to point past this one. Importantly, we do not modify this node's Next or Prev. // We want to ensure that an enumeration through all of the registrations can still // progress through this one. Prev?.Next = Next; Next?.Prev = Prev; // Decrement the ref count. In the common case, there's no in-flight invocation for // this handler. However, in the uncommon case that there is, we need to wait for // that invocation to complete. To do that, initialize the _disposeTcs. It's created // with RunContinuationsAsynchronously so that completing it doesn't run the continuation // under any held locks. if (--_refCount != 0) { Debug.Assert(_disposeTcs is null); _disposeTcs = new(TaskCreationOptions.RunContinuationsAsynchronously); } } } // Ensure that DisposeAsync doesn't complete until all in-flight invocations have completed, // unless our call chain includes one of those in-flight invocations, in which case waiting // would deadlock. if (_disposeTcs is not null && s_invokingAncestor.Value == 0) { await _disposeTcs.Task.ConfigureAwait(false); } } /// Invoke the handler associated with the registration. public ValueTask InvokeAsync(JsonRpcNotification notification, CancellationToken cancellationToken) { // For permanent registrations, skip all the tracking overhead and just invoke the handler. if (!_temporary) { return _handler(notification, cancellationToken); } // For temporary registrations, track the invocation and coordinate with disposal. return InvokeTemporaryAsync(notification, cancellationToken); } /// Invoke the handler associated with the temporary registration. private async ValueTask InvokeTemporaryAsync(JsonRpcNotification notification, CancellationToken cancellationToken) { // Check whether we need to handle this registration. If DisposeAsync has been called, // then even if there are in-flight invocations for it, we avoid adding more. // If DisposeAsync has not been called, then we need to increment the ref count to // signal that there's another in-flight invocation. lock (_handlers.SyncObj) { Debug.Assert(_refCount != 0 || _disposedCalled, $"Expected {nameof(_disposedCalled)} == true when {nameof(_refCount)} == 0"); if (_disposedCalled) { return; } Debug.Assert(_refCount > 0); _refCount++; } // Ensure that if DisposeAsync is called from within the handler, it won't deadlock by waiting // for the in-flight invocation to complete. s_invokingAncestor.Value++; try { // Invoke the handler. await _handler(notification, cancellationToken).ConfigureAwait(false); } finally { // Undo the in-flight tracking. s_invokingAncestor.Value--; // Now decrement the ref count we previously incremented. If that brings the ref count to 0, // DisposeAsync must have been called while this was in-flight, which also means it's now // waiting on _disposeTcs; unblock it. lock (_handlers.SyncObj) { _refCount--; if (_refCount == 0) { Debug.Assert(_disposedCalled); Debug.Assert(_disposeTcs is not null); bool completed = _disposeTcs!.TrySetResult(true); Debug.Assert(completed); } } } } } }