< Summary - CoreWCF Coverage — PR #1766

Information
Class: CoreWCF.Runtime.SynchronizedPool<T>
Assembly: CoreWCF.Primitives
File(s): /home/runner/work/CoreWCF/CoreWCF/src/CoreWCF.Primitives/src/CoreWCF/Runtime/SynchronizedPool.cs
Line coverage
68%
Covered lines: 99
Uncovered lines: 45
Coverable lines: 144
Total lines: 430
Line coverage: 68.7%
Branch coverage
63%
Covered branches: 43
Total branches: 68
Branch coverage: 63.2%
Method coverage

Feature is only available for sponsors

Upgrade to PRO version

Metrics

MethodBranch coverage Cyclomatic complexity NPath complexity Sequence coverage
.ctor(...)100%22100%
Clear()0%220%
HandlePromotionFailure(...)0%220%
PromoteThread(...)50%6675%
RecordReturnToGlobalPool(...)90%101092.3%
RecordTakeFromGlobalPool(...)80%101088.23%
Return(...)50%4466.66%
ReturnToPerThreadPool(...)50%8860%
ReturnToGlobalPool(...)100%11100%
Take()50%4471.42%
TakeFromPerThreadPool(...)62.5%8854.54%
TakeFromGlobalPool(...)100%11100%
.cctor()100%11100%
GetProcessorCount()100%11100%
.ctor(...)100%11100%
DecrementMaxCount()50%2283.33%
Take()75%4485.71%
Return(...)75%4487.5%
Clear()100%110%

File(s)

