#region Header
//********************************************************************
// author: Thomas "zeroflag" Kraemer
// author email: zeroflag@zeroflag.de
//
// Copyright (C) 2006-2007 Thomas "zeroflag" Kraemer
//
// license: (LGPL)
//
// This library is free software; you can redistribute it and/or
// modify it under the terms of the GNU Lesser General Public
// License as published by the Free Software Foundation; either
// version 2.1 of the License, or (at your option) any later version.
//
// This library is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
// Lesser General Public License for more details.
//
// You should have received a copy of the GNU Lesser General Public
// License along with this library; if not, write to the Free Software
// Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
//
// http://www.gnu.org/licenses/lgpl.html#TOC1
//
//*********************************************************************
#endregion Header
using System;
using System.Collections.Generic;
namespace Sage.Threading
{
///
/// A single reader single writer message queues without locking.
///
///
/// Update must be called by the writer when lots (= more than the buffer size, 2048 by default) of messages were sent before but nothing is added.
public class MessageQueue : Sage.Threading.IMessageQueue
where T : class // <-- must be a reference type or atomic type to be threadsafe
{
Queue m_WriteQueue = new Queue();
///
/// Queue for new values before they're added to the buffer (through Update).
///
protected Queue WriteQueue
{
get { return m_WriteQueue; }
}
#region MessageContainer
///
/// A box arround the stored objects.
/// With this one can set a value in the array without causing it to synchronize to the other threads.
///
protected class MessageContainer
{
T m_Value;
///
/// Gets or sets the value.
/// Getting the value will result in it's deletion (null).
///
public T Value
{
get
{
T value = m_Value;
if (value != null)
{
this.m_Value = null;
}
return value;
}
set
{
m_Value = value;
}
}
///
/// Whether the box has a value or not.
///
public bool HasValue
{
get
{
if (m_Value != null)
return true;
else
return false;
}
}
}
#endregion MessageContainer
public MessageQueue(int bufferSize)
{
this.m_Buffer = new MessageContainer[bufferSize];
this.Clear();
}
///
/// Creates a new message queue and initializes it's buffer.
///
public MessageQueue()
{
this.Clear();
}
protected void Clear()
{
for (int i = 0; i < this.Buffer.Length; i++)
{
this.Buffer[i] = new MessageContainer();
}
}
private int ReadIndex = 0;
private int WriteIndex = 0;
MessageContainer[] m_Buffer = new MessageContainer[2048];
///
/// The queue's ring buffer.
/// Default size is 2048 entries.
///
protected MessageContainer[] Buffer
{
get { return m_Buffer; }
}
#region IMessageQueue Members
///
/// Whether there are values ready to be read.
/// *** READER THREAD ***
///
public bool IsEmpty
{
get { return !this.Buffer[this.ReadIndex].HasValue; }
}
///
/// Pops a value from the queue.
/// *** READER THREAD ***
///
///
public T Pop()
{
if (this.Buffer[this.ReadIndex].HasValue)
{
// grab the current value...
T value = this.Buffer[this.ReadIndex].Value;
// continue to the next...
this.ReadIndex++;
if (this.ReadIndex >= this.Buffer.Length)
{
this.ReadIndex = 0;
}
return value;
}
else
{
// no value waiting...
return null;
}
}
///
/// Pushes a new value to the queue and updates the buffer.
/// *** WRITER THREAD ***
///
///
public void Push(T msg)
{
this.WriteQueue.Enqueue(msg);
this.Update();
}
///
/// Fills the buffer from the queue.
/// *** WRITER THREAD ***
///
public void Update()
{
while (this.PushToBuffer()) ;
}
#endregion
///
/// Pushes a single value from the WriteQueue to the Buffer.
///
/// True if a value was pushed, false if it wasn't (like no items in queue or no free cell in buffer).
protected bool PushToBuffer()
{
if (!this.Buffer[this.WriteIndex].HasValue && this.WriteQueue.Count > 0)
{
// get the message...
T value = this.WriteQueue.Dequeue();
// store the message...
this.Buffer[this.WriteIndex].Value = value;
// move the index...
this.WriteIndex++;
if (this.WriteIndex >= this.Buffer.Length)
{
this.WriteIndex = 0;
}
// yes, we pushed a new message from the write-queue to the buffer...
return true;
}
else
{
// no, the exchange table is full...
return false;
}
}
}
}