file.go raw
1 // Copyright 2021 The Go Authors. All rights reserved.
2 // Copyright 2015 Microsoft
3 // Use of this source code is governed by a BSD-style
4 // license that can be found in the LICENSE file.
5
6 //go:build windows
7
8 package namedpipe
9
10 import (
11 "io"
12 "os"
13 "runtime"
14 "sync"
15 "sync/atomic"
16 "time"
17 "unsafe"
18
19 "golang.org/x/sys/windows"
20 )
21
22 type timeoutChan chan struct{}
23
24 var (
25 ioInitOnce sync.Once
26 ioCompletionPort windows.Handle
27 )
28
29 // ioResult contains the result of an asynchronous IO operation
30 type ioResult struct {
31 bytes uint32
32 err error
33 }
34
35 // ioOperation represents an outstanding asynchronous Win32 IO
36 type ioOperation struct {
37 o windows.Overlapped
38 ch chan ioResult
39 }
40
41 func initIo() {
42 h, err := windows.CreateIoCompletionPort(windows.InvalidHandle, 0, 0, 0)
43 if err != nil {
44 panic(err)
45 }
46 ioCompletionPort = h
47 go ioCompletionProcessor(h)
48 }
49
50 // file implements Reader, Writer, and Closer on a Win32 handle without blocking in a syscall.
51 // It takes ownership of this handle and will close it if it is garbage collected.
52 type file struct {
53 handle windows.Handle
54 wg sync.WaitGroup
55 wgLock sync.RWMutex
56 closing atomic.Bool
57 socket bool
58 readDeadline deadlineHandler
59 writeDeadline deadlineHandler
60 }
61
62 type deadlineHandler struct {
63 setLock sync.Mutex
64 channel timeoutChan
65 channelLock sync.RWMutex
66 timer *time.Timer
67 timedout atomic.Bool
68 }
69
70 // makeFile makes a new file from an existing file handle
71 func makeFile(h windows.Handle) (*file, error) {
72 f := &file{handle: h}
73 ioInitOnce.Do(initIo)
74 _, err := windows.CreateIoCompletionPort(h, ioCompletionPort, 0, 0)
75 if err != nil {
76 return nil, err
77 }
78 err = windows.SetFileCompletionNotificationModes(h, windows.FILE_SKIP_COMPLETION_PORT_ON_SUCCESS|windows.FILE_SKIP_SET_EVENT_ON_HANDLE)
79 if err != nil {
80 return nil, err
81 }
82 f.readDeadline.channel = make(timeoutChan)
83 f.writeDeadline.channel = make(timeoutChan)
84 return f, nil
85 }
86
87 // closeHandle closes the resources associated with a Win32 handle
88 func (f *file) closeHandle() {
89 f.wgLock.Lock()
90 // Atomically set that we are closing, releasing the resources only once.
91 if f.closing.Swap(true) == false {
92 f.wgLock.Unlock()
93 // cancel all IO and wait for it to complete
94 windows.CancelIoEx(f.handle, nil)
95 f.wg.Wait()
96 // at this point, no new IO can start
97 windows.Close(f.handle)
98 f.handle = 0
99 } else {
100 f.wgLock.Unlock()
101 }
102 }
103
104 // Close closes a file.
105 func (f *file) Close() error {
106 f.closeHandle()
107 return nil
108 }
109
110 // prepareIo prepares for a new IO operation.
111 // The caller must call f.wg.Done() when the IO is finished, prior to Close() returning.
112 func (f *file) prepareIo() (*ioOperation, error) {
113 f.wgLock.RLock()
114 if f.closing.Load() {
115 f.wgLock.RUnlock()
116 return nil, os.ErrClosed
117 }
118 f.wg.Add(1)
119 f.wgLock.RUnlock()
120 c := &ioOperation{}
121 c.ch = make(chan ioResult)
122 return c, nil
123 }
124
125 // ioCompletionProcessor processes completed async IOs forever
126 func ioCompletionProcessor(h windows.Handle) {
127 for {
128 var bytes uint32
129 var key uintptr
130 var op *ioOperation
131 err := windows.GetQueuedCompletionStatus(h, &bytes, &key, (**windows.Overlapped)(unsafe.Pointer(&op)), windows.INFINITE)
132 if op == nil {
133 panic(err)
134 }
135 op.ch <- ioResult{bytes, err}
136 }
137 }
138
139 // asyncIo processes the return value from ReadFile or WriteFile, blocking until
140 // the operation has actually completed.
141 func (f *file) asyncIo(c *ioOperation, d *deadlineHandler, bytes uint32, err error) (int, error) {
142 if err != windows.ERROR_IO_PENDING {
143 return int(bytes), err
144 }
145
146 if f.closing.Load() {
147 windows.CancelIoEx(f.handle, &c.o)
148 }
149
150 var timeout timeoutChan
151 if d != nil {
152 d.channelLock.Lock()
153 timeout = d.channel
154 d.channelLock.Unlock()
155 }
156
157 var r ioResult
158 select {
159 case r = <-c.ch:
160 err = r.err
161 if err == windows.ERROR_OPERATION_ABORTED {
162 if f.closing.Load() {
163 err = os.ErrClosed
164 }
165 } else if err != nil && f.socket {
166 // err is from Win32. Query the overlapped structure to get the winsock error.
167 var bytes, flags uint32
168 err = windows.WSAGetOverlappedResult(f.handle, &c.o, &bytes, false, &flags)
169 }
170 case <-timeout:
171 windows.CancelIoEx(f.handle, &c.o)
172 r = <-c.ch
173 err = r.err
174 if err == windows.ERROR_OPERATION_ABORTED {
175 err = os.ErrDeadlineExceeded
176 }
177 }
178
179 // runtime.KeepAlive is needed, as c is passed via native
180 // code to ioCompletionProcessor, c must remain alive
181 // until the channel read is complete.
182 runtime.KeepAlive(c)
183 return int(r.bytes), err
184 }
185
186 // Read reads from a file handle.
187 func (f *file) Read(b []byte) (int, error) {
188 c, err := f.prepareIo()
189 if err != nil {
190 return 0, err
191 }
192 defer f.wg.Done()
193
194 if f.readDeadline.timedout.Load() {
195 return 0, os.ErrDeadlineExceeded
196 }
197
198 var bytes uint32
199 err = windows.ReadFile(f.handle, b, &bytes, &c.o)
200 n, err := f.asyncIo(c, &f.readDeadline, bytes, err)
201 runtime.KeepAlive(b)
202
203 // Handle EOF conditions.
204 if err == nil && n == 0 && len(b) != 0 {
205 return 0, io.EOF
206 } else if err == windows.ERROR_BROKEN_PIPE {
207 return 0, io.EOF
208 } else {
209 return n, err
210 }
211 }
212
213 // Write writes to a file handle.
214 func (f *file) Write(b []byte) (int, error) {
215 c, err := f.prepareIo()
216 if err != nil {
217 return 0, err
218 }
219 defer f.wg.Done()
220
221 if f.writeDeadline.timedout.Load() {
222 return 0, os.ErrDeadlineExceeded
223 }
224
225 var bytes uint32
226 err = windows.WriteFile(f.handle, b, &bytes, &c.o)
227 n, err := f.asyncIo(c, &f.writeDeadline, bytes, err)
228 runtime.KeepAlive(b)
229 return n, err
230 }
231
232 func (f *file) SetReadDeadline(deadline time.Time) error {
233 return f.readDeadline.set(deadline)
234 }
235
236 func (f *file) SetWriteDeadline(deadline time.Time) error {
237 return f.writeDeadline.set(deadline)
238 }
239
240 func (f *file) Flush() error {
241 return windows.FlushFileBuffers(f.handle)
242 }
243
244 func (f *file) Fd() uintptr {
245 return uintptr(f.handle)
246 }
247
248 func (d *deadlineHandler) set(deadline time.Time) error {
249 d.setLock.Lock()
250 defer d.setLock.Unlock()
251
252 if d.timer != nil {
253 if !d.timer.Stop() {
254 <-d.channel
255 }
256 d.timer = nil
257 }
258 d.timedout.Store(false)
259
260 select {
261 case <-d.channel:
262 d.channelLock.Lock()
263 d.channel = make(chan struct{})
264 d.channelLock.Unlock()
265 default:
266 }
267
268 if deadline.IsZero() {
269 return nil
270 }
271
272 timeoutIO := func() {
273 d.timedout.Store(true)
274 close(d.channel)
275 }
276
277 now := time.Now()
278 duration := deadline.Sub(now)
279 if deadline.After(now) {
280 // Deadline is in the future, set a timer to wait
281 d.timer = time.AfterFunc(duration, timeoutIO)
282 } else {
283 // Deadline is in the past. Cancel all pending IO now.
284 timeoutIO()
285 }
286 return nil
287 }
288