diff --git a/docs/wasm-runtime-architecture.md b/docs/wasm-runtime-architecture.md index 6878738..910e0eb 100644 --- a/docs/wasm-runtime-architecture.md +++ b/docs/wasm-runtime-architecture.md @@ -214,6 +214,14 @@ hostcall loop therefore cannot renew an invocation indefinitely. Blocking host helpers retain operation-specific limits and cap them to the remaining invocation budget where the underlying operation is cancellable. Forked task callbacks receive a fresh invocation deadline capped by the task lifetime. +The Memcached ASCII hostcall applies one deadline to the complete command write +and protocol-framed response read. Retrieval payload lengths are parsed, so a +payload containing `END` cannot terminate the response; multi-value and `stats` +responses may span socket chunks up to the 8 MiB response bound. Timeout, +malformed framing, or overflow shuts down the retained descriptor before +returning an empty result, preventing unread response bytes from poisoning the +next command on that connection. The native fallback uses the same parser and +deadline contract. Synchronous compiler locks, transitive `#load` compilation, and compiler child processes consume that same deadline. Compiler children run in a dedicated process group; timeout kills the group, retains the previous generation, and diff --git a/scripts/test_wasm_compile_timeout.sh b/scripts/test_wasm_compile_timeout.sh index 9361c9a..c5be09e 100755 --- a/scripts/test_wasm_compile_timeout.sh +++ b/scripts/test_wasm_compile_timeout.sh @@ -45,7 +45,7 @@ HTTP_PORT= HTTP_DOCUMENT_ROOT=$site SESSION_PATH=$root/session TMP_UPLOAD_PATH=$root/upload -WASM_CORE_PATH=$(pwd)/bin/wasm/core.wasm +WASM_CORE_PATH=$(pwd -P)/bin/wasm/core.wasm WASM_COMPILE_SCRIPT=$root/compile WASM_INVOCATION_TIMEOUT_MS=2000 WASM_EPOCH_PERIOD_MS=20 diff --git a/site/doc/pages/memcache_command.txt b/site/doc/pages/memcache_command.txt index 2cba7dd..70b423e 100644 --- a/site/doc/pages/memcache_command.txt +++ b/site/doc/pages/memcache_command.txt @@ -13,6 +13,7 @@ return value : string containing the Memcache server's response Executes a raw command on an open Memcache connection and returns the server response as a string. This is the low-level escape hatch for Memcache operations that are not covered by the dedicated helpers. +UCE reads the complete framed ASCII response, including multi-value retrievals and `stats`, rather than returning a socket chunk. The complete write and read share a one-second deadline and responses are limited to 8 MiB. A timeout, malformed frame, or oversized response returns an empty string and makes that connection unusable so unread bytes cannot affect a later command. Commands ending in `noreply` and `quit` return immediately with an empty string. :example u64 conn = memcache_connect(); diff --git a/site/tests/services.uce b/site/tests/services.uce index 0fbe4d5..ad3b132 100644 --- a/site/tests/services.uce +++ b/site/tests/services.uce @@ -90,10 +90,25 @@ RENDER(Request& context) StringMap mem_multi = memcache_get_multiple(memfd, {"site-tests-key", "site-tests-key2"}); memcache_delete(memfd, "site-tests-key"); memcache_delete(memfd, "site-tests-key2"); + String large_mem_value(200 * 1024, 'x'); + large_mem_value.replace(1024, 7, "\r\nEND\r\n"); + bool large_mem_stored = memcache_set(memfd, "site-tests-large", large_mem_value); + String large_mem_loaded = memcache_get(memfd, "site-tests-large"); + String second_large_mem_value(160 * 1024, 'y'); + bool second_large_mem_stored = memcache_set(memfd, "site-tests-large-2", second_large_mem_value); + StringMap large_mem_multi = memcache_get_multiple(memfd, {"site-tests-large", "site-tests-large-2"}); + String no_reply = memcache_command(memfd, "set site-tests-noreply 0 60 5 noreply\r\nvalue"); + String no_reply_value = memcache_get(memfd, "site-tests-noreply"); + String stats_mem = memcache_command(memfd, "stats"); + String stats_reset_mem = memcache_command(memfd, "stats reset"); + String after_stats_reset = memcache_get(memfd, "site-tests-noreply"); + bool large_mem_deleted = memcache_delete(memfd, "site-tests-large"); + bool second_large_mem_deleted = memcache_delete(memfd, "site-tests-large-2"); + bool no_reply_deleted = memcache_delete(memfd, "site-tests-noreply"); mark( "memcache_connect() / memcache_command() / memcache_get_multiple()", - mem_value == "value-1" && contains(raw_mem, "VALUE site-tests-key") && mem_multi["site-tests-key"] == "value-1" && mem_multi["site-tests-key2"] == "value-2" ? "pass" : "fail", - "memcache value=" + mem_value + " raw=" + raw_mem + " multi=" + var_dump(mem_multi) + mem_value == "value-1" && contains(raw_mem, "VALUE site-tests-key") && mem_multi["site-tests-key"] == "value-1" && mem_multi["site-tests-key2"] == "value-2" && large_mem_stored && large_mem_loaded == large_mem_value && second_large_mem_stored && large_mem_multi["site-tests-large"] == large_mem_value && large_mem_multi["site-tests-large-2"] == second_large_mem_value && no_reply == "" && no_reply_value == "value" && contains(stats_mem, "STAT ") && trim(stats_reset_mem) == "RESET" && after_stats_reset == "value" && large_mem_deleted && second_large_mem_deleted && no_reply_deleted ? "pass" : "fail", + "memcache value=" + mem_value + " raw=" + raw_mem + " multi=" + var_dump(mem_multi) + " large=" + std::to_string(large_mem_loaded.size()) + " multi-large=" + std::to_string(large_mem_multi["site-tests-large"].size() + large_mem_multi["site-tests-large-2"].size()) + " noreply=" + no_reply_value + " stats=" + std::to_string(stats_mem.size()) + "/" + trim(stats_reset_mem) + " after-reset=" + after_stats_reset + " deleted=" + (large_mem_deleted && second_large_mem_deleted && no_reply_deleted ? "true" : "false") ); } diff --git a/src/lib/sys.cpp b/src/lib/sys.cpp index 9af607c..55fe492 100644 --- a/src/lib/sys.cpp +++ b/src/lib/sys.cpp @@ -737,6 +737,7 @@ StringMap default_config() #include #include #include +#include #include #include "sys.h" #include "hash.h" @@ -1291,6 +1292,185 @@ String socket_read(u64 sockfd, u32 max_length, u32 timeout) return(""); } +static bool memcache_decimal(String value, u64& result) +{ + if(value == "") + return(false); + result = 0; + for(char c : value) + { + if(c < '0' || c > '9' || result > (UINT64_MAX - (u64)(c - '0')) / 10) + return(false); + result = result * 10 + (u64)(c - '0'); + } + return(true); +} + +static String memcache_command_line(const String& command) +{ + size_t end = command.find("\r\n"); + return(trim(command.substr(0, end))); +} + +bool uce_memcache_command_has_no_reply(const String& command) +{ + StringList words = split(memcache_command_line(command), " "); + if(words.size() == 0) + return(false); + String verb = to_lower(words[0]); + return(verb == "quit" || to_lower(words.back()) == "noreply"); +} + +MemcacheResponseState uce_memcache_response_advance(const String& command, const String& response, MemcacheResponseParser& parser) +{ + while(true) + { + if(parser.mode == MemcacheResponseParser::Mode::FirstLine) + { + size_t line_end = response.find("\r\n"); + if(line_end == String::npos) + return(MemcacheResponseState::Incomplete); + String line = response.substr(0, line_end); + if(line == "ERROR" || line.rfind("CLIENT_ERROR", 0) == 0 || line.rfind("SERVER_ERROR", 0) == 0 || line == "RESET" || line == "OK") + return(MemcacheResponseState::Complete); + if(line.rfind("VALUE ", 0) == 0) + parser.mode = MemcacheResponseParser::Mode::RetrievalHeader; + else if(line.rfind("VA ", 0) == 0) + { + StringList fields = split(line, " "); + u64 length = 0; + if(fields.size() < 2 || !memcache_decimal(fields[1], length) || length > SIZE_MAX - line_end - 4) + return(MemcacheResponseState::Malformed); + parser.data_end = line_end + 2 + (size_t)length; + parser.mode = MemcacheResponseParser::Mode::MetaData; + } + else + { + StringList request_words = split(memcache_command_line(command), " "); + String request_verb = request_words.size() ? to_lower(request_words[0]) : String(""); + bool end_terminated = line.rfind("STAT ", 0) == 0 || line.rfind("ITEM ", 0) == 0 || + request_verb == "stats" || (request_verb == "lru_crawler" && request_words.size() > 1 && to_lower(request_words[1]) == "metadump"); + if(!end_terminated) + return(MemcacheResponseState::Complete); + parser.mode = MemcacheResponseParser::Mode::Lines; + } + } + + if(parser.mode == MemcacheResponseParser::Mode::RetrievalHeader) + { + size_t line_end = response.find("\r\n", parser.offset); + if(line_end == String::npos) + return(MemcacheResponseState::Incomplete); + String line = response.substr(parser.offset, line_end - parser.offset); + if(line == "END") + return(MemcacheResponseState::Complete); + StringList fields = split(line, " "); + u64 length = 0; + if(fields.size() < 4 || fields[0] != "VALUE" || !memcache_decimal(fields[3], length) || length > SIZE_MAX - line_end - 4) + return(MemcacheResponseState::Malformed); + parser.data_end = line_end + 2 + (size_t)length; + parser.mode = MemcacheResponseParser::Mode::RetrievalData; + } + + if(parser.mode == MemcacheResponseParser::Mode::RetrievalData) + { + if(response.size() < parser.data_end + 2) + return(MemcacheResponseState::Incomplete); + if(response.compare(parser.data_end, 2, "\r\n") != 0) + return(MemcacheResponseState::Malformed); + parser.offset = parser.data_end + 2; + parser.mode = MemcacheResponseParser::Mode::RetrievalHeader; + } + + if(parser.mode == MemcacheResponseParser::Mode::MetaData) + { + if(response.size() < parser.data_end + 2) + return(MemcacheResponseState::Incomplete); + return(response.compare(parser.data_end, 2, "\r\n") == 0 ? MemcacheResponseState::Complete : MemcacheResponseState::Malformed); + } + + if(parser.mode == MemcacheResponseParser::Mode::Lines) + { + size_t line_end = response.find("\r\n", parser.offset); + if(line_end == String::npos) + return(MemcacheResponseState::Incomplete); + if(response.compare(parser.offset, line_end - parser.offset, "END") == 0) + return(MemcacheResponseState::Complete); + parser.offset = line_end + 2; + } + } +} + +static bool native_socket_wait(int fd, short events, u64 deadline) +{ + while(true) + { + u64 now = monotonic_ms(); + if(now >= deadline) + return(false); + pollfd item = {fd, events, 0}; + int timeout = (int)std::min(deadline - now, INT_MAX); + int ready = poll(&item, 1, timeout); + if(ready > 0) + return((item.revents & events) != 0); + if(ready == 0 || errno != EINTR) + return(false); + } +} + +static String native_memcache_exchange(u64 connection, String command) +{ + static constexpr size_t max_response = 8 * 1024 * 1024; + int fd = (int)connection; + u64 deadline = monotonic_ms() + 1000; + String request = command + "\r\n"; + size_t sent = 0; + while(sent < request.size()) + { + if(!native_socket_wait(fd, POLLOUT, deadline)) + break; + ssize_t count = send(fd, request.data() + sent, request.size() - sent, MSG_DONTWAIT | MSG_NOSIGNAL); + if(count > 0) + sent += (size_t)count; + else if(count < 0 && (errno == EINTR || errno == EAGAIN || errno == EWOULDBLOCK)) + continue; + else + break; + } + if(sent != request.size()) + { + shutdown(fd, SHUT_RDWR); + return(""); + } + if(uce_memcache_command_has_no_reply(command)) + return(""); + + String response; + response.reserve(16 * 1024); + MemcacheResponseParser parser; + while(response.size() < max_response) + { + MemcacheResponseState state = uce_memcache_response_advance(command, response, parser); + if(state == MemcacheResponseState::Complete) + return(response); + if(state == MemcacheResponseState::Malformed || !native_socket_wait(fd, POLLIN, deadline)) + break; + char buffer[64 * 1024]; + size_t remaining = max_response - response.size(); + ssize_t count = recv(fd, buffer, std::min(remaining, sizeof(buffer)), MSG_DONTWAIT); + if(count > 0) + response.append(buffer, (size_t)count); + else if(count < 0 && (errno == EINTR || errno == EAGAIN || errno == EWOULDBLOCK)) + continue; + else + break; + } + if(uce_memcache_response_advance(command, response, parser) == MemcacheResponseState::Complete) + return(response); + shutdown(fd, SHUT_RDWR); + return(""); +} + String memcache_escape_key(String key) { String result; @@ -1320,26 +1500,18 @@ u64 memcache_connect(String host, u16 port) String memcache_command(u64 connection, String command) { - socket_write(connection, command+"\r\n"); - return(socket_read(connection)); // FIXME: do multi-chunk until END line is received! + return(native_memcache_exchange(connection, command)); } bool memcache_set(u64 connection, String key, String value, u64 expires_in) { - socket_write(connection, - // set KEY META_DATA EXPIRY_TIME LENGTH_IN_BYTES - String("set ") + memcache_escape_key(key) + " 0 " + std::to_string(expires_in) + " " + std::to_string(value.length()) + "\r\n" + - value + "\r\n"); - return("STORED" == trim(socket_read(connection))); + return("STORED" == trim(memcache_command(connection, + String("set ") + memcache_escape_key(key) + " 0 " + std::to_string(expires_in) + " " + std::to_string(value.length()) + "\r\n" + value))); } bool memcache_delete(u64 connection, String key) { - socket_write(connection, - // set KEY META_DATA EXPIRY_TIME LENGTH_IN_BYTES - String("delete ") + memcache_escape_key(key) + "\r\n" - ); - return("DELETED" == trim(socket_read(connection))); + return("DELETED" == trim(memcache_command(connection, String("delete ") + memcache_escape_key(key)))); } String memcache_get(u64 connection, String key, String default_value) diff --git a/src/lib/sys.h b/src/lib/sys.h index 8659897..f86f53d 100644 --- a/src/lib/sys.h +++ b/src/lib/sys.h @@ -128,6 +128,15 @@ String signal_name(s32 sig); String memcache_escape_key(String key); StringList memcache_escape_keys(StringList keys); u64 memcache_connect(String host = "127.0.0.1", u16 port = 11211); +enum class MemcacheResponseState { Incomplete, Complete, Malformed }; +struct MemcacheResponseParser +{ + enum class Mode { FirstLine, RetrievalHeader, RetrievalData, Lines, MetaData } mode = Mode::FirstLine; + size_t offset = 0; + size_t data_end = 0; +}; +__attribute__((visibility("hidden"))) bool uce_memcache_command_has_no_reply(const String& command); +__attribute__((visibility("hidden"))) MemcacheResponseState uce_memcache_response_advance(const String& command, const String& response, MemcacheResponseParser& parser); String memcache_command(u64 connection, String command); bool memcache_set(u64 connection, String key, String value, u64 expires_in = 60*60); bool memcache_delete(u64 connection, String key); diff --git a/src/wasm/worker.cpp b/src/wasm/worker.cpp index b4ce6b9..5ddbb4d 100644 --- a/src/wasm/worker.cpp +++ b/src/wasm/worker.cpp @@ -395,10 +395,9 @@ static u64 wasm_socket_connect_bounded(const String& host, u16 port, u64 timeout return((u64)fd); } -static bool wasm_socket_write_bounded(u64 socket_fd, const String& data, u64 timeout_ms) +static bool wasm_socket_write_until(u64 socket_fd, const String& data, u64 deadline) { int fd = (int)socket_fd; - u64 deadline = wasm_deadline_after_ms(timeout_ms); size_t offset = 0; while(offset < data.size()) { @@ -413,6 +412,11 @@ static bool wasm_socket_write_bounded(u64 socket_fd, const String& data, u64 tim return(true); } +static bool wasm_socket_write_bounded(u64 socket_fd, const String& data, u64 timeout_ms) +{ + return(wasm_socket_write_until(socket_fd, data, wasm_deadline_after_ms(timeout_ms))); +} + static String wasm_socket_read_bounded(u64 socket_fd, u32 max_length, u64 timeout_ms) { if(max_length == 0 || !wasm_socket_wait((int)socket_fd, POLLIN, wasm_deadline_after_ms(timeout_ms))) @@ -422,6 +426,42 @@ static String wasm_socket_read_bounded(u64 socket_fd, u32 max_length, u64 timeou return(count > 0 ? String(buffer.data(), (size_t)count) : String("")); } +static String wasm_memcache_exchange(u64 socket_fd, const String& command, u64 timeout_ms) +{ + static constexpr size_t max_response = 8 * 1024 * 1024; + int fd = (int)socket_fd; + u64 deadline = wasm_deadline_after_ms(timeout_ms); + if(!wasm_socket_write_until(socket_fd, command + "\r\n", deadline)) + { + shutdown(fd, SHUT_RDWR); + return(""); + } + if(uce_memcache_command_has_no_reply(command)) + return(""); + String response; + response.reserve(16 * 1024); + MemcacheResponseParser parser; + while(response.size() < max_response) + { + MemcacheResponseState state = uce_memcache_response_advance(command, response, parser); + if(state == MemcacheResponseState::Complete) + return(response); + if(state == MemcacheResponseState::Malformed || !wasm_socket_wait(fd, POLLIN, deadline)) + break; + char buffer[64 * 1024]; + size_t remaining = max_response - response.size(); + ssize_t count = recv(fd, buffer, std::min(remaining, sizeof(buffer)), MSG_DONTWAIT); + if(count > 0) + response.append(buffer, (size_t)count); + else if(count == 0 || (errno != EINTR && errno != EAGAIN && errno != EWOULDBLOCK)) + break; + } + if(uce_memcache_response_advance(command, response, parser) == MemcacheResponseState::Complete) + return(response); + shutdown(fd, SHUT_RDWR); + return(""); +} + static f64 wasm_thread_cpu_time() { struct timespec ts; @@ -4254,8 +4294,7 @@ private: else { u64 socket_fd = (u64)args[0].i64(); - wasm_socket_write_bounded(socket_fd, command + "\r\n", self->bounded_hostcall_timeout_ms(1000)); - out = wasm_socket_read_bounded(socket_fd, 1024 * 128, self->bounded_hostcall_timeout_ms(1000)); + out = wasm_memcache_exchange(socket_fd, command, self->bounded_hostcall_timeout_ms(1000)); if(buf == 0) { self->staged_memcache_key = key;