xref: /dports/lang/mono/mono-5.10.1.57/mcs/class/referencesource/System.ServiceModel.Activities/System/ServiceModel/Activities/Dispatcher/BufferedReceiveManager.cs

//----------------------------------------------------------------
// Copyright (c) Microsoft Corporation.  All rights reserved.
//----------------------------------------------------------------
namespace System.ServiceModel.Activities.Dispatcher
{
    using System.Activities.Hosting;
    using System.Collections.Generic;
    using System.Collections.ObjectModel;
    using System.Runtime;
    using System.Runtime.DurableInstancing;
    using System.ServiceModel.Channels;
    using System.Threading;

    sealed class BufferedReceiveManager : IExtension<ServiceHostBase>
    {
        static AsyncCallback onEndAbandon;
        Dictionary<InstanceKey, List<BufferedReceiveMessageProperty>> bufferedProperties;
        PendingMessageThrottle throttle;
        WorkflowServiceHost host;

        int initialized;

        [Fx.Tag.SynchronizationObject(Blocking = false)]
        object thisLock;

        public BufferedReceiveManager(int maxPendingMessagesPerChannel)
        {
            this.throttle = new PendingMessageThrottle(maxPendingMessagesPerChannel);
            this.thisLock = new object();
        }

        public bool BufferReceive(OperationContext operationContext, ReceiveContext receiveContext, string bookmarkName, BufferedReceiveState state, bool retry)
        {
            Fx.Assert(receiveContext != null, "ReceiveContext must be present in order to perform buffering");

            bool success = false;

            BufferedReceiveMessageProperty property = null;
            if (BufferedReceiveMessageProperty.TryGet(operationContext.IncomingMessageProperties, out property))
            {
                CorrelationMessageProperty correlation = null;
                if (CorrelationMessageProperty.TryGet(operationContext.IncomingMessageProperties, out correlation))
                {
                    InstanceKey instanceKey = correlation.CorrelationKey;
                    int channelKey = operationContext.Channel.GetHashCode();
                    if (this.throttle.Acquire(channelKey))
                    {
                        try
                        {
                            // Tag the property with identifying data to be used during later processing
                            if (UpdateProperty(property, receiveContext, channelKey, bookmarkName, state))
                            {
                                // Cleanup if we are notified the ReceiveContext faulted underneath us
                                receiveContext.Faulted += delegate(object sender, EventArgs e)
                                {
                                    lock (this.thisLock)
                                    {
                                        if (this.bufferedProperties.ContainsKey(instanceKey))
                                        {
                                            if (this.bufferedProperties[instanceKey].Remove(property))
                                            {
                                                try
                                                {
                                                    property.RequestContext.DelayClose(false);
                                                    property.RequestContext.Abort();
                                                }
                                                catch (Exception exception)
                                                {
                                                    if (Fx.IsFatal(exception))
                                                    {
                                                        throw;
                                                    }

                                                    // ---- these exceptions as we are already on the error path
                                                }

                                                this.throttle.Release(channelKey);
                                            }
                                        }
                                    }
                                };

                                // Actual Buffering
                                lock (this.thisLock)
                                {
                                    // Optimistic state check in case we just raced with the receiveContext
                                    // faulting. If the receiveContext still faults after the state check, the above
                                    // cleanup routine will handle things correctly. In both cases, a double-release
                                    // of the throttle is protected.
                                    if (receiveContext.State == ReceiveContextState.Received)
                                    {
                                        bool found = false;
                                        // if the exception indicates retry-able (such as RetryException),
                                        // we will simply retry.  This happens when racing with abort and
                                        // WF informing the client to retry (BufferedReceiveManager is a
                                        // client in this case).
                                        if (retry)
                                        {
                                            property.RequestContext.DelayClose(true);
                                            property.RegisterForReplay(operationContext);
                                            property.ReplayRequest();
                                            property.Notification.NotifyInvokeReceived(property.RequestContext.InnerRequestContext);
                                            found = true;
                                        }
                                        else
                                        {
                                            ReadOnlyCollection<BookmarkInfo> bookmarks = this.host.DurableInstanceManager.PersistenceProviderDirectory.GetBookmarksForInstance(instanceKey);
                                            // Retry in case match the existing bookmark
                                            if (bookmarks != null)
                                            {
                                                for (int i = 0; i < bookmarks.Count; ++i)
                                                {
                                                    BookmarkInfo bookmark = bookmarks[i];
                                                    if (bookmark.BookmarkName == bookmarkName)
                                                    {
                                                        // Found it so retry...
                                                        property.RequestContext.DelayClose(true);
                                                        property.RegisterForReplay(operationContext);
                                                        property.ReplayRequest();
                                                        property.Notification.NotifyInvokeReceived(property.RequestContext.InnerRequestContext);
                                                        found = true;
                                                        break;
                                                    }
                                                }
                                            }
                                        }

                                        if (!found)
                                        {
                                            List<BufferedReceiveMessageProperty> properties;
                                            if (!this.bufferedProperties.TryGetValue(instanceKey, out properties))
                                            {
                                                properties = new List<BufferedReceiveMessageProperty>();
                                                this.bufferedProperties.Add(instanceKey, properties);
                                            }
                                            property.RequestContext.DelayClose(true);
                                            property.RegisterForReplay(operationContext);
                                            properties.Add(property);
                                        }
                                        else
                                        {
                                            this.throttle.Release(channelKey);
                                        }
                                        success = true;
                                    }
                                }
                            }
                        }
                        finally
                        {
                            if (!success)
                            {
                                this.throttle.Release(channelKey);
                            }
                        }
                    }
                }
            }

            return success;
        }

