Auto merge of #11497 - ab22:11467-resource-threads-race-with-shutdown-to-write-out-data, r=Ms2ger

send a reply when thread is done exiting

<!-- Please describe your changes on the following line: -->

---
<!-- Thank you for contributing to Servo! Please replace each `[ ]` by `[X]` when the step is complete, and replace `__` with appropriate data: -->
- [x] `./mach build -d` does not report any errors
- [x] `./mach test-tidy` does not report any errors
- [x] These changes fix #11467

<!-- Either: -->
- [x] There are tests for these changes OR
- [ ] These changes do not require tests because: in this case if code compiles then it's good enough.

<!-- Pull requests that do not address these steps are welcome, but they will require additional verification as part of the review process. -->

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

components/constellation/constellation.rs

@@ -817,11 +817,15 @@ impl<Message, LTF, STF> Constellation<Message, LTF, STF>
     }
 
     fn handle_exit(&mut self) {
+        // Channels to recieve signals when threads are done exiting.
+        let (core_sender, core_receiver) = ipc::channel().expect("Failed to create IPC channel!");
+        let (storage_sender, storage_receiver) = ipc::channel().expect("Failed to create IPC channel!");
+
         for (_id, ref pipeline) in &self.pipelines {
             pipeline.exit();
         }
         self.image_cache_thread.exit();
-        if let Err(e) = self.resource_threads.send(net_traits::CoreResourceMsg::Exit) {
+        if let Err(e) = self.resource_threads.send(net_traits::CoreResourceMsg::Exit(core_sender)) {
             warn!("Exit resource thread failed ({})", e);
         }
         if let Some(ref chan) = self.devtools_chan {
@@ -830,7 +834,7 @@ impl<Message, LTF, STF> Constellation<Message, LTF, STF>
                 warn!("Exit devtools failed ({})", e);
             }
         }
-        if let Err(e) = self.resource_threads.send(StorageThreadMsg::Exit) {
+        if let Err(e) = self.resource_threads.send(StorageThreadMsg::Exit(storage_sender)) {
             warn!("Exit storage thread failed ({})", e);
         }
 
@@ -842,6 +846,15 @@ impl<Message, LTF, STF> Constellation<Message, LTF, STF>
             warn!("Exit bluetooth thread failed ({})", e);
         }
         self.font_cache_thread.exit();
+
+        // Receive exit signals from threads.
+        if let Err(e) = core_receiver.recv() {
+            warn!("Exit resource thread failed ({})", e);
+        }
+        if let Err(e) = storage_receiver.recv() {
+            warn!("Exit storage thread failed ({})", e);
+        }
+
         self.compositor_proxy.send(ToCompositorMsg::ShutdownComplete);
     }
 

components/net/resource_thread.rs

@@ -211,7 +211,7 @@ impl ResourceChannelManager {
                 CoreResourceMsg::Synchronize(sender) => {
                     let _ = sender.send(());
                 }
-                CoreResourceMsg::Exit => {
+                CoreResourceMsg::Exit(sender) => {
                     if let Some(ref config_dir) = opts::get().config_dir {
                         match self.resource_manager.auth_cache.read() {
                             Ok(auth_cache) => write_json_to_file(&*auth_cache, config_dir, "auth_cache.json"),
@@ -226,6 +226,7 @@ impl ResourceChannelManager {
                             Err(_) => warn!("Error writing hsts list to disk"),
                         }
                     }
+                    let _ = sender.send(());
                     break;
                 }
 

components/net/storage_thread.rs

@@ -74,10 +74,11 @@ impl StorageManager {
                 StorageThreadMsg::Clear(sender, url, storage_type) => {
                     self.clear(sender, url, storage_type)
                 }
-                StorageThreadMsg::Exit => {
+                StorageThreadMsg::Exit(sender) => {
                     if let Some(ref config_dir) = opts::get().config_dir {
                         resource_thread::write_json_to_file(&self.local_data, config_dir, "local_data.json");
                     }
+                    let _ = sender.send(());
                     break
                 }
             }

components/net_traits/lib.rs

@@ -340,8 +340,9 @@ pub enum CoreResourceMsg {
     Cancel(ResourceId),
     /// Synchronization message solely for knowing the state of the ResourceChannelManager loop
     Synchronize(IpcSender<()>),
-    /// Break the load handler loop and exit
-    Exit,
+    /// Break the load handler loop, send a reply when done cleaning up local resources
+    //  and exit
+    Exit(IpcSender<()>),
 }
 
 /// Initialized but unsent request. Encapsulates everything necessary to instruct

components/net_traits/storage_thread.rs

@@ -35,6 +35,6 @@ pub enum StorageThreadMsg {
     /// clears the associated storage data by removing all the key/value pairs
     Clear(IpcSender<bool>, Url, StorageType),
 
-    /// shut down this thread
-    Exit
+    /// send a reply when done cleaning up thread resources and then shut it down
+    Exit(IpcSender<()>)
 }

tests/unit/net/resource_thread.rs

@@ -39,13 +39,16 @@ impl LoadOrigin for ResourceTest {
 #[test]
 fn test_exit() {
     let (tx, _rx) = ipc::channel().unwrap();
+    let (sender, receiver) = ipc::channel().unwrap();
     let resource_thread = new_core_resource_thread("".to_owned(), None, ProfilerChan(tx));
-    resource_thread.send(CoreResourceMsg::Exit).unwrap();
+    resource_thread.send(CoreResourceMsg::Exit(sender)).unwrap();
+    receiver.recv().unwrap();
 }
 
 #[test]
 fn test_bad_scheme() {
     let (tx, _rx) = ipc::channel().unwrap();
+    let (sender, receiver) = ipc::channel().unwrap();
     let resource_thread = new_core_resource_thread("".to_owned(), None, ProfilerChan(tx));
     let (start_chan, start) = ipc::channel().unwrap();
     let url = Url::parse("bogus://whatever").unwrap();
@@ -57,7 +60,8 @@ fn test_bad_scheme() {
       ProgressMsg::Done(result) => { assert!(result.is_err()) }
       _ => panic!("bleh")
     }
-    resource_thread.send(CoreResourceMsg::Exit).unwrap();
+    resource_thread.send(CoreResourceMsg::Exit(sender)).unwrap();
+    receiver.recv().unwrap();
 }
 
 #[test]
@@ -223,6 +227,7 @@ fn test_cancelled_listener() {
     });
 
     let (tx, _rx) = ipc::channel().unwrap();
+    let (exit_sender, exit_receiver) = ipc::channel().unwrap();
     let resource_thread = new_core_resource_thread("".to_owned(), None, ProfilerChan(tx));
     let (sender, receiver) = ipc::channel().unwrap();
     let (id_sender, id_receiver) = ipc::channel().unwrap();
@@ -244,5 +249,6 @@ fn test_cancelled_listener() {
     let response = receiver.recv().unwrap();
     assert_eq!(response.progress_port.recv().unwrap(),
                ProgressMsg::Done(Err(NetworkError::LoadCancelled)));
-    resource_thread.send(CoreResourceMsg::Exit).unwrap();
+    resource_thread.send(CoreResourceMsg::Exit(exit_sender)).unwrap();
+    exit_receiver.recv().unwrap();
 }