huahaofeng
/
aspnetcore
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Merged PR 3557: Revert "Wait to Complete Pipe"

This commit is contained in:
Brennan Conroy 2019-10-11 01:46:11 +00:00 committed by Andrew Stanton-Nurse
parent 7fd42c4e94
commit 57e68b069e
8 changed files with 12 additions and 164 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
eng/PatchConfig.props
View File

@ -44,10 +44,4 @@ Later on, this will be checked using this condition:
Microsoft.AspNetCore.CookiePolicy; Microsoft.AspNetCore.CookiePolicy;
</PackagesInPatch> </PackagesInPatch>
</PropertyGroup> </PropertyGroup>
<PropertyGroup Condition=" '$(VersionPrefix)' == '2.1.14' ">
<PackagesInPatch>
Microsoft.AspNetCore.Http.Connections;
Microsoft.AspNetCore.SignalR.Core;
</PackagesInPatch>
</PropertyGroup>
</Project> </Project>

23
src/SignalR/clients/ts/FunctionalTests/selenium/run-tests.ts
View File

@ -1,8 +1,7 @@
import { ChildProcess, spawn } from "child_process"; import { ChildProcess, spawn } from "child_process";
import * as _fs from "fs"; import * as fs from "fs";
import { EOL } from "os"; import { EOL } from "os";
import * as path from "path"; import * as path from "path";
import { promisify } from "util";
import { PassThrough, Readable } from "stream"; import { PassThrough, Readable } from "stream";
import { run } from "../../webdriver-tap-runner/lib"; import { run } from "../../webdriver-tap-runner/lib";
@ -10,16 +9,6 @@ import { run } from "../../webdriver-tap-runner/lib";
import * as _debug from "debug"; import * as _debug from "debug";
const debug = _debug("signalr-functional-tests:run"); const debug = _debug("signalr-functional-tests:run");
const ARTIFACTS_DIR = path.resolve(__dirname, "..", "..", "..", "..", "artifacts");
const LOGS_DIR = path.resolve(ARTIFACTS_DIR, "logs");
// Promisify things from fs we want to use.
const fs = {
createWriteStream: _fs.createWriteStream,
exists: promisify(_fs.exists),
mkdir: promisify(_fs.mkdir),
};
process.on("unhandledRejection", (reason) => { process.on("unhandledRejection", (reason) => {
console.error(`Unhandled promise rejection: ${reason}`); console.error(`Unhandled promise rejection: ${reason}`);
process.exit(1); process.exit(1);
@ -113,13 +102,6 @@ if (chromePath) {
try { try {
const serverPath = path.resolve(__dirname, "..", "bin", configuration, "netcoreapp2.1", "FunctionalTests.dll"); const serverPath = path.resolve(__dirname, "..", "bin", configuration, "netcoreapp2.1", "FunctionalTests.dll");
if (!await fs.exists(ARTIFACTS_DIR)) {
await fs.mkdir(ARTIFACTS_DIR);
}
if (!await fs.exists(LOGS_DIR)) {
await fs.mkdir(LOGS_DIR);
}
debug(`Launching Functional Test Server: ${serverPath}`); debug(`Launching Functional Test Server: ${serverPath}`);
const dotnet = spawn("dotnet", [serverPath], { const dotnet = spawn("dotnet", [serverPath], {
env: { env: {
@ -135,9 +117,6 @@ if (chromePath) {
} }
} }
const logStream = fs.createWriteStream(path.resolve(LOGS_DIR, "ts.functionaltests.dotnet.log"));
dotnet.stdout.pipe(logStream);
process.on("SIGINT", cleanup); process.on("SIGINT", cleanup);
process.on("exit", cleanup); process.on("exit", cleanup);

28
src/SignalR/common/Http.Connections/src/Internal/HttpConnectionContext.cs
View File

@ -274,35 +274,13 @@ namespace Microsoft.AspNetCore.Http.Connections.Internal
// Cancel any pending flushes from back pressure // Cancel any pending flushes from back pressure
Application?.Output.CancelPendingFlush(); Application?.Output.CancelPendingFlush();
// Normally it isn't safe to try and acquire this lock because the Send can hold onto it for a long time if there is backpressure // Shutdown both sides and wait for nothing
// It is safe to wait for this lock now because the Send will be in one of 4 states
// 1. In the middle of a write which is in the middle of being canceled by the CancelPendingFlush above, when it throws
// an OperationCanceledException it will complete the PipeWriter which will make any other Send waiting on the lock
// throw an InvalidOperationException if they call Write
// 2. About to write and see that there is a pending cancel from the CancelPendingFlush, go to 1 to see what happens
// 3. Enters the Send and sees the Dispose state from DisposeAndRemoveAsync and releases the lock
// 4. No Send in progress
await WriteLock.WaitAsync();
try
{
// Complete the applications read loop
Application?.Output.Complete(transportTask.Exception?.InnerException);
}
finally
{
WriteLock.Release();
}
Log.WaitingForTransportAndApplication(_logger, TransportType);
// Wait for application so we can complete the writer safely
await applicationTask.NoThrow();
// Shutdown application side now that it's finished
Transport?.Output.Complete(applicationTask.Exception?.InnerException); Transport?.Output.Complete(applicationTask.Exception?.InnerException);
Application?.Output.Complete(transportTask.Exception?.InnerException);
try try
{ {
Log.WaitingForTransportAndApplication(_logger, TransportType);
// A poorly written application *could* in theory get stuck forever and it'll show up as a memory leak // A poorly written application *could* in theory get stuck forever and it'll show up as a memory leak
await Task.WhenAll(applicationTask, transportTask); await Task.WhenAll(applicationTask, transportTask);
} }

15
src/SignalR/common/Http.Connections/src/Internal/HttpConnectionDispatcher.cs
View File

@ -511,14 +511,6 @@ namespace Microsoft.AspNetCore.Http.Connections.Internal
context.Response.StatusCode = StatusCodes.Status404NotFound; context.Response.StatusCode = StatusCodes.Status404NotFound;
context.Response.ContentType = "text/plain"; context.Response.ContentType = "text/plain";
// There are no writes anymore (since this is the write "loop")
// So it is safe to complete the writer
// We complete the writer here because we already have the WriteLock acquired
// and it's unsafe to complete outside of the lock
// Other code isn't guaranteed to be able to acquire the lock before another write
// even if CancelPendingFlush is called, and the other write could hang if there is backpressure
connection.Application.Output.Complete();
return; return;
} }
@ -557,8 +549,11 @@ namespace Microsoft.AspNetCore.Http.Connections.Internal
Log.TerminatingConection(_logger); Log.TerminatingConection(_logger);
// Dispose the connection, but don't wait for it. We assign it here so we can wait in tests // Complete the receiving end of the pipe
connection.DisposeAndRemoveTask = _manager.DisposeAndRemoveAsync(connection, closeGracefully: false); connection.Application.Output.Complete();
// Dispose the connection gracefully, but don't wait for it. We assign it here so we can wait in tests
connection.DisposeAndRemoveTask = _manager.DisposeAndRemoveAsync(connection, closeGracefully: true);
context.Response.StatusCode = StatusCodes.Status202Accepted; context.Response.StatusCode = StatusCodes.Status202Accepted;
context.Response.ContentType = "text/plain"; context.Response.ContentType = "text/plain";

27
src/SignalR/common/Http.Connections/src/Internal/TaskExtensions.cs
View File

@ -1,27 +0,0 @@
// Copyright (c) .NET Foundation. All rights reserved.
// Licensed under the Apache License, Version 2.0. See License.txt in the project root for license information.
using System.Runtime.CompilerServices;
namespace System.Threading.Tasks
{
internal static class TaskExtensions
{
public static async Task NoThrow(this Task task)
{
await new NoThrowAwaiter(task);
}
}
internal readonly struct NoThrowAwaiter : ICriticalNotifyCompletion
{
private readonly Task _task;
public NoThrowAwaiter(Task task) { _task = task; }
public NoThrowAwaiter GetAwaiter() => this;
public bool IsCompleted => _task.IsCompleted;
// Observe exception
public void GetResult() { _ = _task.Exception; }
public void OnCompleted(Action continuation) => _task.GetAwaiter().OnCompleted(continuation);
public void UnsafeOnCompleted(Action continuation) => OnCompleted(continuation);
}
}

12
src/SignalR/common/Shared/PipeWriterStream.cs
View File

@ -1,4 +1,4 @@
// Copyright (c) .NET Foundation. All rights reserved. // Copyright (c) .NET Foundation. All rights reserved.
// Licensed under the Apache License, Version 2.0. See License.txt in the project root for license information. // Licensed under the Apache License, Version 2.0. See License.txt in the project root for license information.
using System; using System;
@ -76,15 +76,7 @@ namespace System.IO.Pipelines
_length += source.Length; _length += source.Length;
var task = _pipeWriter.WriteAsync(source); var task = _pipeWriter.WriteAsync(source);
if (task.IsCompletedSuccessfully) if (!task.IsCompletedSuccessfully)
{
// Cancellation can be triggered by PipeWriter.CancelPendingFlush
if (task.Result.IsCanceled)
{
throw new OperationCanceledException();
}
}
else if (!task.IsCompletedSuccessfully)
{ {
return WriteSlowAsync(task); return WriteSlowAsync(task);
} }

61
src/SignalR/server/Core/src/HubConnectionContext.cs
View File

@ -33,7 +33,6 @@ namespace Microsoft.AspNetCore.SignalR
private long _lastSendTimestamp = Stopwatch.GetTimestamp(); private long _lastSendTimestamp = Stopwatch.GetTimestamp();
private ReadOnlyMemory<byte> _cachedPingMessage; private ReadOnlyMemory<byte> _cachedPingMessage;
private volatile bool _connectionAborted;
/// <summary> /// <summary>
/// Initializes a new instance of the <see cref="HubConnectionContext"/> class. /// Initializes a new instance of the <see cref="HubConnectionContext"/> class.
@ -100,12 +99,6 @@ namespace Microsoft.AspNetCore.SignalR
return new ValueTask(WriteSlowAsync(message)); return new ValueTask(WriteSlowAsync(message));
} }
if (_connectionAborted)
{
_writeLock.Release();
return default;
}
// This method should never throw synchronously // This method should never throw synchronously
var task = WriteCore(message); var task = WriteCore(message);
@ -136,12 +129,6 @@ namespace Microsoft.AspNetCore.SignalR
return new ValueTask(WriteSlowAsync(message)); return new ValueTask(WriteSlowAsync(message));
} }
if (_connectionAborted)
{
_writeLock.Release();
return default;
}
// This method should never throw synchronously // This method should never throw synchronously
var task = WriteCore(message); var task = WriteCore(message);
@ -171,8 +158,6 @@ namespace Microsoft.AspNetCore.SignalR
{ {
Log.FailedWritingMessage(_logger, ex); Log.FailedWritingMessage(_logger, ex);
Abort();
return new ValueTask<FlushResult>(new FlushResult(isCanceled: false, isCompleted: true)); return new ValueTask<FlushResult>(new FlushResult(isCanceled: false, isCompleted: true));
} }
} }
@ -190,8 +175,6 @@ namespace Microsoft.AspNetCore.SignalR
{ {
Log.FailedWritingMessage(_logger, ex); Log.FailedWritingMessage(_logger, ex);
Abort();
return new ValueTask<FlushResult>(new FlushResult(isCanceled: false, isCompleted: true)); return new ValueTask<FlushResult>(new FlushResult(isCanceled: false, isCompleted: true));
} }
} }
@ -205,8 +188,6 @@ namespace Microsoft.AspNetCore.SignalR
catch (Exception ex) catch (Exception ex)
{ {
Log.FailedWritingMessage(_logger, ex); Log.FailedWritingMessage(_logger, ex);
Abort();
} }
finally finally
{ {
@ -220,11 +201,6 @@ namespace Microsoft.AspNetCore.SignalR
await _writeLock.WaitAsync(); await _writeLock.WaitAsync();
try try
{ {
if (_connectionAborted)
{
return;
}
// Failed to get the lock immediately when entering WriteAsync so await until it is available // Failed to get the lock immediately when entering WriteAsync so await until it is available
await WriteCore(message); await WriteCore(message);
@ -232,8 +208,6 @@ namespace Microsoft.AspNetCore.SignalR
catch (Exception ex) catch (Exception ex)
{ {
Log.FailedWritingMessage(_logger, ex); Log.FailedWritingMessage(_logger, ex);
Abort();
} }
finally finally
{ {
@ -245,11 +219,6 @@ namespace Microsoft.AspNetCore.SignalR
{ {
try try
{ {
if (_connectionAborted)
{
return;
}
// Failed to get the lock immediately when entering WriteAsync so await until it is available // Failed to get the lock immediately when entering WriteAsync so await until it is available
await _writeLock.WaitAsync(); await _writeLock.WaitAsync();
@ -258,8 +227,6 @@ namespace Microsoft.AspNetCore.SignalR
catch (Exception ex) catch (Exception ex)
{ {
Log.FailedWritingMessage(_logger, ex); Log.FailedWritingMessage(_logger, ex);
Abort();
} }
finally finally
{ {
@ -283,11 +250,6 @@ namespace Microsoft.AspNetCore.SignalR
{ {
try try
{ {
if (_connectionAborted)
{
return;
}
await _connectionContext.Transport.Output.WriteAsync(_cachedPingMessage); await _connectionContext.Transport.Output.WriteAsync(_cachedPingMessage);
Log.SentPing(_logger); Log.SentPing(_logger);
@ -295,8 +257,6 @@ namespace Microsoft.AspNetCore.SignalR
catch (Exception ex) catch (Exception ex)
{ {
Log.FailedWritingMessage(_logger, ex); Log.FailedWritingMessage(_logger, ex);
Abort();
} }
finally finally
{ {
@ -333,12 +293,6 @@ namespace Microsoft.AspNetCore.SignalR
/// </summary> /// </summary>
public virtual void Abort() public virtual void Abort()
{ {
_connectionAborted = true;
// Cancel any current writes or writes that are about to happen and have already gone past the _connectionAborted bool
// We have to do this outside of the lock otherwise it could hang if the write is observing backpressure
_connectionContext.Transport.Output.CancelPendingFlush();
// If we already triggered the token then noop, this isn't thread safe but it's good enough // If we already triggered the token then noop, this isn't thread safe but it's good enough
// to avoid spawning a new task in the most common cases // to avoid spawning a new task in the most common cases
if (_connectionAbortedTokenSource.IsCancellationRequested) if (_connectionAbortedTokenSource.IsCancellationRequested)
@ -469,24 +423,9 @@ namespace Microsoft.AspNetCore.SignalR
internal Task AbortAsync() internal Task AbortAsync()
{ {
Abort(); Abort();
// Acquire lock to make sure all writes are completed
if (!_writeLock.Wait(0))
{
return AbortAsyncSlow();
}
_writeLock.Release();
return _abortCompletedTcs.Task; return _abortCompletedTcs.Task;
} }
private async Task AbortAsyncSlow()
{
await _writeLock.WaitAsync();
_writeLock.Release();
await _abortCompletedTcs.Task;
}
private void KeepAliveTick() private void KeepAliveTick()
{ {
var timestamp = Stopwatch.GetTimestamp(); var timestamp = Stopwatch.GetTimestamp();

4
src/SignalR/server/SignalR/test/HubConnectionHandlerTests.cs
View File

@ -79,11 +79,9 @@ namespace Microsoft.AspNetCore.SignalR.Tests
{ {
var connectionHandlerTask = await client.ConnectAsync(connectionHandler); var connectionHandlerTask = await client.ConnectAsync(connectionHandler);
await client.SendInvocationAsync(nameof(AbortHub.Kill)); await client.InvokeAsync(nameof(AbortHub.Kill));
await connectionHandlerTask.OrTimeout(); await connectionHandlerTask.OrTimeout();
Assert.Null(client.TryRead());
} }
} }