Auto merge of #21647 - AgustinCB:fix-space-leak-when-pipeline-is-closed, r=asajeffrey

Fix space leak when pipeline is closed

Add a new control message to drop remove worklets.

Implement `Drop` for `Worklet` and get the worklet thread pool. Use that
pool to send `ExitWorklet` to all the threads with the id of the
`Worklet` that it's dropping.

---
- [x] `./mach build -d` does not report any errors
- [x] `./mach test-tidy` does not report any errors
- [x] These changes fix #17442
- [ ] There are tests for these changes OR
- [x] These changes do not require tests because I don't know how to add a test for this :(

If you consider this needs a test, could you point me out on how to write (and run!) one in this project? I gave this a couple tries, but I wasn't very successful.

<!-- 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/21647)
<!-- Reviewable:end -->

components/script/dom/worklet.rs

@@ -151,6 +151,13 @@ impl WorkletMethods for Worklet {
     }
 }
 
+impl Drop for Worklet {
+    fn drop(&mut self) {
+        let script_thread = ScriptThread::worklet_thread_pool();
+        script_thread.exit_worklet(self.worklet_id);
+    }
+}
+
 /// A guid for worklets.
 #[derive(Clone, Copy, Debug, Eq, Hash, JSTraceable, PartialEq)]
 pub struct WorkletId(Uuid);
@@ -312,10 +319,14 @@ impl WorkletThreadPool {
                 promise: TrustedPromise::new(promise.clone()),
             });
         }
-        // If any of the threads are blocked waiting on data, wake them up.
-        let _ = self.cold_backup_sender.send(WorkletData::WakeUp);
-        let _ = self.hot_backup_sender.send(WorkletData::WakeUp);
-        let _ = self.primary_sender.send(WorkletData::WakeUp);
+        self.wake_threads();
+    }
+
+    pub(crate) fn exit_worklet(&self, worklet_id: WorkletId) {
+        for sender in &[&self.control_sender_0, &self.control_sender_1, &self.control_sender_2] {
+            let _ = sender.send(WorkletControl::ExitWorklet(worklet_id));
+        }
+        self.wake_threads();
     }
 
     /// For testing.
@@ -325,6 +336,13 @@ impl WorkletThreadPool {
         let _ = self.primary_sender.send(msg);
         receiver.recv().expect("Test worklet has died?")
     }
+
+    fn wake_threads(&self) {
+        // If any of the threads are blocked waiting on data, wake them up.
+        let _ = self.cold_backup_sender.send(WorkletData::WakeUp);
+        let _ = self.hot_backup_sender.send(WorkletData::WakeUp);
+        let _ = self.primary_sender.send(WorkletData::WakeUp);
+    }
 }
 
 /// The data messages sent to worklet threads
@@ -338,6 +356,7 @@ enum WorkletData {
 
 /// The control message sent to worklet threads
 enum WorkletControl {
+    ExitWorklet(WorkletId),
     FetchAndInvokeAWorkletScript {
         pipeline_id: PipelineId,
         worklet_id: WorkletId,
@@ -654,6 +673,9 @@ impl WorkletThread {
     /// Process a control message.
     fn process_control(&mut self, control: WorkletControl) {
         match control {
+            WorkletControl::ExitWorklet(worklet_id) => {
+                self.global_scopes.remove(&worklet_id);
+            },
             WorkletControl::FetchAndInvokeAWorkletScript {
                 pipeline_id,
                 worklet_id,