perf(ext/ffi): Avoid receiving on FFI async work channel when no Unsa…

…feCallback exists (#19454)

ext/ffi/callback.rs

@@ -567,7 +567,19 @@ where
   }
 
   let async_work_sender =
-    state.borrow_mut::<FfiState>().async_work_sender.clone();
+    if let Some(ffi_state) = state.try_borrow_mut::<FfiState>() {
+      ffi_state.async_work_sender.clone()
+    } else {
+      let (async_work_sender, async_work_receiver) =
+        mpsc::unbounded::<PendingFfiAsyncWork>();
+
+      state.put(FfiState {
+        async_work_receiver,
+        async_work_sender: async_work_sender.clone(),
+      });
+
+      async_work_sender
+    };
   let callback = v8::Global::new(scope, cb).into_raw();
   let current_context = scope.get_current_context();
   let context = v8::Global::new(scope, current_context).into_raw();

ext/ffi/lib.rs

@@ -114,14 +114,6 @@ deno_core::extension!(deno_ffi,
   state = |state, options| {
     // Stolen from deno_webgpu, is there a better option?
     state.put(Unstable(options.unstable));
-
-    let (async_work_sender, async_work_receiver) =
-      mpsc::unbounded::<PendingFfiAsyncWork>();
-
-    state.put(FfiState {
-      async_work_receiver,
-      async_work_sender,
-    });
   },
   event_loop_middleware = event_loop_middleware,
 );
@@ -133,11 +125,10 @@ fn event_loop_middleware(
   // FFI callbacks coming in from other threads will call in and get queued.
   let mut maybe_scheduling = false;
 
-  let mut work_items: Vec<PendingFfiAsyncWork> = vec![];
-
-  {
-    let mut op_state = op_state_rc.borrow_mut();
-    let ffi_state = op_state.borrow_mut::<FfiState>();
+  let mut op_state = op_state_rc.borrow_mut();
+  if let Some(ffi_state) = op_state.try_borrow_mut::<FfiState>() {
+    // TODO(mmastrac): This should be a SmallVec to avoid allocations in most cases
+    let mut work_items = Vec::with_capacity(1);
 
     while let Ok(Some(async_work_fut)) =
       ffi_state.async_work_receiver.try_next()
@@ -147,10 +138,11 @@ fn event_loop_middleware(
       maybe_scheduling = true;
     }
 
+    // Drop the op_state and ffi_state borrows
     drop(op_state);
-  }
-  while let Some(async_work_fut) = work_items.pop() {
-    async_work_fut();
+    for async_work_fut in work_items.into_iter() {
+      async_work_fut();
+    }
   }
 
   maybe_scheduling