commit_ts.c File Reference

#include "postgres.h"
#include "access/commit_ts.h"
#include "access/htup_details.h"
#include "access/slru.h"
#include "access/transam.h"
#include "access/xloginsert.h"
#include "access/xlogutils.h"
#include "funcapi.h"
#include "miscadmin.h"
#include "storage/shmem.h"
#include "utils/fmgrprotos.h"
#include "utils/guc_hooks.h"
#include "utils/timestamp.h"

Include dependency graph for commit_ts.c:

Go to the source code of this file.

Data Structures
struct	CommitTimestampEntry
struct	CommitTimestampShared

Macros
#define	SizeOfCommitTimestampEntry
#define	COMMIT_TS_XACTS_PER_PAGE    (BLCKSZ / SizeOfCommitTimestampEntry)
#define	TransactionIdToCTsEntry(xid)    ((xid) % (TransactionId) COMMIT_TS_XACTS_PER_PAGE)
#define	CommitTsCtl   (&CommitTsCtlData)

Typedefs
typedef struct CommitTimestampEntry	CommitTimestampEntry
typedef struct CommitTimestampShared	CommitTimestampShared

Functions
static int64	TransactionIdToCTsPage (TransactionId xid)
static void	SetXidCommitTsInPage (TransactionId xid, int nsubxids, TransactionId *subxids, TimestampTz ts, ReplOriginId nodeid, int64 pageno)
static void	TransactionIdSetCommitTs (TransactionId xid, TimestampTz ts, ReplOriginId nodeid, int slotno)
static void	error_commit_ts_disabled (void)
static bool	CommitTsPagePrecedes (int64 page1, int64 page2)
static int	commit_ts_errdetail_for_io_error (const void *opaque_data)
static void	ActivateCommitTs (void)
static void	DeactivateCommitTs (void)
static void	WriteTruncateXlogRec (int64 pageno, TransactionId oldestXid)
void	TransactionTreeSetCommitTsData (TransactionId xid, int nsubxids, TransactionId *subxids, TimestampTz timestamp, ReplOriginId nodeid)
bool	TransactionIdGetCommitTsData (TransactionId xid, TimestampTz *ts, ReplOriginId *nodeid)
TransactionId	GetLatestCommitTsData (TimestampTz *ts, ReplOriginId *nodeid)
Datum	pg_xact_commit_timestamp (PG_FUNCTION_ARGS)
Datum	pg_last_committed_xact (PG_FUNCTION_ARGS)
Datum	pg_xact_commit_timestamp_origin (PG_FUNCTION_ARGS)
static int	CommitTsShmemBuffers (void)
Size	CommitTsShmemSize (void)
void	CommitTsShmemInit (void)
bool	check_commit_ts_buffers (int *newval, void **extra, GucSource source)
void	BootStrapCommitTs (void)
void	StartupCommitTs (void)
void	CompleteCommitTsInitialization (void)
void	CommitTsParameterChange (bool newvalue, bool oldvalue)
void	CheckPointCommitTs (void)
void	ExtendCommitTs (TransactionId newestXact)
void	TruncateCommitTs (TransactionId oldestXact)
void	SetCommitTsLimit (TransactionId oldestXact, TransactionId newestXact)
void	AdvanceOldestCommitTsXid (TransactionId oldestXact)
void	commit_ts_redo (XLogReaderState *record)
int	committssyncfiletag (const FileTag *ftag, char *path)

Variables
static SlruCtlData	CommitTsCtlData
static CommitTimestampShared *	commitTsShared
bool	track_commit_timestamp

Macro Definition Documentation

COMMIT_TS_XACTS_PER_PAGE

#define COMMIT_TS_XACTS_PER_PAGE    (BLCKSZ / SizeOfCommitTimestampEntry)

Definition at line 63 of file commit_ts.c.

{
    return xid / (int64) COMMIT_TS_XACTS_PER_PAGE;
}
 
#define TransactionIdToCTsEntry(xid)    \
    ((xid) % (TransactionId) COMMIT_TS_XACTS_PER_PAGE)
 
/*
 * Link to shared-memory data structures for CommitTs control
 */
static SlruCtlData CommitTsCtlData;
 
#define CommitTsCtl (&CommitTsCtlData)
 
/*
 * We keep a cache of the last value set in shared memory.
 *
 * This is also good place to keep the activation status.  We keep this
 * separate from the GUC so that the standby can activate the module if the
 * primary has it active independently of the value of the GUC.
 *
 * This is protected by CommitTsLock.  In some places, we use commitTsActive
 * without acquiring the lock; where this happens, a comment explains the
 * rationale for it.
 */
typedef struct CommitTimestampShared
{
    TransactionId xidLastCommit;
    CommitTimestampEntry dataLastCommit;
    bool        commitTsActive;
} CommitTimestampShared;
 
static CommitTimestampShared *commitTsShared;
 
 
/* GUC variable */
bool        track_commit_timestamp;
 
static void SetXidCommitTsInPage(TransactionId xid, int nsubxids,
                                 TransactionId *subxids, TimestampTz ts,
                                 ReplOriginId nodeid, int64 pageno);
static void TransactionIdSetCommitTs(TransactionId xid, TimestampTz ts,
                                     ReplOriginId nodeid, int slotno);
static void error_commit_ts_disabled(void);
static bool CommitTsPagePrecedes(int64 page1, int64 page2);
static int  commit_ts_errdetail_for_io_error(const void *opaque_data);
static void ActivateCommitTs(void);
static void DeactivateCommitTs(void);
static void WriteTruncateXlogRec(int64 pageno, TransactionId oldestXid);
 
/*
 * TransactionTreeSetCommitTsData
 *
 * Record the final commit timestamp of transaction entries in the commit log
 * for a transaction and its subtransaction tree, as efficiently as possible.
 *
 * xid is the top level transaction id.
 *
 * subxids is an array of xids of length nsubxids, representing subtransactions
 * in the tree of xid. In various cases nsubxids may be zero.
 * The reason why tracking just the parent xid commit timestamp is not enough
 * is that the subtrans SLRU does not stay valid across crashes (it's not
 * permanent) so we need to keep the information about them here. If the
 * subtrans implementation changes in the future, we might want to revisit the
 * decision of storing timestamp info for each subxid.
 */
void
TransactionTreeSetCommitTsData(TransactionId xid, int nsubxids,
                               TransactionId *subxids, TimestampTz timestamp,
                               ReplOriginId nodeid)
{
    int         i;
    TransactionId headxid;
    TransactionId newestXact;
 
    /*
     * No-op if the module is not active.
     *
     * An unlocked read here is fine, because in a standby (the only place
     * where the flag can change in flight) this routine is only called by the
     * recovery process, which is also the only process which can change the
     * flag.
     */
    if (!commitTsShared->commitTsActive)
        return;
 
    /*
     * Figure out the latest Xid in this batch: either the last subxid if
     * there's any, otherwise the parent xid.
     */
    if (nsubxids > 0)
        newestXact = subxids[nsubxids - 1];
    else
        newestXact = xid;
 
    /*
     * We split the xids to set the timestamp to in groups belonging to the
     * same SLRU page; the first element in each such set is its head.  The
     * first group has the main XID as the head; subsequent sets use the first
     * subxid not on the previous page as head.  This way, we only have to
     * lock/modify each SLRU page once.
     */
    headxid = xid;
    i = 0;
    for (;;)
    {
        int64       pageno = TransactionIdToCTsPage(headxid);
        int         j;
 
        for (j = i; j < nsubxids; j++)
        {
            if (TransactionIdToCTsPage(subxids[j]) != pageno)
                break;
        }
        /* subxids[i..j] are on the same page as the head */
 
        SetXidCommitTsInPage(headxid, j - i, subxids + i, timestamp, nodeid,
                             pageno);
 
        /* if we wrote out all subxids, we're done. */
        if (j >= nsubxids)
            break;
 
        /*
         * Set the new head and skip over it, as well as over the subxids we
         * just wrote.
         */
        headxid = subxids[j];
        i = j + 1;
    }
 
    /* update the cached value in shared memory */
    LWLockAcquire(CommitTsLock, LW_EXCLUSIVE);
    commitTsShared->xidLastCommit = xid;
    commitTsShared->dataLastCommit.time = timestamp;
    commitTsShared->dataLastCommit.nodeid = nodeid;
 
    /* and move forwards our endpoint, if needed */
    if (TransactionIdPrecedes(TransamVariables->newestCommitTsXid, newestXact))
        TransamVariables->newestCommitTsXid = newestXact;
    LWLockRelease(CommitTsLock);
}
 
/*
 * Record the commit timestamp of transaction entries in the commit log for all
 * entries on a single page.  Atomic only on this page.
 */
static void
SetXidCommitTsInPage(TransactionId xid, int nsubxids,
                     TransactionId *subxids, TimestampTz ts,
                     ReplOriginId nodeid, int64 pageno)
{
    LWLock     *lock = SimpleLruGetBankLock(CommitTsCtl, pageno);
    int         slotno;
    int         i;
 
    LWLockAcquire(lock, LW_EXCLUSIVE);
 
    slotno = SimpleLruReadPage(CommitTsCtl, pageno, true, &xid);
 
    TransactionIdSetCommitTs(xid, ts, nodeid, slotno);
    for (i = 0; i < nsubxids; i++)
        TransactionIdSetCommitTs(subxids[i], ts, nodeid, slotno);
 
    CommitTsCtl->shared->page_dirty[slotno] = true;
 
    LWLockRelease(lock);
}
 
