Improvements

Author: mgrunbauer

adagucserverEC/adagucserver.cpp

@@ -88,7 +88,8 @@ int run_adaguc_once(int argc, char **argv, char **envp) {
 
 int main(int argc, char **argv, char **envp) {
   const char *fork_enable = getenv("ADAGUC_FORK_ENABLE");
-  if (fork_enable && std::string(fork_enable) == "TRUE") {
+  bool use_fork_server = fork_enable && std::string(fork_enable) == "TRUE" && argc == 1;
+  if (use_fork_server) {
     // Fork children inherit the mother's stdio buffers.
     // This keeps stdout/stderr unbuffered, so old buffered output cannot be written into a the unix socket before the HTTP headers.
     setvbuf(stdout, NULL, _IONBF, 0);

adagucserverEC/fork_server.cpp

@@ -23,8 +23,6 @@ See also: doc/fork_server.md
 - Methods prefixed with `child_` only get executed by the (forked) children
 */
 
-// Check this many seconds for old left-over processes
-const int CHECK_CHILD_PROC_INTERVAL = 30;
 // Default for ADAGUC_NUMPARALLELPROCESSES, matches the python default
 const int DEFAULT_NUM_PARALLEL_PROCESSES = 4;
 // Default for ADAGUC_MAX_PROC_TIMEOUT, matches the python default.
@@ -38,6 +36,19 @@ typedef struct {
 static std::map<pid_t, child_proc_t> child_procs;
 int self_pipe[2];
 
+/** Close descriptors that belong only to the mother process after a fork. */
+void child_close_mother_file_descriptors(int listen_socket) {
+  close(listen_socket);
+  close(self_pipe[0]);
+  close(self_pipe[1]);
+
+  // Note: `child_procs` comes from the forked mother process but is not shared. Safe to clean everything.
+  for (const auto &[pid, child_proc]: child_procs) {
+    (void)pid;
+    close(child_proc.child_socket_fd);
+  }
+}
+
 /**
  * Gets a positive integer value from the environment.
  *
@@ -241,7 +252,7 @@ void mother_handle_child_exited_events() {
 void mother_kill_old_child_procs(int max_child_proc_timeout) {
   time_t now = time(NULL);
   for (auto it = child_procs.begin(); it != child_procs.end(); ++it) {
-    if (difftime(now, it->second.forked_at) > max_child_proc_timeout) {
+    if (difftime(now, it->second.forked_at) >= max_child_proc_timeout) {
       kill(it->first, SIGKILL);
     }
   }
@@ -317,8 +328,9 @@ int mother_run_as_fork_service(int (*run_adaguc_once)(int, char **, char **), in
     return 1;
   }
 
-  // Keep one extra child slot above the request limit so ping messages can still be handled when Python's request semaphore is full.
-  int max_child_procs = mother_get_env_var_int("ADAGUC_NUMPARALLELPROCESSES", DEFAULT_NUM_PARALLEL_PROCESSES) + 1;
+  // Use `ADAGUC_NUMPARALLELPROCESSES` to set maximum requests, keep one extra slot so PING can be handled when its semaphore is full.
+  int max_request_child_procs = std::max(mother_get_env_var_int("ADAGUC_NUMPARALLELPROCESSES", DEFAULT_NUM_PARALLEL_PROCESSES), 2);
+  int max_child_procs = max_request_child_procs + 1;
   int max_child_proc_timeout = mother_get_env_var_int("ADAGUC_MAX_PROC_TIMEOUT", DEFAULT_MAX_CHILD_PROC_TIMEOUT);
   CDBDebug("Max child processes: %d", max_child_procs);
   CDBDebug("Max child process timeout: %d", max_child_proc_timeout);
@@ -330,8 +342,6 @@ int mother_run_as_fork_service(int (*run_adaguc_once)(int, char **, char **), in
   }
 
   CDBDebug("Entering fork server loop");
-  time_t last_cleanup = time(NULL);
-
   while (1) {
     // fd_set is modified by select(); reinitialize it each loop to monitor listen_socket and self_pipe again
     fd_set readfds;
@@ -359,12 +369,8 @@ int mother_run_as_fork_service(int (*run_adaguc_once)(int, char **, char **), in
       continue;
     }
 
-    // Check for dead processes
-    time_t now = time(NULL);
-    if (now - last_cleanup >= CHECK_CHILD_PROC_INTERVAL) {
-      mother_kill_old_child_procs(max_child_proc_timeout);
-      last_cleanup = now;
-    }
+    // select() wakes at least once per second, first check if there are old processes that need cleaning
+    mother_kill_old_child_procs(max_child_proc_timeout);
 
     // Only run if there is activity on the self_pipe (to handle exit events)
     if (FD_ISSET(self_pipe[0], &readfds)) {
@@ -385,7 +391,7 @@ int mother_run_as_fork_service(int (*run_adaguc_once)(int, char **, char **), in
       // Both if/else paths are taken. Mother process takes pid > 0, child process takes pid == 0.
       if (pid == 0) {
         // Child process handles request. Communication with python happens through `child_socket_fd`
-        close(listen_socket);
+        child_close_mother_file_descriptors(listen_socket);
         child_handle_client(child_socket_fd, run_adaguc_once, argc, argv, envp);
         _exit(1);
       } else if (pid > 0) {

doc/EnvironmentVariablesAndExitCodes.md

@@ -38,6 +38,7 @@ The following exit codes are supported:
 | `ADAGUC_FONT`             | Path to a default TrueType font (TTF) used for rendering text in generated imagery (e.g. `FreeSans.ttf`). | -|
 | `ADAGUC_FORK_ENABLE`      | Enables running adaguc-server in [fork mode](/doc/fork_server.md) when set to `TRUE`. | `FALSE` |
 | `ADAGUC_LOGFILE`          | File where log messages are written. | -|
+| `ADAGUC_MAX_COMMAND_TIMEOUT` | Maximum processing time in seconds for command-style invocations such as `--updatedb` and `--updatelayermetadata`. | `300` |
 | `ADAGUC_MAX_PROC_TIMEOUT` | Maximum allowed processing time (in seconds) for a single adaguc-server request. If a request exceeds this limit, the server terminates the process and returns an HTTP 500 error to the client. This prevents requests from running indefinitely and blocking server resources. | 10 |
 | `ADAGUC_NUMPARALLELPROCESSES` | Number of parallel worker processes used by the adaguc-server. | `4` |
 | `ADAGUC_ONLINERESOURCE`   | Optional override for the [OnlineResource](configuration/OnlineResource.md) URL used by the CGI service. Can also be configured in the XML configuration file. | -|
@@ -51,4 +52,4 @@ The following exit codes are supported:
 | `ADAGUC_TRUSTED_HOSTS`    | When EXTERNALADDRESS is unset, this is the allowed hosts list that can be advertised as online resource in the WMS GetCapabilities | * |
 | `ADAGUC_TRUSTED_PROXIES`  | When EXTERNALADDRESS is unset, this is the list of trusted proxies from which adaguc-server will determine its external address. (Via the x-forwared- headers) | * |
 | `PGBOUNCER_DISABLE_SSL`   | Disables SSL when connecting through PgBouncer. | `true` |
-| `PGBOUNCER_ENABLE`        | Enables or disables PostgreSQL connection pooling through PgBouncer. | `true` |
+| `PGBOUNCER_ENABLE`        | Enables or disables PostgreSQL connection pooling through PgBouncer. | `true` |

doc/Running.md

@@ -42,10 +42,10 @@ bash docker-compose-generate-env.sh \
   -f $ADAGUCDOCKERHOME/adaguc-data \
   -p 443
 # You can view or edit the file ./.env file
-
-For more info on what environment variables you can configure, see [Environment variables](/doc/Environment_Variables.md)
 ```
 
+For more info on what environment variables you can configure, see [Environment variables](EnvironmentVariablesAndExitCodes.md).
+
 ### Step 3. Once the steps above have been done, it is time to start:
 
 ```
@@ -129,4 +129,4 @@ A tip to list the actual contents in the docker container is to do:
 ```
 docker exec -i -t my-adaguc-server bash -c "ls -lrt /data/adaguc-data"
 docker exec -i -t my-adaguc-server bash -c "ls -lrt /data/adaguc-datasets"
-```
+```

doc/fork_server.md

@@ -5,7 +5,7 @@ Each request is handled in a separate child process, while the mother process re
 
 This design reduces per-request overhead by avoiding repeated process initialization.
 
-The fork server is enabled by setting the environment variable `ADAGUC_FORK_ENABLE` to `TRUE`. If this variable is not set to `TRUE`, ADAGUC runs without the fork server.
+The fork server is enabled by setting the environment variable `ADAGUC_FORK_ENABLE` to `TRUE`. If this variable is not set to `TRUE`, ADAGUC runs without the fork server. Command-style invocations with arguments, such as `--updatedb`, `--updatelayermetadata`, `--lint`, and `--report`, always run as normal subprocesses.
 
 # Components
 
@@ -15,7 +15,7 @@ The system consists of three parts:
 - Mother process (C++): A persistent process that listens for requests and manages child processes.
 - Child processes (C++): Short-lived processes created with `fork()`. Each child handles one request.
 
-The maximum number of concurrent children is limited by the environment variable `ADAGUC_NUMPARALLELPROCESSES`.
+The environment variable `ADAGUC_NUMPARALLELPROCESSES` determines the number of concurrent processes, with a minimum of two so a metadata update does not block other requests. The mother process permits one additional child, so the supervisor can perform its `PING` health check while other requests can get handled.
 
 Communication between the Python server and the mother process occurs via a Unix domain socket.
 
@@ -122,4 +122,4 @@ The fork server (mother process) is managed by a Python supervisor `fork_server_
 - Starts the process during application startup
 - Performs periodic health checks via the Unix socket
 - Restarts the process if it becomes unresponsive or exits
-- Terminates the process during application shutdown
+- Terminates the mother and its child process group during application shutdown

python/lib/adaguc/CGIRunner.py

@@ -31,6 +31,7 @@
 ON_POSIX = "posix" in sys.builtin_module_names
 
 MAX_PROC_TIMEOUT = int(os.getenv("ADAGUC_MAX_PROC_TIMEOUT", "10"))
+MAX_COMMAND_TIMEOUT = int(os.getenv("ADAGUC_MAX_COMMAND_TIMEOUT", "300"))
 
 
 class AdagucResponse(NamedTuple):
@@ -84,13 +85,20 @@ async def socket_communicate(url: str, env: dict[str, str]) -> AdagucResponse:
 
     reader, writer = await asyncio.open_unix_connection(get_fork_socket_path())
 
-    writer.write(message_bytes)
-    await writer.drain()
+    try:
+        writer.write(message_bytes)
+        await writer.drain()
 
-    process_output = await reader.read()
+        process_output = await reader.read()
+    finally:
+        writer.close()
+        try:
+            await writer.wait_closed()
+        except OSError:
+            pass
 
-    writer.close()
-    await writer.wait_closed()
+    if len(process_output) < 4:
+        raise RuntimeError("Invalid response from ADAGUC fork server: missing exit status")
 
     # Status code is stored in the last 4 bytes from the received data
     status_code = int.from_bytes(process_output[-4:], sys.byteorder)
@@ -146,7 +154,7 @@ async def run(
         env: dict = {},
         path: str | None = None,
         isCGI: bool = True,
-        timeout: int = MAX_PROC_TIMEOUT,
+        timeout: int | None = None,
     ) -> tuple[int, list[str], bytes | None, bytes]:
         localenv = {}
         if url != None:
@@ -159,8 +167,11 @@ async def run(
             localenv["REQUEST_URI"] = "/myscriptname/" + path
         localenv.update(env)
 
-        # Only use fork server if ADAGUC_FORK_ENABLE=TRUE and adaguc is not executed with extra arguments e.g. `--updatelayermetadata`
-        use_fork = is_fork_enabled() and len(cmds) == 1
+        # Command-style calls, such as `--updatelayermetadata`, always use a subprocess and may run much longer than web requests.
+        is_command = len(cmds) > 1
+        use_fork = is_fork_enabled() and not is_command
+        if timeout is None:
+            timeout = MAX_COMMAND_TIMEOUT if is_command else MAX_PROC_TIMEOUT
 
         async with sem:
             if use_fork:

python/python_fastapi_server/fork_server_supervisor.py

@@ -12,19 +12,22 @@
 
 
 class ForkServerSupervisor:
-    def __init__(self, interval: int = 5):
+    def __init__(self, interval: int = 5, startup_timeout: float = 10.0):
         """Initialize supervisor with binary path and health check interval.
 
         ADAGUC_CONFIG and ADAGUC_ONLINERESOURCE need to be set manually
         """
 
         self.interval = interval
+        self.startup_timeout = startup_timeout
         self.process: asyncio.subprocess.Process | None = None
 
         self.env = os.environ.copy()
         adaguc_env = runAdaguc().getAdagucEnv()
         self.env.update({k: str(v) for k, v in adaguc_env.items()})
-        self.env["ADAGUC_CONFIG"] = f"{self.env.get('ADAGUC_PATH')}/python/lib/adaguc/adaguc-server-config-python-postgres.xml"
+        self.env["ADAGUC_CONFIG"] = os.environ.get(
+            "ADAGUC_CONFIG", f"{self.env.get('ADAGUC_PATH')}/python/lib/adaguc/adaguc-server-config-python-postgres.xml"
+        )
         self.env["ADAGUC_ONLINERESOURCE"] = os.environ.get("EXTERNALADDRESS", "") + "/adaguc-server?"
 
         self.adaguc_binary_path = f"{self.env.get('ADAGUC_PATH')}/bin/adagucserver"
@@ -40,7 +43,23 @@ async def start_process(self):
             return
 
         logger.info("Starting forkserver")
-        self.process = await asyncio.create_subprocess_exec(self.adaguc_binary_path, env=self.env)
+
+        # Isolate the mother and its future forked children in a process group separate from FastAPI.
+        self.process = await asyncio.create_subprocess_exec(self.adaguc_binary_path, env=self.env, start_new_session=True)
+
+    async def wait_for_process_group_exit(self, process_group_id: int, timeout: float) -> bool:
+        """Wait until a process group no longer contains any processes."""
+
+        loop = asyncio.get_running_loop()
+        deadline = loop.time() + timeout
+        while True:
+            try:
+                os.killpg(process_group_id, 0)
+            except ProcessLookupError:
+                return True
+            if loop.time() >= deadline:
+                return False
+            await asyncio.sleep(min(0.05, deadline - loop.time()))
 
     async def stop_process(self):
         """Terminate the subprocess gracefully, force kill if needed."""
@@ -49,14 +68,37 @@ async def stop_process(self):
             return
 
         logger.info("Stopping forkserver")
+        loop = asyncio.get_running_loop()
+        deadline = loop.time() + 2
+
+        try:
+            # Signal the isolated process group, including the mother and every forked request child.
+            os.killpg(self.process.pid, signal.SIGTERM)
+        except ProcessLookupError:
+            pass
+
         if self.process.returncode is None:
-            self.process.send_signal(signal.SIGTERM)
             try:
-                await asyncio.wait_for(self.process.wait(), timeout=2)
+                await asyncio.wait_for(self.process.wait(), timeout=max(0, deadline - loop.time()))
             except asyncio.TimeoutError:
-                logger.warning("Force killing forkserver")
-                self.process.kill()
-                await self.process.wait()
+                pass
+
+        group_exited = await self.wait_for_process_group_exit(self.process.pid, timeout=max(0, deadline - loop.time()))
+        if not group_exited:
+            logger.warning("Force killing forkserver process group")
+            try:
+                # Force-stop anything in the group that did not handle SIGTERM within the grace period.
+                os.killpg(self.process.pid, signal.SIGKILL)
+            except ProcessLookupError:
+                pass
+
+        if self.process.returncode is None:
+            await self.process.wait()
+
+        try:
+            os.unlink(get_fork_socket_path())
+        except FileNotFoundError:
+            pass
 
         self.process = None
 
@@ -68,6 +110,24 @@ async def restart_process(self):
 
         await self.stop_process()
         await self.start_process()
+        if not await self.wait_until_ready():
+            logger.error("Forkserver did not become ready after restart")
+            await self.stop_process()
+
+    async def wait_until_ready(self) -> bool:
+        """Wait until the forkserver answers health checks or the startup deadline expires."""
+
+        loop = asyncio.get_running_loop()
+        deadline = loop.time() + self.startup_timeout
+
+        while not self._stopping and loop.time() < deadline:
+            if not self.process or self.process.returncode is not None:
+                return False
+            if await self.health_check_mother():
+                return True
+            await asyncio.sleep(0.05)
+
+        return False
 
     async def health_check_mother(self, timeout: float = 1.0) -> bool:
         """Check forkserver health via UNIX socket ping."""
@@ -97,7 +157,7 @@ async def _loop(self):
 
                 # Restart if process crashed
                 if self.process and self.process.returncode is not None:
-                    await self.start_process()
+                    await self.restart_process()
 
                 # Health check
                 elif not await self.health_check_mother():
@@ -114,6 +174,10 @@ async def start_monitoring(self):
         self._running = True
         self._stopping = False
         await self.start_process()
+        if not await self.wait_until_ready():
+            await self.stop_process()
+            self._running = False
+            raise RuntimeError("Forkserver did not become ready during application startup")
         self._task = asyncio.create_task(self._loop())
 
     async def stop_monitoring(self):