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