Handle concurrent dispose and WriteAsync calls

src/System.IO.Pipelines/src/Resources/Strings.resx

@@ -120,12 +120,6 @@
   <data name="AdvanceToInvalidCursor" xml:space="preserve">
     <value>The PipeReader has already advanced past the provided position.</value>
   </data>
-  <data name="CannotCompleteWhileReading" xml:space="preserve">
-    <value>Can't complete reader while reading.</value>
-  </data>
-  <data name="CannotCompleteWhiteWriting" xml:space="preserve">
-    <value>Can't complete writer while writing.</value>
-  </data>
   <data name="ConcurrentOperationsNotSupported" xml:space="preserve">
     <value>Concurrent reads or writes are not supported.</value>
   </data>

src/System.IO.Pipelines/src/System/IO/Pipelines/Pipe.cs

@@ -934,16 +934,19 @@ internal ValueTask<FlushResult> WriteAsync(ReadOnlyMemory<byte> source, Cancella
             // state as writing
             AllocateWriteHeadIfNeeded(0);
 
-            if (source.Length <= _writingHeadMemory.Length)
+            lock (_sync)
             {
-                source.CopyTo(_writingHeadMemory);
+                if (source.Length <= _writingHeadMemory.Length)
+                {
+                    source.CopyTo(_writingHeadMemory);
 
-                AdvanceCore(source.Length);
-            }
-            else
-            {
-                // This is the multi segment copy
-                WriteMultiSegment(source.Span);
+                    AdvanceCore(source.Length);
+                }
+                else
+                {
+                    // This is the multi segment copy
+                    WriteMultiSegment(source.Span);
+                }
             }
 
             return FlushAsync(cancellationToken);

src/System.IO.Pipelines/tests/PipeWriterTests.cs

@@ -232,5 +232,33 @@ public async Task WritesUsingGetMemoryWorks()
 
             pipe.Reader.Complete();
         }
+
+        [Fact]
+        public async Task CompleteWithLargeWriteThrows()
+        {
+            var pipe = new Pipe();
+            pipe.Reader.Complete();
+
+            var task = Task.Run(async () =>
+            {
+                await Task.Delay(10);
+                pipe.Writer.Complete();
+            });
+
+            try
+            {
+                for (int i = 0; i < 1000; i++)
+                {
+                    var buffer = new byte[10000000];
+                    await pipe.Writer.WriteAsync(buffer);
+                }
+            }
+            catch (InvalidOperationException)
+            {
+                // Complete while writing
+            }
+
+            await task;
+        }
     }
 }