Corefxlab: IPipeConnection is a Disposable IPipe

[matthewDDennis]

I'm doing some code reading, working up from IPipe and it looks like IPipeConnection is just a Disposable IPipe except for the names of the IPipeReader and IPipeWriter properties.

Conceptually, I think this really would show what an IPipeConnection is, a Pipe.

This would probably simplify some to the code as you would not need separate extension methods for the same functionality.

[davidfowl]

Sorta, they look the same but are kinda different. I actually don't know how important it is to keep IPipeConnection. Regardless, the entire assemblies needs clean up. We're currently in the process of moving the less core "baked" APIs into other assemblies for further research.

IPipe logically represents 2 ends of a single pipe end. IPipeConnection represents a duplex connection, logically Input and Output are both IPipe. IPipeConnection might move into another assembly and the interface might change until further notice. In fact I think I'll do that now.

Also being disposable was about representing things like sockets. Usually, a single end of a pipe gets cleaned up with both the reader and the writer are completed

[matthewDDennis]

Further thoughts,
Actually, an IPipeConnection is a thing the contains two Pipes, only exposing the PipeWriter for the Output Pipe, and the PipeReader for the Input PipeReader. The other ends are read and written from/to by something else such as Socket read and write tasks.
However, there is nothing in the definition of IPipe that says that what goes in the Writer must come out untransformed from the Reader, or that there is any relationship between the Input and the Output.
A Pipe is an IPipe that just has a buffer between the input and output, a very nice buffer at that.
A PipeConnection is an IPipe that has two internal Pipes with the nonexposed ends being read/written by something else. From the application point of view, it still looks like an IPipe.

[davidfowl]

That's true, the interface says nothing about the usage. Let me try removing IPipeConnection completely.

[matthewDDennis]

Be careful, you don't want to lose the IDisposable to ensure connections get cleaned up. Maybe IPipe should be disposable.

[davidfowl]

@matthewDDennis it only matters if we eventually end up making a transport layer abstraction. What is there now is inadequate anyways and needs to be overhauled. Also, with the current server side APIs dispose never needs to be called explicitly, returning from the connection loop disposes internally.

I also didn't want arbitrary code disposing the connection at random places while there are outstanding readers and writers.

[davidfowl]

The biggest problem with that is interop with things like Stream that support out of band disposal.

[davidfowl]

/cc @pakrym @halter73

[Drawaes]

Watch that something like Tls needs managed resources outside the loop dispose comes into play for the stateful connection

[matthewDDennis]

so we would probably need to recommend that things that contain an IPipe do
~ c#
(myPipe as IDisposable)?.Dispose();
~
in case the Pipe has things that should be disposed, like Sockets.

[matthewDDennis]

I notice that the internal Pipe class has a Dispose method, but does not inherit IDisposable.

I think that this is a good indication that IPipe probably should be Disposable, if for nothing other than to ensure that the Pipe returns its buffers to the buffer pool.

[davidfowl]

Watch that something like Tls needs managed resources outside the loop dispose comes into play for the stateful connection

Point me to an example.

in case the Pipe has things that should be disposed, like Sockets.

Not sure what you mean. You can't dispose the pipe.

I notice that the internal Pipe class has a Dispose method, but does not inherit IDisposable. I think that this is a good indication that IPipe probably should be Disposable, if for nothing other than to ensure that the Pipe returns its buffers to the buffer pool.

If you make things IDisposable people will dispose them (probably incorrectly). I don't have any issues doing this after being shown enough overwhelming examples that point in that direction. If you're disposing the pipe out of band, you probably have other issues as it's non trivial to get right. So show me where you are needing to call dispose.

[davidfowl]

Another approach would be to do the safe handle thing and keep track of the dispose call and defer it until it's safe to do so. That may not work well if you were actively trying to clean up resources though, but it would save you from shooting yourself in the foot.

I like the pattern much more now because it really makes you think about when to "Complete" each side of the pipe. Only after you're done reading and writing. Anything else at this point is questionable. Take a look at how dispose is implemented in the transport layers:

https://github.com/dotnet/corefxlab/blob/c65d7ce14edd6772adb9c475255841f1c81141fa/src/System.IO.Pipelines.Networking.Sockets/SocketConnection.cs#L187-L194

https://github.com/dotnet/corefxlab/blob/c65d7ce14edd6772adb9c475255841f1c81141fa/src/System.IO.Pipelines.Networking.Sockets/SocketListener.cs#L140-L154

In this model, the user code never explicitly calls dispose, it just unwinds from an async read loop.

Now you can imagine some cases (like in kestrel) like server shutdown, where connections need to be torn down, but that is done at the connection layer. The underlying transport is torn down so that the read/write loop unwind and complete themselves as they should.

[matthewDDennis]

