Merge pull request #1949 from stephentoub/pipe_buffer_size

Implement System.IO.Pipes buffer sizes on Linux

Author: stephentoub

src/Common/src/Interop/Linux/libc/Interop.FcntlCommands.cs

@@ -0,0 +1,15 @@
+// Copyright (c) Microsoft. All rights reserved.
+// Licensed under the MIT license. See LICENSE file in the project root for full license information.
+
+internal static partial class Interop
+{
+    internal static partial class libc
+    {
+        internal enum FcntlCommands
+        {
+            F_LINUX_SPECIFIC_BASE = 1024,
+            F_SETPIPE_SZ = F_LINUX_SPECIFIC_BASE + 7,
+            F_GETPIPE_SZ = F_LINUX_SPECIFIC_BASE + 8
+        }
+    }
+}

src/Common/src/Interop/Unix/libc/Interop.fcntl.cs

@@ -0,0 +1,21 @@
+// Copyright (c) Microsoft. All rights reserved.
+// Licensed under the MIT license. See LICENSE file in the project root for full license information.
+
+using System;
+using System.Runtime.InteropServices;
+
+internal static partial class Interop
+{
+    internal static partial class libc
+    {
+        // fcntl takes a variable number of arguments, which is difficult to represent in C#.
+        // Instead, since we only have a small and fixed number of call sites, we declare
+        // an overload for each of the specific argument sets we need.
+
+        [DllImport(Libraries.Libc)]
+        internal static extern unsafe int fcntl(int fd, FcntlCommands cmd);
+
+        [DllImport(Libraries.Libc)]
+        internal static extern unsafe int fcntl(int fd, FcntlCommands cmd, int arg1);
+    }
+}

src/System.IO.Pipes/src/System.IO.Pipes.csproj

@@ -206,16 +206,24 @@
   <!-- Linux -->
   <ItemGroup Condition="'$(TargetsLinux)' == 'true'">
     <Compile Include="System\IO\Pipes\AnonymousPipeServerStream.Linux.cs" />
+    <Compile Include="System\IO\Pipes\PipeStream.Linux.cs" />
     <Compile Include="$(CommonPath)\Interop\Linux\Interop.Errors.cs">
       <Link>Common\Interop\Linux\Interop.Errors.cs</Link>
     </Compile>
+    <Compile Include="$(CommonPath)\Interop\Unix\libc\Interop.fcntl.cs">
+      <Link>Common\Interop\Unix\Interop.fcntl.cs</Link>
+    </Compile>
+    <Compile Include="$(CommonPath)\Interop\Linux\libc\Interop.FcntlCommands.cs">
+      <Link>Common\Interop\Linux\Interop.FcntlCommands.cs</Link>
+    </Compile>
     <Compile Include="$(CommonPath)\Interop\Linux\libc\Interop.OpenFlags.cs">
       <Link>Common\Interop\Linux\Interop.OpenFlags.cs</Link>
     </Compile>
   </ItemGroup>
   <!-- OSX -->
   <ItemGroup Condition="'$(TargetsOSX)' == 'true'">
     <Compile Include="System\IO\Pipes\AnonymousPipeServerStream.OSX.cs" />
+    <Compile Include="System\IO\Pipes\PipeStream.OSX.cs" />
     <Compile Include="$(CommonPath)\Interop\OSX\Interop.Errors.cs">
       <Link>Common\Interop\OSX\Interop.Errors.cs</Link>
     </Compile>

src/System.IO.Pipes/src/System/IO/Pipes/AnonymousPipeServerStream.Unix.cs

@@ -38,6 +38,10 @@ private void Create(PipeDirection direction, HandleInheritability inheritability
                 (IntPtr)fds[direction == PipeDirection.In ? Interop.libc.WriteEndOfPipe : Interop.libc.ReadEndOfPipe], 
                 ownsHandle: true);
 
