From 2a0317319a90892a3d0c9a4a46ba5501944792c0 Mon Sep 17 00:00:00 2001 From: Darren Stahl Date: Fri, 5 May 2017 14:53:22 -0700 Subject: [PATCH 1/2] Adding failing tests for deadlines on pending IO Signed-off-by: Darren Stahl --- pipe_test.go | 95 ++++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 95 insertions(+) diff --git a/pipe_test.go b/pipe_test.go index 2a75302..4c6208c 100644 --- a/pipe_test.go +++ b/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 } From 899dbf3ba5f93cde82b16f33eadfd60d5b8b856a Mon Sep 17 00:00:00 2001 From: Darren Stahl Date: Wed, 3 May 2017 16:00:40 -0700 Subject: [PATCH 2/2] 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 --- file.go | 151 +++++++++++++++++++++++++++++++++++++++----------------- pipe.go | 2 +- 2 files changed, 107 insertions(+), 46 deletions(-) diff --git a/file.go b/file.go index b955e88..078a568 100644 --- a/file.go +++ b/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 +} diff --git a/pipe.go b/pipe.go index f85f4a9..5930520 100644 --- a/pipe.go +++ b/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 }