From 1bddaf2de146282ab023fc943f4a15563c1c2cdc Mon Sep 17 00:00:00 2001 From: Bharath Rupireddy Date: Tue, 8 Sep 2026 16:43:36 +0000 Subject: [PATCH v5 1/4] Fix hang and deadlock in concurrent REPACK worker handling. REPACK CONCURRENTLY starts a background worker to decode the changes made to the table while it is being repacked, and the backend running the command coordinates with that worker through shared memory and an error message queue. Previously, the backend went straight to waiting on a condition variable for the worker to set up decoding. Nothing in that wait notices a worker that never arrived. The postmaster does signal the backend when the worker fails to start, for example when fork() fails, but the signal only wakes the sleep, which resumes without raising anything. The backend therefore waits forever while holding its lock on the table. Fix this by waiting for the worker to attach to the error message queue before waiting for it to initialize decoding. The worker's status tells a worker that is gone from one that is merely slow, and whether it ever became the sender on the queue tells whether it left an error behind. A worker that stopped without attaching gets the generic startup failure, and one that attached and reported an error before exiting has that error come through the queue. The backend can also deadlock with the worker while stopping it, because it waits for the worker to exit before detaching from the error message queue. A worker blocked writing into a full queue normally escapes once its send loop acts on the termination, but not if it holds interrupts where it blocks, and the backend holds interrupts of its own across the exit wait. Neither side can be canceled. Fix this by detaching from the error message queue before waiting for the worker to exit, so that the blocked send fails with SHM_MQ_DETACHED and the worker leaves without having to act on an interrupt at all. Teardown only terminates a worker it has a handle for, so a worker that was never registered needs nothing undone. Backpatch to 19, where REPACK CONCURRENTLY was introduced. Reported-by: Nathan Bossart Reported-by: Bharath Rupireddy Author: Bharath Rupireddy Reviewed-by: Antonin Houska Reviewed-by: Masahiko Sawada Discussion: https://postgr.es/m/CALj2ACVAxA9HxvFe8HSspTJ-UO4Aoz%3DkuQdZBeLrod0gqUxH3g%40mail.gmail.com Discussion: https://postgr.es/m/apBpOVZOyqrakEr_@nathan Backpatch-through: 19 --- src/backend/commands/repack.c | 155 +++++++++++++++++++++++++++++----- 1 file changed, 134 insertions(+), 21 deletions(-) diff --git a/src/backend/commands/repack.c b/src/backend/commands/repack.c index 609e2c1d77e..068c6222e3d 100644 --- a/src/backend/commands/repack.c +++ b/src/backend/commands/repack.c @@ -217,6 +217,7 @@ static Oid determine_clustered_index(Relation rel, bool usingindex, const char *indexname); static void start_repack_decoding_worker(Oid relid); +static void wait_for_repack_worker_to_attach(void); static void stop_repack_decoding_worker(void); static void stop_repack_decoding_worker_cb(int code, Datum arg); static Snapshot get_initial_snapshot(DecodingWorker *worker); @@ -3795,6 +3796,19 @@ start_repack_decoding_worker(Oid relid) errmsg("out of background worker slots"), errhint("You might need to increase \"%s\".", "max_worker_processes")); + /* + * Now that the worker is registered, connect the error message queue to + * it. + */ + shm_mq_set_handle(decoding_worker->error_mqh, decoding_worker->handle); + + /* + * Make sure the worker has started before we wait for it to initialize + * decoding below, so that the failure-to-start case does not hang + * forever. + */ + wait_for_repack_worker_to_attach(); + /* * The decoding setup must be done before the caller can have XID assigned * for any reason, otherwise the worker might end up in a deadlock, @@ -3818,6 +3832,89 @@ start_repack_decoding_worker(Oid relid) ConditionVariableCancelSleep(); } +/* + * Wait for the decoding worker to start up, and throw an error if it fails + * to do so. + * + * This is similar to WaitForParallelWorkersToAttach(). The only reliable way + * to tell a worker that failed to start (fork failure, or an exit before it + * attached) from one that is merely slow is to check whether it became the + * sender on the error message queue. If it stopped without attaching, nothing + * was queued and we report the generic failure ourselves. If it attached, any + * error it reported is in the queue and is thrown when we process pending + * messages, either here or later while we wait for it to initialize decoding. + */ +static void +wait_for_repack_worker_to_attach(void) +{ + bool worker_attached = false; + + for (;;) + { + BgwHandleStatus status; + shm_mq *mq; + int rc; + pid_t pid; + + /* + * This will process any repack messages that are pending and it may + * also throw an error propagated from a worker. + */ + CHECK_FOR_INTERRUPTS(); + + /* If the worker is known to have attached, we're done. */ + if (worker_attached) + break; + + /* If error_mqh is NULL, the worker has already exited cleanly. */ + if (decoding_worker->error_mqh == NULL) + { + worker_attached = true; + continue; + } + + status = GetBackgroundWorkerPid(decoding_worker->handle, &pid); + if (status == BGWH_STARTED) + { + /* Has the worker attached to the error message queue? */ + mq = shm_mq_get_queue(decoding_worker->error_mqh); + if (shm_mq_get_sender(mq) != NULL) + worker_attached = true; + } + else if (status == BGWH_STOPPED) + { + /* + * If the worker stopped without attaching to the error message + * queue, throw an error. Otherwise it attached and reported an + * error before exiting, so mark it attached and let the next + * attempt to process pending messages, here or later while the + * initial snapshot is set up, throw that error. + */ + mq = shm_mq_get_queue(decoding_worker->error_mqh); + if (shm_mq_get_sender(mq) == NULL) + ereport(ERROR, + errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE), + errmsg("REPACK decoding worker failed to start"), + errhint("More details may be available in the server log.")); + + worker_attached = true; + } + else + { + /* + * Worker not yet started, so we must wait. The postmaster will + * notify us via bgw_notify_pid if its state changes. + */ + rc = WaitLatch(MyLatch, + WL_LATCH_SET | WL_EXIT_ON_PM_DEATH, + -1, WAIT_EVENT_BGWORKER_STARTUP); + + if (rc & WL_LATCH_SET) + ResetLatch(MyLatch); + } + } +} + /* * Stop the decoding worker and cleanup the related resources. * @@ -3831,13 +3928,43 @@ stop_repack_decoding_worker(void) if (decoding_worker == NULL) return; - /* Terminate the worker process, if one is running. */ + /* Terminate the worker and forget its error message queue. */ + if (decoding_worker->error_mqh != NULL) + { + /* + * The error message queue is attached before the worker is + * registered. + */ + if (decoding_worker->handle != NULL) + TerminateBackgroundWorker(decoding_worker->handle); + + shm_mq_detach(decoding_worker->error_mqh); + decoding_worker->error_mqh = NULL; + } + + /* + * Cancel any sleep on the condition variable before detaching the shared + * memory segment, because the CV lives in that segment. Otherwise later + * cleanup would touch freed memory. + */ + ConditionVariableCancelSleep(); + + /* If we have allocated a shared memory segment, detach it. */ + if (decoding_worker->seg != NULL) + { + dsm_detach(decoding_worker->seg); + decoding_worker->seg = NULL; + } + + /* + * We can't finish the REPACK command until the worker has exited. This + * means, in particular, that we can't respond to interrupts at this + * stage. + */ if (decoding_worker->handle != NULL) { BgwHandleStatus status; - TerminateBackgroundWorker(decoding_worker->handle); - /* The worker should really exit before the REPACK command does. */ HOLD_INTERRUPTS(); status = WaitForBackgroundWorkerShutdown(decoding_worker->handle); RESUME_INTERRUPTS(); @@ -3848,22 +3975,6 @@ stop_repack_decoding_worker(void) errmsg("postmaster exited during REPACK command")); } - /* - * Now detach from our shared memory segment. In error cases there might - * still be messages from the worker in the queue, which ProcessInterrupts - * would try to read; this is pointless (and causes an assertion failure), - * so set the global pointer to NULL to have ProcessRepackMessages ignore - * them. - * - * We must also cancel the current sleep, if one is still set up. This is - * critical because the CV lives in the DSM that we're about to detach, so - * if we omit it, later automatic cleanup tries to clear freed memory. - */ - if (decoding_worker->error_mqh != NULL) - shm_mq_detach(decoding_worker->error_mqh); - ConditionVariableCancelSleep(); - if (decoding_worker->seg != NULL) - dsm_detach(decoding_worker->seg); pfree(decoding_worker); decoding_worker = NULL; } @@ -3968,9 +4079,11 @@ ProcessRepackMessages(void) /* * Nothing to do if we haven't launched the worker yet or have already - * terminated it. + * terminated it. Stopping the worker detaches the error message queue + * before clearing decoding_worker, so also bail out once error_mqh is + * gone. */ - if (decoding_worker == NULL) + if (decoding_worker == NULL || decoding_worker->error_mqh == NULL) return; /* -- 2.47.3