*** a/src/backend/access/transam/twophase.c --- b/src/backend/access/transam/twophase.c *************** *** 287,292 **** MarkAsPreparing(TransactionId xid, const char *gid, --- 287,293 ---- gxact->proc.databaseId = databaseid; gxact->proc.roleId = owner; gxact->proc.inCommit = false; + gxact->proc.signalFlags = 0; gxact->proc.vacuumFlags = 0; gxact->proc.lwWaiting = false; gxact->proc.lwExclusive = false; *** a/src/backend/commands/async.c --- b/src/backend/commands/async.c *************** *** 915,923 **** EnableNotifyInterrupt(void) * a frontend command. Signal handler execution of inbound notifies * is disabled until the next EnableNotifyInterrupt call. * ! * The SIGUSR1 signal handler also needs to call this, so as to ! * prevent conflicts if one signal interrupts the other. So we ! * must return the previous state of the flag. */ bool DisableNotifyInterrupt(void) --- 915,924 ---- * a frontend command. Signal handler execution of inbound notifies * is disabled until the next EnableNotifyInterrupt call. * ! * This also needs to be called when SIGUSR1 with ! * PROCSIG_CATCHUP_INTERRUPT is received, so as to prevent conflicts ! * if one signal interrupts the other. So we must return the previous ! * state of the flag. */ bool DisableNotifyInterrupt(void) *************** *** 954,960 **** ProcessIncomingNotify(void) nulls[Natts_pg_listener]; bool catchup_enabled; ! /* Must prevent SIGUSR1 interrupt while I am running */ catchup_enabled = DisableCatchupInterrupt(); if (Trace_notify) --- 955,961 ---- nulls[Natts_pg_listener]; bool catchup_enabled; ! /* Must prevent catchup interrupt while I am running */ catchup_enabled = DisableCatchupInterrupt(); if (Trace_notify) *** a/src/backend/postmaster/autovacuum.c --- b/src/backend/postmaster/autovacuum.c *************** *** 1477,1483 **** AutoVacWorkerMain(int argc, char *argv[]) pqsignal(SIGALRM, handle_sig_alarm); pqsignal(SIGPIPE, SIG_IGN); ! pqsignal(SIGUSR1, CatchupInterruptHandler); /* We don't listen for async notifies */ pqsignal(SIGUSR2, SIG_IGN); pqsignal(SIGFPE, FloatExceptionHandler); --- 1477,1483 ---- pqsignal(SIGALRM, handle_sig_alarm); pqsignal(SIGPIPE, SIG_IGN); ! pqsignal(SIGUSR1, proc_sigusr1_handler); /* We don't listen for async notifies */ pqsignal(SIGUSR2, SIG_IGN); pqsignal(SIGFPE, FloatExceptionHandler); *** a/src/backend/storage/ipc/sinval.c --- b/src/backend/storage/ipc/sinval.c *************** *** 27,33 **** * need a way to give an idle backend a swift kick in the rear and make * it catch up before the sinval queue overflows and forces it to go * through a cache reset exercise. This is done by sending SIGUSR1 ! * to any backend that gets too far behind. * * State for catchup events consists of two flags: one saying whether * the signal handler is currently allowed to call ProcessCatchupEvent --- 27,34 ---- * need a way to give an idle backend a swift kick in the rear and make * it catch up before the sinval queue overflows and forces it to go * through a cache reset exercise. This is done by sending SIGUSR1 ! * with PROCSIG_CATCHUP_INTERRUPT to any backend that gets too far ! * behind. * * State for catchup events consists of two flags: one saying whether * the signal handler is currently allowed to call ProcessCatchupEvent *************** *** 144,152 **** ReceiveSharedInvalidMessages( /* ! * CatchupInterruptHandler * ! * This is the signal handler for SIGUSR1. * * If we are idle (catchupInterruptEnabled is set), we can safely * invoke ProcessCatchupEvent directly. Otherwise, just set a flag --- 145,154 ---- /* ! * HandleCatchupInterrupt * ! * This is called when SIGUSR1 with PROCSIG_CATCHUP_INTERRUPT is ! * received. * * If we are idle (catchupInterruptEnabled is set), we can safely * invoke ProcessCatchupEvent directly. Otherwise, just set a flag *************** *** 156,168 **** ReceiveSharedInvalidMessages( * since there's no longer any reason to do anything.) */ void ! CatchupInterruptHandler(SIGNAL_ARGS) { - int save_errno = errno; - /* ! * Note: this is a SIGNAL HANDLER. You must be very wary what you do ! * here. */ /* Don't joggle the elbow of proc_exit */ --- 158,168 ---- * since there's no longer any reason to do anything.) */ void ! HandleCatchupInterrupt(void) { /* ! * Note: this is called by a SIGNAL HANDLER. ! * You must be very wary what you do here. */ /* Don't joggle the elbow of proc_exit */ *************** *** 216,223 **** CatchupInterruptHandler(SIGNAL_ARGS) */ catchupInterruptOccurred = 1; } - - errno = save_errno; } /* --- 216,221 ---- *************** *** 289,295 **** DisableCatchupInterrupt(void) /* * ProcessCatchupEvent * ! * Respond to a catchup event (SIGUSR1) from another backend. * * This is called either directly from the SIGUSR1 signal handler, * or the next time control reaches the outer idle loop (assuming --- 287,294 ---- /* * ProcessCatchupEvent * ! * Respond to a catchup event (SIGUSR1 with PROCSIG_CATCHUP_INTERRUPT) ! * from another backend. * * This is called either directly from the SIGUSR1 signal handler, * or the next time control reaches the outer idle loop (assuming *** a/src/backend/storage/ipc/sinvaladt.c --- b/src/backend/storage/ipc/sinvaladt.c *************** *** 21,26 **** --- 21,27 ---- #include "storage/backendid.h" #include "storage/ipc.h" #include "storage/proc.h" + #include "storage/procarray.h" #include "storage/shmem.h" #include "storage/sinvaladt.h" #include "storage/spin.h" *************** *** 136,144 **** /* Per-backend state in shared invalidation structure */ typedef struct ProcState { ! /* procPid is zero in an inactive ProcState array entry. */ ! pid_t procPid; /* PID of backend, for signaling */ ! /* nextMsgNum is meaningless if procPid == 0 or resetState is true. */ int nextMsgNum; /* next message number to read */ bool resetState; /* backend needs to reset its state */ bool signaled; /* backend has been sent catchup signal */ --- 137,145 ---- /* Per-backend state in shared invalidation structure */ typedef struct ProcState { ! /* proc is NULL in an inactive ProcState array entry. */ ! PGPROC *proc; /* PGPROC entry of backend, for signaling */ ! /* nextMsgNum is meaningless if proc == NULL or resetState is true. */ int nextMsgNum; /* next message number to read */ bool resetState; /* backend needs to reset its state */ bool signaled; /* backend has been sent catchup signal */ *************** *** 235,241 **** CreateSharedInvalidationState(void) /* Mark all backends inactive, and initialize nextLXID */ for (i = 0; i < shmInvalBuffer->maxBackends; i++) { ! shmInvalBuffer->procState[i].procPid = 0; /* inactive */ shmInvalBuffer->procState[i].nextMsgNum = 0; /* meaningless */ shmInvalBuffer->procState[i].resetState = false; shmInvalBuffer->procState[i].signaled = false; --- 236,242 ---- /* Mark all backends inactive, and initialize nextLXID */ for (i = 0; i < shmInvalBuffer->maxBackends; i++) { ! shmInvalBuffer->procState[i].proc = NULL; /* inactive */ shmInvalBuffer->procState[i].nextMsgNum = 0; /* meaningless */ shmInvalBuffer->procState[i].resetState = false; shmInvalBuffer->procState[i].signaled = false; *************** *** 266,272 **** SharedInvalBackendInit(void) /* Look for a free entry in the procState array */ for (index = 0; index < segP->lastBackend; index++) { ! if (segP->procState[index].procPid == 0) /* inactive slot? */ { stateP = &segP->procState[index]; break; --- 267,273 ---- /* Look for a free entry in the procState array */ for (index = 0; index < segP->lastBackend; index++) { ! if (segP->procState[index].proc == NULL) /* inactive slot? */ { stateP = &segP->procState[index]; break; *************** *** 278,284 **** SharedInvalBackendInit(void) if (segP->lastBackend < segP->maxBackends) { stateP = &segP->procState[segP->lastBackend]; ! Assert(stateP->procPid == 0); segP->lastBackend++; } else --- 279,285 ---- if (segP->lastBackend < segP->maxBackends) { stateP = &segP->procState[segP->lastBackend]; ! Assert(stateP->proc == NULL); segP->lastBackend++; } else *************** *** 303,309 **** SharedInvalBackendInit(void) nextLocalTransactionId = stateP->nextLXID; /* mark myself active, with all extant messages already read */ ! stateP->procPid = MyProcPid; stateP->nextMsgNum = segP->maxMsgNum; stateP->resetState = false; stateP->signaled = false; --- 304,310 ---- nextLocalTransactionId = stateP->nextLXID; /* mark myself active, with all extant messages already read */ ! stateP->proc = MyProc; stateP->nextMsgNum = segP->maxMsgNum; stateP->resetState = false; stateP->signaled = false; *************** *** 341,347 **** CleanupInvalidationState(int status, Datum arg) stateP->nextLXID = nextLocalTransactionId; /* Mark myself inactive */ ! stateP->procPid = 0; stateP->nextMsgNum = 0; stateP->resetState = false; stateP->signaled = false; --- 342,348 ---- stateP->nextLXID = nextLocalTransactionId; /* Mark myself inactive */ ! stateP->proc = NULL; stateP->nextMsgNum = 0; stateP->resetState = false; stateP->signaled = false; *************** *** 349,355 **** CleanupInvalidationState(int status, Datum arg) /* Recompute index of last active backend */ for (i = segP->lastBackend; i > 0; i--) { ! if (segP->procState[i - 1].procPid != 0) break; } segP->lastBackend = i; --- 350,356 ---- /* Recompute index of last active backend */ for (i = segP->lastBackend; i > 0; i--) { ! if (segP->procState[i - 1].proc != NULL) break; } segP->lastBackend = i; *************** *** 374,380 **** BackendIdIsActive(int backendID) { ProcState *stateP = &segP->procState[backendID - 1]; ! result = (stateP->procPid != 0); } else result = false; --- 375,381 ---- { ProcState *stateP = &segP->procState[backendID - 1]; ! result = (stateP->proc != NULL); } else result = false; *************** *** 590,596 **** SICleanupQueue(bool callerHasWriteLock, int minFree) int n = stateP->nextMsgNum; /* Ignore if inactive or already in reset state */ ! if (stateP->procPid == 0 || stateP->resetState) continue; /* --- 591,597 ---- int n = stateP->nextMsgNum; /* Ignore if inactive or already in reset state */ ! if (stateP->proc == NULL || stateP->resetState) continue; /* *************** *** 644,661 **** SICleanupQueue(bool callerHasWriteLock, int minFree) segP->nextThreshold = (numMsgs / CLEANUP_QUANTUM + 1) * CLEANUP_QUANTUM; /* ! * Lastly, signal anyone who needs a catchup interrupt. Since kill() ! * might not be fast, we don't want to hold locks while executing it. */ if (needSig) { ! pid_t his_pid = needSig->procPid; needSig->signaled = true; LWLockRelease(SInvalReadLock); LWLockRelease(SInvalWriteLock); ! elog(DEBUG4, "sending sinval catchup signal to PID %d", (int) his_pid); ! kill(his_pid, SIGUSR1); if (callerHasWriteLock) LWLockAcquire(SInvalWriteLock, LW_EXCLUSIVE); } --- 645,664 ---- segP->nextThreshold = (numMsgs / CLEANUP_QUANTUM + 1) * CLEANUP_QUANTUM; /* ! * Lastly, signal anyone who needs a catchup interrupt. Since ! * SendProcSignal() might not be fast, we don't want to hold locks while ! * executing it. */ if (needSig) { ! PGPROC *his_proc = needSig->proc; needSig->signaled = true; LWLockRelease(SInvalReadLock); LWLockRelease(SInvalWriteLock); ! elog(DEBUG4, "sending sinval catchup signal to PID %d", ! (int) his_proc->pid); ! SendProcSignal(his_proc, PROCSIG_CATCHUP_INTERRUPT); if (callerHasWriteLock) LWLockAcquire(SInvalWriteLock, LW_EXCLUSIVE); } *** a/src/backend/storage/lmgr/proc.c --- b/src/backend/storage/lmgr/proc.c *************** *** 289,294 **** InitProcess(void) --- 289,295 ---- MyProc->databaseId = InvalidOid; MyProc->roleId = InvalidOid; MyProc->inCommit = false; + MyProc->signalFlags = 0; MyProc->vacuumFlags = 0; if (IsAutoVacuumWorkerProcess()) MyProc->vacuumFlags |= PROC_IS_AUTOVACUUM; *************** *** 428,433 **** InitAuxiliaryProcess(void) --- 429,435 ---- MyProc->databaseId = InvalidOid; MyProc->roleId = InvalidOid; MyProc->inCommit = false; + MyProc->signalFlags = 0; /* we don't set the "is autovacuum" flag in the launcher */ MyProc->vacuumFlags = 0; MyProc->lwWaiting = false; *************** *** 1277,1282 **** ProcSendSignal(int pid) --- 1279,1330 ---- PGSemaphoreUnlock(&proc->sem); } + /* + * SendProcSignal - send the signal with the reason to the process + * (such as backend, autovacuum worker and auxiliary process) + * identified by proc. + */ + void + SendProcSignal(PGPROC *proc, uint8 reason) + { + int pid; + + if (proc == NULL) + return; + + LWLockAcquire(ProcArrayLock, LW_EXCLUSIVE); + pid = proc->pid; + if (pid != 0) + proc->signalFlags |= reason; + LWLockRelease(ProcArrayLock); + + /* Send SIGUSR1 to the process */ + kill(pid, SIGUSR1); + } + + /* + * CheckProcSignal - check to see if the particular reason has been + * signaled, and clear the signal flag. Should be called after + * receiving SIGUSR1. + */ + bool + CheckProcSignal(uint8 reason) + { + LWLockAcquire(ProcArrayLock, LW_EXCLUSIVE); + + /* Careful here --- don't clear flag if we haven't seen it set */ + if (MyProc->signalFlags & reason) + { + MyProc->signalFlags &= ~reason; + LWLockRelease(ProcArrayLock); + return true; + } + + LWLockRelease(ProcArrayLock); + + return false; + } + /***************************************************************************** * SIGALRM interrupt support *** a/src/backend/tcop/postgres.c --- b/src/backend/tcop/postgres.c *************** *** 2437,2442 **** drop_unnamed_stmt(void) --- 2437,2464 ---- */ /* + * proc_sigusr1_handler - handle SIGUSR1 signal. + * + * SIGUSR1 is multiplexed to handle multiple different events. The signalFlags + * bitmask in PGPROC indicates which events have been signaled. + */ + void + proc_sigusr1_handler(SIGNAL_ARGS) + { + int save_errno = errno; + + if (CheckProcSignal(PROCSIG_CATCHUP_INTERRUPT)) + { + /* + * Catchup interrupt has been sent. + */ + HandleCatchupInterrupt(); + } + + errno = save_errno; + } + + /* * quickdie() occurs when signalled SIGQUIT by the postmaster. * * Some backend has bought the farm, *************** *** 3180,3186 **** PostgresMain(int argc, char *argv[], const char *username) * of output during who-knows-what operation... */ pqsignal(SIGPIPE, SIG_IGN); ! pqsignal(SIGUSR1, CatchupInterruptHandler); pqsignal(SIGUSR2, NotifyInterruptHandler); pqsignal(SIGFPE, FloatExceptionHandler); --- 3202,3208 ---- * of output during who-knows-what operation... */ pqsignal(SIGPIPE, SIG_IGN); ! pqsignal(SIGUSR1, proc_sigusr1_handler); pqsignal(SIGUSR2, NotifyInterruptHandler); pqsignal(SIGFPE, FloatExceptionHandler); *** a/src/include/storage/proc.h --- b/src/include/storage/proc.h *************** *** 38,43 **** struct XidCache --- 38,46 ---- TransactionId xids[PGPROC_MAX_CACHED_SUBXIDS]; }; + /* Signals for PGPROC->signalFlags */ + #define PROCSIG_CATCHUP_INTERRUPT 0x01 /* catchup interrupt */ + /* Flags for PGPROC->vacuumFlags */ #define PROC_IS_AUTOVACUUM 0x01 /* is it an autovac worker? */ #define PROC_IN_VACUUM 0x02 /* currently running lazy vacuum */ *************** *** 91,96 **** struct PGPROC --- 94,100 ---- bool inCommit; /* true if within commit critical section */ + uint8 signalFlags; /* bitmask of signals raised, see above */ uint8 vacuumFlags; /* vacuum-related flags, see above */ /* Info about LWLock the process is currently waiting for, if any. */ *************** *** 171,176 **** extern void LockWaitCancel(void); --- 175,183 ---- extern void ProcWaitForSignal(void); extern void ProcSendSignal(int pid); + extern void SendProcSignal(PGPROC *proc, uint8 reason); + extern bool CheckProcSignal(uint8 reason); + extern bool enable_sig_alarm(int delayms, bool is_statement_timeout); extern bool disable_sig_alarm(bool is_statement_timeout); extern void handle_sig_alarm(SIGNAL_ARGS); *** a/src/include/storage/sinval.h --- b/src/include/storage/sinval.h *************** *** 90,96 **** extern void ReceiveSharedInvalidMessages( void (*resetFunction) (void)); /* signal handler for catchup events (SIGUSR1) */ ! extern void CatchupInterruptHandler(SIGNAL_ARGS); /* * enable/disable processing of catchup events directly from signal handler. --- 90,96 ---- void (*resetFunction) (void)); /* signal handler for catchup events (SIGUSR1) */ ! extern void HandleCatchupInterrupt(void); /* * enable/disable processing of catchup events directly from signal handler. *** a/src/include/tcop/tcopprot.h --- b/src/include/tcop/tcopprot.h *************** *** 56,61 **** extern List *pg_plan_queries(List *querytrees, int cursorOptions, --- 56,62 ---- extern bool assign_max_stack_depth(int newval, bool doit, GucSource source); + extern void proc_sigusr1_handler(SIGNAL_ARGS); extern void die(SIGNAL_ARGS); extern void quickdie(SIGNAL_ARGS); extern void authdie(SIGNAL_ARGS);