Add System.IO.Pipelines API #27007

pakrym · 2018-02-09T19:04:46Z

94% coverage of new API

Please check csprojes, configuration.props and other infrastructure files. It's my first time adding a package to corefx so I may have done some mistakes there.

pakrym · 2018-02-09T19:37:06Z

Not sure why System.IO.Pipelines ref doesn't pick up System.Memory reference on the first build.

KrzysztofCwalina · 2018-02-09T21:17:14Z

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

+    /// </summary>
+    public sealed class Pipe
+    {
+        private const int SegmentPoolSize = 16;


Should this be configurable in PipeOptions? The number is kind of small

We decided to try const first with the idea that if you have more you can increase minimum block size and have larger blocks.

I think we should make it into an option, but it's ok to punt for later.

KrzysztofCwalina · 2018-02-09T21:19:28Z

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

+
+        private readonly MemoryPool<byte> _pool;
+        private readonly int _minimumSegmentSize;
+        private readonly long _maximumSizeHigh;


Are these the backpressure thresholds? If yes, could you rename to use the same terminology as the PipeOptions properties.

KrzysztofCwalina · 2018-02-09T21:21:00Z

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

+                ThrowHelper.ThrowInvalidOperationException_NoWritingAllowed();
+            }
+
+            if (minimumSize < 0)


Is zero ok? Why would we allow that? In fact, whenever I write code using our IOutput APIs I need to have a special branch for empty buffer.

Zero is whatever is availible at the moment but not empty

1 is also whatever is available at the moment but not zero, correct? If yes, then default being one and zero throwing would be more transparent API.

No, one is "I need to write one byte", zero is for "whatever is availible"

KrzysztofCwalina · 2018-02-09T21:24:35Z

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