+            // Configure the pipe.  For buffer size, the size applies to the pipe, rather than to 
+            // just one end's file descriptor, so we only need to do this with one of the handles.
+            InitializeBufferSize(serverHandle, bufferSize);
+
             // We're connected.  Finish initialization using the newly created handles.
             InitializeHandle(serverHandle, isExposed: false, isAsync: false);
             _clientHandle = clientHandle;

src/System.IO.Pipes/src/System/IO/Pipes/NamedPipeServerStream.Unix.cs

@@ -99,9 +99,7 @@ public void WaitForConnection()
                 TranslateFlags(_direction, _options, _inheritability), 
                 (int)Interop.libc.Permissions.S_IRWXU);
 
-            // Ignore _inBufferSize and _outBufferSize.  They're optional, and the fcntl F_SETPIPE_SZ for changing 
-            // a pipe's buffer size is Linux specific.
-
+            InitializeBufferSize(serverHandle, _outBufferSize); // there's only one capacity on Linux; just use the out buffer size
             InitializeHandle(serverHandle, isExposed: false, isAsync: (_options & PipeOptions.Asynchronous) != 0);
             State = PipeState.Connected;
         }

src/System.IO.Pipes/src/System/IO/Pipes/PipeStream.Linux.cs

@@ -0,0 +1,33 @@
+// Copyright (c) Microsoft. All rights reserved.
+// Licensed under the MIT license. See LICENSE file in the project root for full license information.
+
+using Microsoft.Win32.SafeHandles;
+
+namespace System.IO.Pipes
+{
+    public abstract partial class PipeStream
+    {
+        internal void InitializeBufferSize(SafePipeHandle handle, int bufferSize)
+        {
+            if (bufferSize > 0)
+            {
+                SysCall(handle, (fd, _, size) => Interop.libc.fcntl(fd, Interop.libc.FcntlCommands.F_SETPIPE_SZ, size), 
+                    IntPtr.Zero, bufferSize);
+            }
+        }
+
+        private int InBufferSizeCore { get { return GetPipeBufferSize(); } }
+
+        private int OutBufferSizeCore { get { return GetPipeBufferSize(); } }
+
+        private int GetPipeBufferSize()
+        {
+            // If we have a handle, get the capacity of the pipe (there's no distinction between in/out direction).
+            // If we don't, the pipe has been created but not yet connected (in the case of named pipes),
+            // so just return the buffer size that was passed to the constructor.
+            return _handle != null ?
+                (int)SysCall(_handle, (fd, _, __) => Interop.libc.fcntl(fd, Interop.libc.FcntlCommands.F_GETPIPE_SZ)) :
+                _outBufferSize;
+        }
+    }
+}

src/System.IO.Pipes/src/System/IO/Pipes/PipeStream.OSX.cs

@@ -0,0 +1,19 @@
+// Copyright (c) Microsoft. All rights reserved.
+// Licensed under the MIT license. See LICENSE file in the project root for full license information.
+
+using Microsoft.Win32.SafeHandles;
+
+namespace System.IO.Pipes
+{
+    public abstract partial class PipeStream
+    {
+        internal void InitializeBufferSize(SafePipeHandle handle, int bufferSize)
+        {
+            // Nop.  We can't configure the capacity, and as it's just advisory, ignore it.
+        }
+
+        private int InBufferSizeCore { get { throw new PlatformNotSupportedException(); } }
+
+        private int OutBufferSizeCore { get { throw new PlatformNotSupportedException(); } }
+    }
+}

src/System.IO.Pipes/src/System/IO/Pipes/PipeStream.Unix.cs

@@ -20,7 +20,7 @@ public abstract partial class PipeStream : Stream
         // Windows, but can't assume a valid handle on Unix.
         internal const bool CheckOperationsRequiresSetHandle = false;
 
-        private const string PipeDirectoryPath = "/tmp/corefxnamedpipes/";
+        private static readonly string PipeDirectoryPath = Path.Combine(Path.GetTempPath(), "corefxnamedpipes");
 
         internal static string GetPipePath(string serverName, string pipeName)
         {
@@ -66,7 +66,7 @@ internal static string GetPipePath(string serverName, string pipeName)
             }
 
             // Return the pipe path
