diff --git a/src/Directory.Packages.props b/src/Directory.Packages.props index 4ba5fa1a9f..9eb964ba07 100644 --- a/src/Directory.Packages.props +++ b/src/Directory.Packages.props @@ -10,7 +10,6 @@ - diff --git a/src/EventStore.Transport.Tcp/EventStore.Transport.Tcp.csproj b/src/EventStore.Transport.Tcp/EventStore.Transport.Tcp.csproj deleted file mode 100644 index 069ee6f242..0000000000 --- a/src/EventStore.Transport.Tcp/EventStore.Transport.Tcp.csproj +++ /dev/null @@ -1,10 +0,0 @@ - - - true - - - - - - - diff --git a/src/EventStore.Transport.Tcp/Formatting/FormatterBase.cs b/src/EventStore.Transport.Tcp/Formatting/FormatterBase.cs deleted file mode 100644 index ffb2ed3346..0000000000 --- a/src/EventStore.Transport.Tcp/Formatting/FormatterBase.cs +++ /dev/null @@ -1,92 +0,0 @@ -using System; -using System.IO; -using EventStore.BufferManagement; - -namespace EventStore.Transport.Tcp.Formatting; - -public abstract class FormatterBase : IMessageFormatter -{ - /// - /// Gets a representing the IMessage provided. - /// - /// The message. - /// A with a representation of the message - public abstract BufferPool ToBufferPool(T message); - - /// - /// converts the message to a - /// - /// The message. - /// - public virtual ArraySegment ToArraySegment(T message) - { - return new ArraySegment(ToArray(message)); - } - - /// - /// Converts the message to a byte array - /// - /// The message. - /// - public virtual byte[] ToArray(T message) - { - using (BufferPool pool = ToBufferPool(message)) - { - return pool.ToByteArray(); - } - } - - /// - /// Gets a message from a - /// - /// The BufferPool to get data from. - /// - public virtual T From(BufferPool bufferPool) - { - if (bufferPool == null) - { - throw new ArgumentNullException("bufferPool"); - } - - var stream = new BufferPoolStream(bufferPool); - return From(stream); - } - - /// - /// Gets a message from a - /// - /// The segment containing the raw data. - /// - public virtual T From(ArraySegment segment) - { - using (var stream = new MemoryStream(segment.Array, segment.Offset, segment.Count, false)) - { - return From(stream); - } - } - - /// - /// Gets a message from a byte array - /// - /// The byte array. - /// - public virtual T From(byte[] array) - { - if (array == null) - { - throw new ArgumentNullException("array"); - } - - using (var stream = new MemoryStream(array, 0, array.Length, false)) - { - return From(stream); - } - } - - /// - /// Creates a message object from the specified stream - /// - /// The stream. - /// - public abstract T From(Stream stream); -} diff --git a/src/EventStore.Transport.Tcp/Formatting/IMessageFormatter.cs b/src/EventStore.Transport.Tcp/Formatting/IMessageFormatter.cs deleted file mode 100644 index ba46714574..0000000000 --- a/src/EventStore.Transport.Tcp/Formatting/IMessageFormatter.cs +++ /dev/null @@ -1,49 +0,0 @@ -using System; -using EventStore.BufferManagement; - -namespace EventStore.Transport.Tcp.Formatting; - -public interface IMessageFormatter -{ - /// - /// Converts the object to a representing a binary format of it. - /// - /// The message to convert. - /// A containing the data representing the object. - BufferPool ToBufferPool(T message); - - /// - /// Converts the object to a representing a binary format of it. - /// - /// The message to convert. - /// A containing the data representing the object. - ArraySegment ToArraySegment(T message); - - /// - /// Converts the object to a byte array representing a binary format of it. - /// - /// The message to convert. - /// A containing the data representing the object. - byte[] ToArray(T message); - - /// - /// Takes a and converts its contents to a message object - /// - /// The buffer pool. - /// A message representing the data given - T From(BufferPool bufferPool); - - /// - /// Takes an ArraySegment and converts its contents to a message object - /// - /// The buffer pool. - /// A message representing the data given - T From(ArraySegment segment); - - /// - /// Takes an Array and converts its contents to a message object - /// - /// The buffer pool. - /// A message representing the data given - T From(byte[] array); -} diff --git a/src/EventStore.Transport.Tcp/Formatting/RawMessageFormatter.cs b/src/EventStore.Transport.Tcp/Formatting/RawMessageFormatter.cs deleted file mode 100644 index f8fd2fcdb7..0000000000 --- a/src/EventStore.Transport.Tcp/Formatting/RawMessageFormatter.cs +++ /dev/null @@ -1,91 +0,0 @@ -using System; -using EventStore.BufferManagement; - -namespace EventStore.Transport.Tcp.Formatting; - -/// -/// Formatter which does not format anything, actually. Just outputs raw byte[]. -/// -public class RawMessageFormatter : IMessageFormatter -{ - private readonly BufferManager _bufferManager; - private readonly int _initialBuffers; - - /// - /// Initializes a new instance of the class. - /// - public RawMessageFormatter() : this(BufferManager.Default, 2) - { - } - - - /// - /// Initializes a new instance of the class. - /// - /// The buffer manager. - public RawMessageFormatter(BufferManager bufferManager) : this(bufferManager, 2) - { - } - - - /// - /// Initializes a new instance of the class. - /// - /// The buffer manager. - /// The number of initial buffers. - public RawMessageFormatter(BufferManager bufferManager, int initialBuffers) - { - _bufferManager = bufferManager; - _initialBuffers = initialBuffers; - } - - public BufferPool ToBufferPool(byte[] message) - { - if (message == null) - { - throw new ArgumentNullException("message"); - } - - var bufferPool = new BufferPool(_initialBuffers, _bufferManager); - var stream = new BufferPoolStream(bufferPool); - stream.Write(message, 0, message.Length); - return bufferPool; - } - - public ArraySegment ToArraySegment(byte[] message) - { - if (message == null) - { - throw new ArgumentNullException("message"); - } - - return new ArraySegment(message, 0, message.Length); - } - - public byte[] ToArray(byte[] message) - { - if (message == null) - { - throw new ArgumentNullException("message"); - } - - return message; - } - - public byte[] From(BufferPool bufferPool) - { - return bufferPool.ToByteArray(); - } - - public byte[] From(ArraySegment segment) - { - var msg = new byte[segment.Count]; - Buffer.BlockCopy(segment.Array, segment.Offset, msg, 0, segment.Count); - return msg; - } - - public byte[] From(byte[] array) - { - return array; - } -} diff --git a/src/EventStore.Transport.Tcp/Framing/IMessageFramer.cs b/src/EventStore.Transport.Tcp/Framing/IMessageFramer.cs deleted file mode 100644 index b254ccc5dc..0000000000 --- a/src/EventStore.Transport.Tcp/Framing/IMessageFramer.cs +++ /dev/null @@ -1,32 +0,0 @@ -using System; -using System.Collections.Generic; -using System.Threading; -using System.Threading.Tasks; - -namespace EventStore.Transport.Tcp.Framing; - -public interface IMessageFramer -{ - bool HasData { get; } - IEnumerable> FrameData(ArraySegment data); - void Reset(); -} - -/// -/// Encodes outgoing messages in frames and decodes incoming frames. -/// For decoding it uses an internal state, raising a registered -/// callback, once full message arrives -/// -public interface IMessageFramer : IMessageFramer -{ - void UnFrameData(IEnumerable> data); - void UnFrameData(ArraySegment data); - void RegisterMessageArrivedCallback(Action handler); -} - -public interface IAsyncMessageFramer : IMessageFramer -{ - ValueTask UnFrameData(IEnumerable> data, CancellationToken token); - ValueTask UnFrameData(ArraySegment data, CancellationToken token); - void RegisterMessageArrivedCallback(Func handler); -} diff --git a/src/EventStore.Transport.Tcp/Framing/LengthPrefixMessageFramer.cs b/src/EventStore.Transport.Tcp/Framing/LengthPrefixMessageFramer.cs deleted file mode 100644 index c86bb34ad7..0000000000 --- a/src/EventStore.Transport.Tcp/Framing/LengthPrefixMessageFramer.cs +++ /dev/null @@ -1,129 +0,0 @@ -using System; -using System.Collections.Generic; -using EventStore.Common.Utils; -using ILogger = Serilog.ILogger; - -namespace EventStore.Transport.Tcp.Framing; - -/// -/// Uses length-prefixed framing to encode outgoing messages and decode -/// incoming messages, using internal state and raising a callback once -/// full message arrives. -/// -public class LengthPrefixMessageFramer : IMessageFramer> -{ - private static readonly ILogger Log = Serilog.Log.ForContext(); - - public const int HeaderLength = sizeof(Int32); - - private byte[] _messageBuffer; - private int _bufferIndex = 0; - private Action> _receivedHandler; - private readonly int _maxPackageSize; - - private int _headerBytes = 0; - private int _packageLength = 0; - - /// - /// Initializes a new instance of the class. - /// - public LengthPrefixMessageFramer(int maxPackageSize = 64 * 1024 * 1024) - { - Ensure.Positive(maxPackageSize, "maxPackageSize"); - _maxPackageSize = maxPackageSize; - } - - public bool HasData => _headerBytes > 0; - - public void Reset() - { - _messageBuffer = null; - _headerBytes = 0; - _packageLength = 0; - _bufferIndex = 0; - } - - public void UnFrameData(IEnumerable> data) - { - if (data == null) - { - throw new ArgumentNullException("data"); - } - - foreach (ArraySegment buffer in data) - { - Parse(buffer); - } - } - - public void UnFrameData(ArraySegment data) - { - Parse(data); - } - - /// - /// Parses a stream chunking based on length-prefixed framing. - /// Calls are re-entrant and hold state internally. Once full message arrives, - /// callback is raised (it is registered via - /// - /// A byte array of data to append - private void Parse(ArraySegment bytes) - { - byte[] data = bytes.Array; - for (int i = bytes.Offset, n = bytes.Offset + bytes.Count; i < n; i++) - { - if (_headerBytes < HeaderLength) - { - _packageLength |= (data[i] << (_headerBytes * 8)); // little-endian order - ++_headerBytes; - if (_headerBytes == HeaderLength) - { - if (_packageLength <= 0 || _packageLength > _maxPackageSize) - { - Log.Error("FRAMING ERROR! Data:\n {data}", Common.Utils.Helper.FormatBinaryDump(bytes)); - throw new PackageFramingException(string.Format( - "Package size is out of bounds: {0} (max: {1}).", - _packageLength, _maxPackageSize)); - } - - _messageBuffer = new byte[_packageLength]; - } - } - else - { - int copyCnt = Math.Min(bytes.Count + bytes.Offset - i, _packageLength - _bufferIndex); - Buffer.BlockCopy(bytes.Array, i, _messageBuffer, _bufferIndex, copyCnt); - _bufferIndex += copyCnt; - i += copyCnt - 1; - - if (_bufferIndex == _packageLength) - { - if (_receivedHandler != null) - { - _receivedHandler(new ArraySegment(_messageBuffer, 0, _bufferIndex)); - } - - _messageBuffer = null; - _headerBytes = 0; - _packageLength = 0; - _bufferIndex = 0; - } - } - } - } - - public IEnumerable> FrameData(ArraySegment data) - { - var length = data.Count; - - yield return new ArraySegment( - new[] { (byte)length, (byte)(length >> 8), (byte)(length >> 16), (byte)(length >> 24) }); - yield return data; - } - - public void RegisterMessageArrivedCallback(Action> handler) - { - Ensure.NotNull(handler, nameof(handler)); - _receivedHandler = handler; - } -} diff --git a/src/EventStore.Transport.Tcp/Framing/LengthPrefixMessageFramerWithBufferPool.cs b/src/EventStore.Transport.Tcp/Framing/LengthPrefixMessageFramerWithBufferPool.cs deleted file mode 100644 index eddc913eb2..0000000000 --- a/src/EventStore.Transport.Tcp/Framing/LengthPrefixMessageFramerWithBufferPool.cs +++ /dev/null @@ -1,126 +0,0 @@ -using System; -using System.Collections.Generic; -using EventStore.BufferManagement; -using EventStore.Common.Utils; -using ILogger = Serilog.ILogger; - -namespace EventStore.Transport.Tcp.Framing; - -public class LengthPrefixMessageFramerWithBufferPool -{ - private static readonly ILogger Log = Serilog.Log.ForContext(); - - private const int PrefixLength = sizeof(int); - - private readonly int _maxPackageSize; - private readonly BufferManager _bufferManager; - private BufferPool _messageBuffer; - private Action _receivedHandler; - - private int _headerBytes; - private int _packageLength; - - /// - /// Initializes a new instance of the class. - /// - public LengthPrefixMessageFramerWithBufferPool(BufferManager bufferManager, - int maxPackageSize = 16 * 1024 * 1024) - { - Ensure.NotNull(bufferManager, "bufferManager"); - Ensure.Positive(maxPackageSize, "maxPackageSize"); - _bufferManager = bufferManager; - _maxPackageSize = maxPackageSize; - } - - public void Reset() - { - _messageBuffer = null; - _headerBytes = 0; - _packageLength = 0; - } - - public void UnFrameData(IEnumerable> data) - { - if (data == null) - { - throw new ArgumentNullException("data"); - } - - foreach (ArraySegment buffer in data) - { - Parse(buffer); - } - } - - public void UnFrameData(ArraySegment data) - { - Parse(data); - } - - /// - /// Parses a stream chunking based on length-prefixed framing. Calls are re-entrant and hold state internally. - /// - /// A byte array of data to append - private void Parse(ArraySegment bytes) - { - byte[] data = bytes.Array; - for (int i = bytes.Offset; i < bytes.Offset + bytes.Count;) - { - if (_headerBytes < PrefixLength) - { - _packageLength |= (data[i] << (_headerBytes * 8)); // little-endian order - ++_headerBytes; - i += 1; - if (_headerBytes == PrefixLength) - { - if (_packageLength <= 0 || _packageLength > _maxPackageSize) - { - Log.Error("FRAMING ERROR! Data:\n {data}", Common.Utils.Helper.FormatBinaryDump(bytes)); - throw new PackageFramingException(string.Format( - "Package size is out of bounds: {0} (max: {1}).", - _packageLength, _maxPackageSize)); - } - - _messageBuffer = new BufferPool(_bufferManager); - } - } - else - { - int copyCnt = Math.Min(bytes.Count + bytes.Offset - i, _packageLength - _messageBuffer.Length); - _messageBuffer.Append(bytes.Array, i, copyCnt); - i += copyCnt; - - if (_messageBuffer.Length == _packageLength) - { - if (_receivedHandler != null) - { - _receivedHandler(_messageBuffer); - } - - _messageBuffer = null; - _headerBytes = 0; - _packageLength = 0; - } - } - } - } - - public IEnumerable> FrameData(ArraySegment data) - { - var length = data.Count; - - yield return new ArraySegment( - new[] { (byte)length, (byte)(length >> 8), (byte)(length >> 16), (byte)(length >> 24) }); - yield return data; - } - - public void RegisterMessageArrivedCallback(Action handler) - { - if (handler == null) - { - throw new ArgumentNullException("handler"); - } - - _receivedHandler = handler; - } -} diff --git a/src/EventStore.Transport.Tcp/Framing/PackageFramingException.cs b/src/EventStore.Transport.Tcp/Framing/PackageFramingException.cs deleted file mode 100644 index 14e62f5cf6..0000000000 --- a/src/EventStore.Transport.Tcp/Framing/PackageFramingException.cs +++ /dev/null @@ -1,11 +0,0 @@ -using System; -using System.Runtime.Serialization; - -namespace EventStore.Transport.Tcp.Framing; - -public class PackageFramingException : Exception -{ - public PackageFramingException(string message) : base(message) - { - } -} diff --git a/src/EventStore.Transport.Tcp/Helper.cs b/src/EventStore.Transport.Tcp/Helper.cs deleted file mode 100644 index 6baed53061..0000000000 --- a/src/EventStore.Transport.Tcp/Helper.cs +++ /dev/null @@ -1,31 +0,0 @@ -using System; -using EventStore.Common.Utils; - -namespace EventStore.Transport.Tcp; - -internal static class Helper -{ - public static void EatException(Action action) - { - try - { - action(); - } - catch (Exception) - { - } - } - - public static T EatException(Func func, T defaultValue = default(T)) - { - Ensure.NotNull(func, "func"); - try - { - return func(); - } - catch (Exception) - { - return defaultValue; - } - } -} diff --git a/src/EventStore.Transport.Tcp/IMonitoredTcpConnection.cs b/src/EventStore.Transport.Tcp/IMonitoredTcpConnection.cs deleted file mode 100644 index 5c1a754fd1..0000000000 --- a/src/EventStore.Transport.Tcp/IMonitoredTcpConnection.cs +++ /dev/null @@ -1,25 +0,0 @@ -using System; - -namespace EventStore.Transport.Tcp; - -public interface IMonitoredTcpConnection -{ - bool IsReadyForSend { get; } - bool IsReadyForReceive { get; } - bool IsInitialized { get; } - bool IsFaulted { get; } - bool IsClosed { get; } - - bool InSend { get; } - bool InReceive { get; } - - DateTime? LastSendStarted { get; } - DateTime? LastReceiveStarted { get; } - - int PendingSendBytes { get; } - int InSendBytes { get; } - int PendingReceivedBytes { get; } - - long TotalBytesSent { get; } - long TotalBytesReceived { get; } -} diff --git a/src/EventStore.Transport.Tcp/ITcpConnection.cs b/src/EventStore.Transport.Tcp/ITcpConnection.cs deleted file mode 100644 index f0947aa9d8..0000000000 --- a/src/EventStore.Transport.Tcp/ITcpConnection.cs +++ /dev/null @@ -1,26 +0,0 @@ -using System; -using System.Collections.Generic; -using System.Net; -using System.Net.Sockets; - -namespace EventStore.Transport.Tcp; - -public interface ITcpConnection -{ - event Action ConnectionClosed; - - Guid ConnectionId { get; } - string ClientConnectionName { get; } - IPEndPoint RemoteEndPoint { get; } - IPEndPoint LocalEndPoint { get; } - int SendQueueSize { get; } - int PendingSendBytes { get; } - long TotalBytesSent { get; } - long TotalBytesReceived { get; } - bool IsClosed { get; } - - void ReceiveAsync(Action>> callback); - void EnqueueSend(IEnumerable> data); - void Close(string reason); - void SetClientConnectionName(string clientConnectionName); -} diff --git a/src/EventStore.Transport.Tcp/SocketArgsPool.cs b/src/EventStore.Transport.Tcp/SocketArgsPool.cs deleted file mode 100644 index c4e4bb1b36..0000000000 --- a/src/EventStore.Transport.Tcp/SocketArgsPool.cs +++ /dev/null @@ -1,52 +0,0 @@ -using System; -using System.Collections.Concurrent; -using System.Net.Sockets; - -namespace EventStore.Transport.Tcp; - -internal class SocketArgsPool -{ - public readonly string Name; - - private readonly Func _socketArgsCreator; - - private readonly ConcurrentStack _socketArgsPool = - new ConcurrentStack(); - - public SocketArgsPool(string name, int initialCount, Func socketArgsCreator) - { - if (socketArgsCreator == null) - { - throw new ArgumentNullException("socketArgsCreator"); - } - - if (initialCount < 0) - { - throw new ArgumentOutOfRangeException("initialCount"); - } - - Name = name; - _socketArgsCreator = socketArgsCreator; - - for (int i = 0; i < initialCount; ++i) - { - _socketArgsPool.Push(socketArgsCreator()); - } - } - - public SocketAsyncEventArgs Get() - { - SocketAsyncEventArgs result; - if (_socketArgsPool.TryPop(out result)) - { - return result; - } - - return _socketArgsCreator(); - } - - public void Return(SocketAsyncEventArgs socketArgs) - { - _socketArgsPool.Push(socketArgs); - } -} diff --git a/src/EventStore.Transport.Tcp/TcpClientConnector.cs b/src/EventStore.Transport.Tcp/TcpClientConnector.cs deleted file mode 100644 index ee9eb567d5..0000000000 --- a/src/EventStore.Transport.Tcp/TcpClientConnector.cs +++ /dev/null @@ -1,241 +0,0 @@ -using System; -using System.Collections.Concurrent; -using System.Net; -using System.Net.Security; -using System.Net.Sockets; -using System.Security.Cryptography.X509Certificates; -using System.Threading; -using EventStore.Common.Utils; -using ILogger = Serilog.ILogger; - -namespace EventStore.Transport.Tcp; - -public class TcpClientConnector -{ - private const int CheckPeriodMs = 200; - - private readonly SocketArgsPool _connectSocketArgsPool; - private readonly ConcurrentDictionary _pendingConnections; - private readonly Timer _timer; - - private static readonly ILogger Log = Serilog.Log.ForContext(); - - public TcpClientConnector() - { - _connectSocketArgsPool = new SocketArgsPool("TcpClientConnector._connectSocketArgsPool", - TcpConfiguration.ConnectPoolSize, - CreateConnectSocketArgs); - _pendingConnections = new ConcurrentDictionary(); - _timer = new Timer(TimerCallback); - // prevent possible null reference exceptions in case of slow initialization - _timer.Change(CheckPeriodMs, Timeout.Infinite); - } - - private SocketAsyncEventArgs CreateConnectSocketArgs() - { - var socketArgs = new SocketAsyncEventArgs(); - socketArgs.Completed += ConnectCompleted; - socketArgs.UserToken = new CallbacksStateToken(); - return socketArgs; - } - - public ITcpConnection ConnectTo(Guid connectionId, - IPEndPoint remoteEndPoint, - TimeSpan connectionTimeout, - Action onConnectionEstablished = null, - Action onConnectionFailed = null, - bool verbose = true) - { - Ensure.NotNull(remoteEndPoint, "remoteEndPoint"); - return TcpConnection.CreateConnectingTcpConnection(connectionId, remoteEndPoint, this, connectionTimeout, - onConnectionEstablished, onConnectionFailed, verbose); - } - - public ITcpConnection ConnectSslTo(Guid connectionId, - string targetHost, - string[] otherNames, - IPEndPoint remoteEndPoint, - TimeSpan connectionTimeout, - CertificateDelegates.ServerCertificateValidator sslServerCertValidator, - Func clientCertificatesSelector, - Action onConnectionEstablished = null, - Action onConnectionFailed = null, - bool verbose = true) - { - Ensure.NotNull(remoteEndPoint, "remoteEndPoint"); - return TcpConnectionSsl.CreateConnectingConnection(connectionId, targetHost, otherNames, remoteEndPoint, sslServerCertValidator, - clientCertificatesSelector, this, connectionTimeout, onConnectionEstablished, onConnectionFailed, verbose); - } - - internal void InitConnect(IPEndPoint serverEndPoint, - Action onSocketAssigned, - Action onConnectionEstablished, - Action onConnectionFailed, - ITcpConnection connection, - TimeSpan connectionTimeout) - { - if (serverEndPoint == null) - { - throw new ArgumentNullException("serverEndPoint"); - } - - if (onSocketAssigned == null) - { - throw new ArgumentNullException("onSocketAssigned"); - } - - if (onConnectionEstablished == null) - { - throw new ArgumentNullException("onConnectionEstablished"); - } - - if (onConnectionFailed == null) - { - throw new ArgumentNullException("onConnectionFailed"); - } - - var socketArgs = _connectSocketArgsPool.Get(); - var connectingSocket = new Socket(serverEndPoint.AddressFamily, SocketType.Stream, ProtocolType.Tcp); - onSocketAssigned(connectingSocket); - socketArgs.RemoteEndPoint = serverEndPoint; - socketArgs.AcceptSocket = connectingSocket; - var callbacks = (CallbacksStateToken)socketArgs.UserToken; - callbacks.OnConnectionEstablished = onConnectionEstablished; - callbacks.OnConnectionFailed = onConnectionFailed; - callbacks.PendingConnection = new PendingConnection(connection, DateTime.UtcNow.Add(connectionTimeout)); - - AddToConnecting(callbacks.PendingConnection); - - try - { - var firedAsync = connectingSocket.ConnectAsync(socketArgs); - if (!firedAsync) - { - ProcessConnect(socketArgs); - } - } - catch (ObjectDisposedException) - { - HandleBadConnect(socketArgs); - } - } - - private void ConnectCompleted(object sender, SocketAsyncEventArgs e) - { - ProcessConnect(e); - } - - private void ProcessConnect(SocketAsyncEventArgs e) - { - if (e.SocketError != SocketError.Success) - { - HandleBadConnect(e); - } - else - { - OnSocketConnected(e); - } - } - - private void HandleBadConnect(SocketAsyncEventArgs socketArgs) - { - var serverEndPoint = socketArgs.RemoteEndPoint; - var socketError = socketArgs.SocketError; - var callbacks = (CallbacksStateToken)socketArgs.UserToken; - var onConnectionFailed = callbacks.OnConnectionFailed; - var pendingConnection = callbacks.PendingConnection; - - Helper.EatException(() => socketArgs.AcceptSocket.Close()); - socketArgs.AcceptSocket = null; - callbacks.Reset(); - _connectSocketArgsPool.Return(socketArgs); - - if (RemoveFromConnecting(pendingConnection)) - { - onConnectionFailed((IPEndPoint)serverEndPoint, socketError); - } - } - - private void OnSocketConnected(SocketAsyncEventArgs socketArgs) - { - var remoteEndPoint = (IPEndPoint)socketArgs.RemoteEndPoint; - var socket = socketArgs.AcceptSocket; - var callbacks = (CallbacksStateToken)socketArgs.UserToken; - var onConnectionEstablished = callbacks.OnConnectionEstablished; - var pendingConnection = callbacks.PendingConnection; - - socketArgs.AcceptSocket = null; - callbacks.Reset(); - _connectSocketArgsPool.Return(socketArgs); - - if (RemoveFromConnecting(pendingConnection)) - { - onConnectionEstablished(remoteEndPoint, socket); - } - } - - private void TimerCallback(object state) - { - foreach (var pendingConnection in _pendingConnections.Values) - { - if (DateTime.UtcNow >= pendingConnection.WhenToKill && RemoveFromConnecting(pendingConnection)) - { - Helper.EatException(() => pendingConnection.Connection.Close("Connection establishment timeout.")); - } - } - - try - { - _timer.Change(CheckPeriodMs, Timeout.Infinite); - } - catch (ObjectDisposedException) - { - // ignore - } - } - - private void AddToConnecting(PendingConnection pendingConnection) - { - _pendingConnections.TryAdd(pendingConnection.Connection.ConnectionId, pendingConnection); - } - - private bool RemoveFromConnecting(PendingConnection pendingConnection) - { - PendingConnection conn; - if (pendingConnection.Connection == null) - { - Log.Warning("Network Card disconnected"); - return false; - } - - return _pendingConnections.TryRemove(pendingConnection.Connection.ConnectionId, out conn) - && Interlocked.CompareExchange(ref conn.Done, 1, 0) == 0; - } - - private class CallbacksStateToken - { - public Action OnConnectionEstablished; - public Action OnConnectionFailed; - public PendingConnection PendingConnection; - - public void Reset() - { - OnConnectionEstablished = null; - OnConnectionFailed = null; - PendingConnection = null; - } - } - - private class PendingConnection - { - public readonly ITcpConnection Connection; - public readonly DateTime WhenToKill; - public int Done; - - public PendingConnection(ITcpConnection connection, DateTime whenToKill) - { - Connection = connection; - WhenToKill = whenToKill; - } - } -} diff --git a/src/EventStore.Transport.Tcp/TcpConfiguration.cs b/src/EventStore.Transport.Tcp/TcpConfiguration.cs deleted file mode 100644 index 85355e5287..0000000000 --- a/src/EventStore.Transport.Tcp/TcpConfiguration.cs +++ /dev/null @@ -1,16 +0,0 @@ -namespace EventStore.Transport.Tcp; - -public static class TcpConfiguration -{ - public const int SocketCloseTimeoutSecs = 1; - - public const int AcceptBacklogCount = 128; - public const int ConcurrentAccepts = 1; - public const int AcceptPoolSize = ConcurrentAccepts * 2; - - public const int ConnectPoolSize = 32; - public const int SendReceivePoolSize = 512; - - public const int BufferChunksCount = 512; - public const int SocketBufferSize = 8 * 1024; -} diff --git a/src/EventStore.Transport.Tcp/TcpConnection.cs b/src/EventStore.Transport.Tcp/TcpConnection.cs deleted file mode 100644 index 4790529436..0000000000 --- a/src/EventStore.Transport.Tcp/TcpConnection.cs +++ /dev/null @@ -1,563 +0,0 @@ -using System; -using System.Collections.Concurrent; -using System.Collections.Generic; -using System.IO; -using System.Net; -using System.Net.Sockets; -using System.Threading; -using EventStore.BufferManagement; -using EventStore.Common.Utils; -using ILogger = Serilog.ILogger; - -namespace EventStore.Transport.Tcp; - -public class TcpConnection : TcpConnectionBase, ITcpConnection -{ - internal const int MaxSendPacketSize = 65535 /*Max IP packet size*/ - 20 /*IP packet header size*/ - 32 /*TCP min header size*/; - - internal static readonly BufferManager BufferManager = - new BufferManager(TcpConfiguration.BufferChunksCount, TcpConfiguration.SocketBufferSize); - - private static readonly ILogger Log = Serilog.Log.ForContext(); - - private static readonly SocketArgsPool SocketArgsPool = new SocketArgsPool("TcpConnection.SocketArgsPool", - TcpConfiguration.SendReceivePoolSize, - () => new SocketAsyncEventArgs()); - - public static ITcpConnection CreateConnectingTcpConnection(Guid connectionId, - IPEndPoint remoteEndPoint, - TcpClientConnector connector, - TimeSpan connectionTimeout, - Action onConnectionEstablished, - Action onConnectionFailed, - bool verbose) - { - var connection = new TcpConnection(connectionId, remoteEndPoint, verbose); - // ReSharper disable ImplicitlyCapturedClosure - connector.InitConnect(remoteEndPoint, - (socket) => - { - connection.InitSocket(socket); - }, - (_, socket) => - { - connection.InitSendReceive(); - if (onConnectionEstablished != null) - { - onConnectionEstablished(connection); - } - }, - (_, socketError) => - { - if (onConnectionFailed != null) - { - onConnectionFailed(connection, socketError); - } - }, connection, connectionTimeout); - // ReSharper restore ImplicitlyCapturedClosure - return connection; - } - - public static ITcpConnection CreateAcceptedTcpConnection(Guid connectionId, IPEndPoint remoteEndPoint, - Socket socket, bool verbose) - { - var connection = new TcpConnection(connectionId, remoteEndPoint, verbose); - connection.InitSocket(socket); - connection.InitSendReceive(); - return connection; - } - - public event Action ConnectionClosed; - - public Guid ConnectionId - { - get { return _connectionId; } - } - - public int SendQueueSize - { - get { return _sendQueue.Count; } - } - - public string ClientConnectionName - { - get { return _clientConnectionName; } - } - - private readonly Guid _connectionId; - private readonly bool _verbose; - private string _clientConnectionName; - - private Socket _socket; - private SocketAsyncEventArgs _receiveSocketArgs; - private SocketAsyncEventArgs _sendSocketArgs; - - private readonly ConcurrentQueueWrapper> _sendQueue = - new ConcurrentQueueWrapper>(); - - private readonly Queue _receiveQueue = new Queue(); - private readonly MemoryStream _memoryStream = new MemoryStream(); - private long _memoryStreamOffset = 0L; - - private readonly object _receivingLock = new object(); - private readonly object _sendLock = new object(); - private readonly object _closeLock = new object(); - private bool _isSending; - private volatile bool _isClosed; - private volatile bool _isClosing; - - private Action>> _receiveCallback; - - private TcpConnection(Guid connectionId, IPEndPoint remoteEndPoint, bool verbose) : base(remoteEndPoint) - { - Ensure.NotEmptyGuid(connectionId, "connectionId"); - - _connectionId = connectionId; - _verbose = verbose; - } - - private void InitSocket(Socket socket) - { - _socket = socket; - } - - private void InitSendReceive() - { - InitConnectionBase(_socket); - lock (_sendLock) - { - try - { - _socket.NoDelay = true; - } - catch (ObjectDisposedException) - { - CloseInternal(SocketError.Shutdown, "Socket disposed."); - return; - } - catch (SocketException) - { - CloseInternal(SocketError.Shutdown, "Socket is disposed."); - return; - } - - var receiveSocketArgs = SocketArgsPool.Get(); - _receiveSocketArgs = receiveSocketArgs; - _receiveSocketArgs.AcceptSocket = _socket; - _receiveSocketArgs.Completed += OnReceiveAsyncCompleted; - - var sendSocketArgs = SocketArgsPool.Get(); - _sendSocketArgs = sendSocketArgs; - _sendSocketArgs.AcceptSocket = _socket; - _sendSocketArgs.Completed += OnSendAsyncCompleted; - } - - StartReceive(); - TrySend(); - } - - public void EnqueueSend(IEnumerable> data) - { - lock (_sendLock) - { - int bytes = 0; - foreach (var segment in data) - { - _sendQueue.Enqueue(segment); - bytes += segment.Count; - } - - NotifySendScheduled(bytes); - } - - TrySend(); - } - - private void TrySend() - { - bool continueSendSynchronously; - try - { - do - { - lock (_sendLock) - { - if (_isSending || (_sendQueue.IsEmpty && _memoryStreamOffset >= _memoryStream.Length) || _sendSocketArgs == null) - { - return; - } - - if (TcpConnectionMonitor.Default.IsSendBlocked()) - { - return; - } - - _isSending = true; - } - - if (_memoryStreamOffset >= _memoryStream.Length) - { - _memoryStream.SetLength(0); - _memoryStreamOffset = 0L; - - ArraySegment sendPiece; - while (_sendQueue.TryDequeue(out sendPiece)) - { - _memoryStream.Write(sendPiece.Array, sendPiece.Offset, sendPiece.Count); - if (_memoryStream.Length >= MaxSendPacketSize) - { - break; - } - } - } - - int sendingBytes = Math.Min((int)_memoryStream.Length - (int)_memoryStreamOffset, MaxSendPacketSize); - - _sendSocketArgs.SetBuffer(_memoryStream.GetBuffer(), (int)_memoryStreamOffset, sendingBytes); - _memoryStreamOffset += sendingBytes; - - NotifySendStarting(_sendSocketArgs.Count); - var firedAsync = _sendSocketArgs.AcceptSocket.SendAsync(_sendSocketArgs); - if (firedAsync) - { - continueSendSynchronously = false; - } - else - { - continueSendSynchronously = ProcessSend(_sendSocketArgs); - } - } while (continueSendSynchronously); - } - catch (ObjectDisposedException) - { - ReturnSendingSocketArgs(); - } - } - - private void OnSendAsyncCompleted(object sender, SocketAsyncEventArgs e) - { - if (ProcessSend(e)) - { - TrySend(); - } - } - - private bool ProcessSend(SocketAsyncEventArgs socketArgs) - { - if (socketArgs.SocketError != SocketError.Success) - { - NotifySendCompleted(0); - ReturnSendingSocketArgs(); - CloseInternal(socketArgs.SocketError, "Socket send error."); - return false; - } - else - { - NotifySendCompleted(socketArgs.Count); - - if (_isClosed) - { - ReturnSendingSocketArgs(); - return false; - } - else - { - lock (_sendLock) - { - _isSending = false; - } - - return true; - } - } - } - - public void ReceiveAsync(Action>> callback) - { - if (callback == null) - { - throw new ArgumentNullException("callback"); - } - - lock (_receivingLock) - { - if (_receiveCallback != null) - { - Log.Fatal("ReceiveAsync called again while previous call was not fulfilled"); - throw new InvalidOperationException( - "ReceiveAsync called again while previous call was not fulfilled"); - } - - _receiveCallback = callback; - } - - TryDequeueReceivedData(); - } - - private void StartReceive() - { - try - { - bool continueReceiveSynchronously = true; - - do - { - var buffer = BufferManager.CheckOut(); - if (buffer.Array == null || buffer.Count == 0 || buffer.Array.Length < buffer.Offset + buffer.Count) - { - throw new Exception("Invalid buffer allocated"); - } - // TODO AN: do we need to lock on _receiveSocketArgs?.. - lock (_receiveSocketArgs) - { - _receiveSocketArgs.SetBuffer(buffer.Array, buffer.Offset, buffer.Count); - if (_receiveSocketArgs.Buffer == null) - { - throw new Exception("Buffer was not set"); - } - } - - NotifyReceiveStarting(); - bool firedAsync; - lock (_receiveSocketArgs) - { - if (_receiveSocketArgs.Buffer == null) - { - throw new Exception("Buffer was lost"); - } - - firedAsync = _receiveSocketArgs.AcceptSocket.ReceiveAsync(_receiveSocketArgs); - } - - if (firedAsync) - { - continueReceiveSynchronously = false; - } - else - { - var processReceiveSuccess = ProcessReceive(_receiveSocketArgs); - if (processReceiveSuccess) - { - TryDequeueReceivedData(); - } - - continueReceiveSynchronously = processReceiveSuccess; - } - } while (continueReceiveSynchronously); - } - catch (ObjectDisposedException) - { - ReturnReceivingSocketArgs(); - } - } - - private void OnReceiveAsyncCompleted(object sender, SocketAsyncEventArgs e) - { - if (ProcessReceive(e)) - { - TryDequeueReceivedData(); - StartReceive(); - } - } - - private bool ProcessReceive(SocketAsyncEventArgs socketArgs) - { - // socket closed normally or some error occurred - if (socketArgs.BytesTransferred == 0 || socketArgs.SocketError != SocketError.Success) - { - NotifyReceiveCompleted(0); - ReturnReceivingSocketArgs(); - CloseInternal(socketArgs.SocketError, - socketArgs.SocketError != SocketError.Success ? "Socket receive error" : "Socket closed"); - return false; - } - - NotifyReceiveCompleted(socketArgs.BytesTransferred); - - lock (_receivingLock) - { - var buf = new ArraySegment(socketArgs.Buffer, socketArgs.Offset, socketArgs.Count); - _receiveQueue.Enqueue(new ReceivedData(buf, socketArgs.BytesTransferred)); - } - - lock (_receiveSocketArgs) - { - if (socketArgs.Buffer == null) - { - throw new Exception("Cleaning already null buffer"); - } - - socketArgs.SetBuffer(null, 0, 0); - } - - return true; - } - - private void TryDequeueReceivedData() - { - Action>> callback; - List res; - lock (_receivingLock) - { - // no awaiting callback or no data to dequeue - if (_receiveCallback == null || _receiveQueue.Count == 0) - { - return; - } - - res = new List(_receiveQueue.Count); - while (_receiveQueue.Count > 0) - { - res.Add(_receiveQueue.Dequeue()); - } - - callback = _receiveCallback; - _receiveCallback = null; - } - - var data = new ArraySegment[res.Count]; - int bytes = 0; - for (int i = 0; i < data.Length; ++i) - { - var d = res[i]; - bytes += d.DataLen; - data[i] = new ArraySegment(d.Buf.Array, d.Buf.Offset, d.DataLen); - } - - lock (_closeLock) - { - if (!_isClosed) - { - callback(this, data); - } - } - - for (int i = 0, n = res.Count; i < n; ++i) - { - BufferManager.CheckIn(res[i].Buf); // dispose buffers - } - - NotifyReceiveDispatched(bytes); - } - - public void Close(string reason) - { - CloseInternal(SocketError.Success, reason ?? "Normal socket close."); // normal socket closing - } - - private void CloseInternal(SocketError socketError, string reason) - { - lock (_closeLock) - { - if (_isClosing) - { - return; - } - - _isClosing = true; - } - - if (_socket != null) - { - Helper.EatException(() => _socket.Shutdown(SocketShutdown.Both)); - Helper.EatException(() => _socket.Close()); - } - - lock (_closeLock) - { - _isClosed = true; - } - - NotifyClosed(); - - if (_verbose) - { - Log.Information( - "ES {connectionType} closed [{dateTime:HH:mm:ss.fff}: N{remoteEndPoint}, L{localEndPoint}, {connectionId:B}]:Received bytes: {totalBytesReceived}, Sent bytes: {totalBytesSent}", - GetType().Name, DateTime.UtcNow, RemoteEndPoint, LocalEndPoint, _connectionId, - TotalBytesReceived, TotalBytesSent); - Log.Information( - "ES {connectionType} closed [{dateTime:HH:mm:ss.fff}: N{remoteEndPoint}, L{localEndPoint}, {connectionId:B}]:Send calls: {sendCalls}, callbacks: {sendCallbacks}", - GetType().Name, DateTime.UtcNow, RemoteEndPoint, LocalEndPoint, _connectionId, - SendCalls, SendCallbacks); - Log.Information( - "ES {connectionType} closed [{dateTime:HH:mm:ss.fff}: N{remoteEndPoint}, L{localEndPoint}, {connectionId:B}]:Receive calls: {receiveCalls}, callbacks: {receiveCallbacks}", - GetType().Name, DateTime.UtcNow, RemoteEndPoint, LocalEndPoint, _connectionId, - ReceiveCalls, ReceiveCallbacks); - Log.Information( - "ES {connectionType} closed [{dateTime:HH:mm:ss.fff}: N{remoteEndPoint}, L{localEndPoint}, {connectionId:B}]:Close reason: [{socketError}] {reason}", - GetType().Name, DateTime.UtcNow, RemoteEndPoint, LocalEndPoint, _connectionId, - socketError, reason); - } - - lock (_sendLock) - { - if (!_isSending) - { - ReturnSendingSocketArgs(); - } - } - - var handler = ConnectionClosed; - if (handler != null) - { - handler(this, socketError); - } - } - - private void ReturnSendingSocketArgs() - { - var socketArgs = Interlocked.Exchange(ref _sendSocketArgs, null); - if (socketArgs != null) - { - socketArgs.Completed -= OnSendAsyncCompleted; - socketArgs.AcceptSocket = null; - if (socketArgs.Buffer != null) - { - socketArgs.SetBuffer(null, 0, 0); - } - - SocketArgsPool.Return(socketArgs); - } - } - - private void ReturnReceivingSocketArgs() - { - var socketArgs = Interlocked.Exchange(ref _receiveSocketArgs, null); - if (socketArgs != null) - { - socketArgs.Completed -= OnReceiveAsyncCompleted; - socketArgs.AcceptSocket = null; - if (socketArgs.Buffer != null) - { - BufferManager.CheckIn( - new ArraySegment(socketArgs.Buffer, socketArgs.Offset, socketArgs.Count)); - socketArgs.SetBuffer(null, 0, 0); - } - - SocketArgsPool.Return(socketArgs); - } - } - - public void SetClientConnectionName(string clientConnectionName) - { - _clientConnectionName = clientConnectionName; - } - - public override string ToString() - { - return RemoteEndPoint.ToString(); - } - - private struct ReceivedData - { - public readonly ArraySegment Buf; - public readonly int DataLen; - - public ReceivedData(ArraySegment buf, int dataLen) - { - Buf = buf; - DataLen = dataLen; - } - } -} diff --git a/src/EventStore.Transport.Tcp/TcpConnectionBase.cs b/src/EventStore.Transport.Tcp/TcpConnectionBase.cs deleted file mode 100644 index 2b9f7d41b6..0000000000 --- a/src/EventStore.Transport.Tcp/TcpConnectionBase.cs +++ /dev/null @@ -1,240 +0,0 @@ -using System; -using System.Net; -using System.Net.Sockets; -using System.Threading; -using EventStore.Common.Utils; - -namespace EventStore.Transport.Tcp; - -public class TcpConnectionBase : IMonitoredTcpConnection -{ - public IPEndPoint RemoteEndPoint - { - get { return _remoteEndPoint; } - } - - public IPEndPoint LocalEndPoint - { - get { return _localEndPoint; } - } - - public bool IsInitialized - { - get { return _socket != null; } - } - - public bool IsClosed - { - get { return _isClosed; } - } - - public bool InSend - { - get { return Interlocked.Read(ref _lastSendStarted) >= 0; } - } - - public bool InReceive - { - get { return Interlocked.Read(ref _lastReceiveStarted) >= 0; } - } - - public int PendingSendBytes - { - get { return _pendingSendBytes; } - } - - public int InSendBytes - { - get { return _inSendBytes; } - } - - public int PendingReceivedBytes - { - get { return _pendingReceivedBytes; } - } - - public long TotalBytesSent - { - get { return Interlocked.Read(ref _totalBytesSent); } - } - - public long TotalBytesReceived - { - get { return Interlocked.Read(ref _totalBytesReceived); } - } - - public int SendCalls - { - get { return _sentAsyncs; } - } - - public int SendCallbacks - { - get { return _sentAsyncCallbacks; } - } - - public int ReceiveCalls - { - get { return _recvAsyncs; } - } - - public int ReceiveCallbacks - { - get { return _recvAsyncCallbacks; } - } - - public bool IsReadyForSend - { - get - { - try - { - return !_isClosed && _socket.Poll(0, SelectMode.SelectWrite); - } - catch (ObjectDisposedException) - { - //TODO: why do we get this? - return false; - } - } - } - - public bool IsReadyForReceive - { - get - { - try - { - return !_isClosed && _socket.Poll(0, SelectMode.SelectRead); - } - catch (ObjectDisposedException) - { - //TODO: why do we get this? - return false; - } - } - } - - public bool IsFaulted - { - get - { - try - { - return !_isClosed && _socket.Poll(0, SelectMode.SelectError); - } - catch (ObjectDisposedException) - { - //TODO: why do we get this? - return false; - } - } - } - - public DateTime? LastSendStarted - { - get - { - var ticks = Interlocked.Read(ref _lastSendStarted); - return ticks >= 0 ? new DateTime(ticks) : (DateTime?)null; - } - } - - public DateTime? LastReceiveStarted - { - get - { - var ticks = Interlocked.Read(ref _lastReceiveStarted); - return ticks >= 0 ? new DateTime(ticks) : (DateTime?)null; - } - } - - private Socket _socket; - protected readonly IPEndPoint _remoteEndPoint; - private IPEndPoint _localEndPoint; - - private long _lastSendStarted = -1; - private long _lastReceiveStarted = -1; - private bool _isClosed; - - private int _pendingSendBytes; - private int _inSendBytes; - private int _pendingReceivedBytes; - private long _totalBytesSent; - private long _totalBytesReceived; - - private int _sentAsyncs; - private int _sentAsyncCallbacks; - private int _recvAsyncs; - private int _recvAsyncCallbacks; - - public TcpConnectionBase(IPEndPoint remoteEndPoint) - { - Ensure.NotNull(remoteEndPoint, "remoteEndPoint"); - _remoteEndPoint = remoteEndPoint; - - TcpConnectionMonitor.Default.Register(this); - } - - protected void InitConnectionBase(Socket socket) - { - Ensure.NotNull(socket, "socket"); - - _socket = socket; - _localEndPoint = Helper.EatException(() => (IPEndPoint)socket.LocalEndPoint); - } - - protected void NotifySendScheduled(int bytes) - { - Interlocked.Add(ref _pendingSendBytes, bytes); - } - - protected void NotifySendStarting(int bytes) - { - if (Interlocked.CompareExchange(ref _lastSendStarted, DateTime.UtcNow.Ticks, -1) != -1) - { - throw new Exception("Concurrent send detected."); - } - - Interlocked.Add(ref _pendingSendBytes, -bytes); - Interlocked.Add(ref _inSendBytes, bytes); - Interlocked.Increment(ref _sentAsyncs); - } - - protected void NotifySendCompleted(int bytes) - { - Interlocked.Exchange(ref _lastSendStarted, -1); - Interlocked.Add(ref _inSendBytes, -bytes); - Interlocked.Add(ref _totalBytesSent, bytes); - Interlocked.Increment(ref _sentAsyncCallbacks); - } - - protected void NotifyReceiveStarting() - { - if (Interlocked.CompareExchange(ref _lastReceiveStarted, DateTime.UtcNow.Ticks, -1) != -1) - { - throw new Exception("Concurrent receive detected."); - } - - Interlocked.Increment(ref _recvAsyncs); - } - - protected void NotifyReceiveCompleted(int bytes) - { - Interlocked.Exchange(ref _lastReceiveStarted, -1); - Interlocked.Add(ref _pendingReceivedBytes, bytes); - Interlocked.Add(ref _totalBytesReceived, bytes); - Interlocked.Increment(ref _recvAsyncCallbacks); - } - - protected void NotifyReceiveDispatched(int bytes) - { - Interlocked.Add(ref _pendingReceivedBytes, -bytes); - } - - protected void NotifyClosed() - { - _isClosed = true; - TcpConnectionMonitor.Default.Unregister(this); - } -} diff --git a/src/EventStore.Transport.Tcp/TcpConnectionMonitor.cs b/src/EventStore.Transport.Tcp/TcpConnectionMonitor.cs deleted file mode 100644 index a4a0b67323..0000000000 --- a/src/EventStore.Transport.Tcp/TcpConnectionMonitor.cs +++ /dev/null @@ -1,213 +0,0 @@ -using System; -using System.Collections.Concurrent; -using System.Linq; -using EventStore.Common.Utils; -using ILogger = Serilog.ILogger; - -namespace EventStore.Transport.Tcp; - -public class TcpConnectionMonitor -{ - public static readonly TcpConnectionMonitor Default = new TcpConnectionMonitor(); - private static readonly ILogger Log = Serilog.Log.ForContext(); - - private readonly object _statsLock = new object(); - - private readonly ConcurrentDictionary _connections = - new ConcurrentDictionary(); - - private long _sentTotal; - private long _receivedTotal; - private long _sentSinceLastRun; - private long _receivedSinceLastRun; - private long _pendingSendOnLastRun; - private long _inSendOnLastRun; - private long _pendingReceivedOnLastRun; - - private bool _anySendBlockedOnLastRun; - private DateTime _lastUpdateTime; - - private TcpConnectionMonitor() - { - } - - public void Register(IMonitoredTcpConnection connection) - { - _connections.TryAdd(connection, new ConnectionData(connection)); - } - - public void Unregister(IMonitoredTcpConnection connection) - { - _connections.TryRemove(connection, out _); - } - - public TcpStats GetTcpStats() - { - ConnectionData[] connections = _connections.Values.ToArray(); - lock (_statsLock) - { - var stats = AnalyzeConnections(connections, DateTime.UtcNow - _lastUpdateTime); - _lastUpdateTime = DateTime.UtcNow; - return stats; - } - } - - public IMonitoredTcpConnection[] GetTcpConnectionStats() - { - GetTcpStats(); - var monitoredConnections = _connections.Values.Select(conn => conn.Connection).ToArray(); - return monitoredConnections; - } - - private TcpStats AnalyzeConnections(ConnectionData[] connections, TimeSpan measurePeriod) - { - _receivedSinceLastRun = 0; - _sentSinceLastRun = 0; - _pendingSendOnLastRun = 0; - _inSendOnLastRun = 0; - _pendingReceivedOnLastRun = 0; - _anySendBlockedOnLastRun = false; - - foreach (var connection in connections) - { - AnalyzeConnection(connection); - } - - var stats = new TcpStats(connections.Length, - _sentTotal, - _receivedTotal, - _sentSinceLastRun, - _receivedSinceLastRun, - _pendingSendOnLastRun, - _inSendOnLastRun, - _pendingReceivedOnLastRun, - measurePeriod); - - return stats; - } - - private void AnalyzeConnection(ConnectionData connectionData) - { - var connection = connectionData.Connection; - if (!connection.IsInitialized) - { - return; - } - - if (connection.IsFaulted) - { - Log.Information("# {connection} is faulted", connection); - return; - } - - UpdateStatistics(connectionData); - - CheckPendingReceived(connection); - CheckPendingSend(connection); - CheckMissingSendCallback(connectionData, connection); - CheckMissingReceiveCallback(connectionData, connection); - } - - private void UpdateStatistics(ConnectionData connectionData) - { - var connection = connectionData.Connection; - long totalBytesSent = connection.TotalBytesSent; - long totalBytesReceived = connection.TotalBytesReceived; - long pendingSend = connection.PendingSendBytes; - long inSend = connection.InSendBytes; - long pendingReceived = connection.PendingReceivedBytes; - - _sentSinceLastRun += totalBytesSent - connectionData.LastTotalBytesSent; - _receivedSinceLastRun += totalBytesReceived - connectionData.LastTotalBytesReceived; - - _sentTotal += _sentSinceLastRun; - _receivedTotal += _receivedSinceLastRun; - - _pendingSendOnLastRun += pendingSend; - _inSendOnLastRun += inSend; - _pendingReceivedOnLastRun = pendingReceived; - - connectionData.LastTotalBytesSent = totalBytesSent; - connectionData.LastTotalBytesReceived = totalBytesReceived; - } - - private static void CheckMissingReceiveCallback(ConnectionData connectionData, - IMonitoredTcpConnection connection) - { - bool inReceive = connection.InReceive; - bool isReadyForReceive = connection.IsReadyForReceive; - DateTime? lastReceiveStarted = connection.LastReceiveStarted; - - int sinceLastReceive = (int)(DateTime.UtcNow - lastReceiveStarted.GetValueOrDefault()).TotalMilliseconds; - bool missingReceiveCallback = inReceive && isReadyForReceive && sinceLastReceive > 500; - - if (missingReceiveCallback && connectionData.LastMissingReceiveCallBack) - { - Log.Error( - "# {connection} {sinceLastReceive}ms since last Receive started. No completion callback received, but socket status is READY_FOR_RECEIVE", - connection, sinceLastReceive); - } - - connectionData.LastMissingReceiveCallBack = missingReceiveCallback; - } - - private void CheckMissingSendCallback(ConnectionData connectionData, IMonitoredTcpConnection connection) - { - // snapshot all data? - bool inSend = connection.InSend; - bool isReadyForSend = connection.IsReadyForSend; - DateTime? lastSendStarted = connection.LastSendStarted; - int inSendBytes = connection.InSendBytes; - - int sinceLastSend = (int)(DateTime.UtcNow - lastSendStarted.GetValueOrDefault()).TotalMilliseconds; - bool missingSendCallback = inSend && isReadyForSend && sinceLastSend > 500; - - if (missingSendCallback && connectionData.LastMissingSendCallBack) - { - // _anySendBlockedOnLastRun = true; - Log.Error( - "# {connection} {sinceLastSend}ms since last send started. No completion callback received, but socket status is READY_FOR_SEND. In send: {inSendBytes}", - connection, sinceLastSend, inSendBytes); - } - - connectionData.LastMissingSendCallBack = missingSendCallback; - } - - private static void CheckPendingSend(IMonitoredTcpConnection connection) - { - int pendingSendBytes = connection.PendingSendBytes; - if (pendingSendBytes > 128 * 1024) - { - Log.Information("# {connection} {pendingSendKiloBytes}kb pending send", connection, pendingSendBytes / 1024); - } - } - - private static void CheckPendingReceived(IMonitoredTcpConnection connection) - { - int pendingReceivedBytes = connection.PendingReceivedBytes; - if (pendingReceivedBytes > 128 * 1024) - { - Log.Information("# {connection} {pendingReceivedKiloBytes}kb are not dispatched", connection, - pendingReceivedBytes / 1024); - } - } - - public bool IsSendBlocked() - { - return _anySendBlockedOnLastRun; - } - - private class ConnectionData - { - public readonly IMonitoredTcpConnection Connection; - public bool LastMissingSendCallBack; - public bool LastMissingReceiveCallBack; - public long LastTotalBytesSent; - public long LastTotalBytesReceived; - - public ConnectionData(IMonitoredTcpConnection connection) - { - Connection = connection; - } - } -} diff --git a/src/EventStore.Transport.Tcp/TcpConnectionSsl.cs b/src/EventStore.Transport.Tcp/TcpConnectionSsl.cs deleted file mode 100644 index ad7d9135c0..0000000000 --- a/src/EventStore.Transport.Tcp/TcpConnectionSsl.cs +++ /dev/null @@ -1,763 +0,0 @@ -using System; -using System.Collections.Generic; -using System.IO; -using System.Net; -using System.Net.Security; -using System.Net.Sockets; -using System.Security.Authentication; -using System.Security.Cryptography.X509Certificates; -using System.Threading; -using System.Threading.Tasks; -using EventStore.Common.Utils; -using ILogger = Serilog.ILogger; - -namespace EventStore.Transport.Tcp; - -public class TcpConnectionSsl : TcpConnectionBase, ITcpConnection -{ - private static readonly ILogger Log = Serilog.Log.ForContext(); - - public static ITcpConnection CreateConnectingConnection(Guid connectionId, - string targetHost, - string[] otherNames, - IPEndPoint remoteEndPoint, - CertificateDelegates.ServerCertificateValidator serverCertValidator, - Func clientCertificatesSelector, - TcpClientConnector connector, - TimeSpan connectionTimeout, - Action onConnectionEstablished, - Action onConnectionFailed, - bool verbose) - { - var connection = new TcpConnectionSsl(connectionId, remoteEndPoint, verbose); - // ReSharper disable ImplicitlyCapturedClosure - connector.InitConnect(remoteEndPoint, - (socket) => - { - connection.InitClientSocket(socket); - }, - (_, socket) => - { - connection.InitSslStream(targetHost, otherNames, serverCertValidator, clientCertificatesSelector, verbose); - onConnectionEstablished?.Invoke(connection); - }, - (_, socketError) => - { - onConnectionFailed?.Invoke(connection, socketError); - }, connection, connectionTimeout); - // ReSharper restore ImplicitlyCapturedClosure - return connection; - } - - public static ITcpConnection CreateServerFromSocket(Guid connectionId, - IPEndPoint remoteEndPoint, - Socket socket, - Func serverCertificateSelector, - Func intermediatesSelector, - CertificateDelegates.ClientCertificateValidator clientCertValidator, - bool verbose) - { - var connection = new TcpConnectionSsl(connectionId, remoteEndPoint, verbose); - connection.InitServerSocket(socket, serverCertificateSelector, intermediatesSelector, clientCertValidator, verbose); - return connection; - } - - public event Action ConnectionClosed; - - public Guid ConnectionId - { - get { return _connectionId; } - } - - public int SendQueueSize - { - get { return _sendQueue.Count; } - } - - public string ClientConnectionName - { - get { return _clientConnectionName; } - } - - private readonly Guid _connectionId; - private readonly bool _verbose; - public string _clientConnectionName; - - private Socket _socket; - - private readonly ConcurrentQueueWrapper> _sendQueue = new(); - - private readonly ConcurrentQueueWrapper - _receiveQueue = new(); - - private readonly MemoryStream _memoryStream = new(); - private long _memoryStreamOffset; - - private readonly object _streamLock = new(); - private readonly object _closeLock = new(); - private bool _isSending; - private int _receiveHandling; - private volatile bool _isClosed; - private volatile bool _isClosing; - - private Action>> _receiveCallback; - - private SslStream _sslStream; - private bool _isAuthenticated; - private int _sendingBytes; - private CertificateDelegates.ServerCertificateValidator _serverCertValidator; - private CertificateDelegates.ClientCertificateValidator _clientCertValidator; - private string[] _otherNames; - private readonly byte[] _receiveBuffer = new byte[TcpConnection.BufferManager.ChunkSize]; - - private TcpConnectionSsl(Guid connectionId, IPEndPoint remoteEndPoint, bool verbose) : base(remoteEndPoint) - { - Ensure.NotEmptyGuid(connectionId, "connectionId"); - - _connectionId = connectionId; - _verbose = verbose; - } - - private void InitServerSocket( - Socket socket, - Func serverCertificateSelector, - Func intermediatesSelector, - CertificateDelegates.ClientCertificateValidator clientCertValidator, - bool verbose) - { - InitConnectionBase(socket); - if (verbose) - { - Console.WriteLine("TcpConnectionSsl::InitClientSocket({0}, L{1})", RemoteEndPoint, LocalEndPoint); - } - - _clientCertValidator = clientCertValidator; - - lock (_streamLock) - { - try - { - socket.NoDelay = true; - } - catch (ObjectDisposedException) - { - CloseInternal(SocketError.Shutdown, "Socket is disposed."); - return; - } - catch (SocketException) - { - CloseInternal(SocketError.Shutdown, "Socket is disposed."); - return; - } - - try - { - _sslStream = new SslStream(new NetworkStream(socket, true), false); - } - catch (IOException exc) - { - Log.Debug(exc, "[S{remoteEndPoint}, L{localEndPoint}]: IOException on NetworkStream. The socket has already been disposed.", RemoteEndPoint, - LocalEndPoint); - return; - } - - Task.Run(async () => await AuthenticateAsServerAsync(serverCertificateSelector, intermediatesSelector)); - } - } - - private async Task AuthenticateAsServerAsync( - Func serverCertificateSelector, - Func intermediatesSelector) - { - try - { - var enabledSslProtocols = SslProtocols.Tls12 | SslProtocols.Tls13; - var certificate = serverCertificateSelector?.Invoke(); - var intermediates = intermediatesSelector?.Invoke(); - Ensure.NotNull(certificate, "certificate"); - - await _sslStream.AuthenticateAsServerAsync(new SslServerAuthenticationOptions - { - ServerCertificateContext = SslStreamCertificateContext.Create( - certificate!, intermediates, offline: true), - ClientCertificateRequired = true, - EnabledSslProtocols = enabledSslProtocols, - CertificateRevocationCheckMode = X509RevocationMode.NoCheck, - RemoteCertificateValidationCallback = ValidateClientCertificate, - ApplicationProtocols = [], - AllowRenegotiation = false, - }); - - lock (_streamLock) - { - if (_verbose) - { - DisplaySslStreamInfo(_sslStream); - } - - _isAuthenticated = true; - } - } - catch (AuthenticationException exc) - { - Log.Information(exc, - "[S{remoteEndPoint}, L{localEndPoint}]: Authentication exception on AuthenticateAsServerAsync.", - RemoteEndPoint, LocalEndPoint); - CloseInternal(SocketError.SocketError, exc.Message); - } - catch (ObjectDisposedException) - { - CloseInternal(SocketError.SocketError, "SslStream disposed."); - } - catch (Exception exc) - { - Log.Information(exc, - "[S{remoteEndPoint}, L{localEndPoint}]: Exception on AuthenticateAsServerAsync.", - RemoteEndPoint, LocalEndPoint); - CloseInternal(SocketError.SocketError, exc.Message); - } - - StartReceive(); - TrySend(); - } - - private void InitClientSocket(Socket socket) => - _socket = socket; - - private void InitSslStream(string targetHost, string[] otherNames, CertificateDelegates.ServerCertificateValidator serverCertValidator, Func clientCertificatesSelector, bool verbose) - { - Ensure.NotNull(targetHost, "targetHost"); - InitConnectionBase(_socket); - if (verbose) - { - Console.WriteLine("TcpConnectionSsl::InitClientSslStream({0}, L{1})", RemoteEndPoint, LocalEndPoint); - } - - _serverCertValidator = serverCertValidator; - _otherNames = otherNames; - - lock (_streamLock) - { - try - { - _socket.NoDelay = true; - } - catch (ObjectDisposedException) - { - CloseInternal(SocketError.Shutdown, "Socket is disposed."); - return; - } - catch (SocketException) - { - CloseInternal(SocketError.Shutdown, "Socket is disposed."); - return; - } - - try - { - _sslStream = new SslStream(new NetworkStream(_socket, true), false, ValidateServerCertificate, null); - } - catch (IOException exc) - { - Log.Debug(exc, "[S{remoteEndPoint}, L{localEndPoint}]: IOException on NetworkStream. The socket has already been disposed.", RemoteEndPoint, - LocalEndPoint); - return; - } - - try - { - var enabledSslProtocols = SslProtocols.Tls12 | SslProtocols.Tls13; - var clientCertificates = clientCertificatesSelector?.Invoke(); - _sslStream.BeginAuthenticateAsClient(targetHost, clientCertificates, enabledSslProtocols, false, OnEndAuthenticateAsClient, _sslStream); - } - catch (AuthenticationException exc) - { - Log.Information(exc, - "[S{remoteEndPoint}, L{localEndPoint}]: Authentication exception on BeginAuthenticateAsClient.", - RemoteEndPoint, LocalEndPoint); - CloseInternal(SocketError.SocketError, exc.Message); - } - catch (ObjectDisposedException) - { - CloseInternal(SocketError.SocketError, "SslStream disposed."); - } - catch (Exception exc) - { - Log.Information(exc, - "[S{remoteEndPoint}, {localEndPoint}]: Exception on BeginAuthenticateAsClient.", RemoteEndPoint, - LocalEndPoint); - CloseInternal(SocketError.SocketError, exc.Message); - } - } - } - - private void OnEndAuthenticateAsClient(IAsyncResult ar) - { - try - { - lock (_streamLock) - { - var sslStream = (SslStream)ar.AsyncState; - sslStream.EndAuthenticateAsClient(ar); - if (_verbose) - { - DisplaySslStreamInfo(sslStream); - } - - _isAuthenticated = true; - } - - StartReceive(); - TrySend(); - } - catch (AuthenticationException exc) - { - Log.Information(exc, - "[S{remoteEndPoint}, L{localEndPoint}]: Authentication exception on EndAuthenticateAsClient.", - RemoteEndPoint, LocalEndPoint); - CloseInternal(SocketError.SocketError, exc.Message); - } - catch (ObjectDisposedException) - { - CloseInternal(SocketError.SocketError, "SslStream disposed."); - } - catch (Exception exc) - { - Log.Information(exc, "[S{remoteEndPoint}, L{localEndPoint}]: Exception on EndAuthenticateAsClient.", - RemoteEndPoint, LocalEndPoint); - CloseInternal(SocketError.SocketError, exc.Message); - } - } - - // The following method is invoked by the RemoteCertificateValidationDelegate. - public bool ValidateServerCertificate(object sender, X509Certificate certificate, X509Chain chain, - SslPolicyErrors sslPolicyErrors) - { - var (isValid, error) = _serverCertValidator(certificate, chain, sslPolicyErrors, _otherNames); - if (!isValid && error != null) - { - Log.Error("Server certificate validation error: {e}", error); - } - return isValid; - } - - public bool ValidateClientCertificate(object sender, X509Certificate certificate, X509Chain chain, - SslPolicyErrors sslPolicyErrors) - { - var (isValid, error) = _clientCertValidator(certificate, chain, sslPolicyErrors); - if (!isValid && error != null) - { - Log.Error("Client certificate validation error: {e}", error); - } - return isValid; - } - - private void DisplaySslStreamInfo(SslStream stream) - { - Log.Information("[S{remoteEndPoint}, L{localEndPoint}]", RemoteEndPoint, LocalEndPoint); - try - { - Log.Verbose("Cipher suite: {cipherSuite}", stream.NegotiatedCipherSuite); - } - catch (NotImplementedException) - { - } - - Log.Information("Protocol: {sslProtocol}", stream.SslProtocol); - Log.Information("Is authenticated: {isAuthenticated} as server? {isServer}", stream.IsAuthenticated, - stream.IsServer); - Log.Information("IsSigned: {isSigned}", stream.IsSigned); - Log.Information("Is Encrypted: {isEncrypted}", stream.IsEncrypted); - Log.Information("Can read: {canRead}, write {canWrite}", stream.CanRead, stream.CanWrite); - Log.Information("Can timeout: {canTimeout}", stream.CanTimeout); - try - { - Log.Information("Certificate revocation list checked: {checkCertRevocationStatus}", - stream.CheckCertRevocationStatus); - } - catch (NotImplementedException) - { - } - - X509Certificate localCert = stream.LocalCertificate; - if (localCert != null) - { - Log.Information( - "Local certificate was issued to {subject} and is valid from {effectiveDate} until {expirationDate}.", - localCert.Subject, localCert.GetEffectiveDateString(), localCert.GetExpirationDateString()); - } - else - { - Log.Information("Local certificate is null."); - } - - // Display the properties of the client's certificate. - X509Certificate remoteCert = stream.RemoteCertificate; - if (remoteCert != null) - { - Log.Information( - "Remote certificate was issued to {subject} and is valid from {remoteCertEffectiveDate} until {remoteCertExpirationDate}.", - remoteCert.Subject, remoteCert.GetEffectiveDateString(), remoteCert.GetExpirationDateString()); - } - else - { - Log.Information("Remote certificate is null."); - } - } - - public void EnqueueSend(IEnumerable> data) - { - lock (_streamLock) - { - int bytes = 0; - foreach (var segment in data) - { - _sendQueue.Enqueue(segment); - bytes += segment.Count; - } - - NotifySendScheduled(bytes); - } - - TrySend(); - } - - private void TrySend() - { - bool continueSendSynchronously = true; - try - { - do - { - lock (_streamLock) - { - if (_isSending || (_sendQueue.IsEmpty && _memoryStreamOffset >= _memoryStream.Length) || _sslStream == null || !_isAuthenticated) - { - return; - } - - if (TcpConnectionMonitor.Default.IsSendBlocked()) - { - return; - } - - _isSending = true; - } - - if (_memoryStreamOffset >= _memoryStream.Length) - { - _memoryStream.SetLength(0); - _memoryStreamOffset = 0L; - - ArraySegment sendPiece; - while (_sendQueue.TryDequeue(out sendPiece)) - { - _memoryStream.Write(sendPiece.Array, sendPiece.Offset, sendPiece.Count); - if (_memoryStream.Length >= TcpConnection.MaxSendPacketSize) - { - break; - } - } - } - - _sendingBytes = Math.Min((int)_memoryStream.Length - (int)_memoryStreamOffset, TcpConnection.MaxSendPacketSize); - - NotifySendStarting(_sendingBytes); - var result = _sslStream.BeginWrite(_memoryStream.GetBuffer(), (int)_memoryStreamOffset, _sendingBytes, OnEndWrite, null); - _memoryStreamOffset += _sendingBytes; - continueSendSynchronously = result.CompletedSynchronously; - if (continueSendSynchronously) - { - EndWrite(result); - } - } while (continueSendSynchronously); - } - catch (SocketException exc) - { - Log.Debug(exc, "SocketException '{e}' during BeginWrite.", exc.SocketErrorCode); - CloseInternal(exc.SocketErrorCode, "SocketException during BeginWrite."); - } - catch (ObjectDisposedException) - { - CloseInternal(SocketError.SocketError, "SslStream disposed."); - } - catch (Exception exc) - { - Log.Debug(exc, "Exception during BeginWrite."); - CloseInternal(SocketError.SocketError, "Exception during BeginWrite"); - } - } - - private void OnEndWrite(IAsyncResult ar) - { - if (ar.CompletedSynchronously) - { - return; - } - - EndWrite(ar); - TrySend(); - } - - private void EndWrite(IAsyncResult ar) - { - try - { - _sslStream.EndWrite(ar); - NotifySendCompleted(_sendingBytes); - - lock (_streamLock) - { - _isSending = false; - } - } - catch (SocketException exc) - { - Log.Debug(exc, "SocketException '{e}' during EndWrite.", exc.SocketErrorCode); - NotifySendCompleted(0); - CloseInternal(exc.SocketErrorCode, "SocketException during EndWrite."); - } - catch (ObjectDisposedException) - { - NotifySendCompleted(0); - CloseInternal(SocketError.SocketError, "SslStream disposed."); - } - catch (Exception exc) - { - Log.Debug(exc, "Exception during EndWrite."); - NotifySendCompleted(0); - CloseInternal(SocketError.SocketError, "Exception during EndWrite."); - } - } - - public void ReceiveAsync(Action>> callback) - { - Ensure.NotNull(callback, "callback"); - - if (Interlocked.Exchange(ref _receiveCallback, callback) != null) - { - Log.Fatal("ReceiveAsync called again while previous call wasn't fulfilled"); - throw new InvalidOperationException("ReceiveAsync called again while previous call wasn't fulfilled"); - } - - TryDequeueReceivedData(); - } - - private void StartReceive() - { - try - { - bool continueReceiveSynchronously = true; - - do - { - NotifyReceiveStarting(); - var result = _sslStream.BeginRead(_receiveBuffer, 0, _receiveBuffer.Length, OnEndRead, null); - continueReceiveSynchronously = result.CompletedSynchronously; - if (continueReceiveSynchronously) - { - EndRead(result); - } - } while (continueReceiveSynchronously); - } - catch (SocketException exc) - { - Log.Debug(exc, "SocketException '{e}' during BeginRead.", exc.SocketErrorCode); - CloseInternal(exc.SocketErrorCode, "SocketException during BeginRead."); - } - catch (ObjectDisposedException) - { - CloseInternal(SocketError.SocketError, "SslStream disposed."); - } - catch (Exception exc) - { - Log.Debug(exc, "Exception during BeginRead."); - CloseInternal(SocketError.SocketError, "Exception during BeginRead."); - } - } - - private void OnEndRead(IAsyncResult ar) - { - if (ar.CompletedSynchronously) - { - return; - } - - EndRead(ar); - StartReceive(); - } - - private void EndRead(IAsyncResult ar) - { - int bytesRead; - try - { - bytesRead = _sslStream.EndRead(ar); - } - catch (SocketException exc) - { - Log.Debug(exc, "SocketException '{e}' during EndRead.", exc.SocketErrorCode); - NotifyReceiveCompleted(0); - CloseInternal(exc.SocketErrorCode, "SocketException during EndRead."); - return; - } - catch (ObjectDisposedException) - { - NotifyReceiveCompleted(0); - CloseInternal(SocketError.SocketError, "SslStream disposed."); - return; - } - catch (Exception exc) - { - Log.Debug(exc, "Exception during EndRead."); - NotifyReceiveCompleted(0); - CloseInternal(SocketError.SocketError, "Exception during EndRead."); - return; - } - - if (bytesRead <= 0) // socket closed normally - { - NotifyReceiveCompleted(0); - CloseInternal(SocketError.Success, "Socket closed."); - return; - } - - NotifyReceiveCompleted(bytesRead); - - var buffer = TcpConnection.BufferManager.CheckOut(); - if (buffer.Array == null || buffer.Count == 0 || buffer.Array.Length < buffer.Offset + buffer.Count) - { - throw new Exception("Invalid buffer allocated."); - } - - Buffer.BlockCopy(_receiveBuffer, 0, buffer.Array, buffer.Offset, bytesRead); - var buf = new ArraySegment(buffer.Array, buffer.Offset, buffer.Count); - _receiveQueue.Enqueue(new ReceivedData(buf, bytesRead)); - - TryDequeueReceivedData(); - } - - private void TryDequeueReceivedData() - { - if (Interlocked.CompareExchange(ref _receiveHandling, 1, 0) != 0) - { - return; - } - - do - { - if (!_receiveQueue.IsEmpty && _receiveCallback != null) - { - var callback = Interlocked.Exchange(ref _receiveCallback, null); - if (callback == null) - { - Log.Fatal("Some threading issue in TryDequeueReceivedData! Callback is null!"); - throw new Exception("Some threading issue in TryDequeueReceivedData! Callback is null!"); - } - - var res = new List(_receiveQueue.Count); - ReceivedData piece; - while (_receiveQueue.TryDequeue(out piece)) - { - res.Add(piece); - } - - var data = new ArraySegment[res.Count]; - int bytes = 0; - for (int i = 0; i < data.Length; ++i) - { - var d = res[i]; - bytes += d.DataLen; - data[i] = new ArraySegment(d.Buf.Array, d.Buf.Offset, d.DataLen); - } - - lock (_closeLock) - { - if (!_isClosed) - { - callback(this, data); - } - } - - for (int i = 0, n = res.Count; i < n; ++i) - { - TcpConnection.BufferManager.CheckIn(res[i].Buf); // dispose buffers - } - - NotifyReceiveDispatched(bytes); - } - - Interlocked.Exchange(ref _receiveHandling, 0); - } while (!_receiveQueue.IsEmpty - && _receiveCallback != null - && Interlocked.CompareExchange(ref _receiveHandling, 1, 0) == 0); - } - - public void Close(string reason) => - CloseInternal(SocketError.Success, reason ?? "Normal socket close."); // normal socket closing - - private void CloseInternal(SocketError socketError, string reason) - { - lock (_closeLock) - { - if (_isClosing) - { - return; - } - - _isClosing = true; - } - - if (_sslStream != null) - { - Helper.EatException(() => _sslStream.Close()); - } - - if (_socket != null) - { - Helper.EatException(() => _socket.Dispose()); - } - - lock (_closeLock) - { - _isClosed = true; - } - - NotifyClosed(); - - if (_verbose) - { - Log.Information( - "ES {connectionType} closed [{dateTime:HH:mm:ss.fff}: N{remoteEndPoint}, L{localEndPoint}, {connectionId:B}]:Received bytes: {totalBytesReceived}, Sent bytes: {totalBytesSent}", - GetType().Name, DateTime.UtcNow, RemoteEndPoint, LocalEndPoint, _connectionId, - TotalBytesReceived, TotalBytesSent); - Log.Information( - "ES {connectionType} closed [{dateTime:HH:mm:ss.fff}: N{remoteEndPoint}, L{localEndPoint}, {connectionId:B}]:Send calls: {sendCalls}, callbacks: {sendCallbacks}", - GetType().Name, DateTime.UtcNow, RemoteEndPoint, LocalEndPoint, _connectionId, - SendCalls, SendCallbacks); - Log.Information( - "ES {connectionType} closed [{dateTime:HH:mm:ss.fff}: N{remoteEndPoint}, L{localEndPoint}, {connectionId:B}]:Receive calls: {receiveCalls}, callbacks: {receiveCallbacks}", - GetType().Name, DateTime.UtcNow, RemoteEndPoint, LocalEndPoint, _connectionId, - ReceiveCalls, ReceiveCallbacks); - Log.Information( - "ES {connectionType} closed [{dateTime:HH:mm:ss.fff}: N{remoteEndPoint}, L{localEndPoint}, {connectionId:B}]:Close reason: [{e}] {reason}", - GetType().Name, DateTime.UtcNow, RemoteEndPoint, LocalEndPoint, _connectionId, - socketError, reason); - } - - var handler = ConnectionClosed; - handler?.Invoke(this, socketError); - } - - public void SetClientConnectionName(string clientConnectionName) => - _clientConnectionName = clientConnectionName; - - public override string ToString() => - "S" + RemoteEndPoint; - - private struct ReceivedData(ArraySegment buf, int dataLen) - { - public readonly ArraySegment Buf = buf; - public readonly int DataLen = dataLen; - } -} diff --git a/src/EventStore.Transport.Tcp/TcpServerListener.cs b/src/EventStore.Transport.Tcp/TcpServerListener.cs deleted file mode 100644 index 338241df8e..0000000000 --- a/src/EventStore.Transport.Tcp/TcpServerListener.cs +++ /dev/null @@ -1,139 +0,0 @@ -using System; -using System.Net; -using System.Net.Sockets; -using EventStore.Common.Utils; -using ILogger = Serilog.ILogger; - -namespace EventStore.Transport.Tcp; - -public class TcpServerListener -{ - private static readonly ILogger Log = Serilog.Log.ForContext(); - - private readonly IPEndPoint _serverEndPoint; - private readonly Socket _listeningSocket; - private readonly SocketArgsPool _acceptSocketArgsPool; - private Action _onSocketAccepted; - - public TcpServerListener(IPEndPoint serverEndPoint) - { - Ensure.NotNull(serverEndPoint, "serverEndPoint"); - - _serverEndPoint = serverEndPoint; - - _listeningSocket = new Socket(serverEndPoint.AddressFamily, SocketType.Stream, ProtocolType.Tcp); - - _acceptSocketArgsPool = new SocketArgsPool("TcpServerListener.AcceptSocketArgsPool", - TcpConfiguration.ConcurrentAccepts * 2, - CreateAcceptSocketArgs); - } - - private SocketAsyncEventArgs CreateAcceptSocketArgs() - { - var socketArgs = new SocketAsyncEventArgs(); - socketArgs.Completed += AcceptCompleted; - return socketArgs; - } - - public void StartListening(Action callback, string securityType) - { - Ensure.NotNull(callback, "callback"); - - _onSocketAccepted = callback; - - Log.Information("Starting {securityType} TCP listening on TCP endpoint: {serverEndPoint}.", securityType, - _serverEndPoint); - try - { - _listeningSocket.ExclusiveAddressUse = true; - _listeningSocket.Bind(_serverEndPoint); - _listeningSocket.Listen(TcpConfiguration.AcceptBacklogCount); - } - catch (Exception) - { - Log.Information("Failed to listen on TCP endpoint: {serverEndPoint}.", _serverEndPoint); - Helper.EatException(() => _listeningSocket.Close(TcpConfiguration.SocketCloseTimeoutSecs)); - throw; - } - - for (int i = 0; i < TcpConfiguration.ConcurrentAccepts; ++i) - { - StartAccepting(); - } - } - - private void StartAccepting() - { - var socketArgs = _acceptSocketArgsPool.Get(); - - try - { - var firedAsync = _listeningSocket.AcceptAsync(socketArgs); - if (!firedAsync) - { - ProcessAccept(socketArgs); - } - } - catch (ObjectDisposedException) - { - HandleBadAccept(socketArgs); - } - } - - private void AcceptCompleted(object sender, SocketAsyncEventArgs e) - { - ProcessAccept(e); - } - - private void ProcessAccept(SocketAsyncEventArgs e) - { - if (e.SocketError != SocketError.Success) - { - HandleBadAccept(e); - } - else - { - var acceptSocket = e.AcceptSocket; - e.AcceptSocket = null; - _acceptSocketArgsPool.Return(e); - - OnSocketAccepted(acceptSocket); - } - - StartAccepting(); - } - - private void HandleBadAccept(SocketAsyncEventArgs socketArgs) - { - Helper.EatException( - () => - { - if (socketArgs.AcceptSocket != null) // avoid annoying exceptions - { - socketArgs.AcceptSocket.Close(TcpConfiguration.SocketCloseTimeoutSecs); - } - }); - socketArgs.AcceptSocket = null; - _acceptSocketArgsPool.Return(socketArgs); - } - - private void OnSocketAccepted(Socket socket) - { - IPEndPoint socketEndPoint; - try - { - socketEndPoint = (IPEndPoint)socket.RemoteEndPoint; - } - catch (Exception) - { - return; - } - - _onSocketAccepted(socketEndPoint, socket); - } - - public void Stop() - { - Helper.EatException(() => _listeningSocket.Close(TcpConfiguration.SocketCloseTimeoutSecs)); - } -} diff --git a/src/EventStore.Transport.Tcp/TcpStats.cs b/src/EventStore.Transport.Tcp/TcpStats.cs deleted file mode 100644 index 7cf7c51d97..0000000000 --- a/src/EventStore.Transport.Tcp/TcpStats.cs +++ /dev/null @@ -1,76 +0,0 @@ -using System; - -namespace EventStore.Transport.Tcp; - -public class TcpStats -{ - /// - ///Number of TCP connections to Event Store - /// - public readonly int Connections; - /// - ///Total bytes sent from TCP connections - /// - public readonly long SentBytesTotal; - /// - ///Total bytes received by TCP connections - /// - public readonly long ReceivedBytesTotal; - /// - ///Total bytes sent to TCP connections since last run - /// - public readonly long SentBytesSinceLastRun; - /// - ///Total bytes received by TCP connections since last run - /// - public readonly long ReceivedBytesSinceLastRun; - /// - ///Sending speed in bytes per second - /// - public readonly double SendingSpeed; - /// - ///Receiving speed in bytes per second. - /// - public readonly double ReceivingSpeed; - /// - ///Number of bytes waiting to be sent to connections - /// - public readonly long PendingSend; - /// - ///Number of bytes sent to connections but not yet acknowledged by the receiving party - /// - public readonly long InSend; - /// - ///Number of bytes waiting to be received by connections - /// - public readonly long PendingReceived; - /// - ///Time elapsed since last stats read - /// - public readonly TimeSpan MeasureTime; - - public TcpStats(int connections, - long sentBytesTotal, - long receivedBytesTotal, - long sentBytesSinceLastRunSinceLastRun, - long receivedBytesSinceLastRun, - long pendingSend, - long inSend, - long pendingReceived, - TimeSpan measureTime) - { - Connections = connections; - SentBytesTotal = sentBytesTotal; - ReceivedBytesTotal = receivedBytesTotal; - SentBytesSinceLastRun = sentBytesSinceLastRunSinceLastRun; - ReceivedBytesSinceLastRun = receivedBytesSinceLastRun; - PendingSend = pendingSend; - InSend = inSend; - PendingReceived = pendingReceived; - MeasureTime = measureTime; - SendingSpeed = (MeasureTime.TotalSeconds < 0.00001) ? 0 : SentBytesSinceLastRun / MeasureTime.TotalSeconds; - ReceivingSpeed = (MeasureTime.TotalSeconds < 0.00001) - ? 0 - : ReceivedBytesSinceLastRun / MeasureTime.TotalSeconds; - } -} diff --git a/src/EventStore.Transport.Tcp/TcpTypedConnection.cs b/src/EventStore.Transport.Tcp/TcpTypedConnection.cs deleted file mode 100644 index 149a4f9bf9..0000000000 --- a/src/EventStore.Transport.Tcp/TcpTypedConnection.cs +++ /dev/null @@ -1,122 +0,0 @@ -using System; -using System.Collections.Generic; -using System.Net; -using System.Net.Sockets; -using EventStore.Transport.Tcp.Formatting; -using EventStore.Transport.Tcp.Framing; -using ILogger = Serilog.ILogger; - -namespace EventStore.Transport.Tcp; - -public class TcpTypedConnection -{ - private static readonly ILogger Log = - Serilog.Log.ForContext(Serilog.Core.Constants.SourceContextPropertyName, "TcpTypedConnection"); - - public event Action, SocketError> ConnectionClosed; - - private readonly ITcpConnection _connection; - private readonly IMessageFormatter _formatter; - private readonly IMessageFramer> _framer; - - private Action, T> _receiveCallback; - - public EndPoint RemoteEndPoint - { - get { return _connection.RemoteEndPoint; } - } - - public EndPoint LocalEndPoint - { - get { return _connection.LocalEndPoint; } - } - - public int SendQueueSize - { - get { return _connection.SendQueueSize; } - } - - public TcpTypedConnection(ITcpConnection connection, - IMessageFormatter formatter, - IMessageFramer> framer) - { - if (formatter == null) - { - throw new ArgumentNullException("formatter"); - } - - if (framer == null) - { - throw new ArgumentNullException("framer"); - } - - _connection = connection; - _formatter = formatter; - _framer = framer; - - connection.ConnectionClosed += OnConnectionClosed; - - //Setup callback for incoming messages - framer.RegisterMessageArrivedCallback(IncomingMessageArrived); - } - - private void OnConnectionClosed(ITcpConnection connection, SocketError socketError) - { - connection.ConnectionClosed -= OnConnectionClosed; - - var handler = ConnectionClosed; - if (handler != null) - { - handler(this, socketError); - } - } - - public void EnqueueSend(T message) - { - var data = _formatter.ToArraySegment(message); - _connection.EnqueueSend(_framer.FrameData(data)); - } - - public void ReceiveAsync(Action, T> callback) - { - if (_receiveCallback != null) - { - throw new InvalidOperationException("ReceiveAsync should be called just once."); - } - - if (callback == null) - { - throw new ArgumentNullException("callback"); - } - - _receiveCallback = callback; - - _connection.ReceiveAsync(OnRawDataReceived); - } - - private void OnRawDataReceived(ITcpConnection connection, IEnumerable> data) - { - try - { - _framer.UnFrameData(data); - } - catch (PackageFramingException exc) - { - Log.Information(exc, "Invalid TCP frame received."); - Close("Invalid TCP frame received."); - return; - } - - connection.ReceiveAsync(OnRawDataReceived); - } - - private void IncomingMessageArrived(ArraySegment message) - { - _receiveCallback(this, _formatter.From(message)); - } - - public void Close(string reason = null) - { - _connection.Close(reason); - } -} diff --git a/src/EventStore.sln b/src/EventStore.sln index ca8cbfa04b..c8b3528ac6 100644 --- a/src/EventStore.sln +++ b/src/EventStore.sln @@ -19,8 +19,6 @@ Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "EventStore.TestClient", "Ev EndProject Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "EventStore.Core", "EventStore.Core\EventStore.Core.csproj", "{9957CA10-8A8A-419F-92BD-7A8BE3014345}" EndProject -Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "EventStore.Transport.Tcp", "EventStore.Transport.Tcp\EventStore.Transport.Tcp.csproj", "{DCAD0184-4AFE-49EA-BE31-5D68CA4B59CA}" -EndProject Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "EventStore.ClusterNode", "EventStore.ClusterNode\EventStore.ClusterNode.csproj", "{FDFCE363-84A4-4A6F-B97C-D9F57D8D124F}" EndProject Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "Solution Items", "Solution Items", "{A19631BB-65CF-4726-8732-4EFD977E9B32}" @@ -171,18 +169,6 @@ Global {9957CA10-8A8A-419F-92BD-7A8BE3014345}.Release|ARM64.Build.0 = Release|ARM64 {9957CA10-8A8A-419F-92BD-7A8BE3014345}.Release|x64.ActiveCfg = Release|x64 {9957CA10-8A8A-419F-92BD-7A8BE3014345}.Release|x64.Build.0 = Release|x64 - {DCAD0184-4AFE-49EA-BE31-5D68CA4B59CA}.Debug|Any CPU.ActiveCfg = Debug|Any CPU - {DCAD0184-4AFE-49EA-BE31-5D68CA4B59CA}.Debug|Any CPU.Build.0 = Debug|Any CPU - {DCAD0184-4AFE-49EA-BE31-5D68CA4B59CA}.Debug|ARM64.ActiveCfg = Debug|ARM64 - {DCAD0184-4AFE-49EA-BE31-5D68CA4B59CA}.Debug|ARM64.Build.0 = Debug|ARM64 - {DCAD0184-4AFE-49EA-BE31-5D68CA4B59CA}.Debug|x64.ActiveCfg = Debug|x64 - {DCAD0184-4AFE-49EA-BE31-5D68CA4B59CA}.Debug|x64.Build.0 = Debug|x64 - {DCAD0184-4AFE-49EA-BE31-5D68CA4B59CA}.Release|Any CPU.ActiveCfg = Release|Any CPU - {DCAD0184-4AFE-49EA-BE31-5D68CA4B59CA}.Release|Any CPU.Build.0 = Release|Any CPU - {DCAD0184-4AFE-49EA-BE31-5D68CA4B59CA}.Release|ARM64.ActiveCfg = Release|ARM64 - {DCAD0184-4AFE-49EA-BE31-5D68CA4B59CA}.Release|ARM64.Build.0 = Release|ARM64 - {DCAD0184-4AFE-49EA-BE31-5D68CA4B59CA}.Release|x64.ActiveCfg = Release|x64 - {DCAD0184-4AFE-49EA-BE31-5D68CA4B59CA}.Release|x64.Build.0 = Release|x64 {FDFCE363-84A4-4A6F-B97C-D9F57D8D124F}.Debug|Any CPU.ActiveCfg = Debug|Any CPU {FDFCE363-84A4-4A6F-B97C-D9F57D8D124F}.Debug|Any CPU.Build.0 = Debug|Any CPU {FDFCE363-84A4-4A6F-B97C-D9F57D8D124F}.Debug|ARM64.ActiveCfg = Debug|ARM64