/*
 * Sets the commit timestamp of a single transaction.
 *
 * Caller must hold the correct SLRU bank lock, will be held at exit
 */
static void
TransactionIdSetCommitTs(TransactionId xid, TimestampTz ts,
                         ReplOriginId nodeid, int slotno)
{
    int         entryno = TransactionIdToCTsEntry(xid);
    CommitTimestampEntry entry;
 
    Assert(TransactionIdIsNormal(xid));
 
    entry.time = ts;
    entry.nodeid = nodeid;
 
    memcpy(CommitTsCtl->shared->page_buffer[slotno] +
           SizeOfCommitTimestampEntry * entryno,
           &entry, SizeOfCommitTimestampEntry);
}
 
/*
 * Interrogate the commit timestamp of a transaction.
 *
 * The return value indicates whether a commit timestamp record was found for
 * the given xid.  The timestamp value is returned in *ts (which may not be
 * null), and the origin node for the Xid is returned in *nodeid, if it's not
 * null.
 */
bool
TransactionIdGetCommitTsData(TransactionId xid, TimestampTz *ts,
                             ReplOriginId *nodeid)
{
    int64       pageno = TransactionIdToCTsPage(xid);
    int         entryno = TransactionIdToCTsEntry(xid);
    int         slotno;
    CommitTimestampEntry entry;
    TransactionId oldestCommitTsXid;
    TransactionId newestCommitTsXid;
 
    if (!TransactionIdIsValid(xid))
        ereport(ERROR,
                (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
                 errmsg("cannot retrieve commit timestamp for transaction %u", xid)));
    else if (!TransactionIdIsNormal(xid))
    {
        /* frozen and bootstrap xids are always committed far in the past */
        *ts = 0;
        if (nodeid)
            *nodeid = 0;
        return false;
    }
 
    LWLockAcquire(CommitTsLock, LW_SHARED);
 
    /* Error if module not enabled */
    if (!commitTsShared->commitTsActive)
        error_commit_ts_disabled();
 
    /*
     * If we're asked for the cached value, return that.  Otherwise, fall
     * through to read from SLRU.
     */
    if (commitTsShared->xidLastCommit == xid)
    {
        *ts = commitTsShared->dataLastCommit.time;
        if (nodeid)
            *nodeid = commitTsShared->dataLastCommit.nodeid;
 
        LWLockRelease(CommitTsLock);
        return *ts != 0;
    }
 
    oldestCommitTsXid = TransamVariables->oldestCommitTsXid;
    newestCommitTsXid = TransamVariables->newestCommitTsXid;
    /* neither is invalid, or both are */
    Assert(TransactionIdIsValid(oldestCommitTsXid) == TransactionIdIsValid(newestCommitTsXid));
    LWLockRelease(CommitTsLock);
 
    /*
     * Return empty if the requested value is outside our valid range.
     */
    if (!TransactionIdIsValid(oldestCommitTsXid) ||
        TransactionIdPrecedes(xid, oldestCommitTsXid) ||
        TransactionIdPrecedes(newestCommitTsXid, xid))
    {
        *ts = 0;
        if (nodeid)
            *nodeid = InvalidReplOriginId;
        return false;
    }
 
    /* lock is acquired by SimpleLruReadPage_ReadOnly */
    slotno = SimpleLruReadPage_ReadOnly(CommitTsCtl, pageno, &xid);
    memcpy(&entry,
           CommitTsCtl->shared->page_buffer[slotno] +
           SizeOfCommitTimestampEntry * entryno,
           SizeOfCommitTimestampEntry);
 
    *ts = entry.time;
    if (nodeid)
        *nodeid = entry.nodeid;
 
    LWLockRelease(SimpleLruGetBankLock(CommitTsCtl, pageno));
    return *ts != 0;
}
 
/*
 * Return the Xid of the latest committed transaction.  (As far as this module
 * is concerned, anyway; it's up to the caller to ensure the value is useful
 * for its purposes.)
 *
 * ts and nodeid are filled with the corresponding data; they can be passed
 * as NULL if not wanted.
 */
TransactionId
GetLatestCommitTsData(TimestampTz *ts, ReplOriginId *nodeid)
{
    TransactionId xid;
 
    LWLockAcquire(CommitTsLock, LW_SHARED);
 
    /* Error if module not enabled */
    if (!commitTsShared->commitTsActive)
        error_commit_ts_disabled();
 
    xid = commitTsShared->xidLastCommit;
    if (ts)
        *ts = commitTsShared->dataLastCommit.time;
    if (nodeid)
        *nodeid = commitTsShared->dataLastCommit.nodeid;
    LWLockRelease(CommitTsLock);
 
    return xid;
}
 
static void
error_commit_ts_disabled(void)
{
    ereport(ERROR,
            (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
             errmsg("could not get commit timestamp data"),
             RecoveryInProgress() ?
             errhint("Make sure the configuration parameter \"%s\" is set on the primary server.",
                     "track_commit_timestamp") :
             errhint("Make sure the configuration parameter \"%s\" is set.",
                     "track_commit_timestamp")));
}
 
/*
 * SQL-callable wrapper to obtain commit time of a transaction
 */
Datum
pg_xact_commit_timestamp(PG_FUNCTION_ARGS)
{
    TransactionId xid = PG_GETARG_TRANSACTIONID(0);
    TimestampTz ts;
    bool        found;
 
    found = TransactionIdGetCommitTsData(xid, &ts, NULL);
 
    if (!found)
        PG_RETURN_NULL();
 
    PG_RETURN_TIMESTAMPTZ(ts);
}
 
 
/*
 * pg_last_committed_xact
 *
 * SQL-callable wrapper to obtain some information about the latest
 * committed transaction: transaction ID, timestamp and replication
 * origin.
 */
Datum
pg_last_committed_xact(PG_FUNCTION_ARGS)
{
    TransactionId xid;
    ReplOriginId nodeid;
    TimestampTz ts;
    Datum       values[3];
    bool        nulls[3];
    TupleDesc   tupdesc;
    HeapTuple   htup;
 
    /* and construct a tuple with our data */
    xid = GetLatestCommitTsData(&ts, &nodeid);
 
    if (get_call_result_type(fcinfo, NULL, &tupdesc) != TYPEFUNC_COMPOSITE)
        elog(ERROR, "return type must be a row type");
 
    if (!TransactionIdIsNormal(xid))
    {
        memset(nulls, true, sizeof(nulls));
    }
    else
    {
        values[0] = TransactionIdGetDatum(xid);
        nulls[0] = false;
 
        values[1] = TimestampTzGetDatum(ts);
        nulls[1] = false;
 
        values[2] = ObjectIdGetDatum((Oid) nodeid);
        nulls[2] = false;
    }
 
    htup = heap_form_tuple(tupdesc, values, nulls);
 
    PG_RETURN_DATUM(HeapTupleGetDatum(htup));
}
 
/*
 * pg_xact_commit_timestamp_origin
 *
 * SQL-callable wrapper to obtain commit timestamp and replication origin
 * of a given transaction.
 */
Datum
pg_xact_commit_timestamp_origin(PG_FUNCTION_ARGS)
{
    TransactionId xid = PG_GETARG_TRANSACTIONID(0);
    ReplOriginId nodeid;
    TimestampTz ts;
    Datum       values[2];
    bool        nulls[2];
    TupleDesc   tupdesc;
    HeapTuple   htup;
    bool        found;
 
    found = TransactionIdGetCommitTsData(xid, &ts, &nodeid);
 
    if (get_call_result_type(fcinfo, NULL, &tupdesc) != TYPEFUNC_COMPOSITE)
        elog(ERROR, "return type must be a row type");
 
    if (!found)
    {
        memset(nulls, true, sizeof(nulls));
    }
    else
    {
        values[0] = TimestampTzGetDatum(ts);
        nulls[0] = false;
 
        values[1] = ObjectIdGetDatum((Oid) nodeid);
        nulls[1] = false;
    }
 
    htup = heap_form_tuple(tupdesc, values, nulls);
 
    PG_RETURN_DATUM(HeapTupleGetDatum(htup));
}
 
/*
 * Number of shared CommitTS buffers.
 *
 * If asked to autotune, use 2MB for every 1GB of shared buffers, up to 8MB.
 * Otherwise just cap the configured amount to be between 16 and the maximum
 * allowed.
 */
static int
CommitTsShmemBuffers(void)
{
    /* auto-tune based on shared buffers */
    if (commit_timestamp_buffers == 0)
        return SimpleLruAutotuneBuffers(512, 1024);
 
    return Min(Max(16, commit_timestamp_buffers), SLRU_MAX_ALLOWED_BUFFERS);
}
 
/*
 * Shared memory sizing for CommitTs
 */
Size
CommitTsShmemSize(void)
{
    return SimpleLruShmemSize(CommitTsShmemBuffers(), 0) +
        sizeof(CommitTimestampShared);
}
 
/*
 * Initialize CommitTs at system startup (postmaster start or standalone
 * backend)
 */
