1 // Copyright 2013-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.
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.
16 use std::rt::rtio::{IoResult, IoError};
18 use super::{retry, keep_going};
22 ////////////////////////////////////////////////////////////////////////////////
23 // sockaddr and misc bindings
24 ////////////////////////////////////////////////////////////////////////////////
26 #[cfg(windows)] pub type sock_t = libc::SOCKET;
27 #[cfg(unix)] pub type sock_t = super::file::fd_t;
29 pub fn htons(u: u16) -> u16 {
32 pub fn ntohs(u: u16) -> u16 {
37 InAddr(libc::in_addr),
38 In6Addr(libc::in6_addr),
41 fn ip_to_inaddr(ip: rtio::IpAddr) -> InAddr {
43 rtio::Ipv4Addr(a, b, c, d) => {
44 let ip = (a as u32 << 24) |
48 InAddr(libc::in_addr {
49 s_addr: mem::from_be32(ip)
52 rtio::Ipv6Addr(a, b, c, d, e, f, g, h) => {
53 In6Addr(libc::in6_addr {
69 fn addr_to_sockaddr(addr: rtio::SocketAddr) -> (libc::sockaddr_storage, uint) {
71 let storage: libc::sockaddr_storage = mem::zeroed();
72 let len = match ip_to_inaddr(addr.ip) {
74 let storage: *mut libc::sockaddr_in = mem::transmute(&storage);
75 (*storage).sin_family = libc::AF_INET as libc::sa_family_t;
76 (*storage).sin_port = htons(addr.port);
77 (*storage).sin_addr = inaddr;
78 mem::size_of::<libc::sockaddr_in>()
81 let storage: *mut libc::sockaddr_in6 = mem::transmute(&storage);
82 (*storage).sin6_family = libc::AF_INET6 as libc::sa_family_t;
83 (*storage).sin6_port = htons(addr.port);
84 (*storage).sin6_addr = inaddr;
85 mem::size_of::<libc::sockaddr_in6>()
88 return (storage, len);
92 fn socket(addr: rtio::SocketAddr, ty: libc::c_int) -> IoResult<sock_t> {
94 let fam = match addr.ip {
95 rtio::Ipv4Addr(..) => libc::AF_INET,
96 rtio::Ipv6Addr(..) => libc::AF_INET6,
98 match libc::socket(fam, ty, 0) {
99 -1 => Err(super::last_error()),
105 fn setsockopt<T>(fd: sock_t, opt: libc::c_int, val: libc::c_int,
106 payload: T) -> IoResult<()> {
108 let payload = &payload as *T as *libc::c_void;
109 let ret = libc::setsockopt(fd, opt, val,
111 mem::size_of::<T>() as libc::socklen_t);
120 pub fn getsockopt<T: Copy>(fd: sock_t, opt: libc::c_int,
121 val: libc::c_int) -> IoResult<T> {
123 let mut slot: T = mem::zeroed();
124 let mut len = mem::size_of::<T>() as libc::socklen_t;
125 let ret = c::getsockopt(fd, opt, val,
126 &mut slot as *mut _ as *mut _,
131 assert!(len as uint == mem::size_of::<T>());
138 pub fn last_error() -> IoError {
140 let code = unsafe { c::WSAGetLastError() as uint };
144 detail: Some(os::error_string(code)),
149 fn last_error() -> IoError {
153 #[cfg(windows)] unsafe fn close(sock: sock_t) { let _ = libc::closesocket(sock); }
154 #[cfg(unix)] unsafe fn close(sock: sock_t) { let _ = libc::close(sock); }
156 fn sockname(fd: sock_t,
157 f: unsafe extern "system" fn(sock_t, *mut libc::sockaddr,
158 *mut libc::socklen_t) -> libc::c_int)
159 -> IoResult<rtio::SocketAddr>
161 let mut storage: libc::sockaddr_storage = unsafe { mem::zeroed() };
162 let mut len = mem::size_of::<libc::sockaddr_storage>() as libc::socklen_t;
164 let storage = &mut storage as *mut libc::sockaddr_storage;
166 storage as *mut libc::sockaddr,
167 &mut len as *mut libc::socklen_t);
169 return Err(last_error())
172 return sockaddr_to_addr(&storage, len as uint);
175 pub fn sockaddr_to_addr(storage: &libc::sockaddr_storage,
176 len: uint) -> IoResult<rtio::SocketAddr> {
177 match storage.ss_family as libc::c_int {
179 assert!(len as uint >= mem::size_of::<libc::sockaddr_in>());
180 let storage: &libc::sockaddr_in = unsafe {
181 mem::transmute(storage)
183 let ip = mem::to_be32(storage.sin_addr.s_addr as u32);
184 let a = (ip >> 24) as u8;
185 let b = (ip >> 16) as u8;
186 let c = (ip >> 8) as u8;
187 let d = (ip >> 0) as u8;
188 Ok(rtio::SocketAddr {
189 ip: rtio::Ipv4Addr(a, b, c, d),
190 port: ntohs(storage.sin_port),
194 assert!(len as uint >= mem::size_of::<libc::sockaddr_in6>());
195 let storage: &libc::sockaddr_in6 = unsafe {
196 mem::transmute(storage)
198 let a = ntohs(storage.sin6_addr.s6_addr[0]);
199 let b = ntohs(storage.sin6_addr.s6_addr[1]);
200 let c = ntohs(storage.sin6_addr.s6_addr[2]);
201 let d = ntohs(storage.sin6_addr.s6_addr[3]);
202 let e = ntohs(storage.sin6_addr.s6_addr[4]);
203 let f = ntohs(storage.sin6_addr.s6_addr[5]);
204 let g = ntohs(storage.sin6_addr.s6_addr[6]);
205 let h = ntohs(storage.sin6_addr.s6_addr[7]);
206 Ok(rtio::SocketAddr {
207 ip: rtio::Ipv6Addr(a, b, c, d, e, f, g, h),
208 port: ntohs(storage.sin6_port),
212 #[cfg(unix)] use ERROR = libc::EINVAL;
213 #[cfg(windows)] use ERROR = libc::WSAEINVAL;
230 use std::rt::mutex::{StaticNativeMutex, NATIVE_MUTEX_INIT};
231 static mut INITIALIZED: bool = false;
232 static mut LOCK: StaticNativeMutex = NATIVE_MUTEX_INIT;
234 let _guard = LOCK.lock();
236 let mut data: c::WSADATA = mem::zeroed();
237 let ret = c::WSAStartup(0x202, // version 2.2
245 ////////////////////////////////////////////////////////////////////////////////
247 ////////////////////////////////////////////////////////////////////////////////
249 pub struct TcpStream {
257 lock: mutex::NativeMutex,
260 pub struct Guard<'a> {
262 pub guard: mutex::LockGuard<'a>,
266 fn new(fd: sock_t) -> Inner {
267 Inner { fd: fd, lock: unsafe { mutex::NativeMutex::new() } }
272 pub fn connect(addr: rtio::SocketAddr,
273 timeout: Option<u64>) -> IoResult<TcpStream> {
274 let fd = try!(socket(addr, libc::SOCK_STREAM));
275 let ret = TcpStream::new(Inner::new(fd));
277 let (addr, len) = addr_to_sockaddr(addr);
278 let addrp = &addr as *_ as *libc::sockaddr;
279 let len = len as libc::socklen_t;
283 try!(util::connect_timeout(fd, addrp, len, timeout));
287 match retry(|| unsafe { libc::connect(fd, addrp, len) }) {
288 -1 => Err(last_error()),
295 fn new(inner: Inner) -> TcpStream {
297 inner: Arc::new(inner),
303 pub fn fd(&self) -> sock_t { self.inner.fd }
305 fn set_nodelay(&mut self, nodelay: bool) -> IoResult<()> {
306 setsockopt(self.fd(), libc::IPPROTO_TCP, libc::TCP_NODELAY,
307 nodelay as libc::c_int)
310 fn set_keepalive(&mut self, seconds: Option<uint>) -> IoResult<()> {
311 let ret = setsockopt(self.fd(), libc::SOL_SOCKET, libc::SO_KEEPALIVE,
312 seconds.is_some() as libc::c_int);
314 Some(n) => ret.and_then(|()| self.set_tcp_keepalive(n)),
319 #[cfg(target_os = "macos")]
320 fn set_tcp_keepalive(&mut self, seconds: uint) -> IoResult<()> {
321 setsockopt(self.fd(), libc::IPPROTO_TCP, libc::TCP_KEEPALIVE,
322 seconds as libc::c_int)
324 #[cfg(target_os = "freebsd")]
325 fn set_tcp_keepalive(&mut self, seconds: uint) -> IoResult<()> {
326 setsockopt(self.fd(), libc::IPPROTO_TCP, libc::TCP_KEEPIDLE,
327 seconds as libc::c_int)
329 #[cfg(not(target_os = "macos"), not(target_os = "freebsd"))]
330 fn set_tcp_keepalive(&mut self, _seconds: uint) -> IoResult<()> {
334 #[cfg(target_os = "linux")]
335 fn lock_nonblocking(&self) {}
337 #[cfg(not(target_os = "linux"))]
338 fn lock_nonblocking<'a>(&'a self) -> Guard<'a> {
341 guard: unsafe { self.inner.lock.lock() },
343 assert!(util::set_nonblocking(self.fd(), true).is_ok());
348 #[cfg(windows)] type wrlen = libc::c_int;
349 #[cfg(not(windows))] type wrlen = libc::size_t;
351 impl rtio::RtioTcpStream for TcpStream {
352 fn read(&mut self, buf: &mut [u8]) -> IoResult<uint> {
354 let dolock = || self.lock_nonblocking();
355 let doread = |nb| unsafe {
356 let flags = if nb {c::MSG_DONTWAIT} else {0};
358 buf.as_mut_ptr() as *mut libc::c_void,
360 flags) as libc::c_int
362 read(fd, self.read_deadline, dolock, doread)
365 fn write(&mut self, buf: &[u8]) -> IoResult<()> {
367 let dolock = || self.lock_nonblocking();
368 let dowrite = |nb: bool, buf: *u8, len: uint| unsafe {
369 let flags = if nb {c::MSG_DONTWAIT} else {0};
371 buf as *mut libc::c_void,
375 match write(fd, self.write_deadline, buf, true, dolock, dowrite) {
380 fn peer_name(&mut self) -> IoResult<rtio::SocketAddr> {
381 sockname(self.fd(), libc::getpeername)
383 fn control_congestion(&mut self) -> IoResult<()> {
384 self.set_nodelay(false)
386 fn nodelay(&mut self) -> IoResult<()> {
387 self.set_nodelay(true)
389 fn keepalive(&mut self, delay_in_seconds: uint) -> IoResult<()> {
390 self.set_keepalive(Some(delay_in_seconds))
392 fn letdie(&mut self) -> IoResult<()> {
393 self.set_keepalive(None)
396 fn clone(&self) -> Box<rtio::RtioTcpStream:Send> {
398 inner: self.inner.clone(),
401 } as Box<rtio::RtioTcpStream:Send>
404 fn close_write(&mut self) -> IoResult<()> {
405 super::mkerr_libc(unsafe { libc::shutdown(self.fd(), libc::SHUT_WR) })
407 fn close_read(&mut self) -> IoResult<()> {
408 super::mkerr_libc(unsafe { libc::shutdown(self.fd(), libc::SHUT_RD) })
411 fn set_timeout(&mut self, timeout: Option<u64>) {
412 let deadline = timeout.map(|a| ::io::timer::now() + a).unwrap_or(0);
413 self.read_deadline = deadline;
414 self.write_deadline = deadline;
416 fn set_read_timeout(&mut self, timeout: Option<u64>) {
417 self.read_deadline = timeout.map(|a| ::io::timer::now() + a).unwrap_or(0);
419 fn set_write_timeout(&mut self, timeout: Option<u64>) {
420 self.write_deadline = timeout.map(|a| ::io::timer::now() + a).unwrap_or(0);
424 impl rtio::RtioSocket for TcpStream {
425 fn socket_name(&mut self) -> IoResult<rtio::SocketAddr> {
426 sockname(self.fd(), libc::getsockname)
430 impl Drop for Inner {
431 fn drop(&mut self) { unsafe { close(self.fd); } }
435 impl<'a> Drop for Guard<'a> {
437 assert!(util::set_nonblocking(self.fd, false).is_ok());
441 ////////////////////////////////////////////////////////////////////////////////
443 ////////////////////////////////////////////////////////////////////////////////
445 pub struct TcpListener {
450 pub fn bind(addr: rtio::SocketAddr) -> IoResult<TcpListener> {
451 let fd = try!(socket(addr, libc::SOCK_STREAM));
452 let ret = TcpListener { inner: Inner::new(fd) };
454 let (addr, len) = addr_to_sockaddr(addr);
455 let addrp = &addr as *_ as *libc::sockaddr;
456 let len = len as libc::socklen_t;
458 // On platforms with Berkeley-derived sockets, this allows
459 // to quickly rebind a socket, without needing to wait for
460 // the OS to clean up the previous one.
462 try!(setsockopt(fd, libc::SOL_SOCKET, libc::SO_REUSEADDR,
466 match unsafe { libc::bind(fd, addrp, len) } {
467 -1 => Err(last_error()),
472 pub fn fd(&self) -> sock_t { self.inner.fd }
474 pub fn native_listen(self, backlog: int) -> IoResult<TcpAcceptor> {
475 match unsafe { libc::listen(self.fd(), backlog as libc::c_int) } {
476 -1 => Err(last_error()),
477 _ => Ok(TcpAcceptor { listener: self, deadline: 0 })
482 impl rtio::RtioTcpListener for TcpListener {
483 fn listen(~self) -> IoResult<Box<rtio::RtioTcpAcceptor:Send>> {
484 self.native_listen(128).map(|a| {
485 box a as Box<rtio::RtioTcpAcceptor:Send>
490 impl rtio::RtioSocket for TcpListener {
491 fn socket_name(&mut self) -> IoResult<rtio::SocketAddr> {
492 sockname(self.fd(), libc::getsockname)
496 pub struct TcpAcceptor {
497 listener: TcpListener,
502 pub fn fd(&self) -> sock_t { self.listener.fd() }
504 pub fn native_accept(&mut self) -> IoResult<TcpStream> {
505 if self.deadline != 0 {
506 try!(util::await(self.fd(), Some(self.deadline), util::Readable));
509 let mut storage: libc::sockaddr_storage = mem::zeroed();
510 let storagep = &mut storage as *mut libc::sockaddr_storage;
511 let size = mem::size_of::<libc::sockaddr_storage>();
512 let mut size = size as libc::socklen_t;
514 libc::accept(self.fd(),
515 storagep as *mut libc::sockaddr,
516 &mut size as *mut libc::socklen_t) as libc::c_int
518 -1 => Err(last_error()),
519 fd => Ok(TcpStream::new(Inner::new(fd))),
525 impl rtio::RtioSocket for TcpAcceptor {
526 fn socket_name(&mut self) -> IoResult<rtio::SocketAddr> {
527 sockname(self.fd(), libc::getsockname)
531 impl rtio::RtioTcpAcceptor for TcpAcceptor {
532 fn accept(&mut self) -> IoResult<Box<rtio::RtioTcpStream:Send>> {
533 self.native_accept().map(|s| box s as Box<rtio::RtioTcpStream:Send>)
536 fn accept_simultaneously(&mut self) -> IoResult<()> { Ok(()) }
537 fn dont_accept_simultaneously(&mut self) -> IoResult<()> { Ok(()) }
538 fn set_timeout(&mut self, timeout: Option<u64>) {
539 self.deadline = timeout.map(|a| ::io::timer::now() + a).unwrap_or(0);
543 ////////////////////////////////////////////////////////////////////////////////
545 ////////////////////////////////////////////////////////////////////////////////
547 pub struct UdpSocket {
554 pub fn bind(addr: rtio::SocketAddr) -> IoResult<UdpSocket> {
555 let fd = try!(socket(addr, libc::SOCK_DGRAM));
556 let ret = UdpSocket {
557 inner: Arc::new(Inner::new(fd)),
562 let (addr, len) = addr_to_sockaddr(addr);
563 let addrp = &addr as *_ as *libc::sockaddr;
564 let len = len as libc::socklen_t;
566 match unsafe { libc::bind(fd, addrp, len) } {
567 -1 => Err(last_error()),
572 pub fn fd(&self) -> sock_t { self.inner.fd }
574 pub fn set_broadcast(&mut self, on: bool) -> IoResult<()> {
575 setsockopt(self.fd(), libc::SOL_SOCKET, libc::SO_BROADCAST,
579 pub fn set_multicast_loop(&mut self, on: bool) -> IoResult<()> {
580 setsockopt(self.fd(), libc::IPPROTO_IP, libc::IP_MULTICAST_LOOP,
584 pub fn set_membership(&mut self, addr: rtio::IpAddr,
585 opt: libc::c_int) -> IoResult<()> {
586 match ip_to_inaddr(addr) {
588 let mreq = libc::ip_mreq {
590 // interface == INADDR_ANY
591 imr_interface: libc::in_addr { s_addr: 0x0 },
593 setsockopt(self.fd(), libc::IPPROTO_IP, opt, mreq)
596 let mreq = libc::ip6_mreq {
597 ipv6mr_multiaddr: addr,
600 setsockopt(self.fd(), libc::IPPROTO_IPV6, opt, mreq)
605 #[cfg(target_os = "linux")]
606 fn lock_nonblocking(&self) {}
608 #[cfg(not(target_os = "linux"))]
609 fn lock_nonblocking<'a>(&'a self) -> Guard<'a> {
612 guard: unsafe { self.inner.lock.lock() },
614 assert!(util::set_nonblocking(self.fd(), true).is_ok());
619 impl rtio::RtioSocket for UdpSocket {
620 fn socket_name(&mut self) -> IoResult<rtio::SocketAddr> {
621 sockname(self.fd(), libc::getsockname)
625 #[cfg(windows)] type msglen_t = libc::c_int;
626 #[cfg(unix)] type msglen_t = libc::size_t;
628 impl rtio::RtioUdpSocket for UdpSocket {
629 fn recvfrom(&mut self, buf: &mut [u8]) -> IoResult<(uint, rtio::SocketAddr)> {
631 let mut storage: libc::sockaddr_storage = unsafe { mem::zeroed() };
632 let storagep = &mut storage as *mut _ as *mut libc::sockaddr;
633 let mut addrlen: libc::socklen_t =
634 mem::size_of::<libc::sockaddr_storage>() as libc::socklen_t;
636 let dolock = || self.lock_nonblocking();
637 let doread = |nb| unsafe {
638 let flags = if nb {c::MSG_DONTWAIT} else {0};
640 buf.as_mut_ptr() as *mut libc::c_void,
641 buf.len() as msglen_t,
644 &mut addrlen) as libc::c_int
646 let n = try!(read(fd, self.read_deadline, dolock, doread));
647 sockaddr_to_addr(&storage, addrlen as uint).and_then(|addr| {
648 Ok((n as uint, addr))
652 fn sendto(&mut self, buf: &[u8], dst: rtio::SocketAddr) -> IoResult<()> {
653 let (dst, dstlen) = addr_to_sockaddr(dst);
654 let dstp = &dst as *_ as *libc::sockaddr;
655 let dstlen = dstlen as libc::socklen_t;
658 let dolock = || self.lock_nonblocking();
659 let dowrite = |nb, buf: *u8, len: uint| unsafe {
660 let flags = if nb {c::MSG_DONTWAIT} else {0};
662 buf as *libc::c_void,
669 let n = try!(write(fd, self.write_deadline, buf, false, dolock, dowrite));
671 Err(util::short_write(n, "couldn't send entire packet at once"))
677 fn join_multicast(&mut self, multi: rtio::IpAddr) -> IoResult<()> {
679 rtio::Ipv4Addr(..) => {
680 self.set_membership(multi, libc::IP_ADD_MEMBERSHIP)
682 rtio::Ipv6Addr(..) => {
683 self.set_membership(multi, libc::IPV6_ADD_MEMBERSHIP)
687 fn leave_multicast(&mut self, multi: rtio::IpAddr) -> IoResult<()> {
689 rtio::Ipv4Addr(..) => {
690 self.set_membership(multi, libc::IP_DROP_MEMBERSHIP)
692 rtio::Ipv6Addr(..) => {
693 self.set_membership(multi, libc::IPV6_DROP_MEMBERSHIP)
698 fn loop_multicast_locally(&mut self) -> IoResult<()> {
699 self.set_multicast_loop(true)
701 fn dont_loop_multicast_locally(&mut self) -> IoResult<()> {
702 self.set_multicast_loop(false)
705 fn multicast_time_to_live(&mut self, ttl: int) -> IoResult<()> {
706 setsockopt(self.fd(), libc::IPPROTO_IP, libc::IP_MULTICAST_TTL,
709 fn time_to_live(&mut self, ttl: int) -> IoResult<()> {
710 setsockopt(self.fd(), libc::IPPROTO_IP, libc::IP_TTL, ttl as libc::c_int)
713 fn hear_broadcasts(&mut self) -> IoResult<()> {
714 self.set_broadcast(true)
716 fn ignore_broadcasts(&mut self) -> IoResult<()> {
717 self.set_broadcast(false)
720 fn clone(&self) -> Box<rtio::RtioUdpSocket:Send> {
722 inner: self.inner.clone(),
725 } as Box<rtio::RtioUdpSocket:Send>
728 fn set_timeout(&mut self, timeout: Option<u64>) {
729 let deadline = timeout.map(|a| ::io::timer::now() + a).unwrap_or(0);
730 self.read_deadline = deadline;
731 self.write_deadline = deadline;
733 fn set_read_timeout(&mut self, timeout: Option<u64>) {
734 self.read_deadline = timeout.map(|a| ::io::timer::now() + a).unwrap_or(0);
736 fn set_write_timeout(&mut self, timeout: Option<u64>) {
737 self.write_deadline = timeout.map(|a| ::io::timer::now() + a).unwrap_or(0);
741 ////////////////////////////////////////////////////////////////////////////////
744 // The read/write functions below are the helpers for reading/writing a socket
745 // with a possible deadline specified. This is generally viewed as a timed out
748 // From the application's perspective, timeouts apply to the I/O object, not to
749 // the underlying file descriptor (it's one timeout per object). This means that
750 // we can't use the SO_RCVTIMEO and corresponding send timeout option.
752 // The next idea to implement timeouts would be to use nonblocking I/O. An
753 // invocation of select() would wait (with a timeout) for a socket to be ready.
754 // Once its ready, we can perform the operation. Note that the operation *must*
755 // be nonblocking, even though select() says the socket is ready. This is
756 // because some other thread could have come and stolen our data (handles can be
759 // To implement nonblocking I/O, the first option we have is to use the
760 // O_NONBLOCK flag. Remember though that this is a global setting, affecting all
761 // I/O objects, so this was initially viewed as unwise.
763 // It turns out that there's this nifty MSG_DONTWAIT flag which can be passed to
764 // send/recv, but the niftiness wears off once you realize it only works well on
765 // linux [1] [2]. This means that it's pretty easy to get a nonblocking
766 // operation on linux (no flag fidding, no affecting other objects), but not on
769 // To work around this constraint on other platforms, we end up using the
770 // original strategy of flipping the O_NONBLOCK flag. As mentioned before, this
771 // could cause other objects' blocking operations to suddenly become
772 // nonblocking. To get around this, a "blocking operation" which returns EAGAIN
773 // falls back to using the same code path as nonblocking operations, but with an
774 // infinite timeout (select + send/recv). This helps emulate blocking
775 // reads/writes despite the underlying descriptor being nonblocking, as well as
776 // optimizing the fast path of just hitting one syscall in the good case.
778 // As a final caveat, this implementation uses a mutex so only one thread is
779 // doing a nonblocking operation at at time. This is the operation that comes
780 // after the select() (at which point we think the socket is ready). This is
781 // done for sanity to ensure that the state of the O_NONBLOCK flag is what we
782 // expect (wouldn't want someone turning it on when it should be off!). All
783 // operations performed in the lock are *nonblocking* to avoid holding the mutex
786 // So, in summary, linux uses MSG_DONTWAIT and doesn't need mutexes, everyone
787 // else uses O_NONBLOCK and mutexes with some trickery to make sure blocking
788 // reads/writes are still blocking.
792 // [1] http://twistedmatrix.com/pipermail/twisted-commits/2012-April/034692.html
793 // [2] http://stackoverflow.com/questions/19819198/does-send-msg-dontwait
795 pub fn read<T>(fd: sock_t,
798 read: |bool| -> libc::c_int) -> IoResult<uint> {
801 ret = retry(|| read(false));
804 if deadline != 0 || (ret == -1 && util::wouldblock()) {
805 let deadline = match deadline {
810 // With a timeout, first we wait for the socket to become
811 // readable using select(), specifying the relevant timeout for
812 // our previously set deadline.
813 try!(util::await(fd, deadline, util::Readable));
815 // At this point, we're still within the timeout, and we've
816 // determined that the socket is readable (as returned by
817 // select). We must still read the socket in *nonblocking* mode
818 // because some other thread could come steal our data. If we
819 // fail to read some data, we retry (hence the outer loop) and
820 // wait for the socket to become readable again.
822 match retry(|| read(deadline.is_some())) {
823 -1 if util::wouldblock() => { assert!(deadline.is_some()); }
824 -1 => return Err(last_error()),
825 n => { ret = n; break }
831 0 => Err(util::eof()),
832 n if n < 0 => Err(last_error()),
837 pub fn write<T>(fd: sock_t,
840 write_everything: bool,
842 write: |bool, *u8, uint| -> i64) -> IoResult<uint> {
846 if write_everything {
847 ret = keep_going(buf, |inner, len| {
848 written = buf.len() - len;
849 write(false, inner, len)
853 write(false, buf.as_ptr(), buf.len()) as libc::c_int
855 if ret > 0 { written = ret as uint; }
859 if deadline != 0 || (ret == -1 && util::wouldblock()) {
860 let deadline = match deadline {
864 while written < buf.len() && (write_everything || written == 0) {
865 // As with read(), first wait for the socket to be ready for
866 // the I/O operation.
867 match util::await(fd, deadline, util::Writable) {
868 Err(ref e) if e.code == libc::EOF as uint && written > 0 => {
869 assert!(deadline.is_some());
870 return Err(util::short_write(written, "short write"))
872 Err(e) => return Err(e),
876 // Also as with read(), we use MSG_DONTWAIT to guard ourselves
877 // against unforseen circumstances.
879 let ptr = buf.slice_from(written).as_ptr();
880 let len = buf.len() - written;
881 match retry(|| write(deadline.is_some(), ptr, len) as libc::c_int) {
882 -1 if util::wouldblock() => {}
883 -1 => return Err(last_error()),
884 n => { written += n as uint; }