Optimize multipart handling #634, #769

actix-multipart/CHANGES.md

@@ -1,7 +1,9 @@
 # Changes
 
-## [0.1.0-alpha.1] - 2019-04-xx
+## [0.1.0-beta.1] - 2019-04-21
 
 * Do not support nested multipart
 
 * Split multipart support to separate crate
+
+* Optimize multipart handling #634, #769

actix-multipart/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "actix-multipart"
-version = "0.1.0-alpha.1"
+version = "0.1.0-beta.1"
 authors = ["Nikolay Kim <fafhrd91@gmail.com>"]
 description = "Multipart support for actix web framework."
 readme = "README.md"
@@ -18,8 +18,8 @@ name = "actix_multipart"
 path = "src/lib.rs"
 
 [dependencies]
-actix-web = "1.0.0-alpha.6"
+actix-web = "1.0.0-beta.1"
-actix-service = "0.3.4"
+actix-service = "0.3.6"
 bytes = "0.4"
 derive_more = "0.14"
 httparse = "1.3"
@@ -31,4 +31,4 @@ twoway = "0.2"
 
 [dev-dependencies]
 actix-rt = "0.2.2"
-actix-http = "0.1.0"
+actix-http = "0.1.1"

actix-multipart/src/server.rs

@@ -168,7 +168,7 @@ impl InnerMultipart {
         match payload.readline() {
             None => {
                 if payload.eof {
-                    Err(MultipartError::Incomplete)
+                    Ok(Some(true))
                 } else {
                     Ok(None)
                 }
@@ -201,8 +201,7 @@ impl InnerMultipart {
             match payload.readline() {
                 Some(chunk) => {
                     if chunk.is_empty() {
-                        //ValueError("Could not find starting boundary %r"
+                        return Err(MultipartError::Boundary);
-                        //% (self._boundary))
                     }
                     if chunk.len() < boundary.len() {
                         continue;
@@ -505,47 +504,73 @@ impl InnerField {
         payload: &mut PayloadBuffer,
         boundary: &str,
     ) -> Poll<Option<Bytes>, MultipartError> {
-        match payload.read_until(b"\r") {
+        let mut pos = 0;
-            None => {
+
-                if payload.eof {
+        let len = payload.buf.len();
-                    Err(MultipartError::Incomplete)
+        if len == 0 {
+            return Ok(Async::NotReady);
+        }
+
+        // check boundary
+        if len > 4 && payload.buf[0] == b'\r' {
+            let b_len = if &payload.buf[..2] == b"\r\n" && &payload.buf[2..4] == b"--" {
+                Some(4)
+            } else if &payload.buf[1..3] == b"--" {
+                Some(3)
+            } else {
+                None
+            };
+
+            if let Some(b_len) = b_len {
+                let b_size = boundary.len() + b_len;
+                if len < b_size {
+                    return Ok(Async::NotReady);
+                } else {
+                    if &payload.buf[b_len..b_size] == boundary.as_bytes() {
+                        // found boundary
+                        payload.buf.split_to(b_size);
+                        return Ok(Async::Ready(None));
+                    } else {
+                        pos = b_size;
+                    }
+                }
+            }
+        }
+
+        loop {
+            return if let Some(idx) = twoway::find_bytes(&payload.buf[pos..], b"\r") {
+                let cur = pos + idx;
+
+                // check if we have enough data for boundary detection
+                if cur + 4 > len {
+                    if cur > 0 {
+                        Ok(Async::Ready(Some(payload.buf.split_to(cur).freeze())))
                     } else {
                         Ok(Async::NotReady)
                     }
-            }
-            Some(mut chunk) => {
-                if chunk.len() == 1 {
-                    payload.unprocessed(chunk);
-                    match payload.read_exact(boundary.len() + 4) {
-                        None => {
-                            if payload.eof {
-                                Err(MultipartError::Incomplete)
                 } else {
-                                Ok(Async::NotReady)
+                    // check boundary
-                            }
+                    if (&payload.buf[cur..cur + 2] == b"\r\n"
-                        }
+                        && &payload.buf[cur + 2..cur + 4] == b"--")
-                        Some(mut chunk) => {
+                        || (&payload.buf[cur..cur + 1] == b"\r"
-                            if &chunk[..2] == b"\r\n"
+                            && &payload.buf[cur + 1..cur + 3] == b"--")
-                                && &chunk[2..4] == b"--"
-                                && &chunk[4..] == boundary.as_bytes()
                     {
-                                payload.unprocessed(chunk);
+                        if cur != 0 {
-                                Ok(Async::Ready(None))
+                            // return buffer
+                            Ok(Async::Ready(Some(payload.buf.split_to(cur).freeze())))
                         } else {
-                                // \r might be part of data stream
+                            pos = cur + 1;
-                                let ch = chunk.split_to(1);
+                            continue;
-                                payload.unprocessed(chunk);
-                                Ok(Async::Ready(Some(ch)))
                         }
+                    } else {
+                        // not boundary
+                        pos = cur + 1;
+                        continue;
                     }
                 }
             } else {
-                    let to = chunk.len() - 1;
+                return Ok(Async::Ready(Some(payload.buf.take().freeze())));
-                    let ch = chunk.split_to(to);
+            };
-                    payload.unprocessed(chunk);
-                    Ok(Async::Ready(Some(ch)))
-                }
-            }
         }
     }
 
@@ -555,6 +580,7 @@ impl InnerField {
         }
 
         let result = if let Some(payload) = self.payload.as_ref().unwrap().get_mut(s) {
+            if !self.eof {
                 let res = if let Some(ref mut len) = self.length {
                     InnerField::read_len(payload, len)?
                 } else {
@@ -562,10 +588,12 @@ impl InnerField {
                 };
 
                 match res {
-                Async::NotReady => Async::NotReady,
+                    Async::NotReady => return Ok(Async::NotReady),
-                Async::Ready(Some(bytes)) => Async::Ready(Some(bytes)),
+                    Async::Ready(Some(bytes)) => return Ok(Async::Ready(Some(bytes))),
-                Async::Ready(None) => {
+                    Async::Ready(None) => self.eof = true,
-                    self.eof = true;
+                }
+            }
+
             match payload.readline() {
                 None => Async::Ready(None),
                 Some(line) => {
@@ -575,8 +603,6 @@ impl InnerField {
                     Async::Ready(None)
                 }
             }
-                }
-            }
         } else {
             Async::NotReady
         };
@@ -704,7 +730,7 @@ impl PayloadBuffer {
     }
 
     /// Read exact number of bytes
-    #[inline]
+    #[cfg(test)]
     fn read_exact(&mut self, size: usize) -> Option<Bytes> {
         if size <= self.buf.len() {
             Some(self.buf.split_to(size).freeze())