So implementieren Sie die Deno-Kommunikation (mit Code)

Der Inhalt dieses Artikels befasst sich mit der Implementierungsmethode der Deno-Kommunikation (mit Code). Ich hoffe, dass er für Freunde in Not hilfreich ist.

Kommunikationsmethode

Der Deno-Ausführungscode ähnelt dem Knoten, einschließlich synchroner und asynchroner Methoden. Die asynchrone Methode wird über Promise.then implementiert.

Typescript/Javascript ruft Rust auf

Wie im vorherigen Abschnitt erwähnt, initialisiert Deno die v8-Isolatinstanz, wenn sie gestartet wird. Während des Initialisierungsprozesses werden die C++-Funktionen ausgeführt an Instanzen von v8 Isolate gebunden. Wenn v8 Javascript-Code ausführt, können diese gebundenen Funktionen wie Javascript-Funktionen aufgerufen werden. Die spezifische Bindung wird wie folgt implementiert:

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());
}

Nach Abschluss der Bindung kann die Zuordnung von C++-Methoden und Typescript-Methoden in Typescript durch den folgenden Code implementiert werden:

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;

Beim Ausführen von Typescript-Code müssen Sie nur libdeno einführen und die C++-Methode direkt aufrufen, zum Beispiel:

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];
}

Rufen Sie die libdeno.send-Methode auf, um die Daten an C++ zu übergeben, und rufen Sie dann den Rust-Code auf über C++, um den spezifischen Projektbetrieb zu implementieren.

Synchronisierung der Typescript-Ebene und asynchrone Implementierung

Synchronisierung

In Typescript müssen Sie nur den Synchronisierungsparameter der sendInternal-Methode auf true setzen, um dies zu beurteilen Basierend auf dem Sync-Parameter. Wenn sync wahr ist, gibt die libdeono.send-Methode das Ausführungsergebnis zurück. Um Daten zwischen Rust und Typescript zu übertragen, muss die Flatbuffer-Bibliothek verwendet werden Serialisierungsvorgang hier.

const [cmdId, resBuf] = sendInternal(builder, innerType, inner, data, true);

Asynchrone Implementierung

In ähnlicher Weise müssen Sie zum Implementieren des asynchronen Modus nur den Synchronisierungsparameter auf false setzen. Im Vergleich zur Synchronisierung verfügt der asynchrone Betrieb jedoch über mehr Fallback-Methoden bei der Durchführung der asynchronen Kommunikation. Wenn die Operation aufgerufen wird, gibt die Methode libdeno.send eine eindeutige cmdId zurück, die die Operation identifiziert. Gleichzeitig wird nach Abschluss der asynchronen Kommunikation ein Versprechensobjekt erstellt, das cmdId als Schlüssel und Versprechen als Wert verwendet und der Karte hinzugefügt wird. Der Code lautet wie folgt:

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 implementiert Synchronisation und Asynchronität

Wenn die libdeno.send-Methode in Typescript aufgerufen wird, wird die Send-Methode in der C++-Datei binding.cc aufgerufen In Deno initialisiert Es ist an v8 Isolate gebunden. In der Send-Methode wird die Dispatch-Methode in der Datei ops.rs aufgerufen, die die Zuordnung von Nachrichten zu Funktionen implementiert. Jeder Nachrichtentyp entspricht einer Funktion. Beispielsweise entspricht eine Dateilesenachricht einer Dateilesefunktion.

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)
  };

  // ...省略多余的代码
}

Bei jedem Funktionstyp wird die synchrone oder asynchrone Ausführung basierend auf dem Synchronisierungsparameterwert bestimmt, der beim Aufruf der libdeo.send-Methode in Typescript übergeben wird.

let (is_sync, op) = dispatch(isolate, control_buf, zero_copy_buf);

Synchronisierte Ausführung

