Merge pull request #50 from Microsoft/Timeout

Apply read and write deadlines to pending IO

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
-	writeDeadline time.Time
+	readDeadline  deadlineHandler
+	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
-		// code to ioCompletionProcessor, c must remain alive
-		// until the channel read is complete.
-		runtime.KeepAlive(c)
+	if f.closing {
+		cancelIoEx(f.handle, &c.o)
+	}
 
+	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()
-		return int(r.bytes), err
+	case <-timeout:
+		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 {
-	f.readDeadline = t
-	return nil
+func (f *win32File) SetReadDeadline(deadline time.Time) error {
+	return f.readDeadline.set(deadline)
 }
 
-func (f *win32File) SetWriteDeadline(t time.Time) error {
-	f.writeDeadline = t
-	return nil
+func (f *win32File) SetWriteDeadline(deadline time.Time) error {
+	return f.writeDeadline.set(deadline)
 }
 
 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
 	}

pipe_test.go

@@ -12,6 +12,8 @@ import (
 
 var testPipeName = `\\.\pipe\winiotestpipe`
 
+var aLongTimeAgo = time.Unix(1, 0)
+
 func TestDialUnknownFailsImmediately(t *testing.T) {
 	_, err := DialPipe(testPipeName, nil)
 	if err.(*os.PathError).Err != syscall.ENOENT {
@@ -261,6 +263,99 @@ func TestDialTimesOutByDefault(t *testing.T) {
 	}
 }
 
+func TestTimeoutPendingRead(t *testing.T) {
+	l, err := ListenPipe(testPipeName, nil)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer l.Close()
+
+	serverDone := make(chan struct{})
+
+	go func() {
+		s, err := l.Accept()
+		if err != nil {
+			t.Fatal(err)
+		}
+		time.Sleep(1 * time.Second)
+		s.Close()
+		close(serverDone)
+	}()
+
+	client, err := DialPipe(testPipeName, nil)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer client.Close()
+
+	clientErr := make(chan error)
+	go func() {
+		buf := make([]byte, 10)
+		_, err = client.Read(buf)
+		clientErr <- err
+	}()
+
+	time.Sleep(100 * time.Millisecond) // make *sure* the pipe is reading before we set the deadline
+	client.SetReadDeadline(aLongTimeAgo)
+
+	select {
+	case err = <-clientErr:
+		if err != ErrTimeout {
+			t.Fatalf("expected ErrTimeout, got %v", err)
+		}
+	case <-time.After(100 * time.Millisecond):
+		t.Fatalf("timed out while waiting for read to cancel")
+		<-clientErr
+	}
+	<-serverDone
+}
+
+func TestTimeoutPendingWrite(t *testing.T) {
+	l, err := ListenPipe(testPipeName, nil)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer l.Close()
+
+	serverDone := make(chan struct{})
+
+	go func() {
+		s, err := l.Accept()
+		if err != nil {
+			t.Fatal(err)
+		}
+		time.Sleep(1 * time.Second)
+		s.Close()
+		close(serverDone)
+	}()
+
+	client, err := DialPipe(testPipeName, nil)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer client.Close()
+
+	clientErr := make(chan error)
+	go func() {
+		_, err = client.Write([]byte("this should timeout"))
+		clientErr <- err
+	}()
+
+	time.Sleep(100 * time.Millisecond) // make *sure* the pipe is writing before we set the deadline
+	client.SetWriteDeadline(aLongTimeAgo)
+
+	select {
+	case err = <-clientErr:
+		if err != ErrTimeout {
+			t.Fatalf("expected ErrTimeout, got %v", err)
+		}
+	case <-time.After(100 * time.Millisecond):
+		t.Fatalf("timed out while waiting for write to cancel")
+		<-clientErr
+	}
+	<-serverDone
+}
+
 type CloseWriter interface {
 	CloseWrite() error
 }