--- /dev/null
+using System;
+using System.Collections.Concurrent;
+using System.Collections.Generic;
+using System.Net;
+using System.Net.Sockets;
+using System.Threading;
+
+namespace Hazel.Udp.FewerThreads
+{
+ /// <summary>
+ /// Listens for new UDP connections and creates UdpConnections for them.
+ /// </summary>
+ /// <inheritdoc />
+ public class ThreadLimitedUdpConnectionListener : IDisposable
+ {
+ private struct SendMessageInfo
+ {
+ public byte[] Buffer;
+ public EndPoint Recipient;
+ }
+
+ private struct ReceiveMessageInfo
+ {
+ public MessageReader Message;
+ public EndPoint Sender;
+ }
+
+ private const int SendReceiveBufferSize = 1024 * 1024;
+ private const int BufferSize = ushort.MaxValue;
+
+ public event Action<NewConnectionEventArgs> NewConnection;
+
+ /// <summary>
+ /// A callback for early connection rejection.
+ /// * Return false to reject connection.
+ /// * A null response is ok, we just won't send anything.
+ /// </summary>
+ public AcceptConnectionCheck AcceptConnection;
+ public delegate bool AcceptConnectionCheck(IPEndPoint endPoint, byte[] input, out byte[] response);
+
+ private Socket socket;
+ private ILogger Logger;
+
+ public IPEndPoint EndPoint { get; }
+ public IPMode IPMode { get; }
+
+ private Thread reliablePacketThread;
+ private Thread receiveThread;
+ private Thread sendThread;
+ private HazelThreadPool processThreads;
+
+ private ConcurrentDictionary<EndPoint, ThreadLimitedUdpServerConnection> allConnections = new ConcurrentDictionary<EndPoint, ThreadLimitedUdpServerConnection>();
+
+ private Queue<ReceiveMessageInfo> receiveQueue = new Queue<ReceiveMessageInfo>();
+ private Queue<SendMessageInfo> sendQueue = new Queue<SendMessageInfo>();
+
+ public int ConnectionCount { get { return this.allConnections.Count; } }
+ public int SendQueueLength { get { lock(this.sendQueue) return this.sendQueue.Count; } }
+ public int ReceiveQueueLength { get { lock (this.receiveQueue) return this.receiveQueue.Count; } }
+
+ private bool isActive;
+
+ public ThreadLimitedUdpConnectionListener(int numWorkers, IPEndPoint endPoint, ILogger logger, IPMode ipMode = IPMode.IPv4)
+ {
+ this.Logger = logger;
+ this.EndPoint = endPoint;
+ this.IPMode = ipMode;
+
+ this.socket = UdpConnection.CreateSocket(this.IPMode);
+ this.socket.Blocking = false;
+
+ this.socket.ReceiveBufferSize = SendReceiveBufferSize;
+ this.socket.SendBufferSize = SendReceiveBufferSize;
+
+ this.reliablePacketThread = new Thread(ManageReliablePackets);
+ this.sendThread = new Thread(SendLoop);
+ this.receiveThread = new Thread(ReceiveLoop);
+ this.processThreads = new HazelThreadPool(numWorkers, ProcessingLoop);
+ }
+
+ ~ThreadLimitedUdpConnectionListener()
+ {
+ this.Dispose(false);
+ }
+
+ private void ManageReliablePackets()
+ {
+ while (this.isActive)
+ {
+ foreach (var kvp in this.allConnections)
+ {
+ var sock = kvp.Value;
+ sock.ManageReliablePackets();
+ }
+
+ Thread.Sleep(100);
+ }
+ }
+
+ public void Start()
+ {
+ try
+ {
+ socket.Bind(EndPoint);
+ }
+ catch (SocketException e)
+ {
+ throw new HazelException("Could not start listening as a SocketException occurred", e);
+ }
+
+ this.isActive = true;
+ this.reliablePacketThread.Start();
+ this.sendThread.Start();
+ this.receiveThread.Start();
+ this.processThreads.Start();
+ }
+
+ private void ReceiveLoop()
+ {
+ while (this.isActive)
+ {
+ if (this.socket.Poll(Timeout.Infinite, SelectMode.SelectRead))
+ {
+ EndPoint remoteEP = new IPEndPoint(this.EndPoint.Address, this.EndPoint.Port);
+ MessageReader message = MessageReader.GetSized(BufferSize);
+ try
+ {
+ message.Length = socket.ReceiveFrom(message.Buffer, 0, message.Buffer.Length, SocketFlags.None, ref remoteEP);
+ }
+ catch (SocketException sx)
+ {
+ message.Recycle();
+ this.Logger.WriteError("Socket Ex in StartListening: " + sx.Message);
+ continue;
+ }
+ catch (Exception ex)
+ {
+ message.Recycle();
+ this.Logger.WriteError("Stopped due to: " + ex.Message);
+ return;
+ }
+
+ lock (this.receiveQueue)
+ {
+ this.receiveQueue.Enqueue(new ReceiveMessageInfo() { Message = message, Sender = remoteEP });
+ Monitor.Pulse(this.receiveQueue);
+ }
+ }
+ }
+ }
+
+ private void ProcessingLoop()
+ {
+ while (this.isActive)
+ {
+ ReceiveMessageInfo msg;
+ lock (this.receiveQueue)
+ {
+ if (this.receiveQueue.Count == 0)
+ {
+ Monitor.Wait(this.receiveQueue);
+
+ if (this.receiveQueue.Count == 0)
+ {
+ continue;
+ }
+ }
+
+ msg = this.receiveQueue.Dequeue();
+ }
+
+ try
+ {
+ this.ReadCallback(msg.Message, msg.Sender);
+ }
+ catch
+ {
+ }
+ }
+ }
+
+ private void SendLoop()
+ {
+ while (this.isActive)
+ {
+ SendMessageInfo msg;
+ lock (this.sendQueue)
+ {
+ if (this.sendQueue.Count == 0)
+ {
+ Monitor.Wait(this.sendQueue);
+
+ if (this.sendQueue.Count == 0)
+ {
+ continue;
+ }
+ }
+
+ msg = this.sendQueue.Dequeue();
+ }
+
+ try
+ {
+ this.socket.SendTo(msg.Buffer, 0, msg.Buffer.Length, SocketFlags.None, msg.Recipient);
+ }
+ catch { }
+ }
+ }
+
+ void ReadCallback(MessageReader message, EndPoint remoteEndPoint)
+ {
+ int bytesReceived = message.Length;
+ bool aware = true;
+ bool isHello = message.Buffer[0] == (byte)UdpSendOption.Hello;
+
+ // If we're aware of this connection use the one already
+ // If this is a new client then connect with them!
+ ThreadLimitedUdpServerConnection connection;
+ if (!this.allConnections.TryGetValue(remoteEndPoint, out connection))
+ {
+ lock (this.allConnections)
+ {
+ if (!this.allConnections.TryGetValue(remoteEndPoint, out connection))
+ {
+ // Check for malformed connection attempts
+ if (!isHello)
+ {
+ message.Recycle();
+ return;
+ }
+
+ if (AcceptConnection != null)
+ {
+ if (!AcceptConnection((IPEndPoint)remoteEndPoint, message.Buffer, out var response))
+ {
+ message.Recycle();
+ if (response != null)
+ {
+ SendDataRaw(response, remoteEndPoint);
+ }
+
+ return;
+ }
+ }
+
+ aware = false;
+ connection = new ThreadLimitedUdpServerConnection(this, (IPEndPoint)remoteEndPoint, this.IPMode);
+ if (!this.allConnections.TryAdd(remoteEndPoint, connection))
+ {
+ throw new HazelException("Failed to add a connection. This should never happen.");
+ }
+ }
+ }
+ }
+
+ // If it's a new connection invoke the NewConnection event.
+ // This needs to happen before handling the message because in localhost scenarios, the ACK and
+ // subsequent messages can happen before the NewConnection event sets up OnDataRecieved handlers
+ if (!aware)
+ {
+ // Skip header and hello byte;
+ message.Offset = 4;
+ message.Length = bytesReceived - 4;
+ message.Position = 0;
+ this.NewConnection?.Invoke(new NewConnectionEventArgs(message, connection));
+ }
+
+ // Inform the connection of the buffer (new connections need to send an ack back to client)
+ connection.HandleReceive(message, bytesReceived);
+
+ if (isHello)
+ {
+ message.Recycle();
+ }
+ }
+
+ internal void SendDataRaw(byte[] response, EndPoint remoteEndPoint)
+ {
+ lock (this.sendQueue)
+ {
+ this.sendQueue.Enqueue(new SendMessageInfo() { Buffer = response, Recipient = remoteEndPoint });
+ Monitor.Pulse(this.sendQueue);
+ }
+ }
+
+ /// <summary>
+ /// Removes a virtual connection from the list.
+ /// </summary>
+ /// <param name="endPoint">The endpoint of the virtual connection.</param>
+ internal bool RemoveConnectionTo(EndPoint endPoint)
+ {
+ return this.allConnections.TryRemove(endPoint, out var conn);
+ }
+
+ protected virtual void Dispose(bool disposing)
+ {
+ foreach (var kvp in this.allConnections)
+ {
+ kvp.Value.Dispose();
+ }
+
+ try { this.socket.Shutdown(SocketShutdown.Both); } catch { }
+ try { this.socket.Close(); } catch { }
+ try { this.socket.Dispose(); } catch { }
+
+ this.isActive = false;
+
+ lock (this.sendQueue) Monitor.PulseAll(this.sendQueue);
+ lock (this.receiveQueue) Monitor.PulseAll(this.receiveQueue);
+
+ this.reliablePacketThread.Join();
+ this.sendThread.Join();
+ this.receiveThread.Join();
+ this.processThreads.Join();
+ }
+
+ public void Dispose()
+ {
+ this.Dispose(true);
+ }
+ }
+}
--- /dev/null
+using System;
+using System.Net;
+
+namespace Hazel.Udp.FewerThreads
+{
+ /// <summary>
+ /// Represents a servers's connection to a client that uses the UDP protocol.
+ /// </summary>
+ /// <inheritdoc/>
+ internal sealed class ThreadLimitedUdpServerConnection : UdpConnection
+ {
+ /// <summary>
+ /// The connection listener that we use the socket of.
+ /// </summary>
+ /// <remarks>
+ /// Udp server connections utilize the same socket in the listener for sends/receives, this is the listener that
+ /// created this connection and is hence the listener this conenction sends and receives via.
+ /// </remarks>
+ public ThreadLimitedUdpConnectionListener Listener { get; private set; }
+
+ /// <summary>
+ /// Creates a UdpConnection for the virtual connection to the endpoint.
+ /// </summary>
+ /// <param name="listener">The listener that created this connection.</param>
+ /// <param name="endPoint">The endpoint that we are connected to.</param>
+ /// <param name="IPMode">The IPMode we are connected using.</param>
+ internal ThreadLimitedUdpServerConnection(ThreadLimitedUdpConnectionListener listener, IPEndPoint endPoint, IPMode IPMode)
+ : base()
+ {
+ this.Listener = listener;
+ this.RemoteEndPoint = endPoint;
+ this.EndPoint = endPoint;
+ this.IPMode = IPMode;
+
+ State = ConnectionState.Connected;
+ this.InitializeKeepAliveTimer();
+ }
+
+ /// <inheritdoc />
+ protected override void WriteBytesToConnection(byte[] bytes, int length)
+ {
+ if (bytes.Length != length) throw new ArgumentException("I made an assumption here. I hope you see this error.");
+
+ Listener.SendDataRaw(bytes, RemoteEndPoint);
+ }
+
+ /// <inheritdoc />
+ /// <remarks>
+ /// This will always throw a HazelException.
+ /// </remarks>
+ public override void Connect(byte[] bytes = null, int timeout = 5000)
+ {
+ throw new InvalidOperationException("Cannot manually connect a UdpServerConnection, did you mean to use UdpClientConnection?");
+ }
+
+ /// <inheritdoc />
+ /// <remarks>
+ /// This will always throw a HazelException.
+ /// </remarks>
+ public override void ConnectAsync(byte[] bytes = null)
+ {
+ throw new InvalidOperationException("Cannot manually connect a UdpServerConnection, did you mean to use UdpClientConnection?");
+ }
+
+ /// <summary>
+ /// Sends a disconnect message to the end point.
+ /// </summary>
+ protected override bool SendDisconnect(MessageWriter data = null)
+ {
+ if (!Listener.RemoveConnectionTo(RemoteEndPoint)) return false;
+ this._state = ConnectionState.NotConnected;
+
+ var bytes = EmptyDisconnectBytes;
+ if (data != null && data.Length > 0)
+ {
+ if (data.SendOption != SendOption.None) throw new ArgumentException("Disconnect messages can only be unreliable.");
+
+ bytes = data.ToByteArray(true);
+ bytes[0] = (byte)UdpSendOption.Disconnect;
+ }
+
+ try
+ {
+ Listener.SendDataRaw(bytes, RemoteEndPoint);
+ }
+ catch { }
+
+ return true;
+ }
+
+ protected override void Dispose(bool disposing)
+ {
+ if (disposing)
+ {
+ SendDisconnect();
+ }
+
+ base.Dispose(disposing);
+ }
+ }
+}
+++ /dev/null
-using System;
-using System.Collections.Concurrent;
-using System.Collections.Generic;
-using System.Net;
-using System.Net.Sockets;
-using System.Threading;
-
-namespace Hazel.Udp.FewerThreads
-{
- /// <summary>
- /// Listens for new UDP connections and creates UdpConnections for them.
- /// </summary>
- /// <inheritdoc />
- public class UdpConnectionListener2 : IDisposable
- {
- private struct SendMessageInfo
- {
- public byte[] Buffer;
- public EndPoint Recipient;
- }
-
- private struct ReceiveMessageInfo
- {
- public MessageReader Message;
- public EndPoint Sender;
- }
-
- private const int SendReceiveBufferSize = 1024 * 1024;
- private const int BufferSize = ushort.MaxValue;
-
- public event Action<NewConnectionEventArgs> NewConnection;
-
- /// <summary>
- /// A callback for early connection rejection.
- /// * Return false to reject connection.
- /// * A null response is ok, we just won't send anything.
- /// </summary>
- public AcceptConnectionCheck AcceptConnection;
- public delegate bool AcceptConnectionCheck(IPEndPoint endPoint, byte[] input, out byte[] response);
-
- private Socket socket;
- private ILogger Logger;
-
- public IPEndPoint EndPoint { get; }
- public IPMode IPMode { get; }
-
- private Thread reliablePacketThread;
- private Thread receiveThread;
- private Thread sendThread;
- private HazelThreadPool processThreads;
-
- private ConcurrentDictionary<EndPoint, UdpServerConnection2> allConnections = new ConcurrentDictionary<EndPoint, UdpServerConnection2>();
-
- private Queue<ReceiveMessageInfo> receiveQueue = new Queue<ReceiveMessageInfo>();
- private Queue<SendMessageInfo> sendQueue = new Queue<SendMessageInfo>();
-
- public int ConnectionCount { get { return this.allConnections.Count; } }
- public int SendQueueLength { get { lock(this.sendQueue) return this.sendQueue.Count; } }
- public int ReceiveQueueLength { get { lock (this.receiveQueue) return this.receiveQueue.Count; } }
-
- private bool isActive;
-
- public UdpConnectionListener2(IPEndPoint endPoint, ILogger logger, IPMode ipMode = IPMode.IPv4)
- {
- this.Logger = logger;
- this.EndPoint = endPoint;
- this.IPMode = ipMode;
-
- this.socket = UdpConnection.CreateSocket(this.IPMode);
- this.socket.Blocking = false;
-
- this.socket.ReceiveBufferSize = SendReceiveBufferSize;
- this.socket.SendBufferSize = SendReceiveBufferSize;
-
- this.reliablePacketThread = new Thread(ManageReliablePackets);
- this.sendThread = new Thread(SendLoop);
- this.receiveThread = new Thread(ReceiveLoop);
- this.processThreads = new HazelThreadPool(4, ProcessingLoop);
- }
-
- ~UdpConnectionListener2()
- {
- this.Dispose(false);
- }
-
- private void ManageReliablePackets()
- {
- while (this.isActive)
- {
- foreach (var kvp in this.allConnections)
- {
- var sock = kvp.Value;
- sock.ManageReliablePackets();
- }
-
- Thread.Sleep(100);
- }
- }
-
- public void Start()
- {
- try
- {
- socket.Bind(EndPoint);
- }
- catch (SocketException e)
- {
- throw new HazelException("Could not start listening as a SocketException occurred", e);
- }
-
- this.isActive = true;
- this.reliablePacketThread.Start();
- this.sendThread.Start();
- this.receiveThread.Start();
- this.processThreads.Start();
- }
-
- private void ReceiveLoop()
- {
- while (this.isActive)
- {
- if (this.socket.Poll(Timeout.Infinite, SelectMode.SelectRead))
- {
- EndPoint remoteEP = new IPEndPoint(IPMode == IPMode.IPv4 ? IPAddress.Any : IPAddress.IPv6Any, this.EndPoint.Port);
- MessageReader message = MessageReader.GetSized(BufferSize);
- try
- {
- message.Length = socket.ReceiveFrom(message.Buffer, 0, message.Buffer.Length, SocketFlags.None, ref remoteEP);
- }
- catch (SocketException sx)
- {
- message.Recycle();
- this.Logger.WriteError("Socket Ex in StartListening: " + sx.Message);
- continue;
- }
- catch (Exception ex)
- {
- message.Recycle();
- this.Logger.WriteError("Stopped due to: " + ex.Message);
- return;
- }
-
- lock (this.receiveQueue)
- {
- this.receiveQueue.Enqueue(new ReceiveMessageInfo() { Message = message, Sender = remoteEP });
- Monitor.Pulse(this.receiveQueue);
- }
- }
- }
- }
-
- private void ProcessingLoop()
- {
- while (this.isActive)
- {
- ReceiveMessageInfo msg;
- lock (this.receiveQueue)
- {
- if (this.receiveQueue.Count == 0)
- {
- Monitor.Wait(this.receiveQueue);
-
- if (this.receiveQueue.Count == 0)
- {
- continue;
- }
- }
-
- msg = this.receiveQueue.Dequeue();
- }
-
- try
- {
- this.ReadCallback(msg.Message, msg.Sender);
- }
- catch
- {
- }
- }
- }
-
- private void SendLoop()
- {
- while (this.isActive)
- {
- SendMessageInfo msg;
- lock (this.sendQueue)
- {
- if (this.sendQueue.Count == 0)
- {
- Monitor.Wait(this.sendQueue);
-
- if (this.sendQueue.Count == 0)
- {
- continue;
- }
- }
-
- msg = this.sendQueue.Dequeue();
- }
-
- try
- {
- this.socket.SendTo(msg.Buffer, 0, msg.Buffer.Length, SocketFlags.None, msg.Recipient);
- }
- catch { }
- }
- }
-
- void ReadCallback(MessageReader message, EndPoint remoteEndPoint)
- {
- int bytesReceived = message.Length;
- bool aware = true;
- bool isHello = message.Buffer[0] == (byte)UdpSendOption.Hello;
-
- // If we're aware of this connection use the one already
- // If this is a new client then connect with them!
- UdpServerConnection2 connection;
- if (!this.allConnections.TryGetValue(remoteEndPoint, out connection))
- {
- lock (this.allConnections)
- {
- if (!this.allConnections.TryGetValue(remoteEndPoint, out connection))
- {
- // Check for malformed connection attempts
- if (!isHello)
- {
- message.Recycle();
- return;
- }
-
- if (AcceptConnection != null)
- {
- if (!AcceptConnection((IPEndPoint)remoteEndPoint, message.Buffer, out var response))
- {
- message.Recycle();
- if (response != null)
- {
- SendDataRaw(response, remoteEndPoint);
- }
-
- return;
- }
- }
-
- aware = false;
- connection = new UdpServerConnection2(this, (IPEndPoint)remoteEndPoint, this.IPMode);
- if (!this.allConnections.TryAdd(remoteEndPoint, connection))
- {
- throw new HazelException("Failed to add a connection. This should never happen.");
- }
- }
- }
- }
-
- //Inform the connection of the buffer (new connections need to send an ack back to client)
- connection.HandleReceive(message, bytesReceived);
-
- //If it's a new connection invoke the NewConnection event.
- if (!aware)
- {
- // Skip header and hello byte;
- message.Offset = 4;
- message.Length = bytesReceived - 4;
- message.Position = 0;
- this.NewConnection?.Invoke(new NewConnectionEventArgs(message, connection));
- }
- else if (isHello)
- {
- message.Recycle();
- }
- }
-
- internal void SendDataRaw(byte[] response, EndPoint remoteEndPoint)
- {
- lock (this.sendQueue)
- {
- this.sendQueue.Enqueue(new SendMessageInfo() { Buffer = response, Recipient = remoteEndPoint });
- Monitor.Pulse(this.sendQueue);
- }
- }
-
- /// <summary>
- /// Removes a virtual connection from the list.
- /// </summary>
- /// <param name="endPoint">The endpoint of the virtual connection.</param>
- internal bool RemoveConnectionTo(EndPoint endPoint)
- {
- return this.allConnections.TryRemove(endPoint, out var conn);
- }
-
- protected virtual void Dispose(bool disposing)
- {
- foreach (var kvp in this.allConnections)
- {
- kvp.Value.Dispose();
- }
-
- try { this.socket.Shutdown(SocketShutdown.Both); } catch { }
- try { this.socket.Close(); } catch { }
- try { this.socket.Dispose(); } catch { }
-
- this.isActive = false;
-
- lock (this.sendQueue) Monitor.PulseAll(this.sendQueue);
- lock (this.receiveQueue) Monitor.PulseAll(this.receiveQueue);
-
- this.reliablePacketThread.Join();
- this.sendThread.Join();
- this.receiveThread.Join();
- this.processThreads.Join();
- }
-
- public void Dispose()
- {
- this.Dispose(true);
- }
- }
-}
+++ /dev/null
-using System;
-using System.Net;
-
-namespace Hazel.Udp.FewerThreads
-{
- /// <summary>
- /// Represents a servers's connection to a client that uses the UDP protocol.
- /// </summary>
- /// <inheritdoc/>
- internal sealed class UdpServerConnection2 : UdpConnection
- {
- /// <summary>
- /// The connection listener that we use the socket of.
- /// </summary>
- /// <remarks>
- /// Udp server connections utilize the same socket in the listener for sends/receives, this is the listener that
- /// created this connection and is hence the listener this conenction sends and receives via.
- /// </remarks>
- public UdpConnectionListener2 Listener { get; private set; }
-
- /// <summary>
- /// Creates a UdpConnection for the virtual connection to the endpoint.
- /// </summary>
- /// <param name="listener">The listener that created this connection.</param>
- /// <param name="endPoint">The endpoint that we are connected to.</param>
- /// <param name="IPMode">The IPMode we are connected using.</param>
- internal UdpServerConnection2(UdpConnectionListener2 listener, IPEndPoint endPoint, IPMode IPMode)
- : base()
- {
- this.Listener = listener;
- this.RemoteEndPoint = endPoint;
- this.EndPoint = endPoint;
- this.IPMode = IPMode;
-
- State = ConnectionState.Connected;
- this.InitializeKeepAliveTimer();
- }
-
- /// <inheritdoc />
- protected override void WriteBytesToConnection(byte[] bytes, int length)
- {
- if (bytes.Length != length) throw new ArgumentException("I made an assumption here. I hope you see this error.");
-
- Listener.SendDataRaw(bytes, RemoteEndPoint);
- }
-
- /// <inheritdoc />
- /// <remarks>
- /// This will always throw a HazelException.
- /// </remarks>
- public override void Connect(byte[] bytes = null, int timeout = 5000)
- {
- throw new InvalidOperationException("Cannot manually connect a UdpServerConnection, did you mean to use UdpClientConnection?");
- }
-
- /// <inheritdoc />
- /// <remarks>
- /// This will always throw a HazelException.
- /// </remarks>
- public override void ConnectAsync(byte[] bytes = null)
- {
- throw new InvalidOperationException("Cannot manually connect a UdpServerConnection, did you mean to use UdpClientConnection?");
- }
-
- /// <summary>
- /// Sends a disconnect message to the end point.
- /// </summary>
- protected override bool SendDisconnect(MessageWriter data = null)
- {
- if (!Listener.RemoveConnectionTo(RemoteEndPoint)) return false;
- this._state = ConnectionState.NotConnected;
-
- var bytes = EmptyDisconnectBytes;
- if (data != null && data.Length > 0)
- {
- if (data.SendOption != SendOption.None) throw new ArgumentException("Disconnect messages can only be unreliable.");
-
- bytes = data.ToByteArray(true);
- bytes[0] = (byte)UdpSendOption.Disconnect;
- }
-
- try
- {
- Listener.SendDataRaw(bytes, RemoteEndPoint);
- }
- catch { }
-
- return true;
- }
-
- protected override void Dispose(bool disposing)
- {
- if (disposing)
- {
- SendDisconnect();
- }
-
- base.Dispose(disposing);
- }
- }
-}
<LangVersion>7.3</LangVersion>
</PropertyGroup>
<PropertyGroup Condition=" '$(Configuration)|$(Platform)' == 'Release|AnyCPU' ">
- <DebugType>pdbonly</DebugType>
+ <DebugType>portable</DebugType>
<Optimize>true</Optimize>
<OutputPath>bin\Release\</OutputPath>
<DefineConstants>TRACE</DefineConstants>
<AllowUnsafeBlocks>true</AllowUnsafeBlocks>
<Prefer32Bit>false</Prefer32Bit>
<LangVersion>7.3</LangVersion>
+ <DebugSymbols>true</DebugSymbols>
</PropertyGroup>
<PropertyGroup>
<SignAssembly>true</SignAssembly>
<Compile Include="DataReceivedEventArgs.cs" />
<Compile Include="DisconnectedEventArgs.cs" />
<Compile Include="FewerThreads\HazelThreadPool.cs" />
- <Compile Include="FewerThreads\UdpConnectionListener2.cs" />
- <Compile Include="FewerThreads\UdpServerConnection2.cs" />
+ <Compile Include="FewerThreads\ThreadLimitedUdpConnectionListener.cs" />
+ <Compile Include="FewerThreads\ThreadLimitedUdpServerConnection.cs" />
<Compile Include="HazelException.cs" />
<Compile Include="IPMode.cs" />
<Compile Include="IRecyclable.cs" />
<Compile Include="UPnP\NetUtility.cs" />
<Compile Include="UPnP\UPnPHelper.cs" />
</ItemGroup>
- <ItemGroup />
<Import Project="$(MSBuildToolsPath)\Microsoft.CSharp.targets" />
<!-- To modify your build process, add your task inside one of the targets below and uncomment it.
Other similar extension points exist, see Microsoft.Common.targets.
using System;
using System.Collections.Concurrent;
+using System.Collections.Generic;
using System.Threading;
namespace Hazel
public int NumberNotInUse { get { return this.pool.Count; } }
public int Size { get { return this.NumberInUse + this.NumberNotInUse; } }
- // Available objects
+#if HAZEL_BAG
private readonly ConcurrentBag<T> pool = new ConcurrentBag<T>();
+#else
+ private readonly List<T> pool = new List<T>();
+#endif
// Unavailable objects
private readonly ConcurrentDictionary<T, bool> inuse = new ConcurrentDictionary<T, bool>();
/// <returns>An instance of T.</returns>
internal T GetObject()
{
+#if HAZEL_BAG
if (!pool.TryTake(out T item))
{
Interlocked.Increment(ref numberCreated);
item = objectFactory.Invoke();
}
+#else
+ T item;
+ lock (this.pool)
+ {
+ if (this.pool.Count > 0)
+ {
+ var idx = this.pool.Count - 1;
+ item = this.pool[idx];
+ this.pool.RemoveAt(idx);
+ }
+ else
+ {
+ Interlocked.Increment(ref numberCreated);
+ item = objectFactory.Invoke();
+ }
+ }
+#endif
if (!inuse.TryAdd(item, true))
{
{
if (inuse.TryRemove(item, out bool b))
{
+#if HAZEL_BAG
pool.Add(item);
+#else
+ lock (this.pool)
+ {
+ pool.Add(item);
+ }
+#endif
}
else
{
protected void InitializeKeepAliveTimer()
{
keepAliveTimer = new Timer(
- (o) =>
- {
- if (this.State != ConnectionState.Connected) return;
-
- if (this.pingsSinceAck >= this.MissingPingsUntilDisconnect)
- {
- this.DisposeKeepAliveTimer();
- this.DisconnectInternal(HazelInternalErrors.PingsWithoutResponse, $"Sent {this.pingsSinceAck} pings that remote has not responded to.");
- return;
- }
-
- try
- {
- this.pingsSinceAck++;
- SendPing();
- }
- catch
- {
- }
- },
+ HandleKeepAlive,
null,
keepAliveInterval,
keepAliveInterval
);
}
+ private void HandleKeepAlive(object state)
+ {
+ if (this.State != ConnectionState.Connected) return;
+
+ if (this.pingsSinceAck >= this.MissingPingsUntilDisconnect)
+ {
+ this.DisposeKeepAliveTimer();
+ this.DisconnectInternal(HazelInternalErrors.PingsWithoutResponse, $"Sent {this.pingsSinceAck} pings that remote has not responded to.");
+ return;
+ }
+
+ try
+ {
+ this.pingsSinceAck++;
+ SendPing();
+ }
+ catch
+ {
+ }
+ }
+
// Pings are special, quasi-reliable packets.
// We send them to trigger responses that validate our connection is alive
// An unacked ping should never be the sole cause of a disconnect.
internal ConcurrentDictionary<ushort, Packet> reliableDataPacketsSent = new ConcurrentDictionary<ushort, Packet>();
/// <summary>
- /// The last packets that were received.
+ /// Packet ids that have not been received, but are expected.
/// </summary>
private HashSet<ushort> reliableDataPacketsMissing = new HashSet<ushort>();
//If it's new or we've not received anything yet
if (isNew)
{
- //Mark items between the most recent receive and the id received as missing
- for (ushort i = (ushort)(reliableReceiveLast + 1); i < id; i++)
+ // Mark items between the most recent receive and the id received as missing
+ if (id > reliableReceiveLast)
{
- reliableDataPacketsMissing.Add(i);
+ for (ushort i = (ushort)(reliableReceiveLast + 1); i < id; i++)
+ {
+ reliableDataPacketsMissing.Add(i);
+ }
+ }
+ else
+ {
+ int cnt = (ushort.MaxValue - reliableReceiveLast) + id;
+ for (ushort i = 1; i < cnt; ++i)
+ {
+ reliableDataPacketsMissing.Add((ushort)(i + reliableReceiveLast));
+ }
}
//Update the most recently received
{
this.pingsSinceAck = 0;
- // Get ID
ushort id = (ushort)((bytes[1] << 8) + bytes[2]);
AcknowledgeMessageId(id);
recentPackets >>= 1;
}
-
- Statistics.LogReliableReceive(bytesReceived - 4, bytesReceived);
- }
- else
- {
- Statistics.LogReliableReceive(bytesReceived - 3, bytesReceived);
}
+
+ Statistics.LogReliableReceive(0, bytesReceived);
}
private void AcknowledgeMessageId(ushort id)
/// <param name="byte2">The second identification byte.</param>
private void SendAck(ushort id)
{
- const byte Found = 1;
- const byte Missing = 1;
-
byte recentPackets = 0;
lock (this.reliableDataPacketsMissing)
{
for (int i = 1; i <= 8; ++i)
{
- recentPackets |= this.reliableDataPacketsMissing.Contains((ushort)(id - i)) ? Found : Missing;
- recentPackets <<= 1;
+ if (!this.reliableDataPacketsMissing.Contains((ushort)(id - i)))
+ {
+ recentPackets |= (byte)(1 << (i - 1));
+ }
}
}
recentPackets
};
- // Always reply with acknowledgement in order to stop the sender repeatedly sending it
- // TODO: group acks together
try
{
WriteBytesToConnection(bytes, bytes.Length);
this.IPMode = ipMode;
this.socket = UdpConnection.CreateSocket(this.IPMode);
-
+
socket.ReceiveBufferSize = SendReceiveBufferSize;
socket.SendBufferSize = SendReceiveBufferSize;
{
var message = (MessageReader)result.AsyncState;
int bytesReceived;
- EndPoint remoteEndPoint = new IPEndPoint(IPMode == IPMode.IPv4 ? IPAddress.Any : IPAddress.IPv6Any, 0);
+ EndPoint remoteEndPoint = new IPEndPoint(this.EndPoint.Address, this.EndPoint.Port);
//End the receive operation
try
InvokeNewConnection(message, connection);
}
- //Inform the connection of the buffer (new connections need to send an ack back to client)
+ // Inform the connection of the buffer (new connections need to send an ack back to client)
connection.HandleReceive(message, bytesReceived);
if (aware && isHello)
{
private Socket socket;
+ public EndPoint LocalEndpoint { get { return this.socket.LocalEndPoint; } }
+
public UnityUdpClientConnection(IPEndPoint remoteEndPoint, IPMode ipMode = IPMode.IPv4)
: base()
{
this.IPMode = ipMode;
this.socket = CreateSocket(ipMode);
+ this.socket.SetSocketOption(SocketOptionLevel.Socket, SocketOptionName.ExclusiveAddressUse, true);
}
~UnityUdpClientConnection()
var msg = MessageReader.GetSized(ushort.MaxValue);
try
{
- socket.BeginReceive(msg.Buffer, 0, msg.Buffer.Length, SocketFlags.None, ReadCallback, msg);
+ var ep = this.RemoteEndPoint;
+ socket.BeginReceiveFrom(msg.Buffer, 0, msg.Buffer.Length, SocketFlags.None, ref ep, ReadCallback, msg);
}
catch
{
try
{
- msg.Length = socket.EndReceive(result);
+ var ep = this.RemoteEndPoint;
+ msg.Length = socket.EndReceiveFrom(result, ref ep);
}
catch (SocketException e)
{
projects / rhonda / impostor.hazel.git / commitdiff