1 // Copyright 2009 The Go Authors. All rights reserved. 2 // Use of this source code is governed by a BSD-style 3 // license that can be found in the LICENSE file. 4 5 package poll 6 7 import ( 8 "errors" 9 "io" 10 "sync/atomic" 11 "time" 12 ) 13 14 type atomicBool int32 15 16 func (b *atomicBool) isSet() bool { return atomic.LoadInt32((*int32)(b)) != 0 } 17 func (b *atomicBool) setFalse() { atomic.StoreInt32((*int32)(b), 0) } 18 func (b *atomicBool) setTrue() { atomic.StoreInt32((*int32)(b), 1) } 19 20 type FD struct { 21 // Lock sysfd and serialize access to Read and Write methods. 22 fdmu fdMutex 23 24 Destroy func() 25 26 // deadlines 27 raio *asyncIO 28 waio *asyncIO 29 rtimer *time.Timer 30 wtimer *time.Timer 31 rtimedout atomicBool // set true when read deadline has been reached 32 wtimedout atomicBool // set true when write deadline has been reached 33 34 // Whether this is a normal file. 35 // On Plan 9 we do not use this package for ordinary files, 36 // so this is always false, but the field is present because 37 // shared code in fd_mutex.go checks it. 38 isFile bool 39 } 40 41 // We need this to close out a file descriptor when it is unlocked, 42 // but the real implementation has to live in the net package because 43 // it uses os.File's. 44 func (fd *FD) destroy() error { 45 if fd.Destroy != nil { 46 fd.Destroy() 47 } 48 return nil 49 } 50 51 // Close handles the locking for closing an FD. The real operation 52 // is in the net package. 53 func (fd *FD) Close() error { 54 if !fd.fdmu.increfAndClose() { 55 return errClosing(fd.isFile) 56 } 57 return nil 58 } 59 60 // Read implements io.Reader. 61 func (fd *FD) Read(fn func([]byte) (int, error), b []byte) (int, error) { 62 if fd.rtimedout.isSet() { 63 return 0, ErrTimeout 64 } 65 if err := fd.readLock(); err != nil { 66 return 0, err 67 } 68 defer fd.readUnlock() 69 if len(b) == 0 { 70 return 0, nil 71 } 72 fd.raio = newAsyncIO(fn, b) 73 n, err := fd.raio.Wait() 74 fd.raio = nil 75 if isHangup(err) { 76 err = io.EOF 77 } 78 if isInterrupted(err) { 79 err = ErrTimeout 80 } 81 return n, err 82 } 83 84 // Write implements io.Writer. 85 func (fd *FD) Write(fn func([]byte) (int, error), b []byte) (int, error) { 86 if fd.wtimedout.isSet() { 87 return 0, ErrTimeout 88 } 89 if err := fd.writeLock(); err != nil { 90 return 0, err 91 } 92 defer fd.writeUnlock() 93 fd.waio = newAsyncIO(fn, b) 94 n, err := fd.waio.Wait() 95 fd.waio = nil 96 if isInterrupted(err) { 97 err = ErrTimeout 98 } 99 return n, err 100 } 101 102 // SetDeadline sets the read and write deadlines associated with fd. 103 func (fd *FD) SetDeadline(t time.Time) error { 104 return setDeadlineImpl(fd, t, 'r'+'w') 105 } 106 107 // SetReadDeadline sets the read deadline associated with fd. 108 func (fd *FD) SetReadDeadline(t time.Time) error { 109 return setDeadlineImpl(fd, t, 'r') 110 } 111 112 // SetWriteDeadline sets the write deadline associated with fd. 113 func (fd *FD) SetWriteDeadline(t time.Time) error { 114 return setDeadlineImpl(fd, t, 'w') 115 } 116 117 func setDeadlineImpl(fd *FD, t time.Time, mode int) error { 118 d := t.Sub(time.Now()) 119 if mode == 'r' || mode == 'r'+'w' { 120 fd.rtimedout.setFalse() 121 } 122 if mode == 'w' || mode == 'r'+'w' { 123 fd.wtimedout.setFalse() 124 } 125 if t.IsZero() || d < 0 { 126 // Stop timer 127 if mode == 'r' || mode == 'r'+'w' { 128 if fd.rtimer != nil { 129 fd.rtimer.Stop() 130 } 131 fd.rtimer = nil 132 } 133 if mode == 'w' || mode == 'r'+'w' { 134 if fd.wtimer != nil { 135 fd.wtimer.Stop() 136 } 137 fd.wtimer = nil 138 } 139 } else { 140 // Interrupt I/O operation once timer has expired 141 if mode == 'r' || mode == 'r'+'w' { 142 fd.rtimer = time.AfterFunc(d, func() { 143 fd.rtimedout.setTrue() 144 if fd.raio != nil { 145 fd.raio.Cancel() 146 } 147 }) 148 } 149 if mode == 'w' || mode == 'r'+'w' { 150 fd.wtimer = time.AfterFunc(d, func() { 151 fd.wtimedout.setTrue() 152 if fd.waio != nil { 153 fd.waio.Cancel() 154 } 155 }) 156 } 157 } 158 if !t.IsZero() && d < 0 { 159 // Interrupt current I/O operation 160 if mode == 'r' || mode == 'r'+'w' { 161 fd.rtimedout.setTrue() 162 if fd.raio != nil { 163 fd.raio.Cancel() 164 } 165 } 166 if mode == 'w' || mode == 'r'+'w' { 167 fd.wtimedout.setTrue() 168 if fd.waio != nil { 169 fd.waio.Cancel() 170 } 171 } 172 } 173 return nil 174 } 175 176 // On Plan 9 only, expose the locking for the net code. 177 178 // ReadLock wraps FD.readLock. 179 func (fd *FD) ReadLock() error { 180 return fd.readLock() 181 } 182 183 // ReadUnlock wraps FD.readUnlock. 184 func (fd *FD) ReadUnlock() { 185 fd.readUnlock() 186 } 187 188 func isHangup(err error) bool { 189 return err != nil && stringsHasSuffix(err.Error(), "Hangup") 190 } 191 192 func isInterrupted(err error) bool { 193 return err != nil && stringsHasSuffix(err.Error(), "interrupted") 194 } 195 196 // PollDescriptor returns the descriptor being used by the poller, 197 // or ^uintptr(0) if there isn't one. This is only used for testing. 198 func PollDescriptor() uintptr { 199 return ^uintptr(0) 200 } 201 202 // RawControl invokes the user-defined function f for a non-IO 203 // operation. 204 func (fd *FD) RawControl(f func(uintptr)) error { 205 return errors.New("not implemented") 206 } 207 208 // RawRead invokes the user-defined function f for a read operation. 209 func (fd *FD) RawRead(f func(uintptr) bool) error { 210 return errors.New("not implemented") 211 } 212 213 // RawWrite invokes the user-defined function f for a write operation. 214 func (fd *FD) RawWrite(f func(uintptr) bool) error { 215 return errors.New("not implemented") 216 } 217
