MuxedStream.ReadAsync ignores the cancellation token

[freddyrios]

While investigating rare / hard to reproduce hangs, I noticed that MuxedStream.ReadAsync ignores the passed cancellation token. In the related StreamDemuxer class, the reads are done to the underlying web socket by passing a cancellation token that is only related to the disposal of the StreamDemuxer, so there does not seem to be a clear alternative to garantee the reads will not result in a hang in rare cases.

[tg123]

good catch, would you like to send the pr

[k8s-triage-robot]

The Kubernetes project currently lacks enough contributors to adequately respond to all issues.

This bot triages un-triaged issues according to the following rules:

• After 90d of inactivity, lifecycle/stale is applied
• After 30d of inactivity since lifecycle/stale was applied, lifecycle/rotten is applied
• After 30d of inactivity since lifecycle/rotten was applied, the issue is closed

You can:

• Mark this issue as fresh with /remove-lifecycle stale
• Close this issue with /close
• Offer to help out with Issue Triage

Please send feedback to sig-contributor-experience at kubernetes/community.

/lifecycle stale

[k8s-triage-robot]

The Kubernetes project currently lacks enough active contributors to adequately respond to all issues.

This bot triages un-triaged issues according to the following rules:

• After 90d of inactivity, lifecycle/stale is applied
• After 30d of inactivity since lifecycle/stale was applied, lifecycle/rotten is applied
• After 30d of inactivity since lifecycle/rotten was applied, the issue is closed

You can:

• Mark this issue as fresh with /remove-lifecycle rotten
• Close this issue with /close
• Offer to help out with Issue Triage

Please send feedback to sig-contributor-experience at kubernetes/community.

/lifecycle rotten

[freddyrios]

I don't have time to do a PR, but this is the workaround we used for our specific type of usage:

    /// <remarks>
    /// Work arounds the StreamDemuxer's stream bug where it ignores the cancellation token passed to stream.ReadAsync.
    /// When the cancellation token is triggered, the ReadAsync operation is still running underneath, but the caller is expected to Dispose the demuxer which should result in the underlying operation being canceled.
    /// </remarks>
    private static async Task<List<byte>> ReadAllFromDemuxerStream(Stream stream, CancellationToken token)
    {
        var output = new List<byte>();
        var timeoutTask = new TaskCompletionSource();
        using var reg = token.Register(() => timeoutTask.TrySetCanceled());
        var buffer = new byte[4096];
        while (true)
        {
            var readTask = Task.Run(() => stream.ReadAsync(buffer, 0, buffer.Length, token), token);
            if (await Task.WhenAny(readTask, timeoutTask.Task) == timeoutTask.Task)
                throw new OperationCanceledException("Read operation was cancelled.", token);

            var read = await readTask;
            if (read == 0)
                break;

            output.AddRange(buffer.Take(read));
        }

        return output;
    }

Where the caller did this:

        using var demux = new StreamDemuxer(webSocket);
        demux.Start();

        using var stream = demux.GetStream(1, 1);
        var output = await ReadAllFromDemuxerStream(stream, cts.Token);
        return System.Text.Encoding.UTF8.GetString([.. output]);