Mercurial > pub > ImplabNet

comparison Implab/Parallels/BlockingQueue.cs @ 139:041b77711262 v2

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
fixed blocking queue
author cin
date Wed, 18 Feb 2015 18:02:02 +0300
parents 238e15580926
children d6fe09f5592c
comparison
equal deleted inserted replaced
138:f75cfa58e3d4 139:041b77711262
21 Monitor.Pulse(m_lock); 21 Monitor.Pulse(m_lock);
22 } 22 }
23 23
24 public T GetItem(int timeout) { 24 public T GetItem(int timeout) {
25 T item; 25 T item;
26 var t1 = Environment.TickCount; 26
27 var dt = timeout; 27 if (!TryDequeue(out item)) {
28 while (!TryDequeue(out item)) { 28 var t1 = Environment.TickCount;
29 lock (m_lock) 29 var dt = timeout;
30 if (!Monitor.Wait(m_lock, dt)) 30
31 throw new TimeoutException(); 31 lock (m_lock) {
32 if (timeout >= 0) { 32 while (!TryDequeue(out item)) {
33 dt = timeout - Environment.TickCount + t1; 33 if (!Monitor.Wait(m_lock, dt))
34 if (dt < 0) 34 throw new TimeoutException();
35 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 return item; 43 return item;
39 } 44 }
40 45
41 public T GetItem() { 46 public T GetItem() {
42 T item; 47 T item;
43 while (!TryDequeue(out item)) 48 if (!TryDequeue(out item))
44 lock (m_lock) 49 lock (m_lock) {
45 Monitor.Wait(m_lock); 50 while (!TryDequeue(out item))
51 Monitor.Wait(m_lock);
52 }
46 return item; 53 return item;
47 } 54 }
48 55
49 public T[] GetRange(int max, int timeout) { 56 public T[] GetRange(int max, int timeout) {
50 Safe.ArgumentInRange(max, 1, Int32.MaxValue, "max"); 57 Safe.ArgumentInRange(max, 1, Int32.MaxValue, "max");
51 58
52 var buffer = new T[max]; 59 var buffer = new T[max];
53 int actual; 60 int actual;
54 var t1 = Environment.TickCount; 61 if (!TryDequeueRange(buffer, 0, max, out actual)) {
55 var dt = timeout; 62 var t1 = Environment.TickCount;
56 while (!TryDequeueRange(buffer,0,max,out actual)) { 63 var dt = timeout;
57 lock (m_lock) 64
58 if (!Monitor.Wait(m_lock, dt)) 65 lock (m_lock) {
59 throw new TimeoutException(); 66 while (!TryDequeueRange(buffer, 0, max, out actual)) {
60 if (timeout >= 0) { 67
61 dt = timeout - Environment.TickCount + t1; 68 if (!Monitor.Wait(m_lock, dt))
62 if (dt < 0) 69 throw new TimeoutException();
63 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 var data = new T[actual]; 80 var data = new T[actual];
68 Array.Copy(buffer, data, actual); 81 Array.Copy(buffer, data, actual);
72 public T[] GetRange(int max) { 85 public T[] GetRange(int max) {
73 Safe.ArgumentInRange(max, 1, Int32.MaxValue, "max"); 86 Safe.ArgumentInRange(max, 1, Int32.MaxValue, "max");
74 87
75 var buffer = new T[max]; 88 var buffer = new T[max];
76 int actual; 89 int actual;
77 while (!TryDequeueRange(buffer, 0, max, out actual)) 90 if (!TryDequeueRange(buffer, 0, max, out actual))
78 lock (m_lock) 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 var data = new T[actual]; 95 var data = new T[actual];
82 Array.Copy(buffer, data, actual); 96 Array.Copy(buffer, data, actual);
83 return data; 97 return data;
84 } 98 }