Auto merge of #9850 - nikkisquared:2_async_2_furious, r=jdm

Set response.body Asynchronously In Fetch

Following having finished making Fetch asynchronous, response.body should be set asynchronously, since it's the major goal of calling Fetch. So far, I've made the body wrapped in Arc<Mutex<>>, and I've wrapped a new thread around the part where it's set. I've also discovered that the fetch_async function makes step 8 of Main Fetch obsolete, and I've commented it appropriately.

I'm currently having a hard time with the thread for setting response.body, though. @jdm suggested I have the body set continually, block by block, but my implementation for that runs so slow that I can't finish running my fetch test suite in reasonable time. @KiChjang pointed out that a lot of the lag is due to how response.body currently stores everything inside a Vec. Changing the storage container seems to be both necessary and beyond the scope of the time I have to work on this.

<!-- Reviewable:start -->
[<img src="https://reviewable.io/review_button.svg" height="40" alt="Review on Reviewable"/>](https://reviewable.io/reviews/servo/servo/9850)
<!-- Reviewable:end -->

components/net/fetch/methods.rs

@@ -5,7 +5,7 @@
 use fetch::cors_cache::{BasicCORSCache, CORSCache, CacheRequestDetails};
 use fetch::response::ResponseMethods;
 use http_loader::{NetworkHttpRequestFactory, WrappedHttpResponse};
-use http_loader::{create_http_connector, obtain_response};
+use http_loader::{create_http_connector, obtain_response, read_block, ReadResult};
 use hyper::client::response::Response as HyperResponse;
 use hyper::header::{Accept, CacheControl, IfMatch, IfRange, IfUnmodifiedSince, Location};
 use hyper::header::{AcceptLanguage, ContentLength, ContentLanguage, HeaderView, Pragma};
@@ -27,6 +27,7 @@ use std::cell::RefCell;
 use std::io::Read;
 use std::rc::Rc;
 use std::str::FromStr;
+use std::sync::{Arc, Mutex};
 use std::thread;
 use url::idna::domain_to_ascii;
 use url::{Origin as UrlOrigin, OpaqueOrigin, Url, UrlParser, whatwg_scheme_type_mapper};
@@ -35,8 +36,9 @@ use util::thread::spawn_named;
 pub fn fetch_async(request: Request, listener: Box<AsyncFetchListener + Send>) {
     spawn_named(format!("fetch for {:?}", request.current_url_string()), move || {
         let request = Rc::new(request);
-        let res = fetch(request);
+        let fetch_response = fetch(request);
-        listener.response_available(res);
+        fetch_response.wait_until_done();
+        listener.response_available(fetch_response);
     })
 }
 
@@ -140,9 +142,7 @@ fn main_fetch(request: Rc<Request>, cors_flag: bool, recursive_flag: bool) -> Re
     // TODO this step
 
     // Step 8
-    if !request.synchronous && !recursive_flag {
+    // this step is obsoleted by fetch_async
-        // TODO run the remaining steps in parallel
-    }
 
     // Step 9
     let mut response = if response.is_none() {
@@ -228,9 +228,10 @@ fn main_fetch(request: Rc<Request>, cors_flag: bool, recursive_flag: bool) -> Re
                 Method::Head | Method::Connect => true,
                 _ => false })
             {
-            // when the Fetch implementation does asynchronous retrieval of the body,
+            // when Fetch is used only asynchronously, we will need to make sure
-            // we will need to make sure nothing tries to write to the body at this point
+            // that nothing tries to write to the body at this point
-            *internal_response.body.borrow_mut() = ResponseBody::Empty;
+            let mut body = internal_response.body.lock().unwrap();
+            *body = ResponseBody::Empty;
         }
 
         // Step 15
@@ -238,7 +239,7 @@ fn main_fetch(request: Rc<Request>, cors_flag: bool, recursive_flag: bool) -> Re
         // if !response.is_network_error() {
 
         //     // Substep 1
-        //     // TODO wait for response
+        //     response.wait_until_done();
 
         //     // Substep 2
         //     if response.termination_reason.is_none() {
@@ -250,7 +251,7 @@ fn main_fetch(request: Rc<Request>, cors_flag: bool, recursive_flag: bool) -> Re
 
     // Step 16
     if request.synchronous {
-        // TODO wait for internal_response
+        response.get_actual_response().wait_until_done();
         return response;
     }
 
@@ -274,22 +275,14 @@ fn main_fetch(request: Rc<Request>, cors_flag: bool, recursive_flag: bool) -> Re
         // Step 18
         // TODO this step
 