Nach der Ausführung der Dispatch-Methode wird die Variable is_sync zurückgegeben. Wenn is_sync wahr ist, bedeutet dies, dass die Methode synchron ausgeführt wird, und op bedeutet das zurückgegebene Ergebnis. Der Rust-Code ruft die deno_respond-Methode in der C++-Datei api.cc auf, um die Ausführungsergebnisse zurückzusynchronisieren. Die deno_respond-Methode bestimmt anhand des Werts von current_args_, ob es sich um eine Synchronisationsnachricht handelt. Wenn current_args_ einen Wert hat, wird das Ergebnis direkt zurückgegeben .

Asynchrone Ausführung

In Deno wird die Ausführung asynchroner Vorgänge über das Tokio-Modul von Rust implementiert. Wenn es sich nach dem Aufruf der Dispatch-Methode um eine asynchrone Operation handelt, ist der Wert von is_sync falsch , und die op tut es nicht Dann gibt es das Ausführungsergebnis, sondern eine Ausführungsfunktion. Verwenden Sie das tokio-Modul, um einen Thread abzuleiten, um die Funktion asynchron auszuführen.

    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);

Wenn deno initialisiert wird, wird eine Pipe erstellt. Der Code lautet wie folgt:

let (tx, rx) = mpsc::channel::<(usize, Buf)>();

Die Pipe kann die Kommunikation zwischen verschiedenen Threads realisieren, da asynchrone Vorgänge einen neuen Thread zur Ausführung erstellen. Daher kann der Unterthread nicht direkt mit dem Hauptthread kommunizieren und muss über den Pipeline-Mechanismus implementiert werden. Nachdem die Ausführung des asynchronen Codes abgeschlossen ist, rufen Sie die Methode tx.send auf, um die Ausführungsergebnisse zur Pipeline hinzuzufügen. Die Ereignisschleife liest die Ergebnisse jedes Mal aus der Pipeline und gibt sie zurück.

Ereignisschleife

Da asynchrone Operationen auf Ereignisschleifen beruhen, erklären wir zunächst die Ereignisschleife in Deno. Tatsächlich handelt es sich bei der Ereignisschleife um einen Code, der ausgeführt wird in einer Schleife. Wenn die Bedingungen erfüllt sind, wird die Ausführung der Schleife beendet. Der Code der Hauptereignisschleife in deno wird wie folgt implementiert: Die Methode

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 wird verwendet, um zu bestimmen, ob alle asynchronen Vorgänge ausgeführt wurden Wenn alle asynchronen Vorgänge abgeschlossen sind, stoppen Sie die Ereignisschleife. Der Code der is_idle-Methode lautet wie folgt:

fn is_idle(&self) -> bool {
    self.ntasks.get() == 0 && self.get_timeout_due().is_none()
  }

Wenn ein asynchroner Methodenaufruf generiert wird, wird die folgende Methode aufgerufen, um den internen Wert zu erhöhen von ntasks um 1,

fn ntasks_increment(&self) {
    assert!(self.ntasks.get() >= 0);
    self.ntasks.set(self.ntasks.get() + 1);
  }

In der Ereignisschleife wird jedes Mal der Wert aus der Pipeline abgerufen. Dabei fungiert die Ereignisschleife als Verbraucher und der Unterthread, der die asynchrone Methode ausführt, fungiert als Produzent. Wenn in einer Ereignisschleife ein Ausführungsergebnis erhalten wird, wird die Methode ntasks_decrement aufgerufen, um den internen Wert von ntasks um 1 zu verringern. Wenn der Wert von ntasks 0 ist, beendet die Ereignisschleife die Ausführung. In jeder Schleife wird der in der Pipeline erhaltene Wert als Parameter verwendet, die Methode Complete_op aufgerufen und das Ergebnis zurückgegeben.

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中的代码执行。

Das obige ist der detaillierte Inhalt vonSo implementieren Sie die Deno-Kommunikation (mit Code). Für weitere Informationen folgen Sie bitte anderen verwandten Artikeln auf der PHP chinesischen Website!