void
CommitTsShmemInit(void)
{
    bool        found;
 
    /* If auto-tuning is requested, now is the time to do it */
    if (commit_timestamp_buffers == 0)
    {
        char        buf[32];
 
        snprintf(buf, sizeof(buf), "%d", CommitTsShmemBuffers());
        SetConfigOption("commit_timestamp_buffers", buf, PGC_POSTMASTER,
                        PGC_S_DYNAMIC_DEFAULT);
 
        /*
         * We prefer to report this value's source as PGC_S_DYNAMIC_DEFAULT.
         * However, if the DBA explicitly set commit_timestamp_buffers = 0 in
         * the config file, then PGC_S_DYNAMIC_DEFAULT will fail to override
         * that and we must force the matter with PGC_S_OVERRIDE.
         */
        if (commit_timestamp_buffers == 0)  /* failed to apply it? */
            SetConfigOption("commit_timestamp_buffers", buf, PGC_POSTMASTER,
                            PGC_S_OVERRIDE);
    }
    Assert(commit_timestamp_buffers != 0);
 
    CommitTsCtl->PagePrecedes = CommitTsPagePrecedes;
    CommitTsCtl->errdetail_for_io_error = commit_ts_errdetail_for_io_error;
    SimpleLruInit(CommitTsCtl, "commit_timestamp", CommitTsShmemBuffers(), 0,
                  "pg_commit_ts", LWTRANCHE_COMMITTS_BUFFER,
                  LWTRANCHE_COMMITTS_SLRU,
                  SYNC_HANDLER_COMMIT_TS,
                  false);
    SlruPagePrecedesUnitTests(CommitTsCtl, COMMIT_TS_XACTS_PER_PAGE);
 
    commitTsShared = ShmemInitStruct("CommitTs shared",
                                     sizeof(CommitTimestampShared),
                                     &found);
 
    if (!IsUnderPostmaster)
    {
        Assert(!found);
 
        commitTsShared->xidLastCommit = InvalidTransactionId;
        TIMESTAMP_NOBEGIN(commitTsShared->dataLastCommit.time);
        commitTsShared->dataLastCommit.nodeid = InvalidReplOriginId;
        commitTsShared->commitTsActive = false;
    }
    else
        Assert(found);
}
 
/*
 * GUC check_hook for commit_timestamp_buffers
 */
bool
check_commit_ts_buffers(int *newval, void **extra, GucSource source)
{
    return check_slru_buffers("commit_timestamp_buffers", newval);
}
 
/*
 * This function must be called ONCE on system install.
 *
 * (The CommitTs directory is assumed to have been created by initdb, and
 * CommitTsShmemInit must have been called already.)
 */
void
BootStrapCommitTs(void)
{
    /*
     * Nothing to do here at present, unlike most other SLRU modules; segments
     * are created when the server is started with this module enabled. See
     * ActivateCommitTs.
     */
}
 
/*
 * This must be called ONCE during postmaster or standalone-backend startup,
 * after StartupXLOG has initialized TransamVariables->nextXid.
 */
void
StartupCommitTs(void)
{
    ActivateCommitTs();
}
 
/*
 * This must be called ONCE during postmaster or standalone-backend startup,
 * after recovery has finished.
 */
void
CompleteCommitTsInitialization(void)
{
    /*
     * If the feature is not enabled, turn it off for good.  This also removes
     * any leftover data.
     *
     * Conversely, we activate the module if the feature is enabled.  This is
     * necessary for primary and standby as the activation depends on the
     * control file contents at the beginning of recovery or when a
     * XLOG_PARAMETER_CHANGE is replayed.
     */
    if (!track_commit_timestamp)
        DeactivateCommitTs();
    else
        ActivateCommitTs();
}
 
/*
 * Activate or deactivate CommitTs' upon reception of a XLOG_PARAMETER_CHANGE
 * XLog record during recovery.
 */
void
CommitTsParameterChange(bool newvalue, bool oldvalue)
{
    /*
     * If the commit_ts module is disabled in this server and we get word from
     * the primary server that it is enabled there, activate it so that we can
     * replay future WAL records involving it; also mark it as active on
     * pg_control.  If the old value was already set, we already did this, so
     * don't do anything.
     *
     * If the module is disabled in the primary, disable it here too, unless
     * the module is enabled locally.
     *
     * Note this only runs in the recovery process, so an unlocked read is
     * fine.
     */
    if (newvalue)
    {
        if (!commitTsShared->commitTsActive)
            ActivateCommitTs();
    }
    else if (commitTsShared->commitTsActive)
        DeactivateCommitTs();
}
 
/*
 * Activate this module whenever necessary.
 *      This must happen during postmaster or standalone-backend startup,
 *      or during WAL replay anytime the track_commit_timestamp setting is
 *      changed in the primary.
 *
 * The reason why this SLRU needs separate activation/deactivation functions is
 * that it can be enabled/disabled during start and the activation/deactivation
 * on the primary is propagated to the standby via replay. Other SLRUs don't
 * have this property and they can be just initialized during normal startup.
 *
 * This is in charge of creating the currently active segment, if it's not
 * already there.  The reason for this is that the server might have been
 * running with this module disabled for a while and thus might have skipped
 * the normal creation point.
 */
static void
ActivateCommitTs(void)
{
    TransactionId xid;
    int64       pageno;
 
    /*
     * During bootstrap, we should not register commit timestamps so skip the
     * activation in this case.
     */
    if (IsBootstrapProcessingMode())
        return;
 
    /* If we've done this already, there's nothing to do */
    LWLockAcquire(CommitTsLock, LW_EXCLUSIVE);
    if (commitTsShared->commitTsActive)
    {
        LWLockRelease(CommitTsLock);
        return;
    }
    LWLockRelease(CommitTsLock);
 
    xid = XidFromFullTransactionId(TransamVariables->nextXid);
    pageno = TransactionIdToCTsPage(xid);
 
    /*
     * Re-Initialize our idea of the latest page number.
     */
    pg_atomic_write_u64(&CommitTsCtl->shared->latest_page_number, pageno);
 
    /*
     * If CommitTs is enabled, but it wasn't in the previous server run, we
     * need to set the oldest and newest values to the next Xid; that way, we
     * will not try to read data that might not have been set.
     *
     * XXX does this have a problem if a server is started with commitTs
     * enabled, then started with commitTs disabled, then restarted with it
     * enabled again?  It doesn't look like it does, because there should be a
     * checkpoint that sets the value to InvalidTransactionId at end of
     * recovery; and so any chance of injecting new transactions without
     * CommitTs values would occur after the oldestCommitTsXid has been set to
     * Invalid temporarily.
     */
    LWLockAcquire(CommitTsLock, LW_EXCLUSIVE);
    if (TransamVariables->oldestCommitTsXid == InvalidTransactionId)
    {
        TransamVariables->oldestCommitTsXid =
            TransamVariables->newestCommitTsXid = ReadNextTransactionId();
    }
    LWLockRelease(CommitTsLock);
 
    /* Create the current segment file, if necessary */
    if (!SimpleLruDoesPhysicalPageExist(CommitTsCtl, pageno))
        SimpleLruZeroAndWritePage(CommitTsCtl, pageno);
 
    /* Change the activation status in shared memory. */
    LWLockAcquire(CommitTsLock, LW_EXCLUSIVE);
    commitTsShared->commitTsActive = true;
    LWLockRelease(CommitTsLock);
}
 
/*
 * Deactivate this module.
 *
 * This must be called when the track_commit_timestamp parameter is turned off.
 * This happens during postmaster or standalone-backend startup, or during WAL
 * replay.
 *
 * Resets CommitTs into invalid state to make sure we don't hand back
 * possibly-invalid data; also removes segments of old data.
 */
static void
DeactivateCommitTs(void)
{
    /*
     * Cleanup the status in the shared memory.
     *
     * We reset everything in the commitTsShared record to prevent user from
     * getting confusing data about last committed transaction on the standby
     * when the module was activated repeatedly on the primary.
     */
    LWLockAcquire(CommitTsLock, LW_EXCLUSIVE);
 
    commitTsShared->commitTsActive = false;
    commitTsShared->xidLastCommit = InvalidTransactionId;
    TIMESTAMP_NOBEGIN(commitTsShared->dataLastCommit.time);
    commitTsShared->dataLastCommit.nodeid = InvalidReplOriginId;
 
    TransamVariables->oldestCommitTsXid = InvalidTransactionId;
    TransamVariables->newestCommitTsXid = InvalidTransactionId;
 
    /*
     * Remove *all* files.  This is necessary so that there are no leftover
     * files; in the case where this feature is later enabled after running
     * with it disabled for some time there may be a gap in the file sequence.
     * (We can probably tolerate out-of-sequence files, as they are going to
     * be overwritten anyway when we wrap around, but it seems better to be
     * tidy.)
     *
     * Note that we do this with CommitTsLock acquired in exclusive mode. This
     * is very heavy-handed, but since this routine can only be called in the
     * replica and should happen very rarely, we don't worry too much about
     * it.  Note also that no process should be consulting this SLRU if we
     * have just deactivated it.
     */
    (void) SlruScanDirectory(CommitTsCtl, SlruScanDirCbDeleteAll, NULL);
 
    LWLockRelease(CommitTsLock);
}
 
/*
 * Perform a checkpoint --- either during shutdown, or on-the-fly
 */
void
CheckPointCommitTs(void)
{
    /*
     * Write dirty CommitTs pages to disk.  This may result in sync requests
     * queued for later handling by ProcessSyncRequests(), as part of the
     * checkpoint.
     */
    SimpleLruWriteAll(CommitTsCtl, true);
}
 
