Skip to content

Add async send queue and write timeout #626

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
AngeloTadeucci merged 1 commit into from
Feb 4, 2026
Merged

Add async send queue and write timeout #626

Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
89 changes: 80 additions & 9 deletions Maple2.Server.Core/Network/Session.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,6 +26,9 @@ public abstract class Session : IDisposable {
private const int HANDSHAKE_SIZE = 19;
private const int STOP_TIMEOUT = 2000;

// Send timeout to prevent indefinite blocking (in milliseconds)
private const int SEND_TIMEOUT_MS = 5000;

public SessionState State { get; set; }

public EventHandler<string>? OnError;
Expand All @@ -47,6 +50,9 @@ public abstract class Session : IDisposable {
private readonly QueuedPipeScheduler pipeScheduler;
private readonly Pipe recvPipe;

// Send queue for non-blocking sends
private readonly BlockingCollection<(byte[] packet, int length)> sendQueue = new(new ConcurrentQueue<(byte[], int)>());
Comment on lines +53 to +54
Copy link
Contributor

@coderabbitai coderabbitai bot Feb 3, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue | 🟡 Minor

BlockingCollection is not disposed, causing potential resource leak.

BlockingCollection<T> implements IDisposable and should be disposed in Dispose(bool). While CompleteAdding() is called, the underlying resources are not released.

Additionally, the coding guidelines recommend using System.IO.Pipelines for high-performance async I/O in the Session networking layer. Consider using a Pipe for the send path similar to recvPipe, which would provide better integration with the existing pipeline architecture.

🔧 Proposed fix to dispose sendQueue 
     protected virtual void Dispose(bool disposing) {
         if (disposed) return;

         disposed = true;
         State = SessionState.Disconnected;
         try {
             Complete();
         } catch (Exception ex) {
             Logger.Debug(ex, "Complete() threw during Dispose");
         }
+        try {
+            sendQueue.Dispose();
+        } catch (Exception ex) {
+            Logger.Debug(ex, "sendQueue.Dispose() failed");
+        }
         try {
             thread.Join(STOP_TIMEOUT);

As per coding guidelines: "Use System.IO.Pipelines for high-performance async I/O in Session networking layer"

🤖 Prompt for AI Agents 
In `@Maple2.Server.Core/Network/Session.cs` around lines 53 - 54, The sendQueue
BlockingCollection<(byte[] packet, int length)> is not disposed, causing a
resource leak; update the Session.Dispose(bool) implementation to call
sendQueue.CompleteAdding() (if not already) and then sendQueue.Dispose() to
release resources, and ensure any producers/consumers handle
ObjectDisposedException accordingly; additionally, consider refactoring the send
path to use a System.IO.Pipelines.Pipe (similar to the existing recvPipe) by
replacing sendQueue usage in the send loop and producers with a
PipeWriter/PipeReader pair for higher-performance async I/O and better
integration with the Session networking pipeline.


public long AccountId { get; protected set; }
public long CharacterId { get; protected set; }
private readonly ConcurrentDictionary<SendOp, byte[]> lastSentPackets = [];
Expand Down Expand Up @@ -77,6 +83,13 @@ protected Session(TcpClient tcpClient) {
networkStream = tcpClient.GetStream();
sendCipher = new MapleCipher.Encryptor(VERSION, siv, BLOCK_IV);
recvCipher = new MapleCipher.Decryptor(VERSION, riv, BLOCK_IV);

// Start send worker thread
var sendWorkerThread1 = new Thread(SendWorker) {
Name = $"SendWorker-{name}",
IsBackground = true,
};
sendWorkerThread1.Start();
Comment on lines +86 to +92
Copy link
Contributor

@coderabbitai coderabbitai bot Feb 3, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue | 🟡 Minor

Send worker thread is not joined during disposal.

The Dispose method joins the main thread but not sendWorkerThread1. This could leave the send worker running briefly after disposal completes. Consider storing the thread reference and joining it in Dispose.

🔧 Proposed fix to track and join send worker

Add field:

     private readonly Thread thread;
+    private readonly Thread sendWorkerThread;
     private readonly QueuedPipeScheduler pipeScheduler;

Update constructor:

-        var sendWorkerThread1 = new Thread(SendWorker) {
+        sendWorkerThread = new Thread(SendWorker) {
             Name = $"SendWorker-{name}",
             IsBackground = true,
         };
-        sendWorkerThread1.Start();
+        sendWorkerThread.Start();

Add join in Dispose(bool):

         try {
             thread.Join(STOP_TIMEOUT);
         } catch (Exception ex) {
             Logger.Debug(ex, "thread.Join failed");
         }
+        try {
+            sendWorkerThread.Join(STOP_TIMEOUT);
+        } catch (Exception ex) {
+            Logger.Debug(ex, "sendWorkerThread.Join failed");
+        }
📝 Committable suggestion

‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.

Suggested change
// Start send worker thread
var sendWorkerThread1 = new Thread(SendWorker) {
Name = $"SendWorker-{name}",
IsBackground = true,
};
sendWorkerThread1.Start();
// Start send worker thread
sendWorkerThread = new Thread(SendWorker) {
Name = $"SendWorker-{name}",
IsBackground = true,
};
sendWorkerThread.Start();
🤖 Prompt for AI Agents 
In `@Maple2.Server.Core/Network/Session.cs` around lines 86 - 92, The SendWorker
thread started as sendWorkerThread1 is not stored or joined on disposal causing
it to potentially run after Dispose completes; update the class to store the
thread as a field (e.g., private Thread _sendWorkerThread), assign it when
creating the thread in the constructor where SendWorker is started, and in
Dispose(bool) ensure you check _sendWorkerThread != null and call Join (or
interrupt then Join) similar to how the main thread field is handled so the send
worker is properly stopped before disposal returns.

}

~Session() => Dispose(false);
Expand Down Expand Up @@ -112,13 +125,14 @@ protected void Complete() {
recvPipe.Writer.Complete();
recvPipe.Reader.Complete();
pipeScheduler.Complete();
sendQueue.CompleteAdding();
}

public void Disconnect([CallerMemberName] string caller = "", [CallerLineNumber] int line = 0, [CallerFilePath] string filePath = "") {
if (disposed) return;

Logger.Information("Disconnected {Session} at {Caller} in {FilePath} on line {LineNumber}", this, caller, filePath, line);
if (Interlocked.Exchange(ref disconnecting, 1) == 1) return;
Logger.Information("Disconnected {Session} at {Caller} in {FilePath} on line {LineNumber}", this, caller, filePath, line);
Dispose();
}

Expand Down Expand Up @@ -159,6 +173,8 @@ private void StartInternal() {
Task.WhenAll(writeTask, readTask).ContinueWith(t => {
if (t.IsFaulted) {
Logger.Debug(t.Exception, "Pipeline aggregate fault account={AccountId} char={CharacterId}", AccountId, CharacterId);
} else if (t.IsCanceled) {
Logger.Debug("Pipeline tasks cancelled account={AccountId} char={CharacterId}", AccountId, CharacterId);
}
CloseClient();
});
Expand Down Expand Up @@ -202,7 +218,10 @@ private async Task WriteRecvPipe(Socket socket, PipeWriter writer) {

result = await writer.FlushAsync();
} while (!disposed && !result.IsCompleted);
} catch (Exception ex) { Logger.Debug(ex, "WriteRecvPipe exception account={AccountId} char={CharacterId}", AccountId, CharacterId); Disconnect(); }
} catch (Exception ex) {
Logger.Debug(ex, "WriteRecvPipe exception account={AccountId} char={CharacterId}", AccountId, CharacterId);
Disconnect();
}
}

private async Task ReadRecvPipe(PipeReader reader) {
Expand Down Expand Up @@ -262,24 +281,74 @@ private void SendInternal(byte[] packet, int length) {
lastSentPackets[op] = packet.Take(length).ToArray();
}

lock (sendCipher) {
// re-check after potential delay acquiring lock
if (disposed || disconnecting == 1) return;
using PoolByteWriter encryptedPacket = sendCipher.Encrypt(packet, 0, length);
SendRaw(encryptedPacket);
// Queue the raw packet for background processing
// Make a copy since the caller may reuse the buffer
byte[] packetCopy = packet.Take(length).ToArray();
try {
sendQueue.Add((packetCopy, length));
} catch (InvalidOperationException) {
// Queue was completed/disposed
Logger.Debug("SendQueue add failed - queue completed");
}
}

private void SendRaw(ByteWriter packet) {
if (disposed || disconnecting == 1) return;

try {
networkStream.Write(packet.Buffer, 0, packet.Length);
// Use async write with timeout to prevent indefinite blocking
Task writeTask = networkStream.WriteAsync(packet.Buffer, 0, packet.Length);
if (!writeTask.Wait(SEND_TIMEOUT_MS)) {
Logger.Warning("SendRaw timeout after {Timeout}ms, disconnecting account={AccountId} char={CharacterId}",
SEND_TIMEOUT_MS, AccountId, CharacterId);

// Observe the task exception to prevent unobserved task exception
// when the task eventually completes/faults after timeout
_ = writeTask.ContinueWith(t => {
if (t.IsFaulted) {
Logger.Debug(t.Exception, "WriteAsync faulted after timeout account={AccountId} char={CharacterId}",
AccountId, CharacterId);
}
}, TaskContinuationOptions.OnlyOnFaulted);

Disconnect();
return;
}

// Check if write actually failed
if (writeTask.IsFaulted) {
throw writeTask.Exception?.GetBaseException() ?? new Exception("Write task faulted");
}
} catch (Exception ex) {
Logger.Debug(ex, "[LIFECYCLE] SendRaw write failed account={AccountId} char={CharacterId}", AccountId, CharacterId);
Logger.Warning(ex, "[LIFECYCLE] SendRaw write failed account={AccountId} char={CharacterId}", AccountId, CharacterId);
Disconnect();
}
}

private void SendWorker() {
try {
foreach ((byte[] packet, int length) in sendQueue.GetConsumingEnumerable()) {
if (disposed || disconnecting == 1) break;

// Encrypt outside lock, then send with timeout
PoolByteWriter encryptedPacket;
lock (sendCipher) {
if (disposed || disconnecting == 1) break;
encryptedPacket = sendCipher.Encrypt(packet, 0, length);
}
try {
SendRaw(encryptedPacket);
} finally {
encryptedPacket.Dispose();
}
}
} catch (Exception ex) {
if (!disposed) {
Logger.Error(ex, "SendWorker exception account={AccountId} char={CharacterId}", AccountId, CharacterId);
}
}
}

public byte[]? GetLastSentPacket(SendOp op) {
lastSentPackets.TryGetValue(op, out byte[]? packet);
return packet;
Expand All @@ -304,6 +373,7 @@ private void LogSend(byte[] packet, int length) {
case SendOp.FurnishingInventory:
case SendOp.FurnishingStorage:
case SendOp.Vibrate:
case SendOp.Insignia:
break;
default:
Logger.Verbose("{Mode} ({Name} - {OpCode}): {Packet}", "SEND".ColorRed(), opcode, $"0x{op:X4}", packet.ToHexString(length, ' '));
Expand All @@ -320,6 +390,7 @@ private void LogRecv(byte[] packet) {
case RecvOp.GuideObjectSync:
case RecvOp.RideSync:
case RecvOp.ResponseHeartbeat:
case RecvOp.Insignia:
break;
default:
Logger.Verbose("{Mode} ({Name} - {OpCode}): {Packet}", "RECV".ColorGreen(), opcode, $"0x{op:X4}", packet.ToHexString(packet.Length, ' '));
Expand Down
Loading