Mercurial > pub > ImplabNet
annotate Implab/Parallels/AsyncQueue.cs @ 123:f4d6ea6969cc v2
async queue improvements
| author | cin |
|---|---|
| date | Tue, 13 Jan 2015 01:42:38 +0300 |
| parents | 0c8685c8b56b |
| children | a336cb13c6a9 |
| rev | line source |
|---|---|
| 119 | 1 using System.Threading; |
| 2 using System.Collections.Generic; | |
| 3 using System; | |
| 4 using System.Collections; | |
| 5 | |
| 6 namespace Implab.Parallels { | |
| 7 public class AsyncQueue<T> : IEnumerable<T> { | |
| 8 class Chunk { | |
| 9 public Chunk next; | |
| 10 | |
| 11 int m_low; | |
| 12 int m_hi; | |
| 13 int m_alloc; | |
| 14 readonly int m_size; | |
| 15 readonly T[] m_data; | |
| 16 | |
| 17 public Chunk(int size) { | |
| 18 m_size = size; | |
| 19 m_data = new T[size]; | |
| 20 } | |
| 21 | |
| 22 public Chunk(int size, T value) { | |
| 23 m_size = size; | |
| 24 m_hi = 1; | |
| 25 m_alloc = 1; | |
| 26 m_data = new T[size]; | |
| 27 m_data[0] = value; | |
| 28 } | |
| 29 | |
| 121 | 30 public Chunk(int size, T[] data, int offset, int length, int alloc) { |
| 31 m_size = size; | |
| 32 m_hi = length; | |
| 33 m_alloc = alloc; | |
| 34 m_data = new T[size]; | |
| 35 Array.Copy(data, offset, m_data, 0, length); | |
| 36 } | |
| 37 | |
| 119 | 38 public int Low { |
| 39 get { return m_low; } | |
| 40 } | |
| 41 | |
| 42 public int Hi { | |
| 43 get { return m_hi; } | |
| 44 } | |
| 45 | |
| 121 | 46 public bool TryEnqueue(T value, out bool extend) { |
| 120 | 47 var alloc = Interlocked.Increment(ref m_alloc) - 1; |
| 119 | 48 |
| 120 | 49 if (alloc >= m_size) { |
| 50 extend = alloc == m_size; | |
| 119 | 51 return false; |
| 52 } | |
| 120 | 53 |
| 54 extend = false; | |
| 119 | 55 m_data[alloc] = value; |
| 56 | |
| 57 while (alloc != Interlocked.CompareExchange(ref m_hi, alloc + 1, alloc)) { | |
| 58 // spin wait for commit | |
| 59 } | |
| 60 return true; | |
| 61 } | |
| 62 | |
| 121 | 63 public bool TryDequeue(out T value, out bool recycle) { |
| 119 | 64 int low; |
| 65 do { | |
| 66 low = m_low; | |
| 67 if (low >= m_hi) { | |
| 68 value = default(T); | |
| 69 recycle = (low == m_size); | |
| 70 return false; | |
| 71 } | |
| 72 } while(low != Interlocked.CompareExchange(ref m_low, low + 1, low)); | |
| 73 | |
| 74 recycle = (low == m_size - 1); | |
| 75 value = m_data[low]; | |
| 76 | |
| 77 return true; | |
| 78 } | |
| 79 | |
| 120 | 80 public bool TryEnqueueBatch(T[] batch, int offset, int length, out int enqueued, out bool extend) { |
|
122
0c8685c8b56b
minor fixes and improvements of AsyncQueue, additional tests
cin
parents:
121
diff
changeset
|
81 //int alloc; |
|
0c8685c8b56b
minor fixes and improvements of AsyncQueue, additional tests
cin
parents:
121
diff
changeset
|
82 //int allocSize; |
| 120 | 83 |
|
122
0c8685c8b56b
minor fixes and improvements of AsyncQueue, additional tests
cin
parents:
121
diff
changeset
|
84 var alloc = Interlocked.Add(ref m_alloc, length) - length; |
|
0c8685c8b56b
minor fixes and improvements of AsyncQueue, additional tests
cin
parents:
121
diff
changeset
|
85 if (alloc > m_size) { |
|
0c8685c8b56b
minor fixes and improvements of AsyncQueue, additional tests
cin
parents:
121
diff
changeset
|
86 // the chunk is full and someone already |
|
0c8685c8b56b
minor fixes and improvements of AsyncQueue, additional tests
cin
parents:
121
diff
changeset
|
87 // creating the new one |
|
0c8685c8b56b
minor fixes and improvements of AsyncQueue, additional tests
cin
parents:
121
diff
changeset
|
88 enqueued = 0; // nothing was added |
|
0c8685c8b56b
minor fixes and improvements of AsyncQueue, additional tests
cin
parents:
121
diff
changeset
|
89 extend = false; // the caller shouldn't try to extend the queue |
|
0c8685c8b56b
minor fixes and improvements of AsyncQueue, additional tests
cin
parents:
121
diff
changeset
|
90 return false; // nothing was added |
|
0c8685c8b56b
minor fixes and improvements of AsyncQueue, additional tests
cin
parents:
121
diff
changeset
|
91 } |
| 120 | 92 |
|
122
0c8685c8b56b
minor fixes and improvements of AsyncQueue, additional tests
cin
parents:
121
diff
changeset
|
93 enqueued = Math.Min(m_size - alloc, length); |
|
0c8685c8b56b
minor fixes and improvements of AsyncQueue, additional tests
cin
parents:
121
diff
changeset
|
94 extend = length > enqueued; |
| 120 | 95 |
|
122
0c8685c8b56b
minor fixes and improvements of AsyncQueue, additional tests
cin
parents:
121
diff
changeset
|
96 if (enqueued == 0) |
| 120 | 97 return false; |
| 98 | |
| 99 | |
|
122
0c8685c8b56b
minor fixes and improvements of AsyncQueue, additional tests
cin
parents:
121
diff
changeset
|
100 Array.Copy(batch, offset, m_data, alloc, enqueued); |
|
0c8685c8b56b
minor fixes and improvements of AsyncQueue, additional tests
cin
parents:
121
diff
changeset
|
101 |
|
0c8685c8b56b
minor fixes and improvements of AsyncQueue, additional tests
cin
parents:
121
diff
changeset
|
102 while (alloc != Interlocked.CompareExchange(ref m_hi, alloc + enqueued, alloc)) { |
| 120 | 103 // spin wait for commit |
| 104 } | |
|
122
0c8685c8b56b
minor fixes and improvements of AsyncQueue, additional tests
cin
parents:
121
diff
changeset
|
105 |
| 120 | 106 return true; |
| 107 } | |
| 108 | |
| 121 | 109 public bool TryDequeueBatch(T[] buffer, int offset, int length,out int dequeued, out bool recycle) { |
| 110 int low, hi, batchSize; | |
| 111 | |
| 112 do { | |
| 113 low = m_low; | |
| 114 hi = m_hi; | |
| 115 if (low >= hi) { | |
| 116 dequeued = 0; | |
| 117 recycle = (low == m_size); // recycling could be restarted and we need to signal again | |
| 118 return false; | |
| 119 } | |
| 120 batchSize = Math.Min(hi - low, length); | |
| 121 } while(low != Interlocked.CompareExchange(ref m_low, low + batchSize, low)); | |
| 122 | |
| 123 recycle = (low == m_size - batchSize); | |
| 124 dequeued = batchSize; | |
| 125 | |
| 126 Array.Copy(m_data, low, buffer, offset, batchSize); | |
| 127 | |
| 128 return true; | |
| 129 } | |
| 130 | |
| 119 | 131 public T GetAt(int pos) { |
| 132 return m_data[pos]; | |
| 133 } | |
| 134 } | |
| 135 | |
| 136 public const int DEFAULT_CHUNK_SIZE = 32; | |
| 121 | 137 public const int MAX_CHUNK_SIZE = 262144; |
| 119 | 138 |
| 139 readonly int m_chunkSize = DEFAULT_CHUNK_SIZE; | |
| 140 | |
| 141 Chunk m_first; | |
| 142 Chunk m_last; | |
| 143 | |
| 144 public AsyncQueue() { | |
| 145 m_last = m_first = new Chunk(m_chunkSize); | |
| 146 } | |
| 147 | |
| 121 | 148 /// <summary> |
| 149 /// Adds the specified value to the queue. | |
| 150 /// </summary> | |
| 151 /// <param name="value">Tha value which will be added to the queue.</param> | |
| 119 | 152 public void Enqueue(T value) { |
| 153 var last = m_last; | |
| 154 // spin wait to the new chunk | |
| 155 bool extend = true; | |
| 121 | 156 while (last == null || !last.TryEnqueue(value, out extend)) { |
| 119 | 157 // try to extend queue |
| 158 if (extend || last == null) { | |
| 159 var chunk = new Chunk(m_chunkSize, value); | |
| 160 if (EnqueueChunk(last, chunk)) | |
|
122
0c8685c8b56b
minor fixes and improvements of AsyncQueue, additional tests
cin
parents:
121
diff
changeset
|
161 break; // success! exit! |
| 119 | 162 last = m_last; |
| 163 } else { | |
| 121 | 164 while (last == m_last) { |
| 119 | 165 Thread.MemoryBarrier(); |
| 166 } | |
| 121 | 167 last = m_last; |
| 119 | 168 } |
| 169 } | |
| 170 } | |
| 171 | |
| 121 | 172 /// <summary> |
| 173 /// Adds the specified data to the queue. | |
| 174 /// </summary> | |
| 175 /// <param name="data">The buffer which contains the data to be enqueued.</param> | |
| 176 /// <param name="offset">The offset of the data in the buffer.</param> | |
| 177 /// <param name="length">The size of the data to read from the buffer.</param> | |
| 178 public void EnqueueRange(T[] data, int offset, int length) { | |
| 179 if (data == null) | |
| 180 throw new ArgumentNullException("data"); | |
| 181 if (offset < 0) | |
| 182 throw new ArgumentOutOfRangeException("offset"); | |
| 183 if (length < 1 || offset + length > data.Length) | |
| 184 throw new ArgumentOutOfRangeException("length"); | |
| 185 | |
| 186 var last = m_last; | |
| 187 | |
| 188 bool extend; | |
| 189 int enqueued; | |
| 190 | |
| 191 while (length > 0) { | |
| 192 extend = true; | |
| 193 if (last != null && last.TryEnqueueBatch(data, offset, length, out enqueued, out extend)) { | |
| 194 length -= enqueued; | |
| 195 offset += enqueued; | |
| 196 } | |
| 197 | |
| 198 if (extend) { | |
| 199 // there was no enough space in the chunk | |
| 200 // or there was no chunks in the queue | |
| 201 | |
| 202 while (length > 0) { | |
| 203 | |
| 204 var size = Math.Min(length, MAX_CHUNK_SIZE); | |
| 205 | |
| 206 var chunk = new Chunk( | |
| 207 Math.Max(size, m_chunkSize), | |
| 208 data, | |
| 209 offset, | |
| 210 size, | |
| 211 length // length >= size | |
| 212 ); | |
| 213 | |
| 214 if (!EnqueueChunk(last, chunk)) { | |
| 215 // looks like the queue has been updated then proceed from the beginning | |
| 216 last = m_last; | |
| 217 break; | |
| 218 } | |
| 219 | |
| 220 // we have successfully added the new chunk | |
| 221 last = chunk; | |
| 222 length -= size; | |
| 223 offset += size; | |
| 224 } | |
| 225 } else { | |
| 226 // we don't need to extend the queue, if we successfully enqueued data | |
| 227 if (length == 0) | |
| 228 break; | |
| 229 | |
| 230 // if we need to wait while someone is extending the queue | |
| 231 // spinwait | |
| 232 while (last == m_last) { | |
| 233 Thread.MemoryBarrier(); | |
| 234 } | |
| 235 | |
| 236 last = m_last; | |
| 237 } | |
| 238 } | |
| 239 } | |
| 240 | |
| 241 /// <summary> | |
| 242 /// Tries to retrieve the first element from the queue. | |
| 243 /// </summary> | |
| 244 /// <returns><c>true</c>, if element is dequeued, <c>false</c> otherwise.</returns> | |
| 245 /// <param name="value">The value of the dequeued element.</param> | |
| 119 | 246 public bool TryDequeue(out T value) { |
| 247 var chunk = m_first; | |
| 248 bool recycle; | |
| 249 while (chunk != null) { | |
| 250 | |
| 251 var result = chunk.TryDequeue(out value, out recycle); | |
| 252 | |
| 253 if (recycle) // this chunk is waste | |
| 254 RecycleFirstChunk(chunk); | |
| 255 else | |
| 256 return result; // this chunk is usable and returned actual result | |
| 257 | |
| 258 if (result) // this chunk is waste but the true result is always actual | |
| 259 return true; | |
| 260 | |
| 261 // try again | |
| 262 chunk = m_first; | |
| 263 } | |
| 264 | |
| 265 // the queue is empty | |
| 266 value = default(T); | |
| 267 return false; | |
| 268 } | |
| 269 | |
| 121 | 270 /// <summary> |
| 271 /// Tries to dequeue the specified amount of data from the queue. | |
| 272 /// </summary> | |
| 273 /// <returns><c>true</c>, if data was deuqueued, <c>false</c> otherwise.</returns> | |
| 274 /// <param name="buffer">The buffer to which the data will be written.</param> | |
| 275 /// <param name="offset">The offset in the buffer at which the data will be written.</param> | |
| 276 /// <param name="length">The maximum amount of data to be retrieved.</param> | |
| 277 /// <param name="dequeued">The actual amout of the retrieved data.</param> | |
| 278 public bool TryDequeueRange(T[] buffer, int offset, int length, out int dequeued) { | |
| 279 if (buffer == null) | |
| 280 throw new ArgumentNullException("buffer"); | |
| 281 if (offset < 0) | |
| 282 throw new ArgumentOutOfRangeException("offset"); | |
| 283 if (length < 1 || offset + length > buffer.Length) | |
| 284 throw new ArgumentOutOfRangeException("length"); | |
| 285 | |
| 286 var chunk = m_first; | |
| 287 bool recycle; | |
| 288 dequeued = 0; | |
| 289 while (chunk != null) { | |
| 290 | |
| 291 int actual; | |
| 292 if (chunk.TryDequeueBatch(buffer, offset, length, out actual, out recycle)) { | |
| 293 offset += actual; | |
| 294 length -= actual; | |
| 295 dequeued += actual; | |
| 296 } | |
| 297 | |
| 298 if (recycle) // this chunk is waste | |
| 299 RecycleFirstChunk(chunk); | |
| 300 else if (actual == 0) | |
| 301 break; // the chunk is usable, but doesn't contain any data (it's the last chunk in the queue) | |
| 302 | |
| 303 if (length == 0) | |
| 304 return true; | |
| 305 | |
| 306 // we still may dequeue something | |
| 307 // try again | |
| 308 chunk = m_first; | |
| 309 } | |
| 310 | |
| 311 return dequeued != 0; | |
| 312 } | |
| 313 | |
| 314 /// <summary> | |
| 315 /// Tries to dequeue all remaining data in the first chunk. | |
| 316 /// </summary> | |
| 317 /// <returns><c>true</c>, if data was dequeued, <c>false</c> otherwise.</returns> | |
|
122
0c8685c8b56b
minor fixes and improvements of AsyncQueue, additional tests
cin
parents:
121
diff
changeset
|
318 /// <param name="buffer">The buffer to which the data will be written.</param> |
| 121 | 319 /// <param name="offset">The offset in the buffer at which the data will be written.</param> |
| 320 /// <param name="length">Tha maximum amount of the data to be dequeued.</param> | |
| 321 /// <param name="dequeued">The actual amount of the dequeued data.</param> | |
| 322 public bool TryDequeueChunk(T[] buffer, int offset, int length, out int dequeued) { | |
| 323 if (buffer == null) | |
| 324 throw new ArgumentNullException("buffer"); | |
| 325 if (offset < 0) | |
| 326 throw new ArgumentOutOfRangeException("offset"); | |
| 327 if (length < 1 || offset + length > buffer.Length) | |
| 328 throw new ArgumentOutOfRangeException("length"); | |
| 329 | |
| 330 var chunk = m_first; | |
| 331 bool recycle; | |
| 332 dequeued = 0; | |
| 333 | |
| 334 while (chunk != null) { | |
| 335 | |
| 336 int actual; | |
| 337 if (chunk.TryDequeueBatch(buffer, offset, length, out actual, out recycle)) { | |
| 338 dequeued = actual; | |
| 339 } | |
| 340 | |
| 341 if (recycle) // this chunk is waste | |
| 342 RecycleFirstChunk(chunk); | |
| 343 | |
| 344 // if we have dequeued any data, then return | |
| 345 if (dequeued != 0) | |
| 346 return true; | |
| 347 | |
| 348 // we still may dequeue something | |
| 349 // try again | |
| 350 chunk = m_first; | |
| 351 } | |
| 352 | |
| 353 return false; | |
| 354 } | |
| 355 | |
| 119 | 356 bool EnqueueChunk(Chunk last, Chunk chunk) { |
| 357 if (Interlocked.CompareExchange(ref m_last, chunk, last) != last) | |
| 358 return false; | |
| 359 | |
| 360 if (last != null) | |
| 361 last.next = chunk; | |
| 362 else | |
| 363 m_first = chunk; | |
| 364 return true; | |
| 365 } | |
| 366 | |
| 367 void RecycleFirstChunk(Chunk first) { | |
| 368 var next = first.next; | |
| 369 | |
| 370 if (next == null) { | |
| 371 // looks like this is the last chunk | |
| 372 if (first != Interlocked.CompareExchange(ref m_last, null, first)) { | |
| 373 // race | |
| 374 // maybe someone already recycled this chunk | |
| 375 // or a new chunk has been appedned to the queue | |
| 376 | |
| 377 return; // give up | |
| 378 } | |
| 379 // the tail is updated | |
| 380 } | |
| 381 | |
| 382 // we need to update the head | |
| 383 Interlocked.CompareExchange(ref m_first, next, first); | |
| 384 // if the head is already updated then give up | |
| 385 return; | |
| 386 | |
| 387 } | |
| 388 | |
| 123 | 389 public void Clear() { |
| 390 // start the new queue | |
| 391 var t = new Chunk(m_chunkSize); | |
| 392 Thread.MemoryBarrier(); | |
| 393 m_last = t; | |
| 394 Thread.MemoryBarrier(); | |
| 395 | |
| 396 // make the new queue available to the readers, and stop the old one | |
| 397 m_first = t; | |
| 398 Thread.MemoryBarrier(); | |
| 399 } | |
| 400 | |
| 401 public T[] Drain() { | |
| 402 // start the new queue | |
| 403 var t = new Chunk(m_chunkSize); | |
| 404 Thread.MemoryBarrier(); | |
| 405 m_last = t; | |
| 406 Thread.MemoryBarrier(); | |
| 407 | |
| 408 // make the new queue available to the readers, and stop the old one | |
| 409 Chunk first; | |
| 410 | |
| 411 do { | |
| 412 first = m_first; | |
| 413 } while(first != Interlocked.CompareExchange(ref m_first | |
| 414 Thread.MemoryBarrier(); | |
| 415 | |
| 416 | |
| 417 } | |
| 418 | |
| 419 T[] ReadChunks(Chunk chunk) { | |
| 420 var result = new List<T>(); | |
| 421 var buffer = new T[m_chunkSize]; | |
| 422 int actual; | |
| 423 bool recycle; | |
| 424 while (chunk != null) { | |
| 425 // we need to read the chunk using this way | |
| 426 // since some client still may completing the dequeue | |
| 427 // operation, such clients most likely won't get results | |
| 428 while (chunk.TryDequeueBatch(buffer, 0, buffer.Length, out actual, out recycle)) | |
| 429 result.AddRange(new ArraySegmentCollection(buffer, 0, actual)); | |
| 430 | |
| 431 chunk = chunk.next; | |
| 432 } | |
| 433 | |
| 434 return result.ToArray(); | |
| 435 } | |
| 436 | |
| 437 struct ArraySegmentCollection : ICollection<T> { | |
| 438 readonly T[] m_data; | |
| 439 readonly int m_offset; | |
| 440 readonly int m_length; | |
| 441 | |
| 442 public ArraySegmentCollection(T[] data, int offset, int length) { | |
| 443 m_data = data; | |
| 444 m_offset = offset; | |
| 445 m_length = length; | |
| 446 } | |
| 447 | |
| 448 #region ICollection implementation | |
| 449 | |
| 450 public void Add(T item) { | |
| 451 throw new InvalidOperationException(); | |
| 452 } | |
| 453 | |
| 454 public void Clear() { | |
| 455 throw new InvalidOperationException(); | |
| 456 } | |
| 457 | |
| 458 public bool Contains(T item) { | |
| 459 return false; | |
| 460 } | |
| 461 | |
| 462 public void CopyTo(T[] array, int arrayIndex) { | |
| 463 Array.Copy(m_data,m_offset,array,arrayIndex, m_length); | |
| 464 } | |
| 465 | |
| 466 public bool Remove(T item) { | |
| 467 throw new NotImplementedException(); | |
| 468 } | |
| 469 | |
| 470 public int Count { | |
| 471 get { | |
| 472 return m_length; | |
| 473 } | |
| 474 } | |
| 475 | |
| 476 public bool IsReadOnly { | |
| 477 get { | |
| 478 return true; | |
| 479 } | |
| 480 } | |
| 481 | |
| 482 #endregion | |
| 483 | |
| 484 #region IEnumerable implementation | |
| 485 | |
| 486 public IEnumerator<T> GetEnumerator() { | |
| 487 for (int i = m_offset; i < m_length + m_offset; i++) | |
| 488 yield return m_data[i]; | |
| 489 } | |
| 490 | |
| 491 #endregion | |
| 492 | |
| 493 #region IEnumerable implementation | |
| 494 | |
| 495 IEnumerator IEnumerable.GetEnumerator() { | |
| 496 return GetEnumerator(); | |
| 497 } | |
| 498 | |
| 499 #endregion | |
| 500 } | |
| 501 | |
| 119 | 502 #region IEnumerable implementation |
| 503 | |
| 504 class Enumerator : IEnumerator<T> { | |
| 505 Chunk m_current; | |
| 506 int m_pos = -1; | |
| 507 | |
| 508 public Enumerator(Chunk fisrt) { | |
| 509 m_current = fisrt; | |
| 510 } | |
| 511 | |
| 512 #region IEnumerator implementation | |
| 513 | |
| 514 public bool MoveNext() { | |
| 515 if (m_current == null) | |
| 516 return false; | |
| 517 | |
| 518 if (m_pos == -1) | |
| 519 m_pos = m_current.Low; | |
| 520 else | |
| 521 m_pos++; | |
| 522 if (m_pos == m_current.Hi) { | |
| 523 m_pos = 0; | |
| 524 m_current = m_current.next; | |
| 525 } | |
| 526 | |
| 527 return true; | |
| 528 } | |
| 529 | |
| 530 public void Reset() { | |
| 531 throw new NotSupportedException(); | |
| 532 } | |
| 533 | |
| 534 object IEnumerator.Current { | |
| 535 get { | |
| 536 return Current; | |
| 537 } | |
| 538 } | |
| 539 | |
| 540 #endregion | |
| 541 | |
| 542 #region IDisposable implementation | |
| 543 | |
| 544 public void Dispose() { | |
| 545 } | |
| 546 | |
| 547 #endregion | |
| 548 | |
| 549 #region IEnumerator implementation | |
| 550 | |
| 551 public T Current { | |
| 552 get { | |
| 553 if (m_pos == -1 || m_current == null) | |
| 554 throw new InvalidOperationException(); | |
| 555 return m_current.GetAt(m_pos); | |
| 556 } | |
| 557 } | |
| 558 | |
| 559 #endregion | |
| 560 } | |
| 561 | |
| 562 public IEnumerator<T> GetEnumerator() { | |
| 563 return new Enumerator(m_first); | |
| 564 } | |
| 565 | |
| 566 #endregion | |
| 567 | |
| 568 #region IEnumerable implementation | |
| 569 | |
| 570 IEnumerator IEnumerable.GetEnumerator() { | |
| 571 return GetEnumerator(); | |
| 572 } | |
| 573 | |
| 574 #endregion | |
| 575 } | |
| 576 } |