diff --git a/Cargo.toml b/Cargo.toml
index 1f2396d..deb08a2 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -1,7 +1,10 @@
 [package]
 name = "rusb"
 version = "0.8.0"
-authors = ["David Cuddeback <david.cuddeback@gmail.com>", "Ilya Averyanov <a1ien.n3t@gmail.com>"]
+authors = [
+    "David Cuddeback <david.cuddeback@gmail.com>",
+    "Ilya Averyanov <a1ien.n3t@gmail.com>",
+]
 description = "Rust library for accessing USB devices."
 license = "MIT"
 homepage = "https://github.com/a1ien/rusb"
@@ -15,7 +18,7 @@ build = "build.rs"
 travis-ci = { repository = "a1ien/rusb" }
 
 [features]
-vendored = [ "libusb1-sys/vendored" ]
+vendored = ["libusb1-sys/vendored"]
 
 [workspace]
 members = ["libusb1-sys"]
@@ -23,6 +26,8 @@ members = ["libusb1-sys"]
 [dependencies]
 libusb1-sys = { path = "libusb1-sys", version = "0.5.0" }
 libc = "0.2"
+log = "0.4"
+thiserror = "1"
 
 [dev-dependencies]
 regex = "1"
diff --git a/examples/read_async.rs b/examples/read_async.rs
new file mode 100644
index 0000000..4e0cfee
--- /dev/null
+++ b/examples/read_async.rs
@@ -0,0 +1,68 @@
+use rusb::{AsyncPool, Context, UsbContext};
+
+use std::str::FromStr;
+use std::time::Duration;
+
+fn main() {
+    let args: Vec<String> = std::env::args().collect();
+
+    if args.len() < 4 {
+        eprintln!("Usage: read_async <vendor-id> <product-id> <endpoint>");
+        return;
+    }
+
+    let vid: u16 = FromStr::from_str(args[1].as_ref()).unwrap();
+    let pid: u16 = FromStr::from_str(args[2].as_ref()).unwrap();
+    let endpoint: u8 = FromStr::from_str(args[3].as_ref()).unwrap();
+
+    let ctx = Context::new().expect("Could not initialize libusb");
+    let device = ctx
+        .open_device_with_vid_pid(vid, pid)
+        .expect("Could not find device");
+
+    const NUM_TRANSFERS: usize = 32;
+    const BUF_SIZE: usize = 1024;
+
+    let mut buffers = Vec::new();
+    for _ in 0..NUM_TRANSFERS {
+        let buf = Vec::with_capacity(BUF_SIZE);
+        buffers.push(buf);
+    }
+
+    let mut async_pool =
+        AsyncPool::new_bulk(device, endpoint, buffers).expect("Failed to create async pool!");
+
+    let mut swap_vec = Vec::with_capacity(BUF_SIZE);
+    let timeout = Duration::from_secs(10);
+
+    let mut num_bytes = 0u64;
+    let mut num_transfers = 0u64;
+    let mut last_time = std::time::Instant::now();
+    loop {
+        let poll_result = async_pool.poll(timeout, swap_vec);
+        match poll_result {
+            Ok(data) => {
+                num_bytes += data.len() as u64;
+                swap_vec = data
+            }
+            Err((err, buf)) => {
+                eprintln!("Error: {}", err);
+                swap_vec = buf
+            }
+        }
+        num_transfers += 1;
+
+        let elapsed = last_time.elapsed();
+        if elapsed >= Duration::from_millis(1000) {
+            let elapsed = elapsed.as_secs_f32();
+            println!(
+                "KiB per second: {}, \tTx per second: {}",
+                num_bytes as f32 / 1024. / elapsed,
+                num_transfers as f32 / elapsed
+            );
+            num_bytes = 0;
+            num_transfers = 0;
+            last_time = std::time::Instant::now();
+        }
+    }
+}
diff --git a/src/device.rs b/src/device.rs
index bf1a9c3..d703952 100644
--- a/src/device.rs
+++ b/src/device.rs
@@ -10,9 +10,9 @@ use crate::{
     config_descriptor::{self, ConfigDescriptor},
     device_descriptor::{self, DeviceDescriptor},
     device_handle::DeviceHandle,
+    error,
     fields::{self, Speed},
     Error, UsbContext,
-    error,
 };
 
 /// A reference to a USB device.
