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