/*
 * Make sure that CommitTs has room for a newly-allocated XID.
 *
 * NB: this is called while holding XidGenLock.  We want it to be very fast
 * most of the time; even when it's not so fast, no actual I/O need happen
 * unless we're forced to write out a dirty CommitTs or xlog page to make room
 * in shared memory.
 *
 * NB: the current implementation relies on track_commit_timestamp being
 * PGC_POSTMASTER.
 */
void
ExtendCommitTs(TransactionId newestXact)
{
    int64       pageno;
    LWLock     *lock;
 
    /*
     * Nothing to do if module not enabled.  Note we do an unlocked read of
     * the flag here, which is okay because this routine is only called from
     * GetNewTransactionId, which is never called in a standby.
     */
    Assert(!InRecovery);
    if (!commitTsShared->commitTsActive)
        return;
 
    /*
     * No work except at first XID of a page.  But beware: just after
     * wraparound, the first XID of page zero is FirstNormalTransactionId.
     */
    if (TransactionIdToCTsEntry(newestXact) != 0 &&
        !TransactionIdEquals(newestXact, FirstNormalTransactionId))
        return;
 
    pageno = TransactionIdToCTsPage(newestXact);
 
    lock = SimpleLruGetBankLock(CommitTsCtl, pageno);
 
    LWLockAcquire(lock, LW_EXCLUSIVE);
 
    /* Zero the page ... */
    SimpleLruZeroPage(CommitTsCtl, pageno);
 
    /* and make a WAL entry about that, unless we're in REDO */
    if (!InRecovery)
        XLogSimpleInsertInt64(RM_COMMIT_TS_ID, COMMIT_TS_ZEROPAGE, pageno);
 
    LWLockRelease(lock);
}
 
/*
 * Remove all CommitTs segments before the one holding the passed
 * transaction ID.
 *
 * Note that we don't need to flush XLOG here.
 */
void
TruncateCommitTs(TransactionId oldestXact)
{
    int64       cutoffPage;
 
    /*
     * The cutoff point is the start of the segment containing oldestXact. We
     * pass the *page* containing oldestXact to SimpleLruTruncate.
     */
    cutoffPage = TransactionIdToCTsPage(oldestXact);
 
    /* Check to see if there's any files that could be removed */
    if (!SlruScanDirectory(CommitTsCtl, SlruScanDirCbReportPresence,
                           &cutoffPage))
        return;                 /* nothing to remove */
 
    /* Write XLOG record */
    WriteTruncateXlogRec(cutoffPage, oldestXact);
 
    /* Now we can remove the old CommitTs segment(s) */
    SimpleLruTruncate(CommitTsCtl, cutoffPage);
}
 
/*
 * Set the limit values between which commit TS can be consulted.
 */
void
SetCommitTsLimit(TransactionId oldestXact, TransactionId newestXact)
{
    /*
     * Be careful not to overwrite values that are either further into the
     * "future" or signal a disabled committs.
     */
    LWLockAcquire(CommitTsLock, LW_EXCLUSIVE);
    if (TransamVariables->oldestCommitTsXid != InvalidTransactionId)
    {
        if (TransactionIdPrecedes(TransamVariables->oldestCommitTsXid, oldestXact))
            TransamVariables->oldestCommitTsXid = oldestXact;
        if (TransactionIdPrecedes(newestXact, TransamVariables->newestCommitTsXid))
            TransamVariables->newestCommitTsXid = newestXact;
    }
    else
    {
        Assert(TransamVariables->newestCommitTsXid == InvalidTransactionId);
        TransamVariables->oldestCommitTsXid = oldestXact;
        TransamVariables->newestCommitTsXid = newestXact;
    }
    LWLockRelease(CommitTsLock);
}
 
/*
 * Move forwards the oldest commitTS value that can be consulted
 */
void
AdvanceOldestCommitTsXid(TransactionId oldestXact)
{
    LWLockAcquire(CommitTsLock, LW_EXCLUSIVE);
    if (TransamVariables->oldestCommitTsXid != InvalidTransactionId &&
        TransactionIdPrecedes(TransamVariables->oldestCommitTsXid, oldestXact))
        TransamVariables->oldestCommitTsXid = oldestXact;
    LWLockRelease(CommitTsLock);
}
 
 
/*
 * Decide whether a commitTS page number is "older" for truncation purposes.
 * Analogous to CLOGPagePrecedes().
 *
 * At default BLCKSZ, (1 << 31) % COMMIT_TS_XACTS_PER_PAGE == 128.  This
 * introduces differences compared to CLOG and the other SLRUs having (1 <<
 * 31) % per_page == 0.  This function never tests exactly
 * TransactionIdPrecedes(x-2^31, x).  When the system reaches xidStopLimit,
 * there are two possible counts of page boundaries between oldestXact and the
 * latest XID assigned, depending on whether oldestXact is within the first
 * 128 entries of its page.  Since this function doesn't know the location of
 * oldestXact within page2, it returns false for one page that actually is
 * expendable.  This is a wider (yet still negligible) version of the
 * truncation opportunity that CLOGPagePrecedes() cannot recognize.
 *
 * For the sake of a worked example, number entries with decimal values such
 * that page1==1 entries range from 1.0 to 1.999.  Let N+0.15 be the number of
 * pages that 2^31 entries will span (N is an integer).  If oldestXact=N+2.1,
 * then the final safe XID assignment leaves newestXact=1.95.  We keep page 2,
 * because entry=2.85 is the border that toggles whether entries precede the
 * last entry of the oldestXact page.  While page 2 is expendable at
 * oldestXact=N+2.1, it would be precious at oldestXact=N+2.9.
 */
static bool
CommitTsPagePrecedes(int64 page1, int64 page2)
{
    TransactionId xid1;
    TransactionId xid2;
 
    xid1 = ((TransactionId) page1) * COMMIT_TS_XACTS_PER_PAGE;
    xid1 += FirstNormalTransactionId + 1;
    xid2 = ((TransactionId) page2) * COMMIT_TS_XACTS_PER_PAGE;
    xid2 += FirstNormalTransactionId + 1;
 
    return (TransactionIdPrecedes(xid1, xid2) &&
            TransactionIdPrecedes(xid1, xid2 + COMMIT_TS_XACTS_PER_PAGE - 1));
}
 
static int
commit_ts_errdetail_for_io_error(const void *opaque_data)
{
    TransactionId xid = *(const TransactionId *) opaque_data;
 
    return errdetail("Could not access commit timestamp of transaction %u.", xid);
}
 
/*
 * Write a TRUNCATE xlog record
 */
static void
WriteTruncateXlogRec(int64 pageno, TransactionId oldestXid)
{
    xl_commit_ts_truncate xlrec;
 
    xlrec.pageno = pageno;
    xlrec.oldestXid = oldestXid;
 
    XLogBeginInsert();
    XLogRegisterData(&xlrec, SizeOfCommitTsTruncate);
    (void) XLogInsert(RM_COMMIT_TS_ID, COMMIT_TS_TRUNCATE);
}
 
/*
 * CommitTS resource manager's routines
 */
void
commit_ts_redo(XLogReaderState *record)
{
    uint8       info = XLogRecGetInfo(record) & ~XLR_INFO_MASK;
 
    /* Backup blocks are not used in commit_ts records */
    Assert(!XLogRecHasAnyBlockRefs(record));
 
    if (info == COMMIT_TS_ZEROPAGE)
    {
        int64       pageno;
 
        memcpy(&pageno, XLogRecGetData(record), sizeof(pageno));
        SimpleLruZeroAndWritePage(CommitTsCtl, pageno);
    }
    else if (info == COMMIT_TS_TRUNCATE)
    {
        xl_commit_ts_truncate *trunc = (xl_commit_ts_truncate *) XLogRecGetData(record);
 
        AdvanceOldestCommitTsXid(trunc->oldestXid);
 
        /*
         * During XLOG replay, latest_page_number isn't set up yet; insert a
         * suitable value to bypass the sanity test in SimpleLruTruncate.
         */
        pg_atomic_write_u64(&CommitTsCtl->shared->latest_page_number,
                            trunc->pageno);
 
        SimpleLruTruncate(CommitTsCtl, trunc->pageno);
    }
    else
        elog(PANIC, "commit_ts_redo: unknown op code %u", info);
}
 
/*
 * Entrypoint for sync.c to sync commit_ts files.
 */
int
committssyncfiletag(const FileTag *ftag, char *path)
{
    return SlruSyncFileTag(CommitTsCtl, ftag, path);
}

CommitTsCtl

#define CommitTsCtl   (&CommitTsCtlData)

Definition at line 85 of file commit_ts.c.

SizeOfCommitTimestampEntry

#define SizeOfCommitTimestampEntry

Value:

                                    (offsetof(CommitTimestampEntry, nodeid) + \
									sizeof(ReplOriginId))

Definition at line 60 of file commit_ts.c.

TransactionIdToCTsEntry

#define TransactionIdToCTsEntry

(

xid

)

((xid) % (TransactionId) COMMIT_TS_XACTS_PER_PAGE)

Definition at line 77 of file commit_ts.c.

Typedef Documentation

CommitTimestampEntry

typedef struct CommitTimestampEntry CommitTimestampEntry

CommitTimestampShared

typedef struct CommitTimestampShared CommitTimestampShared

Function Documentation

ActivateCommitTs()

static void ActivateCommitTs ( void  )

static

Definition at line 683 of file commit_ts.c.

