From b8239158c47f5632805b3655540846ffb054e033 Mon Sep 17 00:00:00 2001 From: Ajin Cherian Date: Wed, 3 Mar 2021 05:44:59 -0500 Subject: [PATCH v47] Fix apply worker empty prepare. By sad timing of apply/tablesync workers it is possible to have a "consistent snapshot" that spans prepare/commit in such a way that the tablesync did not do the prepare (because snapshot not consistent) and the apply worker does the prepare ('b') but it skips all the prepared operations [e.g. inserts] while the tablesync was still busy (see the condition of should_apply_changes_for_rel). Later, at the commit prepared time when the apply worker does the commit prepare ('K'), there is nothing committed (because the inserts were skipped earlier). This patch implements a two-part fix as suggested [1] on hackers. Part 1 - The begin_prepare handler of apply will always wait for any busy tablesync workers to acheive SYNCDONE/READY state. Part 2 - If (after Part 1) the apply-worker's prepare is found to be lagging behind any of the sync-workers then the subsequent prepared operations will be spooled to a file to be replayed at commit_prepared time. Discussion: [1] https://www.postgresql.org/message-id/CAA4eK1L%3DdhuCRvyDvrXX5wZgc7s1hLRD29CKCK6oaHtVCPgiFA%40mail.gmail.com --- src/backend/replication/logical/tablesync.c | 178 ++++++-- src/backend/replication/logical/worker.c | 668 +++++++++++++++++++++++++++- src/include/replication/worker_internal.h | 3 + src/tools/pgindent/typedefs.list | 1 + 4 files changed, 815 insertions(+), 35 deletions(-) diff --git a/src/backend/replication/logical/tablesync.c b/src/backend/replication/logical/tablesync.c index 50c3ea7..9e2b48c 100644 --- a/src/backend/replication/logical/tablesync.c +++ b/src/backend/replication/logical/tablesync.c @@ -116,6 +116,9 @@ #include "utils/snapmgr.h" static bool table_states_valid = false; +static List *table_states_not_ready = NIL; +static List *table_states_all = NIL; +static void FetchTableStates(bool *started_tx); StringInfo copybuf = NULL; @@ -359,7 +362,6 @@ process_syncing_tables_for_apply(XLogRecPtr current_lsn) Oid relid; TimestampTz last_start_time; }; - static List *table_states = NIL; static HTAB *last_start_times = NULL; ListCell *lc; bool started_tx = false; @@ -367,42 +369,14 @@ process_syncing_tables_for_apply(XLogRecPtr current_lsn) Assert(!IsTransactionState()); /* We need up-to-date sync state info for subscription tables here. */ - if (!table_states_valid) - { - MemoryContext oldctx; - List *rstates; - ListCell *lc; - SubscriptionRelState *rstate; - - /* Clean the old list. */ - list_free_deep(table_states); - table_states = NIL; - - StartTransactionCommand(); - started_tx = true; - - /* Fetch all non-ready tables. */ - rstates = GetSubscriptionNotReadyRelations(MySubscription->oid); - - /* Allocate the tracking info in a permanent memory context. */ - oldctx = MemoryContextSwitchTo(CacheMemoryContext); - foreach(lc, rstates) - { - rstate = palloc(sizeof(SubscriptionRelState)); - memcpy(rstate, lfirst(lc), sizeof(SubscriptionRelState)); - table_states = lappend(table_states, rstate); - } - MemoryContextSwitchTo(oldctx); - - table_states_valid = true; - } + FetchTableStates(&started_tx); /* * Prepare a hash table for tracking last start times of workers, to avoid * immediate restarts. We don't need it if there are no tables that need * syncing. */ - if (table_states && !last_start_times) + if (table_states_not_ready && !last_start_times) { HASHCTL ctl; @@ -416,7 +390,7 @@ process_syncing_tables_for_apply(XLogRecPtr current_lsn) * Clean up the hash table when we're done with all tables (just to * release the bit of memory). */ - else if (!table_states && last_start_times) + else if (!table_states_not_ready && last_start_times) { hash_destroy(last_start_times); last_start_times = NULL; @@ -425,7 +399,7 @@ process_syncing_tables_for_apply(XLogRecPtr current_lsn) /* * Process all tables that are being synchronized. */ - foreach(lc, table_states) + foreach(lc, table_states_not_ready) { SubscriptionRelState *rstate = (SubscriptionRelState *) lfirst(lc); @@ -1137,3 +1111,141 @@ copy_table_done: wait_for_worker_state_change(SUBREL_STATE_CATCHUP); return slotname; } + +/* + * Common code to fetch the up-to-date sync state info into the static lists. + */ +static void +FetchTableStates(bool *started_tx) +{ + *started_tx = false; + + if (!table_states_valid) + { + MemoryContext oldctx; + List *rstates; + ListCell *lc; + SubscriptionRelState *rstate; + + /* Clean the old lists. */ + list_free_deep(table_states_all); + table_states_all = NIL; + list_free_deep(table_states_not_ready); + table_states_not_ready = NIL; + + StartTransactionCommand(); + *started_tx = true; + + /* Fetch all tables. */ + rstates = GetSubscriptionRelations(MySubscription->oid); + + /* Allocate the tracking info in a permanent memory context. */ + oldctx = MemoryContextSwitchTo(CacheMemoryContext); + foreach(lc, rstates) + { + SubscriptionRelState *cur_rstate = (SubscriptionRelState *) lfirst(lc); + + /* List of all states */ + rstate = palloc(sizeof(SubscriptionRelState)); + memcpy(rstate, cur_rstate, sizeof(SubscriptionRelState)); + table_states_all = lappend(table_states_all, rstate); + + /* List of only not-ready states */ + if (cur_rstate->state != SUBREL_STATE_READY) + { + rstate = palloc(sizeof(SubscriptionRelState)); + memcpy(rstate, cur_rstate, sizeof(SubscriptionRelState)); + table_states_not_ready = lappend(table_states_not_ready, rstate); + } + } + MemoryContextSwitchTo(oldctx); + + table_states_valid = true; + } +} + +/* + * Are there any tablesyncs which have still not yet reached SYNCDONE/READY state? + */ +bool +AnyTablesyncInProgress() +{ + bool found_busy = false; + bool started_tx = false; + int count = 0; + ListCell *lc; + + /* We need up-to-date sync state info for subscription tables here. */ + FetchTableStates(&started_tx); + + /* + * Process all not-READY tables to see if any are also not-SYNCDONE + */ + foreach(lc, table_states_not_ready) + { + SubscriptionRelState *rstate = (SubscriptionRelState *) lfirst(lc); + + count++; + elog(DEBUG1, + "AnyTablesyncInProgress?: #%d. Table relid %u has state '%c'", + count, + rstate->relid, + rstate->state); + + /* + * When the process_syncing_tables_for_apply changes the state + * from SYNCDONE to READY, that change is actually written directly + * into the list element of table_states_not_ready. + * + * So the "table_states_not_ready" list might end up having a READY + * state it it even though there was none when it was initially + * created. This is reason why we need to check for READY below. + */ + if (rstate->state != SUBREL_STATE_SYNCDONE && + rstate->state != SUBREL_STATE_READY) + { + found_busy = true; + break; + } + } + + if (started_tx) + { + CommitTransactionCommand(); + pgstat_report_stat(false); + } + + elog(DEBUG1, + "AnyTablesyncInProgress?: Scanned %d tables, and found busy = %s", + count, + found_busy ? "true" : "false"); + + return found_busy; +} + +/* + * What is the biggest LSN from the all the known tablesyncs? + */ +XLogRecPtr +BiggestTablesyncLSN() +{ + XLogRecPtr biggest_lsn = InvalidXLogRecPtr; + ListCell *lc; + int count = 0; + + foreach(lc, table_states_all) + { + SubscriptionRelState *rstate = (SubscriptionRelState *) lfirst(lc); + + count++; + if (rstate->lsn > biggest_lsn) + biggest_lsn = rstate->lsn; + } + + elog(DEBUG1, + "BiggestTablesyncLSN: Scanned %d tables. Biggest lsn found = %X/%X", + count, + LSN_FORMAT_ARGS(biggest_lsn)); + + return biggest_lsn; +} diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c index ad537f4..fcf7bc2 100644 --- a/src/backend/replication/logical/worker.c +++ b/src/backend/replication/logical/worker.c @@ -209,6 +209,45 @@ static void subxact_info_add(TransactionId xid); static inline void cleanup_subxact_info(void); /* + * The following are for the support of a spoolfile for prepared messages. + */ + +/* + * A contest for the prepare spooling + */ +static MemoryContext PsfContext = NULL; + +/* + * A Prepare spoolfile hash entry. We create this entry in the psf_hash. This is + * for maintaining a mapping between the name of the prepared spoolfile, and the + * corresponding fileset handles of same. + */ +typedef struct PsfHashEntry +{ + char name[MAXPGPATH]; /* Hash key --- must be first */ + SharedFileSet *fileset; /* shared file set for prepare spoolfile */ +} PsfHashEntry; + +/* + * Hash table for storing the Prepared spoolfile info along with shared fileset. + */ +static HTAB *psf_hash = NULL; + +/* + * The spoolfile handle is only valid between begin_prepare and prepare. + */ +static BufFile *psf_fd = NULL; + +static void prepare_spoolfile_create(char *path); +static void prepare_spoolfile_write(char action, StringInfo s); +static void prepare_spoolfile_close(void); +static void prepare_spoolfile_cleanup(char *path); +static bool prepare_spoolfile_exists(char *path); +static void prepare_spoolfile_name(char *path, int szpath, Oid subid, char *gid); +static int prepare_spoolfile_replay_messages(char *path, LogicalRepPreparedTxnData *pdata); +static bool prepare_spoolfile_handler(LogicalRepMsgType action, StringInfo s); + +/* * Serialize and deserialize changes for a toplevel transaction. */ static void stream_cleanup_files(Oid subid, TransactionId xid); @@ -730,6 +769,9 @@ apply_handle_begin_prepare(StringInfo s) { LogicalRepBeginPrepareData begin_data; + /* Tablesync should never receive prepare. */ + Assert(!am_tablesync_worker()); + logicalrep_read_begin_prepare(s, &begin_data); /* @@ -741,6 +783,81 @@ apply_handle_begin_prepare(StringInfo s) errmsg("transaction identifier \"%s\" is already in use", begin_data.gid))); + /* + * A Problem: + * + * By sad timing of apply/tablesync workers it is possible to have a + * “consistent snapshot” that spans prepare/commit in such a way that + * the tablesync did not do the prepare (because snapshot not consistent) + * and the apply worker does the begin prepare (‘b’) but it skips all + * the prepared operations [e.g. inserts] while the tablesync was still + * busy (see the condition of should_apply_changes_for_rel). Later at the + * commit prepared time when the apply worker does the commit prepare + * (‘K’), there is nothing in it (because the inserts were skipped + * earlier). + * + * The following code has a 2-part workaround for that scenario. + */ + if (!am_tablesync_worker()) + { + /* + * Workaround Part 1 of 2: + * + * Make sure every tablesync has reached at least SYNCDONE state + * before letting the apply worker proceed. + */ + elog(DEBUG1, + "apply_handle_begin_prepare, end_lsn = %X/%X, final_lsn = %X/%X, lstate_lsn = %X/%X", + LSN_FORMAT_ARGS(begin_data.end_lsn), + LSN_FORMAT_ARGS(begin_data.final_lsn), + LSN_FORMAT_ARGS(MyLogicalRepWorker->relstate_lsn)); + + while (AnyTablesyncInProgress()) + { + process_syncing_tables(begin_data.final_lsn); + + /* This latch is to prevent 100% CPU looping. */ + (void) WaitLatch(MyLatch, + WL_LATCH_SET | WL_TIMEOUT | WL_EXIT_ON_PM_DEATH, + 1000L, WAIT_EVENT_LOGICAL_SYNC_STATE_CHANGE); + ResetLatch(MyLatch); + } + + /* + * Workaround Part 2 of 2: + * + * If (when getting to SYNCDONE/READY state) some tablesync went + * beyond this begin_prepare LSN then set all messages (until + * prepared) will be saved to a spoolfile for replay later at + * commit_prepared time. + */ + if (begin_data.final_lsn < BiggestTablesyncLSN()) + { + char psfpath[MAXPGPATH]; + StringInfoData sid; + + /* The begin_prepare's LSN has been overtaken. */ + + /* + * We need a transaction for handling the buffile, used for + * serializing prepared messages. This transaction lasts until the + * commit_prepared/ rollback_prepared. + */ + ensure_transaction(); + + prepare_spoolfile_name(psfpath, sizeof(psfpath), + MyLogicalRepWorker->subid, begin_data.gid); + prepare_spoolfile_create(psfpath); + + /* + * Write BEGIN_PREPARE as the first message of the psf file. + */ + initStringInfo(&sid); + appendBinaryStringInfo(&sid, (char *)&begin_data, sizeof(begin_data)); + Assert(prepare_spoolfile_handler(LOGICAL_REP_MSG_BEGIN_PREPARE, &sid)); + } + } + remote_final_lsn = begin_data.final_lsn; in_remote_transaction = true; @@ -756,6 +873,38 @@ apply_handle_prepare(StringInfo s) { LogicalRepPreparedTxnData prepare_data; + /* + * If we were using a psf spoolfile, then write the PREPARE as the final + * message. This prepare information will be used at commit_prepared time. + */ + if (psf_fd) + { + /* Write the PREPARE info to the psf file. */ + Assert(prepare_spoolfile_handler(LOGICAL_REP_MSG_PREPARE, s)); + + /* + * TODO - Flush the spoolfile so changes can survive a restart. + */ + + /* + * The psf_fd is meaningful only between begin_prepare and prepared. + * So close it now. If we had been writing any messages to the psf_fd + * (the spoolfile) then those will be applied later during + * handle_commit_prepared. + */ + prepare_spoolfile_close(); + + /* + * And end the transaction that was created by begin_prepare for + * working with the psf buffiles. + */ + Assert(IsTransactionState()); + CommitTransactionCommand(); + + in_remote_transaction = false; + return; + } + logicalrep_read_prepare(s, &prepare_data); Assert(prepare_data.prepare_lsn == remote_final_lsn); @@ -805,9 +954,61 @@ static void apply_handle_commit_prepared(StringInfo s) { LogicalRepPreparedTxnData prepare_data; + char psfpath[MAXPGPATH]; + logicalrep_read_commit_prepared(s, &prepare_data); + /* + * If this prepare's messages were being spooled to a file, then replay + * them all now. + */ + prepare_spoolfile_name(psfpath, sizeof(psfpath), + MyLogicalRepWorker->subid, prepare_data.gid); + if (prepare_spoolfile_exists(psfpath)) + { + int nchanges; + LogicalRepPreparedTxnData pdata; + + /* + * 1. replay the spooled messages + */ + + ensure_transaction(); + + nchanges = prepare_spoolfile_replay_messages(psfpath, &pdata); + elog(DEBUG1, + "apply_handle_commit_prepared: replayed %d (all) changes.", + nchanges); + + prepare_spoolfile_cleanup(psfpath); + + /* + * 2. mark as PREPARED (use prepare_data info from the psf file) + */ + + /* + * BeginTransactionBlock is necessary to balance the + * EndTransactionBlock called within the PrepareTransactionBlock + * below. + */ + BeginTransactionBlock(); + CommitTransactionCommand(); + + /* + * Update origin state so we can restart streaming from correct + * position in case of crash. + */ + replorigin_session_origin_lsn = pdata.end_lsn; + replorigin_session_origin_timestamp = pdata.preparetime; + + PrepareTransactionBlock(pdata.gid); + CommitTransactionCommand(); + pgstat_report_stat(false); + + store_flush_position(pdata.end_lsn); + } + /* there is no transaction when COMMIT PREPARED is called */ ensure_transaction(); @@ -838,15 +1039,53 @@ static void apply_handle_rollback_prepared(StringInfo s) { LogicalRepRollbackPreparedTxnData rollback_data; + bool using_psf; + char psfpath[MAXPGPATH]; logicalrep_read_rollback_prepared(s, &rollback_data); /* + * If this prepare's messages were being spooled to a file, then cleanup + * the file. + */ + prepare_spoolfile_name(psfpath, sizeof(psfpath), + MyLogicalRepWorker->subid, rollback_data.gid); + using_psf = prepare_spoolfile_exists(psfpath); + if (using_psf) + { + if (psf_fd) + { + /* + * XXX - For some reason it is currently possible (due to bug?) it + * is possibe to get here, after a restart, when there was a + * begin_prepare but there was NO prepare. Since there was no + * prepare, the psf_fd and the transaction are still lingering so + * they need to be cleaned up now. + */ + prepare_spoolfile_close(); + + /* + * And end the transaction that was created by the begin_prepare + * for working with psf buffiles. + */ + Assert(IsTransactionState()); + AbortCurrentTransaction(); + } + + prepare_spoolfile_cleanup(psfpath); + } + + /* * It is possible that we haven't received prepare because it occurred * before walsender reached a consistent point in which case we need to * skip rollback prepared. + * + * And we also skip the FinishPreparedTransaction if we're using the + * Prepare Spoolfile (using_psf) because in that case there is no matching + * PrepareTransactionBlock done yet. */ - if (LookupGXact(rollback_data.gid, rollback_data.prepare_end_lsn, + if (!using_psf && + LookupGXact(rollback_data.gid, rollback_data.prepare_end_lsn, rollback_data.preparetime)) { /* @@ -889,10 +1128,33 @@ apply_handle_stream_prepare(StringInfo s) Assert(!in_streamed_transaction); + /* Tablesync should never receive prepare. */ + Assert(!am_tablesync_worker()); + xid = logicalrep_read_stream_prepare(s, &prepare_data); elog(DEBUG1, "received prepare for streamed transaction %u", xid); /* + * Wait for all the sync workers to reach the SYNCDONE/READY state. + * + * This is same waiting logic as in apply_handle_begin_prepare function + * (see that function for more details about this). + */ + if (!am_tablesync_worker()) + { + while (AnyTablesyncInProgress()) + { + process_syncing_tables(prepare_data.end_lsn); + + /* This latch is to prevent 100% CPU looping. */ + (void) WaitLatch(MyLatch, + WL_LATCH_SET | WL_TIMEOUT | WL_EXIT_ON_PM_DEATH, + 1000L, WAIT_EVENT_LOGICAL_SYNC_STATE_CHANGE); + ResetLatch(MyLatch); + } + } + + /* * * -------------------------------------------------------------------------- * 1. Replay all the spooled operations - Similar code as for @@ -959,6 +1221,7 @@ apply_handle_origin(StringInfo s) * remote transaction and before any actual writes. */ if (!in_streamed_transaction && + psf_fd == NULL && (!in_remote_transaction || (IsTransactionState() && !am_tablesync_worker()))) ereport(ERROR, @@ -1393,6 +1656,9 @@ apply_handle_insert(StringInfo s) TupleTableSlot *remoteslot; MemoryContext oldctx; + if (prepare_spoolfile_handler(LOGICAL_REP_MSG_INSERT, s)) + return; + if (handle_streamed_transaction(LOGICAL_REP_MSG_INSERT, s)) return; @@ -1514,6 +1780,9 @@ apply_handle_update(StringInfo s) RangeTblEntry *target_rte; MemoryContext oldctx; + if (prepare_spoolfile_handler(LOGICAL_REP_MSG_UPDATE, s)) + return; + if (handle_streamed_transaction(LOGICAL_REP_MSG_UPDATE, s)) return; @@ -1672,6 +1941,9 @@ apply_handle_delete(StringInfo s) TupleTableSlot *remoteslot; MemoryContext oldctx; + if (prepare_spoolfile_handler(LOGICAL_REP_MSG_DELETE, s)) + return; + if (handle_streamed_transaction(LOGICAL_REP_MSG_DELETE, s)) return; @@ -2041,6 +2313,9 @@ apply_handle_truncate(StringInfo s) List *relids_logged = NIL; ListCell *lc; + if (prepare_spoolfile_handler(LOGICAL_REP_MSG_TRUNCATE, s)) + return; + if (handle_streamed_transaction(LOGICAL_REP_MSG_TRUNCATE, s)) return; @@ -2334,6 +2609,23 @@ LogicalRepApplyLoop(XLogRecPtr last_received) } /* + * Initialize the psf_hash table if we haven't yet. This will be used for + * the entire duration of the apply worker so create it in permanent + * context. + */ + if (psf_hash == NULL) + { + HASHCTL hash_ctl; + PsfHashEntry *hentry; + + hash_ctl.keysize = sizeof(hentry->name); + hash_ctl.entrysize = sizeof(PsfHashEntry); + hash_ctl.hcxt = ApplyContext; + psf_hash = hash_create("PrepareSpoolfileHash", 1024, &hash_ctl, + HASH_ELEM | HASH_BLOBS | HASH_CONTEXT); + } + + /* * Init the ApplyMessageContext which we clean up after each replication * protocol message. */ @@ -2349,6 +2641,14 @@ LogicalRepApplyLoop(XLogRecPtr last_received) "LogicalStreamingContext", ALLOCSET_DEFAULT_SIZES); + /* + * This memory context is used when the prepare spooling is used. It is + * reset at prepare commit/rollback time. + */ + PsfContext = AllocSetContextCreate(ApplyContext, + "PsfContext", + ALLOCSET_DEFAULT_SIZES); + /* mark as idle, before starting to loop */ pgstat_report_activity(STATE_IDLE, NULL); @@ -2453,7 +2753,7 @@ LogicalRepApplyLoop(XLogRecPtr last_received) /* confirm all writes so far */ send_feedback(last_received, false, false); - if (!in_remote_transaction && !in_streamed_transaction) + if (!in_remote_transaction && !in_streamed_transaction && psf_fd == NULL) { /* * If we didn't get any transactions for a while there might be @@ -3378,3 +3678,367 @@ IsLogicalWorker(void) { return MyLogicalRepWorker != NULL; } + +/* + * Handle the PREPARE spoolfile (if any) + * + * It can be necessary to redirect the PREPARE messages to a spoolfile (see + * apply_handle_begin_prepare) and then replay them back at the COMMIT PREPARED + * time. If needed, this is the common function to do that file redirection. + * + * Returns true if the message was redirected to the spoolfile, false + * otherwise (regular mode). + */ +static bool +prepare_spoolfile_handler(LogicalRepMsgType action, StringInfo s) +{ + elog(DEBUG1, + "prepare_spoolfile_handler for action '%c'. %s write to spool file", + action, + psf_fd ? "Do" : "Don't"); + + if (psf_fd == NULL) + return false; + + Assert(!in_streamed_transaction); + + /* write the change to the current file */ + prepare_spoolfile_write(action, s); + + return true; +} + +/* + * Create the spoolfile used to serialize the prepare messages. + */ +static void +prepare_spoolfile_create(char *path) +{ + MemoryContext oldctx; + bool found; + PsfHashEntry *hentry; + + elog(DEBUG1, "creating file \"%s\" for prepare changes", path); + + Assert(psf_fd == NULL); + + /* create or find the prepare spoolfile entry in the psf_hash */ + hentry = (PsfHashEntry *) hash_search(psf_hash, + path, + HASH_ENTER | HASH_FIND, + &found); + + /* + * Create/open the bufFiles under the Prepare Spoolfile Context so that we + * have those files until prepare commit/rollback. + */ + oldctx = MemoryContextSwitchTo(PsfContext); + + if (!found) + { + MemoryContext savectx; + SharedFileSet *fileset; + + elog(DEBUG1, "Not found file \"%s\". Create it.", path); + savectx = MemoryContextSwitchTo(ApplyContext); + fileset = palloc(sizeof(SharedFileSet)); + + SharedFileSetInit(fileset, NULL); + MemoryContextSwitchTo(savectx); + + psf_fd = BufFileCreateShared(fileset, path); + + /* Remember this path's fileset for the next time */ + memcpy(hentry->name, path, MAXPGPATH); + hentry->fileset = fileset; + } + else + { + /* + * Open the file and seek to the end of the file because we always + * append the changes file. + */ + elog(DEBUG1, "Found file \"%s\". Open for append.", path); + psf_fd = BufFileOpenShared(hentry->fileset, path, O_RDWR); + BufFileSeek(psf_fd, 0, 0, SEEK_END); + } + + MemoryContextSwitchTo(oldctx); + + /* Sanity check */ + Assert(prepare_spoolfile_exists(path)); +} + +/* + * Close the "current" spoolfile and unset the fd. + */ +static void +prepare_spoolfile_close() +{ + if (psf_fd) + BufFileClose(psf_fd); + psf_fd = NULL; +} + +/* + * Delete the specified psf spoolfile. + */ +static void +prepare_spoolfile_cleanup(char *path) +{ + PsfHashEntry *hentry; + + /* The current psf should be closed already, but make sure anyway. */ + prepare_spoolfile_close(); + + /* And remove the path entry from the psf_hash */ + hentry = (PsfHashEntry *) hash_search(psf_hash, + path, + HASH_REMOVE, + NULL); + + /* By this time we must have created the entry */ + Assert(hentry != NULL); + + /* Delete the file and release the fileset memory */ + SharedFileSetDeleteAll(hentry->fileset); + pfree(hentry->fileset); + hentry->fileset = NULL; + + /* Reset the memory context used during the file creation */ + MemoryContextReset(PsfContext); +} + +/* + * Serialize a change to the prepare spoolfile for the current toplevel transaction. + * + * The change is serialized in a simple format, with length (not including + * the length), action code (identifying the message type) and message + * contents (without the subxact TransactionId value). + */ +static void +prepare_spoolfile_write(char action, StringInfo s) +{ + int len; + + Assert(psf_fd != NULL); + + elog(DEBUG1, "prepare_spoolfile_write: writing action '%c'", action); + + /* total on-disk size, including the action type character */ + len = (s->len - s->cursor) + sizeof(char); + + /* first write the size */ + BufFileWrite(psf_fd, &len, sizeof(len)); + + /* then the action */ + BufFileWrite(psf_fd, &action, sizeof(action)); + + /* and finally the remaining part of the buffer (after the XID) */ + len = (s->len - s->cursor); + + BufFileWrite(psf_fd, &s->data[s->cursor], len); +} + +/* + * Is there a prepare spoolfile for the specified gid? + */ +static bool +prepare_spoolfile_exists(char *path) +{ + bool found; + + /* Find the prepare spoolfile entry in the psf_hash */ + hash_search(psf_hash, + path, + HASH_FIND, + &found); + + return found; +} + +/* + * Replay (apply) all the prepared messages that are in the prepare spoolfile. + * + * [Note: this is similar to apply_spooled_messages function] + */ +static int +prepare_spoolfile_replay_messages(char *path, LogicalRepPreparedTxnData *pdata) +{ + StringInfoData s2; + int nchanges = 0; + char *buffer = NULL; + MemoryContext oldctx, + oldctx2; + bool found = false; + PsfHashEntry *hentry; + BufFile *fd; + + elog(DEBUG1, + "prepare_spoolfile_replay_messages: replaying changes from file \"%s\"", + path); + + /* + * Allocate memory required to process all the messages in + * TopTransactionContext to avoid it getting reset after each message is + * processed. + */ + oldctx = MemoryContextSwitchTo(TopTransactionContext); + + /* Open the spool file */ + hentry = (PsfHashEntry *) hash_search(psf_hash, + path, + HASH_FIND, + &found); + Assert(found); + fd = BufFileOpenShared(hentry->fileset, path, O_RDONLY); + + buffer = palloc(BLCKSZ); + initStringInfo(&s2); + + MemoryContextSwitchTo(oldctx); + + /* + * Read the entries one by one and pass them through the same logic as in + * apply_dispatch. + */ + while (true) + { + int nbytes; + int len; + + CHECK_FOR_INTERRUPTS(); + + /* read length of the on-disk record */ + nbytes = BufFileRead(fd, &len, sizeof(len)); + + /* have we reached end of the file? */ + if (nbytes == 0) + break; + + /* do we have a correct length? */ + if (nbytes != sizeof(len)) + ereport(ERROR, + (errcode_for_file_access(), + errmsg("could not read from prepared spoolfile \"%s\": %m", + path))); + + Assert(len > 0); + + /* make sure we have sufficiently large buffer */ + buffer = repalloc(buffer, len); + + /* and finally read the data into the buffer */ + if (BufFileRead(fd, buffer, len) != len) + ereport(ERROR, + (errcode_for_file_access(), + errmsg("could not read from prepared spoolfile \"%s\": %m", + path))); + + /* copy the buffer to the stringinfo and call apply_dispatch */ + resetStringInfo(&s2); + appendBinaryStringInfo(&s2, buffer, len); + + /* + * The psf spoolfile contents will have first and last messages as + * BEGIN_PREPARE and PREPARE message respectively. These two are + * processed specially within this function. + * + * BEGIN_PREPARE msg: This will be the first message of the psf file. + * Use this begin_data information to set the remote_final_lsn. + * + * PREPARE msg: The prepare_data information is returned so that the + * prepare lsn values are available to the caller (commit_prepared). + * Unfortunately, just dispatching the PREPARE message is problematic + * because its transaction commits have side effects on this replay + * loop which is still running. + * + * All other message content (between the BEGIN_PREPARE and the PREARE) + * will be delivered to apply_dispatch as they normally would be. + */ + if (s2.data[0] == LOGICAL_REP_MSG_BEGIN_PREPARE) + { + LogicalRepBeginPrepareData bdata; + + /* read/skip the action byte. */ + Assert(pq_getmsgbyte(&s2) == LOGICAL_REP_MSG_BEGIN_PREPARE); + + /* read the begin_data. */ + logicalrep_read_begin_prepare(&s2, &bdata); + + elog(DEBUG1, "BEGIN_PREPARE info: gid = '%s', final_lsn = %X/%X, end_lsn = %X/%X", + bdata.gid, + LSN_FORMAT_ARGS(bdata.final_lsn), + LSN_FORMAT_ARGS(bdata.end_lsn)); + + /* + * Make sure the handle apply_dispatch methods are aware we're in a remote + * transaction. + */ + remote_final_lsn = bdata.final_lsn; + in_remote_transaction = true; + pgstat_report_activity(STATE_RUNNING, NULL); + } + else if (s2.data[0] == LOGICAL_REP_MSG_PREPARE) + { + /* read/skip the action byte. */ + Assert(pq_getmsgbyte(&s2) == LOGICAL_REP_MSG_PREPARE); + + /* read and return the prepare_data info to the caller */ + logicalrep_read_prepare(&s2, pdata); + + elog(DEBUG1, "PREPARE info: gid = '%s', prepare_lsn = %X/%X, end_lsn = %X/%X", + pdata->gid, + LSN_FORMAT_ARGS(pdata->prepare_lsn), + LSN_FORMAT_ARGS(pdata->end_lsn)); + } + else + { + /* Ensure we are reading the data into our memory context. */ + oldctx2 = MemoryContextSwitchTo(ApplyMessageContext); + + apply_dispatch(&s2); + + MemoryContextReset(ApplyMessageContext); + + MemoryContextSwitchTo(oldctx2); + + nchanges++; + + if (nchanges % 1000 == 0) + elog(DEBUG1, "replayed %d changes from file '%s'", + nchanges, path); + } + } + + BufFileClose(fd); + + pfree(buffer); + pfree(s2.data); + + elog(DEBUG1, "replayed %d (all) changes from file \"%s\"", + nchanges, path); + + return nchanges; +} + +/* + * Format the filename for the prepare spoolfile. + */ +static void +prepare_spoolfile_name(char *path, int szpath, Oid subid, char *gid) +{ + PsfHashEntry *hentry; + + /* + * This name is used as the key in the psf_hash HTAB. + * + * Therefore, the name and the key must be exactly same lengths and padded + * with '\0' so garbage does not impact the HTAB lookups. + */ + Assert(sizeof(hentry->name) == MAXPGPATH); + Assert(szpath == MAXPGPATH); + memset(path, '\0', MAXPGPATH); + + snprintf(path, MAXPGPATH, "%u-%s.prep_changes", subid, gid); +} diff --git a/src/include/replication/worker_internal.h b/src/include/replication/worker_internal.h index 1cac75e..95d78e9 100644 --- a/src/include/replication/worker_internal.h +++ b/src/include/replication/worker_internal.h @@ -86,6 +86,9 @@ extern void ReplicationOriginNameForTablesync(Oid suboid, Oid relid, char *originname, int szorgname); extern char *LogicalRepSyncTableStart(XLogRecPtr *origin_startpos); +extern bool AnyTablesyncInProgress(void); +extern XLogRecPtr BiggestTablesyncLSN(void); + void process_syncing_tables(XLogRecPtr current_lsn); void invalidate_syncing_table_states(Datum arg, int cacheid, uint32 hashvalue); diff --git a/src/tools/pgindent/typedefs.list b/src/tools/pgindent/typedefs.list index 745b51d..1b0c25b 100644 --- a/src/tools/pgindent/typedefs.list +++ b/src/tools/pgindent/typedefs.list @@ -1958,6 +1958,7 @@ ProtocolVersion PrsStorage PruneState PruneStepResult +PsfContext PsqlScanCallbacks PsqlScanQuoteType PsqlScanResult -- 1.8.3.1