mirror of
https://github.com/simplex-chat/simplex-chat.git
synced 2026-09-27 00:39:00 +00:00
core, libs: configurable queue size, library fixes (#7542)
* core: add chat_migrate_init_queue FFI export * bots: fix BadgeServiceErrorCode API type * nodejs: pass required command fields * nodejs: fix migration error types * nodejs: install libsimplex from SIMPLEX_LIBS_DIR * nodejs: add queue size option * nodejs: regenerate docs * python: add queue size option * bots: pass incognito in APIConnect * nodejs: accept documented success responses * python: accept documented success responses * nodejs: dispatch each bot message once * nodejs: fix startChat events loop lifecycle * nodejs, python: parse multi-line bot commands * nodejs: fix file buffer handling in addon * nodejs: keep events loop when chat stop fails * python: fix send_and_wait race, load lib off loop * python: make queue size export optional * nodejs: receive events on a dedicated thread * nodejs: release haskell thread after receive * python: receive on a dedicated thread per chat * python: test receive shutdown order * nodejs: one receive thread per chat controller * nodejs: harden receiver shutdown and tests * nodejs: stop chat before closing store * python: stop chat before closing store * nodejs, python: harden close regression and retry * python: free results with ucrt on windows * nodejs: enable c++ exceptions on mac and windows
This commit is contained in:
sh
@@ -3,6 +3,14 @@
|
||||
#include <string>
|
||||
#include <functional>
|
||||
#include <cstdlib>
|
||||
#include <climits>
|
||||
#include <memory>
|
||||
#include <thread>
|
||||
#include <system_error>
|
||||
#include <mutex>
|
||||
#include <condition_variable>
|
||||
#include <deque>
|
||||
#include <unordered_map>
|
||||
#include "simplex.h"
|
||||
|
||||
namespace simplex {
|
||||
@@ -81,6 +89,11 @@ class ResultAsyncWorker : public AsyncWorker {
|
||||
return ctrl_;
|
||||
}
|
||||
|
||||
// the worker thread reads this object's memory until the worker completes
|
||||
void KeepAlive(Object obj) {
|
||||
keep_alive_ = Persistent(obj);
|
||||
}
|
||||
|
||||
protected:
|
||||
std::string result_;
|
||||
uintptr_t ctrl_ = 0;
|
||||
@@ -88,6 +101,7 @@ class ResultAsyncWorker : public AsyncWorker {
|
||||
private:
|
||||
ExecuteFn execute_fn_;
|
||||
ResultProcessor result_processor_;
|
||||
ObjectReference keep_alive_;
|
||||
};
|
||||
|
||||
class BinaryAsyncWorker : public AsyncWorker {
|
||||
@@ -97,21 +111,25 @@ class BinaryAsyncWorker : public AsyncWorker {
|
||||
BinaryAsyncWorker(Function& callback, ExecuteFn execute_fn)
|
||||
: AsyncWorker(callback), execute_fn_(std::move(execute_fn)) {}
|
||||
|
||||
~BinaryAsyncWorker() {
|
||||
free(original_buf);
|
||||
}
|
||||
|
||||
void Execute() override {
|
||||
execute_fn_(this);
|
||||
}
|
||||
|
||||
void OnOK() override {
|
||||
HandleScope scope(Env());
|
||||
if (original_buf == nullptr || binary_len == 0) {
|
||||
Callback().Call({Env().Null(), Env().Undefined()});
|
||||
char* buf = original_buf;
|
||||
original_buf = nullptr;
|
||||
if (binary_len == 0) {
|
||||
free(buf);
|
||||
Callback().Call({Env().Null(), Buffer<char>::New(Env(), 0)});
|
||||
return;
|
||||
}
|
||||
char* data_ptr = original_buf + 5;
|
||||
auto finalizer = [](Napi::Env env, char* finalize_data, char* orig) {
|
||||
free(orig);
|
||||
};
|
||||
Napi::Buffer<char> buffer = Napi::Buffer<char>::New(Env(), data_ptr, binary_len, finalizer, original_buf);
|
||||
// Copies when the runtime forbids external buffers (Electron); the finalizer then runs immediately.
|
||||
Buffer<char> buffer = Buffer<char>::NewOrCopy(Env(), buf + 5, binary_len, [](Napi::Env, char*, char* orig) { free(orig); }, buf);
|
||||
Callback().Call({Env().Null(), buffer});
|
||||
}
|
||||
|
||||
@@ -176,6 +194,159 @@ Napi::Promise CreatePromiseAndCallback(Env env, Function& cb_out) {
|
||||
return deferred.Promise();
|
||||
}
|
||||
|
||||
const char* const RECEIVER_STOPPED = "chat receiver stopped";
|
||||
|
||||
struct RecvRequest {
|
||||
int wait = 0;
|
||||
std::shared_ptr<Promise::Deferred> deferred;
|
||||
};
|
||||
|
||||
// Holds the event loop open only while receives are pending; used only on the JS main thread.
|
||||
class PendingReceives {
|
||||
public:
|
||||
explicit PendingReceives(ThreadSafeFunction tsfn) : tsfn_(tsfn) {}
|
||||
|
||||
const ThreadSafeFunction& Tsfn() const {
|
||||
return tsfn_;
|
||||
}
|
||||
|
||||
void Add(Napi::Env env) {
|
||||
if (count_++ == 0) tsfn_.Ref(env);
|
||||
}
|
||||
|
||||
void Remove(Napi::Env env) {
|
||||
if (--count_ == 0) tsfn_.Unref(env);
|
||||
}
|
||||
|
||||
private:
|
||||
ThreadSafeFunction tsfn_;
|
||||
size_t count_ = 0;
|
||||
};
|
||||
|
||||
// A blocking receive would hold a libuv pool thread for up to `wait`, stalling fs, dns and crypto.
|
||||
class Receiver {
|
||||
public:
|
||||
// Returns nullptr with a pending JS exception if the TSFN cannot be created, throws std::system_error if the thread cannot start.
|
||||
static std::shared_ptr<Receiver> Start(Napi::Env env, chat_ctrl ctrl) {
|
||||
ThreadSafeFunction tsfn = ThreadSafeFunction::New(env, Function::New(env, [](const CallbackInfo&) {}), "chat_recv_msg_wait", 0, 1);
|
||||
if (env.IsExceptionPending()) {
|
||||
return nullptr;
|
||||
}
|
||||
auto receiver = std::make_shared<Receiver>(ctrl, tsfn);
|
||||
try {
|
||||
receiver->thread_ = std::thread(&Receiver::Run, receiver.get());
|
||||
} catch (const std::system_error&) {
|
||||
tsfn.Release();
|
||||
throw;
|
||||
}
|
||||
return receiver;
|
||||
}
|
||||
|
||||
Receiver(chat_ctrl ctrl, ThreadSafeFunction tsfn) : ctrl_(ctrl), pending_(std::make_shared<PendingReceives>(tsfn)) {}
|
||||
|
||||
Receiver(const Receiver&) = delete;
|
||||
Receiver& operator=(const Receiver&) = delete;
|
||||
|
||||
~Receiver() {
|
||||
Stop();
|
||||
}
|
||||
|
||||
void Enqueue(Napi::Env env, RecvRequest request) {
|
||||
pending_->Add(env);
|
||||
{
|
||||
std::lock_guard<std::mutex> lock(mutex_);
|
||||
queue_.push_back(std::move(request));
|
||||
}
|
||||
cv_.notify_one();
|
||||
}
|
||||
|
||||
void RequestStop() {
|
||||
{
|
||||
std::lock_guard<std::mutex> lock(mutex_);
|
||||
stop_ = true;
|
||||
}
|
||||
cv_.notify_one();
|
||||
}
|
||||
|
||||
// Waits for the receive in progress, so it must not run on the JS main thread outside env teardown.
|
||||
void Stop() {
|
||||
RequestStop();
|
||||
if (thread_.joinable()) {
|
||||
thread_.join();
|
||||
}
|
||||
}
|
||||
|
||||
private:
|
||||
void Run() {
|
||||
for (;;) {
|
||||
RecvRequest request;
|
||||
{
|
||||
std::unique_lock<std::mutex> lock(mutex_);
|
||||
cv_.wait(lock, [this] { return stop_ || !queue_.empty(); });
|
||||
if (stop_) break;
|
||||
request = std::move(queue_.front());
|
||||
queue_.pop_front();
|
||||
}
|
||||
char* c_res = chat_recv_msg_wait(ctrl_, request.wait);
|
||||
napi_status status = Settle(request.deferred, [c_res](Napi::Env env, Promise::Deferred& deferred) {
|
||||
if (c_res == nullptr) {
|
||||
deferred.Reject(Error::New(env, "chat_recv_msg_wait failed").Value());
|
||||
} else {
|
||||
deferred.Resolve(String::New(env, c_res));
|
||||
free(c_res);
|
||||
}
|
||||
});
|
||||
if (status != napi_ok) {
|
||||
free(c_res);
|
||||
}
|
||||
}
|
||||
std::deque<RecvRequest> unserved;
|
||||
{
|
||||
std::lock_guard<std::mutex> lock(mutex_);
|
||||
unserved.swap(queue_);
|
||||
}
|
||||
for (RecvRequest& request : unserved) {
|
||||
Settle(request.deferred, [](Napi::Env env, Promise::Deferred& deferred) {
|
||||
deferred.Reject(Error::New(env, RECEIVER_STOPPED).Value());
|
||||
});
|
||||
}
|
||||
pending_->Tsfn().Release();
|
||||
// Each OS thread that enters Haskell keeps an RTS task record until it calls hs_thread_done.
|
||||
hs_thread_done();
|
||||
}
|
||||
|
||||
template <typename SettleFn>
|
||||
napi_status Settle(std::shared_ptr<Promise::Deferred> deferred, SettleFn settle) {
|
||||
// The callback may run after this Receiver is destroyed, so it owns the pending count.
|
||||
std::shared_ptr<PendingReceives> pending = pending_;
|
||||
return pending->Tsfn().BlockingCall([pending, deferred, settle](Napi::Env env, Function) {
|
||||
settle(env, *deferred);
|
||||
pending->Remove(env);
|
||||
});
|
||||
}
|
||||
|
||||
const chat_ctrl ctrl_;
|
||||
const std::shared_ptr<PendingReceives> pending_;
|
||||
std::mutex mutex_;
|
||||
std::condition_variable cv_;
|
||||
std::deque<RecvRequest> queue_;
|
||||
bool stop_ = false;
|
||||
std::thread thread_;
|
||||
};
|
||||
|
||||
// Keyed by chat_ctrl, accessed only on the JS main thread.
|
||||
using Receivers = std::unordered_map<uintptr_t, std::shared_ptr<Receiver>>;
|
||||
|
||||
std::shared_ptr<Receiver> TakeReceiver(Receivers& receivers, chat_ctrl ctrl) {
|
||||
auto it = receivers.find(reinterpret_cast<uintptr_t>(ctrl));
|
||||
if (it == receivers.end()) {
|
||||
return nullptr;
|
||||
}
|
||||
std::shared_ptr<Receiver> receiver = std::move(it->second);
|
||||
receivers.erase(it);
|
||||
return receiver;
|
||||
}
|
||||
|
||||
// Common result processors
|
||||
ResultAsyncWorker::ResultProcessor MigrateResultProcessor() {
|
||||
return [](ResultAsyncWorker* worker, Napi::Env env) {
|
||||
@@ -215,6 +386,39 @@ Value ChatMigrateInit(const CallbackInfo& args) {
|
||||
return promise;
|
||||
}
|
||||
|
||||
Value ChatMigrateInitQueue(const CallbackInfo& args) {
|
||||
Env env = args.Env();
|
||||
if (args.Length() < 4 || !args[0].IsString() || !args[1].IsString() || !args[2].IsString() || !args[3].IsNumber()) {
|
||||
TypeError::New(env, "Expected three string arguments and number").ThrowAsJavaScriptException();
|
||||
return env.Undefined();
|
||||
}
|
||||
|
||||
std::string path = args[0].As<String>().Utf8Value();
|
||||
std::string key = args[1].As<String>().Utf8Value();
|
||||
std::string confirm = args[2].As<String>().Utf8Value();
|
||||
Number queue_size_arg = args[3].As<Number>();
|
||||
int queue_size = queue_size_arg.Int32Value();
|
||||
if (static_cast<double>(queue_size) != queue_size_arg.DoubleValue()) {
|
||||
RangeError::New(env, "Expected 32-bit integer queue size").ThrowAsJavaScriptException();
|
||||
return env.Undefined();
|
||||
}
|
||||
|
||||
Function cb;
|
||||
Promise promise = CreatePromiseAndCallback(env, cb);
|
||||
|
||||
auto execute_fn = [path, key, confirm, queue_size](ResultAsyncWorker* worker) {
|
||||
chat_ctrl ctrl = nullptr;
|
||||
char* c_res = chat_migrate_init_queue(path.c_str(), key.c_str(), confirm.c_str(), queue_size, &ctrl);
|
||||
worker->SetCtrl(reinterpret_cast<uintptr_t>(ctrl));
|
||||
HandleCResult(worker, c_res, "chat_migrate_init_queue");
|
||||
};
|
||||
|
||||
ResultAsyncWorker* worker = new ResultAsyncWorker(cb, std::move(execute_fn), MigrateResultProcessor());
|
||||
worker->Queue();
|
||||
|
||||
return promise;
|
||||
}
|
||||
|
||||
Value ChatCloseStore(const CallbackInfo& args) {
|
||||
Env env = args.Env();
|
||||
if (args.Length() < 1 || !args[0].IsBigInt()) {
|
||||
@@ -223,11 +427,15 @@ Value ChatCloseStore(const CallbackInfo& args) {
|
||||
}
|
||||
|
||||
chat_ctrl ctrl = FromChatCtrlBigInt(args[0]);
|
||||
std::shared_ptr<Receiver> receiver = TakeReceiver(*static_cast<Receivers*>(args.Data()), ctrl);
|
||||
|
||||
Function cb;
|
||||
Promise promise = CreatePromiseAndCallback(env, cb);
|
||||
|
||||
auto execute_fn = [ctrl](ResultAsyncWorker* worker) {
|
||||
auto execute_fn = [ctrl, receiver](ResultAsyncWorker* worker) {
|
||||
if (receiver) {
|
||||
receiver->Stop();
|
||||
}
|
||||
char* c_res = chat_close_store(ctrl);
|
||||
HandleCResult(worker, c_res, "chat_close_store");
|
||||
};
|
||||
@@ -271,44 +479,63 @@ Value ChatRecvMsgWait(const CallbackInfo& args) {
|
||||
|
||||
chat_ctrl ctrl = FromChatCtrlBigInt(args[0]);
|
||||
int wait = static_cast<int>(args[1].As<Number>().Int32Value());
|
||||
Receivers& receivers = *static_cast<Receivers*>(args.Data());
|
||||
|
||||
Function cb;
|
||||
Promise promise = CreatePromiseAndCallback(env, cb);
|
||||
auto deferred = std::make_shared<Promise::Deferred>(Promise::Deferred::New(env));
|
||||
auto it = receivers.find(reinterpret_cast<uintptr_t>(ctrl));
|
||||
if (it == receivers.end()) {
|
||||
std::shared_ptr<Receiver> receiver;
|
||||
try {
|
||||
receiver = Receiver::Start(env, ctrl);
|
||||
} catch (const std::system_error& e) {
|
||||
deferred->Reject(Error::New(env, e.what()).Value());
|
||||
return deferred->Promise();
|
||||
}
|
||||
if (!receiver) {
|
||||
return env.Undefined();
|
||||
}
|
||||
it = receivers.emplace(reinterpret_cast<uintptr_t>(ctrl), std::move(receiver)).first;
|
||||
}
|
||||
it->second->Enqueue(env, {wait, deferred});
|
||||
|
||||
auto execute_fn = [ctrl, wait](ResultAsyncWorker* worker) {
|
||||
char* c_res = chat_recv_msg_wait(ctrl, wait);
|
||||
HandleCResult(worker, c_res, "chat_recv_msg_wait");
|
||||
};
|
||||
|
||||
ResultAsyncWorker* worker = new ResultAsyncWorker(cb, std::move(execute_fn));
|
||||
worker->Queue();
|
||||
|
||||
return promise;
|
||||
return deferred->Promise();
|
||||
}
|
||||
|
||||
Value ChatWriteFile(const CallbackInfo& args) {
|
||||
Env env = args.Env();
|
||||
if (args.Length() < 3 || !args[0].IsBigInt() || !args[1].IsString() || !args[2].IsArrayBuffer()) {
|
||||
TypeError::New(env, "Expected bigint (ctrl), string (path), ArrayBuffer").ThrowAsJavaScriptException();
|
||||
if (args.Length() < 3 || !args[0].IsBigInt() || !args[1].IsString() || !(args[2].IsArrayBuffer() || args[2].IsTypedArray())) {
|
||||
TypeError::New(env, "Expected bigint (ctrl), string (path), ArrayBuffer or Uint8Array").ThrowAsJavaScriptException();
|
||||
return env.Undefined();
|
||||
}
|
||||
|
||||
chat_ctrl ctrl = FromChatCtrlBigInt(args[0]);
|
||||
std::string path = args[1].As<String>().Utf8Value();
|
||||
ArrayBuffer ab = args[2].As<ArrayBuffer>();
|
||||
char* data = static_cast<char*>(ab.Data());
|
||||
size_t len = ab.ByteLength();
|
||||
char* data;
|
||||
size_t len;
|
||||
if (args[2].IsArrayBuffer()) {
|
||||
ArrayBuffer ab = args[2].As<ArrayBuffer>();
|
||||
data = static_cast<char*>(ab.Data());
|
||||
len = ab.ByteLength();
|
||||
} else {
|
||||
TypedArray view = args[2].As<TypedArray>();
|
||||
data = static_cast<char*>(view.ArrayBuffer().Data()) + view.ByteOffset();
|
||||
len = view.ByteLength();
|
||||
}
|
||||
if (len > static_cast<size_t>(INT_MAX)) {
|
||||
RangeError::New(env, "Buffer is too large").ThrowAsJavaScriptException();
|
||||
return env.Undefined();
|
||||
}
|
||||
|
||||
Function cb;
|
||||
Promise promise = CreatePromiseAndCallback(env, cb);
|
||||
|
||||
auto execute_fn = [ctrl, path, ab, data, len](ResultAsyncWorker* worker) {
|
||||
(void)ab; // to keep ArrayBuffer alive
|
||||
auto execute_fn = [ctrl, path, data, len](ResultAsyncWorker* worker) {
|
||||
char* c_res = chat_write_file(ctrl, path.c_str(), data, static_cast<int>(len));
|
||||
HandleCResult(worker, c_res, "chat_write_file");
|
||||
};
|
||||
|
||||
ResultAsyncWorker* worker = new ResultAsyncWorker(cb, std::move(execute_fn));
|
||||
worker->KeepAlive(args[2].As<Object>());
|
||||
worker->Queue();
|
||||
|
||||
return promise;
|
||||
@@ -410,10 +637,19 @@ Value ChatDecryptFile(const CallbackInfo& args) {
|
||||
|
||||
Object Init(Env env, Object exports) {
|
||||
haskell_init();
|
||||
auto* receivers = new Receivers();
|
||||
// Stopping all receivers before joining any bounds teardown by the longest in-flight receive.
|
||||
env.AddCleanupHook([receivers]() {
|
||||
for (auto& entry : *receivers) {
|
||||
entry.second->RequestStop();
|
||||
}
|
||||
delete receivers;
|
||||
});
|
||||
exports.Set("chat_migrate_init", Function::New(env, ChatMigrateInit));
|
||||
exports.Set("chat_close_store", Function::New(env, ChatCloseStore));
|
||||
exports.Set("chat_migrate_init_queue", Function::New(env, ChatMigrateInitQueue));
|
||||
exports.Set("chat_close_store", Function::New(env, ChatCloseStore, "chat_close_store", receivers));
|
||||
exports.Set("chat_send_cmd", Function::New(env, ChatSendCmd));
|
||||
exports.Set("chat_recv_msg_wait", Function::New(env, ChatRecvMsgWait));
|
||||
exports.Set("chat_recv_msg_wait", Function::New(env, ChatRecvMsgWait, "chat_recv_msg_wait", receivers));
|
||||
exports.Set("chat_write_file", Function::New(env, ChatWriteFile));
|
||||
exports.Set("chat_read_file", Function::New(env, ChatReadFile));
|
||||
exports.Set("chat_encrypt_file", Function::New(env, ChatEncryptFile));
|
||||
|
||||
Reference in New Issue
Block a user