diff --git a/src/reader.rs b/src/reader.rs
index a286cc1..c64a7f5 100644
--- a/src/reader.rs
+++ b/src/reader.rs
@@ -6,6 +6,7 @@ use std::task::{Context, Poll};
 use tokio::io::{self, AsyncRead, ReadBuf};
 
 /// The read half of the pipe which implements [`AsyncRead`](https://docs.rs/tokio/0.2.15/tokio/io/trait.AsyncRead.html).
+#[derive(Clone)]
 pub struct PipeReader {
     pub(crate) state: Arc<Mutex<State>>,
 }
@@ -59,6 +60,7 @@ impl PipeReader {
         let len = data.len.min(buf.capacity());
         unsafe {
             ptr::copy_nonoverlapping(data.ptr, buf.initialize_unfilled().as_mut_ptr(), len);
+            buf.advance(len);
         }
         len
     }
diff --git a/src/writer.rs b/src/writer.rs
index 119fcca..a8b3e5d 100644
--- a/src/writer.rs
+++ b/src/writer.rs
@@ -6,6 +6,7 @@ use std::task::{Context, Poll};
 use tokio::io::{self, AsyncWrite};
 
 /// The write half of the pipe which implements [`AsyncWrite`](https://docs.rs/tokio/0.2.16/tokio/io/trait.AsyncWrite.html).
+#[derive(Clone)]
 pub struct PipeWriter {
     pub(crate) state: Arc<Mutex<State>>,
 }