mcs/class/System/System.Collections.Concurrent/ConcurrentBag.cs

   1 //
   2 // ConcurrentBag.cs
   3 //
   4 // Author:
   5 //       Jérémie "Garuma" Laval <jeremie.laval@gmail.com>
   6 //
   7 // Copyright (c) 2009 Jérémie "Garuma" Laval
   8 //
   9 // Permission is hereby granted, free of charge, to any person obtaining a copy
  10 // of this software and associated documentation files (the "Software"), to deal
  11 // in the Software without restriction, including without limitation the rights
  12 // to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
  13 // copies of the Software, and to permit persons to whom the Software is
  14 // furnished to do so, subject to the following conditions:
  15 //
  16 // The above copyright notice and this permission notice shall be included in
  17 // all copies or substantial portions of the Software.
  18 //
  19 // THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
  20 // IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
  21 // FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
  22 // AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
  23 // LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
  24 // OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
  25 // THE SOFTWARE.
  26
  27 #if NET_4_0
  28 using System;
  29 using System.Collections;
  30 using System.Collections.Generic;
  31 using System.Diagnostics;
  32 using System.Runtime.InteropServices;
  33
  34 using System.Threading;
  35 using System.Threading.Tasks;
  36
  37 namespace System.Collections.Concurrent
  38 {
  39         [ComVisible (false)]
  40         [DebuggerDisplay ("Count={Count}")]
  41         [DebuggerTypeProxy (typeof (CollectionDebuggerView<>))]
  42         public class ConcurrentBag<T> : IProducerConsumerCollection<T>, IEnumerable<T>, IEnumerable
  43         {
  44                 // We store hints in an int
  45                 int hints;
  46
  47                 int count;
  48                 // The container area is where bag are added foreach thread
  49                 ConcurrentDictionary<int, CyclicDeque<T>> container = new ConcurrentDictionary<int, CyclicDeque<T>> ();
  50                 // The staging area is where non-empty bag are located for fast iteration
  51                 ConcurrentDictionary<int, CyclicDeque<T>> staging = new ConcurrentDictionary<int, CyclicDeque<T>> ();
  52
  53                 public ConcurrentBag ()
  54                 {
  55                 }
  56
  57                 public ConcurrentBag (IEnumerable<T> collection) : this ()
  58                 {
  59                         foreach (T item in collection)
  60                                 Add (item);
  61                 }
  62
  63                 public void Add (T item)
  64                 {
  65                         int index;
  66                         CyclicDeque<T> bag = GetBag (out index);
  67                         bag.PushBottom (item);
  68                         staging.TryAdd (index, bag);
  69                         AddHint (index);
  70                         Interlocked.Increment (ref count);
  71                 }
  72
  73                 bool IProducerConsumerCollection<T>.TryAdd (T element)
  74                 {
  75                         Add (element);
  76                         return true;
  77                 }
  78
  79                 public bool TryTake (out T result)
  80                 {
  81                         result = default (T);
  82
  83                         if (count == 0)
  84                                 return false;
  85
  86                         int hintIndex;
  87                         CyclicDeque<T> bag = GetBag (out hintIndex, false);
  88                         bool ret = true;
  89
  90                         if (bag == null || bag.PopBottom (out result) != PopResult.Succeed) {
  91                                 var self = bag;
  92                                 ret = false;
  93                                 foreach (var other in staging) {
  94                                         // Try to retrieve something based on a hint
  95                                         ret = TryGetHint (out hintIndex) && (bag = container[hintIndex]).PopTop (out result) == PopResult.Succeed;
  96
  97                                         // We fall back to testing our slot
  98                                         if (!ret && other.Value != self) {
  99                                                 var status = other.Value.PopTop (out result);
 100                                                 while (status == PopResult.Abort)
 101                                                         status = other.Value.PopTop (out result);
 102                                                 ret = status == PopResult.Succeed;
 103                                                 hintIndex = other.Key;
 104                                                 bag = other.Value;
 105                                         }
 106
 107                                         // If we found something, stop
 108                                         if (ret)
 109                                                 break;
 110                                 }
 111                         }
 112
 113                         if (ret) {
 114                                 TidyBag (hintIndex, bag);
 115                                 Interlocked.Decrement (ref count);
 116                         }
 117
 118                         return ret;
 119                 }
 120
 121                 public bool TryPeek (out T result)
 122                 {
 123                         result = default (T);
 124
 125                         if (count == 0)
 126                                 return false;
 127
 128                         int hintIndex;
 129                         CyclicDeque<T> bag = GetBag (out hintIndex, false);
 130                         bool ret = true;
 131
 132                         if (bag == null || !bag.PeekBottom (out result)) {
 133                                 var self = bag;
 134                                 ret = false;
 135                                 foreach (var other in staging) {
 136                                         // Try to retrieve something based on a hint
 137                                         ret = TryGetHint (out hintIndex) && container[hintIndex].PeekTop (out result);
 138
 139                                         // We fall back to testing our slot
 140                                         if (!ret && other.Value != self)
 141                                                 ret = other.Value.PeekTop (out result);
 142
 143                                         // If we found something, stop
 144                                         if (ret)
 145                                                 break;
 146                                 }
 147                         }
 148
 149                         return ret;
 150                 }
 151
 152                 void AddHint (int index)
 153                 {
 154                         // We only take thread index that can be stored in 5 bits (i.e. thread ids 1-15)
 155                         if (index > 0xF)
 156                                 return;
 157                         var hs = hints;
 158                         // If cas failed then we don't retry
 159                         Interlocked.CompareExchange (ref hints, (int)(((uint)hs) << 4 | (uint)index), (int)hs);
 160                 }
 161
 162                 bool TryGetHint (out int index)
 163                 {
 164                         /* Funny little thing to know, since hints is signed (because CAS has no uint overload),
 165                          * a shift-right operation is an arithmetic shift which might set high-order right bits
 166                          * to 1 instead of 0 if the number turns negative.
 167                          */
 168                         var hs = hints;
 169                         index = 0;
 170
 171                         if (Interlocked.CompareExchange (ref hints, (int)(((uint)hs) >> 4), hs) == hs)
 172                                 index = (int)(hs & 0xF);
 173
 174                         return index > 0;
 175                 }
 176
 177                 public int Count {
 178                         get {
 179                                 return count;
 180                         }
 181                 }
 182
 183                 public bool IsEmpty {
 184                         get {
 185                                 return count == 0;
 186                         }
 187                 }
 188
 189                 object System.Collections.ICollection.SyncRoot  {
 190                         get {
 191                                 return this;
 192                         }
 193                 }
 194
 195                 bool System.Collections.ICollection.IsSynchronized  {
 196                         get {
 197                                 return true;
 198                         }
 199                 }
 200
 201                 IEnumerator IEnumerable.GetEnumerator ()
 202                 {
 203                         return GetEnumeratorInternal ();
 204                 }
 205
 206                 public IEnumerator<T> GetEnumerator ()
 207                 {
 208                         return GetEnumeratorInternal ();
 209                 }
 210
 211                 IEnumerator<T> GetEnumeratorInternal ()
 212                 {
 213                         foreach (var bag in container)
 214                                 foreach (T item in bag.Value.GetEnumerable ())
 215                                         yield return item;
 216                 }
 217
 218                 void System.Collections.ICollection.CopyTo (Array array, int index)
 219                 {
 220                         T[] a = array as T[];
 221                         if (a == null)
 222                                 return;
 223
 224                         CopyTo (a, index);
 225                 }
 226
 227                 public void CopyTo (T[] array, int index)
 228                 {
 229                         int c = count;
 230                         if (array.Length < c + index)
 231                                 throw new InvalidOperationException ("Array is not big enough");
 232
 233                         CopyTo (array, index, c);
 234                 }
 235
 236                 void CopyTo (T[] array, int index, int num)
 237                 {
 238                         int i = index;
 239
 240                         foreach (T item in this) {
 241                                 if (i >= num)
 242                                         break;
 243
 244                                 array[i++] = item;
 245                         }
 246                 }
 247
 248                 public T[] ToArray ()
 249                 {
 250                         int c = count;
 251                         T[] temp = new T[c];
 252
 253                         CopyTo (temp, 0, c);
 254
 255                         return temp;
 256                 }
 257
 258                 int GetIndex ()
 259                 {
 260                         return Thread.CurrentThread.ManagedThreadId;
 261                 }
 262
 263                 CyclicDeque<T> GetBag (out int index, bool createBag = true)
 264                 {
 265                         index = GetIndex ();
 266                         CyclicDeque<T> value;
 267                         if (container.TryGetValue (index, out value))
 268                                 return value;
 269
 270                         return createBag ? container.GetOrAdd (index, new CyclicDeque<T> ()) : null;
 271                 }
 272
 273                 void TidyBag (int index, CyclicDeque<T> bag)
 274                 {
 275                         if (bag != null && bag.IsEmpty) {
 276                                 if (staging.TryRemove (index, out bag) && !bag.IsEmpty)
 277                                         staging.TryAdd (index, bag);
 278                         }
 279                 }
 280         }
 281 }
 282 #endif