-            return PipeDirectoryPath + pipeName;
+            return Path.Combine(PipeDirectoryPath, pipeName);
         }
 
         /// <summary>Throws an exception if the supplied handle does not represent a valid pipe.</summary>
@@ -213,10 +213,7 @@ public virtual int InBufferSize
                 {
                     throw new NotSupportedException(SR.NotSupported_UnreadableStream);
                 }
-
-                // On Linux this could be retrieved using F_GETPIPE_SZ with fcntl, but that's non-conforming
-                // and works only on recent versions of Linux.  For now, we'll leave this as unsupported.
-                throw new PlatformNotSupportedException();
+                return InBufferSizeCore;
             }
         }
 
@@ -235,9 +232,7 @@ public virtual int OutBufferSize
                 {
                     throw new NotSupportedException(SR.NotSupported_UnwritableStream);
                 }
-
-                // See comments in inBufferSize
-                throw new PlatformNotSupportedException();
+                return OutBufferSizeCore;
             }
         }
 

src/System.IO.Pipes/tests/AnonymousPipesSimpleTest.cs

@@ -3,6 +3,7 @@
 
 using Microsoft.Win32.SafeHandles;
 using System;
+using System.IO;
 using System.IO.Pipes;
 using System.Threading.Tasks;
 using Xunit;
