From 87fa57753cb667ebcabee48b3fead835de72d6e1 Mon Sep 17 00:00:00 2001 From: Petr Jelinek Date: Sun, 30 Dec 2018 02:07:12 +0100 Subject: [PATCH 1/2] Synchronize logical replication slots from primary to standby --- src/backend/commands/subscriptioncmds.c | 4 +- src/backend/postmaster/bgworker.c | 3 + src/backend/postmaster/pgstat.c | 3 + .../libpqwalreceiver/libpqwalreceiver.c | 74 ++++ src/backend/replication/logical/Makefile | 3 +- src/backend/replication/logical/launcher.c | 207 ++++++++---- src/backend/replication/logical/slotsync.c | 316 ++++++++++++++++++ src/backend/replication/logical/tablesync.c | 12 +- src/backend/replication/repl_gram.y | 13 +- src/backend/replication/repl_scanner.l | 1 + src/backend/replication/slotfuncs.c | 2 +- src/backend/replication/walsender.c | 164 +++++++++ src/include/nodes/nodes.h | 1 + src/include/nodes/replnodes.h | 8 + src/include/pgstat.h | 1 + src/include/replication/slot.h | 3 + src/include/replication/walreceiver.h | 12 + src/include/replication/worker_internal.h | 8 +- 18 files changed, 764 insertions(+), 71 deletions(-) create mode 100644 src/backend/replication/logical/slotsync.c diff --git a/src/backend/commands/subscriptioncmds.c b/src/backend/commands/subscriptioncmds.c index 9021463a4c..bec6d11457 100644 --- a/src/backend/commands/subscriptioncmds.c +++ b/src/backend/commands/subscriptioncmds.c @@ -598,7 +598,7 @@ AlterSubscription_refresh(Subscription *sub, bool copy_data) { RemoveSubscriptionRel(sub->oid, relid); - logicalrep_worker_stop_at_commit(sub->oid, relid); + logicalrep_worker_stop_at_commit(MyDatabaseId, sub->oid, relid); ereport(DEBUG1, (errmsg("table \"%s.%s\" removed from subscription \"%s\"", @@ -940,7 +940,7 @@ DropSubscription(DropSubscriptionStmt *stmt, bool isTopLevel) { LogicalRepWorker *w = (LogicalRepWorker *) lfirst(lc); - logicalrep_worker_stop(w->subid, w->relid); + logicalrep_worker_stop(w->dbid, w->subid, w->relid); } list_free(subworkers); diff --git a/src/backend/postmaster/bgworker.c b/src/backend/postmaster/bgworker.c index d2b695e146..3b2a5d04cd 100644 --- a/src/backend/postmaster/bgworker.c +++ b/src/backend/postmaster/bgworker.c @@ -129,6 +129,9 @@ static const struct }, { "ApplyWorkerMain", ApplyWorkerMain + }, + { + "ReplSlotSyncMain", ReplSlotSyncMain } }; diff --git a/src/backend/postmaster/pgstat.c b/src/backend/postmaster/pgstat.c index 8676088e57..5387f4f2fa 100644 --- a/src/backend/postmaster/pgstat.c +++ b/src/backend/postmaster/pgstat.c @@ -3494,6 +3494,9 @@ pgstat_get_wait_activity(WaitEventActivity w) case WAIT_EVENT_LOGICAL_LAUNCHER_MAIN: event_name = "LogicalLauncherMain"; break; + case WAIT_EVENT_REPL_SLOT_SYNC_MAIN: + event_name = "ReplSlotSyncMain"; + break; case WAIT_EVENT_PGSTAT_MAIN: event_name = "PgStatMain"; break; diff --git a/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c b/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c index 9b75711ebd..3c77a76ea6 100644 --- a/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c +++ b/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c @@ -23,6 +23,7 @@ #include "pqexpbuffer.h" #include "access/xlog.h" #include "catalog/pg_type.h" +#include "commands/dbcommands.h" #include "funcapi.h" #include "mb/pg_wchar.h" #include "miscadmin.h" @@ -58,6 +59,7 @@ static void libpqrcv_get_senderinfo(WalReceiverConn *conn, static char *libpqrcv_identify_system(WalReceiverConn *conn, TimeLineID *primary_tli, int *server_version); +static List *libpqrcv_list_slots(WalReceiverConn *conn); static void libpqrcv_readtimelinehistoryfile(WalReceiverConn *conn, TimeLineID tli, char **filename, char **content, int *len); @@ -86,6 +88,7 @@ static WalReceiverFunctionsType PQWalReceiverFunctions = { libpqrcv_get_conninfo, libpqrcv_get_senderinfo, libpqrcv_identify_system, + libpqrcv_list_slots, libpqrcv_readtimelinehistoryfile, libpqrcv_startstreaming, libpqrcv_endstreaming, @@ -348,6 +351,77 @@ libpqrcv_identify_system(WalReceiverConn *conn, TimeLineID *primary_tli, return primary_sysid; } +/* + * Get list of slots from primary. + */ +static List * +libpqrcv_list_slots(WalReceiverConn *conn) +{ + PGresult *res; + int i; + List *slots = NIL; + int ntuples; + WalRecvReplicationSlotData *slot_data; + + res = libpqrcv_PQexec(conn->streamConn, "LIST_SLOTS"); + if (PQresultStatus(res) != PGRES_TUPLES_OK) + { + PQclear(res); + ereport(ERROR, + (errmsg("could not receive list of slots the primary server: %s", + pchomp(PQerrorMessage(conn->streamConn))))); + } + if (PQnfields(res) < 10) + { + int nfields = PQnfields(res); + + PQclear(res); + ereport(ERROR, + (errmsg("invalid response from primary server"), + errdetail("Could not get list of slots: got %d fields, expected %d or more fields.", + nfields, 10))); + } + + ntuples = PQntuples(res); + for (i = 0; i < ntuples; i++) + { + char *slot_type; + + slot_data = palloc0(sizeof(WalRecvReplicationSlotData)); + namestrcpy(&slot_data->name, PQgetvalue(res, i, 0)); + if (!PQgetisnull(res, i, 1)) + namestrcpy(&slot_data->plugin, PQgetvalue(res, i, 1)); + slot_type = PQgetvalue(res, i, 2); + if (!PQgetisnull(res, i, 3)) + slot_data->database = atooid(PQgetvalue(res, i, 3)); + if (strcmp(slot_type, "physical") == 0) + { + if (OidIsValid(slot_data->database)) + elog(ERROR, "unexpected physical replication slot with database set"); + } + if (pg_strtoint32(PQgetvalue(res, i, 5)) == 1) + slot_data->persistency = RS_TEMPORARY; + else + slot_data->persistency = RS_PERSISTENT; + if (!PQgetisnull(res, i, 6)) + slot_data->xmin = atooid(PQgetvalue(res, i, 6)); + if (!PQgetisnull(res, i, 7)) + slot_data->catalog_xmin = atooid(PQgetvalue(res, i, 7)); + if (!PQgetisnull(res, i, 8)) + slot_data->restart_lsn = pg_strtouint64(PQgetvalue(res, i, 8), + NULL, 10); + if (!PQgetisnull(res, i, 9)) + slot_data->confirmed_flush = pg_strtouint64(PQgetvalue(res, i, 9), + NULL, 10); + + slots = lappend(slots, slot_data); + } + + PQclear(res); + + return slots; +} + /* * Start streaming WAL data from given streaming options. * diff --git a/src/backend/replication/logical/Makefile b/src/backend/replication/logical/Makefile index bb417b042e..385782bd67 100644 --- a/src/backend/replication/logical/Makefile +++ b/src/backend/replication/logical/Makefile @@ -15,6 +15,7 @@ include $(top_builddir)/src/Makefile.global override CPPFLAGS := -I$(srcdir) $(CPPFLAGS) OBJS = decode.o launcher.o logical.o logicalfuncs.o message.o origin.o \ - proto.o relation.o reorderbuffer.o snapbuild.o tablesync.o worker.o + proto.o relation.o reorderbuffer.o slotsync.o snapbuild.o \ + tablesync.o worker.o include $(top_srcdir)/src/backend/common.mk diff --git a/src/backend/replication/logical/launcher.c b/src/backend/replication/logical/launcher.c index 3a84d8ca86..c47d4ee403 100644 --- a/src/backend/replication/logical/launcher.c +++ b/src/backend/replication/logical/launcher.c @@ -26,6 +26,7 @@ #include "access/htup_details.h" #include "access/xact.h" +#include "catalog/pg_authid.h" #include "catalog/pg_subscription.h" #include "catalog/pg_subscription_rel.h" @@ -75,6 +76,7 @@ LogicalRepCtxStruct *LogicalRepCtx; typedef struct LogicalRepWorkerId { + Oid dbid; Oid subid; Oid relid; } LogicalRepWorkerId; @@ -239,7 +241,7 @@ WaitForReplicationWorkerAttach(LogicalRepWorker *worker, * subscription id and relid. */ LogicalRepWorker * -logicalrep_worker_find(Oid subid, Oid relid, bool only_running) +logicalrep_worker_find(Oid dbid, Oid subid, Oid relid, bool only_running) { int i; LogicalRepWorker *res = NULL; @@ -251,8 +253,8 @@ logicalrep_worker_find(Oid subid, Oid relid, bool only_running) { LogicalRepWorker *w = &LogicalRepCtx->workers[i]; - if (w->in_use && w->subid == subid && w->relid == relid && - (!only_running || w->proc)) + if (w->in_use && w->dbid == dbid && w->subid == subid && + w->relid == relid && (!only_running || w->proc)) { res = w; break; @@ -302,9 +304,13 @@ logicalrep_worker_launch(Oid dbid, Oid subid, const char *subname, Oid userid, int nsyncworkers; TimestampTz now; - ereport(DEBUG1, - (errmsg("starting logical replication worker for subscription \"%s\"", - subname))); + if (OidIsValid(subid)) + ereport(DEBUG1, + (errmsg("starting logical replication worker for subscription \"%s\"", + subname))); + else + ereport(DEBUG1, + (errmsg("starting replication slot synchronization worker"))); /* Report this after the initial starting message for consistency. */ if (max_replication_slots == 0) @@ -341,7 +347,9 @@ retry: * reason we do this is because if some worker failed to start up and its * parent has crashed while waiting, the in_use state was never cleared. */ - if (worker == NULL || nsyncworkers >= max_sync_workers_per_subscription) + if (worker == NULL || + (OidIsValid(relid) && + nsyncworkers >= max_sync_workers_per_subscription)) { bool did_cleanup = false; @@ -375,7 +383,7 @@ retry: * silently as we might get here because of an otherwise harmless race * condition. */ - if (nsyncworkers >= max_sync_workers_per_subscription) + if (OidIsValid(relid) && nsyncworkers >= max_sync_workers_per_subscription) { LWLockRelease(LogicalRepWorkerLock); return; @@ -421,15 +429,22 @@ retry: memset(&bgw, 0, sizeof(bgw)); bgw.bgw_flags = BGWORKER_SHMEM_ACCESS | BGWORKER_BACKEND_DATABASE_CONNECTION; - bgw.bgw_start_time = BgWorkerStart_RecoveryFinished; + bgw.bgw_start_time = BgWorkerStart_ConsistentState; snprintf(bgw.bgw_library_name, BGW_MAXLEN, "postgres"); - snprintf(bgw.bgw_function_name, BGW_MAXLEN, "ApplyWorkerMain"); + if (OidIsValid(subid)) + snprintf(bgw.bgw_function_name, BGW_MAXLEN, "ApplyWorkerMain"); + else + snprintf(bgw.bgw_function_name, BGW_MAXLEN, "ReplSlotSyncMain"); if (OidIsValid(relid)) snprintf(bgw.bgw_name, BGW_MAXLEN, "logical replication worker for subscription %u sync %u", subid, relid); - else + else if (OidIsValid(subid)) snprintf(bgw.bgw_name, BGW_MAXLEN, "logical replication worker for subscription %u", subid); + else + snprintf(bgw.bgw_name, BGW_MAXLEN, + "replication slot synchronization worker"); + snprintf(bgw.bgw_type, BGW_MAXLEN, "logical replication worker"); bgw.bgw_restart_time = BGW_NEVER_RESTART; @@ -460,14 +475,14 @@ retry: * it detaches from the slot. */ void -logicalrep_worker_stop(Oid subid, Oid relid) +logicalrep_worker_stop(Oid dbid, Oid subid, Oid relid) { LogicalRepWorker *worker; uint16 generation; LWLockAcquire(LogicalRepWorkerLock, LW_SHARED); - worker = logicalrep_worker_find(subid, relid, false); + worker = logicalrep_worker_find(dbid, subid, relid, false); /* No worker, nothing to do. */ if (!worker) @@ -557,7 +572,7 @@ logicalrep_worker_stop(Oid subid, Oid relid) * Request worker for specified sub/rel to be stopped on commit. */ void -logicalrep_worker_stop_at_commit(Oid subid, Oid relid) +logicalrep_worker_stop_at_commit(Oid dbid, Oid subid, Oid relid) { int nestDepth = GetCurrentTransactionNestLevel(); LogicalRepWorkerId *wid; @@ -590,6 +605,7 @@ logicalrep_worker_stop_at_commit(Oid subid, Oid relid) * subtransaction. */ wid = palloc(sizeof(LogicalRepWorkerId)); + wid->dbid = dbid; wid->subid = subid; wid->relid = relid; on_commit_stop_workers->workers = @@ -602,13 +618,13 @@ logicalrep_worker_stop_at_commit(Oid subid, Oid relid) * Wake up (using latch) any logical replication worker for specified sub/rel. */ void -logicalrep_worker_wakeup(Oid subid, Oid relid) +logicalrep_worker_wakeup(Oid dbid, Oid subid, Oid relid) { LogicalRepWorker *worker; LWLockAcquire(LogicalRepWorkerLock, LW_SHARED); - worker = logicalrep_worker_find(subid, relid, true); + worker = logicalrep_worker_find(dbid, subid, relid, true); if (worker) logicalrep_worker_wakeup_ptr(worker); @@ -795,7 +811,7 @@ ApplyLauncherRegister(void) memset(&bgw, 0, sizeof(bgw)); bgw.bgw_flags = BGWORKER_SHMEM_ACCESS | BGWORKER_BACKEND_DATABASE_CONNECTION; - bgw.bgw_start_time = BgWorkerStart_RecoveryFinished; + bgw.bgw_start_time = BgWorkerStart_ConsistentState; snprintf(bgw.bgw_library_name, BGW_MAXLEN, "postgres"); snprintf(bgw.bgw_function_name, BGW_MAXLEN, "ApplyLauncherMain"); snprintf(bgw.bgw_name, BGW_MAXLEN, @@ -873,7 +889,7 @@ AtEOXact_ApplyLauncher(bool isCommit) { LogicalRepWorkerId *wid = lfirst(lc); - logicalrep_worker_stop(wid->subid, wid->relid); + logicalrep_worker_stop(wid->dbid, wid->subid, wid->relid); } } @@ -964,6 +980,115 @@ ApplyLauncherWakeup(void) kill(LogicalRepCtx->launcher_pid, SIGUSR1); } +static void +ApplyLauncherStartSlotSync(TimestampTz *last_start_time, long *wait_time) +{ + TimestampTz now; + char *err; + List *slots; + ListCell *lc; + MemoryContext tmpctx; + MemoryContext oldctx; + + if (numsynchronize_slot_names == 0) + return; + + wrconn = walrcv_connect(PrimaryConnInfo, false, + "Logical Replication Launcher", &err); + if (!wrconn) + ereport(ERROR, + (errmsg("could not connect to the primary server: %s", err))); + + /* Use temporary context for the slot list and worker info. */ + tmpctx = AllocSetContextCreate(TopMemoryContext, + "Logical Replication Launcher slot sync ctx", + ALLOCSET_DEFAULT_SIZES); + oldctx = MemoryContextSwitchTo(tmpctx); + + slots = walrcv_list_slots(wrconn); + + now = GetCurrentTimestamp(); + + foreach (lc, slots) + { + WalRecvReplicationSlotData *slot_data = lfirst(lc); + LogicalRepWorker *w; + + if (!OidIsValid(slot_data->database)) + continue; + + LWLockAcquire(LogicalRepWorkerLock, LW_SHARED); + w = logicalrep_worker_find(slot_data->database, InvalidOid, + InvalidOid, false); + LWLockRelease(LogicalRepWorkerLock); + + if (w == NULL) + { + *last_start_time = now; + *wait_time = wal_retrieve_retry_interval; + + logicalrep_worker_launch(slot_data->database, InvalidOid, NULL, + BOOTSTRAP_SUPERUSERID, InvalidOid); + } + } + + /* Switch back to original memory context. */ + MemoryContextSwitchTo(oldctx); + /* Clean the temporary memory. */ + MemoryContextDelete(tmpctx); + + walrcv_disconnect(wrconn); +} + +static void +ApplyLauncherStartSubs(TimestampTz *last_start_time, long *wait_time) +{ + TimestampTz now; + List *sublist; + ListCell *lc; + MemoryContext subctx; + MemoryContext oldctx; + + now = GetCurrentTimestamp(); + + /* Use temporary context for the database list and worker info. */ + subctx = AllocSetContextCreate(TopMemoryContext, + "Logical Replication Launcher sublist", + ALLOCSET_DEFAULT_SIZES); + oldctx = MemoryContextSwitchTo(subctx); + + /* search for subscriptions to start or stop. */ + sublist = get_subscription_list(); + + /* Start the missing workers for enabled subscriptions. */ + foreach(lc, sublist) + { + Subscription *sub = (Subscription *) lfirst(lc); + LogicalRepWorker *w; + + if (!sub->enabled) + continue; + + LWLockAcquire(LogicalRepWorkerLock, LW_SHARED); + w = logicalrep_worker_find(sub->dbid, sub->oid, InvalidOid, false); + LWLockRelease(LogicalRepWorkerLock); + + if (w == NULL) + { + *last_start_time = now; + *wait_time = wal_retrieve_retry_interval; + + logicalrep_worker_launch(sub->dbid, sub->oid, sub->name, + sub->owner, InvalidOid); + } + } + + /* Switch back to original memory context. */ + MemoryContextSwitchTo(oldctx); + /* Clean the temporary memory. */ + MemoryContextDelete(subctx); +} + /* * Main loop for the apply launcher process. */ @@ -991,14 +1116,12 @@ ApplyLauncherMain(Datum main_arg) */ BackgroundWorkerInitializeConnection(NULL, NULL, 0); + load_file("libpqwalreceiver", false); + /* Enter main loop */ for (;;) { int rc; - List *sublist; - ListCell *lc; - MemoryContext subctx; - MemoryContext oldctx; TimestampTz now; long wait_time = DEFAULT_NAPTIME_PER_CYCLE; @@ -1010,42 +1133,10 @@ ApplyLauncherMain(Datum main_arg) if (TimestampDifferenceExceeds(last_start_time, now, wal_retrieve_retry_interval)) { - /* Use temporary context for the database list and worker info. */ - subctx = AllocSetContextCreate(TopMemoryContext, - "Logical Replication Launcher sublist", - ALLOCSET_DEFAULT_SIZES); - oldctx = MemoryContextSwitchTo(subctx); - - /* search for subscriptions to start or stop. */ - sublist = get_subscription_list(); - - /* Start the missing workers for enabled subscriptions. */ - foreach(lc, sublist) - { - Subscription *sub = (Subscription *) lfirst(lc); - LogicalRepWorker *w; - - if (!sub->enabled) - continue; - - LWLockAcquire(LogicalRepWorkerLock, LW_SHARED); - w = logicalrep_worker_find(sub->oid, InvalidOid, false); - LWLockRelease(LogicalRepWorkerLock); - - if (w == NULL) - { - last_start_time = now; - wait_time = wal_retrieve_retry_interval; - - logicalrep_worker_launch(sub->dbid, sub->oid, sub->name, - sub->owner, InvalidOid); - } - } - - /* Switch back to original memory context. */ - MemoryContextSwitchTo(oldctx); - /* Clean the temporary memory. */ - MemoryContextDelete(subctx); + if (!RecoveryInProgress()) + ApplyLauncherStartSubs(&last_start_time, &wait_time); + else + ApplyLauncherStartSlotSync(&last_start_time, &wait_time); } else { diff --git a/src/backend/replication/logical/slotsync.c b/src/backend/replication/logical/slotsync.c new file mode 100644 index 0000000000..eda9ad0add --- /dev/null +++ b/src/backend/replication/logical/slotsync.c @@ -0,0 +1,316 @@ +/*------------------------------------------------------------------------- + * slotsync.c + * PostgreSQL worker for synchronizing slots to a standby from primary + * + * Copyright (c) 2016-2018, PostgreSQL Global Development Group + * + * IDENTIFICATION + * src/backend/replication/logical/slotsync.c + * + *------------------------------------------------------------------------- + */ +#include "postgres.h" + +#include "access/xact.h" +#include "catalog/pg_type_d.h" +#include "commands/dbcommands.h" +#include "miscadmin.h" +#include "pgstat.h" +#include "postmaster/bgworker.h" +#include "replication/logicalworker.h" +#include "replication/slot.h" +#include "replication/walreceiver.h" +#include "replication/worker_internal.h" +#include "storage/ipc.h" +#include "storage/procarray.h" +#include "utils/builtins.h" +#include "utils/guc.h" +#include "utils/pg_lsn.h" + + +/* + * Wait for remote slot to pass localy reserved position. + */ +static void +wait_for_primary_slot_catchup(WalReceiverConn *wrconn, char *slot_name, + XLogRecPtr min_lsn) +{ + WalRcvExecResult *res; + TupleTableSlot *slot; + Oid slotRow[1] = {LSNOID}; + StringInfoData cmd; + bool isnull; + XLogRecPtr restart_lsn; + + for (;;) + { + int rc; + + CHECK_FOR_INTERRUPTS(); + + initStringInfo(&cmd); + appendStringInfo(&cmd, + "SELECT restart_lsn" + " FROM pg_catalog.pg_replication_slots" + " WHERE slot_name = %s", + quote_literal_cstr(slot_name)); + res = walrcv_exec(wrconn, cmd.data, 1, slotRow); + + if (res->status != WALRCV_OK_TUPLES) + ereport(ERROR, + (errmsg("could not fetch slot info for slot \"%s\" from primary: %s", + slot_name, res->err))); + + slot = MakeSingleTupleTableSlot(res->tupledesc, &TTSOpsMinimalTuple); + if (!tuplestore_gettupleslot(res->tuplestore, true, false, slot)) + ereport(ERROR, + (errmsg("slot \"%s\" disapeared from provider", + slot_name))); + + restart_lsn = DatumGetLSN(slot_getattr(slot, 1, &isnull)); + Assert(!isnull); + + ExecClearTuple(slot); + walrcv_clear_result(res); + + if (restart_lsn >= min_lsn) + break; + + rc = WaitLatch(MyLatch, + WL_LATCH_SET | WL_TIMEOUT | WL_POSTMASTER_DEATH, + wal_retrieve_retry_interval, + WAIT_EVENT_REPL_SLOT_SYNC_MAIN); + + ResetLatch(MyLatch); + + /* emergency bailout if postmaster has died */ + if (rc & WL_POSTMASTER_DEATH) + proc_exit(1); + } +} + +/* + * Synchronize single slot to given position. + * + * This optionally creates new slot if there is no existing one. + */ +static void +synchronize_one_slot(WalReceiverConn *wrconn, char *slot_name, char *database, + char *plugin_name, XLogRecPtr target_lsn) +{ + int i; + bool found = false; + XLogRecPtr endlsn; + + /* Search for the named slot and mark it active if we find it. */ + LWLockAcquire(ReplicationSlotControlLock, LW_SHARED); + for (i = 0; i < max_replication_slots; i++) + { + ReplicationSlot *s = &ReplicationSlotCtl->replication_slots[i]; + + if (!s->in_use) + continue; + + if (strcmp(NameStr(s->data.name), slot_name) == 0) + { + found = true; + break; + } + } + LWLockRelease(ReplicationSlotControlLock); + + StartTransactionCommand(); + + /* Already existing slot, acquire */ + if (found) + { + ReplicationSlotAcquire(slot_name, true); + + if (target_lsn < MyReplicationSlot->data.confirmed_flush) + { + elog(DEBUG1, + "not synchronizing slot %s; synchronization would move it backward", + slot_name); + + ReplicationSlotRelease(); + CommitTransactionCommand(); + return; + } + } + /* Otherwise create the slot first. */ + else + { + TransactionId xmin_horizon = InvalidTransactionId; + ReplicationSlot *slot; + + ReplicationSlotCreate(slot_name, true, RS_EPHEMERAL); + slot = MyReplicationSlot; + + SpinLockAcquire(&slot->mutex); + slot->data.database = get_database_oid(database, false); + StrNCpy(NameStr(slot->data.plugin), plugin_name, NAMEDATALEN); + SpinLockRelease(&slot->mutex); + + ReplicationSlotReserveWal(); + + LWLockAcquire(ProcArrayLock, LW_EXCLUSIVE); + xmin_horizon = GetOldestSafeDecodingTransactionId(true); + slot->effective_catalog_xmin = xmin_horizon; + slot->data.catalog_xmin = xmin_horizon; + ReplicationSlotsComputeRequiredXmin(true); + LWLockRelease(ProcArrayLock); + + if (target_lsn < MyReplicationSlot->data.restart_lsn) + { + elog(LOG, "waiting for remote slot %s lsn (%X/%X) to pass local slot lsn (%X/%X)", + slot_name, + (uint32) (target_lsn >> 32), + (uint32) (target_lsn), + (uint32) (MyReplicationSlot->data.restart_lsn >> 32), + (uint32) (MyReplicationSlot->data.restart_lsn)); + + wait_for_primary_slot_catchup(wrconn, slot_name, + MyReplicationSlot->data.restart_lsn); + } + + ReplicationSlotPersist(); + } + + endlsn = pg_logical_replication_slot_advance(target_lsn); + + elog(DEBUG3, "synchronized slot %s to lsn (%X/%X)", + slot_name, (uint32) (endlsn >> 32), (uint32) (endlsn)); + + ReplicationSlotRelease(); + CommitTransactionCommand(); +} + +static void +synchronize_slots(void) +{ + WalRcvExecResult *res; + WalReceiverConn *wrconn = NULL; + TupleTableSlot *slot; + Oid slotRow[3] = {TEXTOID, TEXTOID, LSNOID}; + StringInfoData s; + char *database; + char *err; + MemoryContext oldctx = CurrentMemoryContext; + + if (!WalRcv) + return; + + /* syscache access needs a transaction env. */ + StartTransactionCommand(); + /* make dbname live outside TX context */ + MemoryContextSwitchTo(oldctx); + + database = get_database_name(MyDatabaseId); + initStringInfo(&s); + appendStringInfo(&s, "%s dbname=%s", PrimaryConnInfo, database); + wrconn = walrcv_connect(s.data, true, "slot_sync", &err); + + if (wrconn == NULL) + ereport(ERROR, + (errmsg("could not connect to the primary server: %s", err))); + + resetStringInfo(&s); + /* TODO filter slot names? */ + appendStringInfo(&s, + "SELECT slot_name, plugin, confirmed_flush_lsn" + " FROM pg_catalog.pg_replication_slots" + " WHERE database = %s", + quote_literal_cstr(database)); + res = walrcv_exec(wrconn, s.data, 3, slotRow); + pfree(s.data); + + if (res->status != WALRCV_OK_TUPLES) + ereport(ERROR, + (errmsg("could not fetch slot info from primary: %s", + res->err))); + + CommitTransactionCommand(); + /* CommitTransactionCommand switches to TopMemoryContext */ + MemoryContextSwitchTo(oldctx); + + slot = MakeSingleTupleTableSlot(res->tupledesc, &TTSOpsMinimalTuple); + while (tuplestore_gettupleslot(res->tuplestore, true, false, slot)) + { + char *slot_name; + char *plugin_name; + XLogRecPtr confirmed_flush_lsn; + bool isnull; + + slot_name = TextDatumGetCString(slot_getattr(slot, 1, &isnull)); + Assert(!isnull); + + plugin_name = TextDatumGetCString(slot_getattr(slot, 2, &isnull)); + Assert(!isnull); + + confirmed_flush_lsn = DatumGetLSN(slot_getattr(slot, 3, &isnull)); + Assert(!isnull); + + synchronize_one_slot(wrconn, slot_name, database, plugin_name, + confirmed_flush_lsn); + + ExecClearTuple(slot); + } + + walrcv_clear_result(res); + pfree(database); + + walrcv_disconnect(wrconn); +} + +/* + * The main loop of our worker process. + */ +void +ReplSlotSyncMain(Datum main_arg) +{ + int worker_slot = DatumGetInt32(main_arg); + + /* Attach to slot */ + logicalrep_worker_attach(worker_slot); + + /* Establish signal handlers. */ + BackgroundWorkerUnblockSignals(); + + /* Load the libpq-specific functions */ + load_file("libpqwalreceiver", false); + + /* Connect to our database. */ + BackgroundWorkerInitializeConnectionByOid(MyLogicalRepWorker->dbid, + MyLogicalRepWorker->userid, + 0); + + StartTransactionCommand(); + ereport(LOG, + (errmsg("replication slot synchronization worker for database \"%s\" has started", + get_database_name(MyLogicalRepWorker->dbid)))); + CommitTransactionCommand(); + + /* Main wait loop. */ + for (;;) + { + int rc; + + CHECK_FOR_INTERRUPTS(); + + if (!RecoveryInProgress()) + return; + + synchronize_slots(); + + rc = WaitLatch(MyLatch, + WL_LATCH_SET | WL_TIMEOUT | WL_POSTMASTER_DEATH, + wal_retrieve_retry_interval, + WAIT_EVENT_REPL_SLOT_SYNC_MAIN); + + ResetLatch(MyLatch); + + /* emergency bailout if postmaster has died */ + if (rc & WL_POSTMASTER_DEATH) + proc_exit(1); + } +} diff --git a/src/backend/replication/logical/tablesync.c b/src/backend/replication/logical/tablesync.c index 38ae1b9ab8..816468a782 100644 --- a/src/backend/replication/logical/tablesync.c +++ b/src/backend/replication/logical/tablesync.c @@ -141,7 +141,8 @@ finish_sync_worker(void) CommitTransactionCommand(); /* Find the main apply worker and signal it. */ - logicalrep_worker_wakeup(MyLogicalRepWorker->subid, InvalidOid); + logicalrep_worker_wakeup(MyLogicalRepWorker->dbid, + MyLogicalRepWorker->subid, InvalidOid); /* Stop gracefully */ proc_exit(0); @@ -184,7 +185,8 @@ wait_for_relation_state_change(Oid relid, char expected_state) LWLockAcquire(LogicalRepWorkerLock, LW_SHARED); /* Check if the opposite worker is still running and bail if not. */ - worker = logicalrep_worker_find(MyLogicalRepWorker->subid, + worker = logicalrep_worker_find(MyLogicalRepWorker->dbid, + MyLogicalRepWorker->subid, am_tablesync_worker() ? InvalidOid : relid, false); LWLockRelease(LogicalRepWorkerLock); @@ -232,7 +234,8 @@ wait_for_worker_state_change(char expected_state) * waiting. */ LWLockAcquire(LogicalRepWorkerLock, LW_SHARED); - worker = logicalrep_worker_find(MyLogicalRepWorker->subid, + worker = logicalrep_worker_find(MyLogicalRepWorker->dbid, + MyLogicalRepWorker->subid, InvalidOid, false); if (worker && worker->proc) logicalrep_worker_wakeup_ptr(worker); @@ -432,7 +435,8 @@ process_syncing_tables_for_apply(XLogRecPtr current_lsn) */ LWLockAcquire(LogicalRepWorkerLock, LW_SHARED); - syncworker = logicalrep_worker_find(MyLogicalRepWorker->subid, + syncworker = logicalrep_worker_find(MyLogicalRepWorker->dbid, + MyLogicalRepWorker->subid, rstate->relid, false); if (syncworker) diff --git a/src/backend/replication/repl_gram.y b/src/backend/replication/repl_gram.y index 843a878ff3..95d91bf236 100644 --- a/src/backend/replication/repl_gram.y +++ b/src/backend/replication/repl_gram.y @@ -87,11 +87,12 @@ static SQLCmd *make_sqlcmd(void); %token K_EXPORT_SNAPSHOT %token K_NOEXPORT_SNAPSHOT %token K_USE_SNAPSHOT +%token K_LIST_SLOTS %type command %type base_backup start_replication start_logical_replication create_replication_slot drop_replication_slot identify_system - timeline_history show sql_cmd + timeline_history show sql_cmd list_slots %type base_backup_opt_list %type base_backup_opt %type opt_timeline @@ -124,6 +125,7 @@ command: | drop_replication_slot | timeline_history | show + | list_slots | sql_cmd ; @@ -136,6 +138,15 @@ identify_system: $$ = (Node *) makeNode(IdentifySystemCmd); } ; +/* + * LIST_SLOTS + */ +list_slots: + K_LIST_SLOTS + { + $$ = (Node *) makeNode(ListSlotsCmd); + } + ; /* * SHOW setting diff --git a/src/backend/replication/repl_scanner.l b/src/backend/replication/repl_scanner.l index 7bb501f594..2d8accd985 100644 --- a/src/backend/replication/repl_scanner.l +++ b/src/backend/replication/repl_scanner.l @@ -85,6 +85,7 @@ identifier {ident_start}{ident_cont}* BASE_BACKUP { return K_BASE_BACKUP; } FAST { return K_FAST; } IDENTIFY_SYSTEM { return K_IDENTIFY_SYSTEM; } +LIST_SLOTS { return K_LIST_SLOTS; } SHOW { return K_SHOW; } LABEL { return K_LABEL; } NOWAIT { return K_NOWAIT; } diff --git a/src/backend/replication/slotfuncs.c b/src/backend/replication/slotfuncs.c index 8782bad4a2..729935fc60 100644 --- a/src/backend/replication/slotfuncs.c +++ b/src/backend/replication/slotfuncs.c @@ -351,7 +351,7 @@ pg_physical_replication_slot_advance(XLogRecPtr moveto) * WAL and removal of old catalog tuples. As decoding is done in fast_forward * mode, no changes are generated anyway. */ -static XLogRecPtr +XLogRecPtr pg_logical_replication_slot_advance(XLogRecPtr moveto) { LogicalDecodingContext *ctx; diff --git a/src/backend/replication/walsender.c b/src/backend/replication/walsender.c index d1a8113cb6..6ca304b8f5 100644 --- a/src/backend/replication/walsender.c +++ b/src/backend/replication/walsender.c @@ -426,6 +426,166 @@ IdentifySystem(void) end_tup_output(tstate); } +/* + * Handle the LIST_SLOTS command. + */ +static void +ListSlots(void) +{ + DestReceiver *dest; + TupOutputState *tstate; + TupleDesc tupdesc; + int slotno; + + dest = CreateDestReceiver(DestRemoteSimple); + + /* need a tuple descriptor representing four columns */ + tupdesc = CreateTemplateTupleDesc(10); + TupleDescInitBuiltinEntry(tupdesc, (AttrNumber) 1, "slot_name", + TEXTOID, -1, 0); + TupleDescInitBuiltinEntry(tupdesc, (AttrNumber) 2, "plugin", + TEXTOID, -1, 0); + TupleDescInitBuiltinEntry(tupdesc, (AttrNumber) 3, "slot_type", + TEXTOID, -1, 0); + TupleDescInitBuiltinEntry(tupdesc, (AttrNumber) 4, "datoid", + INT8OID, -1, 0); + TupleDescInitBuiltinEntry(tupdesc, (AttrNumber) 5, "database", + TEXTOID, -1, 0); + TupleDescInitBuiltinEntry(tupdesc, (AttrNumber) 6, "temporary", + INT4OID, -1, 0); + TupleDescInitBuiltinEntry(tupdesc, (AttrNumber) 7, "xmin", + INT8OID, -1, 0); + TupleDescInitBuiltinEntry(tupdesc, (AttrNumber) 8, "catalog_xmin", + INT8OID, -1, 0); + TupleDescInitBuiltinEntry(tupdesc, (AttrNumber) 9, "restart_lsn", + TEXTOID, -1, 0); + TupleDescInitBuiltinEntry(tupdesc, (AttrNumber) 10, "confirmed_flush", + TEXTOID, -1, 0); + + /* prepare for projection of tuples */ + tstate = begin_tup_output_tupdesc(dest, tupdesc, &TTSOpsVirtual); + + LWLockAcquire(ReplicationSlotControlLock, LW_SHARED); + for (slotno = 0; slotno < max_replication_slots; slotno++) + { + ReplicationSlot *slot = &ReplicationSlotCtl->replication_slots[slotno]; + char restart_lsn_str[MAXFNAMELEN]; + char confirmed_flush_lsn_str[MAXFNAMELEN]; + Datum values[10]; + bool nulls[10]; + + ReplicationSlotPersistency persistency; + TransactionId xmin; + TransactionId catalog_xmin; + XLogRecPtr restart_lsn; + XLogRecPtr confirmed_flush_lsn; + Oid datoid; + NameData slot_name; + NameData plugin; + int i; + int64 tmpbigint; + + if (!slot->in_use) + continue; + + SpinLockAcquire(&slot->mutex); + + xmin = slot->data.xmin; + catalog_xmin = slot->data.catalog_xmin; + datoid = slot->data.database; + restart_lsn = slot->data.restart_lsn; + confirmed_flush_lsn = slot->data.confirmed_flush; + namecpy(&slot_name, &slot->data.name); + namecpy(&plugin, &slot->data.plugin); + persistency = slot->data.persistency; + + SpinLockRelease(&slot->mutex); + + memset(nulls, 0, sizeof(nulls)); + + i = 0; + values[i++] = CStringGetTextDatum(NameStr(slot_name)); + + if (datoid == InvalidOid) + nulls[i++] = true; + else + values[i++] = CStringGetTextDatum(NameStr(plugin)); + + if (datoid == InvalidOid) + values[i++] = CStringGetTextDatum("physical"); + else + values[i++] = CStringGetTextDatum("logical"); + + if (datoid == InvalidOid) + nulls[i++] = true; + else + { + tmpbigint = datoid; + values[i++] = Int64GetDatum(tmpbigint); + } + + if (datoid == InvalidOid) + nulls[i++] = true; + else + { + MemoryContext cur = CurrentMemoryContext; + + /* syscache access needs a transaction env. */ + StartTransactionCommand(); + /* make dbname live outside TX context */ + MemoryContextSwitchTo(cur); + values[i++] = CStringGetTextDatum(get_database_name(datoid)); + CommitTransactionCommand(); + /* CommitTransactionCommand switches to TopMemoryContext */ + MemoryContextSwitchTo(cur); + } + + values[i++] = Int32GetDatum(persistency == RS_TEMPORARY ? 1 : 0); + + if (xmin != InvalidTransactionId) + { + tmpbigint = xmin; + values[i++] = Int64GetDatum(tmpbigint); + } + else + nulls[i++] = true; + + if (catalog_xmin != InvalidTransactionId) + { + tmpbigint = catalog_xmin; + values[i++] = Int64GetDatum(tmpbigint); + } + else + nulls[i++] = true; + + if (restart_lsn != InvalidXLogRecPtr) + { + snprintf(restart_lsn_str, sizeof(restart_lsn_str), "%X/%X", + (uint32) (restart_lsn >> 32), + (uint32) restart_lsn); + values[i++] = CStringGetTextDatum(restart_lsn_str); + } + else + nulls[i++] = true; + + if (confirmed_flush_lsn != InvalidXLogRecPtr) + { + snprintf(confirmed_flush_lsn_str, sizeof(confirmed_flush_lsn_str), + "%X/%X", + (uint32) (confirmed_flush_lsn >> 32), + (uint32) confirmed_flush_lsn); + values[i++] = CStringGetTextDatum(confirmed_flush_lsn_str); + } + else + nulls[i++] = true; + + /* send it to dest */ + do_tup_output(tstate, values, nulls); + } + LWLockRelease(ReplicationSlotControlLock); + + end_tup_output(tstate); +} /* * Handle TIMELINE_HISTORY command. @@ -1551,6 +1711,10 @@ exec_replication_command(const char *cmd_string) } break; + case T_ListSlotsCmd: + ListSlots(); + break; + case T_SQLCmd: if (MyDatabaseId == InvalidOid) ereport(ERROR, diff --git a/src/include/nodes/nodes.h b/src/include/nodes/nodes.h index cac6ff0eda..a773acab79 100644 --- a/src/include/nodes/nodes.h +++ b/src/include/nodes/nodes.h @@ -486,6 +486,7 @@ typedef enum NodeTag T_StartReplicationCmd, T_TimeLineHistoryCmd, T_SQLCmd, + T_ListSlotsCmd, /* * TAGS FOR RANDOM OTHER STUFF diff --git a/src/include/nodes/replnodes.h b/src/include/nodes/replnodes.h index 66c948d85e..1101f71bcc 100644 --- a/src/include/nodes/replnodes.h +++ b/src/include/nodes/replnodes.h @@ -33,6 +33,14 @@ typedef struct IdentifySystemCmd NodeTag type; } IdentifySystemCmd; +/* ---------------------- + * LIST_SLOTS command + * ---------------------- + */ +typedef struct ListSlotsCmd +{ + NodeTag type; +} ListSlotsCmd; /* ---------------------- * BASE_BACKUP command diff --git a/src/include/pgstat.h b/src/include/pgstat.h index f1c10d16b8..52bdee746e 100644 --- a/src/include/pgstat.h +++ b/src/include/pgstat.h @@ -761,6 +761,7 @@ typedef enum WAIT_EVENT_CHECKPOINTER_MAIN, WAIT_EVENT_LOGICAL_APPLY_MAIN, WAIT_EVENT_LOGICAL_LAUNCHER_MAIN, + WAIT_EVENT_REPL_SLOT_SYNC_MAIN, WAIT_EVENT_PGSTAT_MAIN, WAIT_EVENT_RECOVERY_WAL_ALL, WAIT_EVENT_RECOVERY_WAL_STREAM, diff --git a/src/include/replication/slot.h b/src/include/replication/slot.h index 7964ae254f..1a0fd44876 100644 --- a/src/include/replication/slot.h +++ b/src/include/replication/slot.h @@ -205,4 +205,7 @@ extern void CheckPointReplicationSlots(void); extern void CheckSlotRequirements(void); +extern XLogRecPtr pg_logical_replication_slot_advance(XLogRecPtr moveto); + + #endif /* SLOT_H */ diff --git a/src/include/replication/walreceiver.h b/src/include/replication/walreceiver.h index 5913b580c2..9014f92bfa 100644 --- a/src/include/replication/walreceiver.h +++ b/src/include/replication/walreceiver.h @@ -17,6 +17,7 @@ #include "fmgr.h" #include "getaddrinfo.h" /* for NI_MAXHOST */ #include "replication/logicalproto.h" +#include "replication/slot.h" #include "replication/walsender.h" #include "storage/latch.h" #include "storage/spin.h" @@ -167,6 +168,13 @@ typedef struct } proto; } WalRcvStreamOptions; +/* + * Slot information receiver from remote. + * + * Currently this is same as ReplicationSlotPersistentData + */ +#define WalRecvReplicationSlotData ReplicationSlotPersistentData + struct WalReceiverConn; typedef struct WalReceiverConn WalReceiverConn; @@ -211,6 +219,7 @@ typedef void (*walrcv_get_senderinfo_fn) (WalReceiverConn *conn, typedef char *(*walrcv_identify_system_fn) (WalReceiverConn *conn, TimeLineID *primary_tli, int *server_version); +typedef List *(*walrcv_list_slots_fn) (WalReceiverConn *conn); typedef void (*walrcv_readtimelinehistoryfile_fn) (WalReceiverConn *conn, TimeLineID tli, char **filename, @@ -240,6 +249,7 @@ typedef struct WalReceiverFunctionsType walrcv_get_conninfo_fn walrcv_get_conninfo; walrcv_get_senderinfo_fn walrcv_get_senderinfo; walrcv_identify_system_fn walrcv_identify_system; + walrcv_list_slots_fn walrcv_list_slots; walrcv_readtimelinehistoryfile_fn walrcv_readtimelinehistoryfile; walrcv_startstreaming_fn walrcv_startstreaming; walrcv_endstreaming_fn walrcv_endstreaming; @@ -262,6 +272,8 @@ extern PGDLLIMPORT WalReceiverFunctionsType *WalReceiverFunctions; WalReceiverFunctions->walrcv_get_senderinfo(conn, sender_host, sender_port) #define walrcv_identify_system(conn, primary_tli, server_version) \ WalReceiverFunctions->walrcv_identify_system(conn, primary_tli, server_version) +#define walrcv_list_slots(conn) \ + WalReceiverFunctions->walrcv_list_slots(conn) #define walrcv_readtimelinehistoryfile(conn, tli, filename, content, size) \ WalReceiverFunctions->walrcv_readtimelinehistoryfile(conn, tli, filename, content, size) #define walrcv_startstreaming(conn, options) \ diff --git a/src/include/replication/worker_internal.h b/src/include/replication/worker_internal.h index ef079111cd..ba2d3f1fca 100644 --- a/src/include/replication/worker_internal.h +++ b/src/include/replication/worker_internal.h @@ -69,14 +69,14 @@ extern LogicalRepWorker *MyLogicalRepWorker; extern bool in_remote_transaction; extern void logicalrep_worker_attach(int slot); -extern LogicalRepWorker *logicalrep_worker_find(Oid subid, Oid relid, +extern LogicalRepWorker *logicalrep_worker_find(Oid dbid, Oid subid, Oid relid, bool only_running); extern List *logicalrep_workers_find(Oid subid, bool only_running); extern void logicalrep_worker_launch(Oid dbid, Oid subid, const char *subname, Oid userid, Oid relid); -extern void logicalrep_worker_stop(Oid subid, Oid relid); -extern void logicalrep_worker_stop_at_commit(Oid subid, Oid relid); -extern void logicalrep_worker_wakeup(Oid subid, Oid relid); +extern void logicalrep_worker_stop(Oid dbid, Oid subid, Oid relid); +extern void logicalrep_worker_stop_at_commit(Oid dbid, Oid subid, Oid relid); +extern void logicalrep_worker_wakeup(Oid dbid, Oid subid, Oid relid); extern void logicalrep_worker_wakeup_ptr(LogicalRepWorker *worker); extern int logicalrep_sync_worker_count(Oid subid); -- 2.17.1