Auto merge of #19274 - Manishearth:xhr-cancel, r=jdm

Fetch cancellation

This PR implements cancellation for fetch, and uses it for XHR. This means that fetch clients can now send a message to the fetch task asking for the network request to be aborted.

Previously, clients like XHR had abort functionality but would implement it by simply ignoring future messages from the network task; and would not actually cancel the network fetch.

<!-- Reviewable:start -->
---
This change is [<img src="https://reviewable.io/review_button.svg" height="34" align="absmiddle" alt="Reviewable"/>](https://reviewable.io/reviews/servo/servo/19274)
<!-- Reviewable:end -->

components/constellation/network_listener.rs

@@ -57,14 +57,14 @@ impl NetworkListener {
             Some(ref res_init_) => CoreResourceMsg::FetchRedirect(
                                    self.req_init.clone(),
                                    res_init_.clone(),
-                                   ipc_sender),
+                                   ipc_sender, None),
             None => {
                 set_default_accept(Destination::Document, &mut listener.req_init.headers);
                 set_default_accept_language(&mut listener.req_init.headers);
 
                 CoreResourceMsg::Fetch(
                 listener.req_init.clone(),
-                FetchChannels::ResponseMsg(ipc_sender))
+                FetchChannels::ResponseMsg(ipc_sender, None))
             }
         };
 

components/net/fetch/methods.rs