@@ -130,7 +131,7 @@ public static void StartClient(PipeDirection direction, SafePipeHandle clientPip
     public static async Task ServerPInvokeChecks()
     {
         // calling every API related to server and client to detect any bad PInvokes
-        using (AnonymousPipeServerStream server = new AnonymousPipeServerStream(PipeDirection.Out))
+        using (AnonymousPipeServerStream server = new AnonymousPipeServerStream(PipeDirection.Out, HandleInheritability.None, 4096))
         {
             Task clientTask = Task.Run(() => StartClient(PipeDirection.In, server.ClientSafePipeHandle));
 
@@ -141,9 +142,9 @@ public static async Task ServerPInvokeChecks()
             Assert.False(string.IsNullOrWhiteSpace(server.GetClientHandleAsString()));
             Assert.False(server.IsAsync);
             Assert.True(server.IsConnected);
-            if (Interop.IsWindows)
+            if (Interop.IsWindows || Interop.IsLinux)
             {
-                Assert.Equal(0, server.OutBufferSize);
+                Assert.True(server.OutBufferSize > 0);
             }
             else
             {
@@ -177,6 +178,10 @@ public static async Task ServerPInvokeChecks()
             {
                 Assert.Equal(4096, server.InBufferSize);
             }
+            else if (Interop.IsLinux)
+            {
+                Assert.True(server.InBufferSize > 0);
+            }
             else
             {
                 Assert.Throws<PlatformNotSupportedException>(() => server.InBufferSize);
@@ -194,7 +199,7 @@ public static async Task ServerPInvokeChecks()
     [Fact]
     public static async Task ClientPInvokeChecks()
     {
-        using (AnonymousPipeServerStream server = new AnonymousPipeServerStream(PipeDirection.In))
+        using (AnonymousPipeServerStream server = new AnonymousPipeServerStream(PipeDirection.In, System.IO.HandleInheritability.None, 4096))
         {
             using (AnonymousPipeClientStream client = new AnonymousPipeClientStream(PipeDirection.Out, server.ClientSafePipeHandle))
             {
@@ -210,6 +215,10 @@ public static async Task ClientPInvokeChecks()
                 {
                     Assert.Equal(0, client.OutBufferSize);
                 }
+                else if (Interop.IsLinux)
+                {
+                    Assert.True(client.OutBufferSize > 0);
+                }
                 else
                 {
                     Assert.Throws<PlatformNotSupportedException>(() => client.OutBufferSize);
@@ -244,6 +253,10 @@ public static async Task ClientPInvokeChecks()
                 {
                     Assert.Equal(4096, client.InBufferSize);
                 }
+                else if (Interop.IsLinux)
+                {
+                    Assert.True(client.InBufferSize > 0);
+                }
                 else
                 {
                     Assert.Throws<PlatformNotSupportedException>(() => client.InBufferSize);
@@ -258,4 +271,31 @@ public static async Task ClientPInvokeChecks()
             }
         }
     }
+
+    [Fact]
+    [PlatformSpecific(PlatformID.Linux)]
+    public static void BufferSizeRoundtrips()
+    {
+        int desiredBufferSize;
+        using (var server = new AnonymousPipeServerStream(PipeDirection.Out))
+        {
+            desiredBufferSize = server.OutBufferSize * 2;
+            Assert.True(desiredBufferSize > 0);
+        }
+
+        using (var server = new AnonymousPipeServerStream(PipeDirection.Out, HandleInheritability.None, desiredBufferSize))
+        using (var client = new AnonymousPipeClientStream(PipeDirection.In, server.ClientSafePipeHandle))
+        {
+            Assert.Equal(desiredBufferSize, server.OutBufferSize);
+            Assert.Equal(desiredBufferSize, client.InBufferSize);
+        }
+
+        using (var server = new AnonymousPipeServerStream(PipeDirection.In, HandleInheritability.None, desiredBufferSize))
+        using (var client = new AnonymousPipeClientStream(PipeDirection.Out, server.ClientSafePipeHandle))
+        {
+            Assert.Equal(desiredBufferSize, server.InBufferSize);
+            Assert.Equal(desiredBufferSize, client.OutBufferSize);
+        }
+    }
+
 }

src/System.IO.Pipes/tests/NamedPipesSimpleTest.cs

@@ -223,6 +223,10 @@ public static async Task ServerPInvokeChecks()
             {
                 Assert.Equal(0, server.OutBufferSize);
             }
+            else if (Interop.IsLinux)
+            {
+                Assert.True(server.OutBufferSize > 0);
+            }
             else
             {
                 Assert.Throws<PlatformNotSupportedException>(() => server.OutBufferSize);
@@ -255,6 +259,10 @@ public static async Task ServerPInvokeChecks()
             {
                 Assert.Equal(0, server.InBufferSize);
             }
+            else if (Interop.IsLinux)
+            {
+                Assert.True(server.InBufferSize > 0);
+            }
             else
             {
                 Assert.Throws<PlatformNotSupportedException>(() => server.InBufferSize);
@@ -270,7 +278,7 @@ public static async Task ServerPInvokeChecks()
     [Fact]
     public static async Task ClientPInvokeChecks()
     {
-        using (NamedPipeServerStream server = new NamedPipeServerStream("foo", PipeDirection.In))
+        using (NamedPipeServerStream server = new NamedPipeServerStream("foo", PipeDirection.In, 1, PipeTransmissionMode.Byte, PipeOptions.None, 4096, 4096))
         {
             using (NamedPipeClientStream client = new NamedPipeClientStream(".", "foo", PipeDirection.Out))
             {
@@ -287,6 +295,10 @@ public static async Task ClientPInvokeChecks()
                 {
                     Assert.Equal(0, client.OutBufferSize);
                 }
+                else if (Interop.IsLinux)
+                {
+                    Assert.True(client.OutBufferSize > 0);
+                }
                 else
                 {
                     Assert.Throws<PlatformNotSupportedException>(() => client.OutBufferSize);
@@ -322,6 +334,10 @@ public static async Task ClientPInvokeChecks()
                 {
                     Assert.Equal(0, client.InBufferSize);
                 }
+                else if (Interop.IsLinux)
+                {
+                    Assert.True(client.InBufferSize > 0);
+                }
                 else
                 {
                     Assert.Throws<PlatformNotSupportedException>(() => client.InBufferSize);