#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; } } } }