*** a/doc/src/sgml/config.sgml --- b/doc/src/sgml/config.sgml *************** *** 1760,1809 **** SET ENABLE_SEQSCAN TO OFF; - - commit_delay (integer) - - commit_delay configuration parameter - - - - When the commit data for a transaction is flushed to disk, any - additional commits ready at that time are also flushed out. - commit_delay adds a time delay, set in - microseconds, before a transaction attempts to - flush the WAL buffer out to disk. A nonzero delay can allow more - transactions to be committed with only one flush operation, if - system load is high enough that additional transactions become - ready to commit within the given interval. But the delay is - just wasted if no other transactions become ready to - commit. Therefore, the delay is only performed if at least - commit_siblings other transactions are - active at the instant that a server process has written its - commit record. - The default commit_delay is zero (no delay). - Since all pending commit data will be written at every flush - regardless of this setting, it is rare that adding delay - by increasing this parameter will actually improve performance. - - - - - - commit_siblings (integer) - - commit_siblings configuration parameter - - - - Minimum number of concurrent open transactions to require - before performing the commit_delay delay. A larger - value makes it more probable that at least one other - transaction will become ready to commit during the delay - interval. The default is five transactions. - - - - --- 1760,1765 ---- *** a/doc/src/sgml/wal.sgml --- b/doc/src/sgml/wal.sgml *************** *** 367,386 **** of data corruption. - - also sounds very similar to - asynchronous commit, but it is actually a synchronous commit method - (in fact, commit_delay is ignored during an - asynchronous commit). commit_delay causes a delay - just before a synchronous commit attempts to flush - WAL to disk, in the hope that a single flush - executed by one such transaction can also serve other transactions - committing at about the same time. Setting commit_delay - can only help when there are many concurrently committing transactions, - and it is difficult to tune it to a value that actually helps rather - than hurt throughput. - - --- 367,372 ---- *************** *** 556,579 **** - The parameter defines for how many - microseconds the server process will sleep after writing a commit - record to the log with LogInsert but before - performing a LogFlush. This delay allows other - server processes to add their commit records to the log so as to have all - of them flushed with a single log sync. No sleep will occur if - - is not enabled, or if fewer than - other sessions are currently in active transactions; this avoids - sleeping when it's unlikely that any other session will commit soon. - Note that on most platforms, the resolution of a sleep request is - ten milliseconds, so that any nonzero commit_delay - setting between 1 and 10000 microseconds would have the same effect. - Good values for these parameters are not yet clear; experimentation - is encouraged. - - - The parameter determines how PostgreSQL will ask the kernel to force WAL updates out to disk. --- 542,547 ---- *** a/src/backend/access/transam/Makefile --- b/src/backend/access/transam/Makefile *************** *** 13,19 **** top_builddir = ../../../.. include $(top_builddir)/src/Makefile.global OBJS = clog.o transam.o varsup.o xact.o rmgr.o slru.o subtrans.o multixact.o \ ! twophase.o twophase_rmgr.o xlog.o xlogfuncs.o xlogutils.o include $(top_srcdir)/src/backend/common.mk --- 13,19 ---- include $(top_builddir)/src/Makefile.global OBJS = clog.o transam.o varsup.o xact.o rmgr.o slru.o subtrans.o multixact.o \ ! twophase.o twophase_rmgr.o xlog.o xlogfuncs.o xlogutils.o groupcommit.o include $(top_srcdir)/src/backend/common.mk *** /dev/null --- b/src/backend/access/transam/groupcommit.c *************** *** 0 **** --- 1,346 ---- + /*------------------------------------------------------------------------- + * + * groupcommit.c + * + * Group Commit is new as of PostgreSQL 9.2. It is a performance optimization + * where the fsyncing of WAL is batched across many more-or-less concurrently + * committing transactions, thereby amortizing the cost of that notoriously + * expensive operation, and potentially greatly increasing transaction + * throughput. + * + * Transactions requesting XlogFlush() wait until the WALWriter has performed + * the write for them, then will be woken to continue. This module contains the + * code for waiting and release of backends. + * + * Manage waiting backends by having a single ordered queue of waiting backends, + * so that we can avoid searching through all waiters each time the WAL Writer + * writes/fsyncs. + * + * Copyright (c) 2010-2012, PostgreSQL Global Development Group + * + * IDENTIFICATION + * src/backend/access/transam/groupcommit.c + * + *------------------------------------------------------------------------- + */ + + #include "postgres.h" + + #include + + #include "access/groupcommit.h" + #include "access/xact.h" + #include "access/xlog.h" + #include "access/xlog_internal.h" + #include "miscadmin.h" + #include "storage/pmsignal.h" + #include "storage/proc.h" + #include "tcop/tcopprot.h" + #include "utils/builtins.h" + + /* + * Shared memory structure + */ + typedef struct + { + SHM_QUEUE queue; /* list of processes waiting to be woken up */ + PGPROC *driver; /* process currently flushing the WAL */ + } GroupCommitData; + + static GroupCommitData *gcShared; + + + static bool amdriving = false; + + /* Shmem functions */ + Size + GroupCommitShmemSize(void) + { + return sizeof(GroupCommitData); + } + + void + GroupCommitShmemInit(void) + { + bool found; + + /* + * Create or attach to the GroupCommitShared structure. + * + */ + gcShared = (GroupCommitData *) + ShmemInitStruct("Group Commit", GroupCommitShmemSize(), &found); + + if (!found) + { + /* First time through, so initialize it */ + SHMQueueInit(&gcShared->queue); + gcShared->driver = NULL; + } + } + + /* + * ================================================ + * Group Commit functions for normal user backends + * ================================================ + */ + + /* + * Wait for group commit, and then return true, if group commit serviced the + * request (not necessarily successfully). Otherwise, return false and fastpath + * out of here, allowing the backend to make alternative arrangements to flush + * its WAL in a more granular fashion. This can happen because the record that + * the backend requests to have flushed in so far into the future that to group + * commit it would + * + * Initially backends start in state GROUP_COMMIT_INIT. Backends that service + * database connections will then change that state to GROUP_COMMIT_NOT_WAITING + * before adding itself to the wait queue. During GroupCommitWakeQueue(), the + * WAL Writer changes the state to GROUP_COMMIT_WAIT_COMPLETE once write/fsync + * is confirmed. + * + * The backend then resets its state to GROUP_COMMIT_NOT_WAITING. + */ + void + GroupCommitWaitForLSN(XLogRecPtr XactCommitLSN) + { + Assert(SHMQueueIsDetached(&(MyProc->waitLSNLinks))); + Assert(MyProc->grpCommitState == GROUP_COMMIT_NOT_WAITING); + Assert(!LWLockHeldByMe(SyncRepLock)); + + Assert(!amdriving); + + LWLockAcquire(GroupCommitLock, LW_EXCLUSIVE); + + /* See if we have to drive */ + if (gcShared->driver == NULL) + { + amdriving = true; + gcShared->driver = MyProc; + LWLockRelease(GroupCommitLock); + + DoXLogFlush(XactCommitLSN); + return; + } + else + { + /* + * Set our waitLSN, add ourselves to the queue, and Wait for the + * driver to wake us up + */ + MyProc->waitLSN = XactCommitLSN; + MyProc->grpCommitState = GROUP_COMMIT_WAITING; + BatchCommitQueueInsert(&gcShared->queue); + Assert(BatchCommitQueueIsOrderedByLSN(&gcShared->queue)); + + LWLockRelease(GroupCommitLock); + + #ifdef GROUP_COMMIT_DEBUG + elog(LOG, "waiting for LSN %X/%X", + XactCommitLSN.xlogid, + XactCommitLSN.xrecoff); + #endif + + /* + * Wait for specified LSN to be confirmed. + * + * Each proc has its own wait latch, so we perform a normal latch + * check/wait loop here. + */ + for (;;) + { + GroupCommitState CommitState; + + /* + * Wait until the driver services our request to flush WAL (or + * wakes us up to become the next driver). + */ + PGSemaphoreLock(&MyProc->sem, true); + + /* + * Try checking the state without the lock first. There's no + * guarantee that we'll read the most up-to-date value, so if it + * looks like we're still waiting, recheck while holding the lock. + * But if it looks like we're done, we must really be done, because + * once the driver updates our state to GROUP_COMMIT_WAIT_COMPLETE, + * it will never update it again, so we can't be seeing a stale + * value in that case. Control will never reach here if its value + * is GROUP_COMMIT_INIT, as it is in the case of auxiliary + * processes. + */ + CommitState = ((volatile PGPROC *) MyProc)->grpCommitState; + Assert(CommitState != GROUP_COMMIT_NOT_WAITING); + if (CommitState == GROUP_COMMIT_WAIT_COMPLETE) + break; + if (CommitState == GROUP_COMMIT_WAITING) + { + LWLockAcquire(GroupCommitLock, LW_EXCLUSIVE); + CommitState = MyProc->grpCommitState; + + /* + * If we're still not finished, and no-one is driving, + * take the driver's seat + */ + if (CommitState == GROUP_COMMIT_WAIT_COMPLETE) + { + LWLockRelease(GroupCommitLock); + break; + } + if (CommitState == GROUP_COMMIT_WAITING && gcShared->driver == NULL) + { + /* Remove ourself from the queue */ + SHMQueueDelete(&(MyProc->waitLSNLinks)); + MyProc->grpCommitState = GROUP_COMMIT_NOT_WAITING; + MyProc->waitLSN.xlogid = 0; + MyProc->waitLSN.xrecoff = 0; + + amdriving = true; + gcShared->driver = MyProc; + LWLockRelease(GroupCommitLock); + + DoXLogFlush(XactCommitLSN); + return; + } + + LWLockRelease(GroupCommitLock); + } + } + + /* + * The WAL has been flushed up to our LSN and we have been removed from + * the queue. Clean up state and leave. It's OK to reset these shared + * memory fields without holding GroupCommitLock, because the WAL + * writer will ignore us anyway when we're not on the queue. + */ + Assert(SHMQueueIsDetached(&(MyProc->waitLSNLinks))); + MyProc->grpCommitState = GROUP_COMMIT_NOT_WAITING; + MyProc->waitLSN.xlogid = 0; + MyProc->waitLSN.xrecoff = 0; + } + } + + /* + * Walk queue from head. Set the state of any backends that need to be woken + * due to waiting on an LSN up to flushLSN, then remove them from the queue and + * wake them up. + * + * Caller must hold GroupCommitLock in exclusive mode. + */ + void + GroupCommitReleaseWaiters(XLogRecPtr flushLSN) + { + PGPROC *proc = NULL; + PGPROC *thisproc = NULL; + int numprocs = 0; + + LWLockAcquire(GroupCommitLock, LW_EXCLUSIVE); + + if (gcShared->driver == MyProc) + gcShared->driver = NULL; + amdriving = false; + + Assert(BatchCommitQueueIsOrderedByLSN(&gcShared->queue)); + Assert(!LWLockHeldByMe(SyncRepLock)); + + proc = (PGPROC *) SHMQueueNext(&gcShared->queue, + &gcShared->queue, + offsetof(PGPROC, waitLSNLinks)); + + while (proc) + { + /* + * Assume the queue is ordered by LSN (assumption is verified elsewhere + * by assertions). + */ + if (XLByteLT(flushLSN, proc->waitLSN)) + break; + + /* + * Move to next proc, so we can delete thisproc from the queue. + * thisproc is valid, proc may be NULL after this. + */ + thisproc = proc; + proc = (PGPROC *) SHMQueueNext(&gcShared->queue, + &(proc->waitLSNLinks), + offsetof(PGPROC, waitLSNLinks)); + + /* + * Set state to complete; see GroupCommitWaitForLSN() for discussion of + * the various states. + */ + thisproc->grpCommitState = GROUP_COMMIT_WAIT_COMPLETE; + /* + * Remove thisproc from queue. + */ + SHMQueueDelete(&(thisproc->waitLSNLinks)); + + /* + * Wake only when we have set state and removed from queue. + */ + PGSemaphoreUnlock(&thisproc->sem); + + numprocs++; + } + + /* + * Finally, wake up the process that's *last* in the queue, with the + * highest LSN. He's not done yet, but will become the next driver. + */ + if (gcShared->driver == NULL) + { + proc = (PGPROC *) SHMQueuePrev(&gcShared->queue, + &gcShared->queue, + offsetof(PGPROC, waitLSNLinks)); + if (proc) + PGSemaphoreUnlock(&proc->sem); + } + + LWLockRelease(GroupCommitLock); + + #ifdef GROUP_COMMIT_DEBUG + elog(LOG, "released %d procs up to %X/%X", + numprocs, + flushLSN.xlogid, + flushLSN.xrecoff); + #endif + } + + void + GroupCommitCleanupAtProcExit(void) + { + PGPROC *proc; + + /* XXX: check if we're driving, and clear if so */ + if (!SHMQueueIsDetached(&(MyProc->waitLSNLinks))) + { + Assert(MyProc->grpCommitState != GROUP_COMMIT_NOT_WAITING); + LWLockAcquire(GroupCommitLock, LW_EXCLUSIVE); + SHMQueueDelete(&(MyProc->waitLSNLinks)); + LWLockRelease(GroupCommitLock); + } + if (amdriving) + { + amdriving = false; + LWLockAcquire(GroupCommitLock, LW_EXCLUSIVE); + /* + * Zap the old driver. Presumably it was myself, but since we + * shouldn't get here in the first place, maybe we're all confused. + * In any case, it's always safe to clear the driver field, as long + * as we wake up someone from the wait queue to become the new driver. + * It's ok if two backends think they're driving. + */ + gcShared->driver = NULL; + + /* + * If the queue is not empty, wake up someone to become the new + * driver. + */ + proc = (PGPROC *) SHMQueueNext(&gcShared->queue, + &gcShared->queue, + offsetof(PGPROC, waitLSNLinks)); + if (proc) + PGSemaphoreUnlock(&proc->sem); + + LWLockRelease(GroupCommitLock); + } + } *** a/src/backend/access/transam/xact.c --- b/src/backend/access/transam/xact.c *************** *** 67,75 **** bool XactDeferrable; int synchronous_commit = SYNCHRONOUS_COMMIT_ON; - int CommitDelay = 0; /* precommit delay in microseconds */ - int CommitSiblings = 5; /* # concurrent xacts needed to sleep */ - /* * MyXactAccessedTempRel is set when a temporary relation is accessed. * We don't allow PREPARE TRANSACTION in that case. (This is global --- 67,72 ---- *************** *** 1094,1115 **** RecordTransactionCommit(void) if ((wrote_xlog && synchronous_commit > SYNCHRONOUS_COMMIT_OFF) || forceSyncCommit || nrels > 0) { - /* - * Synchronous commit case: - * - * Sleep before flush! So we can flush more than one commit records - * per single fsync. (The idea is some other backend may do the - * XLogFlush while we're sleeping. This needs work still, because on - * most Unixen, the minimum select() delay is 10msec or more, which is - * way too long.) - * - * We do not sleep if enableFsync is not turned on, nor if there are - * fewer than CommitSiblings other backends with active transactions. - */ - if (CommitDelay > 0 && enableFsync && - MinimumActiveBackends(CommitSiblings)) - pg_usleep(CommitDelay); - XLogFlush(XactLastRecEnd); /* --- 1091,1096 ---- *** a/src/backend/access/transam/xlog.c --- b/src/backend/access/transam/xlog.c *************** *** 24,29 **** --- 24,30 ---- #include #include "access/clog.h" + #include "access/groupcommit.h" #include "access/multixact.h" #include "access/subtrans.h" #include "access/transam.h" *************** *** 2020,2033 **** UpdateMinRecoveryPoint(XLogRecPtr lsn, bool force) * Ensure that all XLOG data through the given position is flushed to disk. * * NOTE: this differs from XLogWrite mainly in that the WALWriteLock is not ! * already held, and we try to avoid acquiring it if possible. */ void XLogFlush(XLogRecPtr record) { - XLogRecPtr WriteRqstPtr; - XLogwrtRqst WriteRqst; - /* * During REDO, we are reading not writing WAL. Therefore, instead of * trying to flush the WAL, we should update minRecoveryPoint instead. We --- 2021,2036 ---- * Ensure that all XLOG data through the given position is flushed to disk. * * NOTE: this differs from XLogWrite mainly in that the WALWriteLock is not ! * already held, and we try to avoid acquiring it if possible. Moreover, ! * XLogFlush is the main and only entry point for group commit. ! * ! * We'll actually call XLogWrite from here in the event of not being able to ! * participate in a group commit, due to it a backend requesting the flushing of ! * an LSN that is very far from the flush pointer's current position. */ void XLogFlush(XLogRecPtr record) { /* * During REDO, we are reading not writing WAL. Therefore, instead of * trying to flush the WAL, we should update minRecoveryPoint instead. We *************** *** 2059,2079 **** XLogFlush(XLogRecPtr record) * Since fsync is usually a horribly expensive operation, we try to * piggyback as much data as we can on each fsync: if we see any more data * entered into the xlog buffer, we'll write and fsync that too, so that ! * the final value of LogwrtResult.Flush is as large as possible. This ! * gives us some chance of avoiding another fsync immediately after. */ - /* initialize to given target; may increase below */ - WriteRqstPtr = record; - /* read LogwrtResult and update local state */ { /* use volatile pointer to prevent code rearrangement */ volatile XLogCtlData *xlogctl = XLogCtl; SpinLockAcquire(&xlogctl->info_lck); ! if (XLByteLT(WriteRqstPtr, xlogctl->LogwrtRqst.Write)) ! WriteRqstPtr = xlogctl->LogwrtRqst.Write; LogwrtResult = xlogctl->LogwrtResult; SpinLockRelease(&xlogctl->info_lck); } --- 2062,2083 ---- * Since fsync is usually a horribly expensive operation, we try to * piggyback as much data as we can on each fsync: if we see any more data * entered into the xlog buffer, we'll write and fsync that too, so that ! * the final value of LogwrtResult. Flush is as large as is reasonable; in ! * the group commit case (i.e. ordinary multiuser backends that a group ! * commit would be useful for), that will be as many backends as are queued ! * up. */ /* read LogwrtResult and update local state */ { /* use volatile pointer to prevent code rearrangement */ volatile XLogCtlData *xlogctl = XLogCtl; SpinLockAcquire(&xlogctl->info_lck); ! if (XLByteLT(xlogctl->LogwrtRqst.Write, record)) ! xlogctl->LogwrtRqst.Write = record; ! if (XLByteLT(xlogctl->LogwrtRqst.Flush, record)) ! xlogctl->LogwrtRqst.Flush = record; LogwrtResult = xlogctl->LogwrtResult; SpinLockRelease(&xlogctl->info_lck); } *************** *** 2081,2116 **** XLogFlush(XLogRecPtr record) /* done already? */ if (!XLByteLE(record, LogwrtResult.Flush)) { ! /* now wait for the write lock */ ! LWLockAcquire(WALWriteLock, LW_EXCLUSIVE); ! LogwrtResult = XLogCtl->Write.LogwrtResult; ! if (!XLByteLE(record, LogwrtResult.Flush)) ! { ! /* try to write/flush later additions to XLOG as well */ ! if (LWLockConditionalAcquire(WALInsertLock, LW_EXCLUSIVE)) ! { ! XLogCtlInsert *Insert = &XLogCtl->Insert; ! uint32 freespace = INSERT_FREESPACE(Insert); ! if (freespace < SizeOfXLogRecord) /* buffer is full */ ! WriteRqstPtr = XLogCtl->xlblocks[Insert->curridx]; ! else ! { ! WriteRqstPtr = XLogCtl->xlblocks[Insert->curridx]; ! WriteRqstPtr.xrecoff -= freespace; ! } ! LWLockRelease(WALInsertLock); ! WriteRqst.Write = WriteRqstPtr; ! WriteRqst.Flush = WriteRqstPtr; ! } ! else ! { ! WriteRqst.Write = WriteRqstPtr; ! WriteRqst.Flush = record; ! } ! XLogWrite(WriteRqst, false, false); ! } ! LWLockRelease(WALWriteLock); } END_CRIT_SECTION(); --- 2085,2100 ---- /* done already? */ if (!XLByteLE(record, LogwrtResult.Flush)) { ! /* use volatile pointer to prevent code rearrangement */ ! volatile XLogCtlData *xlogctl = XLogCtl; ! /* group commit */ ! GroupCommitWaitForLSN(record); ! ! /* update our result */ ! SpinLockAcquire(&xlogctl->info_lck); ! LogwrtResult = xlogctl->LogwrtResult; ! SpinLockRelease(&xlogctl->info_lck); } END_CRIT_SECTION(); *************** *** 2144,2149 **** XLogFlush(XLogRecPtr record) --- 2128,2180 ---- } /* + * A lower-level version of XLogFlush(). This is used from group commit code + * to actually perform a flush, while XLogFlush() is the facade for the + * group commit code. Got it? + */ + XLogRecPtr + DoXLogFlush(XLogRecPtr record) + { + XLogRecPtr WriteRqstPtr; + XLogwrtRqst WriteRqst; + + WriteRqstPtr = record; + + /* read LogwrtResult and update local state */ + { + /* use volatile pointer to prevent code rearrangement */ + volatile XLogCtlData *xlogctl = XLogCtl; + + SpinLockAcquire(&xlogctl->info_lck); + if (XLByteLT(xlogctl->LogwrtRqst.Write, record)) + WriteRqstPtr = xlogctl->LogwrtRqst.Write; + LogwrtResult = xlogctl->LogwrtResult; + SpinLockRelease(&xlogctl->info_lck); + } + + /* done already? */ + if (!XLByteLE(record, LogwrtResult.Flush)) + { + /* now wait for the write lock */ + LWLockAcquire(WALWriteLock, LW_EXCLUSIVE); + + LogwrtResult = XLogCtl->Write.LogwrtResult; + if (!XLByteLE(record, LogwrtResult.Flush)) + { + WriteRqst.Write = WriteRqstPtr; + WriteRqst.Flush = WriteRqstPtr; + XLogWrite(WriteRqst, false, false); + } + LWLockRelease(WALWriteLock); + } + + /* Wake up everyone else that we flushed as a side-effect */ + GroupCommitReleaseWaiters(LogwrtResult.Flush); + + return LogwrtResult.Flush; + } + + /* * Flush xlog, but without specifying exactly where to flush to. * * We normally flush only completed blocks; but if there is nothing to do on *************** *** 2235,2240 **** XLogBackgroundFlush(void) --- 2266,2281 ---- LWLockRelease(WALWriteLock); END_CRIT_SECTION(); + + /* + * Wake up everyone else that we flushed as a side-effect. If there are + * active commits happening, the driver will get the WALWriteLock as + * soon as we release it, and if it no longer has any work to do it will + * wake up everyone. But if it does, ie. if our flush didn't cover the + * current driver, it's good that we wake up everyone that we did flush, + * or they will have to wait for another flush cycle. + */ + GroupCommitReleaseWaiters(LogwrtResult.Flush); } /* *** a/src/backend/replication/syncrep.c --- b/src/backend/replication/syncrep.c *************** *** 34,39 **** --- 34,45 ---- * take some time. Once caught up, the current highest priority standby * will release waiters from the queue. * + * As of Postgres 9.2, some of the functions that were previously private to + * this module are generalized and exposed as generic "batch commit" + * infrastructure, for use by the group commit feature, which similarly batches + * commits, though it does so merely as a local performance optimization for + * user backends. + * * Portions Copyright (c) 2010-2012, PostgreSQL Global Development Group * * IDENTIFICATION *************** *** 65,81 **** char *SyncRepStandbyNames; #define SyncRepRequested() \ (max_wal_senders > 0 && synchronous_commit > SYNCHRONOUS_COMMIT_LOCAL_FLUSH) static bool announce_next_takeover = true; - static void SyncRepQueueInsert(void); static void SyncRepCancelWait(void); static int SyncRepGetStandbyPriority(void); - #ifdef USE_ASSERT_CHECKING - static bool SyncRepQueueIsOrderedByLSN(void); - #endif - /* * =========================================================== * Synchronous Replication functions for normal user backends --- 71,86 ---- #define SyncRepRequested() \ (max_wal_senders > 0 && synchronous_commit > SYNCHRONOUS_COMMIT_LOCAL_FLUSH) + /* Convenience macros for SyncRep use of batch commit infrastructure */ + #define SyncRepQueueInsert() BatchCommitQueueInsert(&(WalSndCtl->SyncRepQueue)) + #define SyncRepQueueIsOrderedByLSN() BatchCommitQueueIsOrderedByLSN(&(WalSndCtl->SyncRepQueue)) + static bool announce_next_takeover = true; static void SyncRepCancelWait(void); static int SyncRepGetStandbyPriority(void); /* * =========================================================== * Synchronous Replication functions for normal user backends *************** *** 105,112 **** SyncRepWaitForLSN(XLogRecPtr XactCommitLSN) if (!SyncRepRequested() || !SyncStandbysDefined()) return; ! Assert(SHMQueueIsDetached(&(MyProc->syncRepLinks))); Assert(WalSndCtl != NULL); LWLockAcquire(SyncRepLock, LW_EXCLUSIVE); Assert(MyProc->syncRepState == SYNC_REP_NOT_WAITING); --- 110,118 ---- if (!SyncRepRequested() || !SyncStandbysDefined()) return; ! Assert(SHMQueueIsDetached(&(MyProc->waitLSNLinks))); Assert(WalSndCtl != NULL); + Assert(!LWLockHeldByMe(GroupCommitLock)); LWLockAcquire(SyncRepLock, LW_EXCLUSIVE); Assert(MyProc->syncRepState == SYNC_REP_NOT_WAITING); *************** *** 253,259 **** SyncRepWaitForLSN(XLogRecPtr XactCommitLSN) * holding SyncRepLock, because any walsenders will ignore us anyway when * we're not on the queue. */ ! Assert(SHMQueueIsDetached(&(MyProc->syncRepLinks))); MyProc->syncRepState = SYNC_REP_NOT_WAITING; MyProc->waitLSN.xlogid = 0; MyProc->waitLSN.xrecoff = 0; --- 259,265 ---- * holding SyncRepLock, because any walsenders will ignore us anyway when * we're not on the queue. */ ! Assert(SHMQueueIsDetached(&(MyProc->waitLSNLinks))); MyProc->syncRepState = SYNC_REP_NOT_WAITING; MyProc->waitLSN.xlogid = 0; MyProc->waitLSN.xrecoff = 0; *************** *** 267,285 **** SyncRepWaitForLSN(XLogRecPtr XactCommitLSN) } /* ! * Insert MyProc into SyncRepQueue, maintaining sorted invariant. * * Usually we will go at tail of queue, though it's possible that we arrive * here out of order, so start at tail and work back to insertion point. */ ! static void ! SyncRepQueueInsert(void) { PGPROC *proc; ! proc = (PGPROC *) SHMQueuePrev(&(WalSndCtl->SyncRepQueue), ! &(WalSndCtl->SyncRepQueue), ! offsetof(PGPROC, syncRepLinks)); while (proc) { --- 273,293 ---- } /* ! * Insert MyProc into a queue (usually SyncRepQueue), maintaining sorted ! * invariant. * * Usually we will go at tail of queue, though it's possible that we arrive * here out of order, so start at tail and work back to insertion point. */ ! void ! BatchCommitQueueInsert(SHM_QUEUE *waitQueue) { PGPROC *proc; ! ! proc = (PGPROC *) SHMQueuePrev(waitQueue, ! waitQueue, ! offsetof(PGPROC, waitLSNLinks)); while (proc) { *************** *** 290,304 **** SyncRepQueueInsert(void) if (XLByteLT(proc->waitLSN, MyProc->waitLSN)) break; ! proc = (PGPROC *) SHMQueuePrev(&(WalSndCtl->SyncRepQueue), ! &(proc->syncRepLinks), ! offsetof(PGPROC, syncRepLinks)); } if (proc) ! SHMQueueInsertAfter(&(proc->syncRepLinks), &(MyProc->syncRepLinks)); else ! SHMQueueInsertAfter(&(WalSndCtl->SyncRepQueue), &(MyProc->syncRepLinks)); } /* --- 298,312 ---- if (XLByteLT(proc->waitLSN, MyProc->waitLSN)) break; ! proc = (PGPROC *) SHMQueuePrev(waitQueue, ! &(proc->waitLSNLinks), ! offsetof(PGPROC, waitLSNLinks)); } if (proc) ! SHMQueueInsertAfter(&(proc->waitLSNLinks), &(MyProc->waitLSNLinks)); else ! SHMQueueInsertAfter(waitQueue, &(MyProc->waitLSNLinks)); } /* *************** *** 307,315 **** SyncRepQueueInsert(void) static void SyncRepCancelWait(void) { LWLockAcquire(SyncRepLock, LW_EXCLUSIVE); ! if (!SHMQueueIsDetached(&(MyProc->syncRepLinks))) ! SHMQueueDelete(&(MyProc->syncRepLinks)); MyProc->syncRepState = SYNC_REP_NOT_WAITING; LWLockRelease(SyncRepLock); } --- 315,324 ---- static void SyncRepCancelWait(void) { + Assert(!LWLockHeldByMe(GroupCommitLock)); LWLockAcquire(SyncRepLock, LW_EXCLUSIVE); ! if (!SHMQueueIsDetached(&(MyProc->waitLSNLinks))) ! SHMQueueDelete(&(MyProc->waitLSNLinks)); MyProc->syncRepState = SYNC_REP_NOT_WAITING; LWLockRelease(SyncRepLock); } *************** *** 317,326 **** SyncRepCancelWait(void) void SyncRepCleanupAtProcExit(void) { ! if (!SHMQueueIsDetached(&(MyProc->syncRepLinks))) { LWLockAcquire(SyncRepLock, LW_EXCLUSIVE); ! SHMQueueDelete(&(MyProc->syncRepLinks)); LWLockRelease(SyncRepLock); } } --- 326,335 ---- void SyncRepCleanupAtProcExit(void) { ! if (!SHMQueueIsDetached(&(MyProc->waitLSNLinks))) { LWLockAcquire(SyncRepLock, LW_EXCLUSIVE); ! SHMQueueDelete(&(MyProc->waitLSNLinks)); LWLockRelease(SyncRepLock); } } *************** *** 388,393 **** SyncRepReleaseWaiters(void) --- 397,403 ---- * change pg_stat_get_wal_senders(). */ LWLockAcquire(SyncRepLock, LW_EXCLUSIVE); + Assert(!LWLockHeldByMe(GroupCommitLock)); for (i = 0; i < max_wal_senders; i++) { *************** *** 522,531 **** SyncRepWakeQueue(bool all) int numprocs = 0; Assert(SyncRepQueueIsOrderedByLSN()); proc = (PGPROC *) SHMQueueNext(&(WalSndCtl->SyncRepQueue), &(WalSndCtl->SyncRepQueue), ! offsetof(PGPROC, syncRepLinks)); while (proc) { --- 532,542 ---- int numprocs = 0; Assert(SyncRepQueueIsOrderedByLSN()); + Assert(!LWLockHeldByMe(GroupCommitLock)); proc = (PGPROC *) SHMQueueNext(&(WalSndCtl->SyncRepQueue), &(WalSndCtl->SyncRepQueue), ! offsetof(PGPROC, waitLSNLinks)); while (proc) { *************** *** 541,548 **** SyncRepWakeQueue(bool all) */ thisproc = proc; proc = (PGPROC *) SHMQueueNext(&(WalSndCtl->SyncRepQueue), ! &(proc->syncRepLinks), ! offsetof(PGPROC, syncRepLinks)); /* * Set state to complete; see SyncRepWaitForLSN() for discussion of --- 552,559 ---- */ thisproc = proc; proc = (PGPROC *) SHMQueueNext(&(WalSndCtl->SyncRepQueue), ! &(proc->waitLSNLinks), ! offsetof(PGPROC, waitLSNLinks)); /* * Set state to complete; see SyncRepWaitForLSN() for discussion of *************** *** 553,559 **** SyncRepWakeQueue(bool all) /* * Remove thisproc from queue. */ ! SHMQueueDelete(&(thisproc->syncRepLinks)); /* * Wake only when we have set state and removed from queue. --- 564,570 ---- /* * Remove thisproc from queue. */ ! SHMQueueDelete(&(thisproc->waitLSNLinks)); /* * Wake only when we have set state and removed from queue. *************** *** 604,611 **** SyncRepUpdateSyncStandbysDefined(void) } #ifdef USE_ASSERT_CHECKING ! static bool ! SyncRepQueueIsOrderedByLSN(void) { PGPROC *proc = NULL; XLogRecPtr lastLSN; --- 615,622 ---- } #ifdef USE_ASSERT_CHECKING ! bool ! BatchCommitQueueIsOrderedByLSN(SHM_QUEUE *waitQueue) { PGPROC *proc = NULL; XLogRecPtr lastLSN; *************** *** 613,621 **** SyncRepQueueIsOrderedByLSN(void) lastLSN.xlogid = 0; lastLSN.xrecoff = 0; ! proc = (PGPROC *) SHMQueueNext(&(WalSndCtl->SyncRepQueue), ! &(WalSndCtl->SyncRepQueue), ! offsetof(PGPROC, syncRepLinks)); while (proc) { --- 624,632 ---- lastLSN.xlogid = 0; lastLSN.xrecoff = 0; ! proc = (PGPROC *) SHMQueueNext(waitQueue, ! waitQueue, ! offsetof(PGPROC, waitLSNLinks)); while (proc) { *************** *** 628,636 **** SyncRepQueueIsOrderedByLSN(void) lastLSN = proc->waitLSN; ! proc = (PGPROC *) SHMQueueNext(&(WalSndCtl->SyncRepQueue), ! &(proc->syncRepLinks), ! offsetof(PGPROC, syncRepLinks)); } return true; --- 639,647 ---- lastLSN = proc->waitLSN; ! proc = (PGPROC *) SHMQueueNext(waitQueue, ! &(proc->waitLSNLinks), ! offsetof(PGPROC, waitLSNLinks)); } return true; *** a/src/backend/storage/ipc/ipci.c --- b/src/backend/storage/ipc/ipci.c *************** *** 126,131 **** CreateSharedMemoryAndSemaphores(bool makePrivate, int port) --- 126,132 ---- size = add_size(size, BTreeShmemSize()); size = add_size(size, SyncScanShmemSize()); size = add_size(size, AsyncShmemSize()); + size = add_size(size, GroupCommitShmemSize()); #ifdef EXEC_BACKEND size = add_size(size, ShmemBackendArraySize()); #endif *************** *** 190,195 **** CreateSharedMemoryAndSemaphores(bool makePrivate, int port) --- 191,197 ---- * Set up xlog, clog, and buffers */ XLOGShmemInit(); + GroupCommitShmemInit(); CLOGShmemInit(); SUBTRANSShmemInit(); MultiXactShmemInit(); *** a/src/backend/storage/lmgr/proc.c --- b/src/backend/storage/lmgr/proc.c *************** *** 35,40 **** --- 35,41 ---- #include #include + #include "access/groupcommit.h" #include "access/transam.h" #include "access/twophase.h" #include "access/xact.h" *************** *** 377,387 **** InitProcess(void) #endif MyProc->recoveryConflictPending = false; ! /* Initialize fields for sync rep */ MyProc->waitLSN.xlogid = 0; MyProc->waitLSN.xrecoff = 0; MyProc->syncRepState = SYNC_REP_NOT_WAITING; ! SHMQueueElemInit(&(MyProc->syncRepLinks)); /* * Acquire ownership of the PGPROC's latch, so that we can use WaitLatch. --- 378,389 ---- #endif MyProc->recoveryConflictPending = false; ! /* Initialize fields for waiting on WAL LSNs */ MyProc->waitLSN.xlogid = 0; MyProc->waitLSN.xrecoff = 0; MyProc->syncRepState = SYNC_REP_NOT_WAITING; ! MyProc->grpCommitState = GROUP_COMMIT_NOT_WAITING; ! SHMQueueElemInit(&(MyProc->waitLSNLinks)); /* * Acquire ownership of the PGPROC's latch, so that we can use WaitLatch. *************** *** 739,746 **** ProcKill(int code, Datum arg) Assert(MyProc != NULL); ! /* Make sure we're out of the sync rep lists */ SyncRepCleanupAtProcExit(); #ifdef USE_ASSERT_CHECKING if (assert_enabled) --- 741,749 ---- Assert(MyProc != NULL); ! /* Make sure we're out of the wait for LSN lists */ SyncRepCleanupAtProcExit(); + GroupCommitCleanupAtProcExit(); #ifdef USE_ASSERT_CHECKING if (assert_enabled) *************** *** 776,781 **** ProcKill(int code, Datum arg) --- 779,786 ---- MyProc->links.next = (SHM_QUEUE *) procglobal->freeProcs; procglobal->freeProcs = MyProc; } + /* Invalidate group commit state */ + MyProc->grpCommitState = GROUP_COMMIT_NOT_WAITING; /* PGPROC struct isn't mine anymore */ MyProc = NULL; *** a/src/backend/utils/misc/guc.c --- b/src/backend/utils/misc/guc.c *************** *** 125,132 **** /* XXX these should appear in other modules' header files */ extern bool Log_disconnections; - extern int CommitDelay; - extern int CommitSiblings; extern char *default_tablespace; extern char *temp_tablespaces; extern bool synchronize_seqscans; --- 125,130 ---- *************** *** 2019,2046 **** static struct config_int ConfigureNamesInt[] = }, { - {"commit_delay", PGC_USERSET, WAL_SETTINGS, - gettext_noop("Sets the delay in microseconds between transaction commit and " - "flushing WAL to disk."), - NULL - }, - &CommitDelay, - 0, 0, 100000, - NULL, NULL, NULL - }, - - { - {"commit_siblings", PGC_USERSET, WAL_SETTINGS, - gettext_noop("Sets the minimum concurrent open transactions before performing " - "commit_delay."), - NULL - }, - &CommitSiblings, - 5, 0, 1000, - NULL, NULL, NULL - }, - - { {"extra_float_digits", PGC_USERSET, CLIENT_CONN_LOCALE, gettext_noop("Sets the number of digits displayed for floating-point values."), gettext_noop("This affects real, double precision, and geometric data types. " --- 2017,2022 ---- *** a/src/backend/utils/misc/postgresql.conf.sample --- b/src/backend/utils/misc/postgresql.conf.sample *************** *** 171,179 **** # (change requires restart) #wal_writer_delay = 200ms # 1-10000 milliseconds - #commit_delay = 0 # range 0-100000, in microseconds - #commit_siblings = 5 # range 1-1000 - # - Checkpoints - #checkpoint_segments = 3 # in logfile segments, min 1, 16MB each --- 171,176 ---- *** /dev/null --- b/src/include/access/groupcommit.h *************** *** 0 **** --- 1,42 ---- + /*------------------------------------------------------------------------- + * + * groupcommit.h + * Group commit interface + * + * Exports from access/transam/groupcommit.c. + * + * Copyright (c) 2010-2011, PostgreSQL Global Development Group + * + * IDENTIFICATION + * src/include/access/groupcommit.h + * + *------------------------------------------------------------------------- + */ + #ifndef _GROUPCOMMIT_H + #define _GROUPCOMMIT_H + + #include "access/xlogdefs.h" + #include "replication/syncrep.h" + + typedef enum GroupCommitState + { + GROUP_COMMIT_NOT_WAITING, + GROUP_COMMIT_WAITING, + GROUP_COMMIT_WAIT_COMPLETE + } GroupCommitState; + + extern PGDLLIMPORT bool UseGroupCommit; + + /* function prototypes */ + extern Size GroupCommitShmemSize(void); + extern void GroupCommitShmemInit(void); + + /* called by user backends */ + extern void GroupCommitWaitForLSN(XLogRecPtr XactCommitLSN); + /* called at backend exit */ + extern void GroupCommitCleanupAtProcExit(void); + + /* called by the driver */ + extern void GroupCommitReleaseWaiters(XLogRecPtr flushLSN); + + #endif /* _GROUPCOMMIT_H */ *** a/src/include/access/xlog.h --- b/src/include/access/xlog.h *************** *** 265,270 **** extern CheckpointStatsData CheckpointStats; --- 265,271 ---- extern XLogRecPtr XLogInsert(RmgrId rmid, uint8 info, XLogRecData *rdata); extern void XLogFlush(XLogRecPtr RecPtr); + extern XLogRecPtr DoXLogFlush(XLogRecPtr RecPtr); extern void XLogBackgroundFlush(void); extern bool XLogNeedsFlush(XLogRecPtr RecPtr); extern int XLogFileInit(uint32 log, uint32 seg, *** a/src/include/replication/syncrep.h --- b/src/include/replication/syncrep.h *************** *** 14,19 **** --- 14,20 ---- #define _SYNCREP_H #include "utils/guc.h" + #include "storage/shmem.h" /* syncRepState */ #define SYNC_REP_NOT_WAITING 0 *************** *** 41,44 **** extern int SyncRepWakeQueue(bool all); --- 42,52 ---- extern bool check_synchronous_standby_names(char **newval, void **extra, GucSource source); + /* Batch commit infrastructure, used by sync rep and group commit */ + extern void BatchCommitQueueInsert(SHM_QUEUE *waitQueue); + + #ifdef USE_ASSERT_CHECKING + extern bool BatchCommitQueueIsOrderedByLSN(SHM_QUEUE *waitQueue); + #endif + #endif /* _SYNCREP_H */ *** a/src/include/storage/latch.h --- b/src/include/storage/latch.h *************** *** 25,31 **** * and must be initialized at postmaster startup by InitSharedLatch. Before * a shared latch can be waited on, it must be associated with a process * with OwnLatch. Only the process owning the latch can wait on it, but any ! * process can set it. * * There are three basic operations on a latch: * --- 25,36 ---- * and must be initialized at postmaster startup by InitSharedLatch. Before * a shared latch can be waited on, it must be associated with a process * with OwnLatch. Only the process owning the latch can wait on it, but any ! * process can set it. Note that the use of the process latch (which is a field ! * in PGPROC) is generally preferred to using an ad-hoc shared latch, as generic ! * signal handlers will call SetLatch on the process latch. Signals have ! * the potential to invalidate the Latch timeout on certain platforms, resulting ! * in a denial-of-service, so it is important to verify that all signal handlers ! * within the process call SetLatch(). * * There are three basic operations on a latch: * *** a/src/include/storage/lwlock.h --- b/src/include/storage/lwlock.h *************** *** 79,84 **** typedef enum LWLockId --- 79,85 ---- SerializablePredicateLockListLock, OldSerXidLock, SyncRepLock, + GroupCommitLock, /* Individual lock IDs end here */ FirstBufMappingLock, FirstLockMgrLock = FirstBufMappingLock + NUM_BUFFER_PARTITIONS, *** a/src/include/storage/proc.h --- b/src/include/storage/proc.h *************** *** 119,126 **** struct PGPROC * syncRepLinks used only while holding SyncRepLock. */ XLogRecPtr waitLSN; /* waiting for this LSN or higher */ int syncRepState; /* wait state for sync rep */ ! SHM_QUEUE syncRepLinks; /* list link if process is in syncrep queue */ /* * All PROCLOCK objects for locks held or awaited by this backend are --- 119,127 ---- * syncRepLinks used only while holding SyncRepLock. */ XLogRecPtr waitLSN; /* waiting for this LSN or higher */ + int grpCommitState; /* wait state for group commit */ int syncRepState; /* wait state for sync rep */ ! SHM_QUEUE waitLSNLinks; /* list link if process is in WAL queue */ /* * All PROCLOCK objects for locks held or awaited by this backend are