Introducing System.IO.Pipelines #980

davidfowl · 2016-11-15T08:29:09Z

Golang took the name channels so now we have pipelines

- Golang took the name channels so no we have pipelines

jchannon · 2016-11-15T12:26:07Z

So what if Golang named it first, why can't 2 languages have the same named concept?

If a golang user comes to .net they will immediately know what channels is.

JesperTreetop · 2016-11-15T13:04:58Z

@jchannon Yes, they will have an idea of what channels should be and be highly confused, because channels in Go are a completely different concept. What is now called Pipelines in .NET is an optimized specialization of streams. In Go, "Channels are a typed conduit through which you can send and receive values with the channel operator, <-."

stephentoub · 2016-11-15T13:17:24Z

The project at https://github.com/dotnet/corefxlab/tree/master/src/System.Threading.Tasks.Channels maps much more closely in concept to Go's channel.

stephentoub · 2016-11-15T13:24:29Z

src/System.IO.Pipelines/DefaultReadableBufferExtensions.cs

+                ArraySegment<byte> data;
+                if (memory.TryGetArray(out data))
+                {
+                    await stream.WriteAsync(data.Array, data.Offset, data.Count);


All of the awaits on tasks should use ConfigureAwait(false)

Oh god, ConfigureAwait....

Is this still true?

stephentoub · 2016-11-15T13:25:49Z

src/System.IO.Pipelines/DefaultReadableBufferExtensions.cs

+        /// <returns></returns>
+        public static async Task CopyToAsync(this ReadableBuffer buffer, Stream stream)
+        {
+            foreach (var memory in buffer)


Is it common/likely for buffer to have more than one "memory"? If so, you could special case that and return the result of stream.WriteAsync directly, avoiding the extra async method frame.

stephentoub · 2016-11-15T13:28:27Z

src/System.IO.Pipelines/Gate.cs

+        /// <summary>
+        /// Returns a boolean indicating if the gate is "open"
+        /// </summary>
+        public bool IsCompleted => _gateState == _gateIsOpen;


_gateState == _gateIsOpen [](start = 35, length = 25)

Unless it's misused, checking here and in other places in the file != null should provide the same behavior as == _gateIsOpen, and without needing to load the static field.

I'll file bugs for the Gate/Signal changes. I want to get this in asap since it's a giant "move the code over" PR. I'd prefer if there was a reusable TCS in the framework, these 2 types are basically that.

I'd prefer if there was a reusable TCS in the framework, these 2 types are basically that.

I've considered that several times over the years. I think it'd be reasonable to add something to TCS that would essentially allow it to dump its existing task and create a new one, but that's really not sufficient to replace these custom awaitables... for that, you'd not only need to make TCS reusable, but you'd need to allow a Task to go from completed to not completed, which could potentially break a whole lot of code if used incorrectly. Such a task-resetting feature could only be used in a closed environment where the producer and consumer of the task coordinate, as otherwise the producer won't know when it's safe to reset (the same kind of issue that exists with pooled data structures in general).

Agreed, this is one of those systems. We just need primitives that give the power. Maybe a new namespace? System.*.KnowWhatYouAreDoing

stephentoub · 2016-11-15T13:28:53Z

src/System.IO.Pipelines/Gate.cs

+        public void GetResult()
+        {
+            // Clear the active continuation to "reset" the state of this event
+            Interlocked.Exchange(ref _gateState, null);


Interlocked.Exchange [](start = 12, length = 20)

Why an interlocked?

stephentoub · 2016-11-15T13:32:27Z

src/System.IO.Pipelines/MemoryPool.cs

+        private MemoryPoolBlock Lease(
+            [CallerMemberName] string memberName = "",
+            [CallerFilePath] string sourceFilePath = "",
+            [CallerLineNumber] int sourceLineNumber = 0)


Did you intend to store these in DEBUG-only fields?

We'll need to figure out a way to get these diagnostics without being in debug mode. But these are fine for now

But these are fine for now

How so? They don't appear to be stored or used anywhere, even when compiled for DEBUG, so they don't appear to be useful right now at all, even in DEBUG.

You're right, these are remnants of the past.

khalidabuhakmeh · 2016-11-15T13:32:46Z

I like it!

stephentoub · 2016-11-15T13:34:19Z

src/System.IO.Pipelines/PipelineConnectionExtensions.cs

+                else if (fin)
+                {
+                    return new ValueTask<int>(0);
+                }


Couldn't this just be:

if (actual != 0 || fin) { return new ValueTask<int>(actual); }

?

fin being true doesn't mean the buffer is empty. I'd rather keep the logic split for now and do further testing to make sure that length is 0 when fin is true (in this case).

stephentoub · 2016-11-15T13:39:07Z

src/System.IO.Pipelines/PipelineConnectionStream.cs

+            else
+            {
+                // Needs .AsTask to match Stream's Async method return types
+                _cachedTask = task.AsTask();


_cachedTask = task.AsTask(); [](start = 16, length = 28)

I'm not sure this is a good thing to do. First, the operation could fail, in which case you've cached a task that has an exception, and the next time you check _cachedTask.Result, it'll throw. Second, you could end up in a situation where a call to ReadAsync blocks waiting on the _cachedTask. I think you only want to cache it in the task.IsCompletedSuccessfully case.

stephentoub · 2016-11-15T13:39:49Z

src/System.IO.Pipelines/PipelineConnectionStream.cs

+        public override Task FlushAsync(CancellationToken cancellationToken)
+        {
+            // No-op since writes are immediate.
+            return Task.FromResult(0);


Task.FromResult(0) [](start = 19, length = 18)

Task.CompletedTask

stephentoub · 2016-11-15T13:40:49Z

src/System.IO.Pipelines/PipelineConnectionStream.cs

+                task.ContinueWith(t => callback.Invoke(t));
+            }
+            return task;
+        }


See https://github.com/dotnet/corefx/blob/master/src/Common/src/System/Threading/Tasks/TaskToApm.cs for what corefx uses to turn Tasks into the Begin/End pattern

stephentoub · 2016-11-15T13:45:34Z

src/System.IO.Pipelines/StreamExtensions.cs

+                {
+                    writer.CompleteReader();
+                }
+            });


Would be better as:

writer.CopyToAsync(stream).ContinueWith((t,s) => { var innerWriter = (PipelineReaderWriter)s; if (t.IsFaulted) innerWriter.CompleteReader(task.Exception.InnerException); else innerWriter.CompleteReader(); }, writer, CancellationToken.None, TaskContinuationOptions.ExecuteSynchronously, TaskScheduler.Default);

That way, you avoid the closure/delegate allocation closing over state, the continuation is invoked as part of the antecedent task's completion (rather than queueing another work item, and the continuation will not be forced back to any custom TaskScheduler that was present. Similar comments apply to other uses of ContinueWith in this PR.

Ah yea, I was being lazy here. Though passing of those arguments is one of the reasons I try to avoid continue with...

stephentoub · 2016-11-15T13:47:24Z

src/System.IO.Pipelines.Compression/Inflater.cs

+    /// <summary>
+    /// Provides a wrapper around the ZLib decompression API
+    /// </summary>
+    internal sealed class Inflater : IDisposable


cc: @ianhays... we should think through exposing Inflater/Deflator in corefx publicly, so that other consumers besides DeflateStream can use it directly.

Yes, but we need to design clean low level APIs.

In reply to: 88018514 [](ancestors = 88018514)

stephentoub · 2016-11-15T13:48:15Z

src/System.IO.Pipelines.File/FileReader.cs

+                Value = value;
+            }
+            public T Value { get; set; }
+        }


