Do not cancel operations after point of no return

sharwell · sharwell · commit 57759e39f555 · 2020-09-30T11:27:46.000-07:00
diff --git a/src/Workspaces/Remote/Core/BrokeredServiceConnection.cs b/src/Workspaces/Remote/Core/BrokeredServiceConnection.cs
@@ -150,6 +150,11 @@ internal static async ValueTask<TResult> InvokeStreamingServiceAsync<TResult>(
             Func<PipeReader, CancellationToken, ValueTask<TResult>> reader,
             CancellationToken cancellationToken)
         {
+            // We can cancel at entry, but once the pipe operations are scheduled we rely on both operations running to
+            // avoid deadlocks (the exception handler in 'writerTask' ensures progress is made in 'readerTask').
+            cancellationToken.ThrowIfCancellationRequested();
+            var mustNotCancelToken = CancellationToken.None;
+
             var pipe = new Pipe();
 
             // Create new tasks that both start executing, rather than invoking the delegates directly
@@ -170,7 +175,7 @@ internal static async ValueTask<TResult> InvokeStreamingServiceAsync<TResult>(
 
                     throw;
                 }
-            }, cancellationToken);
+            }, mustNotCancelToken);
 
             var readerTask = Task.Run(
                 async () =>
@@ -189,7 +194,7 @@ internal static async ValueTask<TResult> InvokeStreamingServiceAsync<TResult>(
                     {
                         await pipe.Reader.CompleteAsync(exception).ConfigureAwait(false);
                     }
-                }, cancellationToken);
+                }, mustNotCancelToken);
 
             await Task.WhenAll(writerTask, readerTask).ConfigureAwait(false);
 
diff --git a/src/Workspaces/Remote/Core/RemoteHostAssetSerialization.cs b/src/Workspaces/Remote/Core/RemoteHostAssetSerialization.cs
@@ -88,6 +88,11 @@ static void WriteAsset(ObjectWriter writer, ISerializerService serializer, Check
 
         public static async ValueTask<ImmutableArray<(Checksum, object)>> ReadDataAsync(PipeReader pipeReader, int scopeId, ISet<Checksum> checksums, ISerializerService serializerService, CancellationToken cancellationToken)
         {
+            // We can cancel at entry, but once the pipe operations are scheduled we rely on both operations running to
+            // avoid deadlocks (the exception handler in 'copyTask' ensures progress is made in the blocking read).
+            cancellationToken.ThrowIfCancellationRequested();
+            var mustNotCancelToken = CancellationToken.None;
+
             // Workaround for ObjectReader not supporting async reading.
             // Unless we read from the RPC stream asynchronously and with cancallation support we might hang when the server cancels.
             // https://github.com/dotnet/roslyn/issues/47861
@@ -113,7 +118,7 @@ static void WriteAsset(ObjectWriter writer, ISerializerService serializer, Check
                     await localPipe.Writer.CompleteAsync(exception).ConfigureAwait(false);
                     await pipeReader.CompleteAsync(exception).ConfigureAwait(false);
                 }
-            }, cancellationToken);
+            }, mustNotCancelToken);
 
             // blocking read from the local pipe on the current thread:
             try
diff --git a/src/Workspaces/Remote/Core/SolutionAssetProvider.cs b/src/Workspaces/Remote/Core/SolutionAssetProvider.cs
@@ -52,6 +52,11 @@ public async ValueTask GetAssetsAsync(PipeWriter pipeWriter, int scopeId, Checks
                 assetMap = await assetStorage.GetAssetsAsync(scopeId, checksums, cancellationToken).ConfigureAwait(false);
             }
 
+            // We can cancel early, but once the pipe operations are scheduled we rely on both operations running to
+            // avoid deadlocks (the exception handler in 'task1' ensures progress is made in 'task2').
+            cancellationToken.ThrowIfCancellationRequested();
+            var mustNotCancelToken = CancellationToken.None;
+
             // Work around the lack of async stream writing in ObjectWriter, which is required when writing to the RPC pipe.
             // Run two tasks - the first synchronously writes to a local pipe and the second asynchronosly transfers the data to the RPC pipe.
             //
