< Summary - CoreWCF Coverage — PR #1766

Information
Class: CoreWCF.Dispatcher.ConcurrencyBehavior
Assembly: CoreWCF.Primitives
File(s): /home/runner/work/CoreWCF/CoreWCF/src/CoreWCF.Primitives/src/CoreWCF/Dispatcher/ConcurrencyBehavior.cs
Line coverage
58%
Covered lines: 46
Uncovered lines: 33
Coverable lines: 79
Total lines: 302
Line coverage: 58.2%
Branch coverage
69%
Covered branches: 29
Total branches: 42
Branch coverage: 69%
Method coverage

Feature is only available for sponsors

Upgrade to PRO version

Metrics

MethodBranch coverage Cyclomatic complexity NPath complexity Sequence coverage
.ctor(...)100%11100%
IsConcurrent(...)100%11100%
IsConcurrent(...)83.33%6685.71%
IsConcurrent(...)90%101083.33%
LockInstanceAsync()62.5%8876.92%
UnlockInstance(...)100%22100%
UnlockInstanceBeforeCallout(...)75%4466.66%
UnlockInstance(...)50%2285.71%
LockInstanceAfterCalloutAsync(...)50%8833.33%
.ctor(...)100%110%
CoreWCF.Dispatcher.ConcurrencyBehavior.IWaiter.Signal()0%220%
.ctor()100%110%
CoreWCF.Dispatcher.ConcurrencyBehavior.IWaiter.Signal()100%110%
Wait()100%110%

File(s)

/home/runner/work/CoreWCF/CoreWCF/src/CoreWCF.Primitives/src/CoreWCF/Dispatcher/ConcurrencyBehavior.cs

