Keep FastCGI available during restarts

src/fastcgi/src/fcgicc.cc

@@ -206,7 +206,6 @@ FastCGIServer::shutdown()
 	{
 		printf("Closing server socket %i\n", *it);
 		close(*it);
-		sleep(1);
 	}
 
 	for (std::vector<std::string>::iterator it = listen_unlink.begin();
@@ -223,7 +222,6 @@ FastCGIServer::shutdown()
 		delete it->second;
 	}
 
-	sleep(1);
 	server_sockets.clear();
 
 }
@@ -264,6 +262,22 @@ FastCGIServer::listen(unsigned tcp_port)
 	return(listen(tcp_port, "0.0.0.0"));
 }
 
+int
+FastCGIServer::adopt_listener(int socket_handle, char type)
+{
+	int accepting = 0;
+	socklen_t accepting_size = sizeof(accepting);
+	if(socket_handle < 0 || getsockopt(socket_handle, SOL_SOCKET, SO_ACCEPTCONN,
+		&accepting, &accepting_size) != 0 || !accepting)
+		throw std::runtime_error("inherited descriptor is not a listening socket");
+	set_socket_nonblocking(socket_handle);
+	server_sockets.push_back(socket_handle);
+	server_socket_types[socket_handle] = type;
+	printf("(P) adopted #%i inherited %s listener\n", socket_handle,
+		type == 'F' ? "FastCGI" : "server");
+	return(socket_handle);
+}
+
 int
 FastCGIServer::listen(unsigned tcp_port, const std::string& bind_address)
 {

src/fastcgi/src/fcgicc.h

@@ -57,6 +57,7 @@ public:
 	int listen_http(unsigned tcp_port, const std::string& bind_address);
 	int listen_cli(const std::string& local_path);
 	int listen(const std::string& local_path);
+	int adopt_listener(int socket_handle, char type = 'F');
 
 	void process(int timeout_ms = -1); // timeout_ms<0 blocks forever
 	void process_forever();

src/lib/sys.cpp

@@ -1549,6 +1549,12 @@ pid_t task(String key, std::function<void()> exec_after_spawn, u64 timeout)
 	{
 		close_locked_file(lock_fd);
 		my_pid = getpid();
+		// The FastCGI worker handles termination to drain accepted requests.
+		// Generic task children do not run that drain loop, so inheriting those
+		// handlers would turn task_kill(SIGTERM) into a no-op.
+		signal(SIGTERM, SIG_DFL);
+		signal(SIGINT, SIG_DFL);
+		signal(SIGHUP, SIG_DFL);
 		signal(SIGALRM, SIG_DFL);
 		if(timeout > 0)
 			alarm(timeout);

src/linux_fastcgi.cpp

@@ -699,15 +699,7 @@ volatile bool termination_signal_received = false;
 
 void on_terminate(int sig)
 {
-	if(termination_signal_received)
-		return;
 	termination_signal_received = true;
-	if(getpid() != parent_pid)
-		exit(1);
-	printf("Terminating... PID %i:%i\n", getpid(), parent_pid);
-	wasm_backend_shutdown();
-	server.shutdown();
-	exit(1);
 }
 
 void clear_shared_unit_cache(ServerState& state)
@@ -1069,7 +1061,7 @@ void run_ws_broker()
 		ws_broker.listen(server_state.config["WS_BROKER_SOCKET_PATH"]);
 		chmod(server_state.config["WS_BROKER_SOCKET_PATH"].c_str(), S_IRWXU | S_IRGRP | S_IWGRP);
 	}
-	for(;;)
+	while(!termination_signal_received)
 	{
 		ws_broker.process(50);
 		ws_broker_drain_outbound(ws_broker_outbound_timeout_seconds);
@@ -1254,7 +1246,7 @@ void run_proactive_compiler()
 	}
 	next_scan_at = time_precise();
 
-	for(;;)
+	while(!termination_signal_received)
 	{
 		try
 		{
@@ -1397,10 +1389,21 @@ void listen_for_connections()
 		printf("(P) wasm worker ready: PID %i in %.3f ms\n", getpid(), wasm_ms);
 	else
 		fprintf(stderr, "(!) wasm worker initialization failed: PID %i in %.3f ms: %s\n", getpid(), wasm_ms, wasm_error.c_str());
-	for(;;)
+	while(!termination_signal_received)
 	{
 		server.process(-1);
 	}
+	close_inherited_server_sockets();
+	f64 drain_deadline = time_precise() + (f64)to_u64(server_state.config["WORKER_DRAIN_TIMEOUT_SECONDS"], 10);
+	while(!server.client_sockets.empty() && time_precise() < drain_deadline)
+		server.process(100);
+	if(!server.client_sockets.empty())
+		fprintf(stderr, "(!) worker PID %i drain deadline reached with %zu client connections\n",
+			getpid(), server.client_sockets.size());
+	else
+		printf("(P) worker PID %i drained cleanly\n", getpid());
+	wasm_backend_shutdown();
+	exit(server.client_sockets.empty() ? 0 : 1);
 }
 
 mode_t configured_socket_mode(String value, mode_t fallback)
@@ -1424,6 +1427,25 @@ void chmod_configured_socket(String path, String mode_value, mode_t fallback)
 		fprintf(stderr, "(!) Could not chmod socket %s to %04o: %s\n", path.c_str(), (unsigned int)mode, strerror(errno));
 }
 
+int systemd_fastcgi_listener()
+{
+	const char* listen_pid = getenv("LISTEN_PID");
+	const char* listen_fds = getenv("LISTEN_FDS");
+	if(!listen_pid || !listen_fds)
+		return(-1);
+	if(to_u64(listen_pid, 0) != (u64)getpid())
+		return(-1);
+	if(to_u64(listen_fds, 0) != 1)
+		throw std::runtime_error("UCE requires exactly one systemd FastCGI listener");
+	const char* names = getenv("LISTEN_FDNAMES");
+	if(names && names[0] != '\0' && String(names) != "fastcgi")
+		throw std::runtime_error("systemd listener must be named fastcgi");
+	unsetenv("LISTEN_PID");
+	unsetenv("LISTEN_FDS");
+	unsetenv("LISTEN_FDNAMES");
+	return(3);
+}
+
 StringMap redacted_server_config_for_log(const StringMap& config)
 {
 	StringMap redacted = config;
@@ -1443,13 +1465,16 @@ void init_base_process()
 	server_state.config = make_server_settings();
 	server_state.config["COMPILER_SYS_PATH"] = cwd_get();
 	printf("Compiler base path: %s\n", server_state.config["COMPILER_SYS_PATH"].c_str());
+	int inherited_fastcgi = systemd_fastcgi_listener();
+	if(inherited_fastcgi >= 0)
+		server.adopt_listener(inherited_fastcgi, 'F');
 
 	if(server_state.config["FCGI_PORT"] != "")
 		server.listen(int_val(server_state.config["FCGI_PORT"]));
 
 	printf("%s\n", var_dump(redacted_server_config_for_log(server_state.config)).c_str());
 
-	if(server_state.config["FCGI_SOCKET_PATH"] != "")
+	if(inherited_fastcgi < 0 && server_state.config["FCGI_SOCKET_PATH"] != "")
 	{
 		server.listen(server_state.config["FCGI_SOCKET_PATH"]);
 		chmod_configured_socket(server_state.config["FCGI_SOCKET_PATH"], server_state.config["FCGI_SOCKET_MODE"], 0666);
@@ -1469,7 +1494,9 @@ void init_base_process()
 	mkdir(server_state.config["SESSION_PATH"]);
 
 	signal(SIGCHLD, on_child_exit);
+	signal(SIGTERM, on_terminate);
 	signal(SIGINT, on_terminate);
+	signal(SIGHUP, on_terminate);
 	signal(SIGPIPE, SIG_IGN);
 	srand(time());
 }
@@ -1489,7 +1516,7 @@ int main(int argc, char** argv)
 	ensure_proactive_compiler();
 	ensure_ws_broker();
 
-	for(;;)
+	while(!termination_signal_received)
 	{
 		if(!proactive_compiler_alive())
 			proactive_compiler_pid = 0;
@@ -1512,5 +1539,22 @@ int main(int argc, char** argv)
 		sleep(1);
 	}
 
+	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);
+	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)
+	{
+		on_child_exit(0);
+		usleep(10000);
+	}
+	for(auto& worker : workers)
+		kill(worker.first, SIGKILL);
+	server.shutdown();
+
 	return 0;
 }

