##// END OF EJS Templates
implab » ImplabNet
RSS
Reworked cancelation handling, if the cancel handler isn't specified the OperationCanceledException will be handled by the error handler...
Reworked cancelation handling, if the cancel handler isn't specified the OperationCanceledException will be handled by the error handler Any unhandled OperationCanceledException will cause the promise cancelation
cin - - Load All Authors

File last commit:

r139:041b77711262 v2
r187:dd4a3590f9c6 ref20160224
Show More
BlockingQueue.cs
101 lines | 3.1 KiB | text/x-csharp | CSharpLexer
/ Implab / Parallels / BlockingQueue.cs
History | Source | Raw |Copy content |Copy permalink
cin
added the blocking queue
 r137 using System;
using System.Threading;
namespace Implab.Parallels {
public class BlockingQueue<T> : AsyncQueue<T> {
readonly object m_lock = new object();
public override void Enqueue(T value) {
base.Enqueue(value);
lock (m_lock)
Monitor.Pulse(m_lock);
}
public override void EnqueueRange(T[] data, int offset, int length) {
base.EnqueueRange(data, offset, length);
if (length > 1)
lock (m_lock)
Monitor.PulseAll(m_lock);
else
lock (m_lock)
Monitor.Pulse(m_lock);
}
public T GetItem(int timeout) {
T item;
cin
fixed blocking queue
 r139
if (!TryDequeue(out item)) {
var t1 = Environment.TickCount;
var dt = timeout;
lock (m_lock) {
while (!TryDequeue(out item)) {
if (!Monitor.Wait(m_lock, dt))
throw new TimeoutException();
if (timeout >= 0) {
dt = timeout - Environment.TickCount + t1;
if (dt < 0)
throw new TimeoutException();
}
}
cin
added the blocking queue
 r137 }
}
return item;
}
public T GetItem() {
T item;
cin
fixed blocking queue
 r139 if (!TryDequeue(out item))
lock (m_lock) {
while (!TryDequeue(out item))
Monitor.Wait(m_lock);
}
cin
added the blocking queue
 r137 return item;
}
public T[] GetRange(int max, int timeout) {
Safe.ArgumentInRange(max, 1, Int32.MaxValue, "max");
var buffer = new T[max];
int actual;
cin
fixed blocking queue
 r139 if (!TryDequeueRange(buffer, 0, max, out actual)) {
var t1 = Environment.TickCount;
var dt = timeout;
lock (m_lock) {
while (!TryDequeueRange(buffer, 0, max, out actual)) {
if (!Monitor.Wait(m_lock, dt))
throw new TimeoutException();
if (timeout >= 0) {
dt = timeout - Environment.TickCount + t1;
if (dt < 0)
throw new TimeoutException();
}
}
cin
added the blocking queue
 r137 }
}
var data = new T[actual];
Array.Copy(buffer, data, actual);
return data;
}
public T[] GetRange(int max) {
Safe.ArgumentInRange(max, 1, Int32.MaxValue, "max");
var buffer = new T[max];
int actual;
cin
fixed blocking queue
 r139 if (!TryDequeueRange(buffer, 0, max, out actual))
cin
added the blocking queue
 r137 lock (m_lock)
cin
fixed blocking queue
 r139 while (!TryDequeueRange(buffer, 0, max, out actual))
Monitor.Wait(m_lock);
cin
added the blocking queue
 r137
var data = new T[actual];
Array.Copy(buffer, data, actual);
return data;
}
}
}