From fe951ed8e53892a9ae545e1d65bb04477e57af70 Mon Sep 17 00:00:00 2001 From: Bharath Rupireddy Date: Tue, 8 Sep 2026 16:43:36 +0000 Subject: [PATCH v6 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. Only the queue is detached early. The shared memory segment is still detached after the worker has exited, because the worker attaches to the shared file set in that segment only after it has connected to the database. Detaching the segment before that would destroy the file set, and a worker that is still starting up would then fail with "could not attach to a SharedFileSet that is already destroyed". 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 Reviewed-by: Shihao Zhong 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 | 152 ++++++++++++++++++++++++++++++---- 1 file changed, 135 insertions(+), 17 deletions(-) diff --git a/src/backend/commands/repack.c b/src/backend/commands/repack.c index 3e972f2cc14..4d19dc06b66 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); @@ -3796,6 +3797,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, @@ -3819,6 +3833,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. * @@ -3832,13 +3929,36 @@ 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(); + + /* + * 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(); @@ -3850,21 +3970,17 @@ stop_repack_decoding_worker(void) } /* - * 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. + * Detach from the shared memory segment only now that the worker is gone. + * The worker attaches to the shared file set well after it maps the + * segment, so detaching any earlier can destroy the file set under a + * worker that is still starting up. */ - if (decoding_worker->error_mqh != NULL) - shm_mq_detach(decoding_worker->error_mqh); - ConditionVariableCancelSleep(); if (decoding_worker->seg != NULL) + { dsm_detach(decoding_worker->seg); + decoding_worker->seg = NULL; + } + pfree(decoding_worker); decoding_worker = NULL; } @@ -3969,9 +4085,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.37.1 (Apple Git-137.1)