{
    TransactionId xid;
    int64       pageno;
 
    /*
     * During bootstrap, we should not register commit timestamps so skip the
     * activation in this case.
     */
    if (IsBootstrapProcessingMode())
        return;
 
    /* If we've done this already, there's nothing to do */
    LWLockAcquire(CommitTsLock, LW_EXCLUSIVE);
    if (commitTsShared->commitTsActive)
    {
        LWLockRelease(CommitTsLock);
        return;
    }
    LWLockRelease(CommitTsLock);
 
    xid = XidFromFullTransactionId(TransamVariables->nextXid);
    pageno = TransactionIdToCTsPage(xid);
 
    /*
     * Re-Initialize our idea of the latest page number.
     */
    pg_atomic_write_u64(&CommitTsCtl->shared->latest_page_number, pageno);
 
    /*
     * If CommitTs is enabled, but it wasn't in the previous server run, we
     * need to set the oldest and newest values to the next Xid; that way, we
     * will not try to read data that might not have been set.
     *
     * XXX does this have a problem if a server is started with commitTs
     * enabled, then started with commitTs disabled, then restarted with it
     * enabled again?  It doesn't look like it does, because there should be a
     * checkpoint that sets the value to InvalidTransactionId at end of
     * recovery; and so any chance of injecting new transactions without
     * CommitTs values would occur after the oldestCommitTsXid has been set to
     * Invalid temporarily.
     */
    LWLockAcquire(CommitTsLock, LW_EXCLUSIVE);
    if (TransamVariables->oldestCommitTsXid == InvalidTransactionId)
    {
        TransamVariables->oldestCommitTsXid =
            TransamVariables->newestCommitTsXid = ReadNextTransactionId();
    }
    LWLockRelease(CommitTsLock);
 
    /* Create the current segment file, if necessary */
    if (!SimpleLruDoesPhysicalPageExist(CommitTsCtl, pageno))
        SimpleLruZeroAndWritePage(CommitTsCtl, pageno);
 
    /* Change the activation status in shared memory. */
    LWLockAcquire(CommitTsLock, LW_EXCLUSIVE);
    commitTsShared->commitTsActive = true;
    LWLockRelease(CommitTsLock);
}

References CommitTimestampShared::commitTsActive, CommitTsCtl, commitTsShared, fb(), InvalidTransactionId, IsBootstrapProcessingMode, LW_EXCLUSIVE, LWLockAcquire(), LWLockRelease(), TransamVariablesData::newestCommitTsXid, TransamVariablesData::nextXid, TransamVariablesData::oldestCommitTsXid, pg_atomic_write_u64(), ReadNextTransactionId(), SimpleLruDoesPhysicalPageExist(), SimpleLruZeroAndWritePage(), TransactionIdToCTsPage(), TransamVariables, and XidFromFullTransactionId.

Referenced by CommitTsParameterChange(), CompleteCommitTsInitialization(), and StartupCommitTs().

AdvanceOldestCommitTsXid()

void AdvanceOldestCommitTsXid

(

TransactionId

oldestXact

)

Definition at line 916 of file commit_ts.c.

{
    LWLockAcquire(CommitTsLock, LW_EXCLUSIVE);
    if (TransamVariables->oldestCommitTsXid != InvalidTransactionId &&
        TransactionIdPrecedes(TransamVariables->oldestCommitTsXid, oldestXact))
        TransamVariables->oldestCommitTsXid = oldestXact;
    LWLockRelease(CommitTsLock);
}

References fb(), InvalidTransactionId, LW_EXCLUSIVE, LWLockAcquire(), LWLockRelease(), TransamVariablesData::oldestCommitTsXid, TransactionIdPrecedes(), and TransamVariables.

Referenced by commit_ts_redo(), and vac_truncate_clog().

BootStrapCommitTs()

void BootStrapCommitTs

(

void

)

Definition at line 596 of file commit_ts.c.

{
    /*
     * Nothing to do here at present, unlike most other SLRU modules; segments
     * are created when the server is started with this module enabled. See
     * ActivateCommitTs.
     */
}

Referenced by BootStrapXLOG().

check_commit_ts_buffers()

bool check_commit_ts_buffers	(	int *	newval,
		void **	extra,
		GucSource	source
	)

Definition at line 584 of file commit_ts.c.

{
    return check_slru_buffers("commit_timestamp_buffers", newval);
}

References check_slru_buffers(), and newval.

CheckPointCommitTs()

void CheckPointCommitTs

(

void

)

Definition at line 796 of file commit_ts.c.

{
    /*
     * Write dirty CommitTs pages to disk.  This may result in sync requests
     * queued for later handling by ProcessSyncRequests(), as part of the
     * checkpoint.
     */
    SimpleLruWriteAll(CommitTsCtl, true);
}

References CommitTsCtl, and SimpleLruWriteAll().

Referenced by CheckPointGuts().

commit_ts_errdetail_for_io_error()

static int commit_ts_errdetail_for_io_error ( const void *  opaque_data )

static

Definition at line 965 of file commit_ts.c.

{
    TransactionId xid = *(const TransactionId *) opaque_data;
 
    return errdetail("Could not access commit timestamp of transaction %u.", xid);
}

References errdetail(), and fb().

Referenced by CommitTsShmemInit().

commit_ts_redo()

void commit_ts_redo

(

XLogReaderState *

record

)

Definition at line 992 of file commit_ts.c.

{
    uint8       info = XLogRecGetInfo(record) & ~XLR_INFO_MASK;
 
    /* Backup blocks are not used in commit_ts records */
    Assert(!XLogRecHasAnyBlockRefs(record));
 
    if (info == COMMIT_TS_ZEROPAGE)
    {
        int64       pageno;
 
        memcpy(&pageno, XLogRecGetData(record), sizeof(pageno));
        SimpleLruZeroAndWritePage(CommitTsCtl, pageno);
    }
    else if (info == COMMIT_TS_TRUNCATE)
    {
        xl_commit_ts_truncate *trunc = (xl_commit_ts_truncate *) XLogRecGetData(record);
 
        AdvanceOldestCommitTsXid(trunc->oldestXid);
 
        /*
         * During XLOG replay, latest_page_number isn't set up yet; insert a
         * suitable value to bypass the sanity test in SimpleLruTruncate.
         */
        pg_atomic_write_u64(&CommitTsCtl->shared->latest_page_number,
                            trunc->pageno);
 
        SimpleLruTruncate(CommitTsCtl, trunc->pageno);
    }
    else
        elog(PANIC, "commit_ts_redo: unknown op code %u", info);
}

References AdvanceOldestCommitTsXid(), Assert, COMMIT_TS_TRUNCATE, COMMIT_TS_ZEROPAGE, CommitTsCtl, elog, fb(), PANIC, pg_atomic_write_u64(), SimpleLruTruncate(), SimpleLruZeroAndWritePage(), XLogRecGetData, XLogRecGetInfo, and XLogRecHasAnyBlockRefs.

CommitTsPagePrecedes()

static bool CommitTsPagePrecedes ( int64  page1, int64  page2  )

static

Definition at line 950 of file commit_ts.c.

{
    TransactionId xid1;
    TransactionId xid2;
 
    xid1 = ((TransactionId) page1) * COMMIT_TS_XACTS_PER_PAGE;
    xid1 += FirstNormalTransactionId + 1;
    xid2 = ((TransactionId) page2) * COMMIT_TS_XACTS_PER_PAGE;
    xid2 += FirstNormalTransactionId + 1;
 
    return (TransactionIdPrecedes(xid1, xid2) &&
            TransactionIdPrecedes(xid1, xid2 + COMMIT_TS_XACTS_PER_PAGE - 1));
}

References COMMIT_TS_XACTS_PER_PAGE, fb(), FirstNormalTransactionId, and TransactionIdPrecedes().

Referenced by CommitTsShmemInit().

CommitTsParameterChange()

void CommitTsParameterChange	(	bool	newvalue,
		bool	oldvalue
	)

Definition at line 642 of file commit_ts.c.

{
    /*
     * If the commit_ts module is disabled in this server and we get word from
     * the primary server that it is enabled there, activate it so that we can
     * replay future WAL records involving it; also mark it as active on
     * pg_control.  If the old value was already set, we already did this, so
     * don't do anything.
     *
     * If the module is disabled in the primary, disable it here too, unless
     * the module is enabled locally.
     *
     * Note this only runs in the recovery process, so an unlocked read is
     * fine.
     */
    if (newvalue)
    {
        if (!commitTsShared->commitTsActive)
            ActivateCommitTs();
    }
    else if (commitTsShared->commitTsActive)
        DeactivateCommitTs();
}

References ActivateCommitTs(), CommitTimestampShared::commitTsActive, commitTsShared, DeactivateCommitTs(), and fb().

Referenced by xlog_redo().

CommitTsShmemBuffers()

static int CommitTsShmemBuffers ( void  )

static

Definition at line 505 of file commit_ts.c.

{
    /* auto-tune based on shared buffers */
    if (commit_timestamp_buffers == 0)
        return SimpleLruAutotuneBuffers(512, 1024);
 
    return Min(Max(16, commit_timestamp_buffers), SLRU_MAX_ALLOWED_BUFFERS);
}

References commit_timestamp_buffers, Max, Min, SimpleLruAutotuneBuffers(), and SLRU_MAX_ALLOWED_BUFFERS.

Referenced by CommitTsShmemInit(), and CommitTsShmemSize().