@@ -150,27 +150,16 @@ impl<T: UsbContext> Device<T> {
     pub fn get_parent(&self) -> Option<Self> {
         let device = unsafe { libusb_get_parent(self.device.as_ptr()) };
         NonNull::new(device)
-            .map(|device|
-                 unsafe {
-                     Device::from_libusb(
-                         self.context.clone(),
-                         device,
-                     )
-                 }
-            )
+            .map(|device| unsafe { Device::from_libusb(self.context.clone(), device) })
     }
 
     ///  Get the list of all port numbers from root for the specified device
     pub fn port_numbers(&self) -> Result<Vec<u8>, Error> {
         // As per the USB 3.0 specs, the current maximum limit for the depth is 7.
-        let mut ports = [0;7];
+        let mut ports = [0; 7];
 
         let result = unsafe {
-            libusb_get_port_numbers(
-                self.device.as_ptr(),
-                ports.as_mut_ptr(),
-                ports.len() as i32
-            )
+            libusb_get_port_numbers(self.device.as_ptr(), ports.as_mut_ptr(), ports.len() as i32)
         };
 
         let ports_number = if result < 0 {
@@ -180,5 +169,4 @@ impl<T: UsbContext> Device<T> {
         };
         Ok(ports[0..ports_number as usize].to_vec())
     }
-
 }
diff --git a/src/device_handle/async_api.rs b/src/device_handle/async_api.rs
new file mode 100644
index 0000000..ba5de60
--- /dev/null
+++ b/src/device_handle/async_api.rs
@@ -0,0 +1,320 @@
+use crate::{DeviceHandle, UsbContext};
+use libusb1_sys as ffi;
+
+use libc::c_void;
+use std::collections::VecDeque;
+use std::convert::TryInto;
+use std::marker::PhantomPinned;
+use std::pin::Pin;
+use std::ptr::NonNull;
+use std::sync::{Arc, Mutex};
+use std::time::Duration;
+use thiserror::Error;
+
+type CompletedQueue = Arc<Mutex<VecDeque<usize>>>;
+type PinnedTransfer<C> = Pin<Box<AsyncTransfer<C>>>;
+
+#[derive(Error, Debug)]
+pub enum AsyncError {
+    #[error("Poll timed out")]
+    PollTimeout,
+    #[error("Transfer is stalled")]
+    Stall,
+    #[error("Device was disconnected")]
+    Disconnected,
+    #[error("Other Error: {0}")]
+    Other(&'static str),
+    #[error("{0}ERRNO: {1}")]
+    Errno(&'static str, i32),
+    #[error("Transfer was cancelled")]
+    Cancelled,
+    #[error("Transfer could not fit into provided buffer with size of {0}")]
+    Overflow(usize),
+}
+
+struct AsyncTransfer<C: UsbContext> {
+    ptr: NonNull<ffi::libusb_transfer>,
+    /// The ID of the transfer, as an idx into the transfer pool
+    pool_id: usize,
+    buffer: Vec<u8>,
+    completed_queue: CompletedQueue,
+    device: Arc<DeviceHandle<C>>,
+    // `ptr` holds a pointer to `buffer`, `device`, and `self`, so we must be `!Unpin`
+    _pin: PhantomPinned,
+}
+impl<C: UsbContext> AsyncTransfer<C> {
+    fn new_bulk(
+        pool_id: usize,
+        completed_queue: CompletedQueue,
+        device: Arc<DeviceHandle<C>>,
+        endpoint: u8,
+        buffer: Vec<u8>,
+    ) -> Pin<Box<Self>> {
+        // non-isochronous endpoints (e.g. control, bulk, interrupt) specify a value of 0
+        // This is step 1 of async API
+        let ptr = unsafe { ffi::libusb_alloc_transfer(0) };
+        let ptr = NonNull::new(ptr).expect("Could not allocate transfer!");
+
+        // Safety: Pinning `result` ensures it doesn't move, but we know that we will
+        // want to access its fields mutably, we just don't want its memory location
+        // changing (or its fields moving!). So routinely we will unsafely interact with
+        // its fields mutably through a shared reference, but this is still sound.
+        let result = Box::pin(Self {
+            ptr,
+            pool_id,
+            buffer,
+            completed_queue,
+            device,
+            _pin: PhantomPinned,
+        });
+        let user_data = std::ptr::addr_of!(*result) as *mut c_void;
+
+        unsafe {
+            // Step 2 of async api
+            ffi::libusb_fill_bulk_transfer(
+                ptr.as_ptr(),
+                result.device.as_raw(),
+                endpoint,
+                result.buffer.as_ptr() as *mut u8,
+                result.buffer.capacity().try_into().unwrap(),
+                Self::transfer_cb,
+                user_data, // We haven't associated with a queue yet, so this is null
+                0,
+            )
+        };
+        result
+    }
+
+    // Part of step 4 of async API the transfer is finished being handled when
+    // `poll()` is called.
+    extern "system" fn transfer_cb(transfer: *mut ffi::libusb_transfer) {
+        // Safety: libusb should never make this null, so this is fine
+        let transfer = unsafe { &mut *transfer };
+
+        // sanity
+        debug_assert_eq!(
+            transfer.transfer_type,
+            ffi::constants::LIBUSB_TRANSFER_TYPE_BULK
+        );
+
+        // Before doing anything else, handle the case where the transfer was cancelled
+        // and we need to clean up the libusb resource
+        // Step 5 of async API
+        if transfer.status == ffi::constants::LIBUSB_TRANSFER_CANCELLED {
+            // NOTE: transfer.user_data has already been deallocated in the destructor.
+            unsafe { ffi::libusb_free_transfer(transfer) }
+            return;
+        }
+
+        let transfer: *mut AsyncTransfer<C> = transfer.user_data.cast();
+        let transfer = unsafe { &mut *transfer };
+        // Mark transfer as completed
+        transfer
+            .completed_queue
+            .lock()
+            .unwrap()
+            .push_back(transfer.pool_id);
+    }
+
+    /// Prerequisite: self.buffer ans self.ptr are both correctly set
+    fn swap_buffer(self: &mut AsyncTransfer<C>, new_buf: Vec<u8>) -> Vec<u8> {
+        let transfer_struct = unsafe { self.ptr.as_mut() };
+
+        let data = std::mem::replace(&mut self.buffer, new_buf);
+
+        // Update transfer struct for new buffer
+        transfer_struct.actual_length = 0; // TODO: Is this necessary?
+        transfer_struct.buffer = self.buffer.as_mut_ptr();
+        transfer_struct.length = self.buffer.capacity() as i32;
+
+        data
+    }
+
+    // Step 3 of async API
+    fn submit(self: &mut AsyncTransfer<C>) -> Result<(), AsyncError> {
+        let transfer_struct = self.ptr;
+        let errno = unsafe { ffi::libusb_submit_transfer(transfer_struct.as_ptr()) };
+
+        use ffi::constants::*;
+        use AsyncError as E;
+        match errno {
+            0 => Ok(()),
+            LIBUSB_ERROR_NO_DEVICE => Err(E::Disconnected),
+            LIBUSB_ERROR_BUSY => {
+                unreachable!("We shouldn't be calling submit on transfers already submitted!")
+            }
+            LIBUSB_ERROR_NOT_SUPPORTED => Err(E::Other("Transfer not supported")),
+            LIBUSB_ERROR_INVALID_PARAM => Err(E::Other("Transfer size bigger than OS supports")),
+            _ => Err(E::Errno("Error while submitting transfer: ", errno)),
+        }
+    }
+}
+impl<C: UsbContext> Drop for AsyncTransfer<C> {
+    fn drop(&mut self) {
+        match unsafe { ffi::libusb_cancel_transfer(self.ptr.as_ptr()) } {
+            // Doesn't actually cancel until the callback completes, so lets block until
+            // then
+            0 => {
+                // We already requested the transfer to be cancelled, so this should
+                // immediately run the callback when libusb_handle_events() is invoked
+                const TIMEOUT: libc::timeval = libc::timeval {
+                    tv_sec: 0,
+                    tv_usec: 0,
+                };
+                let errno = unsafe {
+                    ffi::libusb_handle_events_timeout_completed(
+                        self.device.context().as_raw(),
+                        &TIMEOUT,
+                        std::ptr::null_mut(),
+                    )
+                };
+                assert_eq!(errno, 0, "Failed to cancel transfer! ERRNO: {}", errno);
+            }
+            // transfer is not in progress, already complete, or already cancelled.
+            // Therefore, no need to do anything special.
+            ffi::constants::LIBUSB_ERROR_NOT_FOUND => (),
+            errno => panic!(
+                "Error while requesting transfer cancellation! ERRNO: {}",
+                errno
+            ),
+        }
+    }
+}
+
+/// Represents a pool of asynchronous transfers, that can be polled to completion
+pub struct AsyncPool<C: UsbContext> {
+    /// Contains the pool of AsyncTransfers
+    pool: Vec<PinnedTransfer<C>>,
+    /// Contains the idxs of transfers in `pool` that have completed
+    completed: CompletedQueue,
+    device: Arc<DeviceHandle<C>>, // TODO: We hold refs to this, do we need it to be pinned?
+}
+impl<C: UsbContext> AsyncPool<C> {
+    pub fn new_bulk(
+        device: DeviceHandle<C>,
+        endpoint: u8,
+        buffers: impl IntoIterator<Item = Vec<u8>>,
+    ) -> Result<Self, AsyncError> {
+        let buffers = buffers.into_iter();
+        let mut pool = Vec::with_capacity(buffers.size_hint().0);
+        let completed = Arc::new(Mutex::new(VecDeque::with_capacity(buffers.size_hint().0)));
+        let device = Arc::new(device);
+
+        for (id, buf) in buffers.into_iter().enumerate() {
+            let mut transfer: PinnedTransfer<C> =
+                AsyncTransfer::new_bulk(id, completed.clone(), device.clone(), endpoint, buf);
+            unsafe { transfer.as_mut().get_unchecked_mut() }.submit()?;
+            pool.push(transfer);
+        }
+        Ok(Self {
+            pool,
+            completed,
+            device,
+        })
+    }
+
+    /// Polls for the completion of async transfers. If successful, will swap the buffer
+    /// of the completed transfer with `new_buf`, otherwise returns `(err, new_buf)` so
+    /// that `new_buf` may be repurposed.
+    pub fn poll(
+        &mut self,
+        timeout: Duration,
+        new_buf: Vec<u8>,
+    ) -> Result<Vec<u8>, (AsyncError, Vec<u8>)> {
+        let pop_result = { self.completed.lock().unwrap().pop_front() };
+        if let Some(id) = pop_result {
+            let ref mut transfer = self.pool[id];
+            return Self::handle_completed_transfer(transfer, new_buf);
+        }
+        // No completed transfers, so poll for some new ones
+        poll_transfers(self.device.context(), timeout);
+        let pop_result = { self.completed.lock().unwrap().pop_front() };
+        if let Some(id) = pop_result {
+            let ref mut transfer = self.pool[id];
+            Self::handle_completed_transfer(transfer, new_buf)
+        } else {
+            Err((AsyncError::PollTimeout, new_buf))
+        }
+    }
+
+    /// Returns the number of async transfers in the pool.
+    pub fn size(&self) -> usize {
+        self.pool.len()
+    }
+
+    /// Once a transfer is completed, check the c struct for errors, otherwise swap
+    /// buffers. Step 4 of async API.
+    fn handle_completed_transfer(
+        transfer: &mut PinnedTransfer<C>,
+        new_buf: Vec<u8>,
+    ) -> Result<Vec<u8>, (AsyncError, Vec<u8>)> {
+        use AsyncError as E;
+        let transfer = unsafe { transfer.as_mut().get_unchecked_mut() };
+        let transfer_struct = unsafe { transfer.ptr.as_mut() };
+
+        use ffi::constants::*;
+        let result = match transfer_struct.status {
+            LIBUSB_TRANSFER_COMPLETED => Ok(()),
+            LIBUSB_TRANSFER_CANCELLED => Err(E::Cancelled),
+            LIBUSB_TRANSFER_ERROR => Err(E::Other("Error occurred during transfer execution")),
+            LIBUSB_TRANSFER_TIMED_OUT => {
+                unreachable!("We are using timeout=0 which means no timeout")
+            }
+            LIBUSB_TRANSFER_STALL => Err(E::Stall),
+            LIBUSB_TRANSFER_NO_DEVICE => Err(E::Disconnected),
+            LIBUSB_TRANSFER_OVERFLOW => Err(E::Overflow(transfer.buffer.capacity())),
+            _ => panic!("Found an unexpected error value for transfer status"),
+        };
+        match result {
+            Err(err) => Err((err, new_buf)),
+            Ok(()) => {
+                debug_assert!(transfer_struct.length >= transfer_struct.actual_length); // sanity
+                unsafe {
+                    transfer
+                        .buffer
+                        .set_len(transfer_struct.actual_length as usize)
+                };
+
+                let data = transfer.swap_buffer(new_buf);
+
+                let submit_result = transfer.submit();
+                match submit_result {
+                    Ok(()) => Ok(data),
+                    Err(err) => {
+                        // Take back the original buffer and return the error
+                        let new_buf = transfer.swap_buffer(data);
+                        Err((err, new_buf))
+                    }
+                }
+            }
+        }
+    }
+}
+unsafe impl<C: UsbContext> Send for AsyncPool<C> {}
+unsafe impl<C: UsbContext> Sync for AsyncPool<C> {}
+
+/// Polls for transfers and executes their callbacks. Will block until the
+/// given timeout, or return immediately if timeout is zero.
+fn poll_transfers(ctx: &impl UsbContext, timeout: Duration) {
+    let timeval = libc::timeval {
+        tv_sec: timeout.as_secs().try_into().unwrap(),
+        tv_usec: timeout.subsec_millis().try_into().unwrap(),
+    };
+    unsafe {
+        let errno = ffi::libusb_handle_events_timeout_completed(
+            ctx.as_raw(),
+            std::ptr::addr_of!(timeval),
+            std::ptr::null_mut(),
+        );
+        use ffi::constants::*;
+        match errno {
+            0 => (),
+            LIBUSB_ERROR_INVALID_PARAM => panic!("Provided timeout was unexpectedly invalid"),
+            _ => panic!(
+                "Error when polling transfers. ERRNO: {}, Message: {}",
+                errno,
+                std::ffi::CStr::from_ptr(ffi::libusb_strerror(errno)).to_string_lossy()
+            ),
+        }
+    }
+}
diff --git a/src/device_handle.rs b/src/device_handle/mod.rs
similarity index 99%
rename from src/device_handle.rs
rename to src/device_handle/mod.rs
index 767145d..b80873b 100644
--- a/src/device_handle.rs
+++ b/src/device_handle/mod.rs
@@ -1,3 +1,5 @@
+pub mod async_api;
+
 use std::{mem, ptr::NonNull, time::Duration, u8};
 
 use libc::{c_int, c_uchar, c_uint};
diff --git a/src/device_list.rs b/src/device_list.rs
index d07630b..3eaf757 100644
--- a/src/device_list.rs
+++ b/src/device_list.rs
@@ -102,7 +102,12 @@ impl<'a, T: UsbContext> Iterator for Devices<'a, T> {
             let device = self.devices[self.index];
 
             self.index += 1;
-            Some(unsafe { device::Device::from_libusb(self.context.clone(), std::ptr::NonNull::new_unchecked(device)) })
+            Some(unsafe {
+                device::Device::from_libusb(
+                    self.context.clone(),
+                    std::ptr::NonNull::new_unchecked(device),
+                )
+            })
         } else {
             None
         }
diff --git a/src/lib.rs b/src/lib.rs
index 5dfc530..7f56703 100644
--- a/src/lib.rs
+++ b/src/lib.rs
@@ -7,6 +7,7 @@ pub use crate::{
     context::{Context, GlobalContext, Hotplug, LogLevel, Registration, UsbContext},
     device::Device,
     device_descriptor::DeviceDescriptor,
+    device_handle::async_api::{AsyncError, AsyncPool},
     device_handle::DeviceHandle,
     device_list::{DeviceList, Devices},
     endpoint_descriptor::EndpointDescriptor,
@@ -106,6 +107,11 @@ pub fn open_device_with_vid_pid(
     if handle.is_null() {
         None
     } else {
-        Some(unsafe { DeviceHandle::from_libusb(GlobalContext::default(), std::ptr::NonNull::new_unchecked(handle)) })
+        Some(unsafe {
+            DeviceHandle::from_libusb(
+                GlobalContext::default(),
+                std::ptr::NonNull::new_unchecked(handle),
+            )
+        })
     }
 }