System.Runtime.CompilerServices.StrongBox

stephentoub · 2016-11-15T13:49:42Z

src/System.IO.Pipelines.File/FileReader.cs

+
+        // Win32 file impl
+        // TODO: Other platforms
+        public unsafe void OpenReadFile(string path)


Do you still need this now that FileStream has an optimized CopyToAsync implementation? What further changes would be necessary there to obviate the need for this here?

This was a proof of concept done way before FileStream.CopyToAsync had those changes. Actually it would be better to use FileStream.CTA adapter directly now (at least on .NET Core) because you get the cross platform benefits. There are some questions as to whether we ship this at all but for now its just an experiment. Ideally we'd have a unix impl so we could see what that looks like.

stephentoub · 2016-11-15T13:51:00Z

src/System.IO.Pipelines.Networking.Sockets/Internal/Signal.cs

+    /// a custom awaiter is provided. Works like a <see cref="ManualResetEvent "/> - the <see cref="Reset"/> method must
+    /// be called between operations.
+    /// </summary>
+    internal class Signal : INotifyCompletion


ICriticalNotifyCompletion

stephentoub · 2016-11-15T13:51:43Z

src/System.IO.Pipelines.Networking.Sockets/Internal/Signal.cs

+
+        public void OnCompleted(Action continuation)
+        {
+            if (continuation != null)


if (continuation != null) [](start = 11, length = 26)

This would only happen if it was misused... the compiler will never pass a null continuation.

stephentoub · 2016-11-15T13:52:08Z

src/System.IO.Pipelines.Networking.Sockets/Internal/Signal.cs

+                var oldValue = Interlocked.CompareExchange(ref _continuation, continuation, null);
+
+                if (ReferenceEquals(oldValue, _completedSentinel))
+                {


Same comments as on Gate about this.

stephentoub · 2016-11-15T13:56:08Z

samples/System.IO.Pipelines.Samples/HttpServer/HttpResponseStream.cs

+        private Task WriteAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken, object state)
+        {
+            var tcs = new TaskCompletionSource<object>(state);
+            var task = WriteAsync(buffer, offset, count, cancellationToken);


var task = WriteAsync(buffer, offset, count, cancellationToken); [](start = 12, length = 64)

In cases like this, where the async operation is likely to complete synchronously, you're better off adding a fast path, e.g.

Task task = WriteAsync(...); if (task.Status == TaskStatus.RanToCompletion) { return task; } var tcs = ... ...

adamralph · 2016-11-15T14:24:34Z

Does it matter that golang might introduce something called pipelines later which is again completely different to this?

nzall · 2016-11-15T14:48:38Z

@adamralph golang already has a concept called pipelines, which is a series of stages (goroutines running the same function) connected by channels. Since there is only a limited set of words in the English vocabulary that are suitable for software purposes, chances are that the good ones are already taken.

omariom · 2016-11-15T15:14:32Z

@davidfowl
What is the allocation profile of such approach compared to the traditional streams?

hammett · 2016-11-15T16:09:56Z

samples/System.IO.Pipelines.Samples/Framing/Codec.cs

+            var listener = new UvTcpListener(thread, new IPEndPoint(ip, port));
+            listener.OnConnection(async connection =>
+            {
+                var channel = MakePipeline(connection);


shouldn't you rename "channel" as well?

KrzysztofCwalina · 2016-11-15T16:14:28Z

using System.Buffers;

All the files should have the standard copyright/license notice

Refers to: src/System.IO.Pipelines/ArrayBufferPool.cs:1 in 56ee6e7. [](commit_id = 56ee6e7, deletion_comment = False)

KrzysztofCwalina · 2016-11-15T16:23:00Z

src/System.IO.Pipelines/PipelineConnectionStream.cs

+
+namespace System.IO.Pipelines
+{
+    public class PipelineConnectionStream : Stream


Stream [](start = 44, length = 6)

I think the adapters should be separated (namespace and possibly package wise) from the pure-pipeline APIs

Suggestions?

System.IO.Pipelines.Bridge? System.IO.Pipelines.Streams?

KrzysztofCwalina · 2016-11-15T16:25:29Z

src/System.IO.Pipelines/PipelineReader.cs

+        /// <summary>
+        /// The underlying <see cref="PipelineReaderWriter"/> the <see cref="PipelineReader"/> communicates over.
+        /// </summary>
+        protected readonly PipelineReaderWriter _input;


Yeah, the fact that reader wraps reader/writers shows that we don't yet have a good name for the reader/writer. Having said that, I think we should merge as-is and think about it later.

hammett · 2016-11-15T16:38:20Z

src/System.IO.Pipelines.Networking.Libuv/Interop/SockAddr.cs

+
+namespace System.IO.Pipelines.Networking.Libuv.Interop
+{
+    public struct SockAddr


No need for struct layout?

KrzysztofCwalina · 2016-11-15T16:39:23Z

🕐

adamralph · 2016-11-15T16:50:53Z

With reference to @stephentoub's comment, is the implication that System.IO.Pipelines aligns more closely with golang pipelines (referenced by @nzall)?

stephentoub · 2016-11-15T16:59:09Z

With reference to @stephentoub's comment, is the implication that System.IO.Pipelines aligns more closely with golang pipelines (referenced by @nzall)?

A "pipeline" is a general pattern, a series of processing elements (often called stages) with data handed from one stage to the next. A pipeline can be implemented in many languages/frameworks; in Go, it's typically implemented using a goroutine for each stage and using a channel as the mechanism for handing data between stages. But the concept is in no way specific to Go.

adamralph · 2016-11-15T19:05:30Z

@stephentoub thanks for the explanation. I'm trying to rationalise the rename from "channels" to "pipelines". I can see how the rename probably makes sense in this case, but basing the decision on another platform having "taken" the generic noun you intended to use feels a bit off somehow.

JesperTreetop · 2016-11-16T10:51:26Z

@adamralph Look at the repository @stephentoub linked to - they're not changing the name of this thing "because we can't use names taken in other languages", but because they already are testing something which would be closer to channels and want to name that thing channels instead.

adamralph · 2016-11-16T11:02:10Z

@JesperTreetop that sounds like a good reason. It's a very different reason to that given by the OP.

dazinator · 2016-11-18T01:08:49Z

Couldn't resist commenting.. I have literally no idea what this is.. but I love it +1. Look forward to working it out through experimentation :-)

malachib · 2016-12-09T17:42:14Z

I'm excited about this, it looks like a potentially fresher and better tool than Streams, especially for in-process / short-hop data movement

dazinator · 2016-12-09T18:06:45Z

samples/System.IO.Pipelines.Samples/CompressionSample.cs

+                // Copy from the file reader to the console writer
+                input.CopyToAsync(output).GetAwaiter().GetResult();
+
+                input.Complete();


What happens if you forget to call Complete() ? Do you think there would be any benefit if IPipelineReader / 'IPipelineReader' implemented IDisposable, and called Complete() in it's dispose?

Introducing System.IO.Pipelines

1a81deb

- Golang took the name channels so no we have pipelines

dnfclas added the cla-already-signed label Nov 15, 2016

Remove remaining traces of channels

56ee6e7

stephentoub reviewed Nov 15, 2016

View reviewed changes

hammett reviewed Nov 15, 2016

View reviewed changes

KrzysztofCwalina reviewed Nov 15, 2016

View reviewed changes

hammett reviewed Nov 15, 2016

View reviewed changes

Few code review comments and tweaks

d7ef7b1

davidfowl merged commit df543b2 into master Nov 15, 2016

This was referenced Nov 15, 2016

Consolidate Gate and Signal #981

Closed

Consider not caching the resultant Task in PipelineConnectionStream.ReadAsync #982

Closed

SeanFeldman mentioned this pull request Nov 16, 2016

Cleanup use of ConfigureAwait (Discussion) HTBox/allReady#1313

Closed

JvanderStad mentioned this pull request Nov 29, 2016

Logo for Span<T> should be Spanatee the Manatee #1034

Closed

dazinator reviewed Dec 9, 2016

View reviewed changes

davidfowl deleted the davidfowl/pipelines branch December 23, 2016 16:20

Introducing System.IO.Pipelines #980

Introducing System.IO.Pipelines #980

Conversation

davidfowl commented Nov 15, 2016 • edited Loading

jchannon commented Nov 15, 2016

JesperTreetop commented Nov 15, 2016

stephentoub commented Nov 15, 2016

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

khalidabuhakmeh commented Nov 15, 2016

stephentoub Nov 15, 2016 • 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

adamralph commented Nov 15, 2016

nzall commented Nov 15, 2016

omariom commented Nov 15, 2016

Choose a reason for hiding this comment

KrzysztofCwalina commented Nov 15, 2016

KrzysztofCwalina Nov 15, 2016 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

KrzysztofCwalina Nov 15, 2016 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

KrzysztofCwalina commented Nov 15, 2016

adamralph commented Nov 15, 2016

stephentoub commented Nov 15, 2016 • edited Loading

adamralph commented Nov 15, 2016

JesperTreetop commented Nov 16, 2016

adamralph commented Nov 16, 2016

dazinator commented Nov 18, 2016

malachib commented Dec 9, 2016

dazinator Dec 9, 2016 • edited Loading

Choose a reason for hiding this comment

davidfowl commented Nov 15, 2016 •

edited

Loading

stephentoub Nov 15, 2016 •

edited

Loading

KrzysztofCwalina Nov 15, 2016 •

edited

Loading

KrzysztofCwalina Nov 15, 2016 •

edited

Loading

stephentoub commented Nov 15, 2016 •

edited

Loading

dazinator Dec 9, 2016 •

edited

Loading