Apply read and write deadlines to pending IO

In order to upgrade to go1.8 all IO operations
including previous pending IO must have the deadline
set when SetReadDeadline or SetWriteDeadline is called

Signed-off-by: Darren Stahl <darst@microsoft.com>

file.go

@@ -7,6 +7,7 @@ import (
 	"io"
 	"runtime"
 	"sync"
+	"sync/atomic"
 	"syscall"
 	"time"
 )
@@ -17,6 +18,12 @@ import (
 //sys setFileCompletionNotificationModes(h syscall.Handle, flags uint8) (err error) = SetFileCompletionNotificationModes
 //sys timeBeginPeriod(period uint32) (n int32) = winmm.timeBeginPeriod
 
+type atomicBool int32
+
+func (b *atomicBool) isSet() bool { return atomic.LoadInt32((*int32)(b)) != 0 }
+func (b *atomicBool) setFalse()   { atomic.StoreInt32((*int32)(b), 0) }
+func (b *atomicBool) setTrue()    { atomic.StoreInt32((*int32)(b), 1) }
+
 const (
 	cFILE_SKIP_COMPLETION_PORT_ON_SUCCESS = 1
 	cFILE_SKIP_SET_EVENT_ON_HANDLE        = 2
@@ -33,6 +40,8 @@ func (e *timeoutError) Error() string   { return "i/o timeout" }
 func (e *timeoutError) Timeout() bool   { return true }
 func (e *timeoutError) Temporary() bool { return true }
 
+type timeoutChan chan struct{}
+
 var ioInitOnce sync.Once
 var ioCompletionPort syscall.Handle
 
@@ -63,8 +72,16 @@ type win32File struct {
 	handle        syscall.Handle
 	wg            sync.WaitGroup
 	closing       bool
-	readDeadline  time.Time
+	readDeadline  deadlineHandler
-	writeDeadline time.Time
+	writeDeadline deadlineHandler
+}
+
+type deadlineHandler struct {
+	setLock     sync.Mutex
+	channel     timeoutChan
+	channelLock sync.RWMutex
+	timer       *time.Timer
+	timedout    atomicBool
 }
 
 // makeWin32File makes a new win32File from an existing file handle
@@ -79,6 +96,8 @@ func makeWin32File(h syscall.Handle) (*win32File, error) {
 	if err != nil {
 		return nil, err
 	}
+	f.readDeadline.channel = make(timeoutChan)
+	f.writeDeadline.channel = make(timeoutChan)
 	return f, nil
 }
 
@@ -134,53 +153,47 @@ func ioCompletionProcessor(h syscall.Handle) {
 
 // asyncIo processes the return value from ReadFile or WriteFile, blocking until
 // the operation has actually completed.
-func (f *win32File) asyncIo(c *ioOperation, deadline time.Time, bytes uint32, err error) (int, error) {
+func (f *win32File) asyncIo(c *ioOperation, d *deadlineHandler, bytes uint32, err error) (int, error) {
 	if err != syscall.ERROR_IO_PENDING {
 		f.wg.Done()
 		return int(bytes), err
-	} else {
+	}
-		var r ioResult
-		wait := true
-		timedout := false
-		if f.closing {
-			cancelIoEx(f.handle, &c.o)
-		} else if !deadline.IsZero() {
-			now := time.Now()
-			if !deadline.After(now) {
-				timedout = true
-			} else {
-				timeout := time.After(deadline.Sub(now))
-				select {
-				case r = <-c.ch:
-					wait = false
-				case <-timeout:
-					timedout = true
-				}
-			}
-		}
-		if timedout {
-			cancelIoEx(f.handle, &c.o)
-		}
-		if wait {
-			r = <-c.ch
-		}
 
-		// runtime.KeepAlive is needed, as c is passed via native
+	if f.closing {
-		// code to ioCompletionProcessor, c must remain alive
+		cancelIoEx(f.handle, &c.o)
-		// until the channel read is complete.
+	}
-		runtime.KeepAlive(c)
 
+	var timeout timeoutChan
+	if d != nil {
+		d.channelLock.Lock()
+		timeout = d.channel
+		d.channelLock.Unlock()
+	}
+
+	var r ioResult
+	select {
+	case r = <-c.ch:
 		err = r.err
 		if err == syscall.ERROR_OPERATION_ABORTED {
 			if f.closing {
 				err = ErrFileClosed
-			} else if timedout {
-				err = ErrTimeout
 			}
 		}
-		f.wg.Done()
+	case <-timeout:
-		return int(r.bytes), err
+		cancelIoEx(f.handle, &c.o)
+		r = <-c.ch
+		err = r.err
+		if err == syscall.ERROR_OPERATION_ABORTED {
+			err = ErrTimeout
+		}
 	}
+
+	// runtime.KeepAlive is needed, as c is passed via native
+	// code to ioCompletionProcessor, c must remain alive
+	// until the channel read is complete.
+	runtime.KeepAlive(c)
+	f.wg.Done()
+	return int(r.bytes), err
 }
 
 // Read reads from a file handle.
@@ -189,9 +202,14 @@ func (f *win32File) Read(b []byte) (int, error) {
 	if err != nil {
 		return 0, err
 	}
+
+	if f.readDeadline.timedout.isSet() {
+		return 0, ErrTimeout
+	}
+
 	var bytes uint32
 	err = syscall.ReadFile(f.handle, b, &bytes, &c.o)
-	n, err := f.asyncIo(c, f.readDeadline, bytes, err)
+	n, err := f.asyncIo(c, &f.readDeadline, bytes, err)
 	runtime.KeepAlive(b)
 
 	// Handle EOF conditions.
@@ -210,23 +228,66 @@ func (f *win32File) Write(b []byte) (int, error) {
 	if err != nil {
 		return 0, err
 	}
+	if f.writeDeadline.timedout.isSet() {
+		return 0, ErrTimeout
+	}
+
 	var bytes uint32
 	err = syscall.WriteFile(f.handle, b, &bytes, &c.o)
-	n, err := f.asyncIo(c, f.writeDeadline, bytes, err)
+	n, err := f.asyncIo(c, &f.writeDeadline, bytes, err)
 	runtime.KeepAlive(b)
 	return n, err
 }
 
-func (f *win32File) SetReadDeadline(t time.Time) error {
+func (f *win32File) SetReadDeadline(deadline time.Time) error {
-	f.readDeadline = t
+	return f.readDeadline.set(deadline)
-	return nil
 }
 
-func (f *win32File) SetWriteDeadline(t time.Time) error {
+func (f *win32File) SetWriteDeadline(deadline time.Time) error {
-	f.writeDeadline = t
+	return f.writeDeadline.set(deadline)
-	return nil
 }
 
 func (f *win32File) Flush() error {
 	return syscall.FlushFileBuffers(f.handle)
 }
+
+func (d *deadlineHandler) set(deadline time.Time) error {
+	d.setLock.Lock()
+	defer d.setLock.Unlock()
+
+	if d.timer != nil {
+		if !d.timer.Stop() {
+			<-d.channel
+		}
+		d.timer = nil
+	}
+	d.timedout.setFalse()
+
+	select {
+	case <-d.channel:
+		d.channelLock.Lock()
+		d.channel = make(chan struct{})
+		d.channelLock.Unlock()
+	default:
+	}
+
+	if deadline.IsZero() {
+		return nil
+	}
+
+	timeoutIO := func() {
+		d.timedout.setTrue()
+		close(d.channel)
+	}
+
+	now := time.Now()
+	duration := deadline.Sub(now)
+	if deadline.After(now) {
+		// Deadline is in the future, set a timer to wait
+		d.timer = time.AfterFunc(duration, timeoutIO)
+	} else {
+		// Deadline is in the past. Cancel all pending IO now.
+		timeoutIO()
+	}
+	return nil
+}

pipe.go

@@ -367,7 +367,7 @@ func connectPipe(p *win32File) error {
 		return err
 	}
 	err = connectNamedPipe(p.handle, &c.o)
-	_, err = p.asyncIo(c, time.Time{}, 0, err)
+	_, err = p.asyncIo(c, nil, 0, err)
 	if err != nil && err != cERROR_PIPE_CONNECTED {
 		return err
 	}