Skip to content

[wasm] Cancel pending ReadableStream on Dispose or cancelationToken #64285

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
pavelsavara merged 9 commits into from
Jan 29, 2022
Merged

[wasm] Cancel pending ReadableStream on Dispose or cancelationToken #64285

Show file tree
Hide file tree
Changes from 2 commits
Commits
Show all changes
9 commits
Select commit Hold shift + click to select a range
184846d
- call cancel when disposing ReadableStream
pavelsavara Jan 25, 2022
19a8182
- enabled cancelation tests for browser
pavelsavara Jan 26, 2022
3c9b244
fix
pavelsavara Jan 27, 2022
3e2ed32
end to end test
pavelsavara Jan 27, 2022
4a95748
fix
pavelsavara Jan 27, 2022
d2d74df
revert unnecesary test change
pavelsavara Jan 27, 2022
c23f74f
feedback + netfx CI problem
pavelsavara Jan 28, 2022
7508699
feedback
pavelsavara Jan 28, 2022
b074484
Update src/libraries/Common/tests/System/Net/Http/HttpClientHandlerTe…
lewing Jan 28, 2022
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
13 changes: 11 additions & 2 deletions src/libraries/Common/tests/System/Net/Http/HttpClientHandlerTest.Cancellation.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28,6 +28,7 @@ public HttpClientHandler_Cancellation_Test(ITestOutputHelper output) : base(outp
[Theory]
[InlineData(false, CancellationMode.Token)]
[InlineData(true, CancellationMode.Token)]
[ActiveIssue("https://github.com/dotnet/runtime/issues/64333", TestPlatforms.Browser)] // out of memory
public async Task PostAsync_CancelDuringRequestContentSend_TaskCanceledQuickly(bool chunkedTransfer, CancellationMode mode)
{
if (LoopbackServerFactory.Version >= HttpVersion20.Value && chunkedTransfer)
Expand Down Expand Up @@ -226,12 +227,19 @@ await LoopbackServerFactory.CreateServerAsync(async (server, url) =>
}

await connection.ReadRequestDataAsync();
await connection.SendResponseAsync(HttpStatusCode.OK, headers: headers, isFinal: false);
await connection.SendResponseAsync(HttpStatusCode.OK, headers: headers, isFinal: false, content: chunkedTransfer ? "8\r\nTooShort\r\n" : "TooShort");
await clientFinished.Task;
});

var req = new HttpRequestMessage(HttpMethod.Get, url) { Version = UseVersion };
req.Headers.ConnectionClose = connectionClose;

if (PlatformDetection.IsBrowser)
{
var WebAssemblyEnableStreamingResponseKey = new HttpRequestOptionsKey<bool>("WebAssemblyEnableStreamingResponse");
req.Options.Set(WebAssemblyEnableStreamingResponseKey, true);
}

Task<HttpResponseMessage> getResponse = client.SendAsync(TestAsync, req, HttpCompletionOption.ResponseHeadersRead, cts.Token);
await ValidateClientCancellationAsync(async () =>
{
Expand All @@ -247,7 +255,6 @@ await ValidateClientCancellationAsync(async () =>
cts.Cancel();
await readTask;
});

try
{
clientFinished.SetResult(true);
Expand All @@ -256,11 +263,13 @@ await ValidateClientCancellationAsync(async () =>
});
}
}