        public void Retry(HashSet<InstanceKey> associatedInstances, ReadOnlyCollection<BookmarkInfo> availableBookmarks)
        {
            List<BookmarkInfo> bookmarks = new List<BookmarkInfo>(availableBookmarks);
            foreach (InstanceKey instanceKey in associatedInstances)
            {
                lock (this.thisLock)
                {
                    if (this.bufferedProperties.ContainsKey(instanceKey))
                    {
                        List<BufferedReceiveMessageProperty> properties = this.bufferedProperties[instanceKey];
                        int index = 0;

                        while (index < properties.Count && bookmarks.Count > 0)
                        {
                            BufferedReceiveMessageProperty property = properties[index];

                            // Determine if this property is now ready to be processed
                            int channelKey = 0;
                            bool found = false;
                            for (int i = 0; i < bookmarks.Count; ++i)
                            {
                                BookmarkInfo bookmark = (BookmarkInfo)bookmarks[i];
                                PropertyData data = (PropertyData)property.UserState;
                                if (bookmark.BookmarkName == data.BookmarkName)
                                {
                                    // Found it so retry...
                                    bookmarks.RemoveAt(i);
                                    channelKey = data.ChannelKey;
                                    property.ReplayRequest();
                                    property.Notification.NotifyInvokeReceived(property.RequestContext.InnerRequestContext);
                                    found = true;
                                    break;
                                }
                            }

                            if (!found)
                            {
                                index++;
                            }
                            else
                            {
                                properties.RemoveAt(index);
                                this.throttle.Release(channelKey);
                            }
                        }
                    }
                }

                if (bookmarks.Count == 0)
                {
                    break;
                }
            }
        }

        public void AbandonBufferedReceives(HashSet<InstanceKey> associatedInstances)
        {
            foreach (InstanceKey instanceKey in associatedInstances)
            {
                lock (this.thisLock)
                {
                    if (this.bufferedProperties.ContainsKey(instanceKey))
                    {
                        foreach (BufferedReceiveMessageProperty property in this.bufferedProperties[instanceKey])
                        {
                            PropertyData data = (PropertyData)property.UserState;
                            AbandonReceiveContext(data.ReceiveContext);
                            this.throttle.Release(data.ChannelKey);
                        }

                        this.bufferedProperties.Remove(instanceKey);
                    }
                }
            }
        }

        // clean up any remaining buffered receives as part of ServiceHost close.
        internal void AbandonBufferedReceives()
        {
            lock (this.thisLock)
            {
                foreach (List<BufferedReceiveMessageProperty> value in this.bufferedProperties.Values)
                {
                    foreach (BufferedReceiveMessageProperty property in value)
                    {
                        PropertyData data = (PropertyData)property.UserState;
                        AbandonReceiveContext(data.ReceiveContext);
                        this.throttle.Release(data.ChannelKey);
                    }
                }
                this.bufferedProperties.Clear();
            }
        }

        // Best-effort to abandon the receiveContext
        internal static void AbandonReceiveContext(ReceiveContext receiveContext)
        {
            if (receiveContext != null)
            {
                if (onEndAbandon == null)
                {
                    onEndAbandon = Fx.ThunkCallback(new AsyncCallback(OnEndAbandon));
                }

                try
                {
                    IAsyncResult result = receiveContext.BeginAbandon(
                        TimeSpan.MaxValue, onEndAbandon, receiveContext);
                    if (result.CompletedSynchronously)
                    {
                        HandleEndAbandon(result);
                    }
                }
                catch (Exception exception)
                {
                    if (Fx.IsFatal(exception))
                    {
                        throw;
                    }

                    // We ---- any Abandon exception - best effort.
                    FxTrace.Exception.AsWarning(exception);
                }
            }
        }

        static bool HandleEndAbandon(IAsyncResult result)
        {
            ReceiveContext receiveContext = (ReceiveContext)result.AsyncState;
            receiveContext.EndAbandon(result);
            return true;
        }