-        match *internal_response.body.borrow() {
+        // Step 19
-            // Step 20
+        internal_response.wait_until_done();
-            ResponseBody::Empty => {
-                // Substep 1
-                // Substep 2
-            },
 
-            // Step 19
+        // Step 20
-            _ => {
+        // TODO this step
-                // Substep 1
-                // Substep 2
-            }
-        };
     }
 
-    // TODO remove this line when asynchronous fetches are supported
+    // TODO remove this line when only asynchronous fetches are used
     return response;
 }
 
@@ -544,11 +537,12 @@ fn http_redirect_fetch(request: Rc<Request>,
     let location = match response.get_actual_response().headers.get::<Location>() {
         Some(&Location(ref location)) => location.clone(),
         // Step 4
-        _ => return Response::network_error(),
+        _ => return Response::network_error()
     };
 
     // Step 5
-    let location_url = UrlParser::new().base_url(&request.current_url()).parse(&*location);
+    let response_url = response.get_actual_response().url.as_ref().unwrap();
+    let location_url = UrlParser::new().base_url(response_url).parse(&*location);
 
     // Step 6
     let location_url = match location_url {
@@ -663,29 +657,38 @@ fn http_network_or_cache_fetch(request: Rc<Request>,
         http_request.headers.borrow_mut().set(UserAgent(global_user_agent().to_owned()));
     }
 
-    // Step 9
+    match http_request.cache_mode.get() {
-    if http_request.cache_mode.get() == CacheMode::Default && is_no_store_cache(&http_request.headers.borrow()) {
+
-        http_request.cache_mode.set(CacheMode::NoStore);
+        // Step 9
+        CacheMode::Default if is_no_store_cache(&http_request.headers.borrow()) => {
+            http_request.cache_mode.set(CacheMode::NoStore);
+        },
+
+        // Step 10
+        CacheMode::NoCache if !http_request.headers.borrow().has::<CacheControl>() => {
+            http_request.headers.borrow_mut().set(CacheControl(vec![CacheDirective::MaxAge(0)]));
+        },
+
+        // Step 11
+        CacheMode::Reload => {
+            // Substep 1
+            if !http_request.headers.borrow().has::<Pragma>() {
+                http_request.headers.borrow_mut().set(Pragma::NoCache);
+            }
+
+            // Substep 2
+            if !http_request.headers.borrow().has::<CacheControl>() {
+                http_request.headers.borrow_mut().set(CacheControl(vec![CacheDirective::NoCache]));
+            }
+        },
+
+        _ => {}
     }
 
-    // Step 10
-    if http_request.cache_mode.get() == CacheMode::Reload {
-
-        // Substep 1
-        if !http_request.headers.borrow().has::<Pragma>() {
-            http_request.headers.borrow_mut().set(Pragma::NoCache);
-        }
-
-        // Substep 2
-        if !http_request.headers.borrow().has::<CacheControl>() {
-            http_request.headers.borrow_mut().set(CacheControl(vec![CacheDirective::NoCache]));
-        }
-    }
-
-    // Step 11
-    // modify_request_headers(http_request.headers.borrow());
-
     // Step 12
+    // modify_request_headers(http_request.headers.borrow());
+
+    // Step 13
     // TODO some of this step can't be implemented yet
     if credentials_flag {
 
@@ -723,13 +726,13 @@ fn http_network_or_cache_fetch(request: Rc<Request>,
         }
     }
 
-    // Step 13
-    // TODO this step can't be implemented
-
     // Step 14
-    let mut response: Option<Response> = None;
+    // TODO this step can't be implemented yet
 
     // Step 15
+    let mut response: Option<Response> = None;
+
+    // Step 16
     // TODO have a HTTP cache to check for a completed response
     let complete_http_response_from_cache: Option<Response> = None;
     if http_request.cache_mode.get() != CacheMode::NoStore &&
@@ -761,20 +764,20 @@ fn http_network_or_cache_fetch(request: Rc<Request>,
             // TODO this substep
         }
 
-    // Step 16
+    // Step 17
     // TODO have a HTTP cache to check for a partial response
     } else if http_request.cache_mode.get() == CacheMode::Default ||
         http_request.cache_mode.get() == CacheMode::ForceCache {
         // TODO this substep
     }
 
-    // Step 17
+    // Step 18
     if response.is_none() {
         response = Some(http_network_fetch(request.clone(), http_request.clone(), credentials_flag));
     }
     let response = response.unwrap();
 
-    // Step 18
+    // Step 19
     if let Some(status) = response.status {
         if status == StatusCode::NotModified &&
             (http_request.cache_mode.get() == CacheMode::Default ||
@@ -800,7 +803,7 @@ fn http_network_or_cache_fetch(request: Rc<Request>,
         }
     }
 
-    // Step 19
+    // Step 20
     response
 }
 
@@ -835,14 +838,43 @@ fn http_network_fetch(request: Rc<Request>,
     let mut response = Response::new();
     match wrapped_response {
         Ok(mut res) => {
-            // is it okay for res.version to be unused?
             response.url = Some(res.response.url.clone());
             response.status = Some(res.response.status);
             response.headers = res.response.headers.clone();
 
-            let mut body = vec![];
+            let res_body = response.body.clone();
-            res.response.read_to_end(&mut body);
+            thread::spawn(move || {
-            *response.body.borrow_mut() = ResponseBody::Done(body);
+
+                *res_body.lock().unwrap() = ResponseBody::Receiving(vec![]);
+                let mut new_body = vec![];
+                res.response.read_to_end(&mut new_body);
+
+                let mut body = res_body.lock().unwrap();
+                assert!(*body != ResponseBody::Empty);
+                *body = ResponseBody::Done(new_body);
+
+                // TODO: the vec storage format is much too slow for these operations,
+                // response.body needs to use something else before this code can be used
+                // *res_body.lock().unwrap() = ResponseBody::Receiving(vec![]);
+
+                // loop {
+                //     match read_block(&mut res.response) {
+                //         Ok(ReadResult::Payload(ref mut new_body)) => {
+                //             if let ResponseBody::Receiving(ref mut body) = *res_body.lock().unwrap() {
+                //                 (body).append(new_body);
+                //             }
+                //         },
+                //         Ok(ReadResult::EOF) | Err(_) => break
+                //     }
+
+                // }
+
+                // let mut completed_body = res_body.lock().unwrap();
+                // if let ResponseBody::Receiving(ref body) = *completed_body {
+                //     // TODO cloning seems sub-optimal, but I couldn't figure anything else out
+                //     *res_body.lock().unwrap() = ResponseBody::Done((*body).clone());
+                // }
+            });
         },
         Err(e) =>
             response.termination_reason = Some(TerminationReason::Fatal)

components/net/fetch/response.rs

@@ -9,6 +9,7 @@ use std::ascii::AsciiExt;
 use std::cell::{Cell, RefCell};
 use std::rc::Rc;
 use std::sync::mpsc::Receiver;
+use std::sync::{Arc, Mutex};
 use url::Url;
 
 pub trait ResponseMethods {
@@ -24,7 +25,7 @@ impl ResponseMethods for Response {
             url_list: RefCell::new(Vec::new()),
             status: Some(StatusCode::Ok),
             headers: Headers::new(),
-            body: RefCell::new(ResponseBody::Empty),
+            body: Arc::new(Mutex::new(ResponseBody::Empty)),
             cache_state: CacheState::None,
             https_state: HttpsState::None,
             internal_response: None,

components/net/http_loader.rs

@@ -101,12 +101,12 @@ pub fn factory(user_agent: String,
     }
 }
 
-enum ReadResult {
+pub enum ReadResult {
     Payload(Vec<u8>),
     EOF,
 }
 
-fn read_block<R: Read>(reader: &mut R) -> Result<ReadResult, ()> {
+pub fn read_block<R: Read>(reader: &mut R) -> Result<ReadResult, ()> {
     let mut buf = vec![0; 1024];
 
     match reader.read(&mut buf) {

components/net_traits/response.rs

@@ -8,6 +8,7 @@ use hyper::header::{AccessControlExposeHeaders, Headers};
 use hyper::status::StatusCode;
 use std::ascii::AsciiExt;
 use std::cell::{Cell, RefCell};
+use std::sync::{Arc, Mutex};
 use url::Url;
 
 /// [Response type](https://fetch.spec.whatwg.org/#concept-response-type)
@@ -31,7 +32,7 @@ pub enum TerminationReason {
 
 /// The response body can still be pushed to after fetch
 /// This provides a way to store unfinished response bodies
-#[derive(Clone, Debug)]
+#[derive(Clone, Debug, PartialEq)]
 pub enum ResponseBody {
     Empty, // XXXManishearth is this necessary, or is Done(vec![]) enough?
     Receiving(Vec<u8>),
@@ -81,7 +82,7 @@ pub struct Response {
     /// `None` can be considered a StatusCode of `0`.
     pub status: Option<StatusCode>,
     pub headers: Headers,
-    pub body: RefCell<ResponseBody>,
+    pub body: Arc<Mutex<ResponseBody>>,
     pub cache_state: CacheState,
     pub https_state: HttpsState,
     /// [Internal response](https://fetch.spec.whatwg.org/#concept-internal-response), only used if the Response
@@ -100,7 +101,7 @@ impl Response {
             url_list: RefCell::new(vec![]),
             status: None,
             headers: Headers::new(),
-            body: RefCell::new(ResponseBody::Empty),
+            body: Arc::new(Mutex::new(ResponseBody::Empty)),
             cache_state: CacheState::None,
             https_state: HttpsState::None,
             internal_response: None,
@@ -115,6 +116,11 @@ impl Response {
         }
     }
 
+    pub fn wait_until_done(&self) {
+        while !self.body.lock().unwrap().is_done() && !self.is_network_error() {
+        }
+    }
+
     pub fn get_actual_response(&self) -> &Response {
         if self.return_internal.get() && self.internal_response.is_some() {
             &**self.internal_response.as_ref().unwrap()
@@ -188,14 +194,14 @@ impl Response {
                 response.url = None;
                 response.headers = Headers::new();
                 response.status = None;
-                response.body = RefCell::new(ResponseBody::Empty);
+                response.body = Arc::new(Mutex::new(ResponseBody::Empty));
                 response.cache_state = CacheState::None;
             },
 
             ResponseType::OpaqueRedirect => {
                 response.headers = Headers::new();
                 response.status = None;
-                response.body = RefCell::new(ResponseBody::Empty);
+                response.body = Arc::new(Mutex::new(ResponseBody::Empty));
                 response.cache_state = CacheState::None;
             }
         }

tests/unit/net/fetch.rs

@@ -88,7 +88,7 @@ fn test_fetch_response_body_matches_const_message() {
     assert!(!fetch_response.is_network_error());
     assert_eq!(fetch_response.response_type, ResponseType::Basic);
 
-    match *fetch_response.body.borrow() {
+    match *fetch_response.body.lock().unwrap() {
         ResponseBody::Done(ref body) => {
             assert_eq!(&**body, MESSAGE);
         },
@@ -210,7 +210,7 @@ fn test_fetch_response_is_opaque_filtered() {
     // this also asserts that status message is "the empty byte sequence"
     assert!(fetch_response.status.is_none());
     assert_eq!(fetch_response.headers, Headers::new());
-    match fetch_response.body.into_inner() {
+    match *fetch_response.body.lock().unwrap() {
         ResponseBody::Empty => { },
         _ => panic!()
     }
@@ -260,7 +260,7 @@ fn test_fetch_response_is_opaque_redirect_filtered() {
     // this also asserts that status message is "the empty byte sequence"
     assert!(fetch_response.status.is_none());
     assert_eq!(fetch_response.headers, Headers::new());
-    match fetch_response.body.into_inner() {
+    match *fetch_response.body.lock().unwrap() {
         ResponseBody::Empty => { },
         _ => panic!()
     }
@@ -315,7 +315,7 @@ fn test_fetch_redirect_count_ceiling() {
     assert!(!fetch_response.is_network_error());
     assert_eq!(fetch_response.response_type, ResponseType::Basic);
 
-    match *fetch_response.body.borrow() {
+    match *fetch_response.body.lock().unwrap() {
         ResponseBody::Done(ref body) => {
             assert_eq!(&**body, MESSAGE);
         },
@@ -334,7 +334,7 @@ fn test_fetch_redirect_count_failure() {
 
     assert!(fetch_response.is_network_error());
 
-    match *fetch_response.body.borrow() {
+    match *fetch_response.body.lock().unwrap() {
         ResponseBody::Done(_) | ResponseBody::Receiving(_) => panic!(),
         _ => { }
     };
@@ -438,13 +438,15 @@ fn test_fetch_redirect_updates_method() {
 fn response_is_done(response: &Response) -> bool {
 
     let response_complete = match response.response_type {
-        ResponseType::Default | ResponseType::Basic | ResponseType::CORS => response.body.borrow().is_done(),
+        ResponseType::Default | ResponseType::Basic | ResponseType::CORS => {
+            (*response.body.lock().unwrap()).is_done()
+        }
         // if the internal response cannot have a body, it shouldn't block the "done" state
         ResponseType::Opaque | ResponseType::OpaqueRedirect | ResponseType::Error => true
     };
 
     let internal_complete = if let Some(ref res) = response.internal_response {
-        res.body.borrow().is_done()
+        res.body.lock().unwrap().is_done()
     } else {
         true
     };