Why websocket deadlocks when reading and writing in parallel?

[zxyao145]

Hi, I am writing a K8s web terminal that receives ws requests from the web end and then proxies them to K8s through a wsclient. If I don't use a channel in memory to relay requests from the web, a deadlock will occur. Why is this phenomenon happening?

My architecture model is as follows:

And my pseudo code is as follows:

Invalid code：

WebSocket webSocket;
WebSocket podWebSocket；

var forwardToPod = Task.Run(async () =>
{
    var clientBuf = new byte[2048];
    while (webSocket.State == WebSocketState.Open)
    {
        WebSocketReceiveResult result = await webSocket.ReceiveAsync(clientBuf, CancellationToken.None);
        var data = ConvertFromWebInput(clientBuf, result);
        await podWebSocket.SendAsync(data, WebSocketMessageType.Text, WebSocketMessageFlags.EndOfMessage,
            CancellationToken.None);
    }
});


var forwardToWeb = Task.Run(async () =>
{
    var clientBuf = new byte[2048];
    while (podWebSocket.State == WebSocketState.Open)
    {
        WebSocketReceiveResult result = await podWebSocket.ReceiveAsync(clientBuf, CancellationToken.None);
        var data = ConvertToWebOutput(clientBuf, result);
        await webSocket.SendAsync(data, WebSocketMessageType.Text, WebSocketMessageFlags.EndOfMessage, CancellationToken.None);
    }
});

await Task.WhenAny(forwardToPod, forwardToWeb);

Valid code：

WebSocket webSocket;
WebSocket podWebSocket;

var chInput = Channel.CreateUnbounded<byte[]>();
var chOutput = Channel.CreateUnbounded<byte[]>();


var forwardInToMem = Task.Run(async () =>
{
    var clientBuf = new byte[2048];
    while (webSocket.State == WebSocketState.Open)
    {
        WebSocketReceiveResult result = await webSocket.ReceiveAsync(clientBuf, CancellationToken.None);
        await chInput.Writer.WriteAsync(clientBuf.Take(result.Count));
    }
});
var forwardMemToPod = Task.Run(async () =>
{
    while (await chInput.Reader.WaitToReadAsync())
    {
        if (chInput.Reader.TryRead(out var buffer))
        {
            var data = ConvertFromWebInput(buffer);
            await podWebSocket.SendAsync(data, WebSocketMessageType.Text, WebSocketMessageFlags.EndOfMessage,
                CancellationToken.None);
        }
    }
});


var forwardPodToMem = Task.Run(async () =>
{
    var clientBuf = new byte[2048];
    while (podWebSocket.State == WebSocketState.Open)
    {
        WebSocketReceiveResult result = await podWebSocket.ReceiveAsync(clientBuf, CancellationToken.None);
        await chOutput.Writer.WriteAsync(clientBuf.Take(result.Count));
        
    }
});
var forwardMemToWeb = Task.Run(async () =>
{
    while (await chOutput.Reader.WaitToReadAsync())
    {
        if (chOutput.Reader.TryRead(out var buffer))
        {
            var data = ConvertToWebOutput(clientBuf, result);
            await webSocket.SendAsync(data, WebSocketMessageType.Text, WebSocketMessageFlags.EndOfMessage, CancellationToken.None);
        }
    }
});

await Task.WhenAny(forwardInToMem, forwardPodToMem);

[zxyao145]

I have previously written code for forwarding data between two NetworkStreams, and this issue will not occur. Is there anything special about websocket?

async Task ForwardData(NetworkStream source, NetworkStream destination, CancellationToken? token = null)
{
    token ??= CancellationToken.None;
    byte[] buffer = new byte[4096];
    int bytesRead;
    while ((bytesRead = await source.ReadAsync(buffer, token.Value)) > 0)
    {
        await destination.WriteAsync(
            new ReadOnlyMemory<byte>(buffer, 0, bytesRead),
            token.Value
        );
    }
}


NetworkStream  src;
NetworkStream  dist;

var forwardToRemote = ForwardData(src, dist);
var forwardToClient = ForwardData(dist, src)

await Task.WhenAny(forwardToRemote, forwardToClient);

[davidfowl]

It's legal and should work if you have concurrent reads and writes with the websocket. Can you share a repro? Where does podWebSocket come from?

[zxyao145]

Sorry, I am unable to reproduce it at the moment. If there is any progress, I will synchronize here.