        static void OnEndAbandon(IAsyncResult result)
        {
            if (result.CompletedSynchronously)
            {
                return;
            }

            try
            {
                HandleEndAbandon(result);
            }
            catch (Exception exception)
            {
                if (Fx.IsFatal(exception))
                {
                    throw;
                }

                // We ---- any Abandon exception - best effort.
                FxTrace.Exception.AsWarning(exception);
            }
        }

        void IExtension<ServiceHostBase>.Attach(ServiceHostBase owner)
        {
            if (owner == null)
            {
                throw FxTrace.Exception.AsError(new ArgumentNullException("owner"));
            }

            if (Interlocked.CompareExchange(ref this.initialized, 1, 0) != 0)
            {
                throw FxTrace.Exception.AsError(
                    new InvalidOperationException(SR.BufferedReceiveBehaviorMultipleUse));
            }

            owner.ThrowIfClosedOrOpened();

            Fx.Assert(owner is WorkflowServiceHost, "owner must be of WorkflowServiceHost type!");
            this.host = (WorkflowServiceHost)owner;
            Initialize();
        }

        void IExtension<ServiceHostBase>.Detach(ServiceHostBase owner)
        {
        }

        bool UpdateProperty(BufferedReceiveMessageProperty property, ReceiveContext receiveContext, int channelKey, string bookmarkName, BufferedReceiveState state)
        {
            // If there's data already there make sure the state is allowed
            if (property.UserState == null)
            {
                property.UserState = new PropertyData()
                {
                    ReceiveContext = receiveContext,
                    ChannelKey = channelKey,
                    BookmarkName = bookmarkName,
                    State = state
                };
            }
            else
            {
                PropertyData data = (PropertyData)property.UserState;

                // We should not buffer twice at the same state
                if (data.State == state)
                {
                    return false;
                }

                data.State = state;
            }

            return true;
        }

        void Initialize()
        {
            this.bufferedProperties = new Dictionary<InstanceKey, List<BufferedReceiveMessageProperty>>();
        }

        class PendingMessageThrottle
        {
            [Fx.Tag.SynchronizationObject(Blocking = false)]
            Dictionary<int, ThrottleEntry> pendingMessages;

            int maxPendingMessagesPerChannel;
            int warningRestoreLimit;

            public PendingMessageThrottle(int maxPendingMessagesPerChannel)
            {
                this.maxPendingMessagesPerChannel = maxPendingMessagesPerChannel;
                this.warningRestoreLimit = (int)Math.Floor(0.7 * (double)maxPendingMessagesPerChannel);
                this.pendingMessages = new Dictionary<int, ThrottleEntry>();
            }

            public bool Acquire(int channelKey)
            {
                lock (this.pendingMessages)
                {
                    if (!this.pendingMessages.ContainsKey(channelKey))
                    {
                        this.pendingMessages.Add(channelKey, new ThrottleEntry());
                    }

                    ThrottleEntry entry = this.pendingMessages[channelKey];
                    if (entry.Count < this.maxPendingMessagesPerChannel)
                    {
                        entry.Count++;
                        if (TD.PendingMessagesPerChannelRatioIsEnabled())
                        {
                            TD.PendingMessagesPerChannelRatio(entry.Count, this.maxPendingMessagesPerChannel);
                        }
                        return true;
                    }
                    else
                    {
                        if (TD.MaxPendingMessagesPerChannelExceededIsEnabled())
                        {
                            if (!entry.WarningIssued)
                            {
                                TD.MaxPendingMessagesPerChannelExceeded(this.maxPendingMessagesPerChannel);
                                entry.WarningIssued = true;
                            }
                        }

                        return false;
                    }
                }
            }

            public void Release(int channelKey)
            {
                lock (this.pendingMessages)
                {
                    ThrottleEntry entry = this.pendingMessages[channelKey];
                    Fx.Assert(entry.Count > 0, "The pending message throttle was released too many times");

                    entry.Count--;
                    if (TD.PendingMessagesPerChannelRatioIsEnabled())
                    {
                        TD.PendingMessagesPerChannelRatio(entry.Count, this.maxPendingMessagesPerChannel);
                    }
                    if (entry.Count == 0)
                    {
                        this.pendingMessages.Remove(channelKey);
                    }
                    else if (entry.Count < this.warningRestoreLimit)
                    {
                        entry.WarningIssued = false;
                    }
                }
            }

            class ThrottleEntry
            {
                public ThrottleEntry()
                {
                }

                public bool WarningIssued
                {
                    get;
                    set;
                }

                public int Count
                {
                    get;
                    set;
                }
            }
        }

        class PropertyData
        {
            public PropertyData()
            {
            }

            public ReceiveContext ReceiveContext
            {
                get;
                set;
            }

            public int ChannelKey
            {
                get;
                set;
            }

            public string BookmarkName
            {
                get;
                set;
            }

            public BufferedReceiveState State
            {
                get;
                set;
            }
        }
    }
}