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

   1 #if NET_4_0
   2 //
   3 // ConcurrentBag.cs
   4 //
   5 // Author:
   6 //       Jérémie "Garuma" Laval <jeremie.laval@gmail.com>
   7 //
   8 // Copyright (c) 2009 Jérémie "Garuma" Laval
   9 //
  10 // Permission is hereby granted, free of charge, to any person obtaining a copy
  11 // of this software and associated documentation files (the "Software"), to deal
  12 // in the Software without restriction, including without limitation the rights
  13 // to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
  14 // copies of the Software, and to permit persons to whom the Software is
  15 // furnished to do so, subject to the following conditions:
  16 //
  17 // The above copyright notice and this permission notice shall be included in
  18 // all copies or substantial portions of the Software.
  19 //
  20 // THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
  21 // IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
  22 // FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
  23 // AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
  24 // LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
  25 // OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
  26 // THE SOFTWARE.
  27
  28 using System;
  29 using System.Collections;
  30 using System.Collections.Generic;
  31
  32 using System.Threading;
  33 using System.Threading.Tasks;
  34
  35 namespace System.Collections.Concurrent
  36 {
  37
  38         public class ConcurrentBag<T> : IProducerConsumerCollection<T>, IEnumerable<T>, IEnumerable
  39         {
  40                 int size = Environment.ProcessorCount + 1;
  41                 int multiplier = 2;
  42                 int count;
  43
  44                 CyclicDeque<T>[] container;
  45
  46                 object syncLock = new object ();
  47
  48                 public ConcurrentBag ()
  49                 {
  50                         container = new CyclicDeque<T>[size];
  51                         for (int i = 0; i < container.Length; i++)
  52                                 container[i] = new CyclicDeque<T> ();
  53                 }
  54
  55                 public ConcurrentBag (IEnumerable<T> enumerable) : this ()
  56                 {
  57                         foreach (T item in enumerable)
  58                                 Add (item);
  59                 }
  60
  61                 public bool TryAdd (T item)
  62                 {
  63                         Add (item);
  64
  65                         return true;
  66                 }
  67
  68                 public void Add (T item)
  69                 {
  70                         Interlocked.Increment (ref count);
  71                         GrowIfNecessary ();
  72
  73                         CyclicDeque<T> bag = GetBag ();
  74                         bag.PushBottom (item);
  75                 }
  76
  77                 public bool TryTake (out T item)
  78                 {
  79                         item = default (T);
  80                         CyclicDeque<T> bag = GetBag ();
  81
  82                         if (bag == null || bag.PopBottom (out item) != PopResult.Succeed) {
  83                                 for (int i = 0; i < container.Length; i++) {
  84                                         if (container[i].PopTop (out item) == PopResult.Succeed) {
  85                                                 Interlocked.Decrement (ref count);
  86                                                 return true;
  87                                         }
  88                                 }
  89                         } else {
  90                                 Interlocked.Decrement (ref count);
  91                                 return true;
  92                         }
  93
  94                         return false;
  95                 }
  96
  97                 public int Count {
  98                         get {
  99                                 return count;
 100                         }
 101                 }
 102
 103                 public bool IsEmpty {
 104                         get {
 105                                 return count == 0;
 106                         }
 107                 }
 108
 109                 public object SyncRoot  {
 110                         get {
 111                                 return this;
 112                         }
 113                 }
 114
 115                 public bool IsSynchronized  {
 116                         get {
 117                                 return true;
 118                         }
 119                 }
 120
 121                 IEnumerator IEnumerable.GetEnumerator ()
 122                 {
 123                         return GetEnumeratorInternal ();
 124                 }
 125
 126                 IEnumerator<T> IEnumerable<T>.GetEnumerator ()
 127                 {
 128                         return GetEnumeratorInternal ();
 129                 }
 130
 131                 IEnumerator<T> GetEnumeratorInternal ()
 132                 {
 133                         for (int i = 0; i < size; i++) {
 134                                 CyclicDeque<T> bag = container[i];
 135                                 foreach (T item in bag.GetEnumerable ()) {
 136                                         yield return item;
 137                                 }
 138                         }
 139                 }
 140
 141                 public void CopyTo (Array array, int index)
 142                 {
 143                         T[] a = array as T[];
 144                         if (a == null)
 145                                 return;
 146
 147                         CopyTo (a, index);
 148                 }
 149
 150                 public void CopyTo (T[] array, int index)
 151                 {
 152                         int c = count;
 153                         if (array.Length < c + index)
 154                                 throw new InvalidOperationException ("Array is not big enough");
 155
 156                         CopyTo (array, index, c);
 157                 }
 158
 159                 void CopyTo (T[] array, int index, int num)
 160                 {
 161                         int i = index;
 162
 163                         foreach (T item in this) {
 164                                 if (i >= num)
 165                                         break;
 166
 167                                 array[i++] = item;
 168                         }
 169                 }
 170
 171                 public T[] ToArray ()
 172                 {
 173                         int c = count;
 174                         T[] temp = new T[c];
 175
 176                         CopyTo (temp, 0, c);
 177
 178                         return temp;
 179                 }
 180
 181                 int GetIndex ()
 182                 {
 183                         return Thread.CurrentThread.ManagedThreadId - 1;
 184                 }
 185
 186                 void GrowIfNecessary ()
 187                 {
 188                         int index = GetIndex ();
 189                         int currentSize = size;
 190
 191                         while (index > currentSize - 1) {
 192                                 currentSize = size;
 193                                 Grow (currentSize);
 194                         }
 195                 }
 196
 197                 CyclicDeque<T> GetBag ()
 198                 {
 199                         int i = GetIndex ();
 200
 201                         return i < container.Length ? container[i] : null;
 202                 }
 203
 204                 void Grow (int referenceSize)
 205                 {
 206                         lock (syncLock) {
 207                                 if (referenceSize != size)
 208                                         return;
 209
 210                                 CyclicDeque<T>[] slice = new CyclicDeque<T>[size * multiplier];
 211                                 int i = 0;
 212                                 for (i = 0; i < container.Length; i++)
 213                                         slice[i] = container[i];
 214                                 for (; i < slice.Length; i++)
 215                                         slice[i] = new CyclicDeque<T> ();
 216
 217                                 container = slice;
 218                                 size = slice.Length;
 219                         }
 220                 }
 221         }
 222 }
 223 #endif