From 0c1e4a09ecd7e97381ac98cb3d73902300fa9ee1 Mon Sep 17 00:00:00 2001 From: John Starks Date: Mon, 14 Mar 2016 16:33:35 -0700 Subject: [PATCH 1/3] Add tests that check for EOF on pipe close --- pipe_test.go | 33 +++++++++++++++++++++++++++++++++ 1 file changed, 33 insertions(+) diff --git a/pipe_test.go b/pipe_test.go index 67c86ce..da0e708 100644 --- a/pipe_test.go +++ b/pipe_test.go @@ -2,6 +2,7 @@ package winio import ( "bufio" + "io" "net" "os" "syscall" @@ -174,6 +175,38 @@ func TestCloseAbortsListen(t *testing.T) { } } +func ensureEOFOnClose(t *testing.T, r io.Reader, w io.Closer) { + b := make([]byte, 10) + w.Close() + n, err := r.Read(b) + if n > 0 { + t.Errorf("unexpected byte count %d", n) + } + if err != io.EOF { + t.Errorf("expected EOF: %v", err) + } +} + +func TestCloseClientEOFServer(t *testing.T) { + c, s, err := getConnection() + if err != nil { + t.Fatal(err) + } + defer c.Close() + defer s.Close() + ensureEOFOnClose(t, c, s) +} + +func TestCloseServerEOFClient(t *testing.T) { + c, s, err := getConnection() + if err != nil { + t.Fatal(err) + } + defer c.Close() + defer s.Close() + ensureEOFOnClose(t, s, c) +} + func TestAcceptAfterCloseFails(t *testing.T) { l, err := ListenPipe(testPipeName, "") if err != nil { From 36f4e701c46e5cf84d27ad6c5d6807ade4862027 Mon Sep 17 00:00:00 2001 From: John Starks Date: Mon, 14 Mar 2016 18:27:54 -0700 Subject: [PATCH 2/3] Add support for CloseWrite() to pipes This change adds support for message mode pipes and uses them to support CloseWrite() to better match TCP and UNIX sockets. Message mode pipes support writing and (optionally) reading data in message-sized chunks. This is useful for us because when in this mode a zero-sized message can be read. We use this zero-sized message to signal that no more writes will arrive. This is not standard practice in Windows, but it is a reasonable compromise. --- pipe.go | 164 +++++++++++++++++++++++++++++++++++++++++++-------- pipe_test.go | 52 ++++++++++++---- zsyscall.go | 26 ++++++++ 3 files changed, 205 insertions(+), 37 deletions(-) diff --git a/pipe.go b/pipe.go index 0e398ff..2b9c0cb 100644 --- a/pipe.go +++ b/pipe.go @@ -2,6 +2,7 @@ package winio import ( "errors" + "io" "net" "os" "syscall" @@ -13,6 +14,8 @@ import ( //sys createNamedPipe(name string, flags uint32, pipeMode uint32, maxInstances uint32, outSize uint32, inSize uint32, defaultTimeout uint32, sa *securityAttributes) (handle syscall.Handle, err error) [failretval==syscall.InvalidHandle] = CreateNamedPipeW //sys createFile(name string, access uint32, mode uint32, sa *securityAttributes, createmode uint32, attrs uint32, templatefile syscall.Handle) (handle syscall.Handle, err error) [failretval==syscall.InvalidHandle] = CreateFileW //sys waitNamedPipe(name string, timeout uint32) (err error) = WaitNamedPipeW +//sys getNamedPipeInfo(pipe syscall.Handle, flags *uint32, outSize *uint32, inSize *uint32, maxInstances *uint32) (err error) = GetNamedPipeInfo +//sys getNamedPipeHandleState(pipe syscall.Handle, state *uint32, curInstances *uint32, maxCollectionCount *uint32, collectDataTimeout *uint32, userName *uint16, maxUserNameSize uint32) (err error) = GetNamedPipeHandleStateW type securityAttributes struct { Length uint32 @@ -36,11 +39,18 @@ const ( cNMPWAIT_USE_DEFAULT_WAIT = 0 cNMPWAIT_NOWAIT = 1 + + cPIPE_TYPE_MESSAGE = 4 + + cPIPE_READMODE_MESSAGE = 2 ) var ( - // This error should match net.errClosing since docker takes a dependency on its text + // ErrPipeListenerClosed is returned for pipe operations on listeners that have been closed. + // This error should match net.errClosing since docker takes a dependency on its text. ErrPipeListenerClosed = errors.New("use of closed network connection") + + errPipeWriteClosed = errors.New("pipe has been closed for write") ) type win32Pipe struct { @@ -48,6 +58,12 @@ type win32Pipe struct { path string } +type win32MessageBytePipe struct { + win32Pipe + writeClosed bool + readEOF bool +} + type pipeAddress string func (f *win32Pipe) LocalAddr() net.Addr { @@ -64,6 +80,49 @@ func (f *win32Pipe) SetDeadline(t time.Time) error { return nil } +// CloseWrite closes the write side of a message pipe in byte mode. +func (f *win32MessageBytePipe) CloseWrite() error { + if f.writeClosed { + return errPipeWriteClosed + } + _, err := f.win32File.Write(nil) + if err != nil { + return err + } + f.writeClosed = true + return nil +} + +// Write writes bytes to a message pipe in byte mode. Zero-byte writes are ignored, since +// they are used to implement CloseWrite(). +func (f *win32MessageBytePipe) Write(b []byte) (int, error) { + if f.writeClosed { + return 0, errPipeWriteClosed + } + if len(b) == 0 { + return 0, nil + } + return f.win32File.Write(b) +} + +// Read reads bytes from a message pipe in byte mode. A read of a zero-byte message on a message +// mode pipe will return io.EOF, as will all subsequent reads. +func (f *win32MessageBytePipe) Read(b []byte) (int, error) { + if f.readEOF { + return 0, io.EOF + } + n, err := f.win32File.Read(b) + if err == io.EOF { + // If this was the result of a zero-byte read, then + // it is possible that the read was due to a zero-size + // message. Since we are simulating CloseWrite with a + // zero-byte message, ensure that all future Read() calls + // also return EOF. + f.readEOF = true + } + return n, err +} + func (s pipeAddress) Network() string { return "pipe" } @@ -72,14 +131,6 @@ func (s pipeAddress) String() string { return string(s) } -func makeWin32Pipe(h syscall.Handle, path string) (*win32Pipe, error) { - f, err := makeWin32File(h) - if err != nil { - return nil, err - } - return &win32Pipe{f, path}, nil -} - // DialPipe connects to a named pipe by path, timing out if the connection // takes longer than the specified duration. If timeout is nil, then the timeout // is the default timeout established by the pipe server. @@ -113,18 +164,43 @@ func DialPipe(path string, timeout *time.Duration) (net.Conn, error) { } } if err != nil { - return nil, &os.PathError{"open", path, err} + return nil, &os.PathError{Op: "open", Path: path, Err: err} } - p, err := makeWin32Pipe(h, path) + + var flags uint32 + err = getNamedPipeInfo(h, &flags, nil, nil, nil) + if err != nil { + return nil, err + } + + var state uint32 + err = getNamedPipeHandleState(h, &state, nil, nil, nil, nil, 0) + if err != nil { + return nil, err + } + + if state&cPIPE_READMODE_MESSAGE != 0 { + return nil, &os.PathError{Op: "open", Path: path, Err: errors.New("message readmode pipes not supported")} + } + + f, err := makeWin32File(h) if err != nil { syscall.Close(h) return nil, err } - return p, nil + + // If the pipe is in message mode, return a message byte pipe, which + // supports CloseWrite(). + if flags&cPIPE_TYPE_MESSAGE != 0 { + return &win32MessageBytePipe{ + win32Pipe: win32Pipe{win32File: f, path: path}, + }, nil + } + return &win32Pipe{win32File: f, path: path}, nil } type acceptResponse struct { - p *win32Pipe + f *win32File err error } @@ -132,39 +208,46 @@ type win32PipeListener struct { firstHandle syscall.Handle path string securityDescriptor []byte + config PipeConfig acceptCh chan (chan acceptResponse) closeCh chan int doneCh chan int } -func makeServerPipeHandle(path string, securityDescriptor []byte, first bool) (syscall.Handle, error) { +func makeServerPipeHandle(path string, securityDescriptor []byte, c *PipeConfig, first bool) (syscall.Handle, error) { var flags uint32 = cPIPE_ACCESS_DUPLEX | syscall.FILE_FLAG_OVERLAPPED if first { flags |= cFILE_FLAG_FIRST_PIPE_INSTANCE } + + var mode uint32 = cPIPE_REJECT_REMOTE_CLIENTS + if c.MessageMode { + mode |= cPIPE_TYPE_MESSAGE + } + var sa securityAttributes sa.Length = uint32(unsafe.Sizeof(sa)) if securityDescriptor != nil { sa.SecurityDescriptor = &securityDescriptor[0] } - h, err := createNamedPipe(path, flags, cPIPE_REJECT_REMOTE_CLIENTS, cPIPE_UNLIMITED_INSTANCES, 4096, 4096, 0, &sa) + h, err := createNamedPipe(path, flags, mode, cPIPE_UNLIMITED_INSTANCES, 4096, 4096, 0, &sa) if err != nil { - return 0, &os.PathError{"open", path, err} + return 0, &os.PathError{Op: "open", Path: path, Err: err} } return h, nil } -func (l *win32PipeListener) makeServerPipe() (*win32Pipe, error) { - h, err := makeServerPipeHandle(l.path, l.securityDescriptor, false) +func (l *win32PipeListener) makeServerPipe() (*win32File, error) { + h, err := makeServerPipeHandle(l.path, l.securityDescriptor, &l.config, false) if err != nil { return nil, err } - p, err := makeWin32Pipe(h, l.path) + f, err := makeWin32File(h) if err != nil { syscall.Close(h) return nil, err } - return p, nil + return f, nil } func (l *win32PipeListener) listenerRoutine() { @@ -207,18 +290,37 @@ func (l *win32PipeListener) listenerRoutine() { close(l.doneCh) } -func ListenPipe(path, sddl string) (net.Listener, error) { +// PipeConfig contain configuration for the pipe listener. +type PipeConfig struct { + // SecurityDescriptor contains a Windows security descriptor in SDDL format. + SecurityDescriptor string + + // MessageMode determines whether the pipe is in byte or message mode. In either + // case the pipe is read in byte mode by default. The only practical difference in + // this implementation is that CloseWrite() is only supported for message mode pipes; + // CloseWrite() is implemented as a zero-byte write, but zero-byte writes are only + // transferred to the reader (and returned as io.EOF in this implementation) + // when the pipe is in message mode. + MessageMode bool +} + +// ListenPipe creates a listener on a Windows named pipe path, e.g. \\.\pipe\mypipe. +// The pipe must not already exist. +func ListenPipe(path string, c *PipeConfig) (net.Listener, error) { var ( sd []byte err error ) - if sddl != "" { - sd, err = SddlToSecurityDescriptor(sddl) + if c == nil { + c = &PipeConfig{} + } + if c.SecurityDescriptor != "" { + sd, err = SddlToSecurityDescriptor(c.SecurityDescriptor) if err != nil { return nil, err } } - h, err := makeServerPipeHandle(path, sd, true) + h, err := makeServerPipeHandle(path, sd, c, true) if err != nil { return nil, err } @@ -234,6 +336,7 @@ func ListenPipe(path, sddl string) (net.Listener, error) { firstHandle: h, path: path, securityDescriptor: sd, + config: *c, acceptCh: make(chan (chan acceptResponse)), closeCh: make(chan int), doneCh: make(chan int), @@ -242,7 +345,7 @@ func ListenPipe(path, sddl string) (net.Listener, error) { return l, nil } -func connectPipe(p *win32Pipe) error { +func connectPipe(p *win32File) error { c, err := p.prepareIo() if err != nil { return err @@ -260,7 +363,16 @@ func (l *win32PipeListener) Accept() (net.Conn, error) { select { case l.acceptCh <- ch: response := <-ch - return response.p, response.err + err := response.err + if err != nil { + return nil, err + } + if l.config.MessageMode { + return &win32MessageBytePipe{ + win32Pipe: win32Pipe{win32File: response.f, path: l.path}, + }, nil + } + return &win32Pipe{win32File: response.f, path: l.path}, nil case <-l.doneCh: return nil, ErrPipeListenerClosed } diff --git a/pipe_test.go b/pipe_test.go index da0e708..49dad98 100644 --- a/pipe_test.go +++ b/pipe_test.go @@ -20,7 +20,7 @@ func TestDialUnknownFailsImmediately(t *testing.T) { } func TestDialListenerTimesOut(t *testing.T) { - l, err := ListenPipe(testPipeName, "") + l, err := ListenPipe(testPipeName, nil) if err != nil { t.Fatal(err) } @@ -33,7 +33,10 @@ func TestDialListenerTimesOut(t *testing.T) { } func TestDialAccessDeniedWithRestrictedSD(t *testing.T) { - l, err := ListenPipe(testPipeName, "D:P(A;;0x1200FF;;;WD)") + c := PipeConfig{ + SecurityDescriptor: "D:P(A;;0x1200FF;;;WD)", + } + l, err := ListenPipe(testPipeName, &c) if err != nil { t.Fatal(err) } @@ -44,8 +47,8 @@ func TestDialAccessDeniedWithRestrictedSD(t *testing.T) { } } -func getConnection() (client net.Conn, server net.Conn, err error) { - l, err := ListenPipe(testPipeName, "") +func getConnection(cfg *PipeConfig) (client net.Conn, server net.Conn, err error) { + l, err := ListenPipe(testPipeName, cfg) if err != nil { return } @@ -78,7 +81,7 @@ func getConnection() (client net.Conn, server net.Conn, err error) { } func TestReadTimeout(t *testing.T) { - c, s, err := getConnection() + c, s, err := getConnection(nil) if err != nil { t.Fatal(err) } @@ -117,7 +120,7 @@ func server(l net.Listener, ch chan int) { } func TestFullListenDialReadWrite(t *testing.T) { - l, err := ListenPipe(testPipeName, "") + l, err := ListenPipe(testPipeName, nil) if err != nil { t.Fatal(err) } @@ -155,7 +158,7 @@ func TestFullListenDialReadWrite(t *testing.T) { } func TestCloseAbortsListen(t *testing.T) { - l, err := ListenPipe(testPipeName, "") + l, err := ListenPipe(testPipeName, nil) if err != nil { t.Fatal(err) } @@ -188,7 +191,7 @@ func ensureEOFOnClose(t *testing.T, r io.Reader, w io.Closer) { } func TestCloseClientEOFServer(t *testing.T) { - c, s, err := getConnection() + c, s, err := getConnection(nil) if err != nil { t.Fatal(err) } @@ -198,7 +201,7 @@ func TestCloseClientEOFServer(t *testing.T) { } func TestCloseServerEOFClient(t *testing.T) { - c, s, err := getConnection() + c, s, err := getConnection(nil) if err != nil { t.Fatal(err) } @@ -207,8 +210,35 @@ func TestCloseServerEOFClient(t *testing.T) { ensureEOFOnClose(t, s, c) } +func TestCloseWriteEOF(t *testing.T) { + cfg := &PipeConfig{ + MessageMode: true, + } + c, s, err := getConnection(cfg) + if err != nil { + t.Fatal(err) + } + defer c.Close() + defer s.Close() + + type closeWriter interface { + CloseWrite() error + } + + err = c.(closeWriter).CloseWrite() + if err != nil { + t.Fatal(err) + } + + b := make([]byte, 10) + _, err = s.Read(b) + if err != io.EOF { + t.Fatal(err) + } +} + func TestAcceptAfterCloseFails(t *testing.T) { - l, err := ListenPipe(testPipeName, "") + l, err := ListenPipe(testPipeName, nil) if err != nil { t.Fatal(err) } @@ -220,7 +250,7 @@ func TestAcceptAfterCloseFails(t *testing.T) { } func TestDialTimesOutByDefault(t *testing.T) { - l, err := ListenPipe(testPipeName, "") + l, err := ListenPipe(testPipeName, nil) if err != nil { t.Fatal(err) } diff --git a/zsyscall.go b/zsyscall.go index eab955c..e500f98 100644 --- a/zsyscall.go +++ b/zsyscall.go @@ -19,6 +19,8 @@ var ( procCreateNamedPipeW = modkernel32.NewProc("CreateNamedPipeW") procCreateFileW = modkernel32.NewProc("CreateFileW") procWaitNamedPipeW = modkernel32.NewProc("WaitNamedPipeW") + procGetNamedPipeInfo = modkernel32.NewProc("GetNamedPipeInfo") + procGetNamedPipeHandleStateW = modkernel32.NewProc("GetNamedPipeHandleStateW") procLookupAccountNameW = modadvapi32.NewProc("LookupAccountNameW") procConvertSidToStringSidW = modadvapi32.NewProc("ConvertSidToStringSidW") procConvertStringSecurityDescriptorToSecurityDescriptorW = modadvapi32.NewProc("ConvertStringSecurityDescriptorToSecurityDescriptorW") @@ -165,6 +167,30 @@ func _waitNamedPipe(name *uint16, timeout uint32) (err error) { return } +func getNamedPipeInfo(pipe syscall.Handle, flags *uint32, outSize *uint32, inSize *uint32, maxInstances *uint32) (err error) { + r1, _, e1 := syscall.Syscall6(procGetNamedPipeInfo.Addr(), 5, uintptr(pipe), uintptr(unsafe.Pointer(flags)), uintptr(unsafe.Pointer(outSize)), uintptr(unsafe.Pointer(inSize)), uintptr(unsafe.Pointer(maxInstances)), 0) + if r1 == 0 { + if e1 != 0 { + err = error(e1) + } else { + err = syscall.EINVAL + } + } + return +} + +func getNamedPipeHandleState(pipe syscall.Handle, state *uint32, curInstances *uint32, maxCollectionCount *uint32, collectDataTimeout *uint32, userName *uint16, maxUserNameSize uint32) (err error) { + r1, _, e1 := syscall.Syscall9(procGetNamedPipeHandleStateW.Addr(), 7, uintptr(pipe), uintptr(unsafe.Pointer(state)), uintptr(unsafe.Pointer(curInstances)), uintptr(unsafe.Pointer(maxCollectionCount)), uintptr(unsafe.Pointer(collectDataTimeout)), uintptr(unsafe.Pointer(userName)), uintptr(maxUserNameSize), 0, 0) + if r1 == 0 { + if e1 != 0 { + err = error(e1) + } else { + err = syscall.EINVAL + } + } + return +} + func lookupAccountName(systemName *uint16, accountName string, sid *byte, sidSize *uint32, refDomain *uint16, refDomainSize *uint32, sidNameUse *uint32) (err error) { var _p0 *uint16 _p0, err = syscall.UTF16PtrFromString(accountName) From b12bd3ac52a4a3f1807154259237262f28c8f085 Mon Sep 17 00:00:00 2001 From: John Starks Date: Tue, 15 Mar 2016 10:40:31 -0700 Subject: [PATCH 3/3] Let clients set pipe buffer sizes --- pipe.go | 8 +++++++- 1 file changed, 7 insertions(+), 1 deletion(-) diff --git a/pipe.go b/pipe.go index 2b9c0cb..82db283 100644 --- a/pipe.go +++ b/pipe.go @@ -230,7 +230,7 @@ func makeServerPipeHandle(path string, securityDescriptor []byte, c *PipeConfig, if securityDescriptor != nil { sa.SecurityDescriptor = &securityDescriptor[0] } - h, err := createNamedPipe(path, flags, mode, cPIPE_UNLIMITED_INSTANCES, 4096, 4096, 0, &sa) + h, err := createNamedPipe(path, flags, mode, cPIPE_UNLIMITED_INSTANCES, uint32(c.OutputBufferSize), uint32(c.InputBufferSize), 0, &sa) if err != nil { return 0, &os.PathError{Op: "open", Path: path, Err: err} } @@ -302,6 +302,12 @@ type PipeConfig struct { // transferred to the reader (and returned as io.EOF in this implementation) // when the pipe is in message mode. MessageMode bool + + // InputBufferSize specifies the size the input buffer, in bytes. + InputBufferSize int32 + + // OutputBufferSize specifies the size the input buffer, in bytes. + OutputBufferSize int32 } // ListenPipe creates a listener on a Windows named pipe path, e.g. \\.\pipe\mypipe.