From ea8de98fb727ae2649cfb788b06d4483d1517212 Mon Sep 17 00:00:00 2001 From: Forest Date: Wed, 15 Jan 2020 13:08:15 -0800 Subject: [PATCH] Added a listener with a fixed number of threads. Still testing how that works in practice --- Hazel.UnitTests/UPnPTests.cs | 4 +- Hazel/Connection.cs | 19 +- Hazel/FewerThreads/HazelThreadPool.cs | 44 +++ Hazel/FewerThreads/UdpConnectionListener2.cs | 320 +++++++++++++++++++ Hazel/FewerThreads/UdpServerConnection2.cs | 101 ++++++ Hazel/Hazel.csproj | 3 + Hazel/UPnP/ILogger.cs | 19 +- Hazel/UPnP/UPnPHelper.cs | 12 +- 8 files changed, 506 insertions(+), 16 deletions(-) create mode 100644 Hazel/FewerThreads/HazelThreadPool.cs create mode 100644 Hazel/FewerThreads/UdpConnectionListener2.cs create mode 100644 Hazel/FewerThreads/UdpServerConnection2.cs diff --git a/Hazel.UnitTests/UPnPTests.cs b/Hazel.UnitTests/UPnPTests.cs index 460740d..dfe80fe 100644 --- a/Hazel.UnitTests/UPnPTests.cs +++ b/Hazel.UnitTests/UPnPTests.cs @@ -30,12 +30,12 @@ namespace Hazel.UnitTests { public static readonly ILogger Instance = new Logger(); - public void LogError(string msg) + public void WriteError(string msg) { Console.WriteLine(msg); } - public void LogInfo(string msg) + public void WriteInfo(string msg) { Console.WriteLine(msg); } diff --git a/Hazel/Connection.cs b/Hazel/Connection.cs index f06a15f..b5bfe33 100644 --- a/Hazel/Connection.cs +++ b/Hazel/Connection.cs @@ -164,7 +164,6 @@ namespace Hazel /// The number of milliseconds to wait before giving up on the connect attempt. public abstract void Connect(byte[] bytes = null, int timeout = 5000); - /// /// Connects the connection to a server and begins listening. /// This method does not block. @@ -185,11 +184,15 @@ namespace Hazel /// protected void InvokeDataReceived(MessageReader msg, SendOption sendOption) { - //Make a copy to avoid race condition between null check and invocation + // Make a copy to avoid race condition between null check and invocation Action handler = DataReceived; if (handler != null) { - handler(new DataReceivedEventArgs(this, msg, sendOption)); + try + { + handler(new DataReceivedEventArgs(this, msg, sendOption)); + } + catch { } } else { @@ -209,12 +212,18 @@ namespace Hazel /// protected void InvokeDisconnected(string e, MessageReader reader) { - //Make a copy to avoid race condition between null check and invocation + // Make a copy to avoid race condition between null check and invocation EventHandler handler = Disconnected; if (handler != null) { DisconnectedEventArgs args = new DisconnectedEventArgs(e, reader); - handler.Invoke(this, args); + try + { + handler(this, args); + } + catch + { + } } } diff --git a/Hazel/FewerThreads/HazelThreadPool.cs b/Hazel/FewerThreads/HazelThreadPool.cs new file mode 100644 index 0000000..fb36b00 --- /dev/null +++ b/Hazel/FewerThreads/HazelThreadPool.cs @@ -0,0 +1,44 @@ +using System; +using System.Collections.Generic; +using System.Linq; +using System.Text; +using System.Threading; +using System.Threading.Tasks; + +namespace Hazel +{ + internal class HazelThreadPool + { + private Thread[] threads; + + public HazelThreadPool(int numThreads, ThreadStart action) + { + this.threads = new Thread[numThreads]; + for (int i = 0; i < this.threads.Length; ++i) + { + this.threads[i] = new Thread(action); + } + } + + public void Start() + { + for (int i = 0; i < this.threads.Length; ++i) + { + this.threads[i].Start(); + } + } + + public void Join() + { + for (int i = 0; i < this.threads.Length; ++i) + { + var thread = this.threads[i]; + try + { + thread.Join(); + } + catch { } + } + } + } +} \ No newline at end of file diff --git a/Hazel/FewerThreads/UdpConnectionListener2.cs b/Hazel/FewerThreads/UdpConnectionListener2.cs new file mode 100644 index 0000000..52d65da --- /dev/null +++ b/Hazel/FewerThreads/UdpConnectionListener2.cs @@ -0,0 +1,320 @@ +using System; +using System.Collections.Concurrent; +using System.Collections.Generic; +using System.Net; +using System.Net.Sockets; +using System.Threading; + +namespace Hazel.Udp.FewerThreads +{ + /// + /// Listens for new UDP connections and creates UdpConnections for them. + /// + /// + 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 NewConnection; + + /// + /// A callback for early connection rejection. + /// * Return false to reject connection. + /// * A null response is ok, we just won't send anything. + /// + 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 allConnections = new ConcurrentDictionary(); + + private Queue receiveQueue = new Queue(); + private Queue sendQueue = new Queue(); + + 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); + } + } + + /// + /// Removes a virtual connection from the list. + /// + /// The endpoint of the virtual connection. + 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); + } + } +} diff --git a/Hazel/FewerThreads/UdpServerConnection2.cs b/Hazel/FewerThreads/UdpServerConnection2.cs new file mode 100644 index 0000000..fd5ca4f --- /dev/null +++ b/Hazel/FewerThreads/UdpServerConnection2.cs @@ -0,0 +1,101 @@ +using System; +using System.Net; + +namespace Hazel.Udp.FewerThreads +{ + /// + /// Represents a servers's connection to a client that uses the UDP protocol. + /// + /// + internal sealed class UdpServerConnection2 : UdpConnection + { + /// + /// The connection listener that we use the socket of. + /// + /// + /// 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. + /// + public UdpConnectionListener2 Listener { get; private set; } + + /// + /// Creates a UdpConnection for the virtual connection to the endpoint. + /// + /// The listener that created this connection. + /// The endpoint that we are connected to. + /// The IPMode we are connected using. + 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(); + } + + /// + 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); + } + + /// + /// + /// This will always throw a HazelException. + /// + public override void Connect(byte[] bytes = null, int timeout = 5000) + { + throw new InvalidOperationException("Cannot manually connect a UdpServerConnection, did you mean to use UdpClientConnection?"); + } + + /// + /// + /// This will always throw a HazelException. + /// + public override void ConnectAsync(byte[] bytes = null, int timeout = 5000) + { + throw new InvalidOperationException("Cannot manually connect a UdpServerConnection, did you mean to use UdpClientConnection?"); + } + + /// + /// Sends a disconnect message to the end point. + /// + 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); + } + } +} diff --git a/Hazel/Hazel.csproj b/Hazel/Hazel.csproj index 7e81a65..e249199 100644 --- a/Hazel/Hazel.csproj +++ b/Hazel/Hazel.csproj @@ -72,6 +72,9 @@ + + + diff --git a/Hazel/UPnP/ILogger.cs b/Hazel/UPnP/ILogger.cs index 3a217e1..0f89e9c 100644 --- a/Hazel/UPnP/ILogger.cs +++ b/Hazel/UPnP/ILogger.cs @@ -4,11 +4,24 @@ using System.Linq; using System.Text; using System.Threading.Tasks; -namespace Hazel.UPnP +namespace Hazel { public interface ILogger { - void LogInfo(string msg); - void LogError(string msg); + void WriteError(string msg); + void WriteInfo(string msg); + } + + public class NullLogger : ILogger + { + public static readonly NullLogger Instance = new NullLogger(); + + public void WriteError(string msg) + { + } + + public void WriteInfo(string msg) + { + } } } diff --git a/Hazel/UPnP/UPnPHelper.cs b/Hazel/UPnP/UPnPHelper.cs index 506ac70..771709e 100644 --- a/Hazel/UPnP/UPnPHelper.cs +++ b/Hazel/UPnP/UPnPHelper.cs @@ -77,7 +77,7 @@ namespace Hazel.UPnP } catch(Exception e) { - this.logger.LogInfo("Exception listening for UPnP: " + e.Message); + this.logger.WriteInfo("Exception listening for UPnP: " + e.Message); } } @@ -133,7 +133,7 @@ namespace Hazel.UPnP byte[] buffer = System.Text.Encoding.UTF8.GetBytes(str); - this.logger.LogInfo("Attempting UPnP discovery"); + this.logger.WriteInfo("Attempting UPnP discovery"); socket.SendTo(buffer, new IPEndPoint(NetUtility.GetBroadcastAddress(), 1900)); } @@ -166,14 +166,14 @@ namespace Hazel.UPnP } serviceUrl = CombineUrls(resp, node.Value); - this.logger.LogInfo("UPnP service ready"); + this.logger.WriteInfo("UPnP service ready"); Status = UPnPStatus.Available; discoveryComplete.Set(); return true; } catch (Exception e) { - this.logger.LogError("Exception while parsing UPnP Service URL: " + e.Message); + this.logger.WriteError("Exception while parsing UPnP Service URL: " + e.Message); return false; } } @@ -252,12 +252,12 @@ namespace Hazel.UPnP "", "AddPortMapping"); - this.logger.LogInfo("Sent UPnP port forward request."); + this.logger.WriteInfo("Sent UPnP port forward request."); return true; } catch (Exception ex) { - this.logger.LogError("UPnP port forward failed: " + ex.Message); + this.logger.WriteError("UPnP port forward failed: " + ex.Message); return false; } } -- 2.39.5