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 a long
  45                 long 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                         AddHint (index);
  69                         Interlocked.Increment (ref count);
  70                 }
  71
  72                 bool IProducerConsumerCollection<T>.TryAdd (T element)
  73                 {
  74                         Add (element);
  75                         return true;
  76                 }
  77
  78                 public bool TryTake (out T result)
  79                 {
  80                         result = default (T);
  81
  82                         if (count == 0)
  83                                 return false;
  84
  85                         int hintIndex;
  86                         CyclicDeque<T> bag = GetBag (out hintIndex, false);
  87                         bool ret = true;
  88
  89                         if (bag == null || bag.PopBottom (out result) != PopResult.Succeed) {
  90                                 foreach (var other in staging) {
  91                                         // Try to retrieve something based on a hint
  92                                         ret = TryGetHint (out hintIndex) && (bag = container[hintIndex]).PopTop (out result) == PopResult.Succeed;
  93
  94                                         // We fall back to testing our slot
  95                                         if (!ret && other.Value != bag) {
  96                                                 ret = other.Value.PopTop (out result) == PopResult.Succeed;
  97                                                 hintIndex = other.Key;
  98                                                 bag = other.Value;
  99                                         }
 100
 101                                         // If we found something, stop
 102                                         if (ret)
 103                                                 break;
 104                                 }
 105                         }
 106
 107                         if (ret) {
 108                                 TidyBag (hintIndex, bag);
 109                                 Interlocked.Decrement (ref count);
 110                         }
 111
 112                         return ret;
 113                 }
 114
 115                 public bool TryPeek (out T result)
 116                 {
 117                         result = default (T);
 118
 119                         if (count == 0)
 120                                 return false;
 121
 122                         int hintIndex;
 123                         CyclicDeque<T> bag = GetBag (out hintIndex, false);
 124                         bool ret = true;
 125
 126                         if (bag == null || !bag.PeekBottom (out result)) {
 127                                 foreach (var other in staging) {
 128                                         // Try to retrieve something based on a hint
 129                                         ret = TryGetHint (out hintIndex) && container[hintIndex].PeekTop (out result);
 130
 131                                         // We fall back to testing our slot
 132                                         if (!ret && other.Value != bag)
 133                                                 ret = other.Value.PeekTop (out result);
 134
 135                                         // If we found something, stop
 136                                         if (ret)
 137                                                 break;
 138                                 }
 139                         }
 140
 141                         return ret;
 142                 }
 143
 144                 void AddHint (int index)
 145                 {
 146                         // We only take thread index that can be stored in 5 bits (i.e. thread ids 1-15)
 147                         if (index > 0xF)
 148                                 return;
 149                         var hs = hints;
 150                         // If cas failed then we don't retry
 151                         Interlocked.CompareExchange (ref hints, (hs << 4) | (uint)index, hs);
 152                 }
 153
 154                 bool TryGetHint (out int index)
 155                 {
 156                         var hs = hints;
 157                         index = 0;
 158
 159                         if (Interlocked.CompareExchange (ref hints, hs >> 4, hs) == hs)
 160                                 index = (int)(hs & 0xF);
 161
 162                         return index > 0;
 163                 }
 164
 165                 public int Count {
 166                         get {
 167                                 return count;
 168                         }
 169                 }
 170
 171                 public bool IsEmpty {
 172                         get {
 173                                 return count == 0;
 174                         }
 175                 }
 176
 177                 object System.Collections.ICollection.SyncRoot  {
 178                         get {
 179                                 return this;
 180                         }
 181                 }
 182
 183                 bool System.Collections.ICollection.IsSynchronized  {
 184                         get {
 185                                 return true;
 186                         }
 187                 }
 188
 189                 IEnumerator IEnumerable.GetEnumerator ()
 190                 {
 191                         return GetEnumeratorInternal ();
 192                 }
 193
 194                 public IEnumerator<T> GetEnumerator ()
 195                 {
 196                         return GetEnumeratorInternal ();
 197                 }
 198
 199                 IEnumerator<T> GetEnumeratorInternal ()
 200                 {
 201                         foreach (var bag in container)
 202                                 foreach (T item in bag.Value.GetEnumerable ())
 203                                         yield return item;
 204                 }
 205
 206                 void System.Collections.ICollection.CopyTo (Array array, int index)
 207                 {
 208                         T[] a = array as T[];
 209                         if (a == null)
 210                                 return;
 211
 212                         CopyTo (a, index);
 213                 }
 214
 215                 public void CopyTo (T[] array, int index)
 216                 {
 217                         int c = count;
 218                         if (array.Length < c + index)
 219                                 throw new InvalidOperationException ("Array is not big enough");
 220
 221                         CopyTo (array, index, c);
 222                 }
 223
 224                 void CopyTo (T[] array, int index, int num)
 225                 {
 226                         int i = index;
 227
 228                         foreach (T item in this) {
 229                                 if (i >= num)
 230                                         break;
 231
 232                                 array[i++] = item;
 233                         }
 234                 }
 235
 236                 public T[] ToArray ()
 237                 {
 238                         int c = count;
 239                         T[] temp = new T[c];
 240
 241                         CopyTo (temp, 0, c);
 242
 243                         return temp;
 244                 }
 245
 246                 int GetIndex ()
 247                 {
 248                         return Thread.CurrentThread.ManagedThreadId;
 249                 }
 250
 251                 CyclicDeque<T> GetBag (out int index, bool createBag = true)
 252                 {
 253                         index = GetIndex ();
 254                         CyclicDeque<T> value;
 255                         if (container.TryGetValue (index, out value))
 256                                 return value;
 257
 258                         var bag = createBag ? container.GetOrAdd (index, new CyclicDeque<T> ()) : null;
 259                         if (bag != null)
 260                                 staging.TryAdd (index, bag);
 261                         return bag;
 262                 }
 263
 264                 void TidyBag (int index, CyclicDeque<T> bag)
 265                 {
 266                         if (bag != null && bag.IsEmpty) {
 267                                 if (staging.TryRemove (index, out bag) && !bag.IsEmpty)
 268                                         staging.TryAdd (index, bag);
 269                         }
 270                 }
 271         }
 272 }
 273 #endif