/home/runner/work/CoreWCF/CoreWCF/src/CoreWCF.Primitives/src/CoreWCF/Runtime/SynchronizedPool.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;
 7
 8namespace CoreWCF.Runtime
 9{
 10    // A simple synchronized pool would simply lock a stack and push/pop on return/take.
 11    //
 12    // This implementation tries to reduce locking by exploiting the case where an item
 13    // is taken and returned by the same thread, which turns out to be common in our
 14    // scenarios.
 15    //
 16    // Initially, all the quota is allocated to a global (non-thread-specific) pool,
 17    // which takes locks.  As different threads take and return values, we record their IDs,
 18    // and if we detect that a thread is taking and returning "enough" on the same thread,
 19    // then we decide to "promote" the thread.  When a thread is promoted, we decrease the
 20    // quota of the global pool by one, and allocate a thread-specific entry for the thread
 21    // to store it's value.  Once this entry is allocated, the thread can take and return
 22    // it's value from that entry without taking any locks.  Not only does this avoid
 23    // locks, but it affinitizes pooled items to a particular thread.
 24    //
 25    // There are a couple of additional things worth noting:
 26    //
 27    // It is possible for a thread that we have reserved an entry for to exit.  This means
 28    // we will still have a entry allocated for it, but the pooled item stored there
 29    // will never be used.  After a while, we could end up with a number of these, and
 30    // as a result we would begin to exhaust the quota of the overall pool.  To mitigate this
 31    // case, we throw away the entire per-thread pool, and return all the quota back to
 32    // the global pool if we are unable to promote a thread (due to lack of space).  Then
 33    // the set of active threads will be re-promoted as they take and return items.
 34    //
 35    // You may notice that the code does not immediately promote a thread, and does not
 36    // immediately throw away the entire per-thread pool when it is unable to promote a
 37    // thread.  Instead, it uses counters (based on the number of calls to the pool)
 38    // and a threshold to figure out when to do these operations.  In the case where the
 39    // pool to misconfigured to have too few items for the workload, this avoids constant
 40    // promoting and rebuilding of the per thread entries.
 41    //
 42    // You may also notice that we do not use interlocked methods when adjusting statistics.
 43    // Since the statistics are a heuristic as to how often something is happening, they
 44    // do not need to be perfect.
 45    //
 46    internal class SynchronizedPool<T> where T : class
 47    {
 48        private const int maxPendingEntries = 128;
 49        private const int maxPromotionFailures = 64;
 50        private const int maxReturnsBeforePromotion = 64;
 51        private const int maxThreadItemsPerProcessor = 16;
 52        private Entry[] _entries;
 53        private readonly GlobalPool _globalPool;
 54        private readonly int _maxCount;
 55        private PendingEntry[] _pending;
 56        private int _promotionFailures;
 57
 718958        public SynchronizedPool(int maxCount)
 59        {
 718960            int threadCount = maxCount;
 718961            int maxThreadCount = maxThreadItemsPerProcessor + SynchronizedPoolHelper.ProcessorCount;
 718962            if (threadCount > maxThreadCount)
 63            {
 116564                threadCount = maxThreadCount;
 65            }
 718966            _maxCount = maxCount;
 718967            _entries = new Entry[threadCount];
 718968            _pending = new PendingEntry[4];
 718969            _globalPool = new GlobalPool(maxCount);
 718970        }
 71
 72        private object ThisLock
 73        {
 74            get
 75            {
 176                return this;
 77            }
 78        }
 79
 80        public void Clear()
 81        {
 082            Entry[] entries = _entries;
 83
 084            for (int i = 0; i < entries.Length; i++)
 85            {
 086                entries[i].value = null;
 87            }
 88
 089            _globalPool.Clear();
 090        }
 91
 92        private void HandlePromotionFailure(int thisThreadID)
 93        {
 094            int newPromotionFailures = _promotionFailures + 1;
 95
 096            if (newPromotionFailures >= maxPromotionFailures)
 97            {
 098                lock (ThisLock)
 99                {
 0100                    _entries = new Entry[_entries.Length];
 101
 0102                    _globalPool.MaxCount = _maxCount;
 0103                }
 104
 0105                PromoteThread(thisThreadID);
 106            }
 107            else
 108            {
 0109                _promotionFailures = newPromotionFailures;
 110            }
 0111        }
 112
 113        private bool PromoteThread(int thisThreadID)
 114        {
 1115            lock (ThisLock)
 116            {
 2117                for (int i = 0; i < _entries.Length; i++)
 118                {
 1119                    int threadID = _entries[i].threadID;
 120
 1121                    if (threadID == thisThreadID)
 122                    {
 0123                        return true;
 124                    }
 1125                    else if (threadID == 0)
 126                    {
 1127                        _globalPool.DecrementMaxCount();
 1128                        _entries[i].threadID = thisThreadID;
 1129                        return true;
 130                    }
 131                }
 0132            }
 133
 0134            return false;
 1135        }
 136
 137        private void RecordReturnToGlobalPool(int thisThreadID)
 138        {
 2980139            PendingEntry[] localPending = _pending;
 140
 10452141            for (int i = 0; i < localPending.Length; i++)
 142            {
 5211143                int threadID = localPending[i].threadID;
 144
 5211145                if (threadID == thisThreadID)
 146                {
 2767147                    int newReturnCount = localPending[i].returnCount + 1;
 148
 2767149                    if (newReturnCount >= maxReturnsBeforePromotion)
 150                    {
 1151                        localPending[i].returnCount = 0;
 152
 1153                        if (!PromoteThread(thisThreadID))
 154                        {
 0155                            HandlePromotionFailure(thisThreadID);
 156                        }
 157                    }
 158                    else
 159                    {
 2766160                        localPending[i].returnCount = newReturnCount;
 161                    }
 2766162                    break;
 163                }
 2444164                else if (threadID == 0)
 165                {
 166                    break;
 167                }
 168            }
 214169        }
 170
 171        private void RecordTakeFromGlobalPool(int thisThreadID)
 172        {
 7638173            PendingEntry[] localPending = _pending;
 174
 20504175            for (int i = 0; i < localPending.Length; i++)
 176            {
 10221177                int threadID = localPending[i].threadID;
 178
 10221179                if (threadID == thisThreadID)
 180                {
 4496181                    return;
 182                }
 5725183                else if (threadID == 0)
 184                {
 3113185                    lock (localPending)
 186                    {
 3113187                        if (localPending[i].threadID == 0)
 188                        {
 3111189                            localPending[i].threadID = thisThreadID;
 3111190                            return;
 191                        }
 2192                    }
 193                }
 194            }
 195
 31196            if (localPending.Length >= maxPendingEntries)
 197            {
 0198                _pending = new PendingEntry[localPending.Length];
 199            }
 200            else
 201            {
 31202                PendingEntry[] newPending = new PendingEntry[localPending.Length * 2];
 31203                Array.Copy(localPending, newPending, localPending.Length);
 31204                _pending = newPending;
 205            }
 3142206        }
 207
 208        public bool Return(T value)
 209        {
 2980210            int thisThreadID = Thread.CurrentThread.ManagedThreadId;
 211
 2980212            if (thisThreadID == 0)
 213            {
 0214                return false;
 215            }
 216
 2980217            if (ReturnToPerThreadPool(thisThreadID, value))
 218            {
 0219                return true;
 220            }
 221
 2980222            return ReturnToGlobalPool(thisThreadID, value);
 223        }
 224
 225        private bool ReturnToPerThreadPool(int thisThreadID, T value)
 226        {
 2980227            Entry[] entries = _entries;
 228
 5960229            for (int i = 0; i < entries.Length; i++)
 230            {
 2979231                int threadID = entries[i].threadID;
 232
 2979233                if (threadID == thisThreadID)
 234                {
 0235                    if (entries[i].value == null)
 236                    {
 0237                        entries[i].value = value;
 0238                        return true;
 239                    }
 240                    else
 241                    {
 0242                        return false;
 243                    }
 244                }
 2979245                else if (threadID == 0)
 246                {
 247                    break;
 248                }
 249            }
 250
 2980251            return false;
 252        }
 253
 254        private bool ReturnToGlobalPool(int thisThreadID, T value)
 255        {
 2980256            RecordReturnToGlobalPool(thisThreadID);
 257
 2980258            return _globalPool.Return(value);
 259        }
 260
 261        public T Take()
 262        {
 7638263            int thisThreadID = Thread.CurrentThread.ManagedThreadId;
 264
 7638265            if (thisThreadID == 0)
 266            {
 0267                return null;
 268            }
 269
 7638270            T value = TakeFromPerThreadPool(thisThreadID);
 271
 7638272            if (value != null)
 273            {
 0274                return value;
 275            }
 276
 7638277            return TakeFromGlobalPool(thisThreadID);
 278        }
 279
 280        private T TakeFromPerThreadPool(int thisThreadID)
 281        {
 7638282            Entry[] entries = _entries;
 283
 15278284            for (int i = 0; i < entries.Length; i++)
 285            {
 7638286                int threadID = entries[i].threadID;
 287
 7638288                if (threadID == thisThreadID)
 289                {
 0290                    T value = entries[i].value;
 291
 0292                    if (value != null)
 293                    {
 0294                        entries[i].value = null;
 0295                        return value;
 296                    }
 297                    else
 298                    {
 0299                        return null;
 300                    }
 301                }
 7638302                else if (threadID == 0)
 303                {
 304                    break;
 305                }
 306            }
 307
 7638308            return null;
 309        }
 310
 311        private T TakeFromGlobalPool(int thisThreadID)
 312        {
 7638313            RecordTakeFromGlobalPool(thisThreadID);
 314
 7638315            return _globalPool.Take();
 316        }
 317
 318        private struct Entry
 319        {
 320            public int threadID;
 321            public T value;
 322        }
 323
 324        private struct PendingEntry
 325        {
 326            public int returnCount;
 327            public int threadID;
 328        }
 329
 330        private static class SynchronizedPoolHelper
 331        {
 35332            public static readonly int ProcessorCount = GetProcessorCount();
 333
 334            private static int GetProcessorCount()
 335            {
 35336                return Environment.ProcessorCount;
 337            }
 338        }
 339
 340        private class GlobalPool
 341        {
 342            private readonly Stack<T> _items;
 343            private int _maxCount;
 344
 7189345            public GlobalPool(int maxCount)
 346            {
 7189347                _items = new Stack<T>();
 7189348                _maxCount = maxCount;
 7189349            }
 350
 351            public int MaxCount
 352            {
 353                get
 354                {
 5708355                    return _maxCount;
 356                }
 357                set
 358                {
 0359                    lock (ThisLock)
 360                    {
 0361                        while (_items.Count > value)
 362                        {
 0363                            _items.Pop();
 364                        }
 0365                        _maxCount = value;
 0366                    }
 0367                }
 368            }
 369
 370            private object ThisLock
 371            {
 372                get
 373                {
 3845374                    return this;
 375                }
 376            }
 377
 378            public void DecrementMaxCount()
 379            {
 1380                lock (ThisLock)
 381                {
 1382                    if (_items.Count == _maxCount)
 383                    {
 0384                        _items.Pop();
 385                    }
 1386                    _maxCount--;
 1387                }
 1388            }
 389
 390            public T Take()
 391            {
 7638392                if (_items.Count > 0)
 393                {
 1116394                    lock (ThisLock)
 395                    {
 1116396                        if (_items.Count > 0)
 397                        {
 1116398                            return _items.Pop();
 399                        }
 0400                    }
 401                }
 6522402                return null;
 1116403            }
 404
 405            public bool Return(T value)
 406            {
 2980407                if (_items.Count < MaxCount)
 408                {
 2728409                    lock (ThisLock)
 410                    {
 2728411                        if (_items.Count < MaxCount)
 412                        {
 2728413                            _items.Push(value);
 2728414                            return true;
 415                        }
 0416                    }
 417                }
 252418                return false;
 2728419            }
 420
 421            public void Clear()
 422            {
 0423                lock (ThisLock)
 424                {
 0425                    _items.Clear();
 0426                }
 0427            }
 428        }
 429    }
 430}
Generated by: ReportGenerator 5.5.11.0
08/09/2026 - 10:15:49
GitHub | reportgenerator.io

Methods/Properties

.ctor(System.Int32)
ThisLock()
Clear()
HandlePromotionFailure(System.Int32)
PromoteThread(System.Int32)
RecordReturnToGlobalPool(System.Int32)
RecordTakeFromGlobalPool(System.Int32)
Return(T)
ReturnToPerThreadPool(System.Int32,T)
ReturnToGlobalPool(System.Int32,T)
Take()
TakeFromPerThreadPool(System.Int32)
TakeFromGlobalPool(System.Int32)
.cctor()
GetProcessorCount()
.ctor(System.Int32)
MaxCount()
MaxCount(System.Int32)
ThisLock()
DecrementMaxCount()
Take()
Return(T)
Clear()