looking at the code for Pipe and SocketConnection I see that there is a somewhat complex unwinding of the reader and writer tasks by calling the appropriate Complete methods. The Complete methods end up calling one of the Pipe.DisposeXXX(...) methods to release the buffer memory segments back to the Pool.
Besides violating the Single Responsibility Principle, it could result in memory leaks on bad exception handling or coding.
If IPipe was Disposable, then classes that contain IPipe should be Disposable, and the Analyzers will warn of this.
In some cases, it may not be desirable or possible to unwind the loop and just need to

• kill the reader/writer tasks
• kill/dispose of the socket or connection,
• release the pipe buffers back to the pool (Pipe.Dispose)

using the reader/writer.Complete methods is just one way of accomplishing the first two. It is not obvious that the Complete methods will Dispose the Pipe's buffers.

Question, can a Pipe be Reset so it can be used after a call to Complete or is it effectively Disposed and a new Pipe must be created?

I think that the IDisposable pattern will prevent more issues that it will cause and reduce the learning curve.

[matthewDDennis]

I looked at the SocketConnection.Dispose method and you are really using the Complete methods to Dispose of the Pipes.
Really, the Pipe's Dispose should call the Complete to prevent further Read/Writes when the Pipe has been disposed. It seems like an implementation detail of the Pipe is leaking up into the SocketConnection for Disposal.

[davidfowl]

looking at the code for Pipe and SocketConnection I see that there is a somewhat complex unwinding of the reader and writer tasks by calling the appropriate Complete methods. The Complete methods end up calling one of the Pipe.DisposeXXX(...) methods to release the buffer memory segments back to the Pool.

The code is complex but this code usually is, and if you're implementing a transport you have to be aware of this stuff. On the other hand, if you're just consuming the pipe, why should you care? All you do is complete when you're done.

Question, can a Pipe be Reset so it can be used after a call to Complete or is it effectively Disposed and a new Pipe must be created?

Not today, but the plan is to make it resettable to reduce memory pressure in some cases.

I think that the IDisposable pattern will prevent more issues that it will cause and reduce the learning curve.

I'm not convinced yet, as I said, show me some real code samples.

[benaadams]

Really, the Pipe's Dispose should call the Complete to prevent further Read/Writes when the Pipe has been disposed.

If the writer completes the pipe doesn't want to stop the reads; it just means the writer is not going to write any more.

[davidfowl]

I looked at the SocketConnection.Dispose method and you are really using the Complete methods to Dispose of the Pipes.

Complete is being called here to prevent the consumer from leaking, but TBH it's not the responsibility of the socket to take care of the consuming logic. That should actually be moved into the listener callback. When the callback returns, the consuming half of the pipe should be closed. In fact i'll make that change.

Really, the Pipe's Dispose should call the Complete to prevent further Read/Writes when the Pipe has been disposed. It seems like an implementation detail of the Pipe is leaking up into the SocketConnection for Disposal.

Maybe, as I said, I'm not convinced yet. It started off as IDisposable and slowly changed to use this Complete model instead. It is more user friendly.

[davidfowl]

That should actually be moved into the listener callback. When the callback returns, the consuming half of the pipe should be closed. In fact i'll make that change

Just took another look and actually it's fine as is.

[Drawaes]

Thinking through real situations more I think it's right to complete each side. TCP for example you can shutdown one side of a socket. In TLS if you throw an alert you want to send and shutdown writing but you want the socket to drain the alert not dispose and release the buffers and die.

[matthewDDennis]

In a previous issue #1249 showed a TestConnection I created for testing that the correct data was being sent to a Pipe or the something was processing data correctly from a Pipe. (shown below)
First of all, is this correct for the current implementation?
Secondly, if IPipe was disposable, then the Dispose method would become
~~~ c#
public void Dispose()
{
_output.Dispose();
_input.Dispose();

        if (_ownsFactory) { _factory?.Dispose(); }
        _factory = null;
    }

~~~
and the Pipe takes care of whatever is required to clean up the pipe state to make it safe. This means the cleanup code is in one place and doesn't need to be duplicated.

TestConnection

~~~ c#
using System;
using System.Collections.Generic;
using System.IO.Pipelines;
using System.Text;

namespace Munq.Redis.Client.Tests
{
public class TestConnection : IPipeConnection
{
private readonly bool _ownsFactory;
private PipeFactory _factory;
private IPipe _input, _output;

    public TestConnection(PipeFactory factory = null)
    {
        if (factory == null)
        {
            _ownsFactory = true;
            factory = new PipeFactory();
        }
        _factory = factory;

        _input  = _factory.Create();
        _output = _factory.Create();
    }
    public IPipeReader Input => _input.Reader;

    public IPipeWriter Output => _output.Writer;

    public IPipeReader RemoteInput => _output.Reader;

    public IPipeWriter RemoteOutput => _input.Writer;

    public void Dispose()
    {
        _output.Reader.CancelPendingRead();
        _input.Reader.CancelPendingRead();
        _output.Writer.Complete();
        _input.Writer.Complete();

        if (_ownsFactory) { _factory?.Dispose(); }
        _factory = null;
    }
}

}
~~~~