[Theory]
[InlineData(CancellationMode.CancelPendingRequests, false)]
[InlineData(CancellationMode.DisposeHttpClient, false)]
[InlineData(CancellationMode.CancelPendingRequests, true)]
[InlineData(CancellationMode.DisposeHttpClient, true)]
[SkipOnPlatform(TestPlatforms.Browser, "Browser doesn't have blocking synchronous Stream.ReadByte and so it waits for whole body")]
public async Task GetAsync_CancelPendingRequests_DoesntCancelReadAsyncOnResponseStream(CancellationMode mode, bool copyToAsync)
{
if (IsWinHttpHandler && UseVersion >= HttpVersion20.Value)
Expand Down
98 changes: 56 additions & 42 deletions src/libraries/System.Net.Http/src/System/Net/Http/BrowserHttpHandler/BrowserHttpHandler.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -148,10 +148,10 @@ protected internal override async Task<HttpResponseMessage> SendAsync(HttpReques
{
throw new ArgumentNullException(nameof(request), SR.net_http_handler_norequest);
}

CancellationTokenRegistration? abortRegistration = null;
try
{
var requestObject = new JSObject();
using var requestObject = new JSObject();

if (request.Options.TryGetValue(FetchOptions, out IDictionary<string, object>? fetchOptions))
{
Expand Down Expand Up @@ -221,44 +221,39 @@ protected internal override async Task<HttpResponseMessage> SendAsync(HttpReques
}


WasmHttpReadStream? wasmHttpReadStream = null;

JSObject abortController = new HostObject("AbortController");
JSObject signal = (JSObject)abortController.GetObjectProperty("signal");
using JSObject signal = (JSObject)abortController.GetObjectProperty("signal");
requestObject.SetObjectProperty("signal", signal);
signal.Dispose();

CancellationTokenSource abortCts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
CancellationTokenRegistration abortRegistration = abortCts.Token.Register((Action)(() =>
abortRegistration = cancellationToken.Register(() =>
{
if (abortController.JSHandle != -1)
if (!abortController.IsDisposed)
{
abortController.Invoke("abort");
abortController?.Dispose();
}
wasmHttpReadStream?.Dispose();
abortCts.Dispose();
}));
});

var args = new System.Runtime.InteropServices.JavaScript.Array();
using var args = new System.Runtime.InteropServices.JavaScript.Array();
if (request.RequestUri != null)
{
args.Push(request.RequestUri.ToString());
args.Push(requestObject);
}

requestObject.Dispose();

var response = s_fetch?.Invoke("apply", s_window, args) as Task<object>;
args.Dispose();
if (response == null)
var responseTask = s_fetch?.Invoke("apply", s_window, args) as Task<object>;
if (responseTask == null)
throw new Exception(SR.net_http_marshalling_response_promise_from_fetch);

JSObject t = (JSObject)await response.ConfigureAwait(continueOnCapturedContext: true);
cancellationToken.ThrowIfCancellationRequested();

var status = new WasmFetchResponse(t, abortController, abortCts, abortRegistration);
HttpResponseMessage httpResponse = new HttpResponseMessage((HttpStatusCode)status.Status);
httpResponse.RequestMessage = request;
var fetchResponseJs = (JSObject)await responseTask.ConfigureAwait(continueOnCapturedContext: true);

var fetchResponse = new WasmFetchResponse(fetchResponseJs, abortController, abortRegistration.Value);
abortRegistration = null;
var responseMessage = new HttpResponseMessage((HttpStatusCode)fetchResponse.Status);
responseMessage.RequestMessage = request;

// Here we will set the ReasonPhrase so that it can be evaluated later.
// We do not have a status code but this will signal some type of what happened
Expand All @@ -267,9 +262,9 @@ protected internal override async Task<HttpResponseMessage> SendAsync(HttpReques
// https://developer.mozilla.org/en-US/docs/Web/API/Response/type
// opaqueredirect: The fetch request was made with redirect: "manual".
// The Response's status is 0, headers are empty, body is null and trailer is empty.
if (status.ResponseType == "opaqueredirect")
if (fetchResponse.ResponseType == "opaqueredirect")
{
httpResponse.SetReasonPhraseWithoutValidation(status.ResponseType);
responseMessage.SetReasonPhraseWithoutValidation(fetchResponse.ResponseType);
}

bool streamingEnabled = false;
Expand All @@ -278,9 +273,9 @@ protected internal override async Task<HttpResponseMessage> SendAsync(HttpReques
request.Options.TryGetValue(EnableStreamingResponse, out streamingEnabled);
}

httpResponse.Content = streamingEnabled
? new StreamContent(wasmHttpReadStream = new WasmHttpReadStream(status))
: (HttpContent)new BrowserHttpContent(status);
responseMessage.Content = streamingEnabled
? new StreamContent(new WasmHttpReadStream(fetchResponse))
: new BrowserHttpContent(fetchResponse);

// Fill the response headers
// CORS will only allow access to certain headers.
Expand All @@ -290,7 +285,7 @@ protected internal override async Task<HttpResponseMessage> SendAsync(HttpReques
// View more information https://developers.google.com/web/updates/2015/03/introduction-to-fetch#response_types
//
// Note: Some of the headers may not even be valid header types in .NET thus we use TryAddWithoutValidation
using (JSObject respHeaders = status.Headers)
using (JSObject respHeaders = fetchResponse.Headers)
{
if (respHeaders != null)
{
Expand All @@ -306,8 +301,8 @@ protected internal override async Task<HttpResponseMessage> SendAsync(HttpReques
{
var name = (string)resultValue[0];
var value = (string)resultValue[1];
if (!httpResponse.Headers.TryAddWithoutValidation(name, value))
httpResponse.Content.Headers.TryAddWithoutValidation(name, value);
if (!responseMessage.Headers.TryAddWithoutValidation(name, value))
responseMessage.Content.Headers.TryAddWithoutValidation(name, value);
}
nextResult?.Dispose();
nextResult = (JSObject)entriesIterator.Invoke("next");
Expand All @@ -320,7 +315,7 @@ protected internal override async Task<HttpResponseMessage> SendAsync(HttpReques
}
}
}
return httpResponse;
return responseMessage;

}
catch (OperationCanceledException oce) when (cancellationToken.IsCancellationRequested)
Expand All @@ -331,6 +326,10 @@ protected internal override async Task<HttpResponseMessage> SendAsync(HttpReques
{
throw TranslateJSException(jse, cancellationToken);
}
finally
{
abortRegistration?.Dispose();
}
}

private static Exception TranslateJSException(JSException jse, CancellationToken cancellationToken)
Expand All @@ -350,15 +349,13 @@ private sealed class WasmFetchResponse : IDisposable
{
private readonly JSObject _fetchResponse;
private readonly JSObject _abortController;
private readonly CancellationTokenSource _abortCts;
private readonly CancellationTokenRegistration _abortRegistration;
private bool _isDisposed;

public WasmFetchResponse(JSObject fetchResponse, JSObject abortController, CancellationTokenSource abortCts, CancellationTokenRegistration abortRegistration)
public WasmFetchResponse(JSObject fetchResponse, JSObject abortController, CancellationTokenRegistration abortRegistration)
{
_fetchResponse = fetchResponse ?? throw new ArgumentNullException(nameof(fetchResponse));
_abortController = abortController ?? throw new ArgumentNullException(nameof(abortController));
_abortCts = abortCts;
_abortRegistration = abortRegistration;
}

Expand All @@ -383,10 +380,13 @@ public void Dispose()

_isDisposed = true;

_abortCts.Dispose();
_abortRegistration.Dispose();

_fetchResponse?.Dispose();
if (_abortController != null && !_abortController.IsDisposed)
{
_abortController.Invoke("abort");
}
_abortController?.Dispose();
}
}
Expand Down Expand Up @@ -460,15 +460,15 @@ protected override void Dispose(bool disposing)

private sealed class WasmHttpReadStream : Stream
{
private WasmFetchResponse? _status;
private WasmFetchResponse? _fetchResponse;
private JSObject? _reader;

private byte[]? _bufferedBytes;
private int _position;

public WasmHttpReadStream(WasmFetchResponse status)
public WasmHttpReadStream(WasmFetchResponse fetchResponse)
{
_status = status;
_fetchResponse = fetchResponse;
}

public override bool CanRead => true;
Expand All @@ -489,17 +489,19 @@ public override Task<int> ReadAsync(byte[] buffer, int offset, int count, Cancel

public override async ValueTask<int> ReadAsync(Memory<byte> buffer, CancellationToken cancellationToken)
{
CancellationHelper.ThrowIfCancellationRequested(cancellationToken);

if (_reader == null)
{
// If we've read everything, then _reader and _status will be null
if (_status == null)
if (_fetchResponse == null)
{
return 0;
}

try
{
using (JSObject body = _status.Body)
using (JSObject body = _fetchResponse.Body)
{
_reader = (JSObject)body.Invoke("getReader");
}
Expand All @@ -514,6 +516,11 @@ public override async ValueTask<int> ReadAsync(Memory<byte> buffer, Cancellation
}
}

using var abortRegistration = cancellationToken.Register(() =>
{
_reader.Invoke("cancel");
});

if (_bufferedBytes != null && _position < _bufferedBytes.Length)
{
return ReadBuffered();
Expand All @@ -524,13 +531,20 @@ public override async ValueTask<int> ReadAsync(Memory<byte> buffer, Cancellation
var t = (Task<object>)_reader.Invoke("read");
using (var read = (JSObject)await t.ConfigureAwait(continueOnCapturedContext: true))
{
if (cancellationToken.IsCancellationRequested)
{
_reader.Invoke("cancel");
CancellationHelper.CreateOperationCanceledException(null, cancellationToken);
}

CancellationHelper.ThrowIfCancellationRequested(cancellationToken);
if ((bool)read.GetObjectProperty("done"))
{
_reader.Dispose();
_reader = null;

_status?.Dispose();
_status = null;
_fetchResponse?.Dispose();
_fetchResponse = null;
return 0;
}

Expand Down Expand Up @@ -569,7 +583,7 @@ int ReadBuffered()
protected override void Dispose(bool disposing)
{
_reader?.Dispose();
_status?.Dispose();
_fetchResponse?.Dispose();
}

public override void Flush()
Expand Down
9 changes: 8 additions & 1 deletion src/libraries/System.Net.Http/tests/FunctionalTests/SocketsHttpHandlerTest.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1044,12 +1044,12 @@ public sealed class SocketsHttpHandlerTest_Cookies_Http11 : HttpClientHandlerTes
public SocketsHttpHandlerTest_Cookies_Http11(ITestOutputHelper output) : base(output) { }
}

[SkipOnPlatform(TestPlatforms.Browser, "ConnectTimeout is not supported on Browser")]
public sealed class SocketsHttpHandler_HttpClientHandler_Http11_Cancellation_Test : SocketsHttpHandler_Cancellation_Test
{
public SocketsHttpHandler_HttpClientHandler_Http11_Cancellation_Test(ITestOutputHelper output) : base(output) { }

[Fact]
[SkipOnPlatform(TestPlatforms.Browser, "ConnectTimeout is not supported on Browser")]
public void ConnectTimeout_Default()
{
using (var handler = new SocketsHttpHandler())
Expand All @@ -1062,6 +1062,7 @@ public void ConnectTimeout_Default()
[InlineData(0)]
[InlineData(-2)]
[InlineData(int.MaxValue + 1L)]
[SkipOnPlatform(TestPlatforms.Browser, "ConnectTimeout is not supported on Browser")]
public void ConnectTimeout_InvalidValues(long ms)
{
using (var handler = new SocketsHttpHandler())
Expand All @@ -1075,6 +1076,7 @@ public void ConnectTimeout_InvalidValues(long ms)
[InlineData(1)]
[InlineData(int.MaxValue - 1)]
[InlineData(int.MaxValue)]
[SkipOnPlatform(TestPlatforms.Browser, "ConnectTimeout is not supported on Browser")]
public void ConnectTimeout_ValidValues_Roundtrip(long ms)
{
using (var handler = new SocketsHttpHandler())
Expand All @@ -1085,6 +1087,7 @@ public void ConnectTimeout_ValidValues_Roundtrip(long ms)
}

[Fact]
[SkipOnPlatform(TestPlatforms.Browser, "ConnectTimeout is not supported on Browser")]
public void ConnectTimeout_SetAfterUse_Throws()
{
using (var handler = new SocketsHttpHandler())
Expand All @@ -1098,6 +1101,7 @@ public void ConnectTimeout_SetAfterUse_Throws()
}

[Fact]
[SkipOnPlatform(TestPlatforms.Browser, "ConnectTimeout is not supported on Browser")]
public void Expect100ContinueTimeout_Default()
{
using (var handler = new SocketsHttpHandler())
Expand All @@ -1109,6 +1113,7 @@ public void Expect100ContinueTimeout_Default()
[Theory]
[InlineData(-2)]
[InlineData(int.MaxValue + 1L)]
[SkipOnPlatform(TestPlatforms.Browser, "ConnectTimeout is not supported on Browser")]
public void Expect100ContinueTimeout_InvalidValues(long ms)
{
using (var handler = new SocketsHttpHandler())
Expand All @@ -1122,6 +1127,7 @@ public void Expect100ContinueTimeout_InvalidValues(long ms)
[InlineData(1)]
[InlineData(int.MaxValue - 1)]
[InlineData(int.MaxValue)]
[SkipOnPlatform(TestPlatforms.Browser, "ConnectTimeout is not supported on Browser")]
public void Expect100ContinueTimeout_ValidValues_Roundtrip(long ms)
{
using (var handler = new SocketsHttpHandler())
Expand All @@ -1132,6 +1138,7 @@ public void Expect100ContinueTimeout_ValidValues_Roundtrip(long ms)
}

[Fact]
[SkipOnPlatform(TestPlatforms.Browser, "ConnectTimeout is not supported on Browser")]
public void Expect100ContinueTimeout_SetAfterUse_Throws()
{
using (var handler = new SocketsHttpHandler())
Expand Down