diff --git a/src/lib/sys.cpp b/src/lib/sys.cpp index 77a9bac..85fe0d1 100644 --- a/src/lib/sys.cpp +++ b/src/lib/sys.cpp @@ -21,6 +21,14 @@ int uce_host_task_spawn(const char* key, size_t key_len, uint64_t callback_id, d int uce_host_task_pid(const char* key, size_t key_len); int uce_host_task_kill(int pid, int sig); unsigned int uce_host_sleep_us(uint64_t usec); +uint64_t uce_host_socket_connect(const char* host, size_t host_len, int port); +void uce_host_socket_close(uint64_t sockfd); +int uce_host_socket_write(uint64_t sockfd, const char* data, size_t data_len); +size_t uce_host_socket_read(uint64_t sockfd, uint32_t max_length, uint32_t timeout, char* buf, size_t cap); +int uce_host_server_start_http(const char* key, size_t key_len, const char* bind, size_t bind_len, const char* file, size_t file_len, const char* function, size_t function_len, const char* current, size_t current_len); +int uce_host_server_stop(const char* key, size_t key_len); +size_t uce_host_memcache_command(uint64_t sockfd, const char* command, size_t command_len, char* buf, size_t cap); +size_t uce_host_mysql(const char* in, size_t in_len, char* out, size_t cap); } static String wasm_current_unit_file() @@ -137,10 +145,25 @@ String time_format_relative(u64 timestamp, String format_very_recent, u64 medium return(wasm_time_expand_delta(format_not_recent, timestamp, now_timestamp)); } u64 time_parse(String time_String) { char* end = 0; unsigned long long v = strtoull(time_String.c_str(), &end, 10); return(end && *end == 0 ? (u64)v : 0); } -u64 socket_connect(String host, short port) { (void)host; (void)port; return(0); } -void socket_close(u64 sockfd) { (void)sockfd; } -bool socket_write(u64 sockfd, String data) { (void)sockfd; (void)data; return(false); } -String socket_read(u64 sockfd, u32 max_length, u32 timeout) { (void)sockfd; (void)max_length; (void)timeout; return(""); } +u64 socket_connect(String host, short port) +{ + return(uce_host_socket_connect(host.data(), host.size(), port)); +} +void socket_close(u64 sockfd) { uce_host_socket_close(sockfd); } +bool socket_write(u64 sockfd, String data) +{ + return(uce_host_socket_write(sockfd, data.data(), data.size()) != 0); +} +String socket_read(u64 sockfd, u32 max_length, u32 timeout) +{ + size_t required = uce_host_socket_read(sockfd, max_length, timeout, 0, 0); + if(required == 0) + return(""); + String content(required, 0); + size_t got = uce_host_socket_read(sockfd, max_length, timeout, &content[0], required); + content.resize(got <= required ? got : 0); + return(content); +} String ws_message() { return(context ? context->in : ""); } String ws_connection_id() { return(context ? context->resources.websocket_connection_id : ""); } String ws_scope() { return(context ? context->resources.websocket_scope : ""); } @@ -156,12 +179,78 @@ String capture_backtrace_string(u32 max_frames, u32 skip_frames) { (void)max_fra String signal_name(int sig) { (void)sig; return(""); } String memcache_escape_key(String key) { return(key); } StringList memcache_escape_keys(StringList keys) { return(keys); } -u64 memcache_connect(String host, short port) { (void)host; (void)port; return(0); } -String memcache_command(u64 connection, String command) { (void)connection; (void)command; return(""); } -bool memcache_set(u64 connection, String key, String value, u64 expires_in) { (void)connection; (void)key; (void)value; (void)expires_in; return(false); } -bool memcache_delete(u64 connection, String key) { (void)connection; (void)key; return(false); } -String memcache_get(u64 connection, String key, String default_value) { (void)connection; (void)key; return(default_value); } -StringMap memcache_get_multiple(u64 connection, StringList keys) { (void)connection; (void)keys; return(StringMap()); } +u64 memcache_connect(String host, short port) +{ + if(host == "") + host = "127.0.0.1"; + if(port == 0) + port = 11211; + return(socket_connect(host, port)); +} +String memcache_command(u64 connection, String command) +{ + size_t required = uce_host_memcache_command(connection, command.data(), command.size(), 0, 0); + if(required == 0) + return(""); + String content(required, 0); + size_t got = uce_host_memcache_command(connection, command.data(), command.size(), &content[0], required); + content.resize(got <= required ? got : 0); + return(content); +} +bool memcache_set(u64 connection, String key, String value, u64 expires_in) +{ + String response = memcache_command( + connection, + String("set ") + memcache_escape_key(key) + " 0 " + std::to_string(expires_in) + " " + std::to_string(value.length()) + "\r\n" + value + ); + return("STORED" == trim(response)); +} +bool memcache_delete(u64 connection, String key) +{ + String response = memcache_command(connection, String("delete ") + memcache_escape_key(key)); + return("DELETED" == trim(response)); +} +String memcache_get(u64 connection, String key, String default_value) +{ + String res = memcache_command(connection, String("get ") + memcache_escape_key(key)); + String header_end = "\r\n"; + size_t header_pos = res.find(header_end); + if(res.rfind("VALUE ", 0) != 0 || header_pos == String::npos) + return(default_value); + String header = res.substr(0, header_pos); + StringList parts = split(header, " "); + if(parts.size() < 4) + return(default_value); + u64 length = int_val(parts[3]); + size_t data_pos = header_pos + header_end.size(); + if(res.size() < data_pos + length) + return(default_value); + return(res.substr(data_pos, length)); +} +StringMap memcache_get_multiple(u64 connection, StringList keys) +{ + StringMap result; + String res = memcache_command(connection, String("get ") + join(memcache_escape_keys(keys), " ")); + while(res.rfind("VALUE ", 0) == 0) + { + size_t header_pos = res.find("\r\n"); + if(header_pos == String::npos) + break; + StringList parts = split(res.substr(0, header_pos), " "); + if(parts.size() < 4) + break; + String key = parts[1]; + u64 length = int_val(parts[3]); + size_t data_pos = header_pos + 2; + if(res.size() < data_pos + length) + break; + result[key] = res.substr(data_pos, length); + res = res.substr(data_pos + length); + if(res.rfind("\r\n", 0) == 0) + res = res.substr(2); + } + return(result); +} void on_segfault(int sig) { (void)sig; } static u64 wasm_next_task_callback_id = 1; @@ -195,8 +284,18 @@ pid_t task_repeat(String key, f64 interval, std::function exec_after_spa pid_t task_pid(String key) { return((pid_t)uce_host_task_pid(key.data(), key.size())); } extern "C" unsigned int sleep(unsigned int seconds) { return(uce_host_sleep_us((uint64_t)seconds * 1000000ull)); } extern "C" int usleep(unsigned int usec) { uce_host_sleep_us(usec); return(0); } -pid_t server_start_http(String key, String socket_fn_or_port, String call_uce_filename, String call_function) { (void)key; (void)socket_fn_or_port; (void)call_uce_filename; (void)call_function; return(0); } -bool server_stop(String key) { (void)key; return(false); } +pid_t server_start_http(String key, String socket_fn_or_port, String call_uce_filename, String call_function) +{ + String current = wasm_current_unit_file(); + return((pid_t)uce_host_server_start_http( + key.data(), key.size(), + socket_fn_or_port.data(), socket_fn_or_port.size(), + call_uce_filename.data(), call_uce_filename.size(), + call_function.data(), call_function.size(), + current.data(), current.size() + )); +} +bool server_stop(String key) { return(uce_host_server_stop(key.data(), key.size()) != 0); } StringMap default_config() { StringMap cfg; diff --git a/src/wasm/backend.cpp b/src/wasm/backend.cpp index 12dca80..212cebf 100644 --- a/src/wasm/backend.cpp +++ b/src/wasm/backend.cpp @@ -117,14 +117,13 @@ static bool wasm_backend_native_fallback_uncached(Request* context, const String // - regex_* (host PCRE2 hostcall), xml_*/yaml_*/markdown_* (compiled in), // - unit_render()/component() (host resolver). // What remains is genuinely host-owned / native-only for now: - // - zip, sockets/custom servers, memcache, mysql; - // - unit_call + compiler/unit introspection (need the native toolchain). - // Background tasks and sleep/usleep are host-owned but now have membrane - // hostcalls, so they intentionally do not fallback here. + // - zip pages that use C++ try/catch around zip_* error cases; + // - direct compiler_load_shared_unit() access to native SharedUnit internals. + // Background tasks, sleep/usleep, sockets/custom servers, memcache, mysql, + // and unit_call/unit_compile/unit_info/units_list are host-owned but now have + // membrane hostcalls, so they intentionally do not fallback here. StringList native_only_tokens = { - "zip_", "socket_", "server_start_http(", "server_stop(", - "memcache_", "mysql_", "unit_call(", - "unit_compile(", "unit_info(", "units_list(", "compiler_load_shared_unit(" + "zip_", "compiler_load_shared_unit(" }; for(auto& token : native_only_tokens) if(source.find(token) != String::npos) diff --git a/src/wasm/core.cpp b/src/wasm/core.cpp index b336c42..7a1d382 100644 --- a/src/wasm/core.cpp +++ b/src/wasm/core.cpp @@ -10,32 +10,323 @@ #include "../lib/mysql-connector.h" #include "../lib/sqlite-connector.h" -// ---- W3 connector membrane stubs ------------------------------------------- -// Generated units reference the connector class surface through uce_lib.h, so -// the core must define it. Until the real hostcall connectors land (W5 parity -// scope: the starter uses no database), every operation fails cleanly with an -// explanatory error instead of trapping. +// ---- W3 connector membranes ----------------------------------------------- +// sqlite/mysql run host-side (the host links the native connectors and owns the +// connections in per-workspace handle tables). UCEB1-marshalled hostcalls carry +// operation requests/responses; `connection` holds the host handle (>0). static const char* WASM_DB_UNAVAILABLE = - "database connectors are not yet available in the wasm workspace"; + "database connector is not available in the wasm workspace"; + +extern "C" size_t uce_host_mysql(const char* in, size_t in_len, char* out, size_t cap); +extern "C" size_t uce_host_zip(const char* in, size_t in_len, char* out, size_t cap); +extern "C" size_t uce_host_units(const char* in, size_t in_len, char* out, size_t cap); + +static DValue wasm_sized_hostcall(DValue request, size_t (*hostcall)(const char*, size_t, char*, size_t)) +{ + String encoded = ucb_encode(request); + size_t need = hostcall(encoded.data(), encoded.size(), 0, 0); + if(need == 0) + return(DValue()); + String buffer(need, 0); + size_t got = hostcall(encoded.data(), encoded.size(), &buffer[0], need); + if(got == 0 || got > need) + return(DValue()); + DValue response; + String error; + ucb_decode(String(buffer.data(), got), response, &error); + return(response); +} + +static DValue wasm_zip_call(DValue request) +{ + DValue response = wasm_sized_hostcall(request, uce_host_zip); + return(response); +} + +DValue zip_list(String zip_file_name) +{ + DValue request; + request["op"] = "list"; + request["path"] = zip_file_name; + DValue response = wasm_zip_call(request); + DValue* result = response.key("result"); + return(result ? *result : DValue()); +} + +String zip_read(String zip_file_name, String entry_name) +{ + DValue request; + request["op"] = "read"; + request["path"] = zip_file_name; + request["entry"] = entry_name; + DValue response = wasm_zip_call(request); + return(response["result"].to_string()); +} + +bool zip_create(String zip_file_name, DValue entries) +{ + DValue request; + request["op"] = "create"; + request["path"] = zip_file_name; + request["entries"] = entries; + DValue response = wasm_zip_call(request); + return(response["ok"].to_bool()); +} + +bool zip_extract(String zip_file_name, String destination_directory) +{ + DValue request; + request["op"] = "extract"; + request["path"] = zip_file_name; + request["destination"] = destination_directory; + DValue response = wasm_zip_call(request); + return(response["ok"].to_bool()); +} + +String gz_compress(String src) +{ + DValue request; + request["op"] = "gz_compress"; + request["src"] = src; + DValue response = wasm_zip_call(request); + return(response["result"].to_string()); +} + +String gz_uncompress(String compressed) +{ + DValue request; + request["op"] = "gz_uncompress"; + request["src"] = compressed; + DValue response = wasm_zip_call(request); + return(response["result"].to_string()); +} + +static DValue wasm_units_call(DValue request) +{ + return(wasm_sized_hostcall(request, uce_host_units)); +} + +DValue unit_info(String path) +{ + DValue request; + request["op"] = "info"; + request["path"] = path; + DValue response = wasm_units_call(request); + DValue* result = response.key("result"); + return(result ? *result : DValue()); +} + +StringList units_list() +{ + DValue request; + request["op"] = "list"; + DValue response = wasm_units_call(request); + StringList result; + DValue* items = response.key("result"); + if(items) + items->each([&](const DValue& value, String) { result.push_back(value.to_string()); }); + return(result); +} + +bool unit_compile(String path) +{ + DValue request; + request["op"] = "compile"; + request["path"] = path; + DValue response = wasm_units_call(request); + return(response["ok"].to_bool()); +} + +static DValue wasm_unit_call_result; +DValue* unit_call(String file_name, String function_name, DValue* call_param) +{ + DValue request; + request["op"] = "call"; + request["file"] = file_name; + request["function"] = function_name; + if(call_param) + request["param"] = *call_param; + DValue response = wasm_units_call(request); + String output = response["output"].to_string(); + if(output != "") + print(output); + DValue* result = response.key("result"); + if(result) + { + wasm_unit_call_result = *result; + return(&wasm_unit_call_result); + } + return(0); +} + +SharedUnit* compiler_load_shared_unit(Request* context, String file_name, String current_path, bool opt_so_optional) +{ + (void)context; (void)file_name; (void)current_path; (void)opt_so_optional; + return(0); +} + +static DValue wasm_mysql_call(DValue request) +{ + return(wasm_sized_hostcall(request, uce_host_mysql)); +} bool MySQL::connect(String host, String username, String password) { - (void)host; (void)username; (void)password; - connection = 0; - statement_info = WASM_DB_UNAVAILABLE; + DValue request; + request["op"] = "connect"; + request["host"] = host; + request["username"] = username; + request["password"] = password; + DValue response = wasm_mysql_call(request); + u64 handle = response["handle"].to_u64(); + connection = (void*)(uintptr_t)handle; + _preload_next_error_code = (u32)response["error_code"].to_u64(); + statement_info = response["statement_info"].to_string(); + return(handle != 0 && _preload_next_error_code == 0); +} + +void MySQL::disconnect() +{ + if(connection) + { + DValue request; + request["op"] = "disconnect"; + request["handle"] = (f64)(uintptr_t)connection; + wasm_mysql_call(request); + connection = 0; + } +} + +String MySQL::error() +{ + String result = statement_info; + statement_info = ""; + _preload_next_error_code = 0; + return(result); +} + +String MySQL::escape(String raw, char quote_char) { return(mysql_escape(raw, quote_char)); } + +String mysql_escape(String raw, char quote_char) +{ + DValue request; + request["op"] = "escape"; + request["raw"] = raw; + request["quote_char"] = String(quote_char > 0 ? 1 : 0, quote_char); + DValue response = wasm_mysql_call(request); + DValue* result = response.key("result"); + return(result ? result->to_string() : raw); +} + +String MySQL::parse_query_parameters(String query, StringMap map) +{ + String result; + query.append(1, ' '); + + u8 mode = 0; + char quote = 0; + String identifier; + for(u32 i = 0; i < query.length(); i++) + { + char c = query[i]; + if(mode == 0) + { + if(c == ':') + { + mode = 1; + identifier = ""; + } + else if(c == '"' || c == '\'') + { + result.append(1, c); + mode = 2; + quote = c; + } + else + result.append(1, c); + } + else if(mode == 1) + { + if(isalnum(c)) + identifier.append(1, c); + else + { + result.append(escape(map[identifier])); + result.append(1, c); + mode = 0; + } + } + else if(mode == 2) + { + if(c == quote) + mode = 0; + result.append(1, c); + } + } + + return(result); +} + +static bool wasm_mysql_has_unquoted_positional_placeholder(String query) +{ + bool quoted = false; + char quote = 0; + bool escaped = false; + for(u32 i = 0; i < query.length(); i++) + { + char c = query[i]; + if(quoted) + { + if(escaped) + { + escaped = false; + continue; + } + if(c == '\\') + { + escaped = true; + continue; + } + if(c == quote) + quoted = false; + continue; + } + if(c == '\'' || c == '"') + { + quoted = true; + quote = c; + continue; + } + if(c == '?') + return(true); + } return(false); } -void MySQL::disconnect() { connection = 0; } -String MySQL::error() { return(WASM_DB_UNAVAILABLE); } -String MySQL::escape(String raw, char quote_char) { (void)quote_char; return(raw); } -String MySQL::parse_query_parameters(String query, StringMap m) { (void)m; return(query); } -DValue MySQL::query(String q) { (void)q; statement_info = WASM_DB_UNAVAILABLE; return(DValue()); } -DValue MySQL::query(String q, StringMap params) { (void)q; (void)params; statement_info = WASM_DB_UNAVAILABLE; return(DValue()); } -DValue MySQL::get_pending_result() { return(DValue()); } +DValue MySQL::query(String q) +{ + if(wasm_mysql_has_unquoted_positional_placeholder(q)) + { + _preload_next_error_code = 2000; + statement_info = "mysql positional ? placeholders are not supported; use named :name placeholders"; + return(DValue()); + } + DValue request; + request["op"] = "query"; + request["handle"] = (f64)(uintptr_t)connection; + request["query"] = q; + DValue response = wasm_mysql_call(request); + insert_id = response["insert_id"].to_u64(); + affected_rows = (u32)response["affected"].to_u64(); + _preload_next_error_code = (u32)response["error_code"].to_u64(); + statement_info = response["statement_info"].to_string(); + DValue* result = response.key("result"); + return(result ? *result : DValue()); +} -String mysql_escape(String raw, char quote_char) { (void)quote_char; return(raw); } +DValue MySQL::query(String q, StringMap params) { return(query(parse_query_parameters(q, params))); } +DValue MySQL::get_pending_result() { return(DValue()); } // sqlite runs host-side (the host links libsqlite and owns the connections in // a per-workspace handle table). One UCEB1-marshalled hostcall carries @@ -129,6 +420,7 @@ DValue sqlite_query(SQLite* db, String q, const StringMap& params) { return(db ? u64 sqlite_insert_id(SQLite* db) { return(db ? db->insert_id : 0); } u32 sqlite_affected_rows(SQLite* db) { return(db ? db->affected_rows : 0); } void cleanup_sqlite_connections() { } +void cleanup_mysql_connections() { } static ServerState wasm_server; static Request wasm_request; diff --git a/src/wasm/core_hostcalls.syms b/src/wasm/core_hostcalls.syms index 2540d2c..947337d 100644 --- a/src/wasm/core_hostcalls.syms +++ b/src/wasm/core_hostcalls.syms @@ -7,6 +7,16 @@ uce_host_task_spawn uce_host_task_pid uce_host_task_kill uce_host_sleep_us +uce_host_socket_connect +uce_host_socket_close +uce_host_socket_write +uce_host_socket_read +uce_host_server_start_http +uce_host_server_stop +uce_host_memcache_command +uce_host_mysql +uce_host_zip +uce_host_units uce_host_component_resolve uce_host_file_exists uce_host_file_read diff --git a/src/wasm/worker.cpp b/src/wasm/worker.cpp index 0b07f50..84655d2 100644 --- a/src/wasm/worker.cpp +++ b/src/wasm/worker.cpp @@ -35,6 +35,9 @@ #include #include #include +#include +#include +#include #include #include #include @@ -345,11 +348,15 @@ public: // Host-owned resource handle table (ยง3.1): connections opened by the guest // live here and are closed when the workspace drops at request end. std::vector sqlite_handles; + std::vector mysql_handles; ~WasmWorkspace() { for(auto* db : sqlite_handles) if(db) delete db; // ~SQLite disconnects + for(auto* db : mysql_handles) + if(db) + delete db; // ~MySQL disconnects } #endif @@ -359,6 +366,10 @@ public: // input bytes) and replayed on the second without re-running the op. String staged_hostcall_input; String staged_hostcall_result; + String staged_socket_read_key; + String staged_socket_read_result; + String staged_memcache_key; + String staged_memcache_result; bool hostcall_staged(const String& input, String& out) { if(!staged_hostcall_input.empty() && input == staged_hostcall_input) @@ -1193,6 +1204,120 @@ private: results[0] = Val(ok ? (int32_t)1 : (int32_t)0); return(std::monostate()); })); + if(mod == "env" && name == "uce_host_zip") + return(add([self](Caller, Span args, Span results) -> Result { + String encoded; + self->hostcall_read(args[0].i32(), args[1].i32(), encoded); + String out; + if(!self->hostcall_staged(encoded, out)) + { + DValue request, response; + String decode_error; + try + { + if(ucb_decode(encoded, request, &decode_error)) + { + String op = request["op"].to_string(); + if(op == "list") + { + String path = self->resolve_guest_file(request["path"].to_string()); + if(path == "") throw std::runtime_error("zip_list: path is outside wasm file policy"); + response["result"] = zip_list(path); + } + else if(op == "read") + { + String path = self->resolve_guest_file(request["path"].to_string()); + if(path == "") throw std::runtime_error("zip_read: path is outside wasm file policy"); + response["result"] = zip_read(path, request["entry"].to_string()); + } + else if(op == "create") + { + String path = self->resolve_guest_write(request["path"].to_string(), ""); + if(path == "") throw std::runtime_error("zip_create: path is outside wasm file policy"); + DValue* entries = request.key("entries"); + response["ok"].set_bool(zip_create(path, entries ? *entries : DValue())); + } + else if(op == "extract") + { + String path = self->resolve_guest_file(request["path"].to_string()); + String destination = self->resolve_guest_write(request["destination"].to_string(), ""); + if(path == "" || destination == "") throw std::runtime_error("zip_extract: path is outside wasm file policy"); + response["ok"].set_bool(zip_extract(path, destination)); + } + else if(op == "gz_compress") + response["result"] = gz_compress(request["src"].to_string()); + else if(op == "gz_uncompress") + response["result"] = gz_uncompress(request["src"].to_string()); + } + } + catch(const std::exception& e) + { + response["error"] = e.what(); + } + out = ucb_encode(response); + self->hostcall_stage(encoded, out); + } + u32 cap = (u32)args[3].i32(); + int32_t buf = args[2].i32(); + if(buf != 0 && cap >= out.size()) + self->hostcall_write(buf, out); + results[0] = Val((int32_t)out.size()); + return(std::monostate()); + })); + if(mod == "env" && name == "uce_host_units") + return(add([self](Caller caller, Span args, Span results) -> Result { + String encoded; + self->hostcall_read(args[0].i32(), args[1].i32(), encoded); + String out; + if(!self->hostcall_staged(encoded, out)) + { + DValue request, response; + String decode_error; + try + { + if(ucb_decode(encoded, request, &decode_error)) + { + String op = request["op"].to_string(); + if(op == "info") + response["result"] = unit_info(request["path"].to_string()); + else if(op == "list") + { + StringList paths = units_list(); + for(auto& path : paths) + { + DValue item; + item = path; + response["result"].push(item); + } + } + else if(op == "compile") + response["ok"].set_bool(unit_compile(request["path"].to_string())); + else if(op == "call") + { + DValue* param = request.key("param"); + ob_start(); + DValue* result = unit_call(request["file"].to_string(), request["function"].to_string(), param); + response["output"] = ob_get_close(); + if(result) + response["result"] = *result; + } + } + } + catch(const std::exception& e) + { + response["error"] = e.what(); + } + out = ucb_encode(response); + self->hostcall_stage(encoded, out); + } + u32 cap = (u32)args[3].i32(); + int32_t buf = args[2].i32(); + if(buf != 0 && cap >= out.size()) + self->hostcall_write(buf, out); + caller.context().set_epoch_deadline(self->worker.cfg.epoch_deadline_ticks); + results[0] = Val((int32_t)out.size()); + return(std::monostate()); + })); #ifdef UCE_WASM_HOST_CONNECTORS if(mod == "env" && name == "uce_host_sqlite") return(add([self](Caller, Span args, Span results) -> Result { @@ -1260,6 +1385,99 @@ private: results[0] = Val((int32_t)out.size()); return(std::monostate()); })); + if(mod == "env" && name == "uce_host_memcache_command") + return(add([self](Caller caller, Span args, Span results) -> Result { + String command; + self->hostcall_read(args[1].i32(), args[2].i32(), command); + String key = "memcache:" + std::to_string((u64)args[0].i64()) + ":" + command; + u32 cap = (u32)args[4].i32(); + int32_t buf = args[3].i32(); + String out; + if(buf != 0 && self->staged_memcache_key == key) + { + out = self->staged_memcache_result; + self->staged_memcache_key = ""; + self->staged_memcache_result = ""; + } + else + { + ::socket_write((u64)args[0].i64(), command + "\r\n"); + out = ::socket_read((u64)args[0].i64()); + if(buf == 0) + { + self->staged_memcache_key = key; + self->staged_memcache_result = out; + } + } + if(buf != 0 && cap >= out.size()) + self->hostcall_write(buf, out); + caller.context().set_epoch_deadline(self->worker.cfg.epoch_deadline_ticks); + results[0] = Val((int32_t)out.size()); + return(std::monostate()); + })); + if(mod == "env" && name == "uce_host_mysql") + return(add([self](Caller, Span args, Span results) -> Result { + String encoded; + self->hostcall_read(args[0].i32(), args[1].i32(), encoded); + String out; + if(!self->hostcall_staged(encoded, out)) + { + DValue request, response; + String decode_error; + if(ucb_decode(encoded, request, &decode_error)) + { + String op = request["op"].to_string(); + if(op == "connect") + { + MySQL* db = new MySQL(); + bool ok = db->connect(request["host"].to_string(), request["username"].to_string(), request["password"].to_string()); + u64 handle = 0; + if(ok && db->connection) + { + self->mysql_handles.push_back(db); + handle = self->mysql_handles.size(); + } + response["handle"] = (f64)handle; + response["error_code"] = (f64)db->_preload_next_error_code; + response["statement_info"] = db->error(); + if(handle == 0) + delete db; + } + else if(op == "escape") + { + String quote = request["quote_char"].to_string(); + response["result"] = mysql_escape(request["raw"].to_string(), quote.size() ? quote[0] : 0); + } + else + { + u64 handle = request["handle"].to_u64(); + MySQL* db = (handle >= 1 && handle <= self->mysql_handles.size()) + ? self->mysql_handles[(size_t)handle - 1] : 0; + if(op == "query" && db) + { + response["result"] = db->query(request["query"].to_string()); + response["insert_id"] = (f64)db->insert_id; + response["affected"] = (f64)db->affected_rows; + response["error_code"] = (f64)db->_preload_next_error_code; + response["statement_info"] = db->error(); + } + else if(op == "disconnect" && db) + { + delete db; + self->mysql_handles[(size_t)handle - 1] = 0; + } + } + } + out = ucb_encode(response); + self->hostcall_stage(encoded, out); + } + u32 cap = (u32)args[3].i32(); + int32_t buf = args[2].i32(); + if(buf != 0 && cap >= out.size()) + self->hostcall_write(buf, out); + results[0] = Val((int32_t)out.size()); + return(std::monostate()); + })); #endif if(mod == "env" && name == "uce_host_file_unlink") return(add([self](Caller, Span args, Span) -> Result { @@ -1271,6 +1489,110 @@ private: ::unlink(resolved.c_str()); return(std::monostate()); })); + if(mod == "env" && name == "uce_host_socket_connect") + return(add([self](Caller caller, Span args, Span results) -> Result { + String host; + self->hostcall_read(args[0].i32(), args[1].i32(), host); + int fd = ::socket(AF_INET, SOCK_STREAM, IPPROTO_TCP); + if(fd >= 0) + { + struct sockaddr_in addr = {0}; + addr.sin_family = AF_INET; + addr.sin_port = htons((short)args[2].i32()); + addr.sin_addr.s_addr = inet_addr(host.c_str()); + if(::connect(fd, (struct sockaddr*)&addr, sizeof(addr)) < 0) + { + ::close(fd); + fd = -1; + } + else if(fd == 0) + { + int moved = ::dup(fd); + ::close(fd); + fd = moved; + } + if(fd > 0 && context) + context->resources.sockets.push_back(fd); + } + results[0] = Val((int64_t)(fd > 0 ? fd : 0)); + caller.context().set_epoch_deadline(self->worker.cfg.epoch_deadline_ticks); + return(std::monostate()); + })); + if(mod == "env" && name == "uce_host_socket_close") + return(add([](Caller, Span args, Span) -> Result { + ::socket_close((u64)args[0].i64()); + return(std::monostate()); + })); + if(mod == "env" && name == "uce_host_socket_write") + return(add([self](Caller, Span args, Span results) -> Result { + String data; + self->hostcall_read(args[1].i32(), args[2].i32(), data); + results[0] = Val(::socket_write((u64)args[0].i64(), data) ? (int32_t)1 : (int32_t)0); + return(std::monostate()); + })); + if(mod == "env" && name == "uce_host_socket_read") + return(add([self](Caller caller, Span args, Span results) -> Result { + u64 sockfd = (u64)args[0].i64(); + u32 max_length = (u32)args[1].i32(); + u32 timeout = (u32)args[2].i32(); + int32_t buf = args[3].i32(); + u32 cap = (u32)args[4].i32(); + String key = std::to_string(sockfd) + ":" + std::to_string(max_length) + ":" + std::to_string(timeout); + String out; + if(buf != 0 && self->staged_socket_read_key == key) + { + out = self->staged_socket_read_result; + self->staged_socket_read_key = ""; + self->staged_socket_read_result = ""; + } + else + { + out = ::socket_read(sockfd, max_length, timeout); + if(buf == 0) + { + self->staged_socket_read_key = key; + self->staged_socket_read_result = out; + } + } + if(buf != 0 && cap >= out.size()) + self->hostcall_write(buf, out); + caller.context().set_epoch_deadline(self->worker.cfg.epoch_deadline_ticks); + results[0] = Val((int32_t)out.size()); + return(std::monostate()); + })); + if(mod == "env" && name == "uce_host_server_start_http") + return(add([self](Caller, Span args, Span results) -> Result { + String key, bind, file, function, current; + self->hostcall_read(args[0].i32(), args[1].i32(), key); + self->hostcall_read(args[2].i32(), args[3].i32(), bind); + self->hostcall_read(args[4].i32(), args[5].i32(), file); + self->hostcall_read(args[6].i32(), args[7].i32(), function); + self->hostcall_read(args[8].i32(), args[9].i32(), current); + String resolved = self->resolve_guest_file(file, current); + pid_t pid = 0; + try + { + if(resolved != "") + pid = ::server_start_http(key, bind, resolved, function); + } + catch(const std::exception& e) + { + fprintf(stderr, "[wasm server] start failed for key '%s': %s\n", key.c_str(), e.what()); + } + catch(...) + { + fprintf(stderr, "[wasm server] start failed for key '%s'\n", key.c_str()); + } + results[0] = Val((int32_t)pid); + return(std::monostate()); + })); + if(mod == "env" && name == "uce_host_server_stop") + return(add([self](Caller, Span args, Span results) -> Result { + String key; + self->hostcall_read(args[0].i32(), args[1].i32(), key); + results[0] = Val(::server_stop(key) ? (int32_t)1 : (int32_t)0); + return(std::monostate()); + })); if(mod == "env" && name == "uce_host_task_spawn") return(add([self](Caller, Span args, Span results) -> Result { String key;