deno通信实现的方法(附代码)
通信方式
deno执行代码和node相似,包含同步和异步的方式, 异步方式通过Promise.then实现。
Typescript/Javascript调用rust
在上一节中讲到deno的启动时会初始化v8 isolate实例,在初始化的过程中,会将c++的函数绑定到v8 isolate的实例上,在v8执行Javascript代码时,可以像调用Javascript函数一样调用这些绑定的函数。具体的绑定实现如下:
void InitializeContext(v8::Isolate* isolate, v8::Local<v8::Context> context) { v8::HandleScope handle_scope(isolate); v8::Context::Scope context_scope(context); auto global = context->Global(); auto deno_val = v8::Object::New(isolate); CHECK(global->Set(context, deno::v8_str("libdeno"), deno_val).FromJust()); auto print_tmpl = v8::FunctionTemplate::New(isolate, Print); auto print_val = print_tmpl->GetFunction(context).ToLocalChecked(); CHECK(deno_val->Set(context, deno::v8_str("print"), print_val).FromJust()); auto recv_tmpl = v8::FunctionTemplate::New(isolate, Recv); auto recv_val = recv_tmpl->GetFunction(context).ToLocalChecked(); CHECK(deno_val->Set(context, deno::v8_str("recv"), recv_val).FromJust()); auto send_tmpl = v8::FunctionTemplate::New(isolate, Send); auto send_val = send_tmpl->GetFunction(context).ToLocalChecked(); CHECK(deno_val->Set(context, deno::v8_str("send"), send_val).FromJust()); auto eval_context_tmpl = v8::FunctionTemplate::New(isolate, EvalContext); auto eval_context_val = eval_context_tmpl->GetFunction(context).ToLocalChecked(); CHECK(deno_val->Set(context, deno::v8_str("evalContext"), eval_context_val) .FromJust()); auto error_to_json_tmpl = v8::FunctionTemplate::New(isolate, ErrorToJSON); auto error_to_json_val = error_to_json_tmpl->GetFunction(context).ToLocalChecked(); CHECK(deno_val->Set(context, deno::v8_str("errorToJSON"), error_to_json_val) .FromJust()); CHECK(deno_val->SetAccessor(context, deno::v8_str("shared"), Shared) .FromJust()); }
在完成绑定之后,在Typescript中可以通过如下代码实现c++方法和Typescript方法的映射
libdeno.ts
interface Libdeno { recv(cb: MessageCallback): void; send(control: ArrayBufferView, data?: ArrayBufferView): null | Uint8Array; print(x: string, isErr?: boolean): void; shared: ArrayBuffer; /** Evaluate provided code in the current context. * It differs from eval(...) in that it does not create a new context. * Returns an array: [output, errInfo]. * If an error occurs, `output` becomes null and `errInfo` is non-null. */ // eslint-disable-next-line @typescript-eslint/no-explicit-any evalContext(code: string): [any, EvalErrorInfo | null]; errorToJSON: (e: Error) => string; } export const libdeno = window.libdeno as Libdeno;
在执行Typescript代码时,只需要引入libdeno,就直接调用c++方法,例如:
import { libdeno } from "./libdeno"; function sendInternal( builder: flatbuffers.Builder, innerType: msg.Any, inner: flatbuffers.Offset, data: undefined | ArrayBufferView, sync = true ): [number, null | Uint8Array] { const cmdId = nextCmdId++; msg.Base.startBase(builder); msg.Base.addInner(builder, inner); msg.Base.addInnerType(builder, innerType); msg.Base.addSync(builder, sync); msg.Base.addCmdId(builder, cmdId); builder.finish(msg.Base.endBase(builder)); const res = libdeno.send(builder.asUint8Array(), data); builder.inUse = false; return [cmdId, res]; }
调用libdeno.send方法可以将数据传给c++,然后通过c++去调用rust代码实现具体的工程操作。
Typescript层同步异步实现
同步
在Typescript中只需要设置sendInternal方法的sync参数为true即可,在rust中会根据sync参数去判断是执行同步或者异步操作,如果sync为true,libdeono.send方法会返回执行的结果,rust和typescript之间传递数据需要将数据序列化,这里序列化操作使用的是flatbuffer库。
const [cmdId, resBuf] = sendInternal(builder, innerType, inner, data, true);
异步实现
同理,实现异步方式,只需要设置sync参数为false即可,但是异步操作和同步相比,多了回掉方法,在执行异步通信时,libdeno.send方法会返回一个唯一的cmdId标志这次调用操作。同时在异步通信完成后,会创建一个promise对象,将cmdId作为key,promise作为value,加入map中。代码如下:
const [cmdId, resBuf] = sendInternal(builder, innerType, inner, data, false); util.assert(resBuf == null); const promise = util.createResolvable<msg.Base>(); promiseTable.set(cmdId, promise); return promise;
rust实现同步和异步
当在Typescript中调用libdeno.send方法时,调用了C++文件binding.cc中的Send方法,该方法是在deno初始化时绑定到v8 isolate上去的。在Send方法中去调用了ops.rs文件中的dispatch方法,该方法实现了消息到函数的映射。每个类型的消息对应了一种函数,例如读文件消息对应了读文件的函数。
pub fn dispatch( isolate: &Isolate, control: libdeno::deno_buf, data: libdeno::deno_buf, ) -> (bool, Box<Op>) { let base = msg::get_root_as_base(&control); let is_sync = base.sync(); let inner_type = base.inner_type(); let cmd_id = base.cmd_id(); let op: Box<Op> = if inner_type == msg::Any::SetTimeout { // SetTimeout is an exceptional op: the global timeout field is part of the // Isolate state (not the IsolateState state) and it must be updated on the // main thread. assert_eq!(is_sync, true); op_set_timeout(isolate, &base, data) } else { // Handle regular ops. let op_creator: OpCreator = match inner_type { msg::Any::Accept => op_accept, msg::Any::Chdir => op_chdir, msg::Any::Chmod => op_chmod, msg::Any::Close => op_close, msg::Any::FetchModuleMetaData => op_fetch_module_meta_data, msg::Any::CopyFile => op_copy_file, msg::Any::Cwd => op_cwd, msg::Any::Dial => op_dial, msg::Any::Environ => op_env, msg::Any::Exit => op_exit, msg::Any::Fetch => op_fetch, msg::Any::FormatError => op_format_error, msg::Any::Listen => op_listen, msg::Any::MakeTempDir => op_make_temp_dir, msg::Any::Metrics => op_metrics, msg::Any::Mkdir => op_mkdir, msg::Any::Open => op_open, msg::Any::ReadDir => op_read_dir, msg::Any::ReadFile => op_read_file, msg::Any::Readlink => op_read_link, msg::Any::Read => op_read, msg::Any::Remove => op_remove, msg::Any::Rename => op_rename, msg::Any::ReplReadline => op_repl_readline, msg::Any::ReplStart => op_repl_start, msg::Any::Resources => op_resources, msg::Any::Run => op_run, msg::Any::RunStatus => op_run_status, msg::Any::SetEnv => op_set_env, msg::Any::Shutdown => op_shutdown, msg::Any::Start => op_start, msg::Any::Stat => op_stat, msg::Any::Symlink => op_symlink, msg::Any::Truncate => op_truncate, msg::Any::WorkerGetMessage => op_worker_get_message, msg::Any::WorkerPostMessage => op_worker_post_message, msg::Any::Write => op_write, msg::Any::WriteFile => op_write_file, msg::Any::Now => op_now, msg::Any::IsTTY => op_is_tty, msg::Any::Seek => op_seek, msg::Any::Permissions => op_permissions, msg::Any::PermissionRevoke => op_revoke_permission, _ => panic!(format!( "Unhandled message {}", msg::enum_name_any(inner_type) )), }; op_creator(&isolate, &base, data) }; // ...省略多余的代码 }
在每个类型的函数中会根据在Typescript中调用libdeo.send方法时传入的sync参数值去判断同步执行还是异步执行。
let (is_sync, op) = dispatch(isolate, control_buf, zero_copy_buf);
同步执行
在执行dispatch方法后,会返回is_sync的变量,如果is_sync为true,表示该方法是同步执行的,op表示返回的结果。rust代码会调用c++文件api.cc中的deno_respond方法,将执行结果同步回去,deno_respond方法中根据current_args_的值去判断是否为同步消息,如果current_args_存在值,则直接返回结果。
异步执行
在deno中,执行异步操作是通过rust的Tokio模块来实现的,在调用dispatch方法后,如果是异步操作,is_sync的值为false,op不再是执行结果,而是一个执行函数。通过tokio模块派生一个线程程异步去执行该函数。
let task = op .and_then(move |buf| { let sender = tx; // tx is moved to new thread sender.send((zero_copy_id, buf)).expect("tx.send error"); Ok(()) }).map_err(|_| ()); tokio::spawn(task);
在deno初始化时,会创建一个管道,代码如下:
let (tx, rx) = mpsc::channel::<(usize, Buf)>();
管道可以实现不同线程之间的通信,由于异步操作是创建了一个新的线程去执行的,所以子线程无法直接和主线程之间通信,需要通过管道的机制去实现。在异步代码执行完成后,调用tx.send方法将执行结果加入管道里面,event loop会每次从管道里面去读取结果返回回去。
Event Loop
由于异步操作依赖事件循环,所以先解释一下deno中的事件循环,其实事件循环很简单,就是一段循环执行的代码,当达到条件后,事件循环会结束执行,deno中主要的事件循环代码实现如下:
pub fn event_loop(&self) -> Result<(), JSError> { // Main thread event loop. while !self.is_idle() { match recv_deadline(&self.rx, self.get_timeout_due()) { Ok((zero_copy_id, buf)) => self.complete_op(zero_copy_id, buf), Err(mpsc::RecvTimeoutError::Timeout) => self.timeout(), Err(e) => panic!("recv_deadline() failed: {:?}", e), } self.check_promise_errors(); if let Some(err) = self.last_exception() { return Err(err); } } // Check on done self.check_promise_errors(); if let Some(err) = self.last_exception() { return Err(err); } Ok(()) }
self.is_idle方法用来判断是否所有的异步操作都执行完毕,当所有的异步操作都执行完毕后,停止事件循环,is_idle方法代码如下:
fn is_idle(&self) -> bool { self.ntasks.get() == 0 && self.get_timeout_due().is_none() }
当产生一次异步方法调用时,会调用下面的方法,使ntasks内部的值加1,
fn ntasks_increment(&self) { assert!(self.ntasks.get() >= 0); self.ntasks.set(self.ntasks.get() + 1); }
在event loop循环中,每次从管道中去取值,这里event loop充消费者,执行异步方法的子线程充当生产者。如果在一次事件循环中,获取到了一次执行结果,那么会调用ntasks_decrement方法,使ntasks内部的值减1,当ntasks的值为0的时候,事件循环会退出执行。在每次循环中,将管道中取得的值作为参数,调用complete_op方法,将结果返回回去。
rust中将异步操作结果返回回去
在初始化v8实例时,绑定的c++方法中有一个Recv方法,该方法的作用时暴露一个Typescript的函数给rust,在deno的io.ts文件的start方法中执行libdeno.recv(handleAsyncMsgFromRust),将handleAsyncMsgFromRust函数通过c++方法暴露给rust。具体实现如下:
export function start(source?: string): msg.StartRes { libdeno.recv(handleAsyncMsgFromRust); // First we send an empty `Start` message to let the privileged side know we // are ready. The response should be a `StartRes` message containing the CLI // args and other info. const startResMsg = sendStart(); util.setLogDebug(startResMsg.debugFlag(), source); setGlobals(startResMsg.pid(), startResMsg.noColor(), startResMsg.execPath()!); return startResMsg; }
当异步操作执行完成后,可以在rust中直接调用handleAsyncMsgFromRust方法,将结果返回给Typescript。先看一下handleAsyncMsgFromRust方法的实现细节:
export function handleAsyncMsgFromRust(ui8: Uint8Array): void { // If a the buffer is empty, recv() on the native side timed out and we // did not receive a message. if (ui8 && ui8.length) { const bb = new flatbuffers.ByteBuffer(ui8); const base = msg.Base.getRootAsBase(bb); const cmdId = base.cmdId(); const promise = promiseTable.get(cmdId); util.assert(promise != null, `Expecting promise in table. ${cmdId}`); promiseTable.delete(cmdId); const err = errors.maybeError(base); if (err != null) { promise!.reject(err); } else { promise!.resolve(base); } } // Fire timers that have become runnable. fireTimers(); }
从代码handleAsyncMsgFromRust方法的实现中可以知道,首先通过flatbuffer反序列化返回的结果,然后获取返回结果的cmdId,根据cmdId获取之前创建的promise对象,然后调用promise.resolve方法触发promise.then中的代码执行。
以上就是deno通信实现的方法(附代码)的详细内容,更多请关注其它相关文章!