diff options
| -rw-r--r-- | src/backend/replication/walsender.c | 67 |
1 files changed, 37 insertions, 30 deletions
diff --git a/src/backend/replication/walsender.c b/src/backend/replication/walsender.c index b24f9a1e95b..3a06a4a307a 100644 --- a/src/backend/replication/walsender.c +++ b/src/backend/replication/walsender.c @@ -1066,6 +1066,9 @@ static void WalSndWriteData(LogicalDecodingContext *ctx, XLogRecPtr lsn, TransactionId xid, bool last_write) { + TimestampTz now; + int64 now_int; + /* output previously gathered data in a CopyData packet */ pq_putmessage_noblock('d', ctx->out->data, ctx->out->len); @@ -1075,53 +1078,35 @@ WalSndWriteData(LogicalDecodingContext *ctx, XLogRecPtr lsn, TransactionId xid, * several releases by streaming physical replication. */ resetStringInfo(&tmpbuf); - pq_sendint64(&tmpbuf, GetCurrentIntegerTimestamp()); + now_int = GetCurrentIntegerTimestamp(); + now = IntegerTimestampToTimestampTz(now_int); + pq_sendint64(&tmpbuf, now_int); memcpy(&ctx->out->data[1 + sizeof(int64) + sizeof(int64)], tmpbuf.data, sizeof(int64)); - /* fast path */ + CHECK_FOR_INTERRUPTS(); + /* Try to flush pending output to the client */ if (pq_flush_if_writable() != 0) WalSndShutdown(); - if (!pq_is_send_pending()) + /* Try taking fast path unless we get too close to walsender timeout. */ + if (now < TimestampTzPlusMilliseconds(last_reply_timestamp, + wal_sender_timeout / 2) && + !pq_is_send_pending()) + { return; + } + /* If we have pending write here, go to slow path */ for (;;) { int wakeEvents; long sleeptime; - TimestampTz now; - - /* - * Emergency bailout if postmaster has died. This is to avoid the - * necessity for manual cleanup of all postmaster children. - */ - if (!PostmasterIsAlive()) - exit(1); - - /* Process any requests or signals received recently */ - if (ConfigReloadPending) - { - ConfigReloadPending = false; - ProcessConfigFile(PGC_SIGHUP); - SyncRepInitConfig(); - } /* Check for input from the client */ ProcessRepliesIfAny(); - /* Clear any already-pending wakeups */ - ResetLatch(&MyWalSnd->latch); - - /* Try to flush pending output to the client */ - if (pq_flush_if_writable() != 0) - WalSndShutdown(); - - /* If we finished clearing the buffered data, we're done here. */ - if (!pq_is_send_pending()) - break; - now = GetCurrentTimestamp(); /* die if timeout was reached */ @@ -1130,6 +1115,9 @@ WalSndWriteData(LogicalDecodingContext *ctx, XLogRecPtr lsn, TransactionId xid, /* Send keepalive if the time has come */ WalSndKeepaliveIfNecessary(now); + if (!pq_is_send_pending()) + break; + sleeptime = WalSndComputeSleeptime(now); wakeEvents = WL_LATCH_SET | WL_POSTMASTER_DEATH | @@ -1141,6 +1129,25 @@ WalSndWriteData(LogicalDecodingContext *ctx, XLogRecPtr lsn, TransactionId xid, WaitLatchOrSocket(&MyWalSnd->latch, wakeEvents, MyProcPort->sock, sleeptime); ImmediateInterruptOK = false; + + /* + * Emergency bailout if postmaster has died. This is to avoid the + * necessity for manual cleanup of all postmaster children. + */ + if (!PostmasterIsAlive()) + exit(1); + + /* Process any requests or signals received recently */ + if (ConfigReloadPending) + { + ConfigReloadPending = false; + ProcessConfigFile(PGC_SIGHUP); + SyncRepInitConfig(); + } + + /* Try to flush pending output to the client */ + if (pq_flush_if_writable() != 0) + WalSndShutdown(); } /* reactivate latch so WalSndLoop knows to continue */ |