diff --git a/src/Opc.Ua.Server/Session/Session.cs b/src/Opc.Ua.Server/Session/Session.cs index a56e916608..8e2aa0b92f 100644 --- a/src/Opc.Ua.Server/Session/Session.cs +++ b/src/Opc.Ua.Server/Session/Session.cs @@ -184,6 +184,9 @@ namespace Opc.Ua.Server // initialize diagnostics. DateTime now = m_timeProvider.GetUtcNow().UtcDateTime; m_lastContactTickCount = m_timeProvider.GetTimestampMilliseconds(); + m_lastContactUtcTicks = now.Ticks; + m_sessionTimeoutMs = (long)sessionTimeout; + m_foldedContactUtcTicks = now.Ticks; SessionDiagnostics = new SessionDiagnosticsDataType { SessionId = default, @@ -446,7 +449,9 @@ namespace Opc.Ua.Server lock (m_diagnosticsLock) { + FoldPendingDiagnosticsLocked(); update.Invoke(SessionDiagnostics); + PublishSessionTimeoutLocked(); } } @@ -471,6 +476,7 @@ namespace Opc.Ua.Server lock (m_diagnosticsLock) { + FoldPendingDiagnosticsLocked(); return read.Invoke(SessionDiagnostics); } } @@ -521,14 +527,32 @@ namespace Opc.Ua.Server /// /// Whether the session timeout has elapsed since the last communication from the client. /// + /// + /// Checked at admission and dispatch of every request, so it takes no lock. An + /// observed expiry is recorded in the contact time, so a request that read the + /// clock earlier cannot refresh it afterwards (see ). + /// public bool HasExpired { get { - lock (m_diagnosticsLock) + while (true) { - return m_timeProvider.GetTimestampMilliseconds() - m_lastContactTickCount > - (long)SessionDiagnostics.ActualSessionTimeout; + long contact = ReadContactWord(); + if ((contact & kExpiredFlag) != 0) + { + return true; + } + if (m_timeProvider.GetTimestampMilliseconds() - contact <= + ReadSessionTimeout()) + { + return false; + } + if (Interlocked.CompareExchange( + ref m_lastContactTickCount, contact | kExpiredFlag, contact) == contact) + { + return true; + } } } } @@ -542,6 +566,7 @@ namespace Opc.Ua.Server { lock (m_diagnosticsLock) { + FoldPendingDiagnosticsLocked(); return (DateTime)SessionDiagnostics.ClientLastContactTime; } } @@ -551,21 +576,19 @@ namespace Opc.Ua.Server /// The monotonic tick count (milliseconds) at the last client contact. /// Used for timeout calculations that are immune to system time changes. /// - public long LastContactTickCount - { - get - { - lock (m_diagnosticsLock) - { - return m_lastContactTickCount; - } - } - } + public long LastContactTickCount => ReadContactWord() & ~kExpiredFlag; /// /// Whether the session has been activated. /// - public bool Activated { get; private set; } + /// + /// Set once under the session lock and read without it by request validation. + /// + public bool Activated + { + get => Volatile.Read(ref m_activated); + private set => Volatile.Write(ref m_activated, value); + } /// /// Whether the session is being closed. Closing is entered once and never left, so a @@ -730,7 +753,11 @@ namespace Opc.Ua.Server /// /// Returns the session's SecureChannelId /// - public string SecureChannelId { get; private set; } + public string SecureChannelId + { + get => Volatile.Read(ref m_secureChannelId); + private set => Volatile.Write(ref m_secureChannelId, value); + } /// /// allow derived classes access @@ -753,42 +780,44 @@ namespace Opc.Ua.Server throw new ArgumentNullException(nameof(requestHeader)); } - lock (m_lock) + // Runs for every request, so it takes neither the session lock nor the + // diagnostics lock. The channel id and the activated flag are published with + // volatile writes by Activate, the closing flag is entered once, the contact + // refresh is a compare-and-swap and the counters are interlocked increments. + if (secureChannelContext == null || !IsSecureChannelValid(secureChannelContext.SecureChannelId)) { - if (secureChannelContext == null || !IsSecureChannelValid(secureChannelContext.SecureChannelId)) + UpdateDiagnosticCounters(requestType, true, true); + + if (requestType == RequestType.CloseSession) { - UpdateDiagnosticCounters(requestType, true, true); - - if (requestType == RequestType.CloseSession) - { - throw new ServiceResultException(StatusCodes.BadSessionIdInvalid); - } - - throw new ServiceResultException(StatusCodes.BadSecureChannelIdInvalid); + throw new ServiceResultException(StatusCodes.BadSessionIdInvalid); } - // verify that session has been activated. - if (!Activated && requestType != RequestType.CloseSession) - { - UpdateDiagnosticCounters(requestType, true, true); - throw new ServiceResultException(StatusCodes.BadSessionNotActivated); - } - - // OPC 10000-4 5.7.2.1: a Session that is being closed, or whose timeout has - // elapsed, is terminated and serves no further requests. The expiry check and - // the contact refresh are one step under the diagnostics lock, so a request - // cannot revive a Session that the session monitor has already seen expire. - // CloseSession on an expired Session that is not closing yet is still the - // client's close, so the requested DeleteSubscriptions is honoured. - if (IsClosing || (!TryRefreshLastContact() && requestType != RequestType.CloseSession)) - { - UpdateDiagnosticCounters(requestType, true, false); - throw new ServiceResultException(StatusCodes.BadSessionClosed); - } - - // request accepted. - UpdateDiagnosticCounters(requestType, false, false); + throw new ServiceResultException(StatusCodes.BadSecureChannelIdInvalid); } + + // verify that session has been activated. + if (!Activated && requestType != RequestType.CloseSession) + { + UpdateDiagnosticCounters(requestType, true, true); + throw new ServiceResultException(StatusCodes.BadSessionNotActivated); + } + + // OPC 10000-4 5.7.2.1: a Session that is being closed, or whose timeout has + // elapsed, is terminated and serves no further requests. The expiry check and + // the contact refresh are one compare-and-swap, and an observed expiry is + // recorded in the contact time, so a request cannot revive a Session that the + // session monitor has already seen expire. CloseSession on an expired Session + // that is not closing yet is still the client's close, so the requested + // DeleteSubscriptions is honoured. + if (IsClosing || (!TryRefreshLastContact() && requestType != RequestType.CloseSession)) + { + UpdateDiagnosticCounters(requestType, true, false); + throw new ServiceResultException(StatusCodes.BadSessionClosed); + } + + // request accepted. + UpdateDiagnosticCounters(requestType, false, false); } /// @@ -820,12 +849,13 @@ namespace Opc.Ua.Server /// /// Checks if the secure channel is currently valid. /// + /// + /// Checked for every request; the channel id is written under the session lock + /// and published with a volatile write, so it is read without the lock. + /// public virtual bool IsSecureChannelValid(string secureChannelId) { - lock (m_lock) - { - return SecureChannelId == secureChannelId; - } + return SecureChannelId == secureChannelId; } /// @@ -1169,11 +1199,12 @@ namespace Opc.Ua.Server unusedUserTokenNonce = TakePreparedUserTokenNonce(); m_consumedUserTokenNonce = null; - // update the contact time. - lock (m_diagnosticsLock) + // update the contact time; an expiry already observed stays expired, and + // then the diagnostic contact time is not advanced either. Stamped even when + // the tick count did not advance, so the activation itself is recorded. + if (TryRefreshLastContact()) { - SessionDiagnostics.ClientLastContactTime = m_timeProvider.GetUtcNow().UtcDateTime; - m_lastContactTickCount = m_timeProvider.GetTimestampMilliseconds(); + WriteContactUtcTicks(m_timeProvider.GetUtcNow().UtcDateTime.Ticks); } } retiredUserTokenNonce?.Dispose(); @@ -1245,6 +1276,7 @@ namespace Opc.Ua.Server { lock (m_diagnosticsLock) { + FoldPendingDiagnosticsLocked(); value = Variant.FromStructure(SessionDiagnostics, copy: true); } @@ -1756,167 +1788,235 @@ namespace Opc.Ua.Server /// Refreshes the last contact time unless the session timeout has already elapsed. /// /// false when the session has expired. + /// + /// Lock free. The expiry check and the refresh are one compare-and-swap on the + /// contact time, and an expiry observed here or by is + /// recorded in it (), so a request that read the clock + /// before the expiry was observed cannot refresh it afterwards. Requests within + /// the same millisecond do not write: the contact time already says that. + /// private bool TryRefreshLastContact() { - lock (m_diagnosticsLock) + while (true) { - long now = m_timeProvider.GetTimestampMilliseconds(); - if (now - m_lastContactTickCount > (long)SessionDiagnostics.ActualSessionTimeout) + long contact = ReadContactWord(); + if ((contact & kExpiredFlag) != 0) { return false; } - m_lastContactTickCount = now; - return true; + long now = m_timeProvider.GetTimestampMilliseconds(); + if (now - contact > ReadSessionTimeout()) + { + if (Interlocked.CompareExchange( + ref m_lastContactTickCount, contact | kExpiredFlag, contact) == contact) + { + return false; + } + continue; + } + if (now <= contact) + { + return true; + } + if (Interlocked.CompareExchange(ref m_lastContactTickCount, now, contact) == contact) + { + WriteContactUtcTicks(m_timeProvider.GetUtcNow().UtcDateTime.Ticks); + return true; + } } } + /// + /// Reads the contact time word (tick count plus ); + /// atomic on 32-bit too. + /// + private long ReadContactWord() + { + return IntPtr.Size == 8 + ? Volatile.Read(ref m_lastContactTickCount) + : Interlocked.Read(ref m_lastContactTickCount); + } + + /// + /// The session timeout in milliseconds, read without the diagnostics lock. + /// + /// + /// may be changed under the diagnostics lock + /// (), so the expiry checks do not read its + /// ActualSessionTimeout directly: the value is republished as an atomic integer + /// after every update. + /// + private long ReadSessionTimeout() + { + return IntPtr.Size == 8 + ? Volatile.Read(ref m_sessionTimeoutMs) + : Interlocked.Read(ref m_sessionTimeoutMs); + } + + /// + /// Republishes ActualSessionTimeout for ; called + /// under the diagnostics lock after the diagnostics were updated. + /// + private void PublishSessionTimeoutLocked() + { + long timeout = (long)SessionDiagnostics.ActualSessionTimeout; + if (IntPtr.Size == 8) + { + Volatile.Write(ref m_sessionTimeoutMs, timeout); + } + else + { + Interlocked.Exchange(ref m_sessionTimeoutMs, timeout); + } + } + + /// + /// Publishes the wall-clock time of the last contact; folded into + /// when the diagnostics are read. + /// + private void WriteContactUtcTicks(long ticks) + { + if (IntPtr.Size == 8) + { + Volatile.Write(ref m_lastContactUtcTicks, ticks); + } + else + { + Interlocked.Exchange(ref m_lastContactUtcTicks, ticks); + } + } + + /// + /// Applies the request counts and contact time that requests recorded without the + /// diagnostics lock to . Called under the + /// diagnostics lock before the diagnostics are read or updated. + /// + private void FoldPendingDiagnosticsLocked() + { + long utcTicks = IntPtr.Size == 8 + ? Volatile.Read(ref m_lastContactUtcTicks) + : Interlocked.Read(ref m_lastContactUtcTicks); + if (utcTicks != m_foldedContactUtcTicks) + { + m_foldedContactUtcTicks = utcTicks; + SessionDiagnostics.ClientLastContactTime = new DateTime(utcTicks, DateTimeKind.Utc); + } + + uint total = 0; + uint errors = 0; + uint unauthorized = 0; + for (int ii = 0; ii < kRequestTypeCount; ii++) + { + // Each request is one packed delta, so a fold takes whole requests: an error + // is never counted without its request, nor an unauthorized one without its error. + if (Volatile.Read(ref m_pendingRequestCounts[ii]) == 0) + { + continue; + } + ulong pending = unchecked((ulong)Interlocked.Exchange(ref m_pendingRequestCounts[ii], 0)); + uint requests = (uint)(pending & kTotalMask); + uint failed = (uint)((pending >> kErrorShift) & kErrorMask); + total += requests; + errors += failed; + unauthorized += (uint)(pending >> kUnauthorizedShift); + ServiceCounterDataType? counter = GetServiceCounter((RequestType)ii); + if (counter != null) + { + counter.TotalCount += requests; + counter.ErrorCount += failed; + } + } + SessionDiagnostics.TotalRequestCount.TotalCount += total; + SessionDiagnostics.TotalRequestCount.ErrorCount += errors; + SessionDiagnostics.UnauthorizedRequestCount += unauthorized; + } + /// /// Updates the diagnostic counters associated with the request. /// + /// + /// Runs for every request, so the counts are interlocked increments of per-session + /// slots instead of updates under the diagnostics lock; they are applied to + /// when it is read (). + /// /// private void UpdateDiagnosticCounters( RequestType requestType, bool error, bool authorizationError) { - ServiceCounterDataType? counter = null; - - lock (m_diagnosticsLock) + int slot = (int)requestType; + if ((uint)slot >= (uint)kRequestTypeCount) { - if (!error) + throw ServiceResultException.Unexpected( + $"Unexpected RequestType {requestType}"); + } + + long delta = !error + ? 1L + : authorizationError + ? 1L + (1L << kErrorShift) + (1L << kUnauthorizedShift) + : 1L + (1L << kErrorShift); + ulong pending = unchecked((ulong)Interlocked.Add(ref m_pendingRequestCounts[slot], delta)); + + // Without a diagnostics reader nothing folds the slots; fold before a field can + // carry into the next one. Every field has half its range left as headroom for + // the requests that add concurrently until the fold has taken the slot. + if ((pending & kTotalMask) >= kTotalFoldThreshold || + ((pending >> kErrorShift) & kErrorMask) >= kErrorFoldThreshold) + { + lock (m_diagnosticsLock) { - SessionDiagnostics.ClientLastContactTime = m_timeProvider.GetUtcNow().UtcDateTime; - m_lastContactTickCount = m_timeProvider.GetTimestampMilliseconds(); - } - - SessionDiagnostics.TotalRequestCount.TotalCount++; - - if (error) - { - SessionDiagnostics.TotalRequestCount.ErrorCount++; - - if (authorizationError) - { - SessionDiagnostics.UnauthorizedRequestCount++; - } - } - - switch (requestType) - { - case RequestType.Read: - counter = SessionDiagnostics.ReadCount; - break; - case RequestType.HistoryRead: - counter = SessionDiagnostics.HistoryReadCount; - break; - case RequestType.Write: - counter = SessionDiagnostics.WriteCount; - break; - case RequestType.HistoryUpdate: - counter = SessionDiagnostics.HistoryUpdateCount; - break; - case RequestType.Call: - counter = SessionDiagnostics.CallCount; - break; - case RequestType.CreateMonitoredItems: - counter = SessionDiagnostics.CreateMonitoredItemsCount; - break; - case RequestType.ModifyMonitoredItems: - counter = SessionDiagnostics.ModifyMonitoredItemsCount; - break; - case RequestType.SetMonitoringMode: - counter = SessionDiagnostics.SetMonitoringModeCount; - break; - case RequestType.SetTriggering: - counter = SessionDiagnostics.SetTriggeringCount; - break; - case RequestType.DeleteMonitoredItems: - counter = SessionDiagnostics.DeleteMonitoredItemsCount; - break; - case RequestType.CreateSubscription: - counter = SessionDiagnostics.CreateSubscriptionCount; - break; - case RequestType.ModifySubscription: - counter = SessionDiagnostics.ModifySubscriptionCount; - break; - case RequestType.SetPublishingMode: - counter = SessionDiagnostics.SetPublishingModeCount; - break; - case RequestType.Publish: - counter = SessionDiagnostics.PublishCount; - break; - case RequestType.Republish: - counter = SessionDiagnostics.RepublishCount; - break; - case RequestType.TransferSubscriptions: - counter = SessionDiagnostics.TransferSubscriptionsCount; - break; - case RequestType.DeleteSubscriptions: - counter = SessionDiagnostics.DeleteSubscriptionsCount; - break; - case RequestType.AddNodes: - counter = SessionDiagnostics.AddNodesCount; - break; - case RequestType.AddReferences: - counter = SessionDiagnostics.AddReferencesCount; - break; - case RequestType.DeleteNodes: - counter = SessionDiagnostics.DeleteNodesCount; - break; - case RequestType.DeleteReferences: - counter = SessionDiagnostics.DeleteReferencesCount; - break; - case RequestType.Browse: - counter = SessionDiagnostics.BrowseCount; - break; - case RequestType.BrowseNext: - counter = SessionDiagnostics.BrowseNextCount; - break; - case RequestType.TranslateBrowsePathsToNodeIds: - counter = SessionDiagnostics.TranslateBrowsePathsToNodeIdsCount; - break; - case RequestType.QueryFirst: - counter = SessionDiagnostics.QueryFirstCount; - break; - case RequestType.QueryNext: - counter = SessionDiagnostics.QueryNextCount; - break; - case RequestType.RegisterNodes: - counter = SessionDiagnostics.RegisterNodesCount; - break; - case RequestType.UnregisterNodes: - counter = SessionDiagnostics.UnregisterNodesCount; - break; - case RequestType.Unknown: - case RequestType.FindServers: - case RequestType.GetEndpoints: - case RequestType.CreateSession: - case RequestType.ActivateSession: - case RequestType.CloseSession: - case RequestType.Cancel: - break; - default: - throw ServiceResultException.Unexpected( - $"Unexpected RequestType {requestType}"); - } - - if (counter != null) - { - counter.TotalCount++; - - if (error) - { - counter.ErrorCount++; - } + FoldPendingDiagnosticsLocked(); } } - if (counter != null) + // the services without a counter of their own only count in TotalRequestCount. + if (requestType >= RequestType.Read) { m_server.SessionManager.RaiseSessionDiagnosticsChangedEvent(this); } } + /// + /// The per-service counter of a request type, or null for the services that + /// only count in TotalRequestCount. + /// + private ServiceCounterDataType? GetServiceCounter(RequestType requestType) + { + return requestType switch + { + RequestType.Read => SessionDiagnostics.ReadCount, + RequestType.HistoryRead => SessionDiagnostics.HistoryReadCount, + RequestType.Write => SessionDiagnostics.WriteCount, + RequestType.HistoryUpdate => SessionDiagnostics.HistoryUpdateCount, + RequestType.Call => SessionDiagnostics.CallCount, + RequestType.CreateMonitoredItems => SessionDiagnostics.CreateMonitoredItemsCount, + RequestType.ModifyMonitoredItems => SessionDiagnostics.ModifyMonitoredItemsCount, + RequestType.SetMonitoringMode => SessionDiagnostics.SetMonitoringModeCount, + RequestType.SetTriggering => SessionDiagnostics.SetTriggeringCount, + RequestType.DeleteMonitoredItems => SessionDiagnostics.DeleteMonitoredItemsCount, + RequestType.CreateSubscription => SessionDiagnostics.CreateSubscriptionCount, + RequestType.ModifySubscription => SessionDiagnostics.ModifySubscriptionCount, + RequestType.SetPublishingMode => SessionDiagnostics.SetPublishingModeCount, + RequestType.Publish => SessionDiagnostics.PublishCount, + RequestType.Republish => SessionDiagnostics.RepublishCount, + RequestType.TransferSubscriptions => SessionDiagnostics.TransferSubscriptionsCount, + RequestType.DeleteSubscriptions => SessionDiagnostics.DeleteSubscriptionsCount, + RequestType.AddNodes => SessionDiagnostics.AddNodesCount, + RequestType.AddReferences => SessionDiagnostics.AddReferencesCount, + RequestType.DeleteNodes => SessionDiagnostics.DeleteNodesCount, + RequestType.DeleteReferences => SessionDiagnostics.DeleteReferencesCount, + RequestType.Browse => SessionDiagnostics.BrowseCount, + RequestType.BrowseNext => SessionDiagnostics.BrowseNextCount, + RequestType.TranslateBrowsePathsToNodeIds => SessionDiagnostics.TranslateBrowsePathsToNodeIdsCount, + RequestType.QueryFirst => SessionDiagnostics.QueryFirstCount, + RequestType.QueryNext => SessionDiagnostics.QueryNextCount, + RequestType.RegisterNodes => SessionDiagnostics.RegisterNodesCount, + RequestType.UnregisterNodes => SessionDiagnostics.UnregisterNodesCount, + _ => null + }; + } private readonly Lock m_lock = new(); /// @@ -1978,7 +2078,37 @@ namespace Opc.Ua.Server private readonly CertificateCollection? m_clientIssuerCertificates; private readonly SessionContinuationPoints m_continuationPoints; private readonly SessionSecurityDiagnosticsDataType m_securityDiagnostics; + + /// + /// The tick count (milliseconds) of the last contact, with + /// set once an expiry was observed. + /// private long m_lastContactTickCount; + private long m_lastContactUtcTicks; + private long m_sessionTimeoutMs; + private long m_foldedContactUtcTicks; + + /// + /// Request counts not yet applied to , one packed + /// slot per so that a request is recorded with a single + /// interlocked add: bits 0-31 total, 32-47 errors, 48-63 unauthorized. Unauthorized + /// requests are a subset of the errors, so the error threshold also bounds them. + /// + private readonly long[] m_pendingRequestCounts = new long[kRequestTypeCount]; + private const int kRequestTypeCount = (int)RequestType.UnregisterNodes + 1; + private const int kErrorShift = 32; + private const int kUnauthorizedShift = 48; + private const ulong kTotalMask = 0xFFFF_FFFF; + private const ulong kErrorMask = 0xFFFF; + private const ulong kTotalFoldThreshold = 0x8000_0000; + private const ulong kErrorFoldThreshold = 0x8000; + + /// + /// Marks as expired; tick counts stay far below it. + /// + private const long kExpiredFlag = 1L << 62; + private bool m_activated; + private string m_secureChannelId = null!; private long m_identityGeneration; private long m_identityRefreshGeneration; } diff --git a/src/Opc.Ua.Server/Session/SessionManager.cs b/src/Opc.Ua.Server/Session/SessionManager.cs index d3b48d8ccd..6ee606474e 100644 --- a/src/Opc.Ua.Server/Session/SessionManager.cs +++ b/src/Opc.Ua.Server/Session/SessionManager.cs @@ -1290,15 +1290,15 @@ namespace Opc.Ua.Server SessionBindingContext binding; ByteString originalClientChannelCertificate; - // Runs at admission and again at dispatch of every request. The session index - // and the activation-state table are concurrent collections, so the candidate - // snapshot is taken without the bindings lock; the check under the lock below - // confirms that the binding is still the committed one before it is returned. + // Runs at admission and again at dispatch of every request, so it takes no + // server-wide lock. The session index and the activation-state table are + // concurrent collections and DispatchBinding is published by every writer of + // the binding, so the candidate snapshot is lock free; the re-check at the end + // confirms that it is still the committed binding before it is returned. if (authenticationToken.IsNull || !m_sessions.TryGetValue(authenticationToken, out ISession? currentSession) || !m_sessionActivationStates.TryGetValue(currentSession, out SessionActivationState? currentState) || - currentState.IsCommitting || - currentState.BindingContext is not SessionBindingContext currentBinding) + currentState.DispatchBinding is not SessionBindingContext currentBinding) { return false; } @@ -1323,18 +1323,20 @@ namespace Opc.Ua.Server return false; } - lock (m_bindingsLock) + // The same binding instance with no activation committing means no commit, + // removal or commit in progress replaced it (a commit always creates a new + // instance). The token is mapped to a session object at most once and never + // again after it was removed, so finding it still mapped afterwards means it + // was mapped when the binding was read: both held at that instant, as under + // the lock the writers take. + if (!ReferenceEquals(state.DispatchBinding, binding) || + !m_sessions.TryGetValue(authenticationToken, out ISession? committedSession) || + !ReferenceEquals(committedSession, session)) { - if (state.IsCommitting || - !ReferenceEquals(state.BindingContext, binding) || - !m_sessions.TryGetValue(authenticationToken, out ISession? committedSession) || - !ReferenceEquals(committedSession, session)) - { - return false; - } - context = binding; - return true; + return false; } + context = binding; + return true; } private void CommitSessionBinding( @@ -1344,13 +1346,13 @@ namespace Opc.Ua.Server { lock (m_bindingsLock) { - state.IsCommitting = false; // Shutdown can remove the session while authentication is awaiting a provider, // and a timeout or termination can start closing it. if (!m_sessions.TryGetValue(authenticationToken, out ISession? current) || !ReferenceEquals(current, session) || SessionTermination.IsClosingOrClaimed(session)) { + state.IsCommitting = false; throw new ServiceResultException(StatusCodes.BadSessionClosed); } @@ -1369,6 +1371,10 @@ namespace Opc.Ua.Server m_channelSessionCounts[binding.SecureChannelId] = count + 1; } state.BindingContext = binding; + // Cleared only after the new binding is installed: clearing it first would + // republish the previous binding to lock-free lookups (DispatchBinding) + // while Session.Activate has already changed the session. + state.IsCommitting = false; } } @@ -2515,6 +2521,15 @@ namespace Opc.Ua.Server /// protected virtual void RaiseSessionEvent(ISession session, SessionEventReason reason) { + // Diagnostics changes are raised for every request; without a subscriber they + // must not take the server-wide event lock. Subscriptions are added under the + // lock, so a subscriber racing with this check misses at most this one event. + if (reason == SessionEventReason.DiagnosticsChanged && + Volatile.Read(ref m_SessionDiagnosticsChanged) == null) + { + return; + } + lock (m_eventLock) { SessionEventHandler? handler = null; @@ -2873,9 +2888,50 @@ namespace Opc.Ua.Server public long ActivationSequence { get; set; } - public SessionBindingContext? BindingContext { get; set; } + /// + /// The committed binding; written under the bindings lock. + /// + public SessionBindingContext? BindingContext + { + get => m_bindingContext; + set + { + m_bindingContext = value; + PublishDispatchBinding(); + } + } - public bool IsCommitting { get; set; } + /// + /// An activation is between Session.Activate and the binding commit; written + /// under the bindings lock. + /// + public bool IsCommitting + { + get => m_isCommitting; + set + { + m_isCommitting = value; + PublishDispatchBinding(); + } + } + + /// + /// The binding requests may be dispatched under: + /// while no activation is committing, otherwise null. Republished by + /// every writer of the two, so a single read gives a consistent pair without + /// the bindings lock. A commit always creates a new binding instance, so an + /// unchanged reference means the binding was not replaced. + /// + public SessionBindingContext? DispatchBinding => Volatile.Read(ref m_dispatchBinding); + + private void PublishDispatchBinding() + { + Volatile.Write(ref m_dispatchBinding, m_isCommitting ? null : m_bindingContext); + } + + private SessionBindingContext? m_bindingContext; + private bool m_isCommitting; + private SessionBindingContext? m_dispatchBinding; /// /// An ActivateSession holds the activation gate and was admitted; set and read diff --git a/tests/Opc.Ua.Server.Tests/SessionBindingTests.cs b/tests/Opc.Ua.Server.Tests/SessionBindingTests.cs index 4f2fda710d..939ca51756 100644 --- a/tests/Opc.Ua.Server.Tests/SessionBindingTests.cs +++ b/tests/Opc.Ua.Server.Tests/SessionBindingTests.cs @@ -28,6 +28,7 @@ * ======================================================================*/ using System; +using System.Collections.Concurrent; using System.Collections.Generic; using System.Threading; using System.Threading.Tasks; @@ -344,6 +345,129 @@ namespace Opc.Ua.Server.Tests Assert.That(m_manager.HasSession("one"), Is.True); } + /// + /// Between Session.Activate and the binding commit the session already carries the + /// new activation, so the previous binding must not be handed out for dispatch. + /// + [Test] + public async Task LookupDuringActivationCommitReturnsNoBindingAsync() + { + OperationContext channel = CreateContext("one"); + CreateSessionResult created = await CreateAsync(channel).ConfigureAwait(false); + await ActivateAsync(created, channel).ConfigureAwait(false); + SessionBindingContext original = GetBinding(created, channel); + bool? classifiedDuringCommit = null; + m_manager.OnNextSessionActivated = () => classifiedDuringCommit = m_manager.TryGetSessionContext( + created.AuthenticationToken, channel.ChannelContext!, out _); + + await ActivateAsync(created, channel).ConfigureAwait(false); + + Assert.That(classifiedDuringCommit, Is.False); + Assert.That(GetBinding(created, channel).ActivationSequence, Is.EqualTo(original.ActivationSequence + 1)); + } + + /// + /// The lookup takes no server-wide lock, so lookups racing with transfers between + /// two channels and a final close must still only return a committed binding of the + /// channel they ask for, never one older than an activation that already completed, + /// and nothing once the close completed. + /// + [Test] + public async Task LookupRacingTransfersAndCloseReturnsOnlyCommittedBindingsAsync() + { + const int kTransfers = 1000; + const int kReaders = 4; + OperationContext[] channels = [CreateContext("a"), CreateContext("b")]; + CreateSessionResult created = await CreateAsync(channels[0]).ConfigureAwait(false); + await ActivateAsync(created, channels[0]).ConfigureAwait(false); + long completed = 1; + int closed = 0; + int lookups = 0; + int readersRunning = 0; + var allReadersRunning = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var errors = new ConcurrentQueue(); + using var stop = new CancellationTokenSource(); + + var readers = new Task[kReaders]; + for (int r = 0; r < kReaders; r++) + { + int reader = r; + readers[r] = Task.Run(() => + { + int found = 0; + int attempt = 0; + long lastSequence = 0; + while (!stop.IsCancellationRequested) + { + bool closedBefore = Volatile.Read(ref closed) != 0; + long floor = Volatile.Read(ref completed); + OperationContext channel = channels[(reader + attempt++) & 1]; + bool ok = m_manager.TryGetSessionContext( + created.AuthenticationToken, channel.ChannelContext!, out SessionBindingContext? binding); + long ceiling = Volatile.Read(ref completed) + 1; + Interlocked.Increment(ref lookups); + if (!ok) + { + continue; + } + if (found++ == 0 && Interlocked.Increment(ref readersRunning) == kReaders) + { + allReadersRunning.TrySetResult(true); + } + long sequence = binding!.ActivationSequence; + string expectedChannel = (sequence & 1) == 1 ? "a" : "b"; + if (closedBefore || + binding.SecureChannelId != channel.ChannelContext!.SecureChannelId || + binding.SecureChannelId != expectedChannel || + sequence < floor || + sequence > ceiling || + sequence < lastSequence) + { + errors.Enqueue( + $"closed={closedBefore} asked={channel.ChannelContext!.SecureChannelId} " + + $"got={binding.SecureChannelId} seq={sequence} floor={floor} " + + $"ceiling={ceiling} last={lastSequence}"); + } + lastSequence = sequence; + } + return found; + }); + } + + try + { + // Race only once every reader runs and has seen the committed binding: on + // a runner with few cores the transfers can otherwise finish before some + // readers were ever scheduled. + await allReadersRunning.Task.WaitAsync(TimeSpan.FromSeconds(30)).ConfigureAwait(false); + for (int i = 1; i <= kTransfers; i++) + { + await ActivateAsync(created, channels[i & 1]).ConfigureAwait(false); + Volatile.Write(ref completed, i + 1); + } + await m_manager.CloseSessionAsync(created.SessionId).ConfigureAwait(false); + Volatile.Write(ref closed, 1); + // Let the readers run against the closed session for a while too. + int target = Volatile.Read(ref lookups) + (kReaders * 1000); + while (Volatile.Read(ref lookups) < target && errors.IsEmpty) + { + await Task.Yield(); + } + } + finally + { + stop.Cancel(); + } + int[] found = await Task.WhenAll(readers).ConfigureAwait(false); + + Assert.That(errors, Is.Empty, string.Join(Environment.NewLine, errors)); + Assert.That(found, Has.All.GreaterThan(0), "Every reader must have seen a committed binding."); + AssertNoBinding(created.AuthenticationToken, channels[0]); + AssertNoBinding(created.AuthenticationToken, channels[1]); + Assert.That(m_manager.HasSession("a"), Is.False); + Assert.That(m_manager.HasSession("b"), Is.False); + } + [Test] public async Task ExpiredSessionLosesClassificationAndTimeoutRemovesMembershipAsync() { @@ -702,6 +826,7 @@ namespace Opc.Ua.Server.Tests { m_server = server; m_certificate = certificate; + m_clock = clock; } public bool RejectAuthentication { get; set; } @@ -823,8 +948,59 @@ namespace Opc.Ua.Server.Tests return session; } + /// + /// Runs once on the activating thread after Session.Activate returned and + /// before the activation commits its binding. + /// + public Action? OnNextSessionActivated { get; set; } + + protected override ISession CreateSession( + OperationContext context, IServerInternal server, Certificate serverCertificate, + NodeId sessionCookie, ByteString clientNonce, Nonce serverNonce, string sessionName, + ApplicationDescription clientDescription, string endpointUrl, Certificate clientCertificate, + CertificateCollection clientCertificateChain, double sessionTimeout, + uint maxResponseMessageSize, int maxRequestAge, int maxContinuationPoints) + { + return new HookSession( + this, context, server, serverCertificate, sessionCookie, clientNonce, serverNonce, + sessionName, clientDescription, endpointUrl, clientCertificate, clientCertificateChain, + sessionTimeout, m_clock); + } + + private sealed class HookSession : Opc.Ua.Server.Session, ISession + { + public HookSession( + BindingSessionManager manager, OperationContext context, IServerInternal server, + Certificate serverCertificate, NodeId authenticationToken, ByteString clientNonce, + Nonce serverNonce, string sessionName, ApplicationDescription clientDescription, + string endpointUrl, Certificate clientCertificate, CertificateCollection clientCertificateChain, + double sessionTimeout, TimeProvider clock) + : base( + context, server, serverCertificate, authenticationToken, clientNonce, serverNonce, + sessionName, clientDescription, endpointUrl, clientCertificate, clientCertificateChain, + sessionTimeout, 10, 10, clock) + { + m_manager = manager; + } + + bool ISession.Activate( + OperationContext context, IUserIdentityTokenHandler identityToken, IUserIdentity identity, + IUserIdentity effectiveIdentity, ArrayOf localeIds, Nonce serverNonce) + { + bool changed = Activate( + context, identityToken, identity, effectiveIdentity, localeIds, serverNonce); + Action? hook = m_manager.OnNextSessionActivated; + m_manager.OnNextSessionActivated = null; + hook?.Invoke(); + return changed; + } + + private readonly BindingSessionManager m_manager; + } + private readonly IServerInternal m_server; private readonly Certificate m_certificate; + private readonly TimeProvider m_clock; private bool m_delayCallback; private bool m_delayAuthentication; } diff --git a/tests/Opc.Ua.Server.Tests/SessionLifecycleRegressionTests.cs b/tests/Opc.Ua.Server.Tests/SessionLifecycleRegressionTests.cs index 77cab68c7e..d1fd74b130 100644 --- a/tests/Opc.Ua.Server.Tests/SessionLifecycleRegressionTests.cs +++ b/tests/Opc.Ua.Server.Tests/SessionLifecycleRegressionTests.cs @@ -103,6 +103,178 @@ namespace Opc.Ua.Server.Tests Assert.That(session.HasExpired, Is.True, "A rejected request must not refresh the contact time."); } + /// + /// LC-1: request validation refreshes the contact time without a lock. A request + /// that read the clock before the session monitor observed the expiry must still + /// not refresh the contact time afterwards and revive the session. + /// + [Test] + public async Task RequestRacingObservedExpiryDoesNotReviveTheSessionAsync() + { + using var harness = new Harness(); + CreateSessionResult created = await harness.CreateActivatedSessionAsync().ConfigureAwait(false); + var session = (ServerSession)created.Session; + var channel = new SecureChannelContext(kChannelId, harness.Endpoint, RequestEncoding.Binary); + harness.Clock.Advance(TimeSpan.FromMilliseconds(900)); + bool? expiredInRace = null; + + // The request reads the clock (900 ms, not expired); before it refreshes the + // contact time the timeout elapses and the monitor observes the expiry. + harness.Clock.AfterNextTimestamp(() => + { + harness.Clock.Advance(TimeSpan.FromMilliseconds(200)); + expiredInRace = session.HasExpired; + }); + + Assert.That( + () => session.ValidateRequest( + new RequestHeader { AuthenticationToken = created.AuthenticationToken }, + channel, + RequestType.Read), + Throws.TypeOf() + .With.Property(nameof(ServiceResultException.StatusCode)) + .EqualTo((StatusCode)StatusCodes.BadSessionClosed)); + Assert.That(expiredInRace, Is.True); + Assert.That(session.HasExpired, Is.True, "An observed expiry must stay observed."); + } + + /// + /// The request counters are updated without the diagnostics lock and applied when + /// the diagnostics are read: concurrent requests must all be counted exactly once, + /// and a reader never sees a count go backwards. + /// + [Test] + public async Task ConcurrentRequestsAreCountedExactlyAsync() + { + const int kThreads = 8; + const int kRequests = 2000; + const int kRejected = 100; + using var harness = new Harness(); + CreateSessionResult created = await harness.CreateActivatedSessionAsync().ConfigureAwait(false); + var session = (ServerSession)created.Session; + var channel = new SecureChannelContext(kChannelId, harness.Endpoint, RequestEncoding.Binary); + var otherChannel = new SecureChannelContext("other-channel", harness.Endpoint, RequestEncoding.Binary); + var header = new RequestHeader { AuthenticationToken = created.AuthenticationToken }; + (uint Total, uint Read) baseline = session.ReadDiagnostics( + d => (d.TotalRequestCount.TotalCount, d.ReadCount.TotalCount)); + + using var stop = new CancellationTokenSource(); + Task reader = Task.Run(() => + { + uint last = 0; + while (!stop.IsCancellationRequested) + { + (uint total, bool consistent) = session.ReadDiagnostics(d => ( + d.TotalRequestCount.TotalCount, + d.WriteCount.ErrorCount <= d.WriteCount.TotalCount && + d.UnauthorizedRequestCount <= d.TotalRequestCount.ErrorCount)); + if (total < last || !consistent) + { + return false; + } + last = total; + } + return true; + }); + var workers = new Task[kThreads]; + for (int ii = 0; ii < kThreads; ii++) + { + workers[ii] = Task.Run(() => + { + for (int jj = 0; jj < kRequests; jj++) + { + session.ValidateRequest(header, channel, (jj & 1) == 0 ? RequestType.Read : RequestType.Browse); + } + for (int jj = 0; jj < kRejected; jj++) + { + Assert.Throws( + () => session.ValidateRequest(header, otherChannel, RequestType.Write)); + } + }); + } + await Task.WhenAll(workers).ConfigureAwait(false); + stop.Cancel(); + Assert.That(await reader.ConfigureAwait(false), Is.True, + "The total request count went backwards, or an error was counted without its request."); + + session.ReadDiagnostics(d => + { + Assert.That(d.TotalRequestCount.TotalCount - baseline.Total, + Is.EqualTo((uint)(kThreads * (kRequests + kRejected)))); + Assert.That(d.TotalRequestCount.ErrorCount, Is.EqualTo((uint)(kThreads * kRejected))); + Assert.That(d.UnauthorizedRequestCount, Is.EqualTo((uint)(kThreads * kRejected))); + Assert.That(d.ReadCount.TotalCount - baseline.Read, Is.EqualTo((uint)(kThreads * kRequests / 2))); + Assert.That(d.ReadCount.ErrorCount, Is.Zero); + Assert.That(d.BrowseCount.TotalCount, Is.EqualTo((uint)(kThreads * kRequests / 2))); + Assert.That(d.WriteCount.TotalCount, Is.EqualTo((uint)(kThreads * kRejected))); + Assert.That(d.WriteCount.ErrorCount, Is.EqualTo((uint)(kThreads * kRejected))); + return true; + }); + } + + /// + /// Without a diagnostics reader nothing applies the pending request counts, so the + /// recording request folds them before a packed field could carry into the next: + /// more rejected requests than the error field holds must still be counted exactly. + /// + [Test] + public async Task PendingCountsFoldBeforeTheyOverflowWithoutAReaderAsync() + { + const int kRejected = 70_000; + using var harness = new Harness(); + CreateSessionResult created = await harness.CreateActivatedSessionAsync().ConfigureAwait(false); + var session = (ServerSession)created.Session; + var otherChannel = new SecureChannelContext("other-channel", harness.Endpoint, RequestEncoding.Binary); + var header = new RequestHeader { AuthenticationToken = created.AuthenticationToken }; + uint baseline = session.ReadDiagnostics(d => d.TotalRequestCount.TotalCount); + + for (int ii = 0; ii < kRejected; ii++) + { + try + { + session.ValidateRequest(header, otherChannel, RequestType.Write); + } + catch (ServiceResultException) + { + } + } + + session.ReadDiagnostics(d => + { + Assert.That(d.TotalRequestCount.TotalCount - baseline, Is.EqualTo((uint)kRejected)); + Assert.That(d.TotalRequestCount.ErrorCount, Is.EqualTo((uint)kRejected)); + Assert.That(d.UnauthorizedRequestCount, Is.EqualTo((uint)kRejected)); + Assert.That(d.WriteCount.TotalCount, Is.EqualTo((uint)kRejected)); + Assert.That(d.WriteCount.ErrorCount, Is.EqualTo((uint)kRejected)); + return true; + }); + } + + /// + /// The expiry checks read the session timeout without the diagnostics lock; a + /// timeout changed through UpdateDiagnostics must still take effect for them. + /// + [Test] + public async Task SessionTimeoutChangedThroughDiagnosticsAppliesToExpiryAsync() + { + using var harness = new Harness(); + CreateSessionResult created = await harness.CreateActivatedSessionAsync().ConfigureAwait(false); + var session = (ServerSession)created.Session; + var channel = new SecureChannelContext(kChannelId, harness.Endpoint, RequestEncoding.Binary); + + session.UpdateDiagnostics(d => d.ActualSessionTimeout = 5_000); + harness.Clock.Advance(TimeSpan.FromSeconds(2)); + Assert.That(session.HasExpired, Is.False); + session.ValidateRequest( + new RequestHeader { AuthenticationToken = created.AuthenticationToken }, + channel, + RequestType.Read); + + session.UpdateDiagnostics(d => d.ActualSessionTimeout = 1_000); + harness.Clock.Advance(TimeSpan.FromSeconds(2)); + Assert.That(session.HasExpired, Is.True); + } + /// /// LC-1 / LC-2: a session that is being closed admits no new requests. /// @@ -603,6 +775,27 @@ namespace Opc.Ua.Server.Tests private readonly AnonymousAuthenticator m_inner = new(); } + /// + /// A fake clock that can run a callback right after the next timestamp was read, + /// i.e. between a caller reading the clock and acting on the value. + /// + private sealed class HookClock : FakeTimeProvider + { + public override long GetTimestamp() + { + long timestamp = base.GetTimestamp(); + Interlocked.Exchange(ref m_afterNextTimestamp, null)?.Invoke(); + return timestamp; + } + + public void AfterNextTimestamp(Action callback) + { + m_afterNextTimestamp = callback ?? throw new ArgumentNullException(nameof(callback)); + } + + private Action? m_afterNextTimestamp; + } + /// /// A session manager over a mocked server whose clock, diagnostics and close path /// are observable. @@ -717,7 +910,7 @@ namespace Opc.Ua.Server.Tests public Mock Server { get; } = new(); - public FakeTimeProvider Clock { get; } = new(); + public HookClock Clock { get; } = new(); public SessionManager Manager { get; }