]> git.lizzy.rs Git - rust.git/blob - src/libstd/sys/windows/pipe.rs
Merge pull request #20510 from tshepang/patch-6
[rust.git] / src / libstd / sys / windows / pipe.rs
1 // Copyright 2014 The Rust Project Developers. See the COPYRIGHT
2 // file at the top-level directory of this distribution and at
3 // http://rust-lang.org/COPYRIGHT.
4 //
5 // Licensed under the Apache License, Version 2.0 <LICENSE-APACHE or
6 // http://www.apache.org/licenses/LICENSE-2.0> or the MIT license
7 // <LICENSE-MIT or http://opensource.org/licenses/MIT>, at your
8 // option. This file may not be copied, modified, or distributed
9 // except according to those terms.
10
11 //! Named pipes implementation for windows
12 //!
13 //! If are unfortunate enough to be reading this code, I would like to first
14 //! apologize. This was my first encounter with windows named pipes, and it
15 //! didn't exactly turn out very cleanly. If you, too, are new to named pipes,
16 //! read on as I'll try to explain some fun things that I ran into.
17 //!
18 //! # Unix pipes vs Named pipes
19 //!
20 //! As with everything else, named pipes on windows are pretty different from
21 //! unix pipes on unix. On unix, you use one "server pipe" to accept new client
22 //! pipes. So long as this server pipe is active, new children pipes can
23 //! connect. On windows, you instead have a number of "server pipes", and each
24 //! of these server pipes can throughout their lifetime be attached to a client
25 //! or not. Once attached to a client, a server pipe may then disconnect at a
26 //! later date.
27 //!
28 //! # Accepting clients
29 //!
30 //! As with most other I/O interfaces, our Listener/Acceptor/Stream interfaces
31 //! are built around the unix flavors. This means that we have one "server
32 //! pipe" to which many clients can connect. In order to make this compatible
33 //! with the windows model, each connected client consumes ownership of a server
34 //! pipe, and then a new server pipe is created for the next client.
35 //!
36 //! Note that the server pipes attached to clients are never given back to the
37 //! listener for recycling. This could possibly be implemented with a channel so
38 //! the listener half can re-use server pipes, but for now I err'd on the simple
39 //! side of things. Each stream accepted by a listener will destroy the server
40 //! pipe after the stream is dropped.
41 //!
42 //! This model ends up having a small race or two, and you can find more details
43 //! on the `native_accept` method.
44 //!
45 //! # Simultaneous reads and writes
46 //!
47 //! In testing, I found that two simultaneous writes and two simultaneous reads
48 //! on a pipe ended up working out just fine, but problems were encountered when
49 //! a read was executed simultaneously with a write. After some googling around,
50 //! it sounded like named pipes just weren't built for this kind of interaction,
51 //! and the suggested solution was to use overlapped I/O.
52 //!
53 //! I don't really know what overlapped I/O is, but my basic understanding after
54 //! reading about it is that you have an external Event which is used to signal
55 //! I/O completion, passed around in some OVERLAPPED structures. As to what this
56 //! is, I'm not exactly sure.
57 //!
58 //! This problem implies that all named pipes are created with the
59 //! FILE_FLAG_OVERLAPPED option. This means that all of their I/O is
60 //! asynchronous. Each I/O operation has an associated OVERLAPPED structure, and
61 //! inside of this structure is a HANDLE from CreateEvent. After the I/O is
62 //! determined to be pending (may complete in the future), the
63 //! GetOverlappedResult function is used to block on the event, waiting for the
64 //! I/O to finish.
65 //!
66 //! This scheme ended up working well enough. There were two snags that I ran
67 //! into, however:
68 //!
69 //! * Each UnixStream instance needs its own read/write events to wait on. These
70 //!   can't be shared among clones of the same stream because the documentation
71 //!   states that it unsets the event when the I/O is started (would possibly
72 //!   corrupt other events simultaneously waiting). For convenience's sake,
73 //!   these events are lazily initialized.
74 //!
75 //! * Each server pipe needs to be created with FILE_FLAG_OVERLAPPED in addition
76 //!   to all pipes created through `connect`. Notably this means that the
77 //!   ConnectNamedPipe function is nonblocking, implying that the Listener needs
78 //!   to have yet another event to do the actual blocking.
79 //!
80 //! # Conclusion
81 //!
82 //! The conclusion here is that I probably don't know the best way to work with
83 //! windows named pipes, but the solution here seems to work well enough to get
84 //! the test suite passing (the suite is in libstd), and that's good enough for
85 //! me!
86
87 use prelude::v1::*;
88
89 use libc;
90 use c_str::CString;
91 use mem;
92 use ptr;
93 use sync::{Arc, Mutex};
94 use sync::atomic::{AtomicBool, Ordering};
95 use io::{self, IoError, IoResult};
96
97 use sys_common::{self, eof};
98
99 use super::{c, os, timer, to_utf16, decode_error_detailed};
100
101 struct Event(libc::HANDLE);
102
103 impl Event {
104     fn new(manual_reset: bool, initial_state: bool) -> IoResult<Event> {
105         let event = unsafe {
106             libc::CreateEventW(ptr::null_mut(),
107                                manual_reset as libc::BOOL,
108                                initial_state as libc::BOOL,
109                                ptr::null())
110         };
111         if event as uint == 0 {
112             Err(super::last_error())
113         } else {
114             Ok(Event(event))
115         }
116     }
117
118     fn handle(&self) -> libc::HANDLE { let Event(handle) = *self; handle }
119 }
120
121 impl Drop for Event {
122     fn drop(&mut self) {
123         unsafe { let _ = libc::CloseHandle(self.handle()); }
124     }
125 }
126
127 struct Inner {
128     handle: libc::HANDLE,
129     lock: Mutex<()>,
130     read_closed: AtomicBool,
131     write_closed: AtomicBool,
132 }
133
134 impl Inner {
135     fn new(handle: libc::HANDLE) -> Inner {
136         Inner {
137             handle: handle,
138             lock: Mutex::new(()),
139             read_closed: AtomicBool::new(false),
140             write_closed: AtomicBool::new(false),
141         }
142     }
143 }
144
145 impl Drop for Inner {
146     fn drop(&mut self) {
147         unsafe {
148             let _ = libc::FlushFileBuffers(self.handle);
149             let _ = libc::CloseHandle(self.handle);
150         }
151     }
152 }
153
154 unsafe fn pipe(name: *const u16, init: bool) -> libc::HANDLE {
155     libc::CreateNamedPipeW(
156         name,
157         libc::PIPE_ACCESS_DUPLEX |
158             if init {libc::FILE_FLAG_FIRST_PIPE_INSTANCE} else {0} |
159             libc::FILE_FLAG_OVERLAPPED,
160         libc::PIPE_TYPE_BYTE | libc::PIPE_READMODE_BYTE |
161             libc::PIPE_WAIT,
162         libc::PIPE_UNLIMITED_INSTANCES,
163         65536,
164         65536,
165         0,
166         ptr::null_mut()
167     )
168 }
169
170 pub fn await(handle: libc::HANDLE, deadline: u64,
171              events: &[libc::HANDLE]) -> IoResult<uint> {
172     use libc::consts::os::extra::{WAIT_FAILED, WAIT_TIMEOUT, WAIT_OBJECT_0};
173
174     // If we've got a timeout, use WaitForSingleObject in tandem with CancelIo
175     // to figure out if we should indeed get the result.
176     let ms = if deadline == 0 {
177         libc::INFINITE as u64
178     } else {
179         let now = timer::now();
180         if deadline < now {0} else {deadline - now}
181     };
182     let ret = unsafe {
183         c::WaitForMultipleObjects(events.len() as libc::DWORD,
184                                   events.as_ptr(),
185                                   libc::FALSE,
186                                   ms as libc::DWORD)
187     };
188     match ret {
189         WAIT_FAILED => Err(super::last_error()),
190         WAIT_TIMEOUT => unsafe {
191             let _ = c::CancelIo(handle);
192             Err(sys_common::timeout("operation timed out"))
193         },
194         n => Ok((n - WAIT_OBJECT_0) as uint)
195     }
196 }
197
198 fn epipe() -> IoError {
199     IoError {
200         kind: io::EndOfFile,
201         desc: "the pipe has ended",
202         detail: None,
203     }
204 }
205
206 ////////////////////////////////////////////////////////////////////////////////
207 // Unix Streams
208 ////////////////////////////////////////////////////////////////////////////////
209
210 pub struct UnixStream {
211     inner: Arc<Inner>,
212     write: Option<Event>,
213     read: Option<Event>,
214     read_deadline: u64,
215     write_deadline: u64,
216 }
217
218 unsafe impl Send for UnixStream {}
219 unsafe impl Sync for UnixStream {}
220
221 impl UnixStream {
222     fn try_connect(p: *const u16) -> Option<libc::HANDLE> {
223         // Note that most of this is lifted from the libuv implementation.
224         // The idea is that if we fail to open a pipe in read/write mode
225         // that we try afterwards in just read or just write
226         let mut result = unsafe {
227             libc::CreateFileW(p,
228                 libc::GENERIC_READ | libc::GENERIC_WRITE,
229                 0,
230                 ptr::null_mut(),
231                 libc::OPEN_EXISTING,
232                 libc::FILE_FLAG_OVERLAPPED,
233                 ptr::null_mut())
234         };
235         if result != libc::INVALID_HANDLE_VALUE {
236             return Some(result)
237         }
238
239         let err = unsafe { libc::GetLastError() };
240         if err == libc::ERROR_ACCESS_DENIED as libc::DWORD {
241             result = unsafe {
242                 libc::CreateFileW(p,
243                     libc::GENERIC_READ | libc::FILE_WRITE_ATTRIBUTES,
244                     0,
245                     ptr::null_mut(),
246                     libc::OPEN_EXISTING,
247                     libc::FILE_FLAG_OVERLAPPED,
248                     ptr::null_mut())
249             };
250             if result != libc::INVALID_HANDLE_VALUE {
251                 return Some(result)
252             }
253         }
254         let err = unsafe { libc::GetLastError() };
255         if err == libc::ERROR_ACCESS_DENIED as libc::DWORD {
256             result = unsafe {
257                 libc::CreateFileW(p,
258                     libc::GENERIC_WRITE | libc::FILE_READ_ATTRIBUTES,
259                     0,
260                     ptr::null_mut(),
261                     libc::OPEN_EXISTING,
262                     libc::FILE_FLAG_OVERLAPPED,
263                     ptr::null_mut())
264             };
265             if result != libc::INVALID_HANDLE_VALUE {
266                 return Some(result)
267             }
268         }
269         None
270     }
271
272     pub fn connect(addr: &CString, timeout: Option<u64>) -> IoResult<UnixStream> {
273         let addr = try!(to_utf16(addr.as_str()));
274         let start = timer::now();
275         loop {
276             match UnixStream::try_connect(addr.as_ptr()) {
277                 Some(handle) => {
278                     let inner = Inner::new(handle);
279                     let mut mode = libc::PIPE_TYPE_BYTE |
280                                    libc::PIPE_READMODE_BYTE |
281                                    libc::PIPE_WAIT;
282                     let ret = unsafe {
283                         libc::SetNamedPipeHandleState(inner.handle,
284                                                       &mut mode,
285                                                       ptr::null_mut(),
286                                                       ptr::null_mut())
287                     };
288                     return if ret == 0 {
289                         Err(super::last_error())
290                     } else {
291                         Ok(UnixStream {
292                             inner: Arc::new(inner),
293                             read: None,
294                             write: None,
295                             read_deadline: 0,
296                             write_deadline: 0,
297                         })
298                     }
299                 }
300                 None => {}
301             }
302
303             // On windows, if you fail to connect, you may need to call the
304             // `WaitNamedPipe` function, and this is indicated with an error
305             // code of ERROR_PIPE_BUSY.
306             let code = unsafe { libc::GetLastError() };
307             if code as int != libc::ERROR_PIPE_BUSY as int {
308                 return Err(super::last_error())
309             }
310
311             match timeout {
312                 Some(timeout) => {
313                     let now = timer::now();
314                     let timed_out = (now - start) >= timeout || unsafe {
315                         let ms = (timeout - (now - start)) as libc::DWORD;
316                         libc::WaitNamedPipeW(addr.as_ptr(), ms) == 0
317                     };
318                     if timed_out {
319                         return Err(sys_common::timeout("connect timed out"))
320                     }
321                 }
322
323                 // An example I found on Microsoft's website used 20
324                 // seconds, libuv uses 30 seconds, hence we make the
325                 // obvious choice of waiting for 25 seconds.
326                 None => {
327                     if unsafe { libc::WaitNamedPipeW(addr.as_ptr(), 25000) } == 0 {
328                         return Err(super::last_error())
329                     }
330                 }
331             }
332         }
333     }
334
335     pub fn handle(&self) -> libc::HANDLE { self.inner.handle }
336
337     fn read_closed(&self) -> bool {
338         self.inner.read_closed.load(Ordering::SeqCst)
339     }
340
341     fn write_closed(&self) -> bool {
342         self.inner.write_closed.load(Ordering::SeqCst)
343     }
344
345     fn cancel_io(&self) -> IoResult<()> {
346         match unsafe { c::CancelIoEx(self.handle(), ptr::null_mut()) } {
347             0 if os::errno() == libc::ERROR_NOT_FOUND as uint => {
348                 Ok(())
349             }
350             0 => Err(super::last_error()),
351             _ => Ok(())
352         }
353     }
354
355     pub fn read(&mut self, buf: &mut [u8]) -> IoResult<uint> {
356         if self.read.is_none() {
357             self.read = Some(try!(Event::new(true, false)));
358         }
359
360         let mut bytes_read = 0;
361         let mut overlapped: libc::OVERLAPPED = unsafe { mem::zeroed() };
362         overlapped.hEvent = self.read.as_ref().unwrap().handle();
363
364         // Pre-flight check to see if the reading half has been closed. This
365         // must be done before issuing the ReadFile request, but after we
366         // acquire the lock.
367         //
368         // See comments in close_read() about why this lock is necessary.
369         let guard = unsafe { self.inner.lock.lock() };
370         if self.read_closed() {
371             return Err(eof())
372         }
373
374         // Issue a nonblocking requests, succeeding quickly if it happened to
375         // succeed.
376         let ret = unsafe {
377             libc::ReadFile(self.handle(),
378                            buf.as_ptr() as libc::LPVOID,
379                            buf.len() as libc::DWORD,
380                            &mut bytes_read,
381                            &mut overlapped)
382         };
383         if ret != 0 { return Ok(bytes_read as uint) }
384
385         // If our errno doesn't say that the I/O is pending, then we hit some
386         // legitimate error and return immediately.
387         if os::errno() != libc::ERROR_IO_PENDING as uint {
388             return Err(super::last_error())
389         }
390
391         // Now that we've issued a successful nonblocking request, we need to
392         // wait for it to finish. This can all be done outside the lock because
393         // we'll see any invocation of CancelIoEx. We also call this in a loop
394         // because we're woken up if the writing half is closed, we just need to
395         // realize that the reading half wasn't closed and we go right back to
396         // sleep.
397         drop(guard);
398         loop {
399             // Process a timeout if one is pending
400             let wait_succeeded = await(self.handle(), self.read_deadline,
401                                        &[overlapped.hEvent]);
402
403             let ret = unsafe {
404                 libc::GetOverlappedResult(self.handle(),
405                                           &mut overlapped,
406                                           &mut bytes_read,
407                                           libc::TRUE)
408             };
409             // If we succeeded, or we failed for some reason other than
410             // CancelIoEx, return immediately
411             if ret != 0 { return Ok(bytes_read as uint) }
412             if os::errno() != libc::ERROR_OPERATION_ABORTED as uint {
413                 return Err(super::last_error())
414             }
415
416             // If the reading half is now closed, then we're done. If we woke up
417             // because the writing half was closed, keep trying.
418             if wait_succeeded.is_err() {
419                 return Err(sys_common::timeout("read timed out"))
420             }
421             if self.read_closed() {
422                 return Err(eof())
423             }
424         }
425     }
426
427     pub fn write(&mut self, buf: &[u8]) -> IoResult<()> {
428         if self.write.is_none() {
429             self.write = Some(try!(Event::new(true, false)));
430         }
431
432         let mut offset = 0;
433         let mut overlapped: libc::OVERLAPPED = unsafe { mem::zeroed() };
434         overlapped.hEvent = self.write.as_ref().unwrap().handle();
435
436         while offset < buf.len() {
437             let mut bytes_written = 0;
438
439             // This sequence below is quite similar to the one found in read().
440             // Some careful looping is done to ensure that if close_write() is
441             // invoked we bail out early, and if close_read() is invoked we keep
442             // going after we woke up.
443             //
444             // See comments in close_read() about why this lock is necessary.
445             let guard = unsafe { self.inner.lock.lock() };
446             if self.write_closed() {
447                 return Err(epipe())
448             }
449             let ret = unsafe {
450                 libc::WriteFile(self.handle(),
451                                 buf[offset..].as_ptr() as libc::LPVOID,
452                                 (buf.len() - offset) as libc::DWORD,
453                                 &mut bytes_written,
454                                 &mut overlapped)
455             };
456             let err = os::errno();
457             drop(guard);
458
459             if ret == 0 {
460                 if err != libc::ERROR_IO_PENDING as uint {
461                     return Err(decode_error_detailed(err as i32))
462                 }
463                 // Process a timeout if one is pending
464                 let wait_succeeded = await(self.handle(), self.write_deadline,
465                                            &[overlapped.hEvent]);
466                 let ret = unsafe {
467                     libc::GetOverlappedResult(self.handle(),
468                                               &mut overlapped,
469                                               &mut bytes_written,
470                                               libc::TRUE)
471                 };
472                 // If we weren't aborted, this was a legit error, if we were
473                 // aborted, then check to see if the write half was actually
474                 // closed or whether we woke up from the read half closing.
475                 if ret == 0 {
476                     if os::errno() != libc::ERROR_OPERATION_ABORTED as uint {
477                         return Err(super::last_error())
478                     }
479                     if !wait_succeeded.is_ok() {
480                         let amt = offset + bytes_written as uint;
481                         return if amt > 0 {
482                             Err(IoError {
483                                 kind: io::ShortWrite(amt),
484                                 desc: "short write during write",
485                                 detail: None,
486                             })
487                         } else {
488                             Err(sys_common::timeout("write timed out"))
489                         }
490                     }
491                     if self.write_closed() {
492                         return Err(epipe())
493                     }
494                     continue // retry
495                 }
496             }
497             offset += bytes_written as uint;
498         }
499         Ok(())
500     }
501
502     pub fn close_read(&mut self) -> IoResult<()> {
503         // On windows, there's no actual shutdown() method for pipes, so we're
504         // forced to emulate the behavior manually at the application level. To
505         // do this, we need to both cancel any pending requests, as well as
506         // prevent all future requests from succeeding. These two operations are
507         // not atomic with respect to one another, so we must use a lock to do
508         // so.
509         //
510         // The read() code looks like:
511         //
512         //      1. Make sure the pipe is still open
513         //      2. Submit a read request
514         //      3. Wait for the read request to finish
515         //
516         // The race this lock is preventing is if another thread invokes
517         // close_read() between steps 1 and 2. By atomically executing steps 1
518         // and 2 with a lock with respect to close_read(), we're guaranteed that
519         // no thread will erroneously sit in a read forever.
520         let _guard = unsafe { self.inner.lock.lock() };
521         self.inner.read_closed.store(true, Ordering::SeqCst);
522         self.cancel_io()
523     }
524
525     pub fn close_write(&mut self) -> IoResult<()> {
526         // see comments in close_read() for why this lock is necessary
527         let _guard = unsafe { self.inner.lock.lock() };
528         self.inner.write_closed.store(true, Ordering::SeqCst);
529         self.cancel_io()
530     }
531
532     pub fn set_timeout(&mut self, timeout: Option<u64>) {
533         let deadline = timeout.map(|a| timer::now() + a).unwrap_or(0);
534         self.read_deadline = deadline;
535         self.write_deadline = deadline;
536     }
537     pub fn set_read_timeout(&mut self, timeout: Option<u64>) {
538         self.read_deadline = timeout.map(|a| timer::now() + a).unwrap_or(0);
539     }
540     pub fn set_write_timeout(&mut self, timeout: Option<u64>) {
541         self.write_deadline = timeout.map(|a| timer::now() + a).unwrap_or(0);
542     }
543 }
544
545 impl Clone for UnixStream {
546     fn clone(&self) -> UnixStream {
547         UnixStream {
548             inner: self.inner.clone(),
549             read: None,
550             write: None,
551             read_deadline: 0,
552             write_deadline: 0,
553         }
554     }
555 }
556
557 ////////////////////////////////////////////////////////////////////////////////
558 // Unix Listener
559 ////////////////////////////////////////////////////////////////////////////////
560
561 pub struct UnixListener {
562     handle: libc::HANDLE,
563     name: CString,
564 }
565
566 unsafe impl Send for UnixListener {}
567 unsafe impl Sync for UnixListener {}
568
569 impl UnixListener {
570     pub fn bind(addr: &CString) -> IoResult<UnixListener> {
571         // Although we technically don't need the pipe until much later, we
572         // create the initial handle up front to test the validity of the name
573         // and such.
574         let addr_v = try!(to_utf16(addr.as_str()));
575         let ret = unsafe { pipe(addr_v.as_ptr(), true) };
576         if ret == libc::INVALID_HANDLE_VALUE {
577             Err(super::last_error())
578         } else {
579             Ok(UnixListener { handle: ret, name: addr.clone() })
580         }
581     }
582
583     pub fn listen(self) -> IoResult<UnixAcceptor> {
584         Ok(UnixAcceptor {
585             listener: self,
586             event: try!(Event::new(true, false)),
587             deadline: 0,
588             inner: Arc::new(AcceptorState {
589                 abort: try!(Event::new(true, false)),
590                 closed: AtomicBool::new(false),
591             }),
592         })
593     }
594
595     pub fn handle(&self) -> libc::HANDLE {
596         self.handle
597     }
598 }
599
600 impl Drop for UnixListener {
601     fn drop(&mut self) {
602         unsafe { let _ = libc::CloseHandle(self.handle); }
603     }
604 }
605
606 pub struct UnixAcceptor {
607     inner: Arc<AcceptorState>,
608     listener: UnixListener,
609     event: Event,
610     deadline: u64,
611 }
612
613 unsafe impl Send for UnixAcceptor {}
614 unsafe impl Sync for UnixAcceptor {}
615
616 struct AcceptorState {
617     abort: Event,
618     closed: AtomicBool,
619 }
620
621 unsafe impl Send for AcceptorState {}
622 unsafe impl Sync for AcceptorState {}
623
624 impl UnixAcceptor {
625     pub fn accept(&mut self) -> IoResult<UnixStream> {
626         // This function has some funky implementation details when working with
627         // unix pipes. On windows, each server named pipe handle can be
628         // connected to a one or zero clients. To the best of my knowledge, a
629         // named server is considered active and present if there exists at
630         // least one server named pipe for it.
631         //
632         // The model of this function is to take the current known server
633         // handle, connect a client to it, and then transfer ownership to the
634         // UnixStream instance. The next time accept() is invoked, it'll need a
635         // different server handle to connect a client to.
636         //
637         // Note that there is a possible race here. Once our server pipe is
638         // handed off to a `UnixStream` object, the stream could be closed,
639         // meaning that there would be no active server pipes, hence even though
640         // we have a valid `UnixAcceptor`, no one can connect to it. For this
641         // reason, we generate the next accept call's server pipe at the end of
642         // this function call.
643         //
644         // This provides us an invariant that we always have at least one server
645         // connection open at a time, meaning that all connects to this acceptor
646         // should succeed while this is active.
647         //
648         // The actual implementation of doing this is a little tricky. Once a
649         // server pipe is created, a client can connect to it at any time. I
650         // assume that which server a client connects to is nondeterministic, so
651         // we also need to guarantee that the only server able to be connected
652         // to is the one that we're calling ConnectNamedPipe on. This means that
653         // we have to create the second server pipe *after* we've already
654         // accepted a connection. In order to at least somewhat gracefully
655         // handle errors, this means that if the second server pipe creation
656         // fails that we disconnect the connected client and then just keep
657         // using the original server pipe.
658         let handle = self.listener.handle;
659
660         // If we've had an artificial call to close_accept, be sure to never
661         // proceed in accepting new clients in the future
662         if self.inner.closed.load(Ordering::SeqCst) { return Err(eof()) }
663
664         let name = try!(to_utf16(self.listener.name.as_str()));
665
666         // Once we've got a "server handle", we need to wait for a client to
667         // connect. The ConnectNamedPipe function will block this thread until
668         // someone on the other end connects. This function can "fail" if a
669         // client connects after we created the pipe but before we got down
670         // here. Thanks windows.
671         let mut overlapped: libc::OVERLAPPED = unsafe { mem::zeroed() };
672         overlapped.hEvent = self.event.handle();
673         if unsafe { libc::ConnectNamedPipe(handle, &mut overlapped) == 0 } {
674             let mut err = unsafe { libc::GetLastError() };
675
676             if err == libc::ERROR_IO_PENDING as libc::DWORD {
677                 // Process a timeout if one is pending
678                 let wait_succeeded = await(handle, self.deadline,
679                                            &[self.inner.abort.handle(),
680                                              overlapped.hEvent]);
681
682                 // This will block until the overlapped I/O is completed. The
683                 // timeout was previously handled, so this will either block in
684                 // the normal case or succeed very quickly in the timeout case.
685                 let ret = unsafe {
686                     let mut transfer = 0;
687                     libc::GetOverlappedResult(handle,
688                                               &mut overlapped,
689                                               &mut transfer,
690                                               libc::TRUE)
691                 };
692                 if ret == 0 {
693                     if wait_succeeded.is_ok() {
694                         err = unsafe { libc::GetLastError() };
695                     } else {
696                         return Err(sys_common::timeout("accept timed out"))
697                     }
698                 } else {
699                     // we succeeded, bypass the check below
700                     err = libc::ERROR_PIPE_CONNECTED as libc::DWORD;
701                 }
702             }
703             if err != libc::ERROR_PIPE_CONNECTED as libc::DWORD {
704                 return Err(super::last_error())
705             }
706         }
707
708         // Now that we've got a connected client to our handle, we need to
709         // create a second server pipe. If this fails, we disconnect the
710         // connected client and return an error (see comments above).
711         let new_handle = unsafe { pipe(name.as_ptr(), false) };
712         if new_handle == libc::INVALID_HANDLE_VALUE {
713             let ret = Err(super::last_error());
714             // If our disconnection fails, then there's not really a whole lot
715             // that we can do, so panic
716             let err = unsafe { libc::DisconnectNamedPipe(handle) };
717             assert!(err != 0);
718             return ret;
719         } else {
720             self.listener.handle = new_handle;
721         }
722
723         // Transfer ownership of our handle into this stream
724         Ok(UnixStream {
725             inner: Arc::new(Inner::new(handle)),
726             read: None,
727             write: None,
728             read_deadline: 0,
729             write_deadline: 0,
730         })
731     }
732
733     pub fn set_timeout(&mut self, timeout: Option<u64>) {
734         self.deadline = timeout.map(|i| i + timer::now()).unwrap_or(0);
735     }
736
737     pub fn close_accept(&mut self) -> IoResult<()> {
738         self.inner.closed.store(true, Ordering::SeqCst);
739         let ret = unsafe {
740             c::SetEvent(self.inner.abort.handle())
741         };
742         if ret == 0 {
743             Err(super::last_error())
744         } else {
745             Ok(())
746         }
747     }
748
749     pub fn handle(&self) -> libc::HANDLE {
750         self.listener.handle()
751     }
752 }
753
754 impl Clone for UnixAcceptor {
755     fn clone(&self) -> UnixAcceptor {
756         let name = to_utf16(self.listener.name.as_str()).ok().unwrap();
757         UnixAcceptor {
758             inner: self.inner.clone(),
759             event: Event::new(true, false).ok().unwrap(),
760             deadline: 0,
761             listener: UnixListener {
762                 name: self.listener.name.clone(),
763                 handle: unsafe {
764                     let p = pipe(name.as_ptr(), false) ;
765                     assert!(p != libc::INVALID_HANDLE_VALUE as libc::HANDLE);
766                     p
767                 },
768             },
769         }
770     }
771 }