+                {
+                    BufferSegment nextSegment = CreateSegmentUnsynchronized();
+
+                    nextSegment.SetMemory(_pool.Rent(Math.Max(_minimumSegmentSize, minimumSize)));


Do you need to call rent under the lock? Since pool is an abstraction, we don't know what rent does and so it can cause deadlocks or other synchronization problems.

We need to do everything around it in lock. It might deadlock if pool would try to access the same pipe object on different thread which is not very common situation.

OK. Let's see if we can optimized the locks later.

KrzysztofCwalina · 2018-02-09T21:26:58Z

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

+
+            lock (_sync)
+            {
+                BufferSegment segment = _writingHead ?? AllocateWriteHeadUnsynchronized(minimumSize);


Could this be optimize, if the default scheduler is used? ... in which case the reader is reading on the same thread.
I could imagine doing all but allocations of new segments without a lock if the reader continuations are called on the same thread.

Even if scheduler is inline you don't have any guarantees what thread would call into pipe

Yes, but at that point you could do just a simple interlocked read/write, correct? Anyway, I think the code is good as is, I just noticed how much heavy locking we do and it looks like we could optimize this code in the future.

It's actually really hard to remove the locking, we tried several time and failed, each time we ended up locking more 😄 . That said, I'm happy to look at this again once we have lots of tests in place.

OK. Let's see if we can optimized the locks later.

Is it a valid use case for Pipe to have multiple consumers or producers anyway? Seems like that leads to a lot of undefined behavior down the road.

No, pipe is single producer single consumer.

KrzysztofCwalina · 2018-02-09T21:35:51Z

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

+        [MethodImpl(MethodImplOptions.AggressiveInlining)]
+        internal void Advance(int bytesWritten)
+        {
+            if (_writingHead == null)


When can writing head be null? Is the exception going to propagate to user code? If yes, why?

If you call Advance without calling GetMemory()

I assume the exception will explain the problem.

The exceptions need to be reviewed. They are from before the IOutput refactor.

KrzysztofCwalina · 2018-02-09T21:44:20Z

src/System.IO.Pipelines/src/System/IO/Pipelines/PipeAwaiter.cs

+        /// Create new instance of <see cref="PipeAwaiter{T}"/> that wraps async operation implemented by <paramref name="awaiter"/>
+        /// </summary>
+        public PipeAwaiter(IPipeAwaiter<T> awaiter)
+        {


Maybe Debug.Assert that the awaiter is not null

ghost · 2018-02-09T22:05:26Z

src/System.IO.Pipelines/src/Properties/InternalsVisibleTo.cs

+
+using System.Runtime.CompilerServices;
+
+[assembly: InternalsVisibleTo("System.IO.Pipelines.Tests, PublicKey=002400000480000094000000060200000024000052534131000400000100010015c01ae1f50e8cc09ba9eac9147cf8fd9fce2cfe9f8dce4f7301c4132ca9fb50ce8cbf1df4dc18dd4d210e4345c744ecb3365ed327efdbc52603faa5e21daa11234c8c4a73e51f03bf192544581ebe107adee3a34928e39d04e524a9ce729d5090bfd7dad9d10c722c0def9ccc08ff0a03790e48bcd1f9b6c476063e1966a1c4")]


I think we should avoid this. If tests need to extract internals (and this practice should be used with discretion), it should use Reflection to get just what's needed.

Really? The ASP.NET uses this pattern for tests all the time. In fact thats the only valid use of internals visible to IMO.

ghost · 2018-02-09T22:07:33Z

src/System.IO.Pipelines/src/System/IO/Pipelines/ThrowHelper.cs

+
+namespace System.IO.Pipelines
+{
+    internal class ThrowHelper


Declare as "internal static class" to avoid creation of unnecessary constructor.

ghost · 2018-02-09T22:09:55Z

src/System.IO.Pipelines/src/System/IO/Pipelines/BufferSegment.cs

+
+namespace System.IO.Pipelines
+{
+    internal class BufferSegment : IMemoryList<byte>


I didn't see a class deriving from this. Can this class be sealed?

ghost · 2018-02-09T22:10:28Z

src/System.IO.Pipelines/src/System/IO/Pipelines/PipeCompletionCallbacks.cs

+
+namespace System.IO.Pipelines
+{
+    internal class PipeCompletionCallbacks


I didn't see a class deriving from this. Can this class be sealed?

ghost · 2018-02-09T22:20:50Z

src/System.IO.Pipelines/src/System/IO/Pipelines/ThreadPoolScheduler.cs

+    {
+        public override void Schedule(Action action)
+        {
+#if NETCOREAPP2_1


This #if maze is too hard to follow. I think it'd be better to have conditionally imported files named ThreadPoolScheduler.NetCoreApp21.cs, ThreadPoolScheduler.NetStandard2.cs instead.

Is it worth it? They don't even intersect, three small completely different versions depending on the target.

The lack of commonality is exactly why separate files is preferable to #if's. #if's are our least-preferred mechanism for version-specific coding - mainly to be used when 95% of a file is the same between versions and only a handful of lines need to be tweaked. Here, almost every material line is conditional on something - it's very hard to follow.

KrzysztofCwalina · 2018-02-09T22:24:08Z

src/System.Memory/src/System/Buffers/IBufferWriter.cs

+        /// Requests the <see cref="Memory{Byte}"/> of at least <paramref name="minimumLength"/> in size.
+        /// If <paramref name="minimumLength"/> is equal to <code>0</code>, currently available memory would get returned.
+        /// </summary>
+        Memory<byte> GetMemory(int minimumLength = 0);


We should state in the contract whether returning an empty buffer is ok or not. And I would vote that it's not ok to return empty (at which point we should add some tests to validate it.

It's not ok to return empty. I'll add some.

sounds good.

ghost · 2018-02-09T22:26:19Z

src/System.IO.Pipelines/src/System/IO/Pipelines/PipeCompletion.cs

+    [DebuggerDisplay("IsCompleted: {" + nameof(IsCompleted) + "}")]
+    internal struct PipeCompletion
+    {
+        private static readonly ArrayPool<PipeCompletionCallback> CompletionCallbackPool = ArrayPool<PipeCompletionCallback>.Shared;


static field names should be prefixed with "s_" (saw various instances of this throughout.)

ghost · 2018-02-09T22:35:50Z

src/System.IO.Pipelines/src/System/IO/Pipelines/ThrowHelper.cs

+        [MethodImpl(MethodImplOptions.NoInlining)]
+        private static Exception CreateArgumentNullException(ExceptionArgument argument) { return new ArgumentNullException(argument.ToString()); }
+
+        public static void ThrowInvalidOperationException_NotWritingNoAlloc()


I think the rest of this file should be packed like the first part (use one-line formatting for each method with blank lines only between each group of paired methods.) It's much easier to see the pattern and to figure out what you need to select for copy-pasting when adding a new ThrowHelper.

ghost · 2018-02-09T22:38:30Z

src/System.IO.Pipelines/src/System/IO/Pipelines/ResultFlags.cs

+    internal enum ResultFlags : byte
+    {
+        None = 0,
+        Canceled = 1,


Nit: For [Flags] enums, it's usually clearer to use hex notation for the values.

ghost · 2018-02-09T22:39:44Z

src/System.IO.Pipelines/src/System/IO/Pipelines/PipeReaderState.cs

+        public bool IsActive => _state == State.Active;
+    }
+
+    internal enum State: byte


Could this be a private enum defined inside PipeReaderState (assuming the DebuggerDisplay magic would still work)

ghost · 2018-02-09T22:43:46Z

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

+                    // might be using tailspace
+                    if (consumed.Index == returnEnd.Length && _writingHead != returnEnd)
+                    {
+                        var nextBlock = returnEnd.NextSegment;


Non-compliant use of "var"

ghost · 2018-02-09T22:44:51Z

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

+            }
+        }
+
+        private sealed class DefaultPipeReader : PipeReader, IPipeAwaiter<ReadResult>


These are non-trivial-sized nested classes in a file that's already quite large. Please move these two to separate files.

davidfowl · 2018-02-13T06:57:43Z

src/System.IO.Pipelines/ref/System.IO.Pipelines.cs

+        public bool IsCanceled { get { throw null; } }
+        public bool IsCompleted { get { throw null; } }
+    }
+    public partial interface IDuplexPipe : System.IDisposable


Lets get rid of IDisposable. Not critical for this PR but we'll need to make more changes.

davidfowl · 2018-02-13T07:04:35Z

src/System.Memory/ref/System.Memory.cs

+    public partial interface IBufferWriter<T>
+    {
+        void Advance(int count);
+        System.Memory<byte> GetMemory(int minimumLength = 0);


Bug here these should be T not byte

davidfowl · 2018-02-13T07:04:55Z

src/System.Memory/ref/System.Memory.cs

@@ -273,6 +280,7 @@ public static partial class BuffersExtensions
        public static void CopyTo<T>(this System.Buffers.ReadOnlySequence<T> sequence, System.Span<T> destination) { }
        public static System.Nullable<System.SequencePosition> PositionOf<T>(this System.Buffers.ReadOnlySequence<T> sequence, T value) where T : System.IEquatable<T> { throw null; }
        public static T[] ToArray<T>(this System.Buffers.ReadOnlySequence<T> sequence) { throw null; }
+        public static void Write<T>(this System.Buffers.IBufferWriter<T> bufferWriter, ReadOnlySpan<byte> source) { }


Same here, this should be ReadOnlySpan<T>

joperezr · 2018-02-13T18:00:12Z

src/System.IO.Pipelines/src/System.IO.Pipelines.csproj

+    <Compile Include="System\IO\Pipelines\ResultFlags.cs" />
+    <Compile Include="System\IO\Pipelines\ThrowHelper.cs" />
+  </ItemGroup>
+  <ItemGroup Condition="'$(TargetGroup)'=='netcoreapp2.1'">


This condition should look like Condition="'$(TargetGroup)'=='netcoreapp'" instead

joperezr · 2018-02-13T18:03:52Z

src/System.IO.Pipelines/src/System.IO.Pipelines.csproj

+    <Reference Include="System.Resources.ResourceManager" />
+    <Reference Include="System.Runtime" />
+    <Reference Include="System.Runtime.Extensions" />
+    <Reference Include="System.Threading.Tasks" />


You are also missing a reference to System.Threading.ThreadPool in order to get the WaitCallback

joperezr · 2018-02-13T18:05:58Z

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

+                ThrowHelper.ThrowArgumentOutOfRangeException(ExceptionArgument.minimumSize);
+            }
+
+            lock (_sync)


All of these locks expressions will now throw compiler error CS0656 saying:

System\IO\Pipelines\Pipe.cs(130,13): error CS0656: Missing compiler required member 'System.Threading.Monitor.Exit' [F:\git\corefx\src\System.IO.Pipelines\src\System.IO.Pipelines.csproj] System\IO\Pipelines\Pipe.cs(130,13): error CS0656: Missing compiler required member 'System.Threading.Monitor.Enter' [F:\git\corefx\src\System.IO.Pipelines\src\System.IO.Pipelines.csproj]

Fixed by adding System.Threading reference.

ghost · 2018-02-13T22:04:01Z

My comment was about “cancelation” – it’s apparently an alternate spelling but very rarely used (I’ve never seen it used myself). The BCL api cataloged in https://apisof.net/ uses “cancellation” exclusively.

ahsonkhan · 2018-02-13T22:13:48Z

src/System.Memory/src/System/Buffers/BuffersExtensions.cs

+
+            while (source.Length > 0)
+            {
+                int writeSize;


nit: int writeSize = destination.Length;
And then we can remove the else condition.

ahsonkhan · 2018-02-13T22:14:47Z

src/System.Memory/src/System/Buffers/BuffersExtensions.cs

+            while (source.Length > 0)
+            {
+                int writeSize;
+                if (destination.Length == 0)


Aside from the first time through the loop, this will always be true (since destination = default). Consider special casing the first iteration outside the loop.

ahsonkhan · 2018-02-13T22:15:34Z

src/System.Memory/src/System/Buffers/BuffersExtensions.cs

+                    writeSize = destination.Length;
+                }
+
+                bufferWriter.Advance(writeSize);


nit: advance after the copyto

ahsonkhan

Otherwise, lgtm

joperezr

Assuming you plan on adding the netstandard1.1 stuff later this LGTM

Add System.IO.Pipelines API Commit migrated from dotnet/corefx@3c708ab

pakrym added 4 commits February 7, 2018 19:29

Add System.IO.Pipelines

5573f26

Compiles

dee5c34

Resources

c3a898a

Cover

4acf7f9

pakrym requested review from davidfowl, ahsonkhan and KrzysztofCwalina February 9, 2018 19:04

KrzysztofCwalina reviewed Feb 9, 2018

View reviewed changes

joshfree requested review from joshfree, GrabYourPitchforks and a user February 9, 2018 21:58

ghost reviewed Feb 9, 2018

View reviewed changes

KrzysztofCwalina reviewed Feb 9, 2018

View reviewed changes

ghost reviewed Feb 9, 2018

View reviewed changes

davidfowl reviewed Feb 13, 2018

View reviewed changes

pakrym added 2 commits February 13, 2018 08:22

T

792c5cd

IDuplexPipe

82be16b

joperezr reviewed Feb 13, 2018

View reviewed changes

pakrym and others added 2 commits February 13, 2018 10:39

NetCoreapp

bb0d129

Fixing pkgproj

200bae7

ahsonkhan reviewed Feb 13, 2018

View reviewed changes

ahsonkhan approved these changes Feb 13, 2018

View reviewed changes

joperezr and others added 2 commits February 13, 2018 14:24

Fix test project

bb72f2e

More comments

17b00a5

joperezr approved these changes Feb 14, 2018

View reviewed changes

KrzysztofCwalina merged commit 3c708ab into dotnet:master Feb 14, 2018

ahsonkhan mentioned this pull request Feb 23, 2018

TryGet for ReadOnlysequence #27229

Merged

karelz added this to the 2.1.0 milestone Mar 10, 2018

antonfirsov mentioned this pull request Jun 16, 2018

System.Memory API integration SixLabors/ImageSharp#565

Closed

7 tasks

AndyAyersMS mentioned this pull request Jan 31, 2020

JIT: consider optimizing this unsigned check and compare idiom dotnet/runtime#9724

Open

picenka21 pushed a commit to picenka21/runtime that referenced this pull request Feb 18, 2022

Merge pull request dotnet/corefx#27007 from pakrym/pakrym/pipelines

05b4594

Add System.IO.Pipelines API Commit migrated from dotnet/corefx@3c708ab


		using System.Runtime.CompilerServices;

		[assembly: InternalsVisibleTo("System.IO.Pipelines.Tests, PublicKey=002400000480000094000000060200000024000052534131000400000100010015c01ae1f50e8cc09ba9eac9147cf8fd9fce2cfe9f8dce4f7301c4132ca9fb50ce8cbf1df4dc18dd4d210e4345c744ecb3365ed327efdbc52603faa5e21daa11234c8c4a73e51f03bf192544581ebe107adee3a34928e39d04e524a9ce729d5090bfd7dad9d10c722c0def9ccc08ff0a03790e48bcd1f9b6c476063e1966a1c4")]

Add System.IO.Pipelines API #27007

Add System.IO.Pipelines API #27007

Conversation

pakrym commented Feb 9, 2018

pakrym commented Feb 9, 2018

KrzysztofCwalina Feb 9, 2018 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

KrzysztofCwalina Feb 9, 2018 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

pakrym Feb 9, 2018 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

davidfowl Feb 13, 2018 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

ghost commented Feb 13, 2018 via email

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

ahsonkhan left a comment

Choose a reason for hiding this comment

joperezr left a comment

Choose a reason for hiding this comment

KrzysztofCwalina Feb 9, 2018 •

edited

Loading

KrzysztofCwalina Feb 9, 2018 •

edited

Loading

pakrym Feb 9, 2018 •

edited

Loading

davidfowl Feb 13, 2018 •

edited

Loading