CommitTsShmemInit()

void CommitTsShmemInit

(

void

)

Definition at line 529 of file commit_ts.c.

{
    bool        found;
 
    /* If auto-tuning is requested, now is the time to do it */
    if (commit_timestamp_buffers == 0)
    {
        char        buf[32];
 
        snprintf(buf, sizeof(buf), "%d", CommitTsShmemBuffers());
        SetConfigOption("commit_timestamp_buffers", buf, PGC_POSTMASTER,
                        PGC_S_DYNAMIC_DEFAULT);
 
        /*
         * We prefer to report this value's source as PGC_S_DYNAMIC_DEFAULT.
         * However, if the DBA explicitly set commit_timestamp_buffers = 0 in
         * the config file, then PGC_S_DYNAMIC_DEFAULT will fail to override
         * that and we must force the matter with PGC_S_OVERRIDE.
         */
        if (commit_timestamp_buffers == 0)  /* failed to apply it? */
            SetConfigOption("commit_timestamp_buffers", buf, PGC_POSTMASTER,
                            PGC_S_OVERRIDE);
    }
    Assert(commit_timestamp_buffers != 0);
 
    CommitTsCtl->PagePrecedes = CommitTsPagePrecedes;
    CommitTsCtl->errdetail_for_io_error = commit_ts_errdetail_for_io_error;
    SimpleLruInit(CommitTsCtl, "commit_timestamp", CommitTsShmemBuffers(), 0,
                  "pg_commit_ts", LWTRANCHE_COMMITTS_BUFFER,
                  LWTRANCHE_COMMITTS_SLRU,
                  SYNC_HANDLER_COMMIT_TS,
                  false);
    SlruPagePrecedesUnitTests(CommitTsCtl, COMMIT_TS_XACTS_PER_PAGE);
 
    commitTsShared = ShmemInitStruct("CommitTs shared",
                                     sizeof(CommitTimestampShared),
                                     &found);
 
    if (!IsUnderPostmaster)
    {
        Assert(!found);
 
        commitTsShared->xidLastCommit = InvalidTransactionId;
        TIMESTAMP_NOBEGIN(commitTsShared->dataLastCommit.time);
        commitTsShared->dataLastCommit.nodeid = InvalidReplOriginId;
        commitTsShared->commitTsActive = false;
    }
    else
        Assert(found);
}

References Assert, buf, commit_timestamp_buffers, commit_ts_errdetail_for_io_error(), COMMIT_TS_XACTS_PER_PAGE, CommitTimestampShared::commitTsActive, CommitTsCtl, CommitTsPagePrecedes(), commitTsShared, CommitTsShmemBuffers(), CommitTimestampShared::dataLastCommit, fb(), InvalidReplOriginId, InvalidTransactionId, IsUnderPostmaster, CommitTimestampEntry::nodeid, PGC_POSTMASTER, PGC_S_DYNAMIC_DEFAULT, PGC_S_OVERRIDE, SetConfigOption(), ShmemInitStruct(), SimpleLruInit(), SlruPagePrecedesUnitTests, snprintf, SYNC_HANDLER_COMMIT_TS, CommitTimestampEntry::time, TIMESTAMP_NOBEGIN, and CommitTimestampShared::xidLastCommit.

Referenced by CreateOrAttachShmemStructs().

CommitTsShmemSize()

Size CommitTsShmemSize

(

void

)

Definition at line 518 of file commit_ts.c.

{
    return SimpleLruShmemSize(CommitTsShmemBuffers(), 0) +
        sizeof(CommitTimestampShared);
}

References CommitTsShmemBuffers(), and SimpleLruShmemSize().

Referenced by CalculateShmemSize().

committssyncfiletag()

int committssyncfiletag	(	const FileTag *	ftag,
		char *	path
	)

Definition at line 1029 of file commit_ts.c.

{
    return SlruSyncFileTag(CommitTsCtl, ftag, path);
}

References CommitTsCtl, and SlruSyncFileTag().

CompleteCommitTsInitialization()

void CompleteCommitTsInitialization

(

void

)

Definition at line 620 of file commit_ts.c.

{
    /*
     * If the feature is not enabled, turn it off for good.  This also removes
     * any leftover data.
     *
     * Conversely, we activate the module if the feature is enabled.  This is
     * necessary for primary and standby as the activation depends on the
     * control file contents at the beginning of recovery or when a
     * XLOG_PARAMETER_CHANGE is replayed.
     */
    if (!track_commit_timestamp)
        DeactivateCommitTs();
    else
        ActivateCommitTs();
}

References ActivateCommitTs(), DeactivateCommitTs(), and track_commit_timestamp.

Referenced by StartupXLOG().

DeactivateCommitTs()

static void DeactivateCommitTs ( void  )

static

Definition at line 754 of file commit_ts.c.

{
    /*
     * Cleanup the status in the shared memory.
     *
     * We reset everything in the commitTsShared record to prevent user from
     * getting confusing data about last committed transaction on the standby
     * when the module was activated repeatedly on the primary.
     */
    LWLockAcquire(CommitTsLock, LW_EXCLUSIVE);
 
    commitTsShared->commitTsActive = false;
    commitTsShared->xidLastCommit = InvalidTransactionId;
    TIMESTAMP_NOBEGIN(commitTsShared->dataLastCommit.time);
    commitTsShared->dataLastCommit.nodeid = InvalidReplOriginId;
 
    TransamVariables->oldestCommitTsXid = InvalidTransactionId;
    TransamVariables->newestCommitTsXid = InvalidTransactionId;
 
    /*
     * Remove *all* files.  This is necessary so that there are no leftover
     * files; in the case where this feature is later enabled after running
     * with it disabled for some time there may be a gap in the file sequence.
     * (We can probably tolerate out-of-sequence files, as they are going to
     * be overwritten anyway when we wrap around, but it seems better to be
     * tidy.)
     *
     * Note that we do this with CommitTsLock acquired in exclusive mode. This
     * is very heavy-handed, but since this routine can only be called in the
     * replica and should happen very rarely, we don't worry too much about
     * it.  Note also that no process should be consulting this SLRU if we
     * have just deactivated it.
     */
    (void) SlruScanDirectory(CommitTsCtl, SlruScanDirCbDeleteAll, NULL);
 
    LWLockRelease(CommitTsLock);
}

References CommitTimestampShared::commitTsActive, CommitTsCtl, commitTsShared, CommitTimestampShared::dataLastCommit, fb(), InvalidReplOriginId, InvalidTransactionId, LW_EXCLUSIVE, LWLockAcquire(), LWLockRelease(), TransamVariablesData::newestCommitTsXid, CommitTimestampEntry::nodeid, TransamVariablesData::oldestCommitTsXid, SlruScanDirCbDeleteAll(), SlruScanDirectory(), CommitTimestampEntry::time, TIMESTAMP_NOBEGIN, TransamVariables, and CommitTimestampShared::xidLastCommit.

Referenced by CommitTsParameterChange(), and CompleteCommitTsInitialization().

error_commit_ts_disabled()

static void error_commit_ts_disabled ( void  )

static

Definition at line 380 of file commit_ts.c.

{
    ereport(ERROR,
            (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
             errmsg("could not get commit timestamp data"),
             RecoveryInProgress() ?
             errhint("Make sure the configuration parameter \"%s\" is set on the primary server.",
                     "track_commit_timestamp") :
             errhint("Make sure the configuration parameter \"%s\" is set.",
                     "track_commit_timestamp")));
}

References ereport, errcode(), errhint(), errmsg, ERROR, fb(), and RecoveryInProgress().

Referenced by GetLatestCommitTsData(), and TransactionIdGetCommitTsData().

ExtendCommitTs()

void ExtendCommitTs

(

TransactionId

newestXact

)

Definition at line 818 of file commit_ts.c.

{
    int64       pageno;
    LWLock     *lock;
 
    /*
     * Nothing to do if module not enabled.  Note we do an unlocked read of
     * the flag here, which is okay because this routine is only called from
     * GetNewTransactionId, which is never called in a standby.
     */
    Assert(!InRecovery);
    if (!commitTsShared->commitTsActive)
        return;
 
    /*
     * No work except at first XID of a page.  But beware: just after
     * wraparound, the first XID of page zero is FirstNormalTransactionId.
     */
    if (TransactionIdToCTsEntry(newestXact) != 0 &&
        !TransactionIdEquals(newestXact, FirstNormalTransactionId))
        return;
 
    pageno = TransactionIdToCTsPage(newestXact);
 
    lock = SimpleLruGetBankLock(CommitTsCtl, pageno);
 
    LWLockAcquire(lock, LW_EXCLUSIVE);
 
    /* Zero the page ... */
    SimpleLruZeroPage(CommitTsCtl, pageno);
 
    /* and make a WAL entry about that, unless we're in REDO */
    if (!InRecovery)
        XLogSimpleInsertInt64(RM_COMMIT_TS_ID, COMMIT_TS_ZEROPAGE, pageno);
 
    LWLockRelease(lock);
}

References Assert, COMMIT_TS_ZEROPAGE, CommitTimestampShared::commitTsActive, CommitTsCtl, commitTsShared, fb(), FirstNormalTransactionId, InRecovery, LW_EXCLUSIVE, LWLockAcquire(), LWLockRelease(), SimpleLruGetBankLock(), SimpleLruZeroPage(), TransactionIdEquals, TransactionIdToCTsEntry, TransactionIdToCTsPage(), and XLogSimpleInsertInt64().

