mirror of
https://github.com/scylladb/scylladb.git
synced 2026-04-28 12:17:02 +00:00
2fd6d495fa
The compilation of wasm UDFs is performed by a call to a foreign function, which cannot be divided with yielding points and, as a result, causes long reactor stalls for big UDFs. We avoid them by submitting the compilation task to a non-seastar std::thread, and retrieving the result using seastar::alien. The thread is created at the start of the program. It executes tasks from a queue in an infinite loop. All seastar shards reference the thread through a std::shared_ptr to a `alien_thread_runner`. Considering that the compilation takes a long time anyway, the alien_thread_runner is implemented with focus on simplicity more than on performance. The tasks are stored in an std::queue, reading and writing to it is synchronized using an std::mutex for reading/ writing to the queue, and an std::condition_variable waiting until the queue has elements. When the destructor of the alien runner is called, an std::nullopt sentinel is pushed to the queue, and after all remaining tasks are finished and the sentinel is read, the thread finishes.
68 lines
2.0 KiB
C++
68 lines
2.0 KiB
C++
/*
|
|
* Copyright (C) 2023-present ScyllaDB
|
|
*/
|
|
|
|
/*
|
|
* SPDX-License-Identifier: AGPL-3.0-or-later
|
|
*/
|
|
|
|
#include <exception>
|
|
#include <seastar/core/alien.hh>
|
|
#include <seastar/core/reactor.hh>
|
|
|
|
#include "lang/wasm.hh"
|
|
#include "lang/wasm_alien_thread_runner.hh"
|
|
|
|
namespace wasm {
|
|
|
|
std::optional<wasm_compile_task> task_queue::pop_front() {
|
|
std::unique_lock lock(_mut);
|
|
_cv.wait(lock, [this] { return !_pending.empty(); });
|
|
auto work_item = std::move(_pending.front());
|
|
_pending.pop();
|
|
return work_item;
|
|
}
|
|
|
|
void task_queue::push_back(std::optional<wasm_compile_task> work_item) {
|
|
std::unique_lock lock(_mut);
|
|
_pending.emplace(std::move(work_item));
|
|
lock.unlock();
|
|
_cv.notify_one();
|
|
}
|
|
|
|
alien_thread_runner::alien_thread_runner()
|
|
: _thread([this] {
|
|
for (;;) {
|
|
auto work_item = _pending_queue.pop_front();
|
|
if (work_item) {
|
|
work_item->func();
|
|
} else {
|
|
break;
|
|
}
|
|
}
|
|
})
|
|
{ }
|
|
|
|
alien_thread_runner::~alien_thread_runner() {
|
|
_pending_queue.push_back(std::nullopt);
|
|
_thread.join();
|
|
}
|
|
|
|
void alien_thread_runner::submit(seastar::promise<rust::Box<wasmtime::Module>>& p, std::function<rust::Box<wasmtime::Module>()> f) {
|
|
seastar::noncopyable_function<void()> packaged([f = std::move(f), &p, &alien = seastar::engine().alien(), shard = seastar::this_shard_id()] () mutable {
|
|
try {
|
|
rust::Box<wasmtime::Module> mod = f();
|
|
seastar::alien::run_on(alien, shard, [&p, mod = std::move(mod)] () mutable {
|
|
p.set_value(std::move(mod));
|
|
});
|
|
} catch (...) {
|
|
seastar::alien::run_on(alien, shard, [&, eptr = std::current_exception()] {
|
|
p.set_exception(wasm::exception(format("Compilation failed: {}", eptr)));
|
|
});
|
|
}
|
|
});
|
|
_pending_queue.push_back(wasm_compile_task{.func = std::move(packaged), .done = p, .shard = seastar::this_shard_id()});
|
|
}
|
|
|
|
} // namespace wasm
|