diff --git a/src/backend/replication/logical/applyparallelworker.c b/src/backend/replication/logical/applyparallelworker.c index 3acf9d5fb9..1fc62cdba2 100644 --- a/src/backend/replication/logical/applyparallelworker.c +++ b/src/backend/replication/logical/applyparallelworker.c @@ -74,10 +74,10 @@ * messages, and this wait doesn't appear in lmgr. * * To resolve this issue, we use non-blocking write and wait with a timeout. If - * timeout is exceeded, the LA will serialize the message to a file and + * the timeout is exceeded, the LA will serialize the message to a file and * indicate PA-2 that it needs to read that file for the remaining messages. * Then LA will start waiting for commit which will detect deadlock if any. - * (See pa_send_data() and typedef enum TransApplyAction) + * (See pa_send_data() and enum TransApplyAction) * * 4) Lock type * @@ -1147,9 +1147,10 @@ pa_send_data(ParallelApplyWorkerInfo *winfo, Size nbytes, const void *data) LogicalRepMsgType action; /* - * The parallel apply worker might be stuck for some reason, so - * stop sending data directly to it and start to serialize data to - * files instead. + * The parallel apply worker could be stuck for some reason (say + * waiting on some lock by other backend), so stop trying to send + * data directly to it and instead start to serialize data to + * file instead. */ winfo->serialize_changes = true; diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c index b5fb74572e..c51d92b67f 100644 --- a/src/backend/replication/logical/worker.c +++ b/src/backend/replication/logical/worker.c @@ -1331,7 +1331,6 @@ apply_handle_stream_prepare(StringInfo s) */ pa_unlock_stream(winfo->shared->xid, AccessExclusiveLock); - /* Send STREAM PREPARE message to the parallel apply worker. */ if (apply_action == TRANS_LEADER_SEND_TO_PARALLEL) pa_send_data(winfo, s->len, s->data); else @@ -1339,6 +1338,10 @@ apply_handle_stream_prepare(StringInfo s) LOGICAL_REP_MSG_STREAM_PREPARE, &original_msg); + /* + * It is possible that while sending this change to parallel apply + * worker we need to switch to serialize mode. + */ if (winfo->serialize_changes) pa_set_fileset_state(winfo->shared, FS_READY); @@ -1840,6 +1843,10 @@ apply_handle_stream_abort(StringInfo s) if (toplevel_xact) { + /* + * It is possible that while sending this change to parallel apply + * worker we need to switch to serialize mode. + */ if (winfo->serialize_changes) pa_set_fileset_state(winfo->shared, FS_READY); @@ -2062,13 +2069,16 @@ apply_handle_stream_commit(StringInfo s) */ pa_unlock_stream(xid, AccessExclusiveLock); - /* Send STREAM COMMIT message to the parallel apply worker. */ if (apply_action == TRANS_LEADER_SEND_TO_PARALLEL) pa_send_data(winfo, s->len, s->data); else stream_open_and_write_change(xid, LOGICAL_REP_MSG_STREAM_COMMIT, &original_msg); + /* + * It is possible that while sending this change to parallel apply + * worker we need to switch to serialize mode. + */ if (winfo->serialize_changes) pa_set_fileset_state(winfo->shared, FS_READY);