From 899dbf3ba5f93cde82b16f33eadfd60d5b8b856a Mon Sep 17 00:00:00 2001 From: Darren Stahl Date: Wed, 3 May 2017 16:00:40 -0700 Subject: [PATCH] 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 }