Ensure win32PipeListener.Close() correctly aborts Listen() calls

Signed-off-by: John Starks <jostarks@microsoft.com>

pipe.go

@@ -114,8 +114,9 @@ type win32PipeListener struct {
 	firstHandle        syscall.Handle
 	path               string
 	securityDescriptor string
-	closeCh            chan (chan int)
 	acceptCh           chan (chan acceptResponse)
+	closeCh            chan int
+	doneCh             chan int
 }
 
 func makeServerPipeHandle(path, securityDescriptor string, first bool) (syscall.Handle, error) {
@@ -157,40 +158,43 @@ func (l *win32PipeListener) makeServerPipe() (*win32Pipe, error) {
 }
 
 func (l *win32PipeListener) listenerRoutine() {
-	var closeResponseCh (chan int)
+	closed := false
-Loop:
+	for !closed {
-	for {
-		var responseCh (chan acceptResponse)
 		select {
-		case closeResponseCh = <-l.closeCh:
+		case <-l.closeCh:
-			break Loop
+			closed = true
-		case responseCh = <-l.acceptCh:
+		case responseCh := <-l.acceptCh:
-		}
 			p, err := l.makeServerPipe()
-		if err != nil {
+			if err == nil {
-			responseCh <- acceptResponse{nil, err}
+				// Wait for the client to connect.
-		} else {
 				ch := make(chan error)
 				go func() {
 					ch <- connectPipe(p)
 				}()
 				select {
-			case closeResponseCh = <-l.closeCh:
-				p.Close()
-				<-ch
-				break Loop
 				case err = <-ch:
 					if err != nil {
 						p.Close()
 						p = nil
 					}
-				responseCh <- acceptResponse{p, err}
+				case <-l.closeCh:
+					// Abort the connect request by closing the handle.
+					p.Close()
+					p = nil
+					err = <-ch
+					if err == nil {
+						err = FileClosed
 					}
+					closed = true
+				}
+			}
+			responseCh <- acceptResponse{p, err}
 		}
 	}
 	syscall.CloseHandle(l.firstHandle)
 	l.firstHandle = syscall.Handle(0)
-	closeResponseCh <- 1
+	// Notify Close() and Accept() callers that the handle has been closed.
+	close(l.doneCh)
 }
 
 func ListenPipe(path, securityDescriptor string) (net.Listener, error) {
@@ -210,8 +214,9 @@ func ListenPipe(path, securityDescriptor string) (net.Listener, error) {
 		firstHandle:        h,
 		path:               path,
 		securityDescriptor: securityDescriptor,
-		closeCh:            make(chan (chan int)),
 		acceptCh:           make(chan (chan acceptResponse)),
+		closeCh:            make(chan int),
+		doneCh:             make(chan int),
 	}
 	go l.listenerRoutine()
 	return l, nil
@@ -232,15 +237,21 @@ func connectPipe(p *win32Pipe) error {
 
 func (l *win32PipeListener) Accept() (net.Conn, error) {
 	ch := make(chan acceptResponse)
-	l.acceptCh <- ch
+	select {
+	case l.acceptCh <- ch:
 		response := <-ch
 		return response.p, response.err
+	case <-l.doneCh:
+		return nil, FileClosed
+	}
 }
 
 func (l *win32PipeListener) Close() error {
-	ch := make(chan int)
+	select {
-	l.closeCh <- ch
+	case l.closeCh <- 1:
-	<-ch
+		<-l.doneCh
+	case <-l.doneCh:
+	}
 	return nil
 }
 

pipe_test.go

@@ -93,12 +93,11 @@ func TestReadTimeout(t *testing.T) {
 	}
 }
 
-func server(l net.Listener) {
+func server(l net.Listener, ch chan int) {
 	c, err := l.Accept()
 	if err != nil {
 		panic(err)
 	}
-	defer c.Close()
 	rw := bufio.NewReadWriter(bufio.NewReader(c), bufio.NewWriter(c))
 	s, err := rw.ReadString('\n')
 	if err != nil {
@@ -112,6 +111,8 @@ func server(l net.Listener) {
 	if err != nil {
 		panic(err)
 	}
+	c.Close()
+	ch <- 1
 }
 
 func TestFullListenDialReadWrite(t *testing.T) {
@@ -119,13 +120,16 @@ func TestFullListenDialReadWrite(t *testing.T) {
 	if err != nil {
 		t.Fatal(err)
 	}
+	defer l.Close()
 
-	go server(l)
+	ch := make(chan int)
+	go server(l, ch)
 
 	c, err := DialPipe(testPipeName, nil)
 	if err != nil {
 		t.Fatal(err)
 	}
+	defer c.Close()
 
 	rw := bufio.NewReadWriter(bufio.NewReader(c), bufio.NewWriter(c))
 	_, err = rw.WriteString("hello world\n")
@@ -145,4 +149,39 @@ func TestFullListenDialReadWrite(t *testing.T) {
 	if s != ms {
 		t.Errorf("expected '%s', got '%s'", ms, s)
 	}
+
+	<-ch
+}
+
+func TestCloseAbortsListen(t *testing.T) {
+	l, err := ListenPipe(testPipeName, "")
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	ch := make(chan int)
+	go func() {
+		_, err := l.Accept()
+		if err != syscall.ERROR_OPERATION_ABORTED {
+			t.Fatalf("expected ERROR_OPERATION_ABORTED, got %v", err)
+		}
+		ch <- 1
+	}()
+
+	time.Sleep(30 * time.Millisecond)
+	l.Close()
+
+	<-ch
+}
+
+func TestAcceptAfterCloseFails(t *testing.T) {
+	l, err := ListenPipe(testPipeName, "")
+	if err != nil {
+		t.Fatal(err)
+	}
+	l.Close()
+	_, err = l.Accept()
+	if err != FileClosed {
+		t.Fatalf("expected FileClosed, got %v", err)
+	}
 }