Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
10 changes: 10 additions & 0 deletions src/GameFrameX.SuperSocket.Connection/Sockets/SocketSender.cs
Original file line number Diff line number Diff line change
Expand Up @@ -166,8 +166,18 @@ public void OnCompleted(Action<object> continuation, object state, short token,
/// Attempts to reset the state of the sender.
/// </summary>
/// <returns><c>true</c> if the state was successfully reset; otherwise, <c>false</c>.</returns>
/// <remarks>
/// <see cref="DefaultObjectPool{T}"/> invokes this when an instance is returned to the pool.
/// Besides the buffer, the <see cref="IValueTaskSource"/> continuation and
/// <see cref="SocketAsyncEventArgs.UserToken"/> must also be cleared; otherwise residual state
/// from a previous send makes the next caller's <see cref="GetStatus"/> misjudge the operation
/// as completed, leaving the real native send unobserved and corrupting the overlapped state.
/// </remarks>
public bool TryReset()
{
_continuation = null;
UserToken = null;

if (BufferList != null)
{
BufferList = null;
Expand Down
50 changes: 30 additions & 20 deletions src/GameFrameX.SuperSocket.Connection/TcpPipeConnection.cs
Original file line number Diff line number Diff line change
Expand Up @@ -14,6 +14,8 @@ public class TcpPipeConnection : PipeConnection

private readonly ObjectPool<SocketSender> _socketSenderPool;

private SocketSender _socketSender;

/// <summary>
/// Initializes a new instance of the <see cref="TcpPipeConnection"/> class with the specified socket, options, and socket sender pool.
/// </summary>
Expand All @@ -28,13 +30,38 @@ public TcpPipeConnection(Socket socket, ConnectionOptions options, ObjectPool<So
LocalEndPoint = socket.LocalEndPoint;

_socketSenderPool = socketSenderPool;

// Acquire a dedicated sender for the lifetime of this connection. SocketAsyncEventArgs is bound
// to a single in-flight operation and must not be shared across connections concurrently; the
// previous per-send Get/Return let different connections reuse one instance and corrupted the
// native overlapped, surfacing as the callBack null terminating crash.
_socketSender = socketSenderPool?.Get() ?? new SocketSender();
}

/// <summary>
/// Handles the closure of the connection.
/// </summary>
protected override void OnClosed()
{
var socketSender = _socketSender;
_socketSender = null;

if (socketSender != null)
{
var pool = _socketSenderPool;

if (pool != null)
{
// Returning triggers the now-complete TryReset, clearing any residual
// IValueTaskSource state before the next connection reuses the instance.
pool.Return(socketSender);
}
else
{
socketSender.Dispose();
}
}

_socket = null;
base.OnClosed();
}
Expand Down Expand Up @@ -66,26 +93,9 @@ private async ValueTask<int> ReceiveAsync(Socket socket, Memory<byte> memory, So
/// <returns>The total number of bytes sent.</returns>
protected override async ValueTask<int> SendOverIOAsync(ReadOnlySequence<byte> buffer, CancellationToken cancellationToken)
{
var socketSenderPool = _socketSenderPool;

var socketSender = socketSenderPool?.Get() ?? new SocketSender();

try
{
var sentBytes = await socketSender.SendAsync(_socket, buffer).ConfigureAwait(false);

if (socketSenderPool != null)
{
socketSenderPool.Return(socketSender);
socketSender = null;
}

return sentBytes;
}
finally
{
socketSender?.Dispose();
}
// The sender is exclusive to this connection (see constructor). It is no longer
// borrowed/returned per send, so SAEA instances are never shared across connections.
return await _socketSender.SendAsync(_socket, buffer).ConfigureAwait(false);
}

/// <summary>
Expand Down