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

Parallelize proactive unit compilation

Browse Source
This commit is contained in:
udo
2026-07-18 20:19:37 +00:00
parent d4ae51f54b
commit 33a7b00c78
7 changed files with 303 additions and 64 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/linux_fastcgi.cpp
+117 -56
View File
@@ -16,7 +16,7 @@ ServerState server_state;
#include "fastcgi/src/fcgicc.cc"
FastCGIServer server;
pid_t proactive_compiler_pid = 0;
std::vector<pid_t> proactive_compiler_pids;
pid_t priority_compiler_pid = 0;
// The central WS broker process: owns the WS port + every connection, forwards
@@ -1212,6 +1212,32 @@ bool proactive_compile_queue_has(StringList& queue, String file_name)
return(std::find(queue.begin(), queue.end(), file_name) != queue.end());
}
u64 bounded_compile_jobs(String value, u64 fallback = 2)
{
value = trim(value);
u64 jobs = fallback;
if(value != "")
{
char* end = 0;
errno = 0;
long long parsed = strtoll(value.c_str(), &end, 10);
if(end != value.c_str() && end && *end == '\0' && errno != ERANGE)
jobs = parsed < 1 ? 1 : (u64)parsed;
}
return(std::max<u64>(1, std::min<u64>(jobs, 16)));
}
bool proactive_compile_worker_owns(String file_name, u64 worker, u64 jobs)
{
u64 hash = 1469598103934665603ull;
for(unsigned char c : file_name)
{
hash ^= c;
hash *= 1099511628211ull;
}
return(hash % jobs == worker);
}
void proactive_compile_queue_push(StringList& queue, String file_name)
{
if(file_name == "" || proactive_compile_queue_has(queue, file_name))
@@ -1252,7 +1278,7 @@ bool proactive_compile_unit(Request& context, String file_name, bool& source_mis
return(failed);
}
void run_proactive_compiler()
void run_proactive_compiler(u64 worker, u64 jobs)
{
Request background_context;
StringList compile_queue;
@@ -1281,6 +1307,8 @@ void run_proactive_compiler()
signal(SIGFPE, on_segfault);
signal(SIGPIPE, SIG_IGN);
setpriority(PRIO_PROCESS, 0, 10);
printf("(P) proactive compiler worker %llu/%llu ready: PID %i\n",
(unsigned long long)(worker + 1), (unsigned long long)jobs, getpid());
try
{
@@ -1303,43 +1331,45 @@ void run_proactive_compiler()
{
try
{
auto priority_units = compiler_take_priority_units(&background_context);
for(auto it = priority_units.rbegin(); it != priority_units.rend(); ++it)
{
compile_queue.erase(std::remove(compile_queue.begin(), compile_queue.end(), *it), compile_queue.end());
compile_queue.insert(compile_queue.begin(), *it);
}
if(compile_queue.size() == 0 && time_precise() >= next_scan_at)
{
auto tracked_units = compiler_list_known_units(&background_context);
StringList existing_units;
bool source_generation_changed = false;
for(auto& file_name : tracked_units)
{
bool source_missing = false;
auto retry_it = retry_after.find(file_name);
bool retry_allowed = (retry_it == retry_after.end() || time_precise() >= retry_it->second);
bool needs_compile = compiler_unit_needs_recompile(&background_context, file_name, &source_missing);
if(needs_compile && retry_allowed)
{
proactive_compile_queue_push(compile_queue, file_name);
source_generation_changed = true;
}
if(source_missing)
{
source_generation_changed = true;
printf("(i) proactive compiler forget removed unit %s\n", file_name.c_str());
retry_after.erase(file_name);
continue;
}
existing_units.push_back(file_name);
}
auto tracked_units = compiler_list_known_units(&background_context);
bool source_generation_changed = false;
if(existing_units.size() != tracked_units.size())
compiler_set_known_units(&background_context, existing_units);
if(source_generation_changed)
compiler_mark_source_generation(&background_context);
for(auto& file_name : tracked_units)
{
if(!proactive_compile_worker_owns(file_name, worker, jobs))
continue;
bool source_missing = false;
auto retry_it = retry_after.find(file_name);
bool needs_compile = compiler_unit_needs_recompile(&background_context, file_name, &source_missing);
f64 now = time_precise();
if(needs_compile && retry_it != retry_after.end())
{
retry_after.erase(retry_it);
retry_it = retry_after.end();
}
bool retry_due = retry_it != retry_after.end() && now >= retry_it->second;
if(!needs_compile && retry_it == retry_after.end() && !source_missing &&
compiler_unit_needs_recompile(&background_context, file_name, 0, false, true, true))
retry_after[file_name] = now + failure_retry_interval;
if(needs_compile || retry_due)
{
proactive_compile_queue_push(compile_queue, file_name);
source_generation_changed = true;
}
if(source_missing)
{
source_generation_changed = true;
printf("(i) proactive compiler forget removed unit %s\n", file_name.c_str());
compiler_untrack_known_unit(&background_context, file_name);
retry_after.erase(file_name);
continue;
}
}
if(source_generation_changed)
compiler_mark_source_generation(&background_context);
next_scan_at = time_precise() + check_interval;
}
@@ -1352,7 +1382,8 @@ void run_proactive_compiler()
auto retry_it = retry_after.find(file_name);
if(retry_it != retry_after.end() && time_precise() < retry_it->second)
continue;
bool failed = proactive_compile_unit(background_context, file_name, source_missing);
bool normal_compile = compiler_unit_needs_recompile(&background_context, file_name, &source_missing);
bool failed = proactive_compile_unit(background_context, file_name, source_missing, !normal_compile && !source_missing);
if(source_missing)
retry_after.erase(file_name);
else if(failed)
@@ -1412,9 +1443,9 @@ void run_priority_compiler()
}
}
bool proactive_compiler_alive()
bool proactive_compiler_alive(pid_t pid)
{
return(proactive_compiler_pid > 0 && task_kill(proactive_compiler_pid, 0) == 0);
return(pid > 0 && task_kill(pid, 0) == 0);
}
pid_t spawn_compiler(const char* label, void (*runner)())
@@ -1435,6 +1466,25 @@ pid_t spawn_compiler(const char* label, void (*runner)())
return(p);
}
pid_t spawn_proactive_compiler(u64 worker, u64 jobs)
{
pid_t p = fork();
if(p < 0)
{
perror("fork proactive compiler");
return(0);
}
if(p == 0)
{
prctl(PR_SET_PDEATHSIG, SIGHUP);
run_proactive_compiler(worker, jobs);
exit(0);
}
printf("(P) proactive compiler worker %llu/%llu spawned: PID %i\n",
(unsigned long long)(worker + 1), (unsigned long long)jobs, p);
return(p);
}
void ensure_proactive_compiler()
{
if(!to_bool(server_state.config["PROACTIVE_COMPILE_ENABLED"], true))
@@ -1442,8 +1492,12 @@ void ensure_proactive_compiler()
if(float_val(server_state.config["PROACTIVE_COMPILE_CHECK_INTERVAL"]) <= 0)
return;
if(!proactive_compiler_alive())
proactive_compiler_pid = spawn_compiler("proactive compiler", run_proactive_compiler);
u64 jobs = bounded_compile_jobs(server_state.config["PROACTIVE_COMPILE_JOBS"]);
if(proactive_compiler_pids.size() != jobs)
proactive_compiler_pids.resize(jobs, 0);
for(u64 worker = 0; worker < jobs; worker++)
if(!proactive_compiler_alive(proactive_compiler_pids[worker]))
proactive_compiler_pids[worker] = spawn_proactive_compiler(worker, jobs);
if(priority_compiler_pid <= 0 || task_kill(priority_compiler_pid, 0) != 0)
priority_compiler_pid = spawn_compiler("priority compiler", run_priority_compiler);
}
@@ -1661,16 +1715,7 @@ int precompile_unit_generation()
compiler_set_known_units(&background_context, files);
const char* jobs_env = getenv("UCE_PRECOMPILE_JOBS");
String jobs_text = trim(jobs_env && jobs_env[0] != '\0' ? String(jobs_env) : server_state.config["PRECOMPILE_JOBS"]);
u64 jobs = 2;
if(jobs_text != "")
{
char* end = 0;
errno = 0;
long long parsed = strtoll(jobs_text.c_str(), &end, 10);
if(end != jobs_text.c_str() && end && *end == '\0' && errno != ERANGE)
jobs = parsed < 1 ? 1 : (u64)parsed;
}
jobs = std::max<u64>(1, std::min<u64>(jobs, std::min<u64>(files.size() == 0 ? 1 : files.size(), 16)));
u64 jobs = std::min<u64>(bounded_compile_jobs(jobs_text), files.size() == 0 ? 1 : files.size());
PrecompileWorkerResult total;
bool worker_error = false;
if(jobs == 1)
@@ -1784,8 +1829,9 @@ int main(int argc, char** argv)
while(!termination_signal_received)
{
if(!proactive_compiler_alive())
proactive_compiler_pid = 0;
for(auto& pid : proactive_compiler_pids)
if(!proactive_compiler_alive(pid))
pid = 0;
if(priority_compiler_pid > 0 && task_kill(priority_compiler_pid, 0) != 0)
priority_compiler_pid = 0;
if(!termination_signal_received)
@@ -1810,20 +1856,35 @@ int main(int argc, char** argv)
printf("(P) draining %zu workers before shutdown\n", workers.size());
for(auto& worker : workers)
kill(worker.first, SIGTERM);
if(proactive_compiler_pid > 0)
kill(proactive_compiler_pid, SIGTERM);
for(auto pid : proactive_compiler_pids)
if(pid > 0)
kill(pid, SIGTERM);
if(priority_compiler_pid > 0)
kill(priority_compiler_pid, SIGTERM);
if(ws_broker_pid > 0)
kill(ws_broker_pid, SIGTERM);
f64 drain_deadline = time_precise() + (f64)to_u64(server_state.config["WORKER_DRAIN_TIMEOUT_SECONDS"], 10);
while(!workers.empty() && time_precise() < drain_deadline)
auto background_children_alive = [&]() {
bool alive = priority_compiler_pid > 0 && task_kill(priority_compiler_pid, 0) == 0;
alive = alive || (ws_broker_pid > 0 && task_kill(ws_broker_pid, 0) == 0);
for(auto pid : proactive_compiler_pids)
alive = alive || proactive_compiler_alive(pid);
return(alive);
};
while((!workers.empty() || background_children_alive()) && time_precise() < drain_deadline)
{
on_child_exit(0);
usleep(10000);
}
for(auto& worker : workers)
kill(worker.first, SIGKILL);
for(auto pid : proactive_compiler_pids)
if(proactive_compiler_alive(pid))
kill(pid, SIGKILL);
if(priority_compiler_pid > 0 && task_kill(priority_compiler_pid, 0) == 0)
kill(priority_compiler_pid, SIGKILL);
if(ws_broker_pid > 0 && task_kill(ws_broker_pid, 0) == 0)
kill(ws_broker_pid, SIGKILL);
server.shutdown();
return 0;