From 5d0c4bbe6459e13b6336d6934ba231b6e1bba928 Mon Sep 17 00:00:00 2001 From: Peter Smith Date: Thu, 4 Mar 2021 20:12:24 +1100 Subject: [PATCH v48] 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 | 77 ++++++++++++++++++++++------- 2 files changed, 83 insertions(+), 23 deletions(-) diff --git a/src/backend/replication/logical/tablesync.c b/src/backend/replication/logical/tablesync.c index 97fc399..f3984d4 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) { @@ -1127,6 +1133,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; @@ -1149,6 +1157,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) @@ -1156,12 +1165,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"); + } } /* @@ -1175,6 +1189,8 @@ AnyTablesyncInProgress() int count = 0; ListCell *lc; + elog(LOG, "!!> AnyTablesyncInProgress?"); + /* We need up-to-date sync state info for subscription tables here. */ FetchTableStates(&started_tx); @@ -1186,8 +1202,8 @@ AnyTablesyncInProgress() SubscriptionRelState *rstate = (SubscriptionRelState *) lfirst(lc); count++; - elog(DEBUG1, - "AnyTablesyncInProgress?: #%d. Table relid %u has state '%c'", + elog(LOG, + "!!> AnyTablesyncInProgress?: #%d. Table relid %u has state '%c'", count, rstate->relid, rstate->state); @@ -1204,6 +1220,7 @@ AnyTablesyncInProgress() if (rstate->state != SUBREL_STATE_SYNCDONE && rstate->state != SUBREL_STATE_READY) { + elog(LOG, "!!> AnyTablesyncInProgress?: Table relid %u is busy!", rstate->relid); found_busy = true; break; } @@ -1215,8 +1232,8 @@ AnyTablesyncInProgress() pgstat_report_stat(false); } - elog(DEBUG1, - "AnyTablesyncInProgress?: Scanned %d tables, and found busy = %s", + elog(LOG, + "!!> AnyTablesyncInProgress?: Scanned %d tables, and found busy = %s", count, found_busy ? "true" : "false"); @@ -1242,8 +1259,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 f753e33..3c71f54 100644 --- a/src/backend/replication/logical/worker.c +++ b/src/backend/replication/logical/worker.c @@ -815,14 +815,16 @@ 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, end_lsn = %X/%X, final_lsn = %X/%X, lstate_lsn = %X/%X", + elog(LOG, + "!!> 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()) { + elog(LOG, "!!> apply_handle_begin_prepare - waiting for all sync workers to be DONE/READY"); + process_syncing_tables(begin_data.final_lsn); /* This latch is to prevent 100% CPU looping. */ @@ -840,7 +842,12 @@ apply_handle_begin_prepare(StringInfo s) * prepared) will be saved to a spoolfile for replay later at * commit_prepared time. */ - if (begin_data.final_lsn < BiggestTablesyncLSN()) + if (begin_data.final_lsn < BiggestTablesyncLSN() +#if 0 + || true /* XXX - Add this line to force psf (for + * easier debugging) */ +#endif + ) { char psfpath[MAXPGPATH]; StringInfoData sid; @@ -908,6 +915,8 @@ apply_handle_prepare(StringInfo s) * removal on proc-exit just in case there is an unexpected restart * between now and when commit_prepared happens. */ + elog(LOG, + "!!> apply_handle_prepare: Make sure the spoolfile is not removed on proc-exit"); hentry = (PsfHashEntry *) hash_search(psf_hash, psf_cur.name, HASH_FIND, @@ -990,6 +999,8 @@ apply_handle_commit_prepared(StringInfo s) int nchanges; LogicalRepPreparedTxnData pdata; + elog(LOG, "!!> apply_handle_commit_prepared: replaying the spooled messages"); + /* * 1. replay the spooled messages */ @@ -997,8 +1008,8 @@ apply_handle_commit_prepared(StringInfo s) ensure_transaction(); nchanges = prepare_spoolfile_replay_messages(psfpath, &pdata); - elog(DEBUG1, - "apply_handle_commit_prepared: replayed %d (all) changes.", + elog(LOG, + "!!> apply_handle_commit_prepared: replayed %d (all) changes.", nchanges); /* @@ -1103,6 +1114,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)) @@ -2493,18 +2505,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; @@ -3706,8 +3722,8 @@ IsLogicalWorker(void) static bool prepare_spoolfile_handler(LogicalRepMsgType action, StringInfo s) { - elog(DEBUG1, - "prepare_spoolfile_handler for action '%c'. %s write to spool file", + elog(LOG, + "!!> prepare_spoolfile_handler for action '%c'. %s write to spool file", action, psf_cur.is_spooling ? "Do" : "Don't"); @@ -3731,7 +3747,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_cur.is_spooling); @@ -3743,7 +3759,7 @@ prepare_spoolfile_create(char *path) if (!found) { - elog(DEBUG1, "Not found file \"%s\". Create it.", path); + elog(LOG, "!!> Not found file \"%s\". Create it.", path); psf_cur.vfd = PathNameOpenFile(path, O_RDWR | O_CREAT | O_TRUNC | PG_BINARY); if (psf_cur.vfd < 0) { @@ -3761,7 +3777,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_cur.vfd = PathNameOpenFile(path, O_RDWR | O_CREAT | O_TRUNC | PG_BINARY); if (psf_cur.vfd < 0) { @@ -3786,6 +3802,7 @@ prepare_spoolfile_create(char *path) static void prepare_spoolfile_close() { + elog(LOG, "!!> prepare_spoolfile_close"); if (psf_cur.vfd >= 0) FileClose(psf_cur.vfd); @@ -3801,6 +3818,8 @@ prepare_spoolfile_close() static void prepare_spoolfile_delete(char *path) { + elog(LOG, "!!> prepare_spoolfile_delete: \"%s\"", path); + /* The current psf should be closed already, but make sure anyway. */ prepare_spoolfile_close(); @@ -3826,18 +3845,20 @@ prepare_spoolfile_write(char action, StringInfo s) Assert(psf_cur.is_spooling); - 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); bytes_written = FileWrite(psf_cur.vfd, (char *)&len, sizeof(len), psf_cur.cur_offset, WAIT_EVENT_DATA_FILE_WRITE); Assert(bytes_written == sizeof(len)); psf_cur.cur_offset += bytes_written; /* then the action */ + elog(LOG, "!!> prepare_spoolfile_write: B writing action = %c, %d bytes", action, (int)sizeof(action)); bytes_written = FileWrite(psf_cur.vfd, &action, sizeof(action), psf_cur.cur_offset, WAIT_EVENT_DATA_FILE_WRITE); Assert(bytes_written == sizeof(action)); @@ -3846,6 +3867,7 @@ prepare_spoolfile_write(char action, StringInfo s) /* 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); bytes_written = FileWrite(psf_cur.vfd, &s->data[s->cursor], len, psf_cur.cur_offset, WAIT_EVENT_DATA_FILE_WRITE); Assert(bytes_written == len); @@ -3879,6 +3901,12 @@ prepare_spoolfile_exists(char *path) if (fd >= 0) FileClose(fd); + elog(LOG, + "!!> prepare_spoolfile_exists: Prepared spoolfile \"%s\" was " + "not found in the HTAB, but was %s on the disk.", + path, + found ? "found" : "not found"); + /* * And if it was found on disk then create the HTAB entry for it. */ @@ -3888,10 +3916,16 @@ prepare_spoolfile_exists(char *path) path, HASH_ENTER, NULL); + elog(LOG, "!!> prepare_spoolfile_exists: Created new HTAB entry '%s'", hentry->name); hentry->allow_delete = false; } } + elog(LOG, + "!!> prepare_spoolfile_exists: Prepared spoolfile \"%s\" was %s", + path, + found ? "found" : "not found"); + return found; } @@ -3910,8 +3944,8 @@ prepare_spoolfile_replay_messages(char *path, LogicalRepPreparedTxnData *pdata) oldctx2; PsfFile psf = { .is_spooling = false, .vfd = -1, .cur_offset = 0 }; - elog(DEBUG1, - "prepare_spoolfile_replay_messages: replaying changes from file \"%s\"", + elog(LOG, + "!!> prepare_spoolfile_replay_messages: replaying changes from file \"%s\"", path); /* @@ -3948,6 +3982,7 @@ prepare_spoolfile_replay_messages(char *path, LogicalRepPreparedTxnData *pdata) nbytes = FileRead(psf.vfd, (char *) &len, sizeof(len), psf.cur_offset, WAIT_EVENT_DATA_FILE_READ); psf.cur_offset += nbytes; + elog(LOG, "!!> prepare_spoolfile_replay_messages: nbytes = %d, len = %d", nbytes, len); /* have we reached end of the file? */ if (nbytes == 0) @@ -3966,6 +4001,7 @@ prepare_spoolfile_replay_messages(char *path, LogicalRepPreparedTxnData *pdata) buffer = repalloc(buffer, len); /* and finally read the data into the buffer */ + elog(LOG, "!!> prepare_spoolfile_replay_messages: read %d bytes into buffer", len); nbytes = FileRead(psf.vfd, buffer, len, psf.cur_offset, WAIT_EVENT_DATA_FILE_READ); psf.cur_offset += nbytes; @@ -4000,13 +4036,15 @@ prepare_spoolfile_replay_messages(char *path, LogicalRepPreparedTxnData *pdata) { LogicalRepBeginPrepareData bdata; + elog(LOG, "!!> prepare_spoolfile_replay_messages: Found the BEGIN_PREPARE info"); + /* 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", + elog(LOG, "!!> 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)); @@ -4021,13 +4059,15 @@ prepare_spoolfile_replay_messages(char *path, LogicalRepPreparedTxnData *pdata) } else if (s2.data[0] == LOGICAL_REP_MSG_PREPARE) { + elog(LOG, "!!> prepare_spoolfile_replay_messages: Found the PREPARE info"); + /* 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", + elog(LOG, "!!> 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)); @@ -4046,7 +4086,7 @@ prepare_spoolfile_replay_messages(char *path, LogicalRepPreparedTxnData *pdata) nchanges++; if (nchanges % 1000 == 0) - elog(DEBUG1, "replayed %d changes from file '%s'", + elog(LOG, "!!> replayed %d changes from file '%s'", nchanges, path); } } @@ -4056,7 +4096,7 @@ prepare_spoolfile_replay_messages(char *path, LogicalRepPreparedTxnData *pdata) 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; @@ -4092,6 +4132,8 @@ prepare_spoolfile_on_proc_exit(int status, Datum arg) HASH_SEQ_STATUS seq_status; PsfHashEntry *hentry; + elog(LOG, "!!> prepare_spoolfile_on_proc_exit"); + /* Iterate the HTAB looking for what file can be deleted. */ if (psf_hash) { @@ -4100,6 +4142,7 @@ prepare_spoolfile_on_proc_exit(int status, Datum arg) { char *path = hentry->name; + elog(LOG, "!!> prepare_spoolfile_proc_exit: found '%s'", path); if (hentry->allow_delete) prepare_spoolfile_delete(path); } -- 1.8.3.1