diff --git a/RiptideNetworking/RiptideNetworking/Client.cs b/RiptideNetworking/RiptideNetworking/Client.cs index a78630ff..964aba98 100644 --- a/RiptideNetworking/RiptideNetworking/Client.cs +++ b/RiptideNetworking/RiptideNetworking/Client.cs @@ -96,6 +96,12 @@ public void ChangeTransport(IClient newTransport) transport = newTransport; } + /// + /// Sets whether or not the server should throw an exception when handling messages. If set to , the server will not throw exceptions when handling messages (like normal). Othervise it will throw an exception. + /// + /// New value for the PREVENT_EXCEPTION property. + public void ChangeExceptionPrevention(bool preventExceptions) => Server.SetPreventException(preventExceptions); + /// Attempts to connect to a server at the given host address. /// The host address to connect to. /// How many connection attempts to make before giving up. @@ -107,10 +113,12 @@ public void ChangeTransport(IClient newTransport) /// Setting to will disable the automatic detection and execution of methods with the , which is beneficial if you prefer to handle messages via the event. /// /// if a connection attempt will be made. if an issue occurred (such as being in an invalid format) and a connection attempt will not be made. - public bool Connect(string hostAddress, int maxConnectionAttempts = 5, byte messageHandlerGroupId = 0, Message message = null, bool useMessageHandlers = true) + public bool Connect(string hostAddress, int maxConnectionAttempts = 5, byte messageHandlerGroupId = 0, Message message = null, bool useMessageHandlers = true, bool preventExceptions = true) { Disconnect(); + Server.SetPreventException(preventExceptions); + SubToTransportEvents(); if (!transport.Connect(hostAddress, out connection, out string connectError)) diff --git a/RiptideNetworking/RiptideNetworking/Connection.cs b/RiptideNetworking/RiptideNetworking/Connection.cs index 4184876a..21957923 100644 --- a/RiptideNetworking/RiptideNetworking/Connection.cs +++ b/RiptideNetworking/RiptideNetworking/Connection.cs @@ -6,7 +6,7 @@ using Riptide.Transports; using Riptide.Utils; using System; -using System.Collections.Generic; +using System.Collections.Concurrent; namespace Riptide { @@ -104,7 +104,7 @@ public bool CanTimeout /// The sequencer for reliable messages. private readonly ReliableSequencer reliable; /// The currently pending reliably sent messages whose delivery has not been acknowledged yet. Stored by sequence ID. - private readonly Dictionary pendingMessages; + private readonly ConcurrentDictionary pendingMessages; /// The connection's current state. private ConnectionState state; /// The number of consecutive times that the threshold was exceeded. @@ -136,7 +136,7 @@ protected Connection() MaxSendAttempts = 15; MaxNotifyLoss = 0.05f; // 5% NotifyLossResilience = 64; - pendingMessages = new Dictionary(); + pendingMessages = new ConcurrentDictionary(); } /// Initializes connection data. @@ -184,7 +184,7 @@ public ushort Send(Message message, bool shouldRelease = true) { sequenceId = reliable.NextSequenceId; PendingMessage pendingMessage = PendingMessage.Create(sequenceId, message, this); - pendingMessages.Add(sequenceId, pendingMessage); + pendingMessages.TryAdd(sequenceId, pendingMessage); pendingMessage.TrySend(); Metrics.ReliableUniques++; } @@ -249,11 +249,10 @@ private void ResendMessage(ushort sequenceId) /// The sequence ID that was acknowledged. internal void ClearMessage(ushort sequenceId) { - if (pendingMessages.TryGetValue(sequenceId, out PendingMessage pendingMessage)) + if (pendingMessages.TryRemove(sequenceId, out PendingMessage pendingMessage)) { ReliableDelivered?.Invoke(sequenceId); pendingMessage.Clear(); - pendingMessages.Remove(sequenceId); UpdateSendAttemptsViolations(); } } diff --git a/RiptideNetworking/RiptideNetworking/Exceptions.cs b/RiptideNetworking/RiptideNetworking/Exceptions.cs index 2d010071..62428cd4 100644 --- a/RiptideNetworking/RiptideNetworking/Exceptions.cs +++ b/RiptideNetworking/RiptideNetworking/Exceptions.cs @@ -117,6 +117,65 @@ private static string GetErrorMessage(Type declaringType, string handlerMethodNa } } + /// The exception that is thrown when a message does not contain enough bits to retrieve a value of the specified type. + public class ReadingOrderOrCapacityException : Exception + { + /// + /// The number of unread bits remaining in the message when the exception was thrown. This is not necessarily the number of bits required to retrieve a value of the specified type, as the exception can also be thrown if the reading order is incorrect (e.g. trying to read a byte before reading a preceding bool). + /// + public int UnreadBits; + /// + /// The number of bits required to retrieve a value of the specified type. This is not necessarily the number of unread bits remaining in the message when the exception was thrown, as the exception can also be thrown if the reading order is incorrect (e.g. trying to read a byte before reading a preceding bool). + /// + public int RequiredBits; + /// + /// The name of the value that could not be retrieved. + /// + public readonly string ValueName; + + /// Initializes a new instance. + public ReadingOrderOrCapacityException() { } + /// Initializes a new instance with a specified error message. + /// The error message that explains the reason for the exception. + public ReadingOrderOrCapacityException(string message) : base(message) { } + /// Initializes a new instance with a specified error message and a reference to the inner exception that is the cause of this exception. + /// The error message that explains the reason for the exception. + /// The exception that is the cause of the current exception. If is not a null reference, the current exception is raised in a catch block that handles the inner exception. + public ReadingOrderOrCapacityException(string message, Exception inner) : base(message, inner) { } + /// Initializes a new instance and constructs an error message from the given information. + /// The number of unread bits remaining in the message when the exception was thrown. + /// The name of the type containing the handler method. + public ReadingOrderOrCapacityException(int unreadbits, string declaringType) : base(GetErrorMessage(unreadbits, declaringType)) + { + UnreadBits = unreadbits; + ValueName = declaringType; + } + + /// Initializes a new instance and constructs an error message from the given information. + /// The number of unread bits remaining in the message when the exception was thrown. + /// The name of the type containing the handler method. + public ReadingOrderOrCapacityException(int unreadbits, int requiredBits, string declaringType) : base(GetErrorMessage(unreadbits, requiredBits, declaringType)) + { + UnreadBits = unreadbits; + RequiredBits = requiredBits; + ValueName = declaringType; + } + + + /// Constructs the error message from the given information. + /// The error message. + private static string GetErrorMessage(int unreadBits, string valueName) + { + return $"Message only contains {unreadBits} unread {Helper.CorrectForm(unreadBits, "bit")}, which is not enough to retrieve a value of type '{valueName}'!"; + } + /// Constructs the error message from the given information. + /// The error message. + private static string GetErrorMessage(int unreadBits, int requiredBits, string valueName) + { + return $"Message only contains {unreadBits} unread {Helper.CorrectForm(unreadBits, "bit")}, which is not enough to retrieve a value of type '{valueName}' (requires {requiredBits} bits)!"; + } + } + /// The exception that is thrown when a method with a does not have an acceptable message handler method signature (either or ). public class InvalidHandlerSignatureException : Exception { diff --git a/RiptideNetworking/RiptideNetworking/Message.cs b/RiptideNetworking/RiptideNetworking/Message.cs index 13de2b11..f84df01e 100644 --- a/RiptideNetworking/RiptideNetworking/Message.cs +++ b/RiptideNetworking/RiptideNetworking/Message.cs @@ -6,7 +6,7 @@ using Riptide.Transports; using Riptide.Utils; using System; -using System.Collections.Generic; +using System.Collections.Concurrent; using System.Runtime.CompilerServices; using System.Text; @@ -86,7 +86,7 @@ public static int MaxPayloadSize /// Changes will not affect and instances which are already running until they are restarted. public static byte InstancesPerPeer { get; set; } = 4; /// A pool of reusable message instances. - private static readonly List pool = new List(InstancesPerPeer * 2); + private static readonly ConcurrentDictionary pool = new ConcurrentDictionary(); static Message() { @@ -124,8 +124,29 @@ static Message() /// The next bit to be written. private int writeBit; + /// Determines which id is this Message. + public uint MessageId { get; private set; } = 0; + /// Determines which is the next value fot give to the next mmessage [LIFO]. + private static readonly ConcurrentStack indexedKeys = new ConcurrentStack(); + /// For give unique ID for each message global thread-safe counter. + private static int globalMessageIdCounter = 0; + /// Initializes a reusable instance. - private Message() => data = new ulong[maxArraySize]; + private Message() + { + data = new ulong[maxArraySize]; + + // UNBREAKABLE THRESHOLD: When the constructor is called, any object not in the pool/in use. Then the atomic loop that continuesly run for tries to find and create UNIQUE ID + uint gId; + do + { + // The `interlocked.Increment` int automatically goes negative when it reaches the limit. When it goes negative, the `uint cast` ensures that the transaction volume is at the maximum value the `uint` value can hold, which is 4.2 billion. + gId = (uint)System.Threading.Interlocked.Increment(ref globalMessageIdCounter); + } + while (gId == 0 || pool.ContainsKey(gId)); + + MessageId = gId; + } /// Gets a completely empty message instance with no header. /// An empty message instance. @@ -175,16 +196,15 @@ public static void TrimPool() { // No Servers or Clients are running, empty the list and reset the capacity pool.Clear(); - pool.Capacity = InstancesPerPeer * 2; // x2 so there's some buffer room for extra Message instances in the event that more are needed + indexedKeys.Clear(); } else { // Reset the pool capacity and number of Message instances in the pool to what is appropriate for how many Servers & Clients are active int idealInstanceAmount = Peer.ActiveCount * InstancesPerPeer; - if (pool.Count > idealInstanceAmount) + while (pool.Count > idealInstanceAmount && indexedKeys.TryPop(out uint key)) { - pool.RemoveRange(Peer.ActiveCount * InstancesPerPeer, pool.Count - idealInstanceAmount); - pool.Capacity = idealInstanceAmount * 2; + pool.TryRemove(key, out Message _); } } } @@ -193,26 +213,19 @@ public static void TrimPool() /// A message instance ready to be used for sending or handling. private static Message RetrieveFromPool() { - Message message; - if (pool.Count > 0) - { - message = pool[0]; - pool.RemoveAt(0); - } - else - message = new Message(); - - return message; + return indexedKeys.TryPop(out uint key) && pool.TryGetValue(key, out Message message) ? message : new Message(); } /// Returns the message instance to the internal pool so it can be reused. public void Release() { - if (pool.Count < pool.Capacity) + int idealCap = Peer.ActiveCount < 1 ? InstancesPerPeer * 2 : Peer.ActiveCount * InstancesPerPeer; + if (pool.Count < idealCap) { - // Pool exists and there's room - if (!pool.Contains(this)) - pool.Add(this); // Only add it if it's not already in the list, otherwise this method being called twice in a row for whatever reason could cause *serious* issues + if (pool.ContainsKey(MessageId) == false && pool.TryAdd(MessageId, this)) // Only add it if it's not already in the list, otherwise this method being called twice in a row for whatever reason could cause *serious* issues + { + indexedKeys.Push(MessageId); + } } } #endregion @@ -608,8 +621,12 @@ public byte GetByte() { if (UnreadBits < BitsPerByte) { - RiptideLogger.Log(LogType.Error, NotEnoughBitsError(ByteName, $"{default(byte)}")); - return default; + if (Server.PREVENT_EXCEPTION) + { + RiptideLogger.Log(LogType.Error, NotEnoughBitsError(ByteName, $"{default(byte)}")); + return default; + } + throw new ReadingOrderOrCapacityException(UnreadBits, ByteName); } byte value = Converter.ByteFromBits(data, readBit); @@ -623,8 +640,12 @@ public sbyte GetSByte() { if (UnreadBits < BitsPerByte) { - RiptideLogger.Log(LogType.Error, NotEnoughBitsError(SByteName, $"{default(sbyte)}")); - return default; + if (Server.PREVENT_EXCEPTION) + { + RiptideLogger.Log(LogType.Error, NotEnoughBitsError(SByteName, $"{default(sbyte)}")); + return default; + } + throw new ReadingOrderOrCapacityException(UnwrittenBits, SByteName); } sbyte value = Converter.SByteFromBits(data, readBit); @@ -797,8 +818,12 @@ private void ReadBytes(int amount, byte[] intoArray, int startIndex = 0) { if (UnreadBits < amount * BitsPerByte) { - RiptideLogger.Log(LogType.Error, NotEnoughBitsError(amount, ByteName)); - amount = UnreadBits / BitsPerByte; + if (Server.PREVENT_EXCEPTION) + { + RiptideLogger.Log(LogType.Error, NotEnoughBitsError(amount, ByteName)); + amount = UnreadBits / BitsPerByte; + } + else throw new ReadingOrderOrCapacityException(UnreadBits, amount * BitsPerByte, ByteName); } if (readBit % BitsPerByte == 0) @@ -824,8 +849,12 @@ private void ReadSBytes(int amount, sbyte[] intoArray, int startIndex = 0) { if (UnreadBits < amount * BitsPerByte) { - RiptideLogger.Log(LogType.Error, NotEnoughBitsError(amount, SByteName)); - amount = UnreadBits / BitsPerByte; + if (Server.PREVENT_EXCEPTION) + { + RiptideLogger.Log(LogType.Error, NotEnoughBitsError(amount, SByteName)); + amount = UnreadBits / BitsPerByte; + } + else throw new ReadingOrderOrCapacityException(UnreadBits, amount * BitsPerByte, SByteName); } for (int i = 0; i < amount; i++) @@ -855,8 +884,12 @@ public bool GetBool() { if (UnreadBits < 1) { - RiptideLogger.Log(LogType.Error, NotEnoughBitsError(BoolName, $"{default(bool)}")); - return default; + if (Server.PREVENT_EXCEPTION) + { + RiptideLogger.Log(LogType.Error, NotEnoughBitsError(BoolName, $"{default(bool)}")); + return default; + } + throw new ReadingOrderOrCapacityException(UnreadBits, BoolName); } return Converter.BoolFromBit(data, readBit++); @@ -916,8 +949,12 @@ private void ReadBools(int amount, bool[] intoArray, int startIndex = 0) { if (UnreadBits < amount) { - RiptideLogger.Log(LogType.Error, NotEnoughBitsError(amount, BoolName)); - amount = UnreadBits; + if (Server.PREVENT_EXCEPTION) + { + RiptideLogger.Log(LogType.Error, NotEnoughBitsError(amount, BoolName)); + amount = UnreadBits; + } + else throw new ReadingOrderOrCapacityException(UnreadBits, amount, BoolName); } for (int i = 0; i < amount; i++) @@ -958,8 +995,12 @@ public short GetShort() { if (UnreadBits < sizeof(short) * BitsPerByte) { - RiptideLogger.Log(LogType.Error, NotEnoughBitsError(ShortName, $"{default(short)}")); - return default; + if (Server.PREVENT_EXCEPTION) + { + RiptideLogger.Log(LogType.Error, NotEnoughBitsError(ShortName, $"{default(short)}")); + return default; + } + throw new ReadingOrderOrCapacityException(UnreadBits, ShortName); } short value = Converter.ShortFromBits(data, readBit); @@ -973,8 +1014,12 @@ public ushort GetUShort() { if (UnreadBits < sizeof(ushort) * BitsPerByte) { - RiptideLogger.Log(LogType.Error, NotEnoughBitsError(UShortName, $"{default(ushort)}")); - return default; + if (Server.PREVENT_EXCEPTION) + { + RiptideLogger.Log(LogType.Error, NotEnoughBitsError(UShortName, $"{default(ushort)}")); + return default; + } + throw new ReadingOrderOrCapacityException(UnreadBits, UShortName); } ushort value = Converter.UShortFromBits(data, readBit); @@ -1088,8 +1133,12 @@ private void ReadShorts(int amount, short[] intoArray, int startIndex = 0) { if (UnreadBits < amount * sizeof(short) * BitsPerByte) { - RiptideLogger.Log(LogType.Error, NotEnoughBitsError(amount, ShortName)); - amount = UnreadBits / (sizeof(short) * BitsPerByte); + if (Server.PREVENT_EXCEPTION) + { + RiptideLogger.Log(LogType.Error, NotEnoughBitsError(amount, ShortName)); + amount = UnreadBits / (sizeof(short) * BitsPerByte); + } + else throw new ReadingOrderOrCapacityException(UnreadBits, amount * sizeof(short) * BitsPerByte, ShortName); } for (int i = 0; i < amount; i++) @@ -1107,8 +1156,12 @@ private void ReadUShorts(int amount, ushort[] intoArray, int startIndex = 0) { if (UnreadBits < amount * sizeof(ushort) * BitsPerByte) { - RiptideLogger.Log(LogType.Error, NotEnoughBitsError(amount, UShortName)); - amount = UnreadBits / (sizeof(ushort) * BitsPerByte); + if (Server.PREVENT_EXCEPTION) + { + RiptideLogger.Log(LogType.Error, NotEnoughBitsError(amount, UShortName)); + amount = UnreadBits / (sizeof(ushort) * BitsPerByte); + } + else throw new ReadingOrderOrCapacityException(UnreadBits, amount * sizeof(ushort) * BitsPerByte, UShortName); } for (int i = 0; i < amount; i++) @@ -1152,8 +1205,12 @@ public int GetInt() { if (UnreadBits < sizeof(int) * BitsPerByte) { - RiptideLogger.Log(LogType.Error, NotEnoughBitsError(IntName, $"{default(int)}")); - return default; + if (Server.PREVENT_EXCEPTION) + { + RiptideLogger.Log(LogType.Error, NotEnoughBitsError(IntName, $"{default(int)}")); + return default; + } + throw new ReadingOrderOrCapacityException(UnreadBits, IntName); } int value = Converter.IntFromBits(data, readBit); @@ -1167,8 +1224,12 @@ public uint GetUInt() { if (UnreadBits < sizeof(uint) * BitsPerByte) { - RiptideLogger.Log(LogType.Error, NotEnoughBitsError(UIntName, $"{default(uint)}")); - return default; + if (Server.PREVENT_EXCEPTION) + { + RiptideLogger.Log(LogType.Error, NotEnoughBitsError(UIntName, $"{default(uint)}")); + return default; + } + throw new ReadingOrderOrCapacityException(UnreadBits, UIntName); } uint value = Converter.UIntFromBits(data, readBit); @@ -1282,8 +1343,12 @@ private void ReadInts(int amount, int[] intoArray, int startIndex = 0) { if (UnreadBits < amount * sizeof(int) * BitsPerByte) { - RiptideLogger.Log(LogType.Error, NotEnoughBitsError(amount, IntName)); - amount = UnreadBits / (sizeof(int) * BitsPerByte); + if (Server.PREVENT_EXCEPTION) + { + RiptideLogger.Log(LogType.Error, NotEnoughBitsError(amount, IntName)); + amount = UnreadBits / (sizeof(int) * BitsPerByte); + } + else throw new ReadingOrderOrCapacityException(UnreadBits, amount * sizeof(int) * BitsPerByte, IntName); } for (int i = 0; i < amount; i++) @@ -1301,8 +1366,12 @@ private void ReadUInts(int amount, uint[] intoArray, int startIndex = 0) { if (UnreadBits < amount * sizeof(uint) * BitsPerByte) { - RiptideLogger.Log(LogType.Error, NotEnoughBitsError(amount, UIntName)); - amount = UnreadBits / (sizeof(uint) * BitsPerByte); + if (Server.PREVENT_EXCEPTION) + { + RiptideLogger.Log(LogType.Error, NotEnoughBitsError(amount, UIntName)); + amount = UnreadBits / (sizeof(uint) * BitsPerByte); + } + else throw new ReadingOrderOrCapacityException(UnreadBits, amount * sizeof(uint) * BitsPerByte, UIntName); } for (int i = 0; i < amount; i++) @@ -1346,8 +1415,12 @@ public long GetLong() { if (UnreadBits < sizeof(long) * BitsPerByte) { - RiptideLogger.Log(LogType.Error, NotEnoughBitsError(LongName, $"{default(long)}")); - return default; + if (Server.PREVENT_EXCEPTION) + { + RiptideLogger.Log(LogType.Error, NotEnoughBitsError(LongName, $"{default(long)}")); + return default; + } + throw new ReadingOrderOrCapacityException(UnreadBits, LongName); } long value = Converter.LongFromBits(data, readBit); @@ -1361,8 +1434,12 @@ public ulong GetULong() { if (UnreadBits < sizeof(ulong) * BitsPerByte) { - RiptideLogger.Log(LogType.Error, NotEnoughBitsError(ULongName, $"{default(ulong)}")); - return default; + if (Server.PREVENT_EXCEPTION) + { + RiptideLogger.Log(LogType.Error, NotEnoughBitsError(ULongName, $"{default(ulong)}")); + return default; + } + throw new ReadingOrderOrCapacityException(UnreadBits, ULongName); } ulong value = Converter.ULongFromBits(data, readBit); @@ -1476,8 +1553,12 @@ private void ReadLongs(int amount, long[] intoArray, int startIndex = 0) { if (UnreadBits < amount * sizeof(long) * BitsPerByte) { - RiptideLogger.Log(LogType.Error, NotEnoughBitsError(amount, LongName)); - amount = UnreadBits / (sizeof(long) * BitsPerByte); + if (Server.PREVENT_EXCEPTION) + { + RiptideLogger.Log(LogType.Error, NotEnoughBitsError(amount, LongName)); + amount = UnreadBits / (sizeof(long) * BitsPerByte); + } + else throw new ReadingOrderOrCapacityException(UnreadBits, amount * sizeof(long) * BitsPerByte, LongName); } for (int i = 0; i < amount; i++) @@ -1495,8 +1576,12 @@ private void ReadULongs(int amount, ulong[] intoArray, int startIndex = 0) { if (UnreadBits < amount * sizeof(ulong) * BitsPerByte) { - RiptideLogger.Log(LogType.Error, NotEnoughBitsError(amount, ULongName)); - amount = UnreadBits / (sizeof(ulong) * BitsPerByte); + if (Server.PREVENT_EXCEPTION) + { + RiptideLogger.Log(LogType.Error, NotEnoughBitsError(amount, ULongName)); + amount = UnreadBits / (sizeof(ulong) * BitsPerByte); + } + else throw new ReadingOrderOrCapacityException(UnreadBits, amount * sizeof(ulong) * BitsPerByte, ULongName); } for (int i = 0; i < amount; i++) @@ -1527,8 +1612,12 @@ public float GetFloat() { if (UnreadBits < sizeof(float) * BitsPerByte) { - RiptideLogger.Log(LogType.Error, NotEnoughBitsError(FloatName, $"{default(float)}")); - return default; + if (Server.PREVENT_EXCEPTION) + { + RiptideLogger.Log(LogType.Error, NotEnoughBitsError(FloatName, $"{default(float)}")); + return default; + } + throw new ReadingOrderOrCapacityException(UnreadBits, FloatName); } float value = Converter.FloatFromBits(data, readBit); @@ -1593,8 +1682,12 @@ private void ReadFloats(int amount, float[] intoArray, int startIndex = 0) { if (UnreadBits < amount * sizeof(float) * BitsPerByte) { - RiptideLogger.Log(LogType.Error, NotEnoughBitsError(amount, FloatName)); - amount = UnreadBits / (sizeof(float) * BitsPerByte); + if (Server.PREVENT_EXCEPTION) + { + RiptideLogger.Log(LogType.Error, NotEnoughBitsError(amount, FloatName)); + amount = UnreadBits / (sizeof(float) * BitsPerByte); + } + else throw new ReadingOrderOrCapacityException(UnreadBits, amount * sizeof(float) * BitsPerByte, FloatName); } for (int i = 0; i < amount; i++) @@ -1625,8 +1718,12 @@ public double GetDouble() { if (UnreadBits < sizeof(double) * BitsPerByte) { - RiptideLogger.Log(LogType.Error, NotEnoughBitsError(DoubleName, $"{default(double)}")); - return default; + if (Server.PREVENT_EXCEPTION) + { + RiptideLogger.Log(LogType.Error, NotEnoughBitsError(DoubleName, $"{default(double)}")); + return default; + } + throw new ReadingOrderOrCapacityException(UnreadBits, DoubleName); } double value = Converter.DoubleFromBits(data, readBit); @@ -1691,8 +1788,12 @@ private void ReadDoubles(int amount, double[] intoArray, int startIndex = 0) { if (UnreadBits < amount * sizeof(double) * BitsPerByte) { - RiptideLogger.Log(LogType.Error, NotEnoughBitsError(amount, DoubleName)); - amount = UnreadBits / (sizeof(double) * BitsPerByte); + if (Server.PREVENT_EXCEPTION) + { + RiptideLogger.Log(LogType.Error, NotEnoughBitsError(amount, DoubleName)); + amount = UnreadBits / (sizeof(double) * BitsPerByte); + } + else throw new ReadingOrderOrCapacityException(UnreadBits, amount * sizeof(double) * BitsPerByte, DoubleName); } for (int i = 0; i < amount; i++) @@ -1720,8 +1821,12 @@ public string GetString() int length = (int)GetVarULong(); // Get the length of the string (in bytes, NOT characters) if (UnreadBits < length * BitsPerByte) { - RiptideLogger.Log(LogType.Error, NotEnoughBitsError(StringName, "shortened string")); - length = UnreadBits / BitsPerByte; + if (Server.PREVENT_EXCEPTION) + { + RiptideLogger.Log(LogType.Error, NotEnoughBitsError(StringName, "shortened string")); + length = UnreadBits / BitsPerByte; + } + throw new ReadingOrderOrCapacityException(UnreadBits, length * BitsPerByte, StringName); } string value = Encoding.UTF8.GetString(GetBytes(length), 0, length); diff --git a/RiptideNetworking/RiptideNetworking/PendingMessage.cs b/RiptideNetworking/RiptideNetworking/PendingMessage.cs index 4828250b..98c0ed35 100644 --- a/RiptideNetworking/RiptideNetworking/PendingMessage.cs +++ b/RiptideNetworking/RiptideNetworking/PendingMessage.cs @@ -6,6 +6,7 @@ using Riptide.Transports; using Riptide.Utils; using System; +using System.Collections.Concurrent; using System.Collections.Generic; namespace Riptide @@ -20,7 +21,13 @@ internal class PendingMessage private const float RetryTimeMultiplier = 1.2f; /// A pool of reusable instances. - private static readonly List pool = new List(); + private static readonly ConcurrentDictionary pool = new ConcurrentDictionary(); + /// For give uniue ID to the each pending Message global thread-safe counter. + private static int globalPendingMessageIdCounter = 0; + /// Determines which id is this PendingMessage. + internal uint PendingMessageId { get; private set; } = 0; + /// Determines which is the next value for giving to the next pending message [LIFO]. + private static readonly ConcurrentStack indexedKeys = new ConcurrentStack(); /// The to use to send (and resend) the pending message. private Connection connection; @@ -37,6 +44,16 @@ internal class PendingMessage internal PendingMessage() { data = new byte[Message.MaxSize]; + + // UNBREAKABLE THRESHOLD: When the constructor is called, any object not in the pool/in use. Then the atomic loop that continuesly run for tries to find and create UNIQUE ID + uint gId; + do + { + // The `interlocked.Increment` int automatically goes negative when it reaches the limit. When it goes negative, the `uint cast` ensures that the transaction volume is at the maximum value the `uint` value can hold, which is 4.2 billion. + gId = (uint)System.Threading.Interlocked.Increment(ref globalPendingMessageIdCounter); + } + while (gId == 0 || pool.ContainsKey(gId)); + PendingMessageId = gId; } #region Pooling @@ -63,29 +80,32 @@ internal static PendingMessage Create(ushort sequenceId, Message message, Connec /// A instance. private static PendingMessage RetrieveFromPool() { - PendingMessage message; - if (pool.Count > 0) + if (indexedKeys.TryPop(out uint key)) { - message = pool[0]; - pool.RemoveAt(0); + if (pool.TryRemove(key, out PendingMessage message)) + { + return message; + } } - else - message = new PendingMessage(); - return message; + // constructor will give an unique id as i made it in the Message.cs + return new PendingMessage(); } /// Empties the pool. Does not affect instances which are actively pending and therefore not in the pool. public static void ClearPool() { pool.Clear(); + indexedKeys.Clear(); } /// Returns the instance to the pool so it can be reused. private void Release() { - if (!pool.Contains(this)) - pool.Add(this); // Only add it if it's not already in the list, otherwise this method being called twice in a row for whatever reason could cause *serious* issues + if (pool.ContainsKey(PendingMessageId) == false && pool.TryAdd(PendingMessageId, this)) // Only add it if it's not already in the list, otherwise this method being called twice in a row for whatever reason could cause *serious* issues + { + indexedKeys.Push(PendingMessageId); + } // TODO: consider doing something to decrease pool capacity if there are far more // available instance than are needed, which could occur if a large burst of diff --git a/RiptideNetworking/RiptideNetworking/Server.cs b/RiptideNetworking/RiptideNetworking/Server.cs index 4def3218..9c4b0beb 100644 --- a/RiptideNetworking/RiptideNetworking/Server.cs +++ b/RiptideNetworking/RiptideNetworking/Server.cs @@ -7,6 +7,7 @@ using Riptide.Utils; using System; using System.Collections.Generic; +using System.Collections.Concurrent; using System.Linq; using System.Reflection; @@ -34,7 +35,7 @@ public override int TimeoutTime set { defaultTimeout = value; - foreach (Connection connection in clients.Values) + foreach (Connection connection in Clients) connection.TimeoutTime = defaultTimeout; } } @@ -62,7 +63,7 @@ public override int TimeoutTime /// Currently pending connections which are waiting to be accepted or rejected. private readonly List pendingConnections; /// Currently connected clients. - private Dictionary clients; + private ConcurrentDictionary clients; /// Clients that have timed out and need to be removed from . private readonly List timedOutClients; /// Methods used to handle messages, accessible by their corresponding message IDs. @@ -70,7 +71,15 @@ public override int TimeoutTime /// The underlying transport's server that is used for sending and receiving data. private IServer transport; /// All currently unused client IDs. - private Queue availableClientIds; + private ConcurrentQueue availableClientIds; + + /// This bool using for preventing exceptions. If set to true, the server will not throw exceptions when handling messages (like normal). Othervise it will throw an exception. + public static bool PREVENT_EXCEPTION { get; private set; } = true; + /// + /// Sets whether or not the server should throw an exception when handling messages. If set to , the server will not throw exceptions when handling messages (like normal). Othervise it will throw an exception. + /// + /// New value for the PREVENT_EXCEPTION property. + public static void SetPreventException(bool value) => PREVENT_EXCEPTION = value; /// Handles initial setup. /// The transport to use for sending and receiving data. @@ -79,7 +88,7 @@ public Server(IServer transport, string logName = "SERVER") : base(logName) { this.transport = transport; pendingConnections = new List(); - clients = new Dictionary(); + clients = new ConcurrentDictionary(); timedOutClients = new List(); } /// Handles initial setup using the built-in UDP transport. @@ -88,30 +97,39 @@ public Server(string logName = "SERVER") : this(new Transports.Udp.UdpServer(), /// Stops the server if it's running and swaps out the transport it's using. /// The new underlying transport server to use for sending and receiving data. - /// This method does not automatically restart the server. To continue accepting connections, must be called again. + /// This method does not automatically restart the server. To continue accepting connections, must be called again. public void ChangeTransport(IServer newTransport) { Stop(); transport = newTransport; } + /// + /// Sets whether or not the server should throw an exception when handling messages. If set to , the server will not throw exceptions when handling messages (like normal). Othervise it will throw an exception. + /// + /// New value for the PREVENT_EXCEPTION property. + public void ChangeExceptionPrevention(bool preventExceptions) => SetPreventException(preventExceptions); + /// Starts the server. /// The local port on which to start the server. /// The maximum number of concurrent connections to allow. /// The ID of the group of message handler methods to use when building . /// Whether or not the server should use the built-in message handler system. + /// Whether or not the server should prevent exceptions when handling messages. /// Setting to will disable the automatic detection and execution of methods with the , which is beneficial if you prefer to handle messages via the event. - public void Start(ushort port, ushort maxClientCount, byte messageHandlerGroupId = 0, bool useMessageHandlers = true) + public void Start(ushort port, ushort maxClientCount, byte messageHandlerGroupId = 0, bool useMessageHandlers = true, bool preventExceptions = true) { Stop(); + SetPreventException(preventExceptions); + IncreaseActiveCount(); this.useMessageHandlers = useMessageHandlers; if (useMessageHandlers) CreateMessageHandlersDictionary(messageHandlerGroupId); MaxClientCount = maxClientCount; - clients = new Dictionary(maxClientCount); + clients = new ConcurrentDictionary(); InitializeClientIds(); SubToTransportEvents(); @@ -192,7 +210,7 @@ private void HandleConnect(Connection connection, Message connectMessage) AcceptConnection(connection); else if (ClientCount < MaxClientCount) { - if (!clients.ContainsValue(connection) && !pendingConnections.Contains(connection)) + if (!clients.ContainsKey(connection.Id) && !pendingConnections.Contains(connection)) { pendingConnections.Add(connection); Send(Message.Create(MessageHeader.Connect), connection); // Inform the client we've received the connection attempt @@ -235,17 +253,19 @@ private void AcceptConnection(Connection connection) { if (ClientCount < MaxClientCount) { - if (!clients.ContainsValue(connection)) + ushort clientId = GetAvailableClientId(); + if (clients.TryAdd(clientId, connection)) { - ushort clientId = GetAvailableClientId(); connection.Id = clientId; - clients.Add(clientId, connection); connection.ResetTimeout(); connection.SendWelcome(); return; } else + { Reject(connection, RejectReason.AlreadyConnected); + availableClientIds.Enqueue(clientId); + } } else Reject(connection, RejectReason.ServerFull); @@ -445,8 +465,7 @@ private void LocalDisconnect(Connection client, DisconnectReason reason) transport.Close(client); - if (clients.Remove(client.Id)) - availableClientIds.Enqueue(client.Id); + if (clients.TryRemove(client.Id, out Connection _)) availableClientIds.Enqueue(client.Id); if (client.IsConnected) OnClientDisconnected(client, reason); // Only run if the client was ever actually connected @@ -488,7 +507,7 @@ private void InitializeClientIds() if (MaxClientCount > ushort.MaxValue - 1) throw new Exception($"A server's max client count may not exceed {ushort.MaxValue - 1}!"); - availableClientIds = new Queue(MaxClientCount); + availableClientIds = new ConcurrentQueue(); for (ushort i = 1; i <= MaxClientCount; i++) availableClientIds.Enqueue(i); } @@ -497,8 +516,8 @@ private void InitializeClientIds() /// The client ID. 0 if none were available. private ushort GetAvailableClientId() { - if (availableClientIds.Count > 0) - return availableClientIds.Dequeue(); + if (availableClientIds.TryDequeue(out ushort id)) + return id; RiptideLogger.Log(LogType.Error, LogName, "No available client IDs, assigned 0!"); return 0;