[davidfowl]

I haven't heard acknowledgement of any of the other arguments stated. Just the fact that dispose should exist.

[Drawaes]

Problem is this is a test scenario and if you cancel a pending read won't most reading loops re-read? So is there a race there?

[davidfowl]

Pipe.Dispose just returns the blocks, it doesn't wait for anything to unwind. That logic in the socket connection wouldn't go away. Also, who is unwinding in the test connection? Is that logic pushed into every test?

[matthewDDennis]

The TestConnection is disposed with a using statement.
~~~ c#
[Fact]
public async Task CanWriteASimpleCommand()
{
using (var connection = new TestConnection())
{
var expected = (Utf8String)"*1rn$4Pingrn";
using (var redisConnection = new RedisConnection(connection, null, 1))
{
await redisConnection.WriteCommandAsync("Ping");

                var readBufferAwaitable = await connection.RemoteInput.ReadAsync();
                var buffer = readBufferAwaitable.Buffer;
                var actual = new Utf8String(buffer.First.Span);

                Assert.Equal(expected, actual);
            }
        }
    }

~~~

[davidfowl]

What happens in RedisConnection.Dispose()

[matthewDDennis]

~~~ c#
using System.IO.Pipelines;
using System.Net;

namespace Munq.Redis.Client
{
///

/// This class implements the connection to the Redis Server by wrapping an
/// IPipeConnection, usually a SocketConnection. The wrapped connection is used
/// to stream data to and from the Redis server.
///

public class RedisConnection : IRedisConnection
{
private IPipeConnection _connection;

    public RedisConnection(IPipeConnection connection, IPEndPoint serverEndpoint, int database)
    {
        _connection        = connection;
        Database           = database;
        ServerEndPoint     = serverEndpoint;
        IsDatabaseSelected = false;
    }

    /// <summary>
    /// Gets the EndPoint for the connection to the Redis Server.
    /// </summary>
    public IPEndPoint ServerEndPoint  { get; internal set; }

    /// <summary>
    /// Gets the Database that the connection is talking to.
    /// </summary>
    /// <remarks>The Select command will change this.</remarks>
    public int Database { get; internal set; }

    /// <summary>
    /// Gets a value indicating whether the Database has been selected.
    /// </summary>
    public bool IsDatabaseSelected { get; internal set; }

    /// <summary>
    /// Gets the Input PipeReader from the underlying IPipeConnection.
    /// </summary>
    public IPipeReader Input => _connection.Input;

    /// <summary>
    /// Gets the Output PipeWriter from the underlying IPipeConnection.
    /// </summary>
    public IPipeWriter Output => _connection.Output;

    /// <summary>
    /// Disposes of the Connection.
    /// </summary>
    public void Dispose()
    {
        _connection?.Dispose();
        _connection = null;
    }
}

}
~~~

[matthewDDennis]

Oh yeah, IRedisConnection
~~~ c#
using System.IO.Pipelines;
using System.Net;

namespace Munq.Redis.Client
{
public interface IRedisConnection : IPipeConnection
{
IPEndPoint ServerEndPoint { get; }
int Database { get; }
bool IsDatabaseSelected { get; }
}
}
~~~

[matthewDDennis]

@benaadams, In other words, you need to be able to Complete either or both of the Pipes independent of each other and Disposing of the resources used by the IPipeConnection or IPipe.

In my RedisConnection, It just wraps another IPipeConnection, so I just want to call the Dispose on the wrapped connection in the RedisConnectoin.Dispose. I shouldn't have to know anything about the wrapped connection to dispose of it.

This makes my RedisConnection more testable as I can use any IPipeConnection to test with, including my TestConnection.

[halter73]

@davidfowl If we don't remove IPipeConnection which is understandable since it seems useful for representing a full-duplex connection, I think we should no longer make it implement IDisposable to avoid this kind of confusion.

Instead, the pattern we should encourage is calling both Input.Complete() and Output.Complete() as the one true way to clean up resources. Once both of these methods are called, any connection resources can be cleaned up. Having a Dispose method muddies the waters concerning when and where cleanup should occur.

[davidfowl]

@davidfowl If we don't remove IPipeConnection which is understandable since it seems useful for representing a full-duplex connection, I think we should no longer make it implement IDisposable to avoid this kind of confusion.

+1

[Drawaes]

Having fiddled with channels then pipelines the shutdown has always been a bit murky to me but this issued has clarified so
+1 for me

[matthewDDennis]

But what if you wish to reuse the Connection. I see using a connection, getting and receiving data, completing input and output, but not disconnection the sockets so the live connection can be pooled.
Later, the connection is 'Reset' so the reading and writing can happen on the connection.

In this case , the disposal of the socket resources is separate from the completion.

[Drawaes]

Just don't complete on the socket when you exit your reading/writing loop... pass the socket to the next consumer