pipe: add server backlog for concurrent Accept()

Teach `pipe.go:ListenPipe()` to create multiple instances of
the server pipe in the kernel so that client connections are
less likely to receive a `windows.ERROR_PIPE_BUSY` error.
This is conceptually similar to the `backlog` argument of the
Unix `listen(2)` function.

The current `listenerRoutine()` function works sequentially and
in response to calls to `Accept()`, such that there will only
be at most one unbound server pipe present at any time.  Even
if the server application calls `Accept()` concurrrently from
a pool of application threads, `listenerRoutine()` will process
them sequentially.

In this model and because there is only one `listenerRoutine()`
instance, there is an interval of time where there are no
available unbound/free server pipes.  When `ConnectNamedPipe()`
returns `listenerRoutine()` sends the new pipe handle over a
channel to the caller of `Accept()`.  Application code then has
an opportunity to dispatch/process it and then call `Accept()`
again.  This causes `listenerRoutine()` to create a new unbound
serer pipe and wait for the next connection.  Anytime during
this interval, a client will get a pipe busy error.

Code in `DialPipe()` hides this from GOLANG callers because it
includes a busy retry loop.  However, clients written in other
languages without this assistance are likely to see it and deal
with it.

This change introduces an "accept queue" using a buffered channel
and splits `listenerRoutine()` into a pool of listener worker
threads.  Each worker creates a new unbound pipe and waits for
a client connection.  The NPFS and kernel handle connectioni
delivery to a random listener worker.  The resulting connected
pipe is delivered back to the caller `Accept()` as before.

A `PipeConfig.QueueSize` variable controls the number of listener
worker threads and the maximum number of unbound/free pipes server
pipes that will be present at any given time.  Note that a
listener worker will normally have an unbound/free pipe except
during that same delivery interval.  Having multiple active
workers gives us extra capacity to handle rapidly arriving
connections.

The application is encouraged to call `Accept()` from a pool
of application workers.  The size of the application pool should
be the same or larger than the queue size to take full advantage
of the listener queue.

To preserve backwards compatibility, a queue size of 0 or 1
will behave as before.

Also for backwards compatibility, listener workers are required
to wait for an `Accept()` call so that the worker has a return
channel to send the connected pipe and error code.  This implies
that the number of unbound pipes will be the smaller of the
queue size and the application pool size.

Finally, a Mutex was added to `l.Close()` to ensure that
concurrent threads do not simultaneously try to shutdown the
pipe.

Signed-off-by: Jeff Hostetler <[email protected]>

pipe.go