src/wasm/worker.cpp

@@ -2106,6 +2106,11 @@ private:
 			auto profiled = [self, callback, profile_name](Caller caller, Span<const Val> args, Span<Val> results) mutable -> Result<std::monostate, Trap> {
 				auto started = std::chrono::steady_clock::now();
 				auto result = callback(caller, args, results);
+				// Epoch interruption limits guest CPU, not time spent in native I/O,
+				// process management, hashing, or other host work. Re-arm at the one
+				// membrane every hostcall crosses so newly added blocking imports
+				// cannot silently consume the next guest segment's budget.
+				caller.context().set_epoch_deadline(self->worker.cfg.epoch_deadline_ticks);
 				if(profile_name != "uce_host_request_perf")
 				{
 					u64 elapsed = (u64)std::chrono::duration_cast<std::chrono::microseconds>(std::chrono::steady_clock::now() - started).count();
@@ -2288,9 +2293,9 @@ private:
 		if(mod == "env" && name == "uce_host_crypto_equal")
 			return(add([self](Caller, Span<const Val> args, Span<Val> results) -> Result<std::monostate, Trap> { String a,b; self->hostcall_read(args[0].i32(), args[1].i32(), a); self->hostcall_read(args[2].i32(), args[3].i32(), b); results[0]=Val((int32_t)(crypto_equal_native(a,b)?1:0)); return(std::monostate()); }));
 		if(mod == "env" && name == "uce_host_password_hash")
-			return(add([self](Caller caller, Span<const Val> args, Span<Val> results) -> Result<std::monostate, Trap> { String password; self->hostcall_read(args[0].i32(), args[1].i32(), password); String out=password_hash_native(password); u32 cap=(u32)args[3].i32(); int32_t buf=args[2].i32(); if(buf&&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()); }));
+			return(add([self](Caller, Span<const Val> args, Span<Val> results) -> Result<std::monostate, Trap> { String password; self->hostcall_read(args[0].i32(), args[1].i32(), password); String out=password_hash_native(password); u32 cap=(u32)args[3].i32(); int32_t buf=args[2].i32(); if(buf&&cap>=out.size()) self->hostcall_write(buf,out); results[0]=Val((int32_t)out.size()); return(std::monostate()); }));
 		if(mod == "env" && name == "uce_host_password_verify")
-			return(add([self](Caller caller, Span<const Val> args, Span<Val> results) -> Result<std::monostate, Trap> { String password,encoded; self->hostcall_read(args[0].i32(), args[1].i32(), password); self->hostcall_read(args[2].i32(), args[3].i32(), encoded); bool valid=password_verify_native(password,encoded); caller.context().set_epoch_deadline(self->worker.cfg.epoch_deadline_ticks); results[0]=Val((int32_t)(valid?1:0)); return(std::monostate()); }));
+			return(add([self](Caller, Span<const Val> args, Span<Val> results) -> Result<std::monostate, Trap> { String password,encoded; self->hostcall_read(args[0].i32(), args[1].i32(), password); self->hostcall_read(args[2].i32(), args[3].i32(), encoded); bool valid=password_verify_native(password,encoded); results[0]=Val((int32_t)(valid?1:0)); return(std::monostate()); }));
 		if(mod == "env" && name == "uce_host_password_needs_rehash")
 			return(add([self](Caller, Span<const Val> args, Span<Val> results) -> Result<std::monostate, Trap> { String encoded; self->hostcall_read(args[0].i32(), args[1].i32(), encoded); results[0]=Val((int32_t)(password_needs_rehash_native(encoded)?1:0)); return(std::monostate()); }));
 		if(mod == "env" && name == "uce_host_log")
@@ -2316,7 +2321,6 @@ private:
 				}
 				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());
 			}));
