如何从 webworker 中取消 wasm 进程

Question

我有一个 wasm 进程（从 C++ 编译），用于处理 Web 应用程序中的数据。假设必要的代码如下所示：std::vector<JSONObject> datafor (size_t i = 0; i < data.size(); i++){    process_data(data[i]);    if (i % 1000 == 0) {        bool is_cancelled = check_if_cancelled();        if (is_cancelled) {            break;        }    }}此代码基本上类似于 SQL 查询界面“运行/处理查询”：但是，查询可能需要几分钟来运行/处理，并且用户可以在任何给定时间取消他们的查询。取消过程将发生在正常的 javascript/web 应用程序中，在运行 wasm 的 service Worker 之外。那么我的问题是什么是我们如何知道用户已单击“取消”按钮并将其传达给 wasm 进程以便知道该进程已被取消以便它可以退出的示例？使用worker.terminate()不是一种选择，因为我们需要为此保留所有加载的数据，worker而不能仅仅杀死该工作人员（它需要使用其存储的数据保持活动状态，因此可以运行另一个查询......）。在 javascript 和 worker/wasm/c++ 应用程序之间进行通信的示例方式是什么，以便我们知道何时退出以及如何正确执行？此外，让我们假设一个典型的查询需要 60 秒来运行并使用 cpp/wasm 在浏览器中处理 500MB 的数据。更新：我认为根据一些研究（以及下面的初始答案/评论）和一些反馈，这里有以下可能的解决方案：使用两个 worker，一个 worker 存储数据，另一个 worker 处理数据。通过这种方式可以终止处理工作者，并且数据将始终保留。可行的？不是真的，因为每当它启动时，将超过 500MB 的数据复制到 webworker 会花费太多时间。这可以（以前）使用SharedArrayBuffer 完成，但由于一些安全问题，它的支持现在非常有限/不存在。太糟糕了，因为如果支持，这似乎是迄今为止最好的解决方案......使用Emterpreter和使用emscripten_sleep_with_yield. 可行的？不，在使用 Emterpreter 时会破坏性能（在上面的文档中提到过），并使所有查询速度降低约 4-6 倍。始终运行第二个工作程序，并且在 UI 中只显示最新的。可行的？不，如果它不是共享数据结构并且数据大小为 500MB x 2 = 1GB（在现代桌面浏览器/计算机中运行时，500MB 似乎是一个大但可接受的大小），则可能会遇到很多 OOM 错误。使用对服务器的 API 调用来存储状态并检查查询是否被取消。可行的？是的，尽管从每个正在运行的查询中每秒对网络请求进行长时间轮询似乎很笨拙。使用增量解析方法，一次只解析一行。可行的？是的，但也需要大量重写解析函数，以便每个函数都支持这一点（实际的数据解析在几个函数中处理——过滤、搜索、计算、分组、排序等。使用 IndexedDB 并将状态存储在 javascript 中。在 WASM 中分配一块内存，然后将其指针返回给 JavaScript。然后在那里读取数据库并填充指针。然后用 C++ 处理你的数据。可行的？不确定，尽管如果可以实施，这似乎是最佳解决方案。[还要别的吗？]在赏金中，我想知道三件事：如果以上六项分析似乎普遍有效？我还缺少其他（也许更好）的方法吗？任何人都可以展示一个非常基本的例子#6 - 如果可能并且跨浏览器工作，这似乎是最好的解决方案。

Answer 1

对于 Chrome（仅），您可以使用共享内存（共享缓冲区作为内存）。当您想停下来时，请在内存中举起一面旗帜。不是这个解决方案的忠实粉丝（很复杂，仅在 chrome 中受支持）。它还取决于您的查询如何工作，以及是否有冗长查询可以检查标志的地方。

相反，您可能应该多次调用 c++ 函数（例如，对于每个查询）并检查您是否应该在每次调用后停止（只需向工作人员发送一条消息即可停止）。

我所说的多次是分阶段进行查询（单个查询的多个函数校准）。它可能不适用于您的情况。

无论如何，AFAIK 无法向 Webassembly 执行（例如 Linux kill）发送信号。因此，您必须等待操作完成才能完成取消。

我附上了一个可以解释这个想法的代码片段。

worker.js:

... init webassembly

onmessage = function(q) {

  // query received from main thread.

  const result = ... call webassembly(q);

  postMessage(result);

}

main.js:

const worker = new Worker("worker.js");

const cancel = false;

const processing = false;

worker.onmessage(function(r) {

  // when worker has finished processing the query.

  // r is the results of the processing.

  processing = false;

  if (cancel === true) {

    // processing is done, but result is not required.

    // instead of showing the results, update that the query was canceled.

    cancel = false;

    ... update UI "cancled".

    return;

  }

  

  ... update UI "results r".

}

function onCancel() {

  // Occurs when user clicks on the cancel button.

  if (cancel) {

    // sanity test - prevent this in UI.

    throw "already cancelling";

  }

  

  cancel = true;

  

  ... update UI "canceling". 

}

function onQuery(q) {

  if (processing === true) {

    // sanity test - prevent this in UI.

    throw "already processing";

  }

  

  processing = true;

  // Send the query to the worker.

  // When the worker receives the message it will process the query via webassembly.

  worker.postMessage(q);

}

从用户体验的角度来看一个想法：你可以创建~两个工人。这将占用两倍的内存，但允许您“立即”“取消”一次。（这只是意味着在后端，第二个工作人员将运行下一个查询，当第一个完成取消时，取消将再次立即生效）。

Answer 2

共享线程

由于工作线程和它调用的 C++ 函数共享同一个线程，工作线程也将被阻塞，直到 C++ 循环完成，并且无法处理任何传入的消息。我认为一个可靠的选项可以通过从主应用程序一次初始化一次迭代来最大程度地减少线程被阻塞的时间。

它看起来像这样。

main.js  ->  worker.js  ->  C++ function  ->  worker.js  ->  main.js

打破循环

下面，C++ 有一个初始化为 0 的变量，它将在每次循环迭代时递增并存储在内存中。然后 C++ 函数执行一次循环迭代，增加变量以跟踪循环位置，然后立即中断。

int x;

x = 0; // initialized counter at 0

std::vector<JSONObject> data

for (size_t i = x; i < data.size(); i++)

{

    process_data(data[i]);

    x++ // increment counter

    break; // stop function until told to iterate again starting at x

}

然后你应该能够向 web worker 发布一条消息，然后它会向 main.js 发送一条消息，表明该线程不再被阻塞。

取消操作

至此，main.js 知道 web worker 线程不再被阻塞，并且可以决定是否告诉 web worker 再次执行 C++ 函数（C++ 变量跟踪内存中的循环增量）。

let continueOperation = true

// here you can set to false at any time since the thread is not blocked here

worker.expensiveThreadBlockingFunction()

// results in one iteration of the loop being iterated until message is received below

worker.onmessage = function(e) {

    if (continueOperation) {

        worker.expensiveThreadBlockingFunction()

        // execute worker function again, ultimately continuing the increment in C++

    } {

        return false

        // or send message to worker to reset C++ counter to prepare for next execution

    }

}

继续操作

假设一切正常，并且用户没有取消操作，循环应该继续直到完成。请记住，您还应该发送有关循环是否已完成或需要继续的不同消息，这样您就不会一直阻塞工作线程。