Merge pull request #21235 from Microsoft/jstarks/add_npipe_close_write

Windows: fix various npipe bugs

docker/listeners/listeners_windows.go

@@ -32,7 +32,13 @@ func Init(proto, addr, socketGroup string, tlsConfig *tls.Config) (ls []net.List
 				sddl += fmt.Sprintf("(A;;GRGW;;;%s)", sid)
 			}
 		}
-		l, err := winio.ListenPipe(addr, sddl)
+		c := winio.PipeConfig{
+			SecurityDescriptor: sddl,
+			MessageMode:        true,  // Use message mode so that CloseWrite() is supported
+			InputBufferSize:    65536, // Use 64KB buffers to improve performance
+			OutputBufferSize:   65536,
+		}
+		l, err := winio.ListenPipe(addr, &c)
 		if err != nil {
 			return nil, err
 		}

hack/vendor.sh

@@ -8,7 +8,7 @@ source 'hack/.vendor-helpers.sh'
 # the following lines are in sorted order, FYI
 clone git github.com/Azure/go-ansiterm 70b2c90b260171e829f1ebd7c17f600c11858dbe
 clone git github.com/Microsoft/hcsshim 116e0e9f5ced0cec94ae46d0aa1b3002a325f532
-clone git github.com/Microsoft/go-winio f778f05015353be65d242f3fedc18695756153bb
+clone git github.com/Microsoft/go-winio v0.1.0
 clone git github.com/Sirupsen/logrus v0.9.0 # logrus is a common dependency among multiple deps
 clone git github.com/docker/libtrust 9cbd2a1374f46905c68a4eb3694a130610adc62a
 clone git github.com/go-check/check a625211d932a2a643d0d17352095f03fb7774663 https://github.com/cpuguy83/check.git

vendor/src/github.com/Microsoft/go-winio/file.go

@@ -13,6 +13,7 @@ import (
 //sys createIoCompletionPort(file syscall.Handle, port syscall.Handle, key uintptr, threadCount uint32) (newport syscall.Handle, err error) = CreateIoCompletionPort
 //sys getQueuedCompletionStatus(port syscall.Handle, bytes *uint32, key *uintptr, o **ioOperation, timeout uint32) (err error) = GetQueuedCompletionStatus
 //sys setFileCompletionNotificationModes(h syscall.Handle, flags uint8) (err error) = SetFileCompletionNotificationModes
+//sys timeBeginPeriod(period uint32) (n int32) = winmm.timeBeginPeriod
 
 const (
 	cFILE_SKIP_COMPLETION_PORT_ON_SUCCESS = 1
@@ -117,6 +118,8 @@ func (f *win32File) prepareIo() (*ioOperation, error) {
 
 // ioCompletionProcessor processes completed async IOs forever
 func ioCompletionProcessor(h syscall.Handle) {
+	// Set the timer resolution to 1. This fixes a performance regression in golang 1.6.
+	timeBeginPeriod(1)
 	for {
 		var bytes uint32
 		var key uintptr

vendor/src/github.com/Microsoft/go-winio/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, uint32(c.OutputBufferSize), uint32(c.InputBufferSize), 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,43 @@ 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
+
+	// 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.
+// 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 +342,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 +351,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 +369,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
 	}

vendor/src/github.com/Microsoft/go-winio/zsyscall.go

@@ -9,16 +9,20 @@ var _ unsafe.Pointer
 
 var (
 	modkernel32 = syscall.NewLazyDLL("kernel32.dll")
+	modwinmm    = syscall.NewLazyDLL("winmm.dll")
 	modadvapi32 = syscall.NewLazyDLL("advapi32.dll")
 
 	procCancelIoEx                                           = modkernel32.NewProc("CancelIoEx")
 	procCreateIoCompletionPort                               = modkernel32.NewProc("CreateIoCompletionPort")
 	procGetQueuedCompletionStatus                            = modkernel32.NewProc("GetQueuedCompletionStatus")
 	procSetFileCompletionNotificationModes                   = modkernel32.NewProc("SetFileCompletionNotificationModes")
+	proctimeBeginPeriod                                      = modwinmm.NewProc("timeBeginPeriod")
 	procConnectNamedPipe                                     = modkernel32.NewProc("ConnectNamedPipe")
 	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")
@@ -88,6 +92,12 @@ func setFileCompletionNotificationModes(h syscall.Handle, flags uint8) (err erro
 	return
 }
 
+func timeBeginPeriod(period uint32) (n int32) {
+	r0, _, _ := syscall.Syscall(proctimeBeginPeriod.Addr(), 1, uintptr(period), 0, 0)
+	n = int32(r0)
+	return
+}
+
 func connectNamedPipe(pipe syscall.Handle, o *syscall.Overlapped) (err error) {
 	r1, _, e1 := syscall.Syscall(procConnectNamedPipe.Addr(), 2, uintptr(pipe), uintptr(unsafe.Pointer(o)), 0)
 	if r1 == 0 {
@@ -165,6 +175,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)