@@ -2324,7 +2328,7 @@ private:
 			return(add([self](Caller caller, Span<const Val> args, Span<Val> results) -> Result<std::monostate, Trap> {
 				String encoded; self->hostcall_read(args[0].i32(), args[1].i32(), encoded); u32 cap=(u32)args[3].i32(); int32_t buf=args[2].i32(); String out; String stage_key="http:"+encoded;
 				if(!self->hostcall_staged(stage_key,out)) { DValue req,response; String err; if(ucb_decode(encoded,req,&err)) response=uce_http_request_value(req); else response["error"]="http_request decode failed: "+err; out=ucb_encode(response); if(buf==0) self->hostcall_stage(stage_key,out); }
-				if(buf&&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(buf&&cap>=out.size()) self->hostcall_write(buf,out); results[0]=Val((int32_t)out.size()); return(std::monostate());
 			}));
 		if(mod == "env" && name == "uce_host_http_request_async")
 			return(add([self](Caller, Span<const Val> args, Span<Val> results) -> Result<std::monostate, Trap> { String encoded; self->hostcall_read(args[0].i32(), args[1].i32(), encoded); DValue req; String err; u64 id=0; if(ucb_decode(encoded,req,&err)) id=uce_http_spawn_spec(req); results[0]=Val((int64_t)id); return(std::monostate()); }));
@@ -2335,7 +2339,6 @@ private:
 				String out; String stage_key="shell_dv:"+encoded;
 				if(!self->hostcall_staged(stage_key,out)) { DValue spec, response; String err; if(ucb_decode(encoded,spec,&err)) response=uce_shell_exec_spec(spec); else response["error"]="shell_exec spec decode failed: "+err; out=ucb_encode(response); if(buf==0) self->hostcall_stage(stage_key,out); }
 				if(buf&&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_shell_spawn")
@@ -2347,7 +2350,7 @@ private:
 		if(mod == "env" && name == "uce_host_job_result")
 			return(add([self](Caller, Span<const Val> args, Span<Val> results) -> Result<std::monostate, Trap> { String out=ucb_encode(uce_job_result_value((u64)args[0].i64(), 100)); u32 cap=(u32)args[2].i32(); int32_t buf=args[1].i32(); if(buf&&cap>=out.size()) self->hostcall_write(buf,out); results[0]=Val((int32_t)out.size()); return(std::monostate()); }));
 		if(mod == "env" && name == "uce_host_job_await")
-			return(add([self](Caller caller, Span<const Val> args, Span<Val> results) -> Result<std::monostate, Trap> { u64 timeout=std::min<u64>((u64)args[1].i64(), 30000); String out=ucb_encode(uce_job_result_value((u64)args[0].i64(), timeout)); u32 cap=(u32)args[3].i32(); int32_t buf=args[2].i32(); if(buf&&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()); }));
+			return(add([self](Caller, Span<const Val> args, Span<Val> results) -> Result<std::monostate, Trap> { u64 timeout=std::min<u64>((u64)args[1].i64(), 30000); String out=ucb_encode(uce_job_result_value((u64)args[0].i64(), timeout)); u32 cap=(u32)args[3].i32(); int32_t buf=args[2].i32(); if(buf&&cap>=out.size()) self->hostcall_write(buf,out); results[0]=Val((int32_t)out.size()); return(std::monostate()); }));
 		if(mod == "env" && name == "uce_host_job_cancel")
 			return(add([](Caller, Span<const Val> args, Span<Val> results) -> Result<std::monostate, Trap> { results[0]=Val((int32_t)(uce_job_cancel_value((u64)args[0].i64())?1:0)); return(std::monostate()); }));
 		if(mod == "env" && name == "uce_host_path_real")
@@ -2823,7 +2826,6 @@ private:
 				}
 				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());
 			}));
