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