udo/uce
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Reuse reset MySQL connections per worker

Browse Source
This commit is contained in:
udo
2026-07-16 17:00:28 +00:00
parent b04ce3d005
commit 6a02c07bcb
11 changed files with 185 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/wasm/worker.cpp
+86 -7
View File
@@ -92,6 +92,7 @@ struct WasmWorkerConfig
int64_t memory_limit = 512ll * 1024 * 1024;
u32 table_headroom = 4096;
u64 epoch_deadline_ticks = 200; // ticker period × ticks = CPU budget
u64 mysql_persistent_pool_size = 8;
bool verbose = false;
// uce_host_* names (bare, without the "uce_host_" prefix) the sysadmin has
// disabled via UCE_HOSTCALL_BLOCKLIST. A blocked hostcall resolves to a trap
@@ -102,6 +103,7 @@ struct WasmWorkerConfig
struct WasmMySQLOperation
{
String op;
String source;
u64 elapsed_us = 0;
};
@@ -119,6 +121,9 @@ struct WasmRequestProfile
u64 mysql_hostcall_total_us = 0;
u64 mysql_operation_count = 0;
u64 mysql_operations_dropped = 0;
u64 mysql_connection_open_count = 0;
u64 mysql_connection_reuse_count = 0;
u64 mysql_request_pool_hit_count = 0;
std::vector<WasmMySQLOperation> mysql_operations;
u64 memcache_hostcall_count = 0;
u64 memcache_hostcall_total_us = 0;
@@ -625,6 +630,62 @@ public:
{
}
#ifdef UCE_WASM_HOST_CONNECTORS
~WasmWorker()
{
for(auto* db : mysql_persistent_pool)
delete db;
}
MySQL* mysql_checkout(const String& host, const String& username, const String& password, bool& reused, bool& persistent)
{
reused = false;
persistent = false;
for(size_t i = 0; i < mysql_persistent_pool.size(); i++)
{
MySQL* db = mysql_persistent_pool[i];
if(!db || !db->connection || db->request_host != host || db->request_username != username || db->request_password != password)
continue;
if(db->reset_connection())
{
if(i + 1 < mysql_persistent_pool.size())
{
mysql_persistent_pool.erase(mysql_persistent_pool.begin() + i);
mysql_persistent_pool.push_back(db);
}
reused = true;
persistent = true;
return(db);
}
delete db;
mysql_persistent_pool.erase(mysql_persistent_pool.begin() + i);
break;
}
MySQL* db = new MySQL();
persistent = cfg.mysql_persistent_pool_size > 0;
db->worker_persistent = persistent;
db->request_pooled = true;
db->request_host = host;
db->request_username = username;
db->request_password = password;
if(!db->connect(host, username, password) || !db->connection)
return(db);
if(persistent)
{
while(mysql_persistent_pool.size() >= cfg.mysql_persistent_pool_size)
{
delete mysql_persistent_pool.front();
mysql_persistent_pool.erase(mysql_persistent_pool.begin());
}
mysql_persistent_pool.push_back(db);
}
return(db);
}
std::vector<MySQL*> mysql_persistent_pool;
#endif
String init()
{
std::vector<u8> bytes;
@@ -831,6 +892,7 @@ public:
std::vector<SQLite*> sqlite_handles;
std::vector<MySQL*> mysql_handles;
std::vector<MySQL*> mysql_request_pool;
std::vector<MySQL*> mysql_request_owned;
#endif
~WasmWorkspace()
{
@@ -848,6 +910,9 @@ public:
if(db)
delete db; // ~SQLite disconnects
for(auto* db : mysql_request_pool)
if(db)
db->request_leases = 0;
for(auto* db : mysql_request_owned)
if(db)
delete db; // ~MySQL disconnects
#endif
@@ -1780,11 +1845,16 @@ private:
response["mysql_hostcall_us"] = (f64)self->mysql_hostcall_total_us;
response["mysql_operation_count"] = (f64)self->mysql_operation_count;
response["mysql_operations_dropped"] = (f64)self->mysql_operations_dropped;
response["mysql_connection_open_count"] = (f64)self->mysql_connection_open_count;
response["mysql_connection_reuse_count"] = (f64)self->mysql_connection_reuse_count;
response["mysql_request_pool_hit_count"] = (f64)self->mysql_request_pool_hit_count;
response["mysql_operations"].set_array();
for(auto& operation : self->mysql_operations)
{
DValue item;
item["op"] = operation.op;
if(operation.source != "")
item["source"] = operation.source;
item["us"] = (f64)operation.elapsed_us;
response["mysql_operations"].push(item);
}
@@ -2500,6 +2570,7 @@ private:
DValue request, response;
String decode_error;
String op;
String connection_source;
auto operation_started = std::chrono::steady_clock::now();
if(ucb_decode(encoded, request, &decode_error))
{
@@ -2514,19 +2585,23 @@ private:
if(pooled && pooled->connection && pooled->request_host == host && pooled->request_username == username && pooled->request_password == password)
{
db = pooled;
connection_source = "request";
break;
}
bool ok = db != 0;
if(!db)
{
db = new MySQL();
db->request_pooled = true;
db->request_host = host;
db->request_username = username;
db->request_password = password;
ok = db->connect(host, username, password);
bool reused = false;
bool persistent = false;
db = self->worker.mysql_checkout(host, username, password, reused, persistent);
connection_source = reused ? "worker" : "new";
ok = db && db->connection;
if(ok && db->connection)
{
self->mysql_request_pool.push_back(db);
if(!persistent)
self->mysql_request_owned.push_back(db);
}
}
u64 handle = 0;
if(ok && db->connection)
@@ -2534,11 +2609,14 @@ private:
db->request_leases++;
self->mysql_handles.push_back(db);
handle = self->mysql_handles.size();
if(connection_source == "new") self->mysql_connection_open_count++;
else if(connection_source == "worker") self->mysql_connection_reuse_count++;
else if(connection_source == "request") self->mysql_request_pool_hit_count++;
}
response["handle"] = (f64)handle;
response["error_code"] = (f64)db->_preload_next_error_code;
response["statement_info"] = db->error();
if(handle == 0)
if(handle == 0 && db)
delete db;
}
else if(op == "escape")
@@ -2571,6 +2649,7 @@ private:
{
WasmMySQLOperation operation;
operation.op = op;
operation.source = connection_source;
operation.elapsed_us = (u64)std::chrono::duration_cast<std::chrono::microseconds>(
std::chrono::steady_clock::now() - operation_started).count();
self->mysql_operation_count++;