upstream/postgresql
1
0
Fork 0
mirror of https://github.com/postgres/postgres.git synced 2026-07-09 17:51:00 -04:00
Code Issues Projects Releases Packages Wiki Activity Actions

injection_points: Switch wait/wakeup to rely on atomics

Browse source 
This change switches the implementation of wait and wakeups in the
module injection_points to not rely anymore on condition variables,
using a more primitive implementation based on atomics.  The former
implementation required a PGPROC, making it impossible to inject waits
in the postmaster or during authentication.  A couple of use cases have
popped up for these in the past, where this would have become handy.

The loop in the wait callback that relied on a condition variable is
replaced by an atomic counter, whose check increases over time in an
exponential manner (starts at 10us for quick responsiveness, up to
100ms).

This change may be backpatched at some point depending on how much
testing coverage is wanted.  Let's limit ourselves to HEAD for now,
checking things first with the buildfarm.

Creating a wait still requires the SQL interface.  We are looking at
expanding that with an alternative implementation, so as early startup
or authentication waits would become possible.  This refactoring piece
is mandatory to achieve this goal.

Reviewed-by: Andrey Borodin <x4mmm@yandex-team.ru>
Discussion: https://postgr.es/m/aher0VsjJ8xeNgLq@paquier.xyz
This commit is contained in:
Michael Paquier 2026-07-08 14:34:09 +09:00
parent d92e98340f
commit 8daeaa9b64
1 changed files with 29 additions and 27 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

56
src/test/modules/injection_points/injection_points.c
View file

@ -23,11 +23,11 @@
#include "miscadmin.h"
#include "nodes/pg_list.h"
#include "nodes/value.h"
#include "storage/condition_variable.h"
#include "storage/dsm_registry.h"
#include "storage/ipc.h"
#include "storage/lwlock.h"
#include "storage/shmem.h"
#include "storage/spin.h"
#include "utils/builtins.h"
#include "utils/guc.h"
#include "utils/injection_point.h"
@ -41,6 +41,10 @@ PG_MODULE_MAGIC;
#define INJ_MAX_WAIT 8
#define INJ_NAME_MAXLEN 64
/* Thresholds of waits */
#define INJ_WAIT_INITIAL_US 10 /* 10us */
#define INJ_WAIT_MAX_US 100000 /* 100ms */
/*
* List of injection points stored in TopMemoryContext attached
* locally to this process.
@ -59,13 +63,10 @@ typedef struct InjectionPointSharedState
slock_t lock;
/* Counters advancing when injection_points_wakeup() is called */
uint32 wait_counts[INJ_MAX_WAIT];
pg_atomic_uint32 wait_counts[INJ_MAX_WAIT];
/* Names of injection points attached to wait counters */
char name[INJ_MAX_WAIT][INJ_NAME_MAXLEN];
/* Condition variable used for waits and wakeups */
ConditionVariable wait_point;
} InjectionPointSharedState;
/* Pointer to shared-memory state. */
@ -102,9 +103,9 @@ injection_point_init_state(void *ptr, void *arg)
InjectionPointSharedState *state = (InjectionPointSharedState *) ptr;
SpinLockInit(&state->lock);
memset(state->wait_counts, 0, sizeof(state->wait_counts));
memset(state->name, 0, sizeof(state->name));
ConditionVariableInit(&state->wait_point);
for (int i = 0; i < INJ_MAX_WAIT; i++)
pg_atomic_init_u32(&state->wait_counts[i], 0);
}
static void
@ -222,7 +223,7 @@ injection_notice(const char *name, const void *private_data, void *arg)
elog(NOTICE, "notice triggered for injection point %s", name);
}
/* Wait on a condition variable, awaken by injection_points_wakeup() */
/* Wait until injection_points_wakeup() is called */
void
injection_wait(const char *name, const void *private_data, void *arg)
{
@ -230,6 +231,7 @@ injection_wait(const char *name, const void *private_data, void *arg)
int index = -1;
uint32 injection_wait_event = 0;
const InjectionPointCondition *condition = private_data;
int delay_us = 0;
if (inj_state == NULL)
injection_init_shmem();
@ -254,31 +256,33 @@ injection_wait(const char *name, const void *private_data, void *arg)
{
index = i;
strlcpy(inj_state->name[i], name, INJ_NAME_MAXLEN);
old_wait_counts = inj_state->wait_counts[i];
old_wait_counts = pg_atomic_read_u32(&inj_state->wait_counts[i]);
break;
}
}
SpinLockRelease(&inj_state->lock);
if (index < 0)
elog(ERROR, "could not find free slot for wait of injection point %s ",
elog(ERROR, "could not find free slot for wait of injection point %s",
name);
/* And sleep.. */
ConditionVariablePrepareToSleep(&inj_state->wait_point);
for (;;)
/*
* Wait until the counter is bumped by injection_points_wakeup().
*
* This loop starts with a short delay for responsiveness, enlarged to
* ease the CPU workload in slower environments.
*/
delay_us = INJ_WAIT_INITIAL_US;
pgstat_report_wait_start(injection_wait_event);
while (pg_atomic_read_u32(&inj_state->wait_counts[index]) == old_wait_counts)
{
uint32 new_wait_counts;
SpinLockAcquire(&inj_state->lock);
new_wait_counts = inj_state->wait_counts[index];
SpinLockRelease(&inj_state->lock);
if (old_wait_counts != new_wait_counts)
break;
ConditionVariableSleep(&inj_state->wait_point, injection_wait_event);
CHECK_FOR_INTERRUPTS();
pg_usleep(delay_us);
if (delay_us < INJ_WAIT_MAX_US)
delay_us = Min(delay_us * 2, INJ_WAIT_MAX_US);
}
ConditionVariableCancelSleep();
pgstat_report_wait_end();
/* Remove this injection point from the waiters. */
SpinLockAcquire(&inj_state->lock);
@ -443,7 +447,7 @@ injection_points_wakeup(PG_FUNCTION_ARGS)
if (inj_state == NULL)
injection_init_shmem();
/* First bump the wait counter for the injection point to wake up */
/* Find the injection point then bump its wait counter */
SpinLockAcquire(&inj_state->lock);
for (int i = 0; i < INJ_MAX_WAIT; i++)
{
@ -458,11 +462,9 @@ injection_points_wakeup(PG_FUNCTION_ARGS)
SpinLockRelease(&inj_state->lock);
elog(ERROR, "could not find injection point %s to wake up", name);
}
inj_state->wait_counts[index]++;
SpinLockRelease(&inj_state->lock);
/* And broadcast the change to the waiters */
ConditionVariableBroadcast(&inj_state->wait_point);
pg_atomic_fetch_add_u32(&inj_state->wait_counts[index], 1);
PG_RETURN_VOID();
}