@@ -11,6 +11,7 @@ import (
 	"net"
 	"os"
 	"runtime"
+	"sync"
 	"syscall"
 	"time"
 	"unsafe"
@@ -258,9 +259,30 @@ type win32PipeListener struct {
 	firstHandle syscall.Handle
 	path        string
 	config      PipeConfig
-	acceptCh    chan (chan acceptResponse)
-	closeCh     chan int
-	doneCh      chan int
+
+	// `acceptQueueCh` is a buffered channel (of channels).  Calls to
+	// Accept() will append to this queue to schedule a listener-worker
+	// to create a new named pipe instance in the named pipe file system
+	// (NPFS) and then listen for a connection from a client.
+	//
+	// The resulting connected pipe (or error) will be signalled (back
+	// to `Accept()`) on the channel value's channel.
+	acceptQueueCh chan (chan acceptResponse)
+
+	// `shutdownStartedCh` will be closed to indicate that all listener
+	// workers should shutdown.  `l.Close()` will signal this to begin
+	// a shutdown.
+	shutdownStartedCh chan struct{}
+
+	// `shutdownFinishedCh` will be closed to indicate that `l.listenerRoutine()`
+	// has stopped all of the listener worker threads and has finished the
+	// shutdown.  `l.Close()` must wait for this signal before returning.
+	shutdownFinishedCh chan struct{}
+
+	// `closeMux` is used to create a critical section in `l.Close()` and
+	// coordinate the shutdown and prevent problems if a second thread calls
+	// `l.Close()` while a shutdown is in progress.
+	closeMux sync.Mutex
 }
 
 func makeServerPipeHandle(path string, sd []byte, c *PipeConfig, first bool) (syscall.Handle, error) {
@@ -383,7 +405,7 @@ func (l *win32PipeListener) makeConnectedServerPipe() (*win32File, error) {
 			p.Close()
 			p = nil
 		}
-	case <-l.closeCh:
+	case <-l.shutdownStartedCh:
 		// Abort the connect request by closing the handle.
 		p.Close()
 		p = nil
@@ -395,33 +417,44 @@ func (l *win32PipeListener) makeConnectedServerPipe() (*win32File, error) {
 	return p, err
 }
 
-func (l *win32PipeListener) listenerRoutine() {
-	closed := false
-	for !closed {
+func (l *win32PipeListener) listenerWorker(wg *sync.WaitGroup) {
+	var stop bool
+	for !stop {
 		select {
-		case <-l.closeCh:
-			closed = true
-		case responseCh := <-l.acceptCh:
-			var (
-				p   *win32File
-				err error
-			)
-			for {
-				p, err = l.makeConnectedServerPipe()
-				// If the connection was immediately closed by the client, try
-				// again.
-				if err != windows.ERROR_NO_DATA { //nolint:errorlint // err is Errno
-					break
-				}
-			}
+		case <-l.shutdownStartedCh:
+			stop = true
+		case responseCh := <-l.acceptQueueCh:
+			p, err := l.makeConnectedServerPipe()
 			responseCh <- acceptResponse{p, err}
-			closed = err == ErrPipeListenerClosed //nolint:errorlint // err is Errno
 		}
 	}
+
+	wg.Done()
+}
+
+func (l *win32PipeListener) listenerRoutine(queueSize int) {
+	var wg sync.WaitGroup
+
+	for k := 0; k < queueSize; k++ {
+		wg.Add(1)
+		go l.listenerWorker(&wg)
+	}
+
+	wg.Wait() // for all listenerWorkers to finish.
+
+	// We can assert here that `l.shutdownStartedCh` has been
+	// signalled (since `l.Close()` closed it).
+	//
+	// We might consider draining the `l.acceptQueueCh` and
+	// closing each of the channel instances, but that is not
+	// necessary since the second "select" in `l.Accept()` is
+	// waiting on the `requestCh` and `l.shutdownFinishedCh`.
+	// And we're going to signal the latter in a moment.
+
 	syscall.Close(l.firstHandle)
 	l.firstHandle = 0
 	// Notify Close() and Accept() callers that the handle has been closed.
-	close(l.doneCh)
+	close(l.shutdownFinishedCh)
 }
 
 // PipeConfig contain configuration for the pipe listener.
@@ -442,6 +475,19 @@ type PipeConfig struct {
 
 	// OutputBufferSize specifies the size of the output buffer, in bytes.
 	OutputBufferSize int32
+
+	// QueueSize specifies the maximum number of concurrently active pipe server
+	// handles to allow.  This is conceptually similar to the `backlog` argument
+	// to `listen(2)` on Unix systems. Increasing this value reduces the likelyhood
+	// of a connecting client receiving a `windows.ERROR_PIPE_BUSY` error.
+	// (Assuming that the server is written to call `l.Accept()` using a pool of
+	// application worker threads.)
+	//
+	// This value should be larger than your expected client arrival rate so that
+	// there are always a few extra listener worker threads and (more importantly)
+	// unbound server pipes in the kernel, so that a client "CreateFile()" should
+	// not get a busy signal.
+	QueueSize int32
 }
 
 // ListenPipe creates a listener on a Windows named pipe path, e.g. \\.\pipe\mypipe.
@@ -460,19 +506,30 @@ func ListenPipe(path string, c *PipeConfig) (net.Listener, error) {
 			return nil, err
 		}
 	}
+
+	queueSize := int(c.QueueSize)
+	if queueSize < 1 {
+		// Legacy calls will pass 0 since they won't know to set the queue size.
+		// Default to legacy behavior where we never have more than 1 available
+		// unbound pipe and that is only present when an application thread is
+		// blocked in `l.Accept()`.
+		queueSize = 1
+	}
+
 	h, err := makeServerPipeHandle(path, sd, c, true)
 	if err != nil {
 		return nil, err
 	}
 	l := &win32PipeListener{
-		firstHandle: h,
-		path:        path,
-		config:      *c,
-		acceptCh:    make(chan (chan acceptResponse)),
-		closeCh:     make(chan int),
-		doneCh:      make(chan int),
-	}
-	go l.listenerRoutine()
+		firstHandle:        h,
+		path:               path,
+		config:             *c,
+		acceptQueueCh:      make(chan chan acceptResponse, queueSize),
+		shutdownStartedCh:  make(chan struct{}),
+		shutdownFinishedCh: make(chan struct{}),
+		closeMux:           sync.Mutex{},
+	}
+	go l.listenerRoutine(queueSize)
 	return l, nil
 }
 
@@ -492,31 +549,84 @@ func connectPipe(p *win32File) error {
 }
 
 func (l *win32PipeListener) Accept() (net.Conn, error) {
+tryAgain:
 	ch := make(chan acceptResponse)
+
 	select {
-	case l.acceptCh <- ch:
-		response := <-ch
-		err := response.err
-		if err != nil {
-			return nil, err
+	case l.acceptQueueCh <- ch:
+		// We have queued a request for a worker thread to listen
+		// for a connection.
+	case <-l.shutdownFinishedCh:
+		// The shutdown completed before we could request a connection.
+		return nil, ErrPipeListenerClosed
+	case <-l.shutdownStartedCh:
+		// The shutdown is already in progress. Don't bother trying to
+		// schedule a new request.
+		return nil, ErrPipeListenerClosed
+	}
+
+	// We queued a request. Now wait for a connection signal or a
+	// shutdown while we were waiting.
+
+	select {
+	case response := <-ch:
+		if response.f == nil && response.err == nil {
+			// The listener worker could close our channel instance
+			// to indicate that the listener is shut down.
+			return nil, ErrPipeListenerClosed
+		}
+		if errors.Is(response.err, ErrPipeListenerClosed) {
+			return nil, ErrPipeListenerClosed
+		}
+		if response.err == windows.ERROR_NO_DATA { //nolint:errorlint // err is Errno
+			// If the connection was immediately closed by the client,
+			// try again (without reporting an error or a dead connection
+			// to the `Accept()` caller).  This avoids spurious
+			// "The pipe is being closed." messages.
+			goto tryAgain
+		}
+		if response.err != nil {
+			return nil, response.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:
+	case <-l.shutdownFinishedCh:
+		// The shutdown started and completed while we were waiting for a
+		// connection.
 		return nil, ErrPipeListenerClosed
+
+		// case <-l.shutdownStartedCh:
+		// We DO NOT watch for `l.shutdownStartedCh` because we need
+		// to keep listening on our local `ch` so that the associated
+		// listener worker can signal it without blocking when throwing
+		// an ErrPipeListenerClosed error.
 	}
 }
 
 func (l *win32PipeListener) Close() error {
+	l.closeMux.Lock()
 	select {
-	case l.closeCh <- 1:
-		<-l.doneCh
-	case <-l.doneCh:
+	case <-l.shutdownFinishedCh:
+		// The shutdown has already completed. Nothing to do.
+	default:
+		select {
+		case <-l.shutdownStartedCh:
+			// The shutdown is in progress. We should not get here because
+			// of the Mutex, but either way, we don't want to race here
+			// and accidentally close `l.shutdownStartedCh` twice.
+		default:
+			// Cause all listener workers to abort.
+			close(l.shutdownStartedCh)
+			// Wait for listenerRoutine to stop the workers and clean up.
+			<-l.shutdownFinishedCh
+		}
 	}
+	l.closeMux.Unlock()
+
 	return nil
 }