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