From: Matthew Endsley Date: Sun, 24 Jan 2021 17:30:35 +0000 (-0800) Subject: Use abstract connection identifier X-Git-Tag: 1.0.0~20^2~21 X-Git-Url: https://git.deb.at/?a=commitdiff_plain;h=133af9bcca7dd07ba7ca5d941fae197129a1ae19;p=rhonda%2Fimpostor.hazel.git Use abstract connection identifier This change lays the groundwork for the ThreadLimitedUdpConnectionListener to distinguish different logical sessions from the same remote address. --- diff --git a/Hazel/FewerThreads/ThreadLimitedUdpConnectionListener.cs b/Hazel/FewerThreads/ThreadLimitedUdpConnectionListener.cs index 4ebff2e..dcb4268 100644 --- a/Hazel/FewerThreads/ThreadLimitedUdpConnectionListener.cs +++ b/Hazel/FewerThreads/ThreadLimitedUdpConnectionListener.cs @@ -23,6 +23,7 @@ namespace Hazel.Udp.FewerThreads { public MessageReader Message; public EndPoint Sender; + public ConnectionId ConnectionId; } private const int SendReceiveBufferSize = 1024 * 1024; @@ -49,7 +50,31 @@ namespace Hazel.Udp.FewerThreads private Thread sendThread; private HazelThreadPool processThreads; - private ConcurrentDictionary allConnections = new ConcurrentDictionary(); + public struct ConnectionId + { + public ulong Id; + + public static ConnectionId Create(ulong id) + { + ConnectionId result = new ConnectionId(); + result.Id = id; + return result; + } + + public static ConnectionId CreateFromEndPoint(IPEndPoint endPoint) + { + if (endPoint.AddressFamily != AddressFamily.InterNetwork) + { + throw new ArgumentException("ConnectionId only supports IPv4"); + } + + ulong port = (ulong)endPoint.Port; + ulong address = (ulong)endPoint.Address.Address; + return Create((address << 32) | port); + } + } + + private ConcurrentDictionary allConnections = new ConcurrentDictionary(); private BlockingCollection receiveQueue; private BlockingCollection sendQueue = new BlockingCollection(); @@ -171,7 +196,8 @@ namespace Hazel.Udp.FewerThreads return; } - this.ProcessIncomingMessageFromOtherThread(message, remoteEP); + ConnectionId connectionId = ConnectionId.CreateFromEndPoint((IPEndPoint)remoteEP); + this.ProcessIncomingMessageFromOtherThread(message, remoteEP, connectionId); } } } @@ -183,7 +209,7 @@ namespace Hazel.Udp.FewerThreads try { - this.ReadCallback(msg.Message, msg.Sender); + this.ReadCallback(msg.Message, msg.Sender, msg.ConnectionId); } catch { @@ -191,9 +217,9 @@ namespace Hazel.Udp.FewerThreads } } } - protected virtual void ProcessIncomingMessageFromOtherThread(MessageReader message, EndPoint peerAddress) + protected virtual void ProcessIncomingMessageFromOtherThread(MessageReader message, EndPoint remoteEndPoint, ConnectionId connectionId) { - this.receiveQueue.Add(new ReceiveMessageInfo() { Message = message, Sender = peerAddress }); + this.receiveQueue.Add(new ReceiveMessageInfo() { Message = message, Sender = remoteEndPoint, ConnectionId = connectionId }); } private void SendLoop() @@ -217,7 +243,7 @@ namespace Hazel.Udp.FewerThreads } } - void ReadCallback(MessageReader message, EndPoint remoteEndPoint) + void ReadCallback(MessageReader message, EndPoint remoteEndPoint, ConnectionId connectionId) { int bytesReceived = message.Length; bool aware = true; @@ -226,11 +252,11 @@ namespace Hazel.Udp.FewerThreads // 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)) + if (!this.allConnections.TryGetValue(connectionId.Id, out connection)) { lock (this.allConnections) { - if (!this.allConnections.TryGetValue(remoteEndPoint, out connection)) + if (!this.allConnections.TryGetValue(connectionId.Id, out connection)) { // Check for malformed connection attempts if (!isHello) @@ -254,8 +280,8 @@ namespace Hazel.Udp.FewerThreads } aware = false; - connection = new ThreadLimitedUdpServerConnection(this, (IPEndPoint)remoteEndPoint, this.IPMode); - if (!this.allConnections.TryAdd(remoteEndPoint, connection)) + connection = new ThreadLimitedUdpServerConnection(this, connectionId, (IPEndPoint)remoteEndPoint, this.IPMode); + if (!this.allConnections.TryAdd(connectionId.Id, connection)) { throw new HazelException("Failed to add a connection. This should never happen."); } @@ -297,10 +323,10 @@ namespace Hazel.Udp.FewerThreads /// /// Removes a virtual connection from the list. /// - /// The endpoint of the virtual connection. - internal bool RemoveConnectionTo(EndPoint endPoint) + /// Connection key of the virtual connection. + internal bool RemoveConnectionTo(ConnectionId connectionId) { - return this.allConnections.TryRemove(endPoint, out var conn); + return this.allConnections.TryRemove(connectionId.Id, out var conn); } protected virtual void Dispose(bool disposing) diff --git a/Hazel/FewerThreads/ThreadLimitedUdpServerConnection.cs b/Hazel/FewerThreads/ThreadLimitedUdpServerConnection.cs index 84acf5f..31c9ca2 100644 --- a/Hazel/FewerThreads/ThreadLimitedUdpServerConnection.cs +++ b/Hazel/FewerThreads/ThreadLimitedUdpServerConnection.cs @@ -1,4 +1,4 @@ -using System; +using System; using System.Net; namespace Hazel.Udp.FewerThreads @@ -20,16 +20,19 @@ namespace Hazel.Udp.FewerThreads /// public ThreadLimitedUdpConnectionListener Listener { get; private set; } + private ThreadLimitedUdpConnectionListener.ConnectionId ConnectionId; + /// /// 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 ThreadLimitedUdpServerConnection(ThreadLimitedUdpConnectionListener listener, IPEndPoint endPoint, IPMode IPMode) + internal ThreadLimitedUdpServerConnection(ThreadLimitedUdpConnectionListener listener, ThreadLimitedUdpConnectionListener.ConnectionId connectionId, IPEndPoint endPoint, IPMode IPMode) : base() { this.Listener = listener; + this.ConnectionId = connectionId; this.RemoteEndPoint = endPoint; this.EndPoint = endPoint; this.IPMode = IPMode; @@ -69,7 +72,7 @@ namespace Hazel.Udp.FewerThreads /// protected override bool SendDisconnect(MessageWriter data = null) { - if (!Listener.RemoveConnectionTo(RemoteEndPoint)) return false; + if (!Listener.RemoveConnectionTo(this.ConnectionId)) return false; this._state = ConnectionState.NotConnected; var bytes = EmptyDisconnectBytes;