refactor(core): use JoinSet instead of FuturesUnordered (#19378)

This commit migrates "deno_core" from using "FuturesUnordered" to
"tokio::task::JoinSet". This makes every op to be a separate Tokio task
and should unlock better utilization of kqueue/epoll.

There were two quirks added to this PR:
- because of the fact that "JoinSet" immediately polls spawn tasks,
op sanitizers can give false positives in some cases, this was
alleviated by polling event loop once before running a test with 
"deno test", which gives canceled ops an opportunity to settle
- "JsRuntimeState::waker" was moved to "OpState::waker" so that FFI
API can still use threadsafe functions - without this change the
registered wakers were wrong as they would not wake up the 
whole "JsRuntime" but the task associated with an op

---------

Co-authored-by: Matt Mastracci <matthew@mastracci.com>

1 files changed, 7 insertions, 3 deletions

diff --git a/core/realm.rs b/core/realm.rs
index 94ce77464..d18f41e66 100644
--- a/core/realm.rs
+++ b/core/realm.rs
@@ -5,10 +5,12 @@ use crate::modules::ModuleCode;
 use crate::ops::OpCtx;
 use crate::runtime::exception_to_err_result;
 use crate::runtime::JsRuntimeState;
+use crate::task::MaskResultAsSend;
 use crate::JsRuntime;
-use crate::OpCall;
+use crate::OpId;
+use crate::OpResult;
+use crate::PromiseId;
 use anyhow::Error;
-use futures::stream::FuturesUnordered;
 use std::cell::RefCell;
 use std::collections::HashSet;
 use std::collections::VecDeque;
@@ -16,6 +18,7 @@ use std::hash::BuildHasherDefault;
 use std::hash::Hasher;
 use std::option::Option;
 use std::rc::Rc;
+use tokio::task::JoinSet;
 use v8::HandleScope;
 use v8::Local;
 
@@ -48,7 +51,8 @@ pub(crate) struct ContextState {
   pub(crate) pending_promise_rejections:
     VecDeque<(v8::Global<v8::Promise>, v8::Global<v8::Value>)>,
   pub(crate) unrefed_ops: HashSet<i32, BuildHasherDefault<IdentityHasher>>,
-  pub(crate) pending_ops: FuturesUnordered<OpCall>,
+  pub(crate) pending_ops:
+    JoinSet<MaskResultAsSend<(PromiseId, OpId, OpResult)>>,
   // We don't explicitly re-read this prop but need the slice to live alongside
   // the context
   pub(crate) op_ctxs: Box<[OpCtx]>,