src/libstd/sys/windows/tcp.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 use io::net::ip;
  12 use io::IoResult;
  13 use libc;
  14 use mem;
  15 use ptr;
  16 use prelude::v1::*;
  17 use super::{last_error, last_net_error, retry, sock_t};
  18 use sync::Arc;
  19 use sync::atomic::{AtomicBool, Ordering};
  20 use sys::fs::FileDesc;
  21 use sys::{self, c, set_nonblocking, wouldblock, timer};
  22 use sys_common::{self, timeout, eof, net};
  23
  24 pub use sys_common::net::TcpStream;
  25
  26 pub struct Event(c::WSAEVENT);
  27
  28 unsafe impl Send for Event {}
  29 unsafe impl Sync for Event {}
  30
  31 impl Event {
  32     pub fn new() -> IoResult<Event> {
  33         let event = unsafe { c::WSACreateEvent() };
  34         if event == c::WSA_INVALID_EVENT {
  35             Err(super::last_error())
  36         } else {
  37             Ok(Event(event))
  38         }
  39     }
  40
  41     pub fn handle(&self) -> c::WSAEVENT { let Event(handle) = *self; handle }
  42 }
  43
  44 impl Drop for Event {
  45     fn drop(&mut self) {
  46         unsafe { let _ = c::WSACloseEvent(self.handle()); }
  47     }
  48 }
  49
  50 ////////////////////////////////////////////////////////////////////////////////
  51 // TCP listeners
  52 ////////////////////////////////////////////////////////////////////////////////
  53
  54 pub struct TcpListener { sock: sock_t }
  55
  56 unsafe impl Send for TcpListener {}
  57 unsafe impl Sync for TcpListener {}
  58
  59 impl TcpListener {
  60     pub fn bind(addr: ip::SocketAddr) -> IoResult<TcpListener> {
  61         sys::init_net();
  62
  63         let sock = try!(net::socket(addr, libc::SOCK_STREAM));
  64         let ret = TcpListener { sock: sock };
  65
  66         let mut storage = unsafe { mem::zeroed() };
  67         let len = net::addr_to_sockaddr(addr, &mut storage);
  68         let addrp = &storage as *const _ as *const libc::sockaddr;
  69
  70         match unsafe { libc::bind(sock, addrp, len) } {
  71             -1 => Err(last_net_error()),
  72             _ => Ok(ret),
  73         }
  74     }
  75
  76     pub fn socket(&self) -> sock_t { self.sock }
  77
  78     pub fn listen(self, backlog: int) -> IoResult<TcpAcceptor> {
  79         match unsafe { libc::listen(self.socket(), backlog as libc::c_int) } {
  80             -1 => Err(last_net_error()),
  81
  82             _ => {
  83                 let accept = try!(Event::new());
  84                 let ret = unsafe {
  85                     c::WSAEventSelect(self.socket(), accept.handle(), c::FD_ACCEPT)
  86                 };
  87                 if ret != 0 {
  88                     return Err(last_net_error())
  89                 }
  90                 Ok(TcpAcceptor {
  91                     inner: Arc::new(AcceptorInner {
  92                         listener: self,
  93                         abort: try!(Event::new()),
  94                         accept: accept,
  95                         closed: AtomicBool::new(false),
  96                     }),
  97                     deadline: 0,
  98                 })
  99             }
 100         }
 101     }
 102
 103     pub fn socket_name(&mut self) -> IoResult<ip::SocketAddr> {
 104         net::sockname(self.socket(), libc::getsockname)
 105     }
 106 }
 107
 108 impl Drop for TcpListener {
 109     fn drop(&mut self) {
 110         unsafe { super::close_sock(self.sock); }
 111     }
 112 }
 113
 114 pub struct TcpAcceptor {
 115     inner: Arc<AcceptorInner>,
 116     deadline: u64,
 117 }
 118
 119 unsafe impl Send for TcpAcceptor {}
 120 unsafe impl Sync for TcpAcceptor {}
 121
 122 struct AcceptorInner {
 123     listener: TcpListener,
 124     abort: Event,
 125     accept: Event,
 126     closed: AtomicBool,
 127 }
 128
 129 unsafe impl Send for AcceptorInner {}
 130 unsafe impl Sync for AcceptorInner {}
 131
 132 impl TcpAcceptor {
 133     pub fn socket(&self) -> sock_t { self.inner.listener.socket() }
 134
 135     pub fn accept(&mut self) -> IoResult<TcpStream> {
 136         // Unlink unix, windows cannot invoke `select` on arbitrary file
 137         // descriptors like pipes, only sockets. Consequently, windows cannot
 138         // use the same implementation as unix for accept() when close_accept()
 139         // is considered.
 140         //
 141         // In order to implement close_accept() and timeouts, windows uses
 142         // event handles. An acceptor-specific abort event is created which
 143         // will only get set in close_accept(), and it will never be un-set.
 144         // Additionally, another acceptor-specific event is associated with the
 145         // FD_ACCEPT network event.
 146         //
 147         // These two events are then passed to WaitForMultipleEvents to see
 148         // which one triggers first, and the timeout passed to this function is
 149         // the local timeout for the acceptor.
 150         //
 151         // If the wait times out, then the accept timed out. If the wait
 152         // succeeds with the abort event, then we were closed, and if the wait
 153         // succeeds otherwise, then we do a nonblocking poll via `accept` to
 154         // see if we can accept a connection. The connection is candidate to be
 155         // stolen, so we do all of this in a loop as well.
 156         let events = [self.inner.abort.handle(), self.inner.accept.handle()];
 157
 158         while !self.inner.closed.load(Ordering::SeqCst) {
 159             let ms = if self.deadline == 0 {
 160                 c::WSA_INFINITE as u64
 161             } else {
 162                 let now = timer::now();
 163                 if self.deadline < now {0} else {self.deadline - now}
 164             };
 165             let ret = unsafe {
 166                 c::WSAWaitForMultipleEvents(2, events.as_ptr(), libc::FALSE,
 167                                             ms as libc::DWORD, libc::FALSE)
 168             };
 169             match ret {
 170                 c::WSA_WAIT_TIMEOUT => {
 171                     return Err(timeout("accept timed out"))
 172                 }
 173                 c::WSA_WAIT_FAILED => return Err(last_net_error()),
 174                 c::WSA_WAIT_EVENT_0 => break,
 175                 n => assert_eq!(n, c::WSA_WAIT_EVENT_0 + 1),
 176             }
 177
 178             let mut wsaevents: c::WSANETWORKEVENTS = unsafe { mem::zeroed() };
 179             let ret = unsafe {
 180                 c::WSAEnumNetworkEvents(self.socket(), events[1], &mut wsaevents)
 181             };
 182             if ret != 0 { return Err(last_net_error()) }
 183
 184             if wsaevents.lNetworkEvents & c::FD_ACCEPT == 0 { continue }
 185             match unsafe {
 186                 libc::accept(self.socket(), ptr::null_mut(), ptr::null_mut())
 187             } {
 188                 -1 if wouldblock() => {}
 189                 -1 => return Err(last_net_error()),
 190
 191                 // Accepted sockets inherit the same properties as the caller,
 192                 // so we need to deregister our event and switch the socket back
 193                 // to blocking mode
 194                 socket => {
 195                     let stream = TcpStream::new(socket);
 196                     let ret = unsafe {
 197                         c::WSAEventSelect(socket, events[1], 0)
 198                     };
 199                     if ret != 0 { return Err(last_net_error()) }
 200                     try!(set_nonblocking(socket, false));
 201                     return Ok(stream)
 202                 }
 203             }
 204         }
 205
 206         Err(eof())
 207     }
 208
 209     pub fn socket_name(&mut self) -> IoResult<ip::SocketAddr> {
 210         net::sockname(self.socket(), libc::getsockname)
 211     }
 212
 213     pub fn set_timeout(&mut self, timeout: Option<u64>) {
 214         self.deadline = timeout.map(|a| timer::now() + a).unwrap_or(0);
 215     }
 216
 217     pub fn close_accept(&mut self) -> IoResult<()> {
 218         self.inner.closed.store(true, Ordering::SeqCst);
 219         let ret = unsafe { c::WSASetEvent(self.inner.abort.handle()) };
 220         if ret == libc::TRUE {
 221             Ok(())
 222         } else {
 223             Err(last_net_error())
 224         }
 225     }
 226 }
 227
 228 impl Clone for TcpAcceptor {
 229     fn clone(&self) -> TcpAcceptor {
 230         TcpAcceptor {
 231             inner: self.inner.clone(),
 232             deadline: 0,
 233         }
 234     }
 235 }