##// END OF EJS Templates
implab » ImplabNet
RSS
fixed promise chaining behavior, the error handler doesn't handle result or cancellation handlers exceptions these exceptions are propagated to the next handlers.
fixed promise chaining behavior, the error handler doesn't handle result or cancellation handlers exceptions these exceptions are propagated to the next handlers.
cin - - Load All Authors

File last commit:

r139:041b77711262 v2
r196:40d7fed4a09e default
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;
}
}
}