@@ -2922,7 +2924,6 @@ private:
 				}
 				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());
 			}));
@@ -3033,7 +3034,6 @@ private:
 				}
 				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());
 			}));
@@ -3074,7 +3074,6 @@ private:
 						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")
@@ -3115,7 +3114,6 @@ private:
 				}
 				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());
 			}));
@@ -3208,7 +3206,6 @@ private:
 					unsigned int remaining = ::sleep((unsigned int)(usec / 1000000ull));
 					if(remaining != 0)
 					{
-						caller.context().set_epoch_deadline(self->worker.cfg.epoch_deadline_ticks);
 						results[0] = Val((int32_t)remaining);
 						return(std::monostate());
 					}
@@ -3216,7 +3213,6 @@ private:
 				}
 				if(usec > 0)
 					::usleep((useconds_t)usec);
-				caller.context().set_epoch_deadline(self->worker.cfg.epoch_deadline_ticks);
 				results[0] = Val((int32_t)0);
 				return(std::monostate());
 			}));
@@ -3275,7 +3271,6 @@ private:
 					self->hostcall_write(args[6].i32(), resolved);
 				}
 				results[0] = Val(slot);
-				caller.context().set_epoch_deadline(self->worker.cfg.epoch_deadline_ticks);
 				return(std::monostate());
 			}));