Referenced by GetNewTransactionId().

GetLatestCommitTsData()

TransactionId GetLatestCommitTsData	(	TimestampTz *	ts,
		ReplOriginId *	nodeid
	)

Definition at line 359 of file commit_ts.c.

{
    TransactionId xid;
 
    LWLockAcquire(CommitTsLock, LW_SHARED);
 
    /* Error if module not enabled */
    if (!commitTsShared->commitTsActive)
        error_commit_ts_disabled();
 
    xid = commitTsShared->xidLastCommit;
    if (ts)
        *ts = commitTsShared->dataLastCommit.time;
    if (nodeid)
        *nodeid = commitTsShared->dataLastCommit.nodeid;
    LWLockRelease(CommitTsLock);
 
    return xid;
}

References CommitTimestampShared::commitTsActive, commitTsShared, CommitTimestampShared::dataLastCommit, error_commit_ts_disabled(), fb(), LW_SHARED, LWLockAcquire(), LWLockRelease(), CommitTimestampEntry::nodeid, CommitTimestampEntry::time, and CommitTimestampShared::xidLastCommit.

Referenced by pg_last_committed_xact().

pg_last_committed_xact()

Datum pg_last_committed_xact

(

PG_FUNCTION_ARGS

)

Definition at line 419 of file commit_ts.c.

{
    TransactionId xid;
    ReplOriginId nodeid;
    TimestampTz ts;
    Datum       values[3];
    bool        nulls[3];
    TupleDesc   tupdesc;
    HeapTuple   htup;
 
    /* and construct a tuple with our data */
    xid = GetLatestCommitTsData(&ts, &nodeid);
 
    if (get_call_result_type(fcinfo, NULL, &tupdesc) != TYPEFUNC_COMPOSITE)
        elog(ERROR, "return type must be a row type");
 
    if (!TransactionIdIsNormal(xid))
    {
        memset(nulls, true, sizeof(nulls));
    }
    else
    {
        values[0] = TransactionIdGetDatum(xid);
        nulls[0] = false;
 
        values[1] = TimestampTzGetDatum(ts);
        nulls[1] = false;
 
        values[2] = ObjectIdGetDatum((Oid) nodeid);
        nulls[2] = false;
    }
 
    htup = heap_form_tuple(tupdesc, values, nulls);
 
    PG_RETURN_DATUM(HeapTupleGetDatum(htup));
}

References elog, ERROR, fb(), get_call_result_type(), GetLatestCommitTsData(), heap_form_tuple(), HeapTupleGetDatum(), ObjectIdGetDatum(), PG_RETURN_DATUM, TimestampTzGetDatum(), TransactionIdGetDatum(), TransactionIdIsNormal, TYPEFUNC_COMPOSITE, and values.

pg_xact_commit_timestamp()

Datum pg_xact_commit_timestamp

(

PG_FUNCTION_ARGS

)

Definition at line 396 of file commit_ts.c.

{
    TransactionId xid = PG_GETARG_TRANSACTIONID(0);
    TimestampTz ts;
    bool        found;
 
    found = TransactionIdGetCommitTsData(xid, &ts, NULL);
 
    if (!found)
        PG_RETURN_NULL();
 
    PG_RETURN_TIMESTAMPTZ(ts);
}

References fb(), PG_GETARG_TRANSACTIONID, PG_RETURN_NULL, PG_RETURN_TIMESTAMPTZ, and TransactionIdGetCommitTsData().

pg_xact_commit_timestamp_origin()

Datum pg_xact_commit_timestamp_origin

(

PG_FUNCTION_ARGS

)

Definition at line 463 of file commit_ts.c.

{
    TransactionId xid = PG_GETARG_TRANSACTIONID(0);
    ReplOriginId nodeid;
    TimestampTz ts;
    Datum       values[2];
    bool        nulls[2];
    TupleDesc   tupdesc;
    HeapTuple   htup;
    bool        found;
 
    found = TransactionIdGetCommitTsData(xid, &ts, &nodeid);
 
    if (get_call_result_type(fcinfo, NULL, &tupdesc) != TYPEFUNC_COMPOSITE)
        elog(ERROR, "return type must be a row type");
 
    if (!found)
    {
        memset(nulls, true, sizeof(nulls));
    }
    else
    {
        values[0] = TimestampTzGetDatum(ts);
        nulls[0] = false;
 
        values[1] = ObjectIdGetDatum((Oid) nodeid);
        nulls[1] = false;
    }
 
    htup = heap_form_tuple(tupdesc, values, nulls);
 
    PG_RETURN_DATUM(HeapTupleGetDatum(htup));
}

References elog, ERROR, fb(), get_call_result_type(), heap_form_tuple(), HeapTupleGetDatum(), ObjectIdGetDatum(), PG_GETARG_TRANSACTIONID, PG_RETURN_DATUM, TimestampTzGetDatum(), TransactionIdGetCommitTsData(), TYPEFUNC_COMPOSITE, and values.

SetCommitTsLimit()

void SetCommitTsLimit	(	TransactionId	oldestXact,
		TransactionId	newestXact
	)

Definition at line 889 of file commit_ts.c.

{
    /*
     * Be careful not to overwrite values that are either further into the
     * "future" or signal a disabled committs.
     */
    LWLockAcquire(CommitTsLock, LW_EXCLUSIVE);
    if (TransamVariables->oldestCommitTsXid != InvalidTransactionId)
    {
        if (TransactionIdPrecedes(TransamVariables->oldestCommitTsXid, oldestXact))
            TransamVariables->oldestCommitTsXid = oldestXact;
        if (TransactionIdPrecedes(newestXact, TransamVariables->newestCommitTsXid))
            TransamVariables->newestCommitTsXid = newestXact;
    }
    else
    {
        Assert(TransamVariables->newestCommitTsXid == InvalidTransactionId);
        TransamVariables->oldestCommitTsXid = oldestXact;
        TransamVariables->newestCommitTsXid = newestXact;
    }
    LWLockRelease(CommitTsLock);
}

References Assert, fb(), InvalidTransactionId, LW_EXCLUSIVE, LWLockAcquire(), LWLockRelease(), TransamVariablesData::newestCommitTsXid, TransamVariablesData::oldestCommitTsXid, TransactionIdPrecedes(), and TransamVariables.

Referenced by BootStrapXLOG(), and StartupXLOG().

SetXidCommitTsInPage()

static void SetXidCommitTsInPage ( TransactionId  xid, int  nsubxids, TransactionId *  subxids, TimestampTz  ts, ReplOriginId  nodeid, int64  pageno  )

static

Definition at line 221 of file commit_ts.c.

{
    LWLock     *lock = SimpleLruGetBankLock(CommitTsCtl, pageno);
    int         slotno;
    int         i;
 
    LWLockAcquire(lock, LW_EXCLUSIVE);
 
    slotno = SimpleLruReadPage(CommitTsCtl, pageno, true, &xid);
 
    TransactionIdSetCommitTs(xid, ts, nodeid, slotno);
    for (i = 0; i < nsubxids; i++)
        TransactionIdSetCommitTs(subxids[i], ts, nodeid, slotno);
 
    CommitTsCtl->shared->page_dirty[slotno] = true;
 
    LWLockRelease(lock);
}

References CommitTsCtl, fb(), i, LW_EXCLUSIVE, LWLockAcquire(), LWLockRelease(), SimpleLruGetBankLock(), SimpleLruReadPage(), and TransactionIdSetCommitTs().

Referenced by TransactionTreeSetCommitTsData().

StartupCommitTs()

void StartupCommitTs

(

void

)

Definition at line 610 of file commit_ts.c.

{
    ActivateCommitTs();
}

References ActivateCommitTs().

Referenced by StartupXLOG().

TransactionIdGetCommitTsData()

bool TransactionIdGetCommitTsData	(	TransactionId	xid,
		TimestampTz *	ts,
		ReplOriginId *	nodeid
	)

Definition at line 273 of file commit_ts.c.

{
    int64       pageno = TransactionIdToCTsPage(xid);
    int         entryno = TransactionIdToCTsEntry(xid);
    int         slotno;
    CommitTimestampEntry entry;
    TransactionId oldestCommitTsXid;
    TransactionId newestCommitTsXid;
 
    if (!TransactionIdIsValid(xid))
        ereport(ERROR,
                (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
                 errmsg("cannot retrieve commit timestamp for transaction %u", xid)));
    else if (!TransactionIdIsNormal(xid))
    {
        /* frozen and bootstrap xids are always committed far in the past */
        *ts = 0;
        if (nodeid)
            *nodeid = 0;
        return false;
    }
 
    LWLockAcquire(CommitTsLock, LW_SHARED);
 
    /* Error if module not enabled */
    if (!commitTsShared->commitTsActive)
        error_commit_ts_disabled();
 
    /*
     * If we're asked for the cached value, return that.  Otherwise, fall
     * through to read from SLRU.
     */
    if (commitTsShared->xidLastCommit == xid)
    {
        *ts = commitTsShared->dataLastCommit.time;
        if (nodeid)
            *nodeid = commitTsShared->dataLastCommit.nodeid;
 
        LWLockRelease(CommitTsLock);
        return *ts != 0;
    }
 
    oldestCommitTsXid = TransamVariables->oldestCommitTsXid;
    newestCommitTsXid = TransamVariables->newestCommitTsXid;
    /* neither is invalid, or both are */
    Assert(TransactionIdIsValid(oldestCommitTsXid) == TransactionIdIsValid(newestCommitTsXid));
    LWLockRelease(CommitTsLock);
 
    /*
     * Return empty if the requested value is outside our valid range.
     */
    if (!TransactionIdIsValid(oldestCommitTsXid) ||
        TransactionIdPrecedes(xid, oldestCommitTsXid) ||
        TransactionIdPrecedes(newestCommitTsXid, xid))
    {
        *ts = 0;
        if (nodeid)
            *nodeid = InvalidReplOriginId;
        return false;
    }
 
    /* lock is acquired by SimpleLruReadPage_ReadOnly */
    slotno = SimpleLruReadPage_ReadOnly(CommitTsCtl, pageno, &xid);
    memcpy(&entry,
           CommitTsCtl->shared->page_buffer[slotno] +
           SizeOfCommitTimestampEntry * entryno,
           SizeOfCommitTimestampEntry);
 
    *ts = entry.time;
    if (nodeid)
        *nodeid = entry.nodeid;
 
    LWLockRelease(SimpleLruGetBankLock(CommitTsCtl, pageno));
    return *ts != 0;
}