@@ -71,7 +76,7 @@ public async ValueTask GetAssetsAsync(PipeWriter pipeWriter, int scopeId, Checks
                 {
                     // no-op
                 }
-            }, cancellationToken);
+            }, mustNotCancelToken);
 
             // Complete RPC once we send the initial piece of data and start waiting for the writer to send more,
             // so the client can start reading from the stream. Once CopyPipeDataAsync completes the pipeWriter

Original file line number	Diff line number	Diff line change
`@@ -150,6 +150,11 @@ internal static async ValueTask<TResult> InvokeStreamingServiceAsync<TResult>(`
`150`	`150`	`Func<PipeReader, CancellationToken, ValueTask<TResult>> reader,`
`151`	`151`	`CancellationToken cancellationToken)`
`152`	`152`	`{`
	`153`	`+ // We can cancel at entry, but once the pipe operations are scheduled we rely on both operations running to`
	`154`	`+ // avoid deadlocks (the exception handler in 'writerTask' ensures progress is made in 'readerTask').`
	`155`	`+ cancellationToken.ThrowIfCancellationRequested();`
	`156`	`+ var mustNotCancelToken = CancellationToken.None;`
	`157`	`+`
`153`	`158`	`var pipe = new Pipe();`
`154`	`159`
`155`	`160`	`// Create new tasks that both start executing, rather than invoking the delegates directly`
`@@ -170,7 +175,7 @@ internal static async ValueTask<TResult> InvokeStreamingServiceAsync<TResult>(`
`170`	`175`
`171`	`176`	`throw;`
`172`	`177`	`}`
`173`		`- }, cancellationToken);`
	`178`	`+ }, mustNotCancelToken);`
`174`	`179`
`175`	`180`	`var readerTask = Task.Run(`
`176`	`181`	`async () =>`
`@@ -189,7 +194,7 @@ internal static async ValueTask<TResult> InvokeStreamingServiceAsync<TResult>(`
`189`	`194`	`{`
`190`	`195`	`await pipe.Reader.CompleteAsync(exception).ConfigureAwait(false);`
`191`	`196`	`}`
`192`		`- }, cancellationToken);`
	`197`	`+ }, mustNotCancelToken);`
`193`	`198`
`194`	`199`	`await Task.WhenAll(writerTask, readerTask).ConfigureAwait(false);`
`195`	`200`
Original file line number	Diff line number	Diff line change
`@@ -52,6 +52,11 @@ public async ValueTask GetAssetsAsync(PipeWriter pipeWriter, int scopeId, Checks`
`52`	`52`	`assetMap = await assetStorage.GetAssetsAsync(scopeId, checksums, cancellationToken).ConfigureAwait(false);`
`53`	`53`	`}`
`54`	`54`
	`55`	`+ // We can cancel early, but once the pipe operations are scheduled we rely on both operations running to`
	`56`	`+ // avoid deadlocks (the exception handler in 'task1' ensures progress is made in 'task2').`
	`57`	`+ cancellationToken.ThrowIfCancellationRequested();`
	`58`	`+ var mustNotCancelToken = CancellationToken.None;`
	`59`	`+`
`55`	`60`	`// Work around the lack of async stream writing in ObjectWriter, which is required when writing to the RPC pipe.`
`56`	`61`	`// Run two tasks - the first synchronously writes to a local pipe and the second asynchronosly transfers the data to the RPC pipe.`
`57`	`62`	`//`
`@@ -71,7 +76,7 @@ public async ValueTask GetAssetsAsync(PipeWriter pipeWriter, int scopeId, Checks`
`71`	`76`	`{`
`72`	`77`	`// no-op`
`73`	`78`	`}`
`74`		`- }, cancellationToken);`
	`79`	`+ }, mustNotCancelToken);`
`75`	`80`
`76`	`81`	`// Complete RPC once we send the initial piece of data and start waiting for the writer to send more,`
`77`	`82`	`// so the client can start reading from the stream. Once CopyPipeDataAsync completes the pipeWriter`