William
Tue Aug 29 10:18:37 CDT 2006
Here is a non-generic version. hth
using System;
using System.Collections;
using System.Threading;
namespace WJS.Threading
{
/// <summary>
/// A blocking queue derived from Queue. The dequeue method will block
on an empty queue.
/// Enqueue operations will not block as queue is only bounded by array
size or MaxSize parameter.
/// <remarks>This class is thread safe for multiple consumer and
producer threads.</remarks>
/// </summary>
public class BlockingQueue : Queue
{
private bool opened = true;
private int maxSize = int.MaxValue;
private readonly object syncRoot = new object();
/// <summary>
/// Create new BlockingQueue.
/// </summary>
/// <param name="col">The System.Collections.ICollection to copy
elements from.</param>
public BlockingQueue(ICollection col) : base(col)
{
}
/// <summary>
/// Create new BlockingQueue.
/// </summary>
/// <param name="capacity">The initial number of elements that the
queue can contain.</param>
/// <param name="growFactor">The factor by which the capacity of the
queue is expanded.</param>
public BlockingQueue(int capacity, float growFactor) :
base(capacity, growFactor)
{
}
/// <summary>
/// Create new BlockingQueue.
/// </summary>
/// <param name="capacity">The initial number of elements that the
queue can contain.</param>
public BlockingQueue(int capacity) : base(capacity)
{
}
/// <summary>
/// Create new BlockingQueue.
/// </summary>
public BlockingQueue() : base()
{
}
/// <summary>
/// BlockingQueue Destructor (Close queue, resume any waiting
thread).
/// </summary>
~BlockingQueue()
{
Close();
}
/// <summary>
/// Gets or sets the maximum size of the queue. After MaxSize is
reached, an additional Enqueue operation will
/// throw an exception.
/// </summary>
public int MaxSize
{
get
{
lock ( syncRoot )
{
return this.maxSize;
}
}
set
{
lock ( syncRoot )
{
if ( value < 0 )
throw new ArgumentOutOfRangeException("MaxSize must
be >= 0.");
}
}
}
/// <summary>
/// Remove all objects from the Queue.
/// </summary>
public override void Clear()
{
lock ( syncRoot )
{
base.Clear();
}
}
/// <summary>
/// Remove all objects from the Queue, resume all dequeue threads.
/// </summary>
public void Close()
{
lock ( syncRoot )
{
if ( ! Opened )
return; // Already closed.
opened = false;
base.Clear();
Monitor.PulseAll(syncRoot); // resume any waiting threads
}
}
/// <summary>
/// Gets flag indicating if queue has been closed.
/// </summary>
public bool Opened
{
get
{
lock ( syncRoot )
{
return this.opened;
}
}
}
/// <summary>
/// Removes and returns the object at the beginning of the Queue.
/// </summary>
/// <returns>Object in queue.</returns>
public override object Dequeue()
{
return Dequeue(Timeout.Infinite);
}
/// <summary>
/// Removes and returns the object at the beginning of the Queue.
/// </summary>
/// <param name="timeout">time to wait before returning</param>
/// <returns>Object in queue.</returns>
public object Dequeue(TimeSpan timeout)
{
return Dequeue(timeout.Milliseconds);
}
/// <summary>
/// Removes and returns the object at the beginning of the Queue.
/// </summary>
/// <param name="timeout">Time to wait before returning (in
milliseconds)</param>
/// <returns>Object in queue.</returns>
public object Dequeue(int timeout)
{
lock ( syncRoot )
{
while ( Opened && (base.Count == 0) )
{
if ( ! Monitor.Wait(syncRoot, timeout) )
throw new InvalidOperationException("Timeout");
}
if ( Opened )
return base.Dequeue();
else
throw new InvalidOperationException("Queue closed");
}
}
/// <summary>
/// Adds an object to the end of the Queue.
/// </summary>
/// <param name="obj">Object to put in queue</param>
public override void Enqueue(object obj)
{
lock ( syncRoot )
{
if ( Opened )
throw new InvalidOperationException("Queue closed");
if ( base.Count >= this.maxSize )
throw new InvalidOperationException("Queue full. MaxSize
reached.");
base.Enqueue(obj);
Monitor.Pulse(syncRoot); // Move 1 waiting thread to ready
queue in this monitor object.
} // Exiting lock will start all threads in the ready queue for
this monitor object.
}
}
}
--
William Stacey [MVP]
"Thana N." <ThanaN@discussions.microsoft.com> wrote in message
news:AE066D58-03E6-47BE-9C0A-68CA55BC08B5@microsoft.com...
| Thanks for your reply, but I've forgot to tell you all that I still using
| .NET 1.1 not 2.0. Any suggestion for 1.1.
|
| Rgrds,
| Thana N.
|
| "William Stacey [MVP]" wrote:
|
| > Couple issues here. First, don't use spin locks that way - big waste of
| > cpu. Moreover, the sync wrapper takes a lock for all methods and
| > properties. So the 2nd thread is taking an releasing the lock 4 times
for a
| > single dequeue operation. As Brian said, use a blocking queue for this
kind
| > of producer/consumer pattern. Here is my 1 lock blocking queue below.
| >
| > private void button5_Click(object sender, EventArgs e)
| > {
| > BlockingQueue<int> bq = new BlockingQueue<int>();
| >
| > new Thread(delegate()
| > {
| > int i;
| > while( bq.TryDequeue(200, out i) )
| > Console.WriteLine(i);
| > Console.WriteLine("Consumer thread completed.");
| > }).Start();
| >
| > for (int i = 0; i < 100; i++)
| > {
| > bq.Enqueue(i);
| > }
| >
| > }
| >
| > /// <summary>
| > /// Represents a first-in, first-out collection of objects.
| > /// </summary>
| > /// <typeparam name="T">Type of element queue will
contain.</typeparam>
| > public class BlockingQueue<T> : IEnumerable<T>, ICollection
| > {
| > private bool isOpened = true;
| > private readonly Queue<T> q;
| > private readonly object syncRoot = new object();
| >
| > /// <summary>
| > /// Initializes a new instance of the BlockingQueue class.
| > /// </summary>
| > public BlockingQueue()
| > {
| > q = new Queue<T>();
| > }
| >
| > /// <summary>
| > /// Initializes a new instance of the BlockingQueue class.
| > /// </summary>
| > /// <param name="capacity">The initial number of elements the
queue
| > can contain.</param>
| > public BlockingQueue(int capacity)
| > {
| > q = new Queue<T>(capacity);
| > }
| >
| > /// <summary>
| > /// Initializes a new instance of the BlockingQueue class.
| > /// </summary>
| > /// <param name="collection">A collection whose elements are
copied
| > to the new queue.</param>
| > public BlockingQueue(IEnumerable<T> collection)
| > {
| > q = new Queue<T>(collection);
| > }
| >
| > /// <summary>
| > /// Gets the number of elements in the queue.
| > /// </summary>
| > public int Count
| > {
| > get
| > {
| > lock ( syncRoot )
| > {
| > return q.Count;
| > }
| > }
| > }
| >
| > /// <summary>
| > /// Remove all objects from the BlockingQueue<T>.
| > /// </summary>
| > public void Clear()
| > {
| > lock ( syncRoot )
| > {
| > q.Clear();
| > }
| > }
| >
| > /// <summary>
| > /// Closes the queue.
| > /// </summary>
| > public void Close()
| > {
| > lock ( syncRoot )
| > {
| > if ( ! this.isOpened )
| > return; // Already closed.
| >
| > isOpened = false;
| > q.Clear();
| > Monitor.PulseAll(syncRoot); // resume any waiting
threads
| > so they see the queue is closed.
| > }
| > }
| >
| > /// <summary>
| > /// Gets a value indicating if queue is opened.
| > /// </summary>
| > public bool Opened
| > {
| > get
| > {
| > lock ( syncRoot )
| > {
| > return this.isOpened;
| > }
| > }
| > }
| >
| > /// <summary>
| > /// Determines whether an element is in the
| > System.Collections.Generic.Queue<T>.
| > /// </summary>
| > /// <param name="item">The object to locate in the
| > System.Collections.Generic.Queue<T>. The value can be null for reference
| > types.</param>
| > /// <returns>true if item is found in the
| > System.Collections.Generic.Queue<T>; otherwise, false.</returns>
| > public bool Contains(T item)
| > {
| > lock ( syncRoot )
| > {
| > return q.Contains(item);
| > }
| > }
| >
| > /// <summary>
| > /// Copies the System.Collections.Generic.Queue<T> elements to
an
| > existing one-dimensional System.Array, starting at the specified array
| > index.
| > /// </summary>
| > /// <param name="array">The one-dimensional System.Array that is
the
| > destination of the elements
| > /// copied from System.Collections.Generic.Queue<T>. The
| > System.Array must have zero-based indexing.
| > /// </param>
| > /// <param name="arrayIndex">The zero-based index in array at
which
| > copying begins.</param>
| > public void CopyTo(T[] array, int arrayIndex)
| > {
| > lock ( syncRoot )
| > {
| > q.CopyTo(array, arrayIndex);
| > }
| > }
| >
| > public T[] ToArray()
| > {
| > lock ( syncRoot )
| > {
| > return q.ToArray();
| > }
| > }
| >
| > public IEnumerator<T> GetEnumerator()
| > {
| > return new BlockingQueue<T>.Enumerator(this, -1);
| > }
| > public IEnumerator<T> GetEnumerator(int millisecondsTimeout)
| > {
| > return new BlockingQueue<T>.Enumerator(this,
| > millisecondsTimeout);
| > }
| > IEnumerator IEnumerable.GetEnumerator()
| > {
| > return new BlockingQueue<T>.Enumerator(this, -1);
| > }
| >
| > /// <summary>
| > /// Sets the capacity to the actual number of elements in the
| > System.Collections.Generic.Queue<T>,
| > /// if that number is less than 90 percent of current capacity.
| > /// </summary>
| > public void TrimExcess()
| > {
| > lock ( syncRoot )
| > {
| > q.TrimExcess();
| > }
| > }
| >
| > /// <summary>
| > /// Removes and returns the object at the beginning of the
Queue.
| > /// </summary>
| > /// <returns>Object in queue.</returns>
| > public T Dequeue()
| > {
| > return Dequeue(Timeout.Infinite);
| > }
| >
| > /// <summary>
| > /// Removes and returns the object at the beginning of the
Queue.
| > /// </summary>
| > /// <param name="timeout">Time to wait before returning (in
| > milliseconds).</param>
| > /// <returns>Object in queue.</returns>
| > public T Dequeue(int millisecondsTimeout)
| > {
| > lock ( syncRoot )
| > {
| > while ( isOpened && (q.Count == 0) )
| > {
| > if ( ! Monitor.Wait(syncRoot, millisecondsTimeout) )
| > throw new TimeoutException("Operation timeout");
| > }
| >
| > if ( ! isOpened )
| > throw new InvalidOperationException("Queue closed");
| > return q.Dequeue();
| > }
| > }
| >
| > public bool TryDequeue(int millisecondsTimeout, out T value)
| > {
| > lock (syncRoot)
| > {
| > while (isOpened && (q.Count == 0))
| > {
| > if ( ! Monitor.Wait(syncRoot, millisecondsTimeout) )
| > {
| > value = default(T);
| > return false;
| > }
| > }
| >
| > if (! isOpened)
| > throw new InvalidOperationException("Queue closed");
| > value = q.Dequeue();
| > return true;
| > }
| > }
| >
| > /// <summary>
| > /// Returns the object at the beginning of the BlockingQueue<T>
| > /// without removing it.
| > /// </summary>
| > /// <returns>The object at the beginning of the
| > BlockingQueue<T>.</returns>
| > public T Peek()
| > {
| > return Peek(-1);
| > }
| >
| > /// <summary>
| > /// Returns the object at the beginning of the BlockingQueue<T>
| > /// without removing it.
| > /// </summary>
| > /// <returns>The object at the beginning of the
| > BlockingQueue<T>.</returns>
| > /// <param name="millisecondsTimeout">Time to wait before
returning
| > (in milliseconds).</param>
| > public T Peek(int millisecondsTimeout)
| > {
| > lock ( syncRoot )
| > {
| > while ( isOpened && (q.Count == 0) )
| > {
| > if ( ! Monitor.Wait(syncRoot, millisecondsTimeout) )
| > throw new TimeoutException("Operation timeout");
| > }
| >
| > if ( ! isOpened )
| > throw new InvalidOperationException("Queue closed");
| > return q.Peek();
| > }
| > }
| >
| > /// <summary>
| > /// Adds an object to the end of the Queue.
| > /// </summary>
| > /// <param name="obj">Object to put in queue.</param>
| > public void Enqueue(T item)
| > {
| > lock ( syncRoot )
| > {
| > if ( ! isOpened )
| > throw new InvalidOperationException("Queue closed");
| > q.Enqueue(item);
| > Monitor.Pulse(syncRoot); // Move 1 waiting thread to
the
| > "ready" queue in this monitor object.
| > } // Exiting lock will free thread(s) in the "ready" queue
for
| > this monitor object.
| > }
| >
| > [Serializable, StructLayout(LayoutKind.Sequential)]
| > public struct Enumerator : IEnumerator<T>, IDisposable,
IEnumerator
| > {
| > private BlockingQueue<T> q;
| > private IEnumerator<T> e;
| >