/*************************************************************************** * SendQueue.cs * ------------------- * begin : May 1, 2002 * copyright : (C) The RunUO Software Team * email : info@runuo.com * * $Id$ * ***************************************************************************/ /*************************************************************************** * * This program is free software; you can redistribute it and/or modify * it under the terms of the GNU General Public License as published by * the Free Software Foundation; either version 2 of the License, or * (at your option) any later version. * ***************************************************************************/ using System; using System.Collections.Generic; namespace Server.Network { public class SendQueue { private const int PendingCap = 256 * 1024; private static int m_CoalesceBufferSize = 512; private static BufferPool m_UnusedBuffers = new BufferPool("Coalesced", 2048, m_CoalesceBufferSize); private Gram _buffered; private Queue _pending; public SendQueue() { _pending = new Queue(); } public static int CoalesceBufferSize { get => m_CoalesceBufferSize; set { if (m_CoalesceBufferSize == value) return; BufferPool old = m_UnusedBuffers; lock (old) { m_UnusedBuffers?.Free(); m_CoalesceBufferSize = value; m_UnusedBuffers = new BufferPool("Coalesced", 2048, m_CoalesceBufferSize); } } } public bool IsFlushReady => _pending.Count == 0 && _buffered != null; public bool IsEmpty => _pending.Count == 0 && _buffered == null; public static byte[] AcquireBuffer() { lock (m_UnusedBuffers) { return m_UnusedBuffers.AcquireBuffer(); } } public static void ReleaseBuffer(byte[] buffer) { lock (m_UnusedBuffers) { if (buffer != null && buffer.Length == m_CoalesceBufferSize) m_UnusedBuffers.ReleaseBuffer(buffer); } } public Gram CheckFlushReady() { Gram gram = _buffered; _pending.Enqueue(_buffered); _buffered = null; return gram; } public Gram Dequeue() { Gram gram = null; if (_pending.Count > 0) { _pending.Dequeue().Release(); if (_pending.Count > 0) gram = _pending.Peek(); } return gram; } public Gram Enqueue(byte[] buffer, int length) { return Enqueue(buffer, 0, length); } public Gram Enqueue(byte[] buffer, int offset, int length) { if (buffer == null) throw new ArgumentNullException("buffer"); if (!(offset >= 0 && offset < buffer.Length)) throw new ArgumentOutOfRangeException("offset", offset, "Offset must be greater than or equal to zero and less than the size of the buffer."); if (length < 0 || length > buffer.Length) throw new ArgumentOutOfRangeException("length", length, "Length cannot be less than zero or greater than the size of the buffer."); if (buffer.Length - offset < length) throw new ArgumentException("Offset and length do not point to a valid segment within the buffer."); int existingBytes = _pending.Count * m_CoalesceBufferSize + (_buffered?.Length ?? 0); if (existingBytes + length > PendingCap) throw new CapacityExceededException(); Gram gram = null; while (length > 0) { if (_buffered == null) _buffered = Gram.Acquire(); int bytesWritten = _buffered.Write(buffer, offset, length); offset += bytesWritten; length -= bytesWritten; if (_buffered.IsFull) { if (_pending.Count == 0) gram = _buffered; _pending.Enqueue(_buffered); _buffered = null; } } return gram; } public void Clear() { if (_buffered != null) { _buffered.Release(); _buffered = null; } while (_pending.Count > 0) _pending.Dequeue().Release(); } public class Gram { private static Stack _pool = new Stack(); private Gram() { } public byte[] Buffer{ get; private set; } public int Length{ get; private set; } public int Available => Buffer.Length - Length; public bool IsFull => Length == Buffer.Length; public static Gram Acquire() { lock (_pool) { Gram gram; if (_pool.Count > 0) gram = _pool.Pop(); else gram = new Gram(); gram.Buffer = AcquireBuffer(); gram.Length = 0; return gram; } } public int Write(byte[] buffer, int offset, int length) { int write = Math.Min(length, Available); System.Buffer.BlockCopy(buffer, offset, Buffer, Length, write); Length += write; return write; } public void Release() { lock (_pool) { _pool.Push(this); ReleaseBuffer(Buffer); } } } } [Serializable] public sealed class CapacityExceededException : Exception { public CapacityExceededException() : base("Too much data pending.") { } } }