From a2c4c4b6fd01fe00e8e87fd586e37e8d775a8c4b Mon Sep 17 00:00:00 2001 From: Peter Smith Date: Thu, 25 Feb 2021 17:24:52 +1100 Subject: [PATCH v43] Fix apply worker empty prepare (dev logs). NOT TO BE COMMITTED. This patch is only for adding some developer logging which may help for debugging/testing the "Fix apply worker empty prepare" patch. --- src/backend/replication/logical/tablesync.c | 29 +++++++++++++---- src/backend/replication/logical/worker.c | 50 ++++++++++++++++++++--------- 2 files changed, 58 insertions(+), 21 deletions(-) diff --git a/src/backend/replication/logical/tablesync.c b/src/backend/replication/logical/tablesync.c index 8b20519..5b5b910 100644 --- a/src/backend/replication/logical/tablesync.c +++ b/src/backend/replication/logical/tablesync.c @@ -282,6 +282,12 @@ process_syncing_tables_for_sync(XLogRecPtr current_lsn) { SpinLockAcquire(&MyLogicalRepWorker->relmutex); + elog(LOG, + "!!> process_syncing_tables_for_sync: state = '%c', current_lsn = %X/%X, relstate_lsn = %X/%X", + MyLogicalRepWorker->relstate, + LSN_FORMAT_ARGS(current_lsn), + LSN_FORMAT_ARGS(MyLogicalRepWorker->relstate_lsn)); + if (MyLogicalRepWorker->relstate == SUBREL_STATE_CATCHUP && current_lsn >= MyLogicalRepWorker->relstate_lsn) { @@ -1131,6 +1137,8 @@ FetchTableStates(bool *started_tx) ListCell *lc; SubscriptionRelState *rstate; + elog(LOG, "!!> FetchTableStates: Re-fetching the state list caches"); + /* Clean the old lists. */ list_free_deep(table_states_all); table_states_all = NIL; @@ -1153,6 +1161,7 @@ FetchTableStates(bool *started_tx) rstate = palloc(sizeof(SubscriptionRelState)); memcpy(rstate, cur_rstate, sizeof(SubscriptionRelState)); table_states_all = lappend(table_states_all, rstate); + elog(LOG, "!!> FetchTableStates: table_states_all - added Table relid %u with state '%c'", rstate->relid, rstate->state); /* List of only not-ready states */ if (cur_rstate->state != SUBREL_STATE_READY) @@ -1160,12 +1169,17 @@ FetchTableStates(bool *started_tx) rstate = palloc(sizeof(SubscriptionRelState)); memcpy(rstate, cur_rstate, sizeof(SubscriptionRelState)); table_states_not_ready = lappend(table_states_not_ready, rstate); + elog(LOG, "!!> FetchTableStates: table_states_not_ready - added Table relid %u with state '%c'", rstate->relid, rstate->state); } } MemoryContextSwitchTo(oldctx); table_states_valid = true; } + else + { + elog(LOG, "!!> FetchTableStates: Already up-to-date"); + } } /* @@ -1179,6 +1193,8 @@ BusyTablesyncs() int count = 0; ListCell *lc; + elog(LOG, "!!> BusyTablesyncs"); + /* We need up-to-date sync state info for subscription tables here. */ FetchTableStates(&started_tx); @@ -1190,8 +1206,8 @@ BusyTablesyncs() SubscriptionRelState *rstate = (SubscriptionRelState *) lfirst(lc); count++; - elog(DEBUG1, - "BusyTablesyncs: #%d. Table relid %u has state '%c'", + elog(LOG, + "!!> BusyTablesyncs: #%d. Table relid %u has state '%c'", count, rstate->relid, rstate->state); @@ -1208,6 +1224,7 @@ BusyTablesyncs() if (rstate->state != SUBREL_STATE_SYNCDONE && rstate->state != SUBREL_STATE_READY) { + elog(LOG, "!!> BusyTablesyncs: Table relid %u is busy!", rstate->relid); found_busy = true; break; } @@ -1219,8 +1236,8 @@ BusyTablesyncs() pgstat_report_stat(false); } - elog(DEBUG1, - "BusyTablesyncs: Scanned %d tables, and found busy = %s", + elog(LOG, + "!!> BusyTablesyncs: Scanned %d tables, and found busy = %s", count, found_busy ? "true" : "false"); @@ -1246,8 +1263,8 @@ BiggestTablesyncLSN() biggest_lsn = rstate->lsn; } - elog(DEBUG1, - "BiggestTablesyncLSN: Scanned %d tables. Biggest lsn found = %X/%X", + elog(LOG, + "!!> BiggestTablesyncLSN: Scanned %d tables. Biggest lsn found = %X/%X", count, LSN_FORMAT_ARGS(biggest_lsn)); diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c index d9b7cfa..addcbbf 100644 --- a/src/backend/replication/logical/worker.c +++ b/src/backend/replication/logical/worker.c @@ -820,14 +820,14 @@ apply_handle_begin_prepare(StringInfo s) * Make sure every tablesync has reached at least SYNCDONE state before * letting the apply worker proceed. */ - elog(DEBUG1, - "apply_handle_begin_prepare, begin_data.end_lsn = %X/%X, lstate_lsn = %X/%X", + elog(LOG, + "!!> apply_handle_begin_prepare, begin_data.end_lsn = %X/%X, lstate_lsn = %X/%X", LSN_FORMAT_ARGS(begin_data.end_lsn), LSN_FORMAT_ARGS(MyLogicalRepWorker->relstate_lsn)); while (BusyTablesyncs()) { - elog(DEBUG1, "apply_handle_begin_prepare - waiting for all sync workers to be DONE/READY"); + elog(LOG, "!!> apply_handle_begin_prepare - waiting for all sync workers to be DONE/READY"); process_syncing_tables(begin_data.end_lsn); @@ -985,6 +985,8 @@ apply_handle_commit_prepared(StringInfo s) { int nchanges; + elog(LOG, "!!> apply_handle_commit_prepared: replaying the spooled messages"); + /* * 1. replay the spooled messages */ @@ -992,8 +994,8 @@ apply_handle_commit_prepared(StringInfo s) ensure_transaction(); nchanges = prepare_spoolfile_replay_messages(psfpath, prepare_data.end_lsn); - elog(DEBUG1, - "apply_handle_commit_prepared: replayed %d (all) changes.", + elog(LOG, + "!!> apply_handle_commit_prepared: replayed %d (all) changes.", nchanges); prepare_spoolfile_cleanup(psfpath); @@ -1104,6 +1106,7 @@ apply_handle_rollback_prepared(StringInfo s) * Prepare Spoolfile (using_psf) because in that case there is no matching * PrepareTransactionBlock done yet. */ + elog(LOG, "!!> apply_handle_rollback_prepared: using_psf=%d", using_psf); if (!using_psf && LookupGXact(rollback_data.gid, rollback_data.prepare_end_lsn, rollback_data.preparetime)) @@ -2533,18 +2536,22 @@ apply_dispatch(StringInfo s) return; case LOGICAL_REP_MSG_BEGIN_PREPARE: + elog(LOG, "!!> ------ apply_handle_begin_prepare ------"); apply_handle_begin_prepare(s); return; case LOGICAL_REP_MSG_PREPARE: + elog(LOG, "!!> ------ apply_handle_prepare ------"); apply_handle_prepare(s); return; case LOGICAL_REP_MSG_COMMIT_PREPARED: + elog(LOG, "!!> ------ apply_handle_commit_prepared ------"); apply_handle_commit_prepared(s); return; case LOGICAL_REP_MSG_ROLLBACK_PREPARED: + elog(LOG, "!!> ------ apply_handle_rollback_prepared ------"); apply_handle_rollback_prepared(s); return; @@ -3751,6 +3758,11 @@ IsLogicalWorker(void) static bool prepare_spoolfile_handler(LogicalRepMsgType action, StringInfo s) { + elog(LOG, + "!!> prepare_spoolfile_handler for action '%c'. %s write to spool file", + action, + psf_fd ? "Do" : "Don't"); + if (psf_fd == NULL) return false; @@ -3772,7 +3784,7 @@ prepare_spoolfile_create(char *path) bool found; PsfHashEntry *hentry; - elog(DEBUG1, "creating file \"%s\" for prepare changes", path); + elog(LOG, "!!> creating file \"%s\" for prepare changes", path); Assert(psf_fd == NULL); @@ -3793,7 +3805,7 @@ prepare_spoolfile_create(char *path) MemoryContext savectx; SharedFileSet *fileset; - elog(DEBUG1, "Not found file \"%s\"", path); + elog(LOG, "!!> Not found file \"%s\". Create it.", path); savectx = MemoryContextSwitchTo(ApplyContext); fileset = palloc(sizeof(SharedFileSet)); @@ -3812,7 +3824,7 @@ prepare_spoolfile_create(char *path) * Open the file and seek to the beginning because we always want to * create/overwrite this file. */ - elog(DEBUG1, "Found file \"%s\". Overwrite it.", path); + elog(LOG, "!!> Found file \"%s\". Overwrite it.", path); psf_fd = BufFileOpenShared(hentry->fileset, path, O_RDWR); BufFileSeek(psf_fd, 0, 0L, SEEK_SET); } @@ -3829,6 +3841,7 @@ prepare_spoolfile_create(char *path) static void prepare_spoolfile_close() { + elog(LOG, "!!> prepare_spoolfile_close"); if (psf_fd) BufFileClose(psf_fd); psf_fd = NULL; @@ -3842,6 +3855,8 @@ prepare_spoolfile_cleanup(char *path) { PsfHashEntry *hentry; + elog(LOG, "!!> prepare_spoolfile_cleanup: \"%s\"", path); + /* The current psf should be closed already, but make sure anyway. */ prepare_spoolfile_close(); @@ -3877,20 +3892,23 @@ prepare_spoolfile_write(char action, StringInfo s) Assert(psf_fd != NULL); - elog(DEBUG1, "prepare_spoolfile_write: writing action '%c'", action); + elog(LOG, "!!> 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 */ + elog(LOG, "!!> prepare_spoolfile_write: A writing len bytes = %d", len); BufFileWrite(psf_fd, &len, sizeof(len)); /* then the action */ + elog(LOG, "!!> prepare_spoolfile_write: B writing action = %c, %d bytes", action, (int)sizeof(action)); BufFileWrite(psf_fd, &action, sizeof(action)); /* and finally the remaining part of the buffer (after the XID) */ len = (s->len - s->cursor); + elog(LOG, "!!> prepare_spoolfile_write: C writing len bytes = %d", len); BufFileWrite(psf_fd, &s->data[s->cursor], len); } @@ -3908,8 +3926,8 @@ prepare_spoolfile_exists(char *path) HASH_FIND, &found); - elog(DEBUG1, - "prepare_spoolfile_exists: Prepared spoolfile \"%s\" was %s", + elog(LOG, + "!!> prepare_spoolfile_exists: Prepared spoolfile \"%s\" was %s", path, found ? "found" : "not found"); @@ -3932,8 +3950,8 @@ prepare_spoolfile_replay_messages(char *path, XLogRecPtr lsn) PsfHashEntry *hentry; BufFile *fd; - elog(DEBUG1, - "prepare_spoolfile_replay_messages: replaying changes from file \"%s\"", + elog(LOG, + "!!> prepare_spoolfile_replay_messages: replaying changes from file \"%s\"", path); /* @@ -3978,6 +3996,7 @@ prepare_spoolfile_replay_messages(char *path, XLogRecPtr lsn) /* read length of the on-disk record */ nbytes = BufFileRead(fd, &len, sizeof(len)); + elog(LOG, "!!> prepare_spoolfile_replay_messages: nbytes = %d, len = %d", nbytes, len); /* have we reached end of the file? */ if (nbytes == 0) @@ -3996,6 +4015,7 @@ prepare_spoolfile_replay_messages(char *path, XLogRecPtr lsn) buffer = repalloc(buffer, len); /* and finally read the data into the buffer */ + elog(LOG, "!!> prepare_spoolfile_replay_messages: read %d bytes into buffer", len); if (BufFileRead(fd, buffer, len) != len) ereport(ERROR, (errcode_for_file_access(), @@ -4018,7 +4038,7 @@ prepare_spoolfile_replay_messages(char *path, XLogRecPtr lsn) nchanges++; if (nchanges % 1000 == 0) - elog(DEBUG1, "replayed %d changes from file '%s'", + elog(LOG, "!!> replayed %d changes from file '%s'", nchanges, path); } @@ -4027,7 +4047,7 @@ prepare_spoolfile_replay_messages(char *path, XLogRecPtr lsn) pfree(buffer); pfree(s2.data); - elog(DEBUG1, "replayed %d (all) changes from file \"%s\"", + elog(LOG, "!!> replayed %d (all) changes from file \"%s\"", nchanges, path); return nchanges; -- 1.8.3.1