Magic-Mirror/gecko-dev
1
0
Fork 0
mirror of https://github.com/mozilla/gecko-dev.git synced 2024-10-21 09:15:35 +00:00
Code Issues Actions Packages Projects Releases Wiki Activity

servo: Merge #9850 - Set response.body Asynchronously In Fetch (from nikkisquared:2_async_2_furious); r=jdm

Browse Source 
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.

Source-Repo: https://github.com/servo/servo
Source-Revision: fee7cb179ee7ba2f159d87af07afaf0cd99a2161
This commit is contained in:
Nikki 2016-03-08 04:16:25 +05:01
parent 90e2b2cb80
commit f2cb1a2b2f
5 changed files with 114 additions and 73 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

148
servo/components/net/fetch/methods.rs
View File

@ -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);
listener.response_available(res);
let fetch_response = fetch(request);
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 {
// TODO run the remaining steps in parallel
}
// this step is obsoleted by fetch_async
// 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,
// we will need to make sure nothing tries to write to the body at this point
*internal_response.body.borrow_mut() = ResponseBody::Empty;
// when Fetch is used only asynchronously, we will need to make sure
// that nothing tries to write to the body at this point
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 20
ResponseBody::Empty => {
// Substep 1
// Substep 2
},
// Step 19
internal_response.wait_until_done();
// Step 19
_ => {
// Substep 1
// Substep 2
}
};
// Step 20
// TODO this step
}
// 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
if http_request.cache_mode.get() == CacheMode::Default && is_no_store_cache(&http_request.headers.borrow()) {
http_request.cache_mode.set(CacheMode::NoStore);
match http_request.cache_mode.get() {
// 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![];
res.response.read_to_end(&mut body);
*response.body.borrow_mut() = ResponseBody::Done(body);
let res_body = response.body.clone();
thread::spawn(move || {
*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)

3
servo/components/net/fetch/response.rs
View File

@ -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,

4
servo/components/net/http_loader.rs
View File

@ -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) {

16
servo/components/net_traits/response.rs
View File

@ -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;
}
}

16
servo/tests/unit/net/fetch.rs
View File

@ -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
};