diff --git a/src/KubernetesClient.Classic/CancellationExtensions.cs b/src/KubernetesClient.Classic/CancellationExtensions.cs new file mode 100644 index 000000000..d7f60cca6 --- /dev/null +++ b/src/KubernetesClient.Classic/CancellationExtensions.cs @@ -0,0 +1,56 @@ +namespace k8s +{ + /// + /// Polyfills for the cancellable overloads which are only available on modern .NET, + /// so the shared sources can rely on them unconditionally. + /// + internal static class CancellationExtensions + { + /// + /// Polyfill of Task<TResult>.WaitAsync(CancellationToken). + /// + public static Task WaitAsync(this Task task, CancellationToken cancellationToken) + { + if (task == null) + { + throw new ArgumentNullException(nameof(task)); + } + + if (task.IsCompleted || !cancellationToken.CanBeCanceled) + { + return task; + } + + if (cancellationToken.IsCancellationRequested) + { + return Task.FromCanceled(cancellationToken); + } + + // Observe any exception from the original task to prevent an + // UnobservedTaskException when the continuation below is cancelled + // before the original task faults (e.g. the transport tears down the + // connection after cancellation). + _ = task.ContinueWith( + static t => { _ = t.Exception; }, + CancellationToken.None, + TaskContinuationOptions.OnlyOnFaulted | TaskContinuationOptions.ExecuteSynchronously, + TaskScheduler.Default); + + // here to pass cancellationToken into task + return task.ContinueWith(t => t.GetAwaiter().GetResult(), cancellationToken); + } + + /// + /// Polyfill of TextReader.ReadLineAsync(CancellationToken). + /// + public static Task ReadLineAsync(this TextReader reader, CancellationToken cancellationToken) + { + if (reader == null) + { + throw new ArgumentNullException(nameof(reader)); + } + + return reader.ReadLineAsync().WaitAsync(cancellationToken); + } + } +} diff --git a/src/KubernetesClient.Classic/KubernetesClient.Classic.csproj b/src/KubernetesClient.Classic/KubernetesClient.Classic.csproj index 902dc41dd..8980f5124 100644 --- a/src/KubernetesClient.Classic/KubernetesClient.Classic.csproj +++ b/src/KubernetesClient.Classic/KubernetesClient.Classic.csproj @@ -107,7 +107,6 @@ - diff --git a/src/KubernetesClient.Classic/LineSeparatedHttpContent.cs b/src/KubernetesClient.Classic/LineSeparatedHttpContent.cs new file mode 100644 index 000000000..9206fe4ff --- /dev/null +++ b/src/KubernetesClient.Classic/LineSeparatedHttpContent.cs @@ -0,0 +1,211 @@ +using System.Net; +using System.Net.Http; + +namespace k8s +{ + internal sealed class LineSeparatedHttpContent : HttpContent + { + private readonly HttpContent _originContent; + private readonly CancellationToken _cancellationToken; + private Stream _originStream; + + public LineSeparatedHttpContent(HttpContent originContent, CancellationToken cancellationToken) + { + _originContent = originContent; + _cancellationToken = cancellationToken; + } + + public TextReader StreamReader { get; private set; } + + protected override async Task SerializeToStreamAsync(Stream stream, TransportContext context) + { + _originStream = await _originContent.ReadAsStreamAsync().ConfigureAwait(false); + + var reader = new PeekableStreamReader(new CancelableStream(_originStream, _cancellationToken)); + StreamReader = reader; + + var firstLine = await reader.PeekLineAsync().ConfigureAwait(false); + + var writer = new StreamWriter(stream); + + await writer.WriteAsync(firstLine).ConfigureAwait(false); + await writer.FlushAsync().ConfigureAwait(false); + } + + protected override bool TryComputeLength(out long length) + { + length = 0; + return false; + } + + internal sealed class CancelableStream : Stream + { + private readonly Stream _innerStream; + private readonly CancellationToken _cancellationToken; + + public CancelableStream(Stream innerStream, CancellationToken cancellationToken) + { + _innerStream = innerStream; + _cancellationToken = cancellationToken; + } + + public override void Flush() => + _innerStream.FlushAsync(_cancellationToken).GetAwaiter().GetResult(); + + public override async Task FlushAsync(CancellationToken cancellationToken) + { + using (var cancellationTokenSource = CreateCancellationTokenSource(cancellationToken)) + { + await _innerStream.FlushAsync(cancellationTokenSource.Token).ConfigureAwait(false); + } + } + + public override int Read(byte[] buffer, int offset, int count) => + _innerStream.ReadAsync(buffer, offset, count, _cancellationToken).GetAwaiter().GetResult(); + + public override async Task ReadAsync(byte[] buffer, int offset, int count, + CancellationToken cancellationToken) + { + using (var cancellationTokenSource = CreateCancellationTokenSource(cancellationToken)) + { + return await _innerStream.ReadAsync(buffer, offset, count, cancellationTokenSource.Token) + .ConfigureAwait(false); + } + } + + public override long Seek(long offset, SeekOrigin origin) => _innerStream.Seek(offset, origin); + + public override void SetLength(long value) => _innerStream.SetLength(value); + + public override void Write(byte[] buffer, int offset, int count) => + _innerStream.WriteAsync(buffer, offset, count, _cancellationToken).GetAwaiter().GetResult(); + + public override async Task WriteAsync(byte[] buffer, int offset, int count, + CancellationToken cancellationToken) + { + using (var cancellationTokenSource = CreateCancellationTokenSource(cancellationToken)) + { + await _innerStream.WriteAsync(buffer, offset, count, cancellationTokenSource.Token) + .ConfigureAwait(false); + } + } + + public override bool CanRead => _innerStream.CanRead; + + public override bool CanSeek => _innerStream.CanSeek; + + public override bool CanWrite => _innerStream.CanWrite; + + public override long Length => _innerStream.Length; + + public override long Position + { + get => _innerStream.Position; + set => _innerStream.Position = value; + } + + protected override void Dispose(bool disposing) + { + if (disposing) + { + _innerStream.Dispose(); + } + + base.Dispose(disposing); + } + + private LinkedCancellationTokenSource CreateCancellationTokenSource(CancellationToken userCancellationToken) + { + return new LinkedCancellationTokenSource(_cancellationToken, userCancellationToken); + } + + private readonly struct LinkedCancellationTokenSource : IDisposable + { + private readonly CancellationTokenSource _cancellationTokenSource; + + public LinkedCancellationTokenSource(CancellationToken token1, CancellationToken token2) + { + if (token1.CanBeCanceled && token2.CanBeCanceled) + { + _cancellationTokenSource = CancellationTokenSource.CreateLinkedTokenSource(token1, token2); + Token = _cancellationTokenSource.Token; + } + else + { + _cancellationTokenSource = null; + Token = token1.CanBeCanceled ? token1 : token2; + } + } + + public CancellationToken Token { get; } + + public void Dispose() + { + _cancellationTokenSource?.Dispose(); + } + } + } + + internal sealed class PeekableStreamReader : TextReader + { + private readonly Queue _buffer; + private readonly StreamReader _inner; + + public PeekableStreamReader(Stream stream) + { + _buffer = new Queue(); + _inner = new StreamReader(stream); + } + + public override string ReadLine() => throw new NotImplementedException(); + + public override Task ReadLineAsync() + { + if (_buffer.Count > 0) + { + return Task.FromResult(_buffer.Dequeue()); + } + + return _inner.ReadLineAsync(); + } + + public async Task PeekLineAsync() + { + var line = await ReadLineAsync().ConfigureAwait(false); + if (line == null) + { + throw new EndOfStreamException(); + } + + _buffer.Enqueue(line); + return line; + } + + public override int Read() => throw new NotImplementedException(); + + public override int Read(char[] buffer, int index, int count) => throw new NotImplementedException(); + + public override Task ReadAsync(char[] buffer, int index, int count) => + throw new NotImplementedException(); + + public override int ReadBlock(char[] buffer, int index, int count) => throw new NotImplementedException(); + + public override Task ReadBlockAsync(char[] buffer, int index, int count) => + throw new NotImplementedException(); + + public override string ReadToEnd() => throw new NotImplementedException(); + + public override Task ReadToEndAsync() => throw new NotImplementedException(); + + protected override void Dispose(bool disposing) + { + if (disposing) + { + _inner.Dispose(); + } + + base.Dispose(disposing); + } + } + } +} diff --git a/src/KubernetesClient/Watcher.cs b/src/KubernetesClient/Watcher.cs index 23868d4e0..1008c640a 100644 --- a/src/KubernetesClient/Watcher.cs +++ b/src/KubernetesClient/Watcher.cs @@ -158,23 +158,12 @@ private async Task WatcherLoop(CancellationToken cancellationToken) Action onError = null, [EnumeratorCancellation] CancellationToken cancellationToken = default) { - Task AttachCancellationToken(Task task) - { - if (!task.IsCompleted) - { - // here to pass cancellationToken into task - return task.ContinueWith(t => t.GetAwaiter().GetResult(), cancellationToken); - } - - return task; - } - - using var streamReader = await AttachCancellationToken(streamReaderCreator()).ConfigureAwait(false); + using var streamReader = await streamReaderCreator().WaitAsync(cancellationToken).ConfigureAwait(false); for (; ; ) { // ReadLineAsync will return null when we've reached the end of the stream. - var line = await AttachCancellationToken(streamReader.ReadLineAsync()).ConfigureAwait(false); + var line = await streamReader.ReadLineAsync(cancellationToken).ConfigureAwait(false); cancellationToken.ThrowIfCancellationRequested(); diff --git a/tests/KubernetesClient.Tests/WatchTests.cs b/tests/KubernetesClient.Tests/WatchTests.cs index 53259d770..fe314b3c6 100644 --- a/tests/KubernetesClient.Tests/WatchTests.cs +++ b/tests/KubernetesClient.Tests/WatchTests.cs @@ -1028,5 +1028,70 @@ public async Task AsyncEnumerableWatchErrorHandling() Assert.True(watchCompleted.IsSet); } } + + [Fact] + public async Task CancellationDoesNotLeaveUnobservedTaskException() + { + // Regression test for https://github.com/kubernetes-client/csharp/issues/1813 + // When the cancellation token is cancelled while a read is in flight and the + // underlying task subsequently faults (e.g. transport-level IOException after the + // connection is torn down), the faulting task must be observed so that no + // TaskScheduler.UnobservedTaskException is raised when it is finalized. + var unobservedExceptions = new List(); + void Handler(object sender, UnobservedTaskExceptionEventArgs e) + { + unobservedExceptions.Add(e.Exception); + } + + TaskScheduler.UnobservedTaskException += Handler; + try + { + // Run the cancellation scenario in a separate, non-inlined method so that all + // references to the orphaned task go out of scope before we force a collection. + await RunCancelledWatchAsync().ConfigureAwait(true); + + // Force the orphaned task to be finalized; without observing its exception this + // would raise TaskScheduler.UnobservedTaskException. + for (var i = 0; i < 5; i++) + { + GC.Collect(); + GC.WaitForPendingFinalizers(); + await Task.Delay(50).ConfigureAwait(true); + } + + Assert.Empty(unobservedExceptions); + } + finally + { + TaskScheduler.UnobservedTaskException -= Handler; + } + } + + [System.Runtime.CompilerServices.MethodImpl(System.Runtime.CompilerServices.MethodImplOptions.NoInlining)] + private static async Task RunCancelledWatchAsync() + { + using var cts = new CancellationTokenSource(); + var faultReader = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + + Func> streamReaderCreator = () => faultReader.Task; + + var enumerator = Watcher + .CreateWatchEventEnumerator(streamReaderCreator, onError: null, cancellationToken: cts.Token) + .GetAsyncEnumerator(cts.Token); + + // Start the enumeration; this awaits the (not yet completed) creator task. + var moveNext = enumerator.MoveNextAsync(); + + // Cancel before the creator task completes. The cancellation-aware continuation + // is cancelled, but the original creator task is still pending. + cts.Cancel(); + + await Assert.ThrowsAnyAsync(async () => await moveNext.ConfigureAwait(true)).ConfigureAwait(true); + + await enumerator.DisposeAsync().ConfigureAwait(true); + + // Now fault the original creator task, mimicking the transport tear-down. + faultReader.SetException(new IOException("The request was aborted.")); + } } }