@@ -15,6 +15,7 @@ use hyper::header::{Header, HeaderFormat, HeaderView, Headers, Referer as Refere
 use hyper::method::Method;
 use hyper::mime::{Mime, SubLevel, TopLevel};
 use hyper::status::StatusCode;
+use ipc_channel::ipc::IpcReceiver;
 use mime_guess::guess_mime_type;
 use net_traits::{FetchTaskTarget, NetworkError, ReferrerPolicy};
 use net_traits::request::{CredentialsMode, Destination, Referrer, Request, RequestMode};
@@ -27,7 +28,7 @@ use std::fs::File;
 use std::io::Read;
 use std::mem;
 use std::str;
-use std::sync::Arc;
+use std::sync::{Arc, Mutex};
 use std::sync::mpsc::{Sender, Receiver};
 use subresource_integrity::is_response_integrity_valid;
 
@@ -36,6 +37,7 @@ pub type Target<'a> = &'a mut (FetchTaskTarget + Send);
 pub enum Data {
     Payload(Vec<u8>),
     Done,
+    Cancelled,
 }
 
 pub struct FetchContext {
@@ -43,8 +45,37 @@ pub struct FetchContext {
     pub user_agent: Cow<'static, str>,
     pub devtools_chan: Option<Sender<DevtoolsControlMsg>>,
     pub filemanager: FileManager,
+    pub cancellation_listener: Arc<Mutex<CancellationListener>>,
 }
 
+pub struct CancellationListener {
+    cancel_chan: Option<IpcReceiver<()>>,
+    cancelled: bool,
+}
+
+impl CancellationListener {
+    pub fn new(cancel_chan: Option<IpcReceiver<()>>) -> Self {
+        Self {
+            cancel_chan: cancel_chan,
+            cancelled: false,
+        }
+    }
+
+    pub fn cancelled(&mut self) -> bool {
+        if let Some(ref cancel_chan) = self.cancel_chan {
+            if self.cancelled {
+                true
+            } else if cancel_chan.try_recv().is_ok() {
+                self.cancelled = true;
+                true
+            } else {
+                false
+            }
+        } else {
+            false
+        }
+    }
+}
 pub type DoneChannel = Option<(Sender<Data>, Receiver<Data>)>;
 
 /// [Fetch](https://fetch.spec.whatwg.org#concept-fetch)
@@ -317,7 +348,7 @@ pub fn main_fetch(request: &mut Request,
     };
 
     // Execute deferred rebinding of response.
-    let response = if let Some(error) = internal_error {
+    let mut response = if let Some(error) = internal_error {
         Response::network_error(error)
     } else {
         response
@@ -325,9 +356,9 @@ pub fn main_fetch(request: &mut Request,
 
     // Step 19.
     let mut response_loaded = false;
-    let response = if !response.is_network_error() && !request.integrity_metadata.is_empty() {
+    let mut response = if !response.is_network_error() && !request.integrity_metadata.is_empty() {
         // Step 19.1.
-        wait_for_response(&response, target, done_chan);
+        wait_for_response(&mut response, target, done_chan);
         response_loaded = true;
 
         // Step 19.2.
@@ -346,9 +377,9 @@ pub fn main_fetch(request: &mut Request,
     if request.synchronous {
         // process_response is not supposed to be used
         // by sync fetch, but we overload it here for simplicity
-        target.process_response(&response);
+        target.process_response(&mut response);
         if !response_loaded {
-            wait_for_response(&response, target, done_chan);
+            wait_for_response(&mut response, target, done_chan);
         }
         // overloaded similarly to process_response
         target.process_response_eof(&response);
@@ -370,7 +401,7 @@ pub fn main_fetch(request: &mut Request,
 
     // Step 23.
     if !response_loaded {
-       wait_for_response(&response, target, done_chan);
+       wait_for_response(&mut response, target, done_chan);
     }
 
     // Step 24.
@@ -381,7 +412,7 @@ pub fn main_fetch(request: &mut Request,
     response
 }
 
-fn wait_for_response(response: &Response, target: Target, done_chan: &mut DoneChannel) {
+fn wait_for_response(response: &mut Response, target: Target, done_chan: &mut DoneChannel) {
     if let Some(ref ch) = *done_chan {
         loop {
             match ch.1.recv()
@@ -390,6 +421,10 @@ fn wait_for_response(response: &Response, target: Target, done_chan: &mut DoneCh
                     target.process_response_chunk(vec);
                 },
                 Data::Done => break,
+                Data::Cancelled => {
+                    response.aborted = true;
+                    break;
+                }
             }
         }
     } else {

components/net/http_loader.rs

@@ -1103,6 +1103,10 @@ fn http_network_fetch(request: &Request,
     let devtools_sender = context.devtools_chan.clone();
     let meta_status = meta.status.clone();
     let meta_headers = meta.headers.clone();
+    let cancellation_listener = context.cancellation_listener.clone();
+    if cancellation_listener.lock().unwrap().cancelled() {
+        return Response::network_error(NetworkError::Internal("Fetch aborted".into()))
+    }
     thread::Builder::new().name(format!("fetch worker thread")).spawn(move || {
         match StreamedResponse::from_http_response(res) {
             Ok(mut res) => {
@@ -1125,6 +1129,11 @@ fn http_network_fetch(request: &Request,
                 }
 
                 loop {
+                    if cancellation_listener.lock().unwrap().cancelled() {
+                        *res_body.lock().unwrap() = ResponseBody::Done(vec![]);
+                        let _ = done_sender.send(Data::Cancelled);
+                        return;
+                    }
                     match read_block(&mut res) {
                         Ok(Data::Payload(chunk)) => {
                             if let ResponseBody::Receiving(ref mut body) = *res_body.lock().unwrap() {
@@ -1144,6 +1153,7 @@ fn http_network_fetch(request: &Request,
                             let _ = done_sender.send(Data::Done);
                             break;
                         }
+                        Ok(Data::Cancelled) => unreachable!() // read_block doesn't return Data::Cancelled
                     }
                 }
             }

components/net/resource_thread.rs

@@ -9,7 +9,7 @@ use cookie_rs;
 use cookie_storage::CookieStorage;
 use devtools_traits::DevtoolsControlMsg;
 use fetch::cors_cache::CorsCache;
-use fetch::methods::{FetchContext, fetch};
+use fetch::methods::{CancellationListener, FetchContext, fetch};
 use filemanager_thread::{FileManager, TFDProvider};
 use hsts::HstsList;
 use http_cache::HttpCache;
@@ -36,7 +36,7 @@ use std::fs::File;
 use std::io::prelude::*;
 use std::ops::Deref;
 use std::path::{Path, PathBuf};
-use std::sync::{Arc, RwLock};
+use std::sync::{Arc, Mutex, RwLock};
 use std::sync::mpsc::Sender;
 use std::thread;
 use storage_thread::StorageThreadFactory;
@@ -160,14 +160,14 @@ impl ResourceChannelManager {
         match msg {
             CoreResourceMsg::Fetch(req_init, channels) => {
                 match channels {
-                    FetchChannels::ResponseMsg(sender) =>
+                    FetchChannels::ResponseMsg(sender, cancel_chan) =>
-                        self.resource_manager.fetch(req_init, None, sender, http_state),
+                        self.resource_manager.fetch(req_init, None, sender, http_state, cancel_chan),
                     FetchChannels::WebSocket { event_sender, action_receiver } =>
                         self.resource_manager.websocket_connect(req_init, event_sender, action_receiver, http_state),
                 }
             }
-            CoreResourceMsg::FetchRedirect(req_init, res_init, sender) =>
+            CoreResourceMsg::FetchRedirect(req_init, res_init, sender, cancel_chan) =>
-                self.resource_manager.fetch(req_init, Some(res_init), sender, http_state),
+                self.resource_manager.fetch(req_init, Some(res_init), sender, http_state, cancel_chan),
             CoreResourceMsg::SetCookieForUrl(request, cookie, source) =>
                 self.resource_manager.set_cookie_for_url(&request, cookie.into_inner(), source, http_state),
             CoreResourceMsg::SetCookiesForUrl(request, cookies, source) => {
@@ -332,7 +332,8 @@ impl CoreResourceManager {
              req_init: RequestInit,
              res_init_: Option<ResponseInit>,
              mut sender: IpcSender<FetchResponseMsg>,
-             http_state: &Arc<HttpState>) {
+             http_state: &Arc<HttpState>,
+             cancel_chan: Option<IpcReceiver<()>>) {
         let http_state = http_state.clone();
         let ua = self.user_agent.clone();
         let dc = self.devtools_chan.clone();
@@ -349,6 +350,7 @@ impl CoreResourceManager {
                 user_agent: ua,
                 devtools_chan: dc,
                 filemanager: filemanager,
+                cancellation_listener: Arc::new(Mutex::new(CancellationListener::new(cancel_chan))),
             };
 
             match res_init_ {

components/net_traits/lib.rs

@@ -342,7 +342,7 @@ pub enum WebSocketNetworkEvent {
 #[derive(Deserialize, Serialize)]
 /// IPC channels to communicate with the script thread about network or DOM events.
 pub enum FetchChannels {
-    ResponseMsg(IpcSender<FetchResponseMsg>),
+    ResponseMsg(IpcSender<FetchResponseMsg>, /* cancel_chan */ Option<IpcReceiver<()>>),
     WebSocket {
         event_sender: IpcSender<WebSocketNetworkEvent>,
         action_receiver: IpcReceiver<WebSocketDomAction>,
@@ -353,7 +353,7 @@ pub enum FetchChannels {
 pub enum CoreResourceMsg {
     Fetch(RequestInit, FetchChannels),
     /// Initiate a fetch in response to processing a redirection
-    FetchRedirect(RequestInit, ResponseInit, IpcSender<FetchResponseMsg>),
+    FetchRedirect(RequestInit, ResponseInit, IpcSender<FetchResponseMsg>, /* cancel_chan */ Option<IpcReceiver<()>>),
     /// Store a cookie for a given originating URL
     SetCookieForUrl(ServoUrl, Serde<Cookie<'static>>, CookieSource),
     /// Store a set of cookies for a given originating URL
@@ -383,7 +383,7 @@ pub fn fetch_async<F>(request: RequestInit, core_resource_thread: &CoreResourceT
     ROUTER.add_route(action_receiver.to_opaque(),
                      Box::new(move |message| f(message.to().unwrap())));
     core_resource_thread.send(
-        CoreResourceMsg::Fetch(request, FetchChannels::ResponseMsg(action_sender))).unwrap();
+        CoreResourceMsg::Fetch(request, FetchChannels::ResponseMsg(action_sender, None))).unwrap();
 }
 
 #[derive(Clone, Deserialize, MallocSizeOf, Serialize)]
@@ -478,7 +478,7 @@ pub fn load_whole_resource(request: RequestInit,
                            -> Result<(Metadata, Vec<u8>), NetworkError> {
     let (action_sender, action_receiver) = ipc::channel().unwrap();
     core_resource_thread.send(
-        CoreResourceMsg::Fetch(request, FetchChannels::ResponseMsg(action_sender))).unwrap();
+        CoreResourceMsg::Fetch(request, FetchChannels::ResponseMsg(action_sender, None))).unwrap();
 
     let mut buf = vec![];
     let mut metadata = None;

components/net_traits/response.rs

@@ -112,6 +112,8 @@ pub struct Response {
     pub internal_response: Option<Box<Response>>,
     /// whether or not to try to return the internal_response when asked for actual_response
     pub return_internal: bool,
+    /// https://fetch.spec.whatwg.org/#concept-response-aborted
+    pub aborted: bool,
 }
 
 impl Response {
@@ -133,6 +135,7 @@ impl Response {
             location_url: None,
             internal_response: None,
             return_internal: true,
+            aborted: false,
         }
     }
 
@@ -162,6 +165,7 @@ impl Response {
             location_url: None,
             internal_response: None,
             return_internal: true,
+            aborted: false,
         }
     }
 

components/script/document_loader.rs

@@ -126,7 +126,7 @@ impl DocumentLoader {
                                   request: RequestInit,
                                   fetch_target: IpcSender<FetchResponseMsg>) {
         self.resource_threads.sender().send(
-            CoreResourceMsg::Fetch(request, FetchChannels::ResponseMsg(fetch_target))).unwrap();
+            CoreResourceMsg::Fetch(request, FetchChannels::ResponseMsg(fetch_target, None))).unwrap();
     }
 
     /// Mark an in-progress network request complete.

components/script/dom/eventsource.rs

@@ -491,7 +491,7 @@ impl EventSource {
             listener.notify_fetch(message.to().unwrap());
         }));
         global.core_resource_thread().send(
-            CoreResourceMsg::Fetch(request, FetchChannels::ResponseMsg(action_sender))).unwrap();
+            CoreResourceMsg::Fetch(request, FetchChannels::ResponseMsg(action_sender, None))).unwrap();
         // Step 13
         Ok(ev)
     }
@@ -555,6 +555,6 @@ impl EventSourceTimeoutCallback {
         }
         // Step 5.4
         global.core_resource_thread().send(
-            CoreResourceMsg::Fetch(request, FetchChannels::ResponseMsg(self.action_sender))).unwrap();
+            CoreResourceMsg::Fetch(request, FetchChannels::ResponseMsg(self.action_sender, None))).unwrap();
     }
 }

components/script/dom/xmlhttprequest.rs

@@ -154,6 +154,8 @@ pub struct XMLHttpRequest {
     response_status: Cell<Result<(), ()>>,
     referrer_url: Option<ServoUrl>,
     referrer_policy: Option<ReferrerPolicy>,
+    #[ignore_malloc_size_of = "channels are hard"]
+    cancellation_chan: DomRefCell<Option<ipc::IpcSender<()>>>,
 }
 
 impl XMLHttpRequest {
@@ -198,6 +200,7 @@ impl XMLHttpRequest {
             response_status: Cell::new(Ok(())),
             referrer_url: referrer_url,
             referrer_policy: referrer_policy,
+            cancellation_chan: DomRefCell::new(None),
         }
     }
     pub fn new(global: &GlobalScope) -> DomRoot<XMLHttpRequest> {
@@ -218,7 +221,8 @@ impl XMLHttpRequest {
     fn initiate_async_xhr(context: Arc<Mutex<XHRContext>>,
                           task_source: NetworkingTaskSource,
                           global: &GlobalScope,
-                          init: RequestInit) {
+                          init: RequestInit,
+                          cancellation_chan: ipc::IpcReceiver<()>) {
         impl FetchResponseListener for XHRContext {
             fn process_request_body(&mut self) {
                 // todo
@@ -255,6 +259,7 @@ impl XMLHttpRequest {
         }
 
         let (action_sender, action_receiver) = ipc::channel().unwrap();
+
         let listener = NetworkListener {
             context: context,
             task_source: task_source,
@@ -264,7 +269,7 @@ impl XMLHttpRequest {
             listener.notify_fetch(message.to().unwrap());
         }));
         global.core_resource_thread().send(
-            Fetch(init, FetchChannels::ResponseMsg(action_sender))).unwrap();
+            Fetch(init, FetchChannels::ResponseMsg(action_sender, Some(cancellation_chan)))).unwrap();
     }
 }
 
@@ -1018,6 +1023,12 @@ impl XMLHttpRequest {
     }
 
     fn terminate_ongoing_fetch(&self) {
+        if let Some(ref cancel_chan) = *self.cancellation_chan.borrow() {
+            // The receiver will be destroyed if the request has already completed;
+            // so we throw away the error. Cancellation is a courtesy call,
+            // we don't actually care if the other side heard.
+            let _ = cancel_chan.send(());
+        }
         let GenerationId(prev_id) = self.generation_id.get();
         self.generation_id.set(GenerationId(prev_id + 1));
         self.response_status.set(Ok(()));
@@ -1311,8 +1322,11 @@ impl XMLHttpRequest {
             (global.networking_task_source(), None)
         };
 
+        let (cancel_sender, cancel_receiver) = ipc::channel().unwrap();
+        *self.cancellation_chan.borrow_mut() = Some(cancel_sender);
+
         XMLHttpRequest::initiate_async_xhr(context.clone(), task_source,
-                                           global, init);
+                                           global, init, cancel_receiver);
 
         if let Some(script_port) = script_port {
             loop {

components/script/fetch.rs

@@ -109,7 +109,7 @@ pub fn Fetch(global: &GlobalScope, input: RequestInfo, init: RootedTraceableBox<
         listener.notify_fetch(message.to().unwrap());
     }));
     core_resource_thread.send(
-        NetTraitsFetch(request_init, FetchChannels::ResponseMsg(action_sender))).unwrap();
+        NetTraitsFetch(request_init, FetchChannels::ResponseMsg(action_sender, None))).unwrap();
 
     promise
 }

tests/unit/net/fetch.rs

@@ -25,7 +25,7 @@ use hyper_openssl;
 use msg::constellation_msg::TEST_PIPELINE_ID;
 use net::connector::create_ssl_client;
 use net::fetch::cors_cache::CorsCache;
-use net::fetch::methods::FetchContext;
+use net::fetch::methods::{CancellationListener, FetchContext};
 use net::filemanager_thread::FileManager;
 use net::hsts::HstsEntry;
 use net::test::HttpState;
@@ -538,6 +538,7 @@ fn test_fetch_with_hsts() {
         user_agent: DEFAULT_USER_AGENT.into(),
         devtools_chan: None,
         filemanager: FileManager::new(),
+        cancellation_listener: Arc::new(Mutex::new(CancellationListener::new(None))),
     };
 
     {

tests/unit/net/lib.rs

@@ -36,7 +36,7 @@ use devtools_traits::DevtoolsControlMsg;
 use hyper::server::{Handler, Listening, Server};
 use net::connector::create_ssl_client;
 use net::fetch::cors_cache::CorsCache;
-use net::fetch::methods::{self, FetchContext};
+use net::fetch::methods::{self, CancellationListener, FetchContext};
 use net::filemanager_thread::FileManager;
 use net::test::HttpState;
 use net_traits::FetchTaskTarget;
@@ -44,7 +44,7 @@ use net_traits::request::Request;
 use net_traits::response::Response;
 use servo_config::resource_files::resources_dir_path;
 use servo_url::ServoUrl;
-use std::sync::Arc;
+use std::sync::{Arc, Mutex};
 use std::sync::mpsc::{Sender, channel};
 
 const DEFAULT_USER_AGENT: &'static str = "Such Browser. Very Layout. Wow.";
@@ -61,6 +61,7 @@ fn new_fetch_context(dc: Option<Sender<DevtoolsControlMsg>>) -> FetchContext {
         user_agent: DEFAULT_USER_AGENT.into(),
         devtools_chan: dc,
         filemanager: FileManager::new(),
+        cancellation_listener: Arc::new(Mutex::new(CancellationListener::new(None))),
     }
 }
 impl FetchTaskTarget for FetchResponseCollector {