@@ -1,87 +1,101 | |||
|
1 | 1 | using System; |
|
2 | 2 | using System.Threading; |
|
3 | 3 | |
|
4 | 4 | namespace Implab.Parallels { |
|
5 | 5 | public class BlockingQueue<T> : AsyncQueue<T> { |
|
6 | 6 | readonly object m_lock = new object(); |
|
7 | 7 | |
|
8 | 8 | public override void Enqueue(T value) { |
|
9 | 9 | base.Enqueue(value); |
|
10 | 10 | lock (m_lock) |
|
11 | 11 | Monitor.Pulse(m_lock); |
|
12 | 12 | } |
|
13 | 13 | |
|
14 | 14 | public override void EnqueueRange(T[] data, int offset, int length) { |
|
15 | 15 | base.EnqueueRange(data, offset, length); |
|
16 | 16 | if (length > 1) |
|
17 | 17 | lock (m_lock) |
|
18 | 18 | Monitor.PulseAll(m_lock); |
|
19 | 19 | else |
|
20 | 20 | lock (m_lock) |
|
21 | 21 | Monitor.Pulse(m_lock); |
|
22 | 22 | } |
|
23 | 23 | |
|
24 | 24 | public T GetItem(int timeout) { |
|
25 | 25 | T item; |
|
26 | var t1 = Environment.TickCount; | |
|
27 | var dt = timeout; | |
|
28 | while (!TryDequeue(out item)) { | |
|
29 | lock (m_lock) | |
|
30 | if (!Monitor.Wait(m_lock, dt)) | |
|
31 | throw new TimeoutException(); | |
|
32 |
|
|
|
33 | dt = timeout - Environment.TickCount + t1; | |
|
34 | if (dt < 0) | |
|
35 |
|
|
|
26 | ||
|
27 | if (!TryDequeue(out item)) { | |
|
28 | var t1 = Environment.TickCount; | |
|
29 | var dt = timeout; | |
|
30 | ||
|
31 | lock (m_lock) { | |
|
32 | while (!TryDequeue(out item)) { | |
|
33 | if (!Monitor.Wait(m_lock, dt)) | |
|
34 | throw new TimeoutException(); | |
|
35 | if (timeout >= 0) { | |
|
36 | dt = timeout - Environment.TickCount + t1; | |
|
37 | if (dt < 0) | |
|
38 | throw new TimeoutException(); | |
|
39 | } | |
|
40 | } | |
|
36 | 41 | } |
|
37 | 42 | } |
|
38 | 43 | return item; |
|
39 | 44 | } |
|
40 | 45 | |
|
41 | 46 | public T GetItem() { |
|
42 | 47 | T item; |
|
43 |
|
|
|
44 | lock (m_lock) | |
|
45 | Monitor.Wait(m_lock); | |
|
48 | if (!TryDequeue(out item)) | |
|
49 | lock (m_lock) { | |
|
50 | while (!TryDequeue(out item)) | |
|
51 | Monitor.Wait(m_lock); | |
|
52 | } | |
|
46 | 53 | return item; |
|
47 | 54 | } |
|
48 | 55 | |
|
49 | 56 | public T[] GetRange(int max, int timeout) { |
|
50 | 57 | Safe.ArgumentInRange(max, 1, Int32.MaxValue, "max"); |
|
51 | 58 | |
|
52 | 59 | var buffer = new T[max]; |
|
53 | 60 | int actual; |
|
54 | var t1 = Environment.TickCount; | |
|
55 |
var |
|
|
56 | while (!TryDequeueRange(buffer,0,max,out actual)) { | |
|
57 | lock (m_lock) | |
|
58 |
|
|
|
59 | throw new TimeoutException(); | |
|
60 |
|
|
|
61 | dt = timeout - Environment.TickCount + t1; | |
|
62 | if (dt < 0) | |
|
63 | throw new TimeoutException(); | |
|
61 | if (!TryDequeueRange(buffer, 0, max, out actual)) { | |
|
62 | var t1 = Environment.TickCount; | |
|
63 | var dt = timeout; | |
|
64 | ||
|
65 | lock (m_lock) { | |
|
66 | while (!TryDequeueRange(buffer, 0, max, out actual)) { | |
|
67 | ||
|
68 | if (!Monitor.Wait(m_lock, dt)) | |
|
69 | throw new TimeoutException(); | |
|
70 | ||
|
71 | if (timeout >= 0) { | |
|
72 | dt = timeout - Environment.TickCount + t1; | |
|
73 | if (dt < 0) | |
|
74 | throw new TimeoutException(); | |
|
75 | } | |
|
76 | } | |
|
64 | 77 | } |
|
65 | 78 | } |
|
66 | 79 | |
|
67 | 80 | var data = new T[actual]; |
|
68 | 81 | Array.Copy(buffer, data, actual); |
|
69 | 82 | return data; |
|
70 | 83 | } |
|
71 | 84 | |
|
72 | 85 | public T[] GetRange(int max) { |
|
73 | 86 | Safe.ArgumentInRange(max, 1, Int32.MaxValue, "max"); |
|
74 | 87 | |
|
75 | 88 | var buffer = new T[max]; |
|
76 | 89 | int actual; |
|
77 |
|
|
|
90 | if (!TryDequeueRange(buffer, 0, max, out actual)) | |
|
78 | 91 | lock (m_lock) |
|
79 | Monitor.Wait(m_lock); | |
|
92 | while (!TryDequeueRange(buffer, 0, max, out actual)) | |
|
93 | Monitor.Wait(m_lock); | |
|
80 | 94 | |
|
81 | 95 | var data = new T[actual]; |
|
82 | 96 | Array.Copy(buffer, data, actual); |
|
83 | 97 | return data; |
|
84 | 98 | } |
|
85 | 99 | } |
|
86 | 100 | } |
|
87 | 101 |
General Comments 0
You need to be logged in to leave comments.
Login now