diff options
Diffstat (limited to 'components/net')
| -rw-r--r-- | components/net/Cargo.toml | 4 | ||||
| -rw-r--r-- | components/net/http_cache.rs | 155 | ||||
| -rw-r--r-- | components/net/lib.rs | 5 | ||||
| -rw-r--r-- | components/net/resource_thread.rs | 88 | ||||
| -rw-r--r-- | components/net/tests/resource_thread.rs | 4 |
5 files changed, 181 insertions, 75 deletions
diff --git a/components/net/Cargo.toml b/components/net/Cargo.toml index b2dd5fd1134..0d69e0f3e02 100644 --- a/components/net/Cargo.toml +++ b/components/net/Cargo.toml @@ -24,6 +24,8 @@ immeta = "0.3.6" ipc-channel = "0.10" lazy_static = "1" log = "0.4" +malloc_size_of = { path = "../malloc_size_of" } +malloc_size_of_derive = { path = "../malloc_size_of_derive" } matches = "0.1" mime = "0.2.1" mime_guess = "1.8.0" @@ -33,6 +35,8 @@ openssl = "0.9" profile_traits = {path = "../profile_traits"} serde = "1.0" serde_json = "1.0" +servo_allocator = {path = "../allocator"} +servo_arc = {path = "../servo_arc"} servo_config = {path = "../config"} servo_url = {path = "../url"} servo-websocket = { version = "0.21", default-features = false, features = ["sync"] } diff --git a/components/net/http_cache.rs b/components/net/http_cache.rs index 12fac0ae852..77fa6897ecb 100644 --- a/components/net/http_cache.rs +++ b/components/net/http_cache.rs @@ -14,14 +14,17 @@ use hyper::header::Headers; use hyper::method::Method; use hyper::status::StatusCode; use hyper_serde::Serde; +use malloc_size_of::{MallocSizeOf, MallocSizeOfOps, MallocUnconditionalSizeOf, MallocUnconditionalShallowSizeOf}; +use malloc_size_of::Measurable; use net_traits::{Metadata, FetchMetadata}; use net_traits::request::Request; use net_traits::response::{HttpsState, Response, ResponseBody}; +use servo_arc::Arc; use servo_config::prefs::PREFS; use servo_url::ServoUrl; use std::collections::HashMap; use std::str; -use std::sync::{Arc, Mutex}; +use std::sync::Mutex; use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::mpsc::{channel, Sender}; use time; @@ -29,7 +32,7 @@ use time::{Duration, Tm}; /// The key used to differentiate requests in the cache. -#[derive(Clone, Eq, Hash, PartialEq)] +#[derive(Clone, Eq, Hash, MallocSizeOf, PartialEq )] pub struct CacheKey { url: ServoUrl } @@ -56,9 +59,16 @@ impl CacheKey { /// A complete cached resource. #[derive(Clone)] struct CachedResource { - metadata: CachedMetadata, request_headers: Arc<Mutex<Headers>>, body: Arc<Mutex<ResponseBody>>, + aborted: Arc<AtomicBool>, + awaiting_body: Arc<Mutex<Vec<Sender<Data>>>>, + data: Measurable<MeasurableCachedResource> +} + +#[derive(Clone, MallocSizeOf)] +struct MeasurableCachedResource { + metadata: CachedMetadata, location_url: Option<Result<ServoUrl, String>>, https_state: HttpsState, status: Option<StatusCode>, @@ -66,25 +76,47 @@ struct CachedResource { url_list: Vec<ServoUrl>, expires: Duration, last_validated: Tm, - aborted: Arc<AtomicBool>, - awaiting_body: Arc<Mutex<Vec<Sender<Data>>>> +} + +impl MallocSizeOf for CachedResource { + fn size_of(&self, ops: &mut MallocSizeOfOps) -> usize { + self.request_headers.unconditional_size_of(ops) + + self.body.unconditional_size_of(ops) + + self.aborted.unconditional_size_of(ops) + + self.awaiting_body.unconditional_size_of(ops) + + self.data.size_of(ops) + } } /// Metadata about a loaded resource, such as is obtained from HTTP headers. #[derive(Clone)] struct CachedMetadata { + /// Headers + pub headers: Arc<Mutex<Headers>>, + /// Fields that implement MallocSizeOf + pub data: Measurable<MeasurableCachedMetadata> +} + +#[derive(Clone, MallocSizeOf)] +struct MeasurableCachedMetadata { /// Final URL after redirects. pub final_url: ServoUrl, /// MIME type / subtype. pub content_type: Option<Serde<ContentType>>, /// Character set. pub charset: Option<String>, - /// Headers - pub headers: Arc<Mutex<Headers>>, /// HTTP Status pub status: Option<(u16, Vec<u8>)> } +impl MallocSizeOf for CachedMetadata { + fn size_of(&self, ops: &mut MallocSizeOfOps) -> usize { + self.headers.unconditional_shallow_size_of(ops) + + self.headers.size_of(ops) + + self.data.size_of(ops) + } +} + /// Wrapper around a cached response, including information on re-validation needs pub struct CachedResponse { /// The response constructed from the cached resource @@ -94,6 +126,7 @@ pub struct CachedResponse { } /// A memory cache. +#[derive(MallocSizeOf)] pub struct HttpCache { /// cached responses. entries: HashMap<CacheKey, Vec<CachedResource>>, @@ -278,7 +311,7 @@ fn create_cached_response(request: &Request, cached_headers: &Headers, done_chan: &mut DoneChannel) -> CachedResponse { - let mut response = Response::new(cached_resource.metadata.final_url.clone()); + let mut response = Response::new(cached_resource.data.metadata.data.final_url.clone()); response.headers = cached_headers.clone(); response.body = cached_resource.body.clone(); if let ResponseBody::Receiving(_) = *cached_resource.body.lock().unwrap() { @@ -286,18 +319,18 @@ fn create_cached_response(request: &Request, *done_chan = Some((done_sender.clone(), done_receiver)); cached_resource.awaiting_body.lock().unwrap().push(done_sender); } - response.location_url = cached_resource.location_url.clone(); - response.status = cached_resource.status.clone(); - response.raw_status = cached_resource.raw_status.clone(); - response.url_list = cached_resource.url_list.clone(); - response.https_state = cached_resource.https_state.clone(); + response.location_url = cached_resource.data.location_url.clone(); + response.status = cached_resource.data.status.clone(); + response.raw_status = cached_resource.data.raw_status.clone(); + response.url_list = cached_resource.data.url_list.clone(); + response.https_state = cached_resource.data.https_state.clone(); response.referrer = request.referrer.to_url().cloned(); response.referrer_policy = request.referrer_policy.clone(); response.aborted = cached_resource.aborted.clone(); - let expires = cached_resource.expires; + let expires = cached_resource.data.expires; let adjusted_expires = get_expiry_adjustment_from_request_headers(request, expires); let now = Duration::seconds(time::now().to_timespec().sec); - let last_validated = Duration::seconds(cached_resource.last_validated.to_timespec().sec); + let last_validated = Duration::seconds(cached_resource.data.last_validated.to_timespec().sec); let time_since_validated = now - last_validated; // TODO: take must-revalidate into account <https://tools.ietf.org/html/rfc7234#section-5.2.2.1> // TODO: if this cache is to be considered shared, take proxy-revalidate into account @@ -312,18 +345,20 @@ fn create_cached_response(request: &Request, fn create_resource_with_bytes_from_resource(bytes: &[u8], resource: &CachedResource) -> CachedResource { CachedResource { - metadata: resource.metadata.clone(), request_headers: resource.request_headers.clone(), body: Arc::new(Mutex::new(ResponseBody::Done(bytes.to_owned()))), - location_url: resource.location_url.clone(), - https_state: resource.https_state.clone(), - status: Some(StatusCode::PartialContent), - raw_status: Some((206, b"Partial Content".to_vec())), - url_list: resource.url_list.clone(), - expires: resource.expires.clone(), - last_validated: resource.last_validated.clone(), aborted: Arc::new(AtomicBool::new(false)), - awaiting_body: Arc::new(Mutex::new(vec![])) + awaiting_body: Arc::new(Mutex::new(vec![])), + data: Measurable(MeasurableCachedResource { + metadata: resource.data.metadata.clone(), + location_url: resource.data.location_url.clone(), + https_state: resource.data.https_state.clone(), + status: Some(StatusCode::PartialContent), + raw_status: Some((206, b"Partial Content".to_vec())), + url_list: resource.data.url_list.clone(), + expires: resource.data.expires.clone(), + last_validated: resource.data.last_validated.clone(), + }) } } @@ -334,13 +369,13 @@ fn handle_range_request(request: &Request, done_chan: &mut DoneChannel) -> Option<CachedResponse> { let mut complete_cached_resources = candidates.iter().filter(|resource| { - match resource.raw_status { + match resource.data.raw_status { Some((ref code, _)) => *code == 200, None => false } }); let partial_cached_resources = candidates.iter().filter(|resource| { - match resource.raw_status { + match resource.data.raw_status { Some((ref code, _)) => *code == 206, None => false } @@ -361,7 +396,7 @@ fn handle_range_request(request: &Request, let requested = body.get(b..e); if let Some(bytes) = requested { let new_resource = create_resource_with_bytes_from_resource(bytes, complete_resource); - let cached_headers = new_resource.metadata.headers.lock().unwrap(); + let cached_headers = new_resource.data.metadata.headers.lock().unwrap(); let cached_response = create_cached_response(request, &new_resource, &*cached_headers, done_chan); return Some(cached_response); } @@ -369,7 +404,7 @@ fn handle_range_request(request: &Request, }, (&header::ByteRangeSpec::FromTo(beginning, end), None) => { for partial_resource in partial_cached_resources { - let headers = partial_resource.metadata.headers.lock().unwrap(); + let headers = partial_resource.data.metadata.headers.lock().unwrap(); let content_range = headers.get::<header::ContentRange>(); let (res_beginning, res_end) = match content_range { Some(&header::ContentRange( @@ -401,7 +436,7 @@ fn handle_range_request(request: &Request, let requested = body.get(b..); if let Some(bytes) = requested { let new_resource = create_resource_with_bytes_from_resource(bytes, complete_resource); - let cached_headers = new_resource.metadata.headers.lock().unwrap(); + let cached_headers = new_resource.data.metadata.headers.lock().unwrap(); let cached_response = create_cached_response(request, &new_resource, &*cached_headers, done_chan); return Some(cached_response); } @@ -409,7 +444,7 @@ fn handle_range_request(request: &Request, }, (&header::ByteRangeSpec::AllFrom(beginning), None) => { for partial_resource in partial_cached_resources { - let headers = partial_resource.metadata.headers.lock().unwrap(); + let headers = partial_resource.data.metadata.headers.lock().unwrap(); let content_range = headers.get::<header::ContentRange>(); let (res_beginning, res_end, total) = match content_range { Some(&header::ContentRange( @@ -441,7 +476,7 @@ fn handle_range_request(request: &Request, let requested = body.get(from_byte..); if let Some(bytes) = requested { let new_resource = create_resource_with_bytes_from_resource(bytes, complete_resource); - let cached_headers = new_resource.metadata.headers.lock().unwrap(); + let cached_headers = new_resource.data.metadata.headers.lock().unwrap(); let cached_response = create_cached_response(request, &new_resource, &*cached_headers, done_chan); return Some(cached_response); } @@ -449,7 +484,7 @@ fn handle_range_request(request: &Request, }, (&header::ByteRangeSpec::Last(offset), None) => { for partial_resource in partial_cached_resources { - let headers = partial_resource.metadata.headers.lock().unwrap(); + let headers = partial_resource.data.metadata.headers.lock().unwrap(); let content_range = headers.get::<header::ContentRange>(); let (res_beginning, res_end, total) = match content_range { Some(&header::ContentRange( @@ -501,7 +536,7 @@ impl HttpCache { let mut candidates = vec![]; for cached_resource in resources { let mut can_be_constructed = true; - let cached_headers = cached_resource.metadata.headers.lock().unwrap(); + let cached_headers = cached_resource.data.metadata.headers.lock().unwrap(); let original_request_headers = cached_resource.request_headers.lock().unwrap(); if let Some(vary_data) = cached_headers.get_raw("Vary") { // Calculating Secondary Keys with Vary <https://tools.ietf.org/html/rfc7234#section-4.1> @@ -554,7 +589,7 @@ impl HttpCache { // Returning the first response that can be constructed // TODO: select the most appropriate one, using a known mechanism from a selecting header field, // or using the Date header to return the most recent one. - let cached_headers = cached_resource.metadata.headers.lock().unwrap(); + let cached_headers = cached_resource.data.metadata.headers.lock().unwrap(); let cached_response = create_cached_response(request, cached_resource, &*cached_headers, done_chan); return Some(cached_response); } @@ -589,24 +624,24 @@ impl HttpCache { let entry_key = CacheKey::new(request.clone()); if let Some(cached_resources) = self.entries.get_mut(&entry_key) { for cached_resource in cached_resources.iter_mut() { - let mut stored_headers = cached_resource.metadata.headers.lock().unwrap(); // Received a response with 304 status code, in response to a request that matches a cached resource. // 1. update the headers of the cached resource. // 2. return a response, constructed from the cached resource. - stored_headers.extend(response.headers.iter()); - let mut constructed_response = Response::new(cached_resource.metadata.final_url.clone()); - constructed_response.headers = stored_headers.clone(); + let mut constructed_response = Response::new(cached_resource.data.metadata.data.final_url.clone()); constructed_response.body = cached_resource.body.clone(); - constructed_response.status = cached_resource.status.clone(); - constructed_response.https_state = cached_resource.https_state.clone(); + constructed_response.status = cached_resource.data.status.clone(); + constructed_response.https_state = cached_resource.data.https_state.clone(); constructed_response.referrer = request.referrer.to_url().cloned(); constructed_response.referrer_policy = request.referrer_policy.clone(); - constructed_response.raw_status = cached_resource.raw_status.clone(); - constructed_response.url_list = cached_resource.url_list.clone(); + constructed_response.raw_status = cached_resource.data.raw_status.clone(); + constructed_response.url_list = cached_resource.data.url_list.clone(); // done_chan will have been set to Some by http_network_fetch, // set it back to None since the response returned here replaces the 304 one from the network. *done_chan = None; - cached_resource.expires = get_response_expiry(&constructed_response); + cached_resource.data.expires = get_response_expiry(&constructed_response); + let mut stored_headers = cached_resource.data.metadata.headers.lock().unwrap(); + stored_headers.extend(response.headers.iter()); + constructed_response.headers = stored_headers.clone(); return Some(constructed_response); } } @@ -617,7 +652,7 @@ impl HttpCache { let entry_key = CacheKey::from_servo_url(url); if let Some(cached_resources) = self.entries.get_mut(&entry_key) { for cached_resource in cached_resources.iter_mut() { - cached_resource.expires = Duration::seconds(0i64); + cached_resource.data.expires = Duration::seconds(0i64); } } } @@ -664,25 +699,29 @@ impl HttpCache { } let expiry = get_response_expiry(&response); let cacheable_metadata = CachedMetadata { - final_url: metadata.final_url, - content_type: metadata.content_type, - charset: metadata.charset, - status: metadata.status, - headers: Arc::new(Mutex::new(response.headers.clone())) + headers: Arc::new(Mutex::new(response.headers.clone())), + data: Measurable(MeasurableCachedMetadata { + final_url: metadata.final_url, + content_type: metadata.content_type, + charset: metadata.charset, + status: metadata.status + }) }; let entry_resource = CachedResource { - metadata: cacheable_metadata, request_headers: Arc::new(Mutex::new(request.headers.clone())), body: response.body.clone(), - location_url: response.location_url.clone(), - https_state: response.https_state.clone(), - status: response.status.clone(), - raw_status: response.raw_status.clone(), - url_list: response.url_list.clone(), - expires: expiry, - last_validated: time::now(), aborted: response.aborted.clone(), - awaiting_body: Arc::new(Mutex::new(vec![])) + awaiting_body: Arc::new(Mutex::new(vec![])), + data: Measurable(MeasurableCachedResource { + metadata: cacheable_metadata, + location_url: response.location_url.clone(), + https_state: response.https_state.clone(), + status: response.status.clone(), + raw_status: response.raw_status.clone(), + url_list: response.url_list.clone(), + expires: expiry, + last_validated: time::now() + }) }; let entry = self.entries.entry(entry_key).or_insert(vec![]); entry.push(entry_resource); diff --git a/components/net/lib.rs b/components/net/lib.rs index 666e7bdb2d5..d5dfbf74c4f 100644 --- a/components/net/lib.rs +++ b/components/net/lib.rs @@ -17,6 +17,8 @@ extern crate ipc_channel; #[macro_use] extern crate lazy_static; #[macro_use] extern crate log; +extern crate malloc_size_of; +#[macro_use] extern crate malloc_size_of_derive; #[macro_use] #[no_link] extern crate matches; #[macro_use] extern crate mime; @@ -24,9 +26,12 @@ extern crate mime_guess; extern crate msg; extern crate net_traits; extern crate openssl; +#[macro_use] extern crate profile_traits; #[macro_use] extern crate serde; extern crate serde_json; +extern crate servo_allocator; +extern crate servo_arc; extern crate servo_config; extern crate servo_url; extern crate time; diff --git a/components/net/resource_thread.rs b/components/net/resource_thread.rs index c319db98505..8060b811bed 100644 --- a/components/net/resource_thread.rs +++ b/components/net/resource_thread.rs @@ -16,6 +16,7 @@ use http_cache::HttpCache; use http_loader::{HttpState, http_redirect_fetch}; use hyper_serde::Serde; use ipc_channel::ipc::{self, IpcReceiver, IpcReceiverSet, IpcSender}; +use malloc_size_of::{MallocSizeOf, MallocSizeOfOps}; use net_traits::{CookieSource, CoreResourceThread}; use net_traits::{CoreResourceMsg, CustomResponseMediator, FetchChannels}; use net_traits::{FetchResponseMsg, ResourceThreads, WebSocketDomAction}; @@ -23,9 +24,12 @@ use net_traits::WebSocketNetworkEvent; use net_traits::request::{Request, RequestInit}; use net_traits::response::{Response, ResponseInit}; use net_traits::storage_thread::StorageThreadMsg; +use profile_traits::mem::{Report, ReportsChan, ReportKind}; +use profile_traits::mem::ProfilerChan as MemProfilerChan; use profile_traits::time::ProfilerChan; use serde::{Deserialize, Serialize}; use serde_json; +use servo_allocator; use servo_config::opts; use servo_config::resource_files::resources_dir_path; use servo_url::ServoUrl; @@ -47,13 +51,15 @@ const TFD_PROVIDER: &'static TFDProvider = &TFDProvider; /// Returns a tuple of (public, private) senders to the new threads. pub fn new_resource_threads(user_agent: Cow<'static, str>, devtools_chan: Option<Sender<DevtoolsControlMsg>>, - profiler_chan: ProfilerChan, + time_profiler_chan: ProfilerChan, + mem_profiler_chan: MemProfilerChan, config_dir: Option<PathBuf>) -> (ResourceThreads, ResourceThreads) { let (public_core, private_core) = new_core_resource_thread( user_agent, devtools_chan, - profiler_chan, + time_profiler_chan, + mem_profiler_chan, config_dir.clone()); let storage: IpcSender<StorageThreadMsg> = StorageThreadFactory::new(config_dir); (ResourceThreads::new(public_core, storage.clone()), @@ -64,22 +70,34 @@ pub fn new_resource_threads(user_agent: Cow<'static, str>, /// Create a CoreResourceThread pub fn new_core_resource_thread(user_agent: Cow<'static, str>, devtools_chan: Option<Sender<DevtoolsControlMsg>>, - profiler_chan: ProfilerChan, + time_profiler_chan: ProfilerChan, + mem_profiler_chan: MemProfilerChan, config_dir: Option<PathBuf>) -> (CoreResourceThread, CoreResourceThread) { let (public_setup_chan, public_setup_port) = ipc::channel().unwrap(); let (private_setup_chan, private_setup_port) = ipc::channel().unwrap(); + let (report_chan, report_port) = ipc::channel().unwrap(); + thread::Builder::new().name("ResourceManager".to_owned()).spawn(move || { let resource_manager = CoreResourceManager::new( - user_agent, devtools_chan, profiler_chan + user_agent, devtools_chan, time_profiler_chan ); let mut channel_manager = ResourceChannelManager { resource_manager: resource_manager, config_dir: config_dir, }; - channel_manager.start(public_setup_port, - private_setup_port); + + mem_profiler_chan.run_with_memory_reporting(|| ( + channel_manager.start( + public_setup_port, + private_setup_port, + report_port) + ), + String::from("network-cache-reporter"), + report_chan, + |report_chan| report_chan); + }).expect("Thread spawning failed"); (public_setup_chan, private_setup_chan) } @@ -127,31 +145,69 @@ impl ResourceChannelManager { #[allow(unsafe_code)] fn start(&mut self, public_receiver: IpcReceiver<CoreResourceMsg>, - private_receiver: IpcReceiver<CoreResourceMsg>) { + private_receiver: IpcReceiver<CoreResourceMsg>, + memory_reporter: IpcReceiver<ReportsChan>) { let (public_http_state, private_http_state) = create_http_states(self.config_dir.as_ref().map(Deref::deref)); let mut rx_set = IpcReceiverSet::new().unwrap(); let private_id = rx_set.add(private_receiver).unwrap(); let public_id = rx_set.add(public_receiver).unwrap(); + let reporter_id = rx_set.add(memory_reporter).unwrap(); loop { - for (id, data) in rx_set.select().unwrap().into_iter().map(|m| m.unwrap()) { - let group = if id == private_id { - &private_http_state + for receiver in rx_set.select().unwrap().into_iter() { + // Handles case where profiler thread shuts down before resource thread. + match receiver { + ipc::IpcSelectionResult::ChannelClosed(..) => continue, + _ => {} + } + let (id, data) = receiver.unwrap(); + // If message is memory report, get the size_of of public and private http caches + if id == reporter_id { + if let Ok(msg) = data.to() { + self.process_report(msg, &private_http_state, &public_http_state); + continue; + } } else { - assert_eq!(id, public_id); - &public_http_state - }; - if let Ok(msg) = data.to() { - if !self.process_msg(msg, group) { - return; + let group = if id == private_id { + &private_http_state + } else { + assert_eq!(id, public_id); + &public_http_state + }; + if let Ok(msg) = data.to() { + if !self.process_msg(msg, group) { + return; + } } } } } } + fn process_report(&mut self, + msg: ReportsChan, + public_http_state: &Arc<HttpState>, + private_http_state: &Arc<HttpState>) { + let mut ops = MallocSizeOfOps::new(servo_allocator::usable_size, None, None); + let public_cache = public_http_state.http_cache.read().unwrap(); + let private_cache = private_http_state.http_cache.read().unwrap(); + + let public_report = Report { + path: path!["memory-cache", "public"], + kind: ReportKind::ExplicitJemallocHeapSize, + size: public_cache.size_of(&mut ops) + }; + + let private_report = Report { + path: path!["memory-cache", "private"], + kind: ReportKind::ExplicitJemallocHeapSize, + size: private_cache.size_of(&mut ops) + }; + + msg.send(vec!(public_report, private_report)); + } /// Returns false if the thread should exit. fn process_msg(&mut self, diff --git a/components/net/tests/resource_thread.rs b/components/net/tests/resource_thread.rs index 8ad3081c224..adf1e305698 100644 --- a/components/net/tests/resource_thread.rs +++ b/components/net/tests/resource_thread.rs @@ -6,6 +6,7 @@ use ipc_channel::ipc; use net::resource_thread::new_core_resource_thread; use net::test::parse_hostsfile; use net_traits::CoreResourceMsg; +use profile_traits::mem::ProfilerChan as MemProfilerChan; use profile_traits::time::ProfilerChan; use std::net::IpAddr; @@ -16,9 +17,10 @@ fn ip(s: &str) -> IpAddr { #[test] fn test_exit() { let (tx, _rx) = ipc::channel().unwrap(); + let (mtx, _mrx) = ipc::channel().unwrap(); let (sender, receiver) = ipc::channel().unwrap(); let (resource_thread, _private_resource_thread) = new_core_resource_thread( - "".into(), None, ProfilerChan(tx), None); + "".into(), None, ProfilerChan(tx), MemProfilerChan(mtx), None); resource_thread.send(CoreResourceMsg::Exit(sender)).unwrap(); receiver.recv().unwrap(); } |