Merge pull request #65 from darrenstahlmsft/waitGroup

Prevent adding to a sync.WaitGroup that has a waiter

file.go

@@ -78,6 +78,7 @@ func initIo() {
 type win32File struct {
 	handle        syscall.Handle
 	wg            sync.WaitGroup
+	wgLock        sync.RWMutex
 	closing       atomicBool
 	readDeadline  deadlineHandler
 	writeDeadline deadlineHandler
@@ -114,14 +115,18 @@ func MakeOpenFile(h syscall.Handle) (io.ReadWriteCloser, error) {
 
 // closeHandle closes the resources associated with a Win32 handle
 func (f *win32File) closeHandle() {
+	f.wgLock.Lock()
 	// Atomically set that we are closing, releasing the resources only once.
 	if !f.closing.swap(true) {
+		f.wgLock.Unlock()
 		// cancel all IO and wait for it to complete
 		cancelIoEx(f.handle, nil)
 		f.wg.Wait()
 		// at this point, no new IO can start
 		syscall.Close(f.handle)
 		f.handle = 0
+	} else {
+		f.wgLock.Unlock()
 	}
 }
 
@@ -134,10 +139,13 @@ func (f *win32File) Close() error {
 // prepareIo prepares for a new IO operation.
 // The caller must call f.wg.Done() when the IO is finished, prior to Close() returning.
 func (f *win32File) prepareIo() (*ioOperation, error) {
+	f.wgLock.RLock()
 	if f.closing.isSet() {
+		f.wgLock.RUnlock()
 		return nil, ErrFileClosed
 	}
 	f.wg.Add(1)
+	f.wgLock.RUnlock()
 	c := &ioOperation{}
 	c.ch = make(chan ioResult)
 	return c, nil

pipe_test.go

@@ -367,11 +367,11 @@ func TestEchoWithMessaging(t *testing.T) {
 		OutputBufferSize: 65536,
 	}
 	l, err := ListenPipe(testPipeName, &c)
-
 	if err != nil {
 		t.Fatal(err)
 	}
 	defer l.Close()
+
 	listenerDone := make(chan bool)
 	clientDone := make(chan bool)
 	go func() {
@@ -380,6 +380,8 @@ func TestEchoWithMessaging(t *testing.T) {
 		if e != nil {
 			t.Fatal(e)
 		}
+		defer conn.Close()
+
 		time.Sleep(500 * time.Millisecond) // make *sure* we don't begin to read before eof signal is sent
 		io.Copy(conn, conn)
 		conn.(CloseWriter).CloseWrite()