fix(stdlib): use vector tokio runtime instead of using a worker thread

src/stdlib/dns_lookup.rs

@@ -12,10 +12,7 @@ use crate::compiler::prelude::*;
 #[cfg(not(target_arch = "wasm32"))]
 mod non_wasm {
     use std::collections::BTreeMap;
-    use std::io::Error;
     use std::net::ToSocketAddrs;
-    use std::sync::{mpsc, Arc, LazyLock, Mutex};
-    use std::thread;
     use std::time::Duration;
 
     use domain::base::iana::Class;
@@ -29,79 +26,6 @@ mod non_wasm {
     use crate::compiler::prelude::*;
     use crate::value::Value;
 
-    /// Single threaded worker for executing DNS requests.
-    /// Currently blocks on each request until result is received.
-    static WORKER: LazyLock<Worker> = LazyLock::new(Worker::new);
-    const CHANNEL_CAPACITY: usize = 100;
-
-    type Job<T> = Box<dyn FnOnce() -> T + Send + 'static>;
-
-    struct Worker {
-        thread: Option<thread::JoinHandle<()>>,
-        queue: Option<mpsc::SyncSender<Job<Result<Answer, Error>>>>,
-        result_receiver: Option<Mutex<mpsc::Receiver<Result<Answer, Error>>>>,
-    }
-
-    impl Worker {
-        // Creates a thread and 2 channels - one for jobs and one for results
-        fn new() -> Self {
-            let (sender, receiver) =
-                mpsc::sync_channel::<Job<Result<Answer, Error>>>(CHANNEL_CAPACITY);
-            let (result_sender, result_receiver) =
-                mpsc::sync_channel::<Result<Answer, Error>>(CHANNEL_CAPACITY);
-            let receiver = Arc::new(Mutex::new(receiver));
-            Self {
-                thread: Some(thread::spawn(move || loop {
-                    let job = receiver
-                        .lock()
-                        .expect("Locking job queue failed")
-                        .recv()
-                        .expect("Worker queue closed");
-                    let result = job();
-                    result_sender
-                        .send(result)
-                        .expect("Sending result back from worker failed");
-                })),
-                queue: Some(sender),
-                result_receiver: Some(result_receiver.into()),
-            }
-        }
-
-        // Sends a job to the worker
-        // Blocks until result is received
-        fn execute<F>(&self, f: F) -> Result<Answer, Error>
-        where
-            F: FnOnce() -> Result<Answer, Error> + Send + 'static,
-        {
-            let job = Box::new(f);
-
-            self.queue
-                .as_ref()
-                .expect("Expected queue to be present in the worker")
-                .send(job)
-                .expect("Submitting job to the queue failed");
-            return self
-                .result_receiver
-                .as_ref()
-                .expect("Expected result queue to be present in the worker")
-                .lock()
-                .expect("Locking result receiver failed")
-                .recv()
-                .expect("Job result channel closed");
-        }
-    }
-
-    // Custom drop implementation which stops the started thread
-    impl Drop for Worker {
-        fn drop(&mut self) {
-            drop(self.queue.take());
-            if let Some(thread) = self.thread.take() {
-                thread.join().unwrap();
-            }
-            drop(self.result_receiver.take());
-        }
-    }
-
     fn dns_lookup(value: &Value, qtype: &Value, qclass: &Value, options: Value) -> Resolved {
         let host: Name<Vec<_>> = value
             .try_bytes_utf8_lossy()?
@@ -121,17 +45,17 @@ mod non_wasm {
 
         let map = options.try_object()?;
         let conf = build_options(&map)?;
-        let answer = match Handle::try_current() {
-            Ok(_) => WORKER.execute(move || {
-                StubResolver::run_with_conf(conf, move |stub| async move {
+        let answer = tokio::task::block_in_place(|| {
+            match Handle::try_current() {
+                Ok(handle) => {
+                    handle.block_on(StubResolver::from_conf(conf).query((host, qtype, qclass)))
+                }
+                Err(_) => StubResolver::run_with_conf(conf, move |stub| async move {
                     stub.query((host, qtype, qclass)).await
-                })
-            }),
-            Err(_) => StubResolver::run_with_conf(conf, move |stub| async move {
-                stub.query((host, qtype, qclass)).await
-            }),
-        }
-        .map_err(|err| format!("query failed: {err}"))?;
+                }),
+            }
+            .map_err(|err| format!("query failed: {err}"))
+        })?;
 
         Ok(parse_answer(&answer)?.into())
     }