WIP: change plugin interface to avoid unloading plugin before it's sa…

…fe (denoland#5210)

I seem to have solved the crashes related to the plugin's op dispatchers
and op futures getting dropped after the plugin is unloaded.

However `runTestPluginClose()` still crashes, it's not quite clear to me
yet what causes it.

cli/lib.rs

@@ -46,6 +46,7 @@ pub mod msg;
 pub mod op_error;
 pub mod ops;
 pub mod permissions;
+pub mod plugin;
 mod repl;
 pub mod resolve_addr;
 pub mod signal;

cli/ops/mod.rs

@@ -19,7 +19,7 @@ pub mod net;
 mod net_unix;
 pub mod os;
 pub mod permissions;
-pub mod plugins;
+pub mod plugin;
 pub mod process;
 pub mod random;
 pub mod repl;

cli/ops/plugins.rs → cli/ops/plugin.rs

@@ -2,14 +2,17 @@ use super::dispatch_json::{Deserialize, JsonOp, Value};
 use crate::fs as deno_fs;
 use crate::op_error::OpError;
 use crate::ops::json_op;
+use crate::plugin::InitContext;
 use crate::state::State;
+
 use deno_core::CoreIsolate;
 use deno_core::ZeroCopyBuf;
 use dlopen::symbor::Library;
 use std::ffi::OsStr;
 use std::path::Path;
+use std::rc::Rc;
 
-pub type PluginInitFn = fn(isolate: &mut CoreIsolate);
+pub type PluginInitFn = fn(&mut InitContext);
 
 pub fn init(i: &mut CoreIsolate, s: &State) {
   i.register_op(
@@ -18,13 +21,19 @@ pub fn init(i: &mut CoreIsolate, s: &State) {
   );
 }
 
-fn open_plugin<P: AsRef<OsStr>>(lib_path: P) -> Result<Library, OpError> {
+fn open_plugin<P: AsRef<OsStr>>(lib_path: P) -> Result<Rc<Library>, OpError> {
   debug!("Loading Plugin: {:#?}", lib_path.as_ref());
-  Library::open(lib_path).map_err(OpError::from)
+  Library::open(lib_path).map(Rc::new).map_err(OpError::from)
 }
 
 struct PluginResource {
-  lib: Library,
+  lib: Rc<Library>,
+}
+
+impl PluginResource {
+  fn new(lib: &Rc<Library>) -> Self {
+    Self { lib: lib.clone() }
+  }
 }
 
 #[derive(Deserialize)]
@@ -45,8 +54,8 @@ pub fn op_open_plugin(
 
   state.check_plugin(&filename)?;
 
-  let lib = open_plugin(filename).unwrap();
-  let plugin_resource = PluginResource { lib };
+  let plugin_lib = open_plugin(filename).unwrap();
+  let plugin_resource = PluginResource::new(&plugin_lib);
 
   let mut resource_table = isolate.resource_table.borrow_mut();
   let rid = resource_table.add("plugin", Box::new(plugin_resource));
@@ -60,7 +69,8 @@ pub fn op_open_plugin(
   .unwrap();
   drop(resource_table);
 
-  deno_plugin_init(isolate);
+  let mut context = InitContext::new(isolate, &plugin_lib);
+  deno_plugin_init(&mut context);
 
   Ok(JsonOp::Sync(json!(rid)))
 }

cli/plugin.rs

@@ -0,0 +1,64 @@
+use deno_core::CoreIsolate;
+use deno_core::Op;
+use deno_core::OpAsyncFuture;
+use deno_core::OpDispatcher;
+use deno_core::OpId;
+use dlopen::symbor::Library;
+use futures::prelude::*;
+use std::rc::Rc;
+
+/// Context passed to the `deno_plugin_init()` function that plugins have to
+/// exports. This avoids the need to expose the `Library` type in the public
+/// interface.
+pub struct InitContext<'a> {
+  isolate: &'a mut CoreIsolate,
+  plugin_lib: &'a Rc<Library>,
+}
+
+impl<'a> InitContext<'a> {
+  pub(crate) fn new(
+    isolate: &'a mut CoreIsolate,
+    plugin_lib: &'a Rc<Library>,
+  ) -> Self {
+    Self {
+      isolate,
+      plugin_lib,
+    }
+  }
+}
+
+/// Does the same as `core::Isolate::register_op()`, but additionally makes
+/// the registered op dispatcher, as well as the op futures created by it, keep
+/// a reference to plugin, so that the plugin doesn't get unloaded before all
+/// its op registrations and the futures created by them are dropped.
+pub fn register_op(
+  context: &mut InitContext,
+  name: &str,
+  dispatcher: impl OpDispatcher,
+) -> OpId {
+  let InitContext {
+    isolate,
+    plugin_lib,
+  } = context;
+  let plugin_lib = plugin_lib.clone();
+
+  let wrap_async_op =
+    |plugin_lib: &Rc<Library>, fut: OpAsyncFuture| -> OpAsyncFuture {
+      let plugin_lib = plugin_lib.clone();
+      async move {
+        let _ = plugin_lib;
+        fut.await
+      }
+      .boxed_local()
+    };
+
+  isolate
+    .op_registry
+    .register(name, move |isolate, control, zero_copy| {
+      match dispatcher(isolate, control, zero_copy) {
+        sync_op @ Op::Sync(..) => sync_op,
+        Op::Async(fut) => Op::Async(wrap_async_op(&plugin_lib, fut)),
+        Op::AsyncUnref(fut) => Op::AsyncUnref(wrap_async_op(&plugin_lib, fut)),
+      }
+    })
+}

cli/web_worker.rs

@@ -136,7 +136,7 @@ impl WebWorker {
         ops::runtime_compiler::init(isolate, &state);
         ops::fs::init(isolate, &state);
         ops::fs_events::init(isolate, &state);
-        ops::plugins::init(isolate, &state);
+        ops::plugin::init(isolate, &state);
         ops::net::init(isolate, &state);
         ops::tls::init(isolate, &state);
         ops::os::init(isolate, &state);

cli/worker.rs

@@ -243,7 +243,7 @@ impl MainWorker {
       ops::fs::init(isolate, &state);
       ops::fs_events::init(isolate, &state);
       ops::io::init(isolate, &state);
-      ops::plugins::init(isolate, &state);
+      ops::plugin::init(isolate, &state);
       ops::net::init(isolate, &state);
       ops::tls::init(isolate, &state);
       ops::os::init(isolate, &state);

core/ops.rs

@@ -20,13 +20,19 @@ pub enum Op {
   AsyncUnref(OpAsyncFuture),
 }
 
-/// Main type describing op
-pub type OpDispatcher =
-  dyn Fn(&mut CoreIsolate, &[u8], Option<ZeroCopyBuf>) -> Op + 'static;
+/// Main trait describing op dispatcher.
+pub trait OpDispatcher:
+  Fn(&'_ mut CoreIsolate, &'_ [u8], Option<ZeroCopyBuf>) -> Op + 'static
+{
+}
+impl<F> OpDispatcher for F where
+  F: Fn(&'_ mut CoreIsolate, &'_ [u8], Option<ZeroCopyBuf>) -> Op + 'static
+{
+}
 
 #[derive(Default)]
 pub struct OpRegistry {
-  dispatchers: Vec<Rc<OpDispatcher>>,
+  dispatchers: Vec<Rc<dyn OpDispatcher>>,
   name_to_id: HashMap<String, OpId>,
 }
 
@@ -41,7 +47,7 @@ impl OpRegistry {
     registry
   }
 
-  pub fn register<F>(&mut self, name: &str, op: F) -> OpId
+  pub fn register<F>(&mut self, name: &str, dispatcher: F) -> OpId
   where
     F: Fn(&mut CoreIsolate, &[u8], Option<ZeroCopyBuf>) -> Op + 'static,
   {
@@ -52,7 +58,7 @@ impl OpRegistry {
       existing.is_none(),
       format!("Op already registered: {}", name)
     );
-    self.dispatchers.push(Rc::new(op));
+    self.dispatchers.push(Rc::new(dispatcher));
     op_id
   }
 
@@ -61,7 +67,7 @@ impl OpRegistry {
     op_map_json.as_bytes().to_owned().into_boxed_slice()
   }
 
-  pub fn get(&self, op_id: OpId) -> Option<Rc<OpDispatcher>> {
+  pub fn get(&self, op_id: OpId) -> Option<Rc<dyn OpDispatcher>> {
     self.dispatchers.get(op_id as usize).map(Rc::clone)
   }
 }

test_plugin/src/lib.rs

@@ -1,16 +1,18 @@
 extern crate deno_core;
 extern crate futures;
 
+use deno::plugin::register_op;
+use deno::plugin::InitContext;
 use deno_core::Buf;
 use deno_core::CoreIsolate;
 use deno_core::Op;
 use deno_core::ZeroCopyBuf;
 use futures::future::FutureExt;
 
 #[no_mangle]
-pub fn deno_plugin_init(isolate: &mut CoreIsolate) {
-  isolate.register_op("testSync", op_test_sync);
-  isolate.register_op("testAsync", op_test_async);
+pub fn deno_plugin_init(context: &mut InitContext) {
+  register_op(context, "testSync", op_test_sync);
+  register_op(context, "testAsync", op_test_async);
 }
 
 pub fn op_test_sync(

test_plugin/tests/integration_tests.rs

@@ -1,9 +1,6 @@
 // To run this test manually:
 //   cd test_plugin
-//   ../target/debug/deno --allow-plugin tests/test.js debug
-
-// TODO(ry) Re-enable this test on windows. It is flaky for an unknown reason.
-#![cfg(not(windows))]
+//   ../target/debug/deno run --unstable --allow-plugin tests/test.js debug
 
 use deno::test_util::*;
 use std::process::Command;