References Assert, CommitTimestampShared::commitTsActive, CommitTsCtl, commitTsShared, CommitTimestampShared::dataLastCommit, ereport, errcode(), errmsg, ERROR, error_commit_ts_disabled(), fb(), InvalidReplOriginId, LW_SHARED, LWLockAcquire(), LWLockRelease(), TransamVariablesData::newestCommitTsXid, CommitTimestampEntry::nodeid, TransamVariablesData::oldestCommitTsXid, SimpleLruGetBankLock(), SimpleLruReadPage_ReadOnly(), SizeOfCommitTimestampEntry, CommitTimestampEntry::time, TransactionIdIsNormal, TransactionIdIsValid, TransactionIdPrecedes(), TransactionIdToCTsEntry, TransactionIdToCTsPage(), TransamVariables, and CommitTimestampShared::xidLastCommit.

Referenced by GetTupleTransactionInfo(), pg_xact_commit_timestamp(), pg_xact_commit_timestamp_origin(), and update_most_recent_deletion_info().

TransactionIdSetCommitTs()

static void TransactionIdSetCommitTs ( TransactionId  xid, TimestampTz  ts, ReplOriginId  nodeid, int  slotno  )

static

Definition at line 248 of file commit_ts.c.

{
    int         entryno = TransactionIdToCTsEntry(xid);
    CommitTimestampEntry entry;
 
    Assert(TransactionIdIsNormal(xid));
 
    entry.time = ts;
    entry.nodeid = nodeid;
 
    memcpy(CommitTsCtl->shared->page_buffer[slotno] +
           SizeOfCommitTimestampEntry * entryno,
           &entry, SizeOfCommitTimestampEntry);
}

References Assert, CommitTsCtl, fb(), CommitTimestampEntry::nodeid, SizeOfCommitTimestampEntry, CommitTimestampEntry::time, TransactionIdIsNormal, and TransactionIdToCTsEntry.

Referenced by SetXidCommitTsInPage().

TransactionIdToCTsPage()

static int64 TransactionIdToCTsPage ( TransactionId  xid )

inlinestatic

Definition at line 72 of file commit_ts.c.

{
    return xid / (int64) COMMIT_TS_XACTS_PER_PAGE;
}

References COMMIT_TS_XACTS_PER_PAGE.

Referenced by ActivateCommitTs(), ExtendCommitTs(), TransactionIdGetCommitTsData(), TransactionTreeSetCommitTsData(), and TruncateCommitTs().

TransactionTreeSetCommitTsData()

void TransactionTreeSetCommitTsData	(	TransactionId	xid,
		int	nsubxids,
		TransactionId *	subxids,
		TimestampTz	timestamp,
		ReplOriginId	nodeid
	)

Definition at line 140 of file commit_ts.c.

{
    int         i;
    TransactionId headxid;
    TransactionId newestXact;
 
    /*
     * No-op if the module is not active.
     *
     * An unlocked read here is fine, because in a standby (the only place
     * where the flag can change in flight) this routine is only called by the
     * recovery process, which is also the only process which can change the
     * flag.
     */
    if (!commitTsShared->commitTsActive)
        return;
 
    /*
     * Figure out the latest Xid in this batch: either the last subxid if
     * there's any, otherwise the parent xid.
     */
    if (nsubxids > 0)
        newestXact = subxids[nsubxids - 1];
    else
        newestXact = xid;
 
    /*
     * We split the xids to set the timestamp to in groups belonging to the
     * same SLRU page; the first element in each such set is its head.  The
     * first group has the main XID as the head; subsequent sets use the first
     * subxid not on the previous page as head.  This way, we only have to
     * lock/modify each SLRU page once.
     */
    headxid = xid;
    i = 0;
    for (;;)
    {
        int64       pageno = TransactionIdToCTsPage(headxid);
        int         j;
 
        for (j = i; j < nsubxids; j++)
        {
            if (TransactionIdToCTsPage(subxids[j]) != pageno)
                break;
        }
        /* subxids[i..j] are on the same page as the head */
 
        SetXidCommitTsInPage(headxid, j - i, subxids + i, timestamp, nodeid,
                             pageno);
 
        /* if we wrote out all subxids, we're done. */
        if (j >= nsubxids)
            break;
 
        /*
         * Set the new head and skip over it, as well as over the subxids we
         * just wrote.
         */
        headxid = subxids[j];
        i = j + 1;
    }
 
    /* update the cached value in shared memory */
    LWLockAcquire(CommitTsLock, LW_EXCLUSIVE);
    commitTsShared->xidLastCommit = xid;
    commitTsShared->dataLastCommit.time = timestamp;
    commitTsShared->dataLastCommit.nodeid = nodeid;
 
    /* and move forwards our endpoint, if needed */
    if (TransactionIdPrecedes(TransamVariables->newestCommitTsXid, newestXact))
        TransamVariables->newestCommitTsXid = newestXact;
    LWLockRelease(CommitTsLock);
}

References CommitTimestampShared::commitTsActive, commitTsShared, CommitTimestampShared::dataLastCommit, fb(), i, j, LW_EXCLUSIVE, LWLockAcquire(), LWLockRelease(), TransamVariablesData::newestCommitTsXid, CommitTimestampEntry::nodeid, SetXidCommitTsInPage(), CommitTimestampEntry::time, TransactionIdPrecedes(), TransactionIdToCTsPage(), TransamVariables, and CommitTimestampShared::xidLastCommit.

Referenced by RecordTransactionCommit(), RecordTransactionCommitPrepared(), and xact_redo_commit().

TruncateCommitTs()

void TruncateCommitTs

(

TransactionId

oldestXact

)

Definition at line 863 of file commit_ts.c.

{
    int64       cutoffPage;
 
    /*
     * The cutoff point is the start of the segment containing oldestXact. We
     * pass the *page* containing oldestXact to SimpleLruTruncate.
     */
    cutoffPage = TransactionIdToCTsPage(oldestXact);
 
    /* Check to see if there's any files that could be removed */
    if (!SlruScanDirectory(CommitTsCtl, SlruScanDirCbReportPresence,
                           &cutoffPage))
        return;                 /* nothing to remove */
 
    /* Write XLOG record */
    WriteTruncateXlogRec(cutoffPage, oldestXact);
 
    /* Now we can remove the old CommitTs segment(s) */
    SimpleLruTruncate(CommitTsCtl, cutoffPage);
}

References CommitTsCtl, fb(), SimpleLruTruncate(), SlruScanDirCbReportPresence(), SlruScanDirectory(), TransactionIdToCTsPage(), and WriteTruncateXlogRec().

Referenced by vac_truncate_clog().

WriteTruncateXlogRec()

static void WriteTruncateXlogRec ( int64  pageno, TransactionId  oldestXid  )

static

Definition at line 976 of file commit_ts.c.

{
    xl_commit_ts_truncate xlrec;
 
    xlrec.pageno = pageno;
    xlrec.oldestXid = oldestXid;
 
    XLogBeginInsert();
    XLogRegisterData(&xlrec, SizeOfCommitTsTruncate);
    (void) XLogInsert(RM_COMMIT_TS_ID, COMMIT_TS_TRUNCATE);
}

References COMMIT_TS_TRUNCATE, fb(), xl_commit_ts_truncate::pageno, SizeOfCommitTsTruncate, XLogBeginInsert(), XLogInsert(), and XLogRegisterData().

Referenced by TruncateCommitTs().

Variable Documentation

CommitTsCtlData

SlruCtlData CommitTsCtlData

static

Definition at line 83 of file commit_ts.c.

commitTsShared

CommitTimestampShared* commitTsShared

static

Definition at line 105 of file commit_ts.c.

Referenced by ActivateCommitTs(), CommitTsParameterChange(), CommitTsShmemInit(), DeactivateCommitTs(), ExtendCommitTs(), GetLatestCommitTsData(), TransactionIdGetCommitTsData(), and TransactionTreeSetCommitTsData().

track_commit_timestamp

bool track_commit_timestamp

Definition at line 109 of file commit_ts.c.

Referenced by CheckSubDeadTupleRetention(), CompleteCommitTsInitialization(), FindDeletedTupleInLocalRel(), GetTupleTransactionInfo(), InitControlFile(), and XLogReportParameters().