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