diff --git a/src/Orleans.Core.Abstractions/IDs/ClientGrainId.cs b/src/Orleans.Core.Abstractions/IDs/ClientGrainId.cs index a9dfb9f84f9..a54ce25332e 100644 --- a/src/Orleans.Core.Abstractions/IDs/ClientGrainId.cs +++ b/src/Orleans.Core.Abstractions/IDs/ClientGrainId.cs @@ -59,6 +59,29 @@ public static bool TryParse(GrainId grainId, out ClientGrainId clientId) return true; } + /// + /// Checks if the provided points to the same client as this . + /// + /// The to compare. + /// if the provided corresponds to the same client, otherwise . + public bool IsClientEqual(GrainId other) + { + if (!GrainId.Type.Equals(other.Type)) + { + return false; + } + + // Strip the observer id, if present, using span-based operations to avoid allocations. + var key = other.Key.AsSpan(); + if (key.IndexOf((byte)ObserverGrainId.SegmentSeparator) is int index && index >= 0) + { + key = key[..index]; + } + + // Compare the stripped key with the current GrainId's key. + return GrainId.Key.AsSpan().SequenceEqual(key); + } + /// public override bool Equals(object? obj) => obj is ClientGrainId clientId && GrainId.Equals(clientId.GrainId); diff --git a/src/Orleans.Core/Diagnostics/EventSourceEvents.cs b/src/Orleans.Core/Diagnostics/EventSourceEvents.cs index 7b21ef17bdf..fb388c1aaca 100644 --- a/src/Orleans.Core/Diagnostics/EventSourceEvents.cs +++ b/src/Orleans.Core/Diagnostics/EventSourceEvents.cs @@ -52,7 +52,7 @@ public void OnTargetSiloFail(Message message) /// Indicates that a request completed. /// [NonEvent] - public void DoCallback(Message message) + public void OnResponse(Message message) { if (this.IsEnabled()) { diff --git a/src/Orleans.Core/Messaging/Message.cs b/src/Orleans.Core/Messaging/Message.cs index f57033e8dae..75b731cdc0a 100644 --- a/src/Orleans.Core/Messaging/Message.cs +++ b/src/Orleans.Core/Messaging/Message.cs @@ -9,6 +9,11 @@ namespace Orleans.Runtime [Id(101)] internal sealed class Message : ISpanFormattable { + private const string GatewayRequestOwnerHeader = "#orleans.gateway.request-owner"; + private const string GatewayRequestOwnerSiloHeader = "#orleans.gateway.request-owner-silo"; + private const string GatewayResponseTargetHeader = "#orleans.gateway.response-target"; + private const string GatewayRequestTimeoutHeader = "#orleans.gateway.request-timeout"; + public const int LENGTH_HEADER_SIZE = 8; public const int LENGTH_META_HEADER = 4; internal const int MaxCacheInvalidationHeaderEntries = 16; @@ -16,6 +21,15 @@ internal sealed class Message : ISpanFormattable [NonSerialized] private short _retryCount; + [NonSerialized] + private bool _hasGatewayRequestSource; + + [NonSerialized] + private SiloAddress? _gatewayRequestSource; + + [NonSerialized] + private bool _hasTrustedGatewayResponseTarget; + public CoarseStopwatch _timeToExpiry; public object? BodyObject { get; set; } @@ -257,6 +271,179 @@ public Dictionary? RequestContextData } } + internal static bool IsGatewayRequestContextHeader(string key) + => key is GatewayRequestOwnerHeader + or GatewayRequestOwnerSiloHeader + or GatewayResponseTargetHeader + or GatewayRequestTimeoutHeader; + + internal void SetGatewayRequestTimeout(TimeSpan timeout) + { + var context = RequestContextData ??= []; + context[GatewayRequestTimeoutHeader] = timeout; + } + + internal TimeSpan? GetGatewayRequestTimeout() + => RequestContextData is { } context + && context.TryGetValue(GatewayRequestTimeoutHeader, out var value) + && value is TimeSpan timeout + ? timeout + : null; + + internal void ClearGatewayRequestOwner() + { + ClearGatewayRequestRouting(); + if (RequestContextData is { } context) + { + context.Remove(GatewayRequestTimeoutHeader); + if (context.Count == 0) + { + RequestContextData = null; + } + } + } + + internal void ClearGatewayRequestRouting() + { + if (RequestContextData is { } context) + { + context.Remove(GatewayRequestOwnerHeader); + context.Remove(GatewayRequestOwnerSiloHeader); + context.Remove(GatewayResponseTargetHeader); + if (context.Count == 0) + { + RequestContextData = null; + } + } + + _gatewayRequestSource = null; + _hasGatewayRequestSource = false; + _hasTrustedGatewayResponseTarget = false; + } + + internal void SetGatewayRequestOwner(SiloAddress ownerGateway, SiloAddress ownerSilo) + { + var context = RequestContextData ??= []; + if (!_hasGatewayRequestSource) + { + _gatewayRequestSource = SendingSilo; + _hasGatewayRequestSource = true; + } + + context.Remove(GatewayRequestOwnerHeader); + context.Remove(GatewayRequestOwnerSiloHeader); + context.Remove(GatewayResponseTargetHeader); + context[GatewayRequestOwnerHeader] = ownerGateway; + context[GatewayRequestOwnerSiloHeader] = ownerSilo; + if (_gatewayRequestSource is { } responseTarget) + { + context[GatewayResponseTargetHeader] = responseTarget; + } + + SendingSilo = ownerGateway; + } + + internal void RestoreGatewayRequestSource() + { + if (RequestContextData is not { } context + || !context.Remove(GatewayRequestOwnerHeader)) + { + return; + } + + SendingSilo = context.Remove(GatewayResponseTargetHeader, out var targetValue) + && targetValue is SiloAddress responseTarget + ? responseTarget + : null; + context.Remove(GatewayRequestOwnerSiloHeader); + _gatewayRequestSource = SendingSilo; + + if (context.Count == 0) + { + RequestContextData = null; + } + } + + internal void ApplyGatewayRequestOwner(Message request) + { + if (request.RequestContextData is not { } requestContext + || !requestContext.TryGetValue(GatewayRequestOwnerHeader, out var ownerValue) + || ownerValue is not SiloAddress ownerGateway) + { + return; + } + + var responseContext = RequestContextData ??= []; + responseContext[GatewayRequestOwnerHeader] = ownerGateway; + if (requestContext.TryGetValue(GatewayRequestOwnerSiloHeader, out var ownerSiloValue) + && ownerSiloValue is SiloAddress ownerSilo) + { + responseContext[GatewayRequestOwnerSiloHeader] = ownerSilo; + } + if (requestContext.TryGetValue(GatewayResponseTargetHeader, out var targetValue) + && targetValue is SiloAddress responseTarget) + { + responseContext[GatewayResponseTargetHeader] = responseTarget; + } + else + { + responseContext.Remove(GatewayResponseTargetHeader); + } + + TargetSilo = ownerGateway; + } + + internal bool TryGetGatewayRequestOwner(out SiloAddress ownerGateway, out SiloAddress ownerSilo) + { + ownerGateway = default!; + ownerSilo = default!; + if (RequestContextData is not { } context + || !context.TryGetValue(GatewayRequestOwnerHeader, out var ownerValue) + || ownerValue is not SiloAddress gateway + || !context.TryGetValue(GatewayRequestOwnerSiloHeader, out var ownerSiloValue) + || ownerSiloValue is not SiloAddress silo) + { + return false; + } + + ownerGateway = gateway; + ownerSilo = silo; + return true; + } + + internal void RestoreGatewayResponseTarget(bool preserveRoute = false) + { + if (RequestContextData is not { } context) + { + return; + } + + context.Remove(GatewayRequestOwnerHeader); + context.Remove(GatewayRequestOwnerSiloHeader); + TargetSilo = context.Remove(GatewayResponseTargetHeader, out var targetValue) + && targetValue is SiloAddress responseTarget + ? responseTarget + : null; + _hasTrustedGatewayResponseTarget = preserveRoute && TargetSilo is not null; + + if (context.Count == 0) + { + RequestContextData = null; + } + } + + internal bool TryTakeTrustedGatewayResponseTarget(out SiloAddress target) + { + target = TargetSilo!; + if (!_hasTrustedGatewayResponseTarget || target is null) + { + return false; + } + + _hasTrustedGatewayResponseTarget = false; + return true; + } + public GrainInterfaceType InterfaceType { get => _interfaceType; diff --git a/src/Orleans.Core/Messaging/MessageFactory.cs b/src/Orleans.Core/Messaging/MessageFactory.cs index 1f815944ea9..3026c8e4f73 100644 --- a/src/Orleans.Core/Messaging/MessageFactory.cs +++ b/src/Orleans.Core/Messaging/MessageFactory.cs @@ -41,6 +41,7 @@ public Message CreateMessage(object? body, InvokeMethodOptions options) BodyObject = body, RequestContextData = RequestContextExtensions.Export(_deepCopier), }; + message.ClearGatewayRequestOwner(); _messagingTrace.OnCreateMessage(message); return message; @@ -69,6 +70,7 @@ public Message CreateResponseMessage(Message request) TimeToLive = request.TimeToLive, RequestContextData = RequestContextExtensions.Export(_deepCopier), }; + response.ApplyGatewayRequestOwner(request); _messagingTrace.OnCreateMessage(response); return response; diff --git a/src/Orleans.Core/OrleansContracts.txt b/src/Orleans.Core/OrleansContracts.txt index 8ab2713ef70..3a6fb5c6cc6 100644 --- a/src/Orleans.Core/OrleansContracts.txt +++ b/src/Orleans.Core/OrleansContracts.txt @@ -16,6 +16,8 @@ interface [GrainInterfaceType("Orleans.ClientObservers.IClientGatewayObserver")] interface [GrainInterfaceType("Orleans.IMembershipTableSystemTarget")] Orleans.IMembershipTableSystemTarget [Version(0)] interface [GrainInterfaceType("Orleans.ISiloControl")] Orleans.ISiloControl [Version(0)] + 16D39D91: CompleteGatewayRequest(Orleans.Runtime.GrainId, Orleans.Runtime.GrainId, Orleans.Runtime.CorrelationId) -> Task + B99FB859: DropDisconnectedClients(bool) -> Task 45D07D09: ForceActivationCollection(System.TimeSpan) -> Task F388CED1: ForceGarbageCollection() -> Task 0C7DBD0C: ForceRuntimeStatisticsCollection() -> Task @@ -58,6 +60,7 @@ interface [GrainInterfaceType("Orleans.Runtime.IGrainCallCancellationExtension") FA239824: CancelRequestAsync(Orleans.Runtime.GrainId, Orleans.Runtime.CorrelationId) -> ValueTask interface [GrainInterfaceType("Orleans.Runtime.IManagementGrain")] Orleans.Runtime.IManagementGrain [Version(0)] + 101564A8: DropDisconnectedClients(bool) -> Task 329F9A1B: ForceActivationCollection(Orleans.Runtime.SiloAddress[], System.TimeSpan) -> Task 54E6D1D1: ForceActivationCollection(System.TimeSpan) -> Task 5922EB76: ForceGarbageCollection(Orleans.Runtime.SiloAddress[]) -> Task diff --git a/src/Orleans.Core/Runtime/CallbackData.cs b/src/Orleans.Core/Runtime/CallbackData.cs index bfbfb7a28f4..7e34e890a16 100644 --- a/src/Orleans.Core/Runtime/CallbackData.cs +++ b/src/Orleans.Core/Runtime/CallbackData.cs @@ -211,7 +211,7 @@ public void DoCallback(Message response) return; } - OrleansCallBackDataEvent.Instance.DoCallback(this.Message); + OrleansCallBackDataEvent.Instance.OnResponse(this.Message); this.stopwatch.Stop(); DisposeCancellationRegistration(); @@ -221,6 +221,8 @@ public void DoCallback(Message response) ResponseCallback(response, this.context); } + public void OnResponse(Message response) => DoCallback(response); + private bool TryComplete() => (Interlocked.Or(ref _state, StateCompleted) & StateCompleted) == 0; private void DisposeCancellationRegistration() diff --git a/src/Orleans.Core/Runtime/OutsideRuntimeClient.cs b/src/Orleans.Core/Runtime/OutsideRuntimeClient.cs index bb86127f633..50edffb5bb0 100644 --- a/src/Orleans.Core/Runtime/OutsideRuntimeClient.cs +++ b/src/Orleans.Core/Runtime/OutsideRuntimeClient.cs @@ -281,11 +281,16 @@ public void SendRequest(GrainReference target, IInvokable request, IResponseComp message.TargetSilo = systemTargetGrainId.GetSiloAddress(); } + var responseTimeout = request.GetDefaultResponseTimeout() ?? this.sharedCallbackData.ResponseTimeout; + if (targetGrainId.IsClient()) + { + message.SetGatewayRequestTimeout(responseTimeout); + } + if (this.clientMessagingOptions.DropExpiredMessages && message.IsExpirableMessage()) { // don't set expiration for system target messages. - var ttl = request.GetDefaultResponseTimeout() ?? this.clientMessagingOptions.ResponseTimeout; - message.TimeToLive = ttl; + message.TimeToLive = responseTimeout; } if (!oneWay) diff --git a/src/Orleans.Core/Runtime/RequestContextExtensions.cs b/src/Orleans.Core/Runtime/RequestContextExtensions.cs index b0c09d68e4e..b9a5d2a5595 100644 --- a/src/Orleans.Core/Runtime/RequestContextExtensions.cs +++ b/src/Orleans.Core/Runtime/RequestContextExtensions.cs @@ -18,11 +18,20 @@ public static class RequestContextExtensions /// The context data. public static void Import(Dictionary? contextData) { - var values = contextData switch + Dictionary? values = null; + if (contextData is { Count: > 0 }) { - { Count: > 0 } => contextData.ToDictionary(kvp => kvp.Key, kvp => kvp.Value), - _ => null, - }; + foreach (var (key, value) in contextData) + { + if (Message.IsGatewayRequestContextHeader(key)) + { + continue; + } + + values ??= new(contextData.Count); + values.Add(key, value); + } + } RequestContext.CallContextData.Value = new RequestContext.ContextProperties { diff --git a/src/Orleans.Core/SystemTargetInterfaces/IManagementGrain.cs b/src/Orleans.Core/SystemTargetInterfaces/IManagementGrain.cs index f9cdf0453d4..e819361b8fa 100644 --- a/src/Orleans.Core/SystemTargetInterfaces/IManagementGrain.cs +++ b/src/Orleans.Core/SystemTargetInterfaces/IManagementGrain.cs @@ -141,6 +141,13 @@ public interface IManagementGrain : IGrainWithIntegerKey, IVersionManager /// The hosts to invoke the operation on. /// A task representing the work performed. ValueTask ResetGrainCallFrequencies(SiloAddress[]? hostsIds = null); + + /// + /// Instructs all gateways to drop defunct (disconnected and expired) clients. + /// + /// If true, only clients that have been disconnected for longer than the configured client expiration time will be dropped. + /// A task representing the work performed. + Task DropDisconnectedClients(bool excludeRecent); } /// diff --git a/src/Orleans.Core/SystemTargetInterfaces/ISiloControl.cs b/src/Orleans.Core/SystemTargetInterfaces/ISiloControl.cs index 8cca206a9d3..cc15b91318d 100644 --- a/src/Orleans.Core/SystemTargetInterfaces/ISiloControl.cs +++ b/src/Orleans.Core/SystemTargetInterfaces/ISiloControl.cs @@ -1,6 +1,7 @@ using System; using System.Collections.Generic; using System.Threading.Tasks; +using Orleans.Concurrency; using Orleans.Providers; using Orleans.Runtime; @@ -25,5 +26,8 @@ internal interface ISiloControl : ISystemTarget, IVersionManager Task SendControlCommandToProvider(string providerName, int command, object? arg) where T : IControllable; Task> GetActiveGrains(GrainType grainType); + [OneWay, AlwaysInterleave] + Task CompleteGatewayRequest(GrainId clientId, GrainId sourceId, CorrelationId correlationId); + Task DropDisconnectedClients(bool excludeRecent); } } diff --git a/src/Orleans.Runtime/Core/HostedClient.cs b/src/Orleans.Runtime/Core/HostedClient.cs index a9d0e6a90e4..5d975d970ee 100644 --- a/src/Orleans.Runtime/Core/HostedClient.cs +++ b/src/Orleans.Runtime/Core/HostedClient.cs @@ -26,7 +26,6 @@ internal sealed partial class HostedClient : IGrainContext, IGrainExtensionBinde private readonly object lockObj = new(); #endif private readonly Channel incomingMessages; - private readonly IGrainReferenceRuntime grainReferenceRuntime; private readonly InvokableObjectManager invokableObjects; private readonly InsideRuntimeClient runtimeClient; private readonly ILogger logger; @@ -43,7 +42,6 @@ public HostedClient( InsideRuntimeClient runtimeClient, ILocalSiloDetails siloDetails, ILogger logger, - IGrainReferenceRuntime grainReferenceRuntime, IInternalGrainFactory grainFactory, MessageCenter messageCenter, MessagingTrace messagingTrace, @@ -59,7 +57,6 @@ public HostedClient( }); this.runtimeClient = runtimeClient; - this.grainReferenceRuntime = grainReferenceRuntime; this.grainFactory = grainFactory; this.invokableObjects = new InvokableObjectManager( this, @@ -186,8 +183,9 @@ public void SetComponent(TComponent? instance) where TComponent : cl /// public bool TryDispatchToClient(Message message) { - if (!ClientGrainId.TryParse(message.TargetGrain, out var targetClient) || !this.ClientId.Equals(targetClient)) + if (!ClientId.IsClientEqual(message.TargetGrain)) { + // The message does not target the hosted client. return false; } @@ -197,14 +195,13 @@ public bool TryDispatchToClient(Message message) return true; } - this.ReceiveMessage(message); + ReceiveMessage(message); return true; } public void ReceiveMessage(object message) { var msg = (Message)message; - if (msg.Direction == Message.Directions.Response) { // Requests are made through the runtime client, so deliver responses to the runtime client so that the request callback can be executed. diff --git a/src/Orleans.Runtime/Core/InsideRuntimeClient.cs b/src/Orleans.Runtime/Core/InsideRuntimeClient.cs index d605549c37e..90ef0258dad 100644 --- a/src/Orleans.Runtime/Core/InsideRuntimeClient.cs +++ b/src/Orleans.Runtime/Core/InsideRuntimeClient.cs @@ -25,7 +25,7 @@ namespace Orleans.Runtime /// /// Internal class for system grains to get access to runtime object /// - internal sealed partial class InsideRuntimeClient : IRuntimeClient, ILifecycleParticipant + internal sealed partial class InsideRuntimeClient : IRuntimeClient, ILifecycleParticipant, IDisposable { private readonly ILogger logger; private readonly ILogger invokeExceptionLogger; @@ -36,6 +36,7 @@ internal sealed partial class InsideRuntimeClient : IRuntimeClient, ILifecyclePa private readonly SharedCallbackData sharedCallbackData; private readonly SharedCallbackData systemSharedCallbackData; private readonly PeriodicTimer callbackTimer; + private readonly ReaderWriterLockSlim _requestAdmissionLock = new(LockRecursionPolicy.SupportsRecursion); private int _isStopping; private GrainLocator grainLocator = null!; @@ -176,9 +177,15 @@ public void SendRequest( sharedData = this.sharedCallbackData; } + var responseTimeout = request.GetDefaultResponseTimeout() ?? sharedData.ResponseTimeout; + if (targetGrainId.IsClient()) + { + message.SetGatewayRequestTimeout(responseTimeout); + } + if (this.messagingOptions.DropExpiredMessages && message.IsExpirableMessage()) { - message.TimeToLive = request.GetDefaultResponseTimeout() ?? sharedData.ResponseTimeout; + message.TimeToLive = responseTimeout; } var oneWay = (options & InvokeMethodOptions.OneWay) != 0; @@ -207,16 +214,32 @@ public void SendRequest( } } - // Completing callbacks during shutdown can resume application code which issues follow-up - // calls. Reject those calls so that they cannot outlive the shutdown callback sweep. - if (Volatile.Read(ref _isStopping) != 0) + var rejectForShutdown = false; + _requestAdmissionLock.EnterReadLock(); + try { - callbackData?.OnHostShutdown(); - return; + // Completing callbacks during shutdown can resume application code which issues follow-up + // calls. Admission and sending are serialized with the stopping transition so that no + // request, including a one-way request, can be sent after shutdown begins. + if (Volatile.Read(ref _isStopping) != 0) + { + rejectForShutdown = true; + } + else + { + this.messagingTrace.OnSendRequest(message); + this.MessageCenter.AddressAndSendMessage(message); + } + } + finally + { + _requestAdmissionLock.ExitReadLock(); } - this.messagingTrace.OnSendRequest(message); - this.MessageCenter.AddressAndSendMessage(message); + if (rejectForShutdown) + { + callbackData?.OnHostShutdown(); + } } public void SendResponse(Message request, Response response) @@ -472,7 +495,7 @@ private void ProcessResponseCallback(Message message) { // IMPORTANT: we do not schedule the response callback via the scheduler, since the only thing it does // is to resolve/break the resolver. The continuations/waits that are based on this resolution will be scheduled as work items. - callbackData.DoCallback(message); + callbackData.OnResponse(message); } else { @@ -549,8 +572,7 @@ public void DeleteObjectReference(IAddressable obj) private async Task OnRuntimeInitializeStop(CancellationToken tc) { - Volatile.Write(ref _isStopping, 1); - this.callbackTimer.Dispose(); + StopRequestAdmission(); // Once the silo is shutting down it can no longer receive responses, so any requests which // are still outstanding will never complete. Fault them now so that in-flight grain calls // observe a terminal result instead of hanging forever, which would otherwise deadlock grain @@ -618,6 +640,8 @@ public void Participate(ISiloLifecycle lifecycle) public int GetRunningRequestsCount(GrainInterfaceType grainInterfaceType) => this.callbacks.Count(c => c.Value.Message.InterfaceType == grainInterfaceType); + internal Task CallbackTimerTask => callbackTimerTask ?? Task.CompletedTask; + private async Task MonitorCallbackExpiry() { while (await callbackTimer.WaitForNextTickAsync()) @@ -645,6 +669,26 @@ private async Task MonitorCallbackExpiry() } } + public void Dispose() + { + StopRequestAdmission(); + BreakOutstandingMessages(); + } + + private void StopRequestAdmission() + { + _requestAdmissionLock.EnterWriteLock(); + try + { + Volatile.Write(ref _isStopping, 1); + this.callbackTimer.Dispose(); + } + finally + { + _requestAdmissionLock.ExitWriteLock(); + } + } + [LoggerMessage( Level = LogLevel.Warning, EventId = (int)ErrorCode.IGC_SniffIncomingMessage_Exc, diff --git a/src/Orleans.Runtime/Core/ManagementGrain.cs b/src/Orleans.Runtime/Core/ManagementGrain.cs index b83cf91bd6a..21ade2bda2b 100644 --- a/src/Orleans.Runtime/Core/ManagementGrain.cs +++ b/src/Orleans.Runtime/Core/ManagementGrain.cs @@ -447,5 +447,14 @@ private readonly struct SiloAddressesKeysLogValue(Dictionary GetSiloControlReference(s).DropDisconnectedClients(excludeRecent)); + await Task.WhenAll(actionPromises); + } } } diff --git a/src/Orleans.Runtime/Messaging/Gateway.cs b/src/Orleans.Runtime/Messaging/Gateway.cs index 00c4ed5689b..448e9d0219f 100644 --- a/src/Orleans.Runtime/Messaging/Gateway.cs +++ b/src/Orleans.Runtime/Messaging/Gateway.cs @@ -39,8 +39,11 @@ internal sealed partial class Gateway : IConnectedClientCollection private readonly SiloMessagingOptions messagingOptions; private long clientsCollectionVersion = 0; private readonly TimeSpan clientDropTimeout; + private readonly TimeProvider timeProvider; + private readonly IServiceProvider serviceProvider; public Gateway( + IServiceProvider serviceProvider, MessageCenter messageCenter, ILocalSiloDetails siloDetails, ILoggerFactory loggerFactory, @@ -50,12 +53,14 @@ public Gateway( MessagingInstruments messagingInstruments, [FromKeyedServices(TimeProviderNames.SystemTimers)] TimeProvider timeProvider) { + this.serviceProvider = serviceProvider; this.messageCenter = messageCenter; _messagingInstruments = messagingInstruments; this.messagingOptions = options.Value; this.loggerFactory = loggerFactory; this.logger = this.loggerFactory.CreateLogger(); this.clientDropTimeout = messagingOptions.ClientDropTimeout; + this.timeProvider = timeProvider; clientsReplyRoutingCache = new ClientsReplyRoutingCache(messagingOptions.ResponseTimeout); this.siloAddress = siloDetails.SiloAddress; this.gatewayAddress = siloDetails.GatewayAddress; @@ -87,6 +92,7 @@ private async Task PerformGatewayMaintenance() { try { + DropExpiredClientRequests(); DropDisconnectedClients(); DropExpiredRoutingCachedEntries(); } @@ -177,6 +183,12 @@ internal void RecordClosedConnection(GatewayInboundConnection connection) internal SiloAddress? TryToReroute(Message msg) { + if (msg.Direction == Message.Directions.Response + && msg.TryTakeTrustedGatewayResponseTarget(out var trustedResponseTarget)) + { + return trustedResponseTarget; + } + // ** Special routing rule for system target here ** // When a client make a request/response to/from a SystemTarget, the TargetSilo can be set to either // - the GatewayAddress of the target silo (for example, when the client want get the cluster typemap) @@ -211,6 +223,52 @@ internal void RecordClosedConnection(GatewayInboundConnection connection) return null; } + internal void RecordClientResponse(Message message) + { + if (message.Direction != Message.Directions.Response) + { + return; + } + + if (message.TryGetGatewayRequestOwner(out var ownerGateway, out var ownerSilo)) + { + if (!IsTargetingLocalGateway(ownerGateway)) + { + var grainFactory = serviceProvider.GetRequiredService(); + var siloControl = grainFactory.GetSystemTarget(Constants.SiloControlType, ownerSilo); + siloControl.CompleteGatewayRequest(message.SendingGrain, message.TargetGrain, message.Id).Ignore(); + message.RestoreGatewayResponseTarget(preserveRoute: true); + return; + } + + message.RestoreGatewayResponseTarget(); + } + + if (ClientGrainId.TryParse(message.SendingGrain, out var respondingClientId) + && clients.TryGetValue(respondingClientId, out var respondingClient)) + { + respondingClient.OnClientResponse(message); + } + } + + internal void RemoveTrackedClientRequest(Message message) + { + if (ClientGrainId.TryParse(message.TargetGrain, out var clientId) + && clients.TryGetValue(clientId, out var client)) + { + client.RemoveRequest(message); + } + } + + internal void CompleteTrackedClientRequest(GrainId clientId, GrainId sourceId, CorrelationId correlationId) + { + if (ClientGrainId.TryParse(clientId, out var parsedClientId) + && clients.TryGetValue(parsedClientId, out var client)) + { + client.CompleteRequest(sourceId, correlationId); + } + } + internal void DropExpiredRoutingCachedEntries() { lock (clients) @@ -219,6 +277,20 @@ internal void DropExpiredRoutingCachedEntries() } } + private void DropExpiredClientRequests() + { + foreach (var client in clients.Values) + { + client.DropExpiredRequests(); + } + } + + internal int GetOutstandingRequestCount(ClientGrainId clientId) + => clients.TryGetValue(clientId, out var client) ? client.OutstandingRequestCount : 0; + + internal IReadOnlyCollection<(GrainId GrainId, CorrelationId CorrelationId)> GetOutstandingRequestKeys(ClientGrainId clientId) + => clients.TryGetValue(clientId, out var client) ? client.OutstandingRequestKeys : []; + private bool IsTargetingLocalGateway(SiloAddress siloAddress) { // Special case if the address used by the client was loopback @@ -229,17 +301,17 @@ private bool IsTargetingLocalGateway(SiloAddress siloAddress) } // There is NO need to acquire individual ClientState lock, since we only close an older socket. - internal void DropDisconnectedClients() + internal void DropDisconnectedClients(bool excludeRecent = true) { var trackDroppedClients = GatewayEvents.IsClientDroppedEnabled(); List<(GrainId ClientId, TimeSpan DisconnectedDuration)>? droppedClients = null; foreach (var kv in clients) { - if (kv.Value.ReadyToDrop()) + if (ShouldDrop(excludeRecent, kv.Value)) { lock (clients) { - if (clients.TryGetValue(kv.Key, out var client) && client.ReadyToDrop()) + if (clients.TryGetValue(kv.Key, out var client) && ShouldDrop(excludeRecent, client)) { var disconnectedDuration = client.DisconnectedSince; LogInformationGatewayDroppingClient(logger, kv.Key, disconnectedDuration); @@ -253,13 +325,16 @@ internal void DropDisconnectedClients() droppedClients ??= []; droppedClients.Add((kv.Key.GrainId, disconnectedDuration)); } - } - clientsCollectionVersion++; - _messagingInstruments.ConnectedClient.Add(-1); + clientsCollectionVersion++; + _messagingInstruments.ConnectedClient.Add(-1); + } } } } + + static bool ShouldDrop(bool excludeRecent, ClientState client) + => excludeRecent ? client.ReadyToDrop() : !client.IsConnected; } if (droppedClients is not null) @@ -309,7 +384,9 @@ private class ClientState { private readonly Gateway _gateway; private readonly Task _messageLoop; + private readonly GatewayRequestTracker _requestTracker; private readonly ConcurrentQueue _pendingToSend = new(); + private readonly object _lifecycleLock = new(); private readonly SingleWaiterAutoResetEvent _signal = new() { RunContinuationsAsynchronously = true @@ -325,6 +402,7 @@ internal ClientState(Gateway gateway, ClientGrainId id) using var suppressExecutionContext = new ExecutionContextSuppressor(); _gateway = gateway; + _requestTracker = new(gateway.timeProvider, gateway.messagingOptions.ResponseTimeout); Id = id; _disconnectedSince.Restart(); _messageLoop = Task.Run(RunMessageLoop); @@ -340,6 +418,10 @@ internal ClientState(Gateway gateway, ClientGrainId id) public ClientGrainId Id { get; } + internal int OutstandingRequestCount => _requestTracker.Count; + + internal IReadOnlyCollection<(GrainId GrainId, CorrelationId CorrelationId)> OutstandingRequestKeys => _requestTracker.Keys; + public void RecordDisconnection() { var connection = Interlocked.Exchange(ref _connection, null); @@ -377,14 +459,31 @@ public bool ReadyToDrop() public void Drop() { - Interlocked.Exchange(ref _dropped, 1); - RejectDroppedClientMessages(); + lock (_lifecycleLock) + { + Volatile.Write(ref _dropped, 1); + } + _signal.Signal(); } public void Send(Message msg) { - _pendingToSend.Enqueue(msg); + lock (_lifecycleLock) + { + if (IsDropped) + { + _gateway.messageCenter.RejectMessage( + msg, + Message.RejectionTypes.Transient, + exc: new ClientNotAvailableException(Id.GrainId), + rejectInfo: "Client dropped"); + return; + } + + _pendingToSend.Enqueue(msg); + } + _signal.Signal(); LogTraceQueuedMessage(_gateway.logger, msg, msg.TargetGrain); } @@ -400,7 +499,7 @@ private async Task RunMessageLoop() if (IsDropped) { RejectDroppedClientMessages(); - continue; + return; } var connection = Volatile.Read(ref _connection); @@ -409,18 +508,28 @@ private async Task RunMessageLoop() continue; } - // Send all pending messages. while (_pendingToSend.TryDequeue(out var message)) { + var isRequest = message.Direction == Message.Directions.Request; + if (isRequest) + { + message.SetGatewayRequestOwner(_gateway.gatewayAddress, _gateway.siloAddress); + _requestTracker.Register(message); + } + if (TrySend(connection, message)) { LogTraceSentQueuedMessage(_gateway.logger, message, Id); } else { - // Re-enqueue the message. It's ok that it is at the end of the queue: message ordering is not guaranteed. + if (isRequest) + { + _requestTracker.Remove(message); + } + _pendingToSend.Enqueue(message); - return; + break; } } } @@ -436,9 +545,40 @@ private void RejectDroppedClientMessages() ClientNotAvailableException? exception = null; while (_pendingToSend.TryDequeue(out var message)) { - exception ??= new ClientNotAvailableException(Id.GrainId); - _gateway.messageCenter.RejectMessage(message, Message.RejectionTypes.Transient, exc: exception, rejectInfo: "Client dropped"); + RejectMessage(ref exception, message); } + + foreach (var message in _requestTracker.Drain()) + { + RejectMessage(ref exception, message); + } + + void RejectMessage(ref ClientNotAvailableException? error, Message message) + { + message.RestoreGatewayRequestSource(); + error ??= new ClientNotAvailableException(Id.GrainId); + _gateway.messageCenter.RejectMessage(message, Message.RejectionTypes.Transient, exc: error, rejectInfo: "Client dropped"); + } + } + + internal void OnClientResponse(Message message) + { + _requestTracker.Complete(message); + } + + internal void DropExpiredRequests() + { + _requestTracker.RemoveExpired(); + } + + internal void RemoveRequest(Message message) + { + _requestTracker.Remove(message); + } + + internal void CompleteRequest(GrainId sourceId, CorrelationId correlationId) + { + _requestTracker.Complete(sourceId, correlationId); } private bool TrySend(GatewayInboundConnection connection, Message message) diff --git a/src/Orleans.Runtime/Messaging/GatewayRequestTracker.cs b/src/Orleans.Runtime/Messaging/GatewayRequestTracker.cs new file mode 100644 index 00000000000..9a91792ed30 --- /dev/null +++ b/src/Orleans.Runtime/Messaging/GatewayRequestTracker.cs @@ -0,0 +1,59 @@ +using System; +using System.Collections.Concurrent; +using System.Collections.Generic; +using Orleans.Serialization.Invocation; + +namespace Orleans.Runtime.Messaging; + +internal sealed class GatewayRequestTracker(TimeProvider timeProvider, TimeSpan defaultResponseTimeout) +{ + private readonly ConcurrentDictionary<(GrainId GrainId, CorrelationId CorrelationId), TrackedRequest> _requests = new(); + + internal int Count => _requests.Count; + + internal IReadOnlyCollection<(GrainId GrainId, CorrelationId CorrelationId)> Keys => [.. _requests.Keys]; + + internal void Register(Message request) + { + var timeout = request.TimeToLive + ?? request.GetGatewayRequestTimeout() + ?? (request.BodyObject as IInvokable)?.GetDefaultResponseTimeout() + ?? defaultResponseTimeout; + var deadline = timeProvider.GetTimestamp() + checked((long)Math.Ceiling(timeout.TotalSeconds * timeProvider.TimestampFrequency)); + _requests[(request.SendingGrain, request.Id)] = new(request, deadline); + } + + internal bool Complete(Message response) + => _requests.TryRemove((response.TargetGrain, response.Id), out _); + + internal bool Complete(GrainId sourceId, CorrelationId correlationId) + => _requests.TryRemove((sourceId, correlationId), out _); + + internal bool Remove(Message request) + => _requests.TryRemove((request.SendingGrain, request.Id), out _); + + internal void RemoveExpired() + { + var now = timeProvider.GetTimestamp(); + foreach (var (key, request) in _requests) + { + if (now >= request.Deadline) + { + _requests.TryRemove(key, out _); + } + } + } + + internal IEnumerable Drain() + { + foreach (var (key, request) in _requests) + { + if (_requests.TryRemove(key, out _)) + { + yield return request.Message; + } + } + } + + private readonly record struct TrackedRequest(Message Message, long Deadline); +} diff --git a/src/Orleans.Runtime/Networking/GatewayConnectionListener.cs b/src/Orleans.Runtime/Networking/GatewayConnectionListener.cs index ff8668af2f2..6c41630cdaa 100644 --- a/src/Orleans.Runtime/Networking/GatewayConnectionListener.cs +++ b/src/Orleans.Runtime/Networking/GatewayConnectionListener.cs @@ -4,7 +4,6 @@ using System.Threading.Tasks; using Microsoft.AspNetCore.Connections; using Microsoft.Extensions.DependencyInjection; -using Microsoft.Extensions.Logging; using Microsoft.Extensions.Options; using Orleans.Configuration; @@ -17,7 +16,6 @@ internal sealed class GatewayConnectionListener : ConnectionListener, ILifecycle private readonly MessageCenter messageCenter; private readonly ConnectionCommon connectionShared; private readonly ConnectionPreambleHelper connectionPreambleHelper; - private readonly ILogger logger; private readonly EndpointOptions endpointOptions; private readonly SiloConnectionOptions siloConnectionOptions; private readonly OverloadDetector overloadDetector; @@ -33,8 +31,7 @@ public GatewayConnectionListener( MessageCenter messageCenter, ConnectionManager connectionManager, ConnectionCommon connectionShared, - ConnectionPreambleHelper connectionPreambleHelper, - ILogger logger) + ConnectionPreambleHelper connectionPreambleHelper) : base(serviceProvider.GetRequiredKeyedService(ServicesKey), connectionOptions, connectionManager, connectionShared) { this.siloConnectionOptions = siloConnectionOptions.Value; @@ -44,7 +41,6 @@ public GatewayConnectionListener( this.messageCenter = messageCenter; this.connectionShared = connectionShared; this.connectionPreambleHelper = connectionPreambleHelper; - this.logger = logger; this.endpointOptions = endpointOptions.Value; } diff --git a/src/Orleans.Runtime/Networking/GatewayInboundConnection.cs b/src/Orleans.Runtime/Networking/GatewayInboundConnection.cs index de4102e102c..2b4ce3a5d7a 100644 --- a/src/Orleans.Runtime/Networking/GatewayInboundConnection.cs +++ b/src/Orleans.Runtime/Networking/GatewayInboundConnection.cs @@ -60,6 +60,17 @@ protected override void RecordMessageSend(Message msg, int numTotalBytes, int he protected override void OnReceivedMessage(Message msg) { + this.gateway.RecordClientResponse(msg); + ProcessReceivedMessage(msg); + } + + private void ProcessReceivedMessage(Message msg) + { + if (msg.Direction is Message.Directions.Request or Message.Directions.OneWay) + { + msg.ClearGatewayRequestRouting(); + } + // Don't process messages that have already timed out if (msg.IsExpired) { @@ -164,13 +175,15 @@ public void FailMessage(Message msg, string reason) if (msg.Direction == Message.Directions.Request) { LogSiloRejectingMessage(this.Log, this.myAddress, msg, reason); + this.gateway.RemoveTrackedClientRequest(msg); + msg.RestoreGatewayRequestSource(); // Done retrying, send back an error instead this.messageCenter.SendRejection( msg, Message.RejectionTypes.Transient, - $"Silo {this.myAddress} is rejecting message: {msg}. Reason = {reason}", - new SiloUnavailableException()); + $"Target client {msg.TargetGrain} is unavailable. Message: {msg}. Reason = {reason}", + new ClientNotAvailableException($"Target client {msg.TargetGrain} is unavailable. {reason ?? "Connection terminated."}")); } else { diff --git a/src/Orleans.Runtime/Silo/SiloControl.cs b/src/Orleans.Runtime/Silo/SiloControl.cs index 141736df378..574ec573b21 100644 --- a/src/Orleans.Runtime/Silo/SiloControl.cs +++ b/src/Orleans.Runtime/Silo/SiloControl.cs @@ -12,6 +12,7 @@ using Orleans.Placement; using Orleans.Providers; using Orleans.Runtime.GrainDirectory; +using Orleans.Runtime.Messaging; using Orleans.Runtime.Placement; using Orleans.Runtime.Versions; using Orleans.Runtime.Versions.Compatibility; @@ -387,5 +388,25 @@ void ILifecycleParticipant.Participate(ISiloLifecycle lifecycle) Message = "Could not find a controllable service for type {ProviderTypeFullName} and name {ProviderName}." )] private partial void LogErrorProviderNotFound(string providerTypeFullName, string providerName); + + public Task DropDisconnectedClients(bool excludeRecent) + { + var gateway = this.services.GetRequiredService().Gateway; + if (gateway is null) + { + // No gateway deployed on this silo. + return Task.CompletedTask; + } + + gateway.DropDisconnectedClients(excludeRecent); + return Task.CompletedTask; + } + + public Task CompleteGatewayRequest(GrainId clientId, GrainId sourceId, CorrelationId correlationId) + { + this.services.GetRequiredService().Gateway?.CompleteTrackedClientRequest(clientId, sourceId, correlationId); + return Task.CompletedTask; + } + } } diff --git a/src/Orleans.TestingHost/InProcTestCluster.cs b/src/Orleans.TestingHost/InProcTestCluster.cs index 37b13162baf..e41c8445f27 100644 --- a/src/Orleans.TestingHost/InProcTestCluster.cs +++ b/src/Orleans.TestingHost/InProcTestCluster.cs @@ -1,3 +1,4 @@ +#nullable enable using System; using System.Collections.Generic; using System.Collections.ObjectModel; @@ -26,11 +27,12 @@ using Orleans.Runtime.TestHooks; using Orleans.Configuration.Internal; using Orleans.TestingHost.Logging; +using System.Diagnostics.CodeAnalysis; namespace Orleans.TestingHost; /// -/// A host class for local testing with Orleans using in-process silos. +/// A host class for local testing with Orleans using in-process silos. /// public sealed class InProcessTestCluster : IDisposable, IAsyncDisposable { @@ -40,9 +42,14 @@ public sealed class InProcessTestCluster : IDisposable, IAsyncDisposable private readonly GrainDirectoryObserver _grainDirectoryObserver = new(); private readonly InProcessGrainDirectory _grainDirectory; private readonly InProcessMembershipTable _membershipTable; - private bool _disposed; + private readonly SemaphoreSlim _clientHostsSemaphore = new(1, 1); + private readonly object _disposeLock = new(); + private volatile bool _disposed; + private Task? _disposeTask; private int _startedInstances; + private readonly Dictionary _clientHosts = new(); + /// /// Collection of all known silos. /// @@ -60,7 +67,7 @@ public ReadOnlyCollection Silos /// /// Options used to configure the test cluster. /// - /// This is the options you configured your test cluster with, or the default one. + /// This is the options you configured your test cluster with, or the default one. /// If the cluster is being configured via ClusterConfiguration, then this object may not reflect the true settings. /// public InProcessTestClusterOptions Options { get; } @@ -68,19 +75,28 @@ public ReadOnlyCollection Silos /// /// The internal client interface. /// - internal IHost? ClientHost { get; private set; } + internal IHost? ClientHost + { + get + { + lock (_clientHosts) + { + _clientHosts.TryGetValue("default", out var clientHost); + return clientHost; + } + } + } /// /// The internal client interface. /// - internal IInternalClusterClient? InternalClient => ClientHost?.Services.GetRequiredService(); + internal IInternalClusterClient InternalClient => ClientHost?.Services.GetRequiredService() ?? throw new InvalidOperationException( + "The test cluster client is unavailable because the cluster has not been deployed or has been stopped."); /// /// The client. /// - /// The cluster has not been deployed or the client has been stopped. - public IClusterClient Client => InternalClient ?? throw new InvalidOperationException( - "The test cluster client is unavailable because the cluster has not been deployed or has been stopped."); + public IClusterClient Client => InternalClient; /// /// The port allocator. @@ -120,29 +136,6 @@ public IServiceProvider GetSiloServiceProvider(SiloAddress? silo = null) } } - /// - /// Attempts to find the for the grain with the specified - /// by searching all silos in the cluster. - /// - /// The ID of the grain to find. - /// When this method returns, contains the grain context if found; otherwise, . - /// if the grain was found in one of the silos; otherwise, . - public bool TryGetGrainContext(GrainId grainId, [System.Diagnostics.CodeAnalysis.NotNullWhen(true)] out IGrainContext? grainContext) - { - foreach (var silo in Silos) - { - var activationDirectory = silo.SiloHost.Services.GetRequiredService(); - grainContext = activationDirectory.FindTarget(grainId); - if (grainContext is not null) - { - return true; - } - } - - grainContext = null; - return false; - } - /// /// Gets a that completes when the current activation of the specified grain /// finishes deactivating. If the grain is not currently activated, returns . @@ -363,8 +356,8 @@ public IEnumerable GetActiveSilos() /// Whether recent membership changes we done by graceful Stop. public async Task WaitForLivenessToStabilizeAsync(bool didKill = false) { - var clusterMembershipOptions = Client!.ServiceProvider.GetRequiredService>().Value; // Stabilization requires a deployed client. - TimeSpan stabilizationTime = GetLivenessStabilizationTime(clusterMembershipOptions, didKill); + var clusterMembershipOptions = Client.ServiceProvider.GetRequiredService>().Value; + var stabilizationTime = GetLivenessStabilizationTime(clusterMembershipOptions, didKill); var activeSilos = GetActiveSilos().ToArray(); var testHooks = activeSilos.Select(static silo => (ITestHooks)silo.ServiceProvider.GetRequiredService()).ToArray(); var gatewayManager = Client.ServiceProvider.GetRequiredService(); @@ -372,41 +365,15 @@ public async Task WaitForLivenessToStabilizeAsync(bool didKill = false) GrainDirectoryObserver.CanObserve(activeSilos) ? timeout => _grainDirectoryObserver.WaitForConvergenceAsync(activeSilos, timeout) : null; - WriteLog(Environment.NewLine + Environment.NewLine + "WaitForLivenessToStabilize is waiting up to {0} for {1} active silo(s)", stabilizationTime, activeSilos.Length); - if (await LivenessStabilizationHelper.WaitForExpectedActiveSilosAndGatewaysAsync( - activeSilos, - testHooks, - gatewayManager, - stabilizationTime, - waitForGrainDirectoryConvergence)) - { - WriteLog("WaitForLivenessToStabilize observed stable active silo and gateway views"); - } - else - { - WriteLog("WaitForLivenessToStabilize reached the fallback wait of {0}", stabilizationTime); - } - } - /// - /// Wait for active silos to observe cluster manifest updates for all active silos. - /// - /// Whether recent membership changes were done by graceful Stop. - public async Task WaitForClusterManifestToStabilizeAsync(bool didKill = false) - { - var clusterMembershipOptions = Client!.ServiceProvider.GetRequiredService>().Value; // Stabilization requires a deployed client. - var stabilizationTime = GetLivenessStabilizationTime(clusterMembershipOptions, didKill); - var activeSilos = GetActiveSilos().ToArray(); - var testHooks = activeSilos.Select(static silo => (ITestHooks)silo.ServiceProvider.GetRequiredService()).ToArray(); - - WriteLog(Environment.NewLine + Environment.NewLine + "WaitForClusterManifestToStabilize is waiting up to {0} for {1} active silo manifest(s)", stabilizationTime, activeSilos.Length); - if (await ClusterManifestStabilizationHelper.WaitForExpectedClusterManifestAsync(activeSilos, testHooks, stabilizationTime)) - { - WriteLog("WaitForClusterManifestToStabilize observed stable cluster manifests"); - } - else + if (!await LivenessStabilizationHelper.WaitForExpectedActiveSilosAndGatewaysAsync( + activeSilos, + testHooks, + gatewayManager, + stabilizationTime, + waitForGrainDirectoryConvergence)) { - WriteLog("WaitForClusterManifestToStabilize reached the fallback wait of {0}", stabilizationTime); + WriteLog("WaitForLivenessToStabilize reached the fallback wait of {0}", stabilizationTime); } } @@ -453,6 +420,46 @@ internal async Task WaitForTopologyToConvergeAsync( } } + /// + /// Waits for active silos to observe cluster manifest updates for all active silos. + /// + public async Task WaitForClusterManifestToStabilizeAsync(bool didKill = false) + { + var clusterMembershipOptions = Client.ServiceProvider.GetRequiredService>().Value; + var stabilizationTime = GetLivenessStabilizationTime(clusterMembershipOptions, didKill); + var activeSilos = GetActiveSilos().ToArray(); + var testHooks = activeSilos.Select(static silo => (ITestHooks)silo.ServiceProvider.GetRequiredService()).ToArray(); + + WriteLog(Environment.NewLine + Environment.NewLine + "WaitForClusterManifestToStabilize is waiting up to {0} for {1} active silo manifest(s)", stabilizationTime, activeSilos.Length); + if (await ClusterManifestStabilizationHelper.WaitForExpectedClusterManifestAsync(activeSilos, testHooks, stabilizationTime)) + { + WriteLog("WaitForClusterManifestToStabilize observed stable cluster manifests"); + } + else + { + WriteLog("WaitForClusterManifestToStabilize reached the fallback wait of {0}", stabilizationTime); + } + } + + /// + /// Attempts to find a grain context by searching all silos. + /// + public bool TryGetGrainContext(GrainId grainId, [NotNullWhen(true)] out IGrainContext? grainContext) + { + foreach (var silo in Silos) + { + var activationDirectory = silo.SiloHost.Services.GetRequiredService(); + grainContext = activationDirectory.FindTarget(grainId); + if (grainContext is not null) + { + return true; + } + } + + grainContext = null; + return false; + } + /// /// Get the timeout value to use to wait for the silo liveness sub-system to detect and act on any recent cluster membership changes. /// @@ -614,13 +621,9 @@ public async Task StopSilosAsync(CancellationToken cancellationToken) /// A representing the asynchronous operation. public async Task StopClusterClientAsync(CancellationToken cancellationToken) { - var client = ClientHost; try { - if (client is not null) - { - await client.StopAsync(cancellationToken).ConfigureAwait(false); - } + await RemoveClientAsync("default", cancellationToken); } catch (Exception exc) { @@ -630,11 +633,6 @@ public async Task StopClusterClientAsync(CancellationToken cancellationToken) throw; } } - finally - { - await DisposeAsync(client).ConfigureAwait(false); - ClientHost = null; - } } /// @@ -732,21 +730,35 @@ public async Task KillSiloAsync(InProcessSiloHandle instance, CancellationToken /// public async Task KillClientAsync() { - var client = ClientHost; - if (client != null) + await _clientHostsSemaphore.WaitAsync(); + try { - var cancelled = new CancellationTokenSource(); - cancelled.Cancel(); - try + ThrowIfDisposed(); + + IHost? client; + lock (_clientHosts) { - await client.StopAsync(cancelled.Token).ConfigureAwait(false); + _clientHosts.Remove("default", out client); } - finally + + if (client is not null) { - await DisposeAsync(client); - ClientHost = null; + using var cancelled = new CancellationTokenSource(); + cancelled.Cancel(); + try + { + await client.StopAsync(cancelled.Token).ConfigureAwait(false); + } + finally + { + await DisposeAsync(client); + } } } + finally + { + _clientHostsSemaphore.Release(); + } } /// @@ -755,7 +767,7 @@ public async Task KillClientAsync() /// Silo to be restarted. public async Task RestartSiloAsync(InProcessSiloHandle instance) { - if (instance is not null) + if (instance != null) { var instanceNumber = instance.InstanceNumber; await StopSiloAsync(instance); @@ -805,10 +817,15 @@ public async Task InitializeClientAsync(CancellationToken cancellationToken) await StopClusterClientAsync(cancellationToken); } + await GetClientAsync("default"); + } + + private IHost CreateClientHost(string name, Action? configure = null) + { var hostBuilder = Host.CreateApplicationBuilder(new HostApplicationBuilderSettings { EnvironmentName = Environments.Development, - ApplicationName = "TestClusterClient", + ApplicationName = $"Client_{name}", DisableDefaults = true, }); @@ -816,6 +833,7 @@ public async Task InitializeClientAsync(CancellationToken cancellationToken) { hostDelegate(hostBuilder); } + configure?.Invoke(hostBuilder); hostBuilder.UseOrleansClient(clientBuilder => { @@ -833,18 +851,103 @@ public async Task InitializeClientAsync(CancellationToken cancellationToken) clientBuilder.UseInMemoryConnectionTransport(_transportHub); }); - TryConfigureFileLogging(Options, hostBuilder.Services, "TestClusterClient"); + TryConfigureFileLogging(Options, hostBuilder.Services, $"Client_{name}"); var clientHost = hostBuilder.Build(); + return clientHost; + } + + /// + /// Gets a client by name, or null if not found. + /// + public IClusterClient? GetClient(string name) + { + lock (_clientHosts) + { + ThrowIfDisposed(); + + return _clientHosts.TryGetValue(name, out var host) + ? host.Services.GetRequiredService() + : null; + } + } + + /// + /// Gets the client with the given name, creating a new client if none exists with that name. + /// + public async Task GetClientAsync(string name, Action? configure = null) + { + await _clientHostsSemaphore.WaitAsync(); try { - await clientHost.StartAsync(cancellationToken); - ClientHost = clientHost; + ThrowIfDisposed(); + + IHost? host; + lock (_clientHosts) + { + _clientHosts.TryGetValue(name, out host); + } + + if (host is null) + { + host = CreateClientHost(name, configure); + try + { + await host.StartAsync(); + lock (_clientHosts) + { + ThrowIfDisposed(); + _clientHosts.Add(name, host); + } + } + catch + { + await DisposeAsync(host); + throw; + } + } + + return host.Services.GetRequiredService(); + } + finally + { + _clientHostsSemaphore.Release(); } - catch + } + + /// + /// Removes and disposes a client by name. + /// + public Task RemoveClientAsync(string name) => RemoveClientAsync(name, CancellationToken.None); + + private async Task RemoveClientAsync(string name, CancellationToken cancellationToken) + { + await _clientHostsSemaphore.WaitAsync(cancellationToken); + try { - await DisposeAsync(clientHost); - throw; + ThrowIfDisposed(); + + IHost? host; + lock (_clientHosts) + { + _clientHosts.Remove(name, out host); + } + + if (host is not null) + { + try + { + await host.StopAsync(cancellationToken); + } + finally + { + await DisposeAsync(host); + } + } + } + finally + { + _clientHostsSemaphore.Release(); } } @@ -1076,13 +1179,13 @@ public string GetLog() return _log.ToString(); } - private void ReportUnobservedException(object? sender, UnhandledExceptionEventArgs eventArgs) + private void ReportUnobservedException(object sender, UnhandledExceptionEventArgs eventArgs) { Exception exception = (Exception)eventArgs.ExceptionObject; WriteLog("Unobserved exception: {0}", exception); } - private void WriteLog(string format, params object?[] args) + private void WriteLog(string format, params object[] args) { _log.AppendFormat(format + Environment.NewLine, args); } @@ -1093,50 +1196,103 @@ private void FlushLogToConsole() } /// - public async ValueTask DisposeAsync() + public ValueTask DisposeAsync() => new(StartDisposal()); + + /// + public void Dispose() => StartDisposal().GetAwaiter().GetResult(); + + private Task StartDisposal() { - if (_disposed) + lock (_disposeLock) { - return; + if (_disposeTask is not null) + { + return _disposeTask; + } + + _disposed = true; + _disposeTask = Task.Run(DisposeCoreAsync); + return _disposeTask; } + } - await Task.Run(async () => + private async Task DisposeCoreAsync() + { + await _clientHostsSemaphore.WaitAsync().ConfigureAwait(false); + List clientHosts; + try { - AppDomain.CurrentDomain.UnhandledException -= ReportUnobservedException; - foreach (var handle in Silos) + lock (_clientHosts) + { + clientHosts = [.. _clientHosts.Values]; + _clientHosts.Clear(); + } + } + finally + { + _clientHostsSemaphore.Release(); + } + + List? exceptions = null; + AppDomain.CurrentDomain.UnhandledException -= ReportUnobservedException; + foreach (var handle in Silos) + { + try { await DisposeAsync(handle).ConfigureAwait(false); } + catch (Exception exception) + { + exceptions ??= []; + exceptions.Add(exception); + } + } + + foreach (var clientHost in clientHosts) + { + try + { + await DisposeAsync(clientHost).ConfigureAwait(false); + } + catch (Exception exception) + { + exceptions ??= []; + exceptions.Add(exception); + } + } - await DisposeAsync(ClientHost).ConfigureAwait(false); - ClientHost = null; + try + { + PortAllocator.Dispose(); + } + catch (Exception exception) + { + exceptions ??= []; + exceptions.Add(exception); + } - PortAllocator?.Dispose(); + try + { _grainDirectoryObserver.Dispose(); - }); + } + catch (Exception exception) + { + exceptions ??= []; + exceptions.Add(exception); + } - _disposed = true; + if (exceptions is not null) + { + throw new AggregateException("One or more errors occurred while disposing the test cluster.", exceptions); + } } - /// - public void Dispose() + private void ThrowIfDisposed() { if (_disposed) { - return; + throw new ObjectDisposedException(nameof(InProcessTestCluster)); } - - foreach (var handle in Silos) - { - handle.Dispose(); - } - - ClientHost?.Dispose(); - ClientHost = null; - PortAllocator?.Dispose(); - _grainDirectoryObserver.Dispose(); - - _disposed = true; } private static async Task DisposeAsync(IDisposable? value) diff --git a/src/api/Orleans.Core/Orleans.Core.cs b/src/api/Orleans.Core/Orleans.Core.cs index c8bdc417e50..b790c96088b 100644 --- a/src/api/Orleans.Core/Orleans.Core.cs +++ b/src/api/Orleans.Core/Orleans.Core.cs @@ -1406,6 +1406,7 @@ public partial interface ILocalSiloDetails public partial interface IManagementGrain : IGrainWithIntegerKey, IGrain, IAddressable, IVersionManager { + System.Threading.Tasks.Task DropDisconnectedClients(bool excludeRecent); System.Threading.Tasks.Task ForceActivationCollection(SiloAddress[] hostsIds, System.TimeSpan ageLimit); System.Threading.Tasks.Task ForceActivationCollection(System.TimeSpan ageLimit); System.Threading.Tasks.Task ForceGarbageCollection(SiloAddress[] hostsIds); @@ -3508,6 +3509,22 @@ public void WriteField(ref global::Orleans.Serialization.Buffers. where TBufferWriter : System.Buffers.IBufferWriter { } } + [System.CodeDom.Compiler.GeneratedCode("OrleansCodeGen", "10.0.0.0")] + [System.ComponentModel.EditorBrowsable(System.ComponentModel.EditorBrowsableState.Never)] + [System.Diagnostics.CodeAnalysis.ExcludeFromCodeCoverage] + public sealed partial class Codec_Invokable_IManagementGrain_GrainReference_101564A8 : global::Orleans.Serialization.Codecs.IFieldCodec, global::Orleans.Serialization.Codecs.IFieldCodec + { + public void Deserialize(ref global::Orleans.Serialization.Buffers.Reader reader, Invokable_IManagementGrain_GrainReference_101564A8 instance) { } + + public Invokable_IManagementGrain_GrainReference_101564A8 ReadValue(ref global::Orleans.Serialization.Buffers.Reader reader, global::Orleans.Serialization.WireProtocol.Field field) { throw null; } + + public void Serialize(ref global::Orleans.Serialization.Buffers.Writer writer, Invokable_IManagementGrain_GrainReference_101564A8 instance) + where TBufferWriter : System.Buffers.IBufferWriter { } + + public void WriteField(ref global::Orleans.Serialization.Buffers.Writer writer, uint fieldIdDelta, System.Type expectedType, Invokable_IManagementGrain_GrainReference_101564A8 value) + where TBufferWriter : System.Buffers.IBufferWriter { } + } + [System.CodeDom.Compiler.GeneratedCode("OrleansCodeGen", "10.0.0.0")] [System.ComponentModel.EditorBrowsable(System.ComponentModel.EditorBrowsableState.Never)] [System.Diagnostics.CodeAnalysis.ExcludeFromCodeCoverage] @@ -3843,6 +3860,14 @@ public Copier_Invokable_IManagementGrain_GrainReference_0F06E027(global::Orleans public Invokable_IManagementGrain_GrainReference_0F06E027 DeepCopy(Invokable_IManagementGrain_GrainReference_0F06E027 original, global::Orleans.Serialization.Cloning.CopyContext context) { throw null; } } + [System.CodeDom.Compiler.GeneratedCode("OrleansCodeGen", "10.0.0.0")] + [System.ComponentModel.EditorBrowsable(System.ComponentModel.EditorBrowsableState.Never)] + [System.Diagnostics.CodeAnalysis.ExcludeFromCodeCoverage] + public sealed partial class Copier_Invokable_IManagementGrain_GrainReference_101564A8 : global::Orleans.Serialization.Cloning.IDeepCopier, global::Orleans.Serialization.Cloning.IDeepCopier + { + public Invokable_IManagementGrain_GrainReference_101564A8 DeepCopy(Invokable_IManagementGrain_GrainReference_101564A8 original, global::Orleans.Serialization.Cloning.CopyContext context) { throw null; } + } + [System.CodeDom.Compiler.GeneratedCode("OrleansCodeGen", "10.0.0.0")] [System.ComponentModel.EditorBrowsable(System.ComponentModel.EditorBrowsableState.Never)] [System.Diagnostics.CodeAnalysis.ExcludeFromCodeCoverage] @@ -4044,6 +4069,38 @@ public override void SetArgument(int index, object value) { } public override void SetTarget(global::Orleans.Serialization.Invocation.ITargetHolder holder) { } } + [System.CodeDom.Compiler.GeneratedCode("OrleansCodeGen", "10.0.0.0")] + [System.ComponentModel.EditorBrowsable(System.ComponentModel.EditorBrowsableState.Never)] + [System.Diagnostics.CodeAnalysis.ExcludeFromCodeCoverage] + [global::Orleans.CompoundTypeAlias(new[] { "inv", typeof(global::Orleans.Runtime.GrainReference), typeof(global::Orleans.Runtime.IManagementGrain), "101564A8" })] + public sealed partial class Invokable_IManagementGrain_GrainReference_101564A8 : global::Orleans.Runtime.TaskRequest + { + public bool arg0; + public override void Dispose() { } + + public override string GetActivityName() { throw null; } + + public override object GetArgument(int index) { throw null; } + + public override int GetArgumentCount() { throw null; } + + public override string GetInterfaceName() { throw null; } + + public override System.Type GetInterfaceType() { throw null; } + + public override System.Reflection.MethodInfo GetMethod() { throw null; } + + public override string GetMethodName() { throw null; } + + public override object GetTarget() { throw null; } + + protected override System.Threading.Tasks.Task InvokeInner() { throw null; } + + public override void SetArgument(int index, object value) { } + + public override void SetTarget(global::Orleans.Serialization.Invocation.ITargetHolder holder) { } + } + [System.CodeDom.Compiler.GeneratedCode("OrleansCodeGen", "10.0.0.0")] [System.ComponentModel.EditorBrowsable(System.ComponentModel.EditorBrowsableState.Never)] [System.Diagnostics.CodeAnalysis.ExcludeFromCodeCoverage] diff --git a/src/api/Orleans.TestingHost/Orleans.TestingHost.cs b/src/api/Orleans.TestingHost/Orleans.TestingHost.cs index dcc1157d009..5168587a5d7 100644 --- a/src/api/Orleans.TestingHost/Orleans.TestingHost.cs +++ b/src/api/Orleans.TestingHost/Orleans.TestingHost.cs @@ -110,6 +110,10 @@ public void Dispose() { } public System.Collections.Generic.IEnumerable GetActiveSilos() { throw null; } + public IClusterClient? GetClient(string name) { throw null; } + + public System.Threading.Tasks.Task GetClientAsync(string name, System.Action? configure = null) { throw null; } + public static System.TimeSpan GetLivenessStabilizationTime(Configuration.ClusterMembershipOptions clusterMembershipOptions, bool didKill = false) { throw null; } public string GetLog() { throw null; } @@ -132,6 +136,8 @@ public void Dispose() { } public System.Threading.Tasks.Task MigrateAsync(Runtime.IAddressable grain, Runtime.SiloAddress? targetSilo = null) { throw null; } + public System.Threading.Tasks.Task RemoveClientAsync(string name) { throw null; } + public System.Threading.Tasks.Task RestartSiloAsync(InProcessSiloHandle instance) { throw null; } public System.Threading.Tasks.Task RestartStoppedSecondarySiloAsync(string siloName) { throw null; } diff --git a/test/Orleans.Runtime.Internal.Tests/GatewayRequestTrackerTests.cs b/test/Orleans.Runtime.Internal.Tests/GatewayRequestTrackerTests.cs new file mode 100644 index 00000000000..544bd8c55d5 --- /dev/null +++ b/test/Orleans.Runtime.Internal.Tests/GatewayRequestTrackerTests.cs @@ -0,0 +1,96 @@ +using System; +using System.Net; +using Orleans.Runtime; +using Orleans.Runtime.Messaging; +using Xunit; + +namespace UnitTests; + +[TestSuite("BVT")] +[TestProvider("None")] +[TestArea("Runtime")] +public class GatewayRequestTrackerTests +{ + [Fact] + public void DropExpiredMessagesFalse_RequestIsRemovedAtExplicitDeadline() + { + var timeProvider = new ManualTimeProvider(); + var responseTimeout = TimeSpan.FromSeconds(30); + var tracker = new GatewayRequestTracker(timeProvider, TimeSpan.FromMinutes(5)); + var request = CreateRequest(); + request.SetGatewayRequestTimeout(responseTimeout); + + Assert.Null(request.TimeToLive); + tracker.Register(request); + Assert.Equal(1, tracker.Count); + + timeProvider.Advance(responseTimeout); + tracker.RemoveExpired(); + + Assert.Equal(0, tracker.Count); + Assert.Null(request.TimeToLive); + } + + [Fact] + public void ResponseBucketedToDifferentGateway_CompletesOwningGatewayWithoutLaterRejection() + { + var ownerGateway = CreateSiloAddress(11111); + var bucketGateway = CreateSiloAddress(22222); + var spoofedGateway = CreateSiloAddress(33333); + var tracker = new GatewayRequestTracker(TimeProvider.System, TimeSpan.FromSeconds(30)); + var request = CreateRequest(); + request.SendingSilo = bucketGateway; + request.RequestContextData = new() + { + ["#orleans.gateway.request-owner"] = spoofedGateway, + ["#orleans.gateway.request-owner-silo"] = spoofedGateway, + ["#orleans.gateway.response-target"] = spoofedGateway, + }; + request.ClearGatewayRequestOwner(); + Assert.Null(request.GetGatewayRequestTimeout()); + request.SetGatewayRequestOwner(ownerGateway, CreateSiloAddress(44444)); + Assert.Equal(ownerGateway, request.SendingSilo); + tracker.Register(request); + + var response = new Message + { + Direction = Message.Directions.Response, + Id = request.Id, + SendingGrain = request.TargetGrain, + TargetGrain = request.SendingGrain, + TargetSilo = bucketGateway, + }; + + response.ApplyGatewayRequestOwner(request); + Assert.Equal(ownerGateway, response.TargetSilo); + Assert.True(response.TryGetGatewayRequestOwner(out var restoredOwner, out _)); + Assert.Equal(ownerGateway, restoredOwner); + response.RestoreGatewayResponseTarget(); + Assert.Equal(bucketGateway, response.TargetSilo); + Assert.True(tracker.Complete(response)); + Assert.Equal(0, tracker.Count); + Assert.Empty(tracker.Drain()); + } + + private static Message CreateRequest() => new() + { + Direction = Message.Directions.Request, + Id = new CorrelationId(1234), + SendingGrain = GrainId.Create("source", "1"), + TargetGrain = GrainId.Create("client", "2"), + }; + + private static SiloAddress CreateSiloAddress(int port) + => SiloAddress.New(new IPEndPoint(IPAddress.Loopback, port), 1); + + private sealed class ManualTimeProvider : TimeProvider + { + private long _timestamp; + + public override long TimestampFrequency => TimeSpan.TicksPerSecond; + + public override long GetTimestamp() => _timestamp; + + public void Advance(TimeSpan duration) => _timestamp += duration.Ticks; + } +} diff --git a/test/Orleans.Runtime.Internal.Tests/InsideRuntimeClientDisposalTests.cs b/test/Orleans.Runtime.Internal.Tests/InsideRuntimeClientDisposalTests.cs new file mode 100644 index 00000000000..9bcf3e86867 --- /dev/null +++ b/test/Orleans.Runtime.Internal.Tests/InsideRuntimeClientDisposalTests.cs @@ -0,0 +1,85 @@ +using System; +using System.Threading; +using System.Threading.Tasks; +using Microsoft.Extensions.DependencyInjection; +using Orleans.Metadata; +using Orleans.Runtime; +using Orleans.TestingHost; +using UnitTests.GrainInterfaces; +using Xunit; + +namespace UnitTests; + +[TestSuite("BVT")] +[TestProvider("None")] +[TestArea("Runtime")] +public class InsideRuntimeClientDisposalTests +{ + [Fact] + public async Task Dispose_RacingWithNewRequests_StopsTimerAndCompletesAllCallbacks() + { + var cancellationToken = TestContext.Current.CancellationToken; + await using var cluster = new InProcessTestClusterBuilder(1).Build(); + await cluster.DeployAsync(cancellationToken); + + var services = cluster.Silos[0].ServiceProvider; + var runtimeClient = services.GetRequiredService(); + var grainFactory = services.GetRequiredService(); + var typeResolver = services.GetRequiredService(); + var interfaceType = typeResolver.GetGrainInterfaceType(typeof(ILongRunningTaskGrain)); + var grainId = Guid.NewGuid(); + var grain = grainFactory.GetGrain>(grainId); + var pendingCall = grain.LongRunningTask(1, TimeSpan.FromSeconds(1)); + + await WaitUntilAsync( + () => runtimeClient.GetRunningRequestsCount(interfaceType) == 1, + TimeSpan.FromSeconds(10), + cancellationToken); + + using var start = new ManualResetEventSlim(); + var disposeTask = Task.Run(() => + { + start.Wait(); + runtimeClient.Dispose(); + }, cancellationToken); + var racingCall = Task.Run(async () => + { + start.Wait(); + await grain.LongRunningTask(2, TimeSpan.Zero); + }, cancellationToken); + + start.Set(); + await disposeTask.WaitAsync(TimeSpan.FromSeconds(10), cancellationToken); + + await Assert.ThrowsAsync(() => pendingCall); + try + { + await racingCall.WaitAsync(TimeSpan.FromSeconds(10), cancellationToken); + } + catch (SiloUnavailableException) + { + // The call lost the admission race and was rejected by the stopping runtime. + } + await runtimeClient.CallbackTimerTask.WaitAsync(TimeSpan.FromSeconds(10), cancellationToken); + Assert.Equal(0, runtimeClient.GetRunningRequestsCount(interfaceType)); + await Assert.ThrowsAsync(() => grain.LongRunningTask(3, TimeSpan.Zero)); + + await Task.Delay(TimeSpan.FromSeconds(2), cancellationToken); + var externalGrain = cluster.Client.GetGrain>(grainId); + Assert.NotEqual(3, await externalGrain.GetLastValue()); + } + + private static async Task WaitUntilAsync(Func condition, TimeSpan timeout, CancellationToken cancellationToken) + { + var deadline = DateTime.UtcNow + timeout; + while (!condition()) + { + if (DateTime.UtcNow >= deadline) + { + throw new TimeoutException("The expected runtime-client state was not reached."); + } + + await Task.Delay(10, cancellationToken); + } + } +} diff --git a/test/Orleans.Runtime.Internal.Tests/TimeoutTests.cs b/test/Orleans.Runtime.Internal.Tests/TimeoutTests.cs index 0df04248e0d..869974c3b0c 100644 --- a/test/Orleans.Runtime.Internal.Tests/TimeoutTests.cs +++ b/test/Orleans.Runtime.Internal.Tests/TimeoutTests.cs @@ -9,23 +9,8 @@ namespace UnitTests { - /// - /// Tests for Orleans timeout mechanisms and request cancellation. - /// - /// Orleans implements timeouts to prevent indefinite waiting on grain calls: - /// - Each grain call has a configurable response timeout - /// - If a grain method doesn't complete within the timeout, a TimeoutException is thrown - /// - The original request continues executing on the silo (not cancelled) - /// - Subsequent calls to a busy grain may be dropped to prevent queue buildup - /// - /// These tests verify: - /// - Timeout exceptions are thrown at the appropriate time - /// - Request tracking is properly cleaned up after timeouts - /// - Call dropping behavior for overloaded grains - /// - /// Note: These tests modify global timeout settings, so they should run in isolation. - /// - [TestArea("Runtime")] + // If we parallelize tests, this should run in isolation. + [TestArea("Runtime")] public class TimeoutTests : HostedTestClusterEnsureDefaultStarted, IDisposable { private readonly ITestOutputHelper output; @@ -37,25 +22,15 @@ public TimeoutTests(ITestOutputHelper output, DefaultClusterFixture fixture) : b { this.output = output; this.runtimeClient = this.HostedCluster.ServiceProvider.GetRequiredService(); - // Save original timeout to restore it after tests originalTimeout = this.runtimeClient.GetResponseTimeout(); this.typeResolver = this.HostedCluster.ServiceProvider.GetRequiredService(); } public virtual void Dispose() { - // Restore original timeout to avoid affecting other tests this.runtimeClient.SetResponseTimeout(originalTimeout); } - /// - /// Tests that grain calls timeout correctly when the method takes longer than the response timeout. - /// Verifies: - /// - TimeoutException is thrown after the configured timeout period - /// - The timeout occurs within expected bounds (not too early, not too late) - /// - Request tracking is cleaned up (no lingering requests) - /// - Re-awaiting the same task fails immediately - /// [TestSuite("Functional")] [TestProvider("None")] [Fact, TestCategory("Functional"), TestCategory("Timeout")] @@ -66,16 +41,13 @@ public async Task Timeout_LongMethod() var grainName = typeof (ErrorGrain).FullName; IErrorGrain grain = this.GrainFactory.GetGrain(GetRandomGrainId(), grainName); var errorGrainType = this.typeResolver.GetGrainInterfaceType(typeof(IErrorGrain)); - // Set a 1-second timeout for this test TimeSpan timeout = TimeSpan.FromMilliseconds(1000); this.runtimeClient.SetResponseTimeout(timeout); - // Call a method that takes 4x longer than the timeout Task promise = grain.LongMethod((int)timeout.Multiply(4).TotalMilliseconds); //promise = grain.LongMethodWithError(2000); - // Note: There's a potential race condition in debugger where the call might complete - // Measure how long we wait for the timeout + // there is a race in the test here. If run in debugger, the invocation can actually finish OK Stopwatch stopwatch = new Stopwatch(); stopwatch.Start(); try @@ -99,15 +71,13 @@ public async Task Timeout_LongMethod() } output.WriteLine("Waited for " + stopwatch.Elapsed); Assert.True(!finished); - // Verify timeout occurred within expected bounds (90% to 350% of configured timeout) Assert.True(stopwatch.Elapsed >= timeout.Multiply(0.9), "Waited less than " + timeout.Multiply(0.9) + ". Waited " + stopwatch.Elapsed); Assert.True(stopwatch.Elapsed <= timeout.Multiply(3.5), "Waited longer than " + timeout.Multiply(3.5) + ". Waited " + stopwatch.Elapsed); Assert.True(promise.Status == TaskStatus.Faulted); - // Verify request tracking is cleaned up - no requests should be pending Assert.Equal(expected: 0, actual: this.runtimeClient.GetRunningRequestsCount(errorGrainType)); - // Re-awaiting a timed-out task should fail immediately + // try to re-use the promise and should fail immediately. try { stopwatch = new Stopwatch(); @@ -150,7 +120,7 @@ public async Task CallThatShouldHaveBeenDroppedNotExecutedTest() var target = Client.GetGrain>(Guid.NewGuid()); - // First call: Takes 5 seconds but client times out after 2 seconds + // First call should be successful, but client will not receive the response var delay = TimeSpan.FromSeconds(5); var firstCall = target.LongRunningTask(1, responseTimeout + delay); await Task.Delay(TimeSpan.FromMilliseconds(100), cancellationToken); @@ -172,7 +142,6 @@ public async Task CallThatShouldHaveBeenDroppedNotExecutedTest() // Wait for first call to complete on the silo await Task.Delay(delay, cancellationToken); - // Verify only the first call executed (value = 1), second call was dropped Assert.Equal(1, await target.GetLastValue()); } } diff --git a/test/Orleans.Runtime.Tests/ClientConnectionTests/ClientDisconnectionTests.cs b/test/Orleans.Runtime.Tests/ClientConnectionTests/ClientDisconnectionTests.cs new file mode 100644 index 00000000000..a2ff5e59605 --- /dev/null +++ b/test/Orleans.Runtime.Tests/ClientConnectionTests/ClientDisconnectionTests.cs @@ -0,0 +1,290 @@ +#nullable enable +using System; +using System.Threading.Tasks; +using Microsoft.Extensions.DependencyInjection; +using Orleans.GrainReferences; +using Orleans.Runtime; +using Orleans.Runtime.Messaging; +using Orleans.TestingHost; +using UnitTests.GrainInterfaces; +using Xunit; + +namespace Tester.ClientConnectionTests; + +[TestCategory("BVT"), TestCategory("MultiClient"), TestCategory("Lifecycle")] +[TestSuite("BVT")] +[TestProvider("None")] +[TestArea("Runtime")] +public class ClientDisconnectionTests(ClientDisconnectionTests.Fixture fixture) : IClassFixture +{ + private readonly InProcessTestCluster _cluster = fixture.Cluster; + + public sealed class Fixture : IAsyncLifetime + { + private InProcessTestCluster? _cluster; + public InProcessTestCluster Cluster => _cluster!; + + public async ValueTask InitializeAsync() + { + var builder = new InProcessTestClusterBuilder(2); + _cluster = builder.Build(); + await _cluster.DeployAsync(); + } + + public async ValueTask DisposeAsync() + { + if (_cluster != null) + { + await _cluster.DisposeAsync(); + } + } + + } + + [Fact] + public async Task ResponseAcrossMultipleGateways_ClearsOwningGatewayBeforeClientDrop() + { + var cancellationToken = TestContext.Current.CancellationToken; + var clientA = await _cluster.GetClientAsync("OwnerClientA"); + var clientB = await _cluster.GetClientAsync("OwnerClientB"); + var observerB = new EchoGrainObserver(); + var observerBReference = clientB.CreateObjectReference(observerB); + observerB.SelfReference = observerBReference; + var observerBId = observerBReference.GetGrainId(); + var aToB = (IEchoGrainObserver)clientA.ServiceProvider.GetRequiredService().CreateReference( + observerBId, + GrainInterfaceType.Create("IEchoGrainObserver")); + var responseTask = aToB.EchoAsync("owner-routed response"); + + await observerB.WaitForCallAsync().WaitAsync(TimeSpan.FromSeconds(10), cancellationToken); + Assert.True(ClientGrainId.TryParse(observerBId, out var clientBId)); + var gateways = _cluster.Silos + .Select(static silo => silo.ServiceProvider.GetRequiredService().Gateway!) + .ToArray(); + await WaitUntilAsync( + () => gateways.Sum(gateway => gateway.GetOutstandingRequestCount(clientBId)) == 1, + TimeSpan.FromSeconds(10)); + + var ownerIndex = Array.FindIndex(gateways, gateway => gateway.GetOutstandingRequestCount(clientBId) == 1); + var ownerDetails = _cluster.Silos[ownerIndex].ServiceProvider.GetRequiredService(); + var requestKey = gateways[ownerIndex].GetOutstandingRequestKeys(clientBId).Single(); + var routingRequest = new Message + { + SendingSilo = ownerDetails.SiloAddress, + }; + routingRequest.SetGatewayRequestOwner(ownerDetails.GatewayAddress, ownerDetails.SiloAddress); + var responseThroughOtherGateway = new Message + { + Direction = Message.Directions.Response, + Id = requestKey.CorrelationId, + SendingGrain = observerBId, + TargetGrain = requestKey.GrainId, + }; + responseThroughOtherGateway.ApplyGatewayRequestOwner(routingRequest); + gateways[1 - ownerIndex].RecordClientResponse(responseThroughOtherGateway); + Assert.Equal(ownerDetails.SiloAddress, gateways[1 - ownerIndex].TryToReroute(responseThroughOtherGateway)); + await WaitUntilAsync( + () => gateways[ownerIndex].GetOutstandingRequestCount(clientBId) == 0, + TimeSpan.FromSeconds(10)); + Assert.All(gateways, gateway => Assert.Equal(0, gateway.GetOutstandingRequestCount(clientBId))); + + observerB.UnblockResponse(); + Assert.Equal("owner-routed response", await responseTask); + await WaitUntilAsync( + () => gateways.All(gateway => gateway.GetOutstandingRequestCount(clientBId) == 0), + TimeSpan.FromSeconds(10)); + + await _cluster.RemoveClientAsync("OwnerClientB"); + await clientA.GetGrain(0).DropDisconnectedClients(excludeRecent: false); + Assert.All(gateways, gateway => Assert.Equal(0, gateway.GetOutstandingRequestCount(clientBId))); + } + + [Theory] + [InlineData(true)] + [InlineData(false)] + public async Task ClientReceivesRejectionWhenTargetClientDisconnected(bool hostedClient) + { + var cancellationToken = TestContext.Current.CancellationToken; + var clientA = hostedClient ? _cluster.Silos[0].ServiceProvider.GetRequiredService() : await _cluster.GetClientAsync("ClientA"); + var clientB = await _cluster.GetClientAsync("ClientB"); + + var observerA = new EchoGrainObserver(); + observerA.SelfReference = clientA.CreateObjectReference(observerA); + + var observerB = new EchoGrainObserver(); + observerB.SelfReference = clientB.CreateObjectReference(observerB); + + // Exchange references, so each one has a reference to the other which is bound to its client. + var observerBId = observerB.SelfReference.GetGrainId(); + var aToB = (IEchoGrainObserver)clientA.ServiceProvider.GetRequiredService().CreateReference(observerBId, GrainInterfaceType.Create("IEchoGrainObserver")); + + observerB.UnblockResponse(); + await aToB.EchoAsync("Hi from A."); + + const string message = "Hello from Client A"; + observerB.UnblockResponse(); + var response = await aToB.EchoAsync(message); + Assert.Equal(message, response); + + await _cluster.RemoveClientAsync("ClientB"); + + observerB.UnblockResponse(); + var responseTask = aToB.EchoAsync(message); + await Assert.ThrowsAsync(async () => await responseTask.WaitAsync(TimeSpan.FromMilliseconds(200), cancellationToken)); + Assert.False(responseTask.IsCompleted, "The task should not complete before the client has been dropped."); + + // Use IManagementGrain to force all Gateways to drop defunct clients. + var managementGrain = clientA.GetGrain(0); + await managementGrain.DropDisconnectedClients(excludeRecent: false); + + // The call should promptly fail with a ClientNotAvailableException. + await Assert.ThrowsAsync(() => responseTask); + + // Attempt call from A to B after B disconnected, expect rejection + await Assert.ThrowsAsync(async () => + { + // This call should fail because Client B is gone and the gateway should reject it. + await aToB.EchoAsync("Calling disconnected client"); + }); + } + + [Theory] + [InlineData(true)] + [InlineData(false)] + public async Task ClientReceivesRejectionForResponseWhenTargetClientDisconnected(bool hostedClient) + { + var clientA = hostedClient ? _cluster.Silos[0].ServiceProvider.GetRequiredService() : await _cluster.GetClientAsync("ClientA"); + var clientB = await _cluster.GetClientAsync("ClientB"); + + var observerA = new EchoGrainObserver(); + observerA.SelfReference = clientA.CreateObjectReference(observerA); + + var observerB = new EchoGrainObserver(); + observerB.SelfReference = clientB.CreateObjectReference(observerB); + + // Create references from each to the other. + var aToB = (IEchoGrainObserver)clientA.ServiceProvider.GetRequiredService().CreateReference(observerB.SelfReference.GetGrainId(), GrainInterfaceType.Create("IEchoGrainObserver")); + var bToA = (IEchoGrainObserver)clientB.ServiceProvider.GetRequiredService().CreateReference(observerA.SelfReference.GetGrainId(), GrainInterfaceType.Create("IEchoGrainObserver")); + + // B -> A (blocked) + var responseTask = bToA.EchoAsync("Hi from B."); + + // B disconnects + await _cluster.RemoveClientAsync("ClientB"); + + // B's pending request should be promptly rejected locally. + var exception = await Assert.ThrowsAnyAsync(() => responseTask); + Assert.True( + exception is OperationCanceledException or OrleansMessageRejectionException or SiloUnavailableException, + $"Unexpected exception type: {exception.GetType()}"); + + // A sends response to B. + observerA.UnblockResponse(); + + // Purge disconnected clients (rejecting pending response) + var managementGrain = clientA.GetGrain(0); + await managementGrain.DropDisconnectedClients(excludeRecent: false); + } + + [Fact] + public async Task ClientCannotSendMessageAfterDisconnecting() + { + var clientA = await _cluster.GetClientAsync("ClientA"); + var observerA = new EchoGrainObserver(); + observerA.SelfReference = clientA.CreateObjectReference(observerA); + + await _cluster.RemoveClientAsync("ClientA"); + + // Attempt to send a message after disconnect + await Assert.ThrowsAsync(async () => + { + await observerA.SelfReference.EchoAsync("Should fail"); + }); + } + + [Theory] + [InlineData(true)] + [InlineData(false)] + public async Task MessageToDisconnectingClientIsRejected(bool hostedClient) + { + var clientA = hostedClient ? _cluster.Silos[0].ServiceProvider.GetRequiredService() : await _cluster.GetClientAsync("ClientA"); + var clientB = await _cluster.GetClientAsync("ClientB"); + + var observerA = new EchoGrainObserver(); + observerA.SelfReference = clientA.CreateObjectReference(observerA); + var observerB = new EchoGrainObserver(); + observerB.SelfReference = clientB.CreateObjectReference(observerB); + + // Create references from each to the other. + var aToB = (IEchoGrainObserver)clientA.ServiceProvider.GetRequiredService().CreateReference(observerB.SelfReference.GetGrainId(), GrainInterfaceType.Create("IEchoGrainObserver")); + var bToA = (IEchoGrainObserver)clientB.ServiceProvider.GetRequiredService().CreateReference(observerA.SelfReference.GetGrainId(), GrainInterfaceType.Create("IEchoGrainObserver")); + + // Start a call but disconnect B before it can respond + var responseTask = aToB.EchoAsync("Test message"); + await _cluster.RemoveClientAsync("ClientB"); + + // Purge disconnected clients (rejecting pending response) + var managementGrain = clientA.GetGrain(0); + await managementGrain.DropDisconnectedClients(excludeRecent: false); + + // The call should be rejected + await Assert.ThrowsAsync(async () => await responseTask); + } + + [GrainInterfaceType("IEchoGrainObserver")] + public interface IEchoGrainObserver : IGrainObserver + { + Task EchoAsync(string message); + Task SendSelfReferenceToPeerAsync(IEchoGrainObserver peer); + Task SetPeerReferenceAsync(IEchoGrainObserver other); + } + + public sealed class EchoGrainObserver : IEchoGrainObserver + { + private TaskCompletionSource _entered = new(TaskCreationOptions.RunContinuationsAsynchronously); + private TaskCompletionSource _tcs = new(); + public IEchoGrainObserver? SelfReference { get; set; } + public IEchoGrainObserver? PeerReference { get; private set; } + public void UnblockResponse() + { + _tcs.SetResult(); + } + + public Task WaitForCallAsync() => _entered.Task; + + public async Task EchoAsync(string message) + { + _entered.TrySetResult(); + await _tcs.Task; + _tcs = new(); + _entered = new(TaskCreationOptions.RunContinuationsAsynchronously); + return message; + } + + public Task SetPeerReferenceAsync(IEchoGrainObserver other) + { + PeerReference = other; + return Task.CompletedTask; + } + + public async Task SendSelfReferenceToPeerAsync(IEchoGrainObserver peer) + { + await peer.SetPeerReferenceAsync(SelfReference!); + } + + } + + private static async Task WaitUntilAsync(Func condition, TimeSpan timeout) + { + var deadline = DateTime.UtcNow + timeout; + while (!condition()) + { + if (DateTime.UtcNow >= deadline) + { + throw new TimeoutException("The expected gateway request-tracking state was not reached."); + } + + await Task.Delay(10); + } + } +}