1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338 339 340 341 342 343 344 345 346 347 348 349 350 351 352 353 354 355 356 357 358 359 360 361 362 363 364 365 366 367 368 369 370 371 372 373 374 375 376 377 378 379 380 381 382 383 384 385 386 387
//! Asynchronous I/O //! //! This crate contains the `AsyncRead` and `AsyncWrite` traits, the //! asynchronous analogs to `std::io::{Read, Write}`. The primary difference is //! that these traits integrate with the asynchronous task system. #![no_std] #![warn(missing_docs, missing_debug_implementations)] #![deny(bare_trait_objects)] #![doc(html_root_url = "https://rust-lang-nursery.github.io/futures-api-docs/0.3.0-alpha.3/futures_io")] #![feature(futures_api)] macro_rules! if_std { ($($i:item)*) => ($( #[cfg(feature = "std")] $i )*) } if_std! { use futures_core::task::{self, Poll}; use std::boxed::Box; use std::cmp; use std::io as StdIo; use std::ptr; // Re-export IoVec for convenience pub use iovec::IoVec; // Re-export io::Error so that users don't have to deal // with conflicts when `use`ing `futures::io` and `std::io`. pub use crate::StdIo::Error as Error; pub use crate::StdIo::ErrorKind as ErrorKind; pub use crate::StdIo::Result as Result; /// A type used to conditionally initialize buffers passed to `AsyncRead` /// methods, modeled after `std`. #[derive(Debug)] pub struct Initializer(bool); impl Initializer { /// Returns a new `Initializer` which will zero out buffers. #[inline] pub fn zeroing() -> Initializer { Initializer(true) } /// Returns a new `Initializer` which will not zero out buffers. /// /// # Safety /// /// This method may only be called by `AsyncRead`ers which guarantee /// that they will not read from the buffers passed to `AsyncRead` /// methods, and that the return value of the method accurately reflects /// the number of bytes that have been written to the head of the buffer. #[inline] pub unsafe fn nop() -> Initializer { Initializer(false) } /// Indicates if a buffer should be initialized. #[inline] pub fn should_initialize(&self) -> bool { self.0 } /// Initializes a buffer if necessary. #[inline] pub fn initialize(&self, buf: &mut [u8]) { if self.should_initialize() { unsafe { ptr::write_bytes(buf.as_mut_ptr(), 0, buf.len()) } } } } /// Read bytes asynchronously. /// /// This trait is analogous to the `std::io::Read` trait, but integrates /// with the asynchronous task system. In particular, the `poll_read` /// method, unlike `Read::read`, will automatically queue the current task /// for wakeup and return if data is not yet available, rather than blocking /// the calling thread. pub trait AsyncRead { /// Determines if this `AsyncRead`er can work with buffers of /// uninitialized memory. /// /// The default implementation returns an initializer which will zero /// buffers. /// /// # Safety /// /// This method is `unsafe` because and `AsyncRead`er could otherwise /// return a non-zeroing `Initializer` from another `AsyncRead` type /// without an `unsafe` block. #[inline] unsafe fn initializer(&self) -> Initializer { Initializer::zeroing() } /// Attempt to read from the `AsyncRead` into `buf`. /// /// On success, returns `Ok(Async::Ready(num_bytes_read))`. /// /// If no data is available for reading, the method returns /// `Ok(Async::Pending)` and arranges for the current task (via /// `cx.waker()`) to receive a notification when the object becomes /// readable or is closed. /// /// # Implementation /// /// This function may not return errors of kind `WouldBlock` or /// `Interrupted`. Implementations must convert `WouldBlock` into /// `Async::Pending` and either internally retry or convert /// `Interrupted` into another error kind. fn poll_read(&mut self, cx: &mut task::Context, buf: &mut [u8]) -> Poll<Result<usize>>; /// Attempt to read from the `AsyncRead` into `vec` using vectored /// IO operations. /// /// This method is similar to `poll_read`, but allows data to be read /// into multiple buffers using a single operation. /// /// On success, returns `Ok(Async::Ready(num_bytes_read))`. /// /// If no data is available for reading, the method returns /// `Ok(Async::Pending)` and arranges for the current task (via /// `cx.waker()`) to receive a notification when the object becomes /// readable or is closed. /// By default, this method delegates to using `poll_read` on the first /// buffer in `vec`. Objects which support vectored IO should override /// this method. /// /// # Implementation /// /// This function may not return errors of kind `WouldBlock` or /// `Interrupted`. Implementations must convert `WouldBlock` into /// `Async::Pending` and either internally retry or convert /// `Interrupted` into another error kind. fn poll_vectored_read(&mut self, cx: &mut task::Context, vec: &mut [&mut IoVec]) -> Poll<Result<usize>> { if let Some(ref mut first_iovec) = vec.get_mut(0) { self.poll_read(cx, first_iovec) } else { // `vec` is empty. Poll::Ready(Ok(0)) } } } /// Write bytes asynchronously. /// /// This trait is analogous to the `std::io::Write` trait, but integrates /// with the asynchronous task system. In particular, the `poll_write` /// method, unlike `Write::write`, will automatically queue the current task /// for wakeup and return if data is not yet available, rather than blocking /// the calling thread. pub trait AsyncWrite { /// Attempt to write bytes from `buf` into the object. /// /// On success, returns `Ok(Async::Ready(num_bytes_written))`. /// /// If the object is not ready for writing, the method returns /// `Ok(Async::Pending)` and arranges for the current task (via /// `cx.waker()`) to receive a notification when the object becomes /// readable or is closed. /// /// # Implementation /// /// This function may not return errors of kind `WouldBlock` or /// `Interrupted`. Implementations must convert `WouldBlock` into /// `Async::Pending` and either internally retry or convert /// `Interrupted` into another error kind. fn poll_write(&mut self, cx: &mut task::Context, buf: &[u8]) -> Poll<Result<usize>>; /// Attempt to write bytes from `vec` into the object using vectored /// IO operations. /// /// This method is similar to `poll_write`, but allows data from multiple buffers to be written /// using a single operation. /// /// On success, returns `Ok(Async::Ready(num_bytes_written))`. /// /// If the object is not ready for writing, the method returns /// `Ok(Async::Pending)` and arranges for the current task (via /// `cx.waker()`) to receive a notification when the object becomes /// readable or is closed. /// /// By default, this method delegates to using `poll_write` on the first /// buffer in `vec`. Objects which support vectored IO should override /// this method. /// /// # Implementation /// /// This function may not return errors of kind `WouldBlock` or /// `Interrupted`. Implementations must convert `WouldBlock` into /// `Async::Pending` and either internally retry or convert /// `Interrupted` into another error kind. fn poll_vectored_write(&mut self, cx: &mut task::Context, vec: &[&IoVec]) -> Poll<Result<usize>> { if let Some(ref first_iovec) = vec.get(0) { self.poll_write(cx, &*first_iovec) } else { // `vec` is empty. Poll::Ready(Ok(0)) } } /// Attempt to flush the object, ensuring that any buffered data reach /// their destination. /// /// On success, returns `Ok(Async::Ready(()))`. /// /// If flushing cannot immediately complete, this method returns /// `Ok(Async::Pending)` and arranges for the current task (via /// `cx.waker()`) to receive a notification when the object can make /// progress towards flushing. /// /// # Implementation /// /// This function may not return errors of kind `WouldBlock` or /// `Interrupted`. Implementations must convert `WouldBlock` into /// `Async::Pending` and either internally retry or convert /// `Interrupted` into another error kind. fn poll_flush(&mut self, cx: &mut task::Context) -> Poll<Result<()>>; /// Attempt to close the object. /// /// On success, returns `Ok(Async::Ready(()))`. /// /// If closing cannot immediately complete, this function returns /// `Ok(Async::Pending)` and arranges for the current task (via /// `cx.waker()`) to receive a notification when the object can make /// progress towards closing. /// /// # Implementation /// /// This function may not return errors of kind `WouldBlock` or /// `Interrupted`. Implementations must convert `WouldBlock` into /// `Async::Pending` and either internally retry or convert /// `Interrupted` into another error kind. fn poll_close(&mut self, cx: &mut task::Context) -> Poll<Result<()>>; } macro_rules! deref_async_read { () => { unsafe fn initializer(&self) -> Initializer { (**self).initializer() } fn poll_read(&mut self, cx: &mut task::Context, buf: &mut [u8]) -> Poll<Result<usize>> { (**self).poll_read(cx, buf) } fn poll_vectored_read(&mut self, cx: &mut task::Context, vec: &mut [&mut IoVec]) -> Poll<Result<usize>> { (**self).poll_vectored_read(cx, vec) } } } impl<T: ?Sized + AsyncRead> AsyncRead for Box<T> { deref_async_read!(); } impl<'a, T: ?Sized + AsyncRead> AsyncRead for &'a mut T { deref_async_read!(); } /// `unsafe` because the `StdIo::Read` type must not access the buffer /// before reading data into it. macro_rules! unsafe_delegate_async_read_to_stdio { () => { unsafe fn initializer(&self) -> Initializer { Initializer::nop() } fn poll_read(&mut self, _: &mut task::Context, buf: &mut [u8]) -> Poll<Result<usize>> { Poll::Ready(StdIo::Read::read(self, buf)) } } } impl<'a> AsyncRead for &'a [u8] { unsafe_delegate_async_read_to_stdio!(); } impl AsyncRead for StdIo::Repeat { unsafe_delegate_async_read_to_stdio!(); } impl<T: AsRef<[u8]>> AsyncRead for StdIo::Cursor<T> { unsafe_delegate_async_read_to_stdio!(); } macro_rules! deref_async_write { () => { fn poll_write(&mut self, cx: &mut task::Context, buf: &[u8]) -> Poll<Result<usize>> { (**self).poll_write(cx, buf) } fn poll_vectored_write(&mut self, cx: &mut task::Context, vec: &[&IoVec]) -> Poll<Result<usize>> { (**self).poll_vectored_write(cx, vec) } fn poll_flush(&mut self, cx: &mut task::Context) -> Poll<Result<()>> { (**self).poll_flush(cx) } fn poll_close(&mut self, cx: &mut task::Context) -> Poll<Result<()>> { (**self).poll_close(cx) } } } impl<T: ?Sized + AsyncWrite> AsyncWrite for Box<T> { deref_async_write!(); } impl<'a, T: ?Sized + AsyncWrite> AsyncWrite for &'a mut T { deref_async_write!(); } macro_rules! delegate_async_write_to_stdio { () => { fn poll_write(&mut self, _: &mut task::Context, buf: &[u8]) -> Poll<Result<usize>> { Poll::Ready(StdIo::Write::write(self, buf)) } fn poll_flush(&mut self, _: &mut task::Context) -> Poll<Result<()>> { Poll::Ready(StdIo::Write::flush(self)) } fn poll_close(&mut self, cx: &mut task::Context) -> Poll<Result<()>> { self.poll_flush(cx) } } } impl<T: AsMut<[u8]>> AsyncWrite for StdIo::Cursor<T> { fn poll_write( &mut self, _: &mut task::Context, buf: &[u8], ) -> Poll<Result<usize>> { let position = self.position(); let result = { let out = self.get_mut().as_mut(); let pos = cmp::min(out.len() as u64, position) as usize; StdIo::Write::write(&mut &mut out[pos..], buf) }; if let Ok(offset) = result { self.set_position(position + offset as u64); } Poll::Ready(result) } fn poll_flush(&mut self, _: &mut task::Context) -> Poll<Result<()>> { Poll::Ready(StdIo::Write::flush(&mut self.get_mut().as_mut())) } fn poll_close(&mut self, cx: &mut task::Context) -> Poll<Result<()>> { self.poll_flush(cx) } } impl AsyncWrite for StdIo::Sink { delegate_async_write_to_stdio!(); } }