ModernUO/Server/Network/SendQueue.cs

/***************************************************************************
 *                               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<Gram> _pending;

    public SendQueue()
    {
      _pending = new Queue<Gram>();
    }

    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<Gram> _pool = new Stack<Gram>();

      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.")
    {
    }
  }
}