#LineLine coverage
 1// Licensed to the .NET Foundation under one or more agreements.
 2// The .NET Foundation licenses this file to you under the MIT license.
 3
 4using System;
 5using System.Collections.Generic;
 6using System.Threading;
 7using System.Threading.Tasks;
 8using CoreWCF.Runtime;
 9
 10namespace CoreWCF.Dispatcher
 11{
 12    internal class ConcurrencyBehavior
 13    {
 14        private readonly ConcurrencyMode _concurrencyMode;
 15        private readonly bool _enforceOrderedReceive;
 16
 66117        internal ConcurrencyBehavior(DispatchRuntime runtime)
 18        {
 66119            _concurrencyMode = runtime.ConcurrencyMode;
 66120            _enforceOrderedReceive = runtime.EnsureOrderedDispatch;
 21            //this.supportsTransactedBatch = ConcurrencyBehavior.SupportsTransactedBatch(runtime.ChannelDispatcher);
 66122        }
 23
 24        internal bool IsConcurrent(MessageRpc rpc)
 25        {
 744926            return IsConcurrent(_concurrencyMode, _enforceOrderedReceive, rpc.Channel.HasSession/*, this.supportsTransac
 27        }
 28
 29        internal static bool IsConcurrent(ConcurrencyMode concurrencyMode, bool ensureOrderedDispatch, bool hasSession /
 30        {
 31            //if (supportsTransactedBatch)
 32            //{
 33            //    return false;
 34            //}
 35
 744936            if (concurrencyMode != ConcurrencyMode.Single)
 37            {
 62138                return true;
 39            }
 40
 682841            if (hasSession)
 42            {
 17443                return false;
 44            }
 45
 665446            if (ensureOrderedDispatch)
 47            {
 048                return false;
 49            }
 50
 665451            return true;
 52        }
 53
 54        internal static bool IsConcurrent(ChannelDispatcher runtime, bool hasSession)
 55        {
 231856            bool isConcurrencyModeSingle = true;
 57
 58            //if (ConcurrencyBehavior.SupportsTransactedBatch(runtime))
 59            //{
 60            //    return false;
 61            //}
 62
 927463            foreach (EndpointDispatcher endpointDispatcher in runtime.Endpoints)
 64            {
 231965                if (endpointDispatcher.DispatchRuntime.EnsureOrderedDispatch)
 66                {
 067                    return false;
 68                }
 69
 231970                if (endpointDispatcher.DispatchRuntime.ConcurrencyMode != ConcurrencyMode.Single)
 71                {
 8772                    isConcurrencyModeSingle = false;
 73                }
 74            }
 75
 231876            if (!isConcurrencyModeSingle)
 77            {
 8778                return true;
 79            }
 80
 223181            if (!hasSession)
 82            {
 219183                return true;
 84            }
 85
 4086            return false;
 087        }
 88
 89        internal async Task LockInstanceAsync(MessageRpc rpc)
 90        {
 253691            if (_concurrencyMode != ConcurrencyMode.Multiple)
 92            {
 232793                ConcurrencyInstanceContextFacet resource = rpc.InstanceContext.Concurrency;
 232794                Task waiter = null;
 232795                lock (rpc.InstanceContext.ThisLock)
 96                {
 232797                    if (!resource.Locked)
 98                    {
 232799                        resource.Locked = true;
 100                    }
 101                    else
 102                    {
 0103                        waiter = resource.EnqueueNewMessage();
 104                    }
 2327105                }
 106
 2327107                if (waiter != null)
 108                {
 0109                    await waiter;
 110                }
 111
 112                // TODO: Throw this on setup
 2327113                if (_concurrencyMode == ConcurrencyMode.Reentrant)
 114                {
 0115                    throw new NotSupportedException(nameof(ConcurrencyMode.Reentrant));
 116                }
 117            }
 2536118        }
 119
 120        internal void UnlockInstance(ref MessageRpc rpc)
 121        {
 2536122            if (_concurrencyMode != ConcurrencyMode.Multiple)
 123            {
 2327124                UnlockInstance(rpc.InstanceContext);
 125            }
 2536126        }
 127
 128        internal static void UnlockInstanceBeforeCallout(OperationContext operationContext)
 129        {
 1130            if (operationContext != null && operationContext.IsServiceReentrant)
 131            {
 0132                UnlockInstance(operationContext.InstanceContext);
 133            }
 1134        }
 135
 136        private static void UnlockInstance(InstanceContext instanceContext)
 137        {
 2327138            ConcurrencyInstanceContextFacet resource = instanceContext.Concurrency;
 139
 2327140            lock (instanceContext.ThisLock)
 141            {
 2327142                if (resource.HasWaiters)
 143                {
 0144                    resource.DequeueWaiter();
 145                }
 146                else
 147                {
 148                    //We have no pending Callouts and no new Messages to process
 2327149                    resource.Locked = false;
 150                }
 2327151            }
 2327152        }
 153
 154        // TODO: Make async to remove blocking Wait call
 155        internal static Task LockInstanceAfterCalloutAsync(OperationContext operationContext)
 156        {
 1157            if (operationContext != null)
 158            {
 1159                InstanceContext instanceContext = operationContext.InstanceContext;
 160
 1161                if (operationContext.IsServiceReentrant)
 162                {
 0163                    ConcurrencyInstanceContextFacet resource = instanceContext.Concurrency;
 0164                    bool needToWait = false;
 0165                    lock (instanceContext.ThisLock)
 166                    {
 0167                        if (!resource.Locked)
 168                        {
 0169                            resource.Locked = true;
 170                        }
 0171                    }
 172
 0173                    if (needToWait)
 174                    {
 0175                        return resource.EnqueueCalloutMessage();
 176                    }
 177                }
 178            }
 179
 1180            return Task.CompletedTask;
 181        }
 182
 183        internal interface IWaiter
 184        {
 185            void Signal();
 186        }
 187
 188        private class MessageRpcWaiter : IWaiter
 189        {
 190            private readonly IResumeMessageRpc _resume;
 191
 0192            internal MessageRpcWaiter(IResumeMessageRpc resume)
 193            {
 0194                _resume = resume;
 0195            }
 196
 197            void IWaiter.Signal()
 198            {
 199                try
 200                {
 0201                    _resume.Resume(out bool alreadyResumedNoLock);
 202
 0203                    if (alreadyResumedNoLock)
 204                    {
 205                        Fx.Assert("ConcurrencyBehavior resumed more than once for same call.");
 206                    }
 0207                }
 0208                catch (Exception e)
 209                {
 0210                    if (Fx.IsFatal(e))
 211                    {
 0212                        throw;
 213                    }
 0214                    throw DiagnosticUtility.ExceptionUtility.ThrowHelperCallback(e);
 215                }
 0216            }
 217        }
 218
 219        private class ThreadWaiter : IWaiter
 220        {
 0221            private readonly ManualResetEvent _wait = new ManualResetEvent(false);
 222
 223            void IWaiter.Signal()
 224            {
 0225                _wait.Set();
 0226            }
 227
 228            internal void Wait()
 229            {
 0230                _wait.WaitOne();
 0231                _wait.Dispose();
 0232            }
 233        }
 234    }
 235
 236    internal class ConcurrencyInstanceContextFacet
 237    {
 238        internal bool Locked;
 239        private Queue<TaskCompletionSource<object>> _calloutMessageQueue;
 240        private Queue<TaskCompletionSource<object>> _newMessageQueue;
 241
 242        internal bool HasWaiters
 243        {
 244            get
 245            {
 246                return (((_calloutMessageQueue != null) && (_calloutMessageQueue.Count > 0)) ||
 247                        ((_newMessageQueue != null) && (_newMessageQueue.Count > 0)));
 248            }
 249        }
 250
 251        private TaskCompletionSource<object> DequeueFrom(Queue<TaskCompletionSource<object>> queue)
 252        {
 253            TaskCompletionSource<object> waiter = queue.Dequeue();
 254
 255            if (queue.Count == 0)
 256            {
 257                queue.TrimExcess();
 258            }
 259
 260            return waiter;
 261        }
 262
 263        internal void DequeueWaiter()
 264        {
 265            TaskCompletionSource<object> waiter;
 266            if ((_calloutMessageQueue != null) && (_calloutMessageQueue.Count > 0))
 267            {
 268                waiter = DequeueFrom(_calloutMessageQueue);
 269            }
 270            else
 271            {
 272                waiter = DequeueFrom(_newMessageQueue);
 273            }
 274
 275            waiter.TrySetResult(null);
 276        }
 277
 278        internal Task EnqueueNewMessage()
 279        {
 280            if (_newMessageQueue == null)
 281            {
 282                _newMessageQueue = new Queue<TaskCompletionSource<object>>();
 283            }
 284            // Prevent release of waiter from running the waiter on the releasing thread by using RunContinuationsAsynch
 285            var waiter = new TaskCompletionSource<object>(TaskCreationOptions.RunContinuationsAsynchronously);
 286            _newMessageQueue.Enqueue(waiter);
 287            return waiter.Task;
 288        }
 289
 290        internal Task EnqueueCalloutMessage()
 291        {
 292            if (_calloutMessageQueue == null)
 293            {
 294                _calloutMessageQueue = new Queue<TaskCompletionSource<object>>();
 295            }
 296            // Prevent release of waiter from running the waiter on the releasing thread by using RunContinuationsAsynch
 297            var waiter = new TaskCompletionSource<object>(TaskCreationOptions.RunContinuationsAsynchronously);
 298            _calloutMessageQueue.Enqueue(waiter);
 299            return waiter.Task;
 300        }
 301    }
 302}
Generated by: ReportGenerator 5.5.11.0
08/09/2026 - 10:15:49
GitHub | reportgenerator.io

Methods/Properties

.ctor(CoreWCF.Dispatcher.DispatchRuntime)
IsConcurrent(CoreWCF.Dispatcher.MessageRpc)
IsConcurrent(CoreWCF.ConcurrencyMode,System.Boolean,System.Boolean)
IsConcurrent(CoreWCF.Dispatcher.ChannelDispatcher,System.Boolean)
LockInstanceAsync()
UnlockInstance(CoreWCF.Dispatcher.MessageRpc&)
UnlockInstanceBeforeCallout(CoreWCF.OperationContext)
UnlockInstance(CoreWCF.InstanceContext)
LockInstanceAfterCalloutAsync(CoreWCF.OperationContext)
.ctor(CoreWCF.Dispatcher.IResumeMessageRpc)
CoreWCF.Dispatcher.ConcurrencyBehavior.IWaiter.Signal()
.ctor()
CoreWCF.Dispatcher.ConcurrencyBehavior.IWaiter.Signal()
Wait()