diff --git a/src/backend/access/transam/Makefile b/src/backend/access/transam/Makefile index 94455b2..37523a1 100644 --- a/src/backend/access/transam/Makefile +++ b/src/backend/access/transam/Makefile @@ -15,7 +15,7 @@ include $(top_builddir)/src/Makefile.global OBJS = clog.o commit_ts.o multixact.o parallel.o rmgr.o slru.o subtrans.o \ timeline.o transam.o twophase.o twophase_rmgr.o varsup.o \ xact.o xlog.o xlogarchive.o xlogfuncs.o \ - xloginsert.o xlogreader.o xlogutils.o + xloginsert.o xlogreader.o xlogutils.o xtm.o include $(top_srcdir)/src/backend/common.mk diff --git a/src/backend/access/transam/clog.c b/src/backend/access/transam/clog.c index 06aff18..ba7b09f 100644 --- a/src/backend/access/transam/clog.c +++ b/src/backend/access/transam/clog.c @@ -38,6 +38,7 @@ #include "access/xlog.h" #include "access/xloginsert.h" #include "access/xlogutils.h" +#include "access/xtm.h" #include "miscadmin.h" #include "pg_trace.h" @@ -92,6 +93,12 @@ static void TransactionIdSetStatusBit(TransactionId xid, XidStatus status, static void set_status_by_pages(int nsubxids, TransactionId *subxids, XidStatus status, XLogRecPtr lsn); +void +TransactionIdSetTreeStatus(TransactionId xid, int nsubxids, + TransactionId *subxids, XidStatus status, XLogRecPtr lsn) +{ + return TM->SetTransactionStatus(xid, nsubxids, subxids, status, lsn); +} /* * TransactionIdSetTreeStatus @@ -145,7 +152,7 @@ static void set_status_by_pages(int nsubxids, TransactionId *subxids, * cache yet. */ void -TransactionIdSetTreeStatus(TransactionId xid, int nsubxids, +PgTransactionIdSetTreeStatus(TransactionId xid, int nsubxids, TransactionId *subxids, XidStatus status, XLogRecPtr lsn) { int pageno = TransactionIdToPage(xid); /* get page of parent */ @@ -391,6 +398,12 @@ TransactionIdSetStatusBit(TransactionId xid, XidStatus status, XLogRecPtr lsn, i XidStatus TransactionIdGetStatus(TransactionId xid, XLogRecPtr *lsn) { + return TM->GetTransactionStatus(xid, lsn); +} + +XidStatus +PgTransactionIdGetStatus(TransactionId xid, XLogRecPtr *lsn) +{ int pageno = TransactionIdToPage(xid); int byteno = TransactionIdToByte(xid); int bshift = TransactionIdToBIndex(xid) * CLOG_BITS_PER_XACT; diff --git a/src/backend/access/transam/twophase.c b/src/backend/access/transam/twophase.c index 8a22836..6c2813b 100644 --- a/src/backend/access/transam/twophase.c +++ b/src/backend/access/transam/twophase.c @@ -1381,15 +1381,21 @@ FinishPreparedTransaction(const char *gid, bool isCommit) * callbacks will release the locks the transaction held. */ if (isCommit) + { RecordTransactionCommitPrepared(xid, hdr->nsubxacts, children, hdr->ncommitrels, commitrels, hdr->ninvalmsgs, invalmsgs, hdr->initfileinval); + CallXactCallbacks(XACT_EVENT_COMMIT_PREPARED); + } else + { RecordTransactionAbortPrepared(xid, hdr->nsubxacts, children, hdr->nabortrels, abortrels); + CallXactCallbacks(XACT_EVENT_ABORT_PREPARED); + } ProcArrayRemove(proc, latestXid); @@ -2150,3 +2156,12 @@ RecordTransactionAbortPrepared(TransactionId xid, */ SyncRepWaitForLSN(recptr); } + +/* + * Return identified of current global transaction + */ +const char* +GetLockedGlobalTransactionId(void) +{ + return MyLockedGxact ? MyLockedGxact->gid : NULL; +} diff --git a/src/backend/access/transam/varsup.c b/src/backend/access/transam/varsup.c index 2f7e645..2f8514d 100644 --- a/src/backend/access/transam/varsup.c +++ b/src/backend/access/transam/varsup.c @@ -19,6 +19,7 @@ #include "access/transam.h" #include "access/xact.h" #include "access/xlog.h" +#include "access/xtm.h" #include "commands/dbcommands.h" #include "miscadmin.h" #include "postmaster/autovacuum.h" @@ -33,6 +34,11 @@ /* pointer to "variable cache" in shared memory (set up by shmem.c) */ VariableCache ShmemVariableCache = NULL; +TransactionId +GetNewTransactionId(bool isSubXact) +{ + return TM->GetNewTransactionId(isSubXact); +} /* * Allocate the next XID for a new transaction or subtransaction. @@ -45,7 +51,7 @@ VariableCache ShmemVariableCache = NULL; * issue a warning about XID wrap. */ TransactionId -GetNewTransactionId(bool isSubXact) +PgGetNewTransactionId(bool isSubXact) { TransactionId xid; diff --git a/src/backend/access/transam/xact.c b/src/backend/access/transam/xact.c index b0d5440..c8d58d3 100644 --- a/src/backend/access/transam/xact.c +++ b/src/backend/access/transam/xact.c @@ -301,7 +301,6 @@ static void AtCommit_Memory(void); static void AtStart_Cache(void); static void AtStart_Memory(void); static void AtStart_ResourceOwner(void); -static void CallXactCallbacks(XactEvent event); static void CallSubXactCallbacks(SubXactEvent event, SubTransactionId mySubid, SubTransactionId parentSubid); @@ -1909,6 +1908,7 @@ StartTransaction(void) */ s->state = TRANS_INPROGRESS; + CallXactCallbacks(XACT_EVENT_START); ShowTransactionState("StartTransaction"); } @@ -3309,7 +3309,7 @@ UnregisterXactCallback(XactCallback callback, void *arg) } } -static void +void CallXactCallbacks(XactEvent event) { XactCallbackItem *item; @@ -5607,3 +5607,10 @@ xact_redo(XLogReaderState *record) else elog(PANIC, "xact_redo: unknown op code %u", info); } + +void +MarkAsAborted() +{ + CurrentTransactionState->state = TRANS_INPROGRESS; + CurrentTransactionState->blockState = TBLOCK_STARTED; +} diff --git a/src/backend/access/transam/xtm.c b/src/backend/access/transam/xtm.c new file mode 100644 index 0000000..7915b1e --- /dev/null +++ b/src/backend/access/transam/xtm.c @@ -0,0 +1,58 @@ +/*------------------------------------------------------------------------- + * + * xtm.c + * PostgreSQL implementation of transaction manager protocol + * + * This module defines default iplementaiton of PostgreSQL transaction manager protocol + * + * Portions Copyright (c) 1996-2015, PostgreSQL Global Development Group + * Portions Copyright (c) 1994, Regents of the University of California + * + * src/backend/access/transam/xtm.c + * + *------------------------------------------------------------------------- + */ + +#include "postgres.h" + +#include "access/transam.h" +#include "access/xtm.h" + +TransactionId +PgGetGlobalTransactionId(void) +{ + return InvalidTransactionId; +} + +bool +PgDetectGlobalDeadLock(PGPROC *proc) +{ + return false; +} + +char const * +PgGetTransactionManagerName(void) +{ + return "postgres"; +} + +TransactionManager PgTM = { + PgTransactionIdGetStatus, + PgTransactionIdSetTreeStatus, + PgGetSnapshotData, + PgGetNewTransactionId, + PgGetOldestXmin, + PgTransactionIdIsInProgress, + PgGetGlobalTransactionId, + PgXidInMVCCSnapshot, + PgDetectGlobalDeadLock, + PgGetTransactionManagerName +}; + +TransactionManager *TM = &PgTM; + +TransactionManager * +GetTransactionManager(void) +{ + return TM; +} diff --git a/src/backend/storage/ipc/procarray.c b/src/backend/storage/ipc/procarray.c index 6ded0f0..c2e878c 100644 --- a/src/backend/storage/ipc/procarray.c +++ b/src/backend/storage/ipc/procarray.c @@ -51,6 +51,7 @@ #include "access/twophase.h" #include "access/xact.h" #include "access/xlog.h" +#include "access/xtm.h" #include "catalog/catalog.h" #include "miscadmin.h" #include "storage/proc.h" @@ -971,6 +972,12 @@ ProcArrayApplyXidAssignment(TransactionId topxid, LWLockRelease(ProcArrayLock); } +bool +TransactionIdIsInProgress(TransactionId xid) +{ + return TM->IsInProgress(xid); +} + /* * TransactionIdIsInProgress -- is given transaction running in some backend * @@ -998,7 +1005,7 @@ ProcArrayApplyXidAssignment(TransactionId topxid, * PGXACT again anyway; see GetNewTransactionId). */ bool -TransactionIdIsInProgress(TransactionId xid) +PgTransactionIdIsInProgress(TransactionId xid) { static TransactionId *xids = NULL; int nxids = 0; @@ -1259,6 +1266,12 @@ TransactionIdIsActive(TransactionId xid) } +TransactionId +GetOldestXmin(Relation rel, bool ignoreVacuum) +{ + return TM->GetOldestXmin(rel, ignoreVacuum); +} + /* * GetOldestXmin -- returns oldest transaction that was running * when any current transaction was started. @@ -1308,7 +1321,7 @@ TransactionIdIsActive(TransactionId xid) * GetOldestXmin() move backwards, with no consequences for data integrity. */ TransactionId -GetOldestXmin(Relation rel, bool ignoreVacuum) +PgGetOldestXmin(Relation rel, bool ignoreVacuum) { ProcArrayStruct *arrayP = procArray; TransactionId result; @@ -1470,6 +1483,12 @@ GetMaxSnapshotSubxidCount(void) return TOTAL_MAX_CACHED_SUBXIDS; } +Snapshot +GetSnapshotData(Snapshot snapshot) +{ + return TM->GetSnapshot(snapshot); +} + /* * GetSnapshotData -- returns information about running transactions. * @@ -1506,7 +1525,7 @@ GetMaxSnapshotSubxidCount(void) * not statically allocated (see xip allocation below). */ Snapshot -GetSnapshotData(Snapshot snapshot) +PgGetSnapshotData(Snapshot snapshot) { ProcArrayStruct *arrayP = procArray; TransactionId xmin; diff --git a/src/backend/storage/lmgr/deadlock.c b/src/backend/storage/lmgr/deadlock.c index 69f678b..ec8b77e 100644 --- a/src/backend/storage/lmgr/deadlock.c +++ b/src/backend/storage/lmgr/deadlock.c @@ -30,6 +30,7 @@ #include "pgstat.h" #include "storage/lmgr.h" #include "storage/proc.h" +#include "access/xtm.h" #include "utils/memutils.h" @@ -272,7 +273,7 @@ DeadLockCheck(PGPROC *proc) else if (blocking_autovacuum_proc != NULL) return DS_BLOCKED_BY_AUTOVACUUM; else - return DS_NO_DEADLOCK; + return TM->DetectGlobalDeadLock(proc) ? DS_DISTRIBUTED_DEADLOCK : DS_NO_DEADLOCK; } /* diff --git a/src/backend/storage/lmgr/lock.c b/src/backend/storage/lmgr/lock.c index e3e9599..21bf32b 100644 --- a/src/backend/storage/lmgr/lock.c +++ b/src/backend/storage/lmgr/lock.c @@ -3639,6 +3639,20 @@ GetLockmodeName(LOCKMETHODID lockmethodid, LOCKMODE mode) return LockMethods[lockmethodid]->lockModeNames[mode]; } +void +EnumerateLocks(LockIterator iterator, void* arg) +{ + PROCLOCK *proclock; + HASH_SEQ_STATUS status; + + hash_seq_init(&status, LockMethodProcLockHash); + + while ((proclock = (PROCLOCK *) hash_seq_search(&status)) != NULL) + { + iterator(proclock, arg); + } +} + #ifdef LOCK_DEBUG /* * Dump all locks in the given proc's myProcLocks lists. diff --git a/src/backend/storage/lmgr/proc.c b/src/backend/storage/lmgr/proc.c index 084be5a..8ded531 100644 --- a/src/backend/storage/lmgr/proc.c +++ b/src/backend/storage/lmgr/proc.c @@ -1395,6 +1395,22 @@ ProcSleep(LOCALLOCK *locallock, LockMethod lockMethodTable) "Processes holding the lock: %s. Wait queue: %s.", lockHoldersNum, lock_holders_sbuf.data, lock_waiters_sbuf.data)))); } + else if (deadlock_state == DS_DISTRIBUTED_DEADLOCK) + { + /* + * This message is a bit redundant with the error that will be + * reported subsequently, but in some cases the error report + * might not make it to the log (eg, if it's caught by an + * exception handler), and we want to ensure all long-wait + * events get logged. + */ + ereport(LOG, + (errmsg("process %d detected distributed deadlock while waiting for %s on %s after %ld.%03d ms", + MyProcPid, modename, buf.data, msecs, usecs), + (errdetail_log_plural("Process holding the lock: %s. Wait queue: %s.", + "Processes holding the lock: %s. Wait queue: %s.", + lockHoldersNum, lock_holders_sbuf.data, lock_waiters_sbuf.data)))); + } if (myWaitStatus == STATUS_WAITING) ereport(LOG, @@ -1419,7 +1435,7 @@ ProcSleep(LOCALLOCK *locallock, LockMethod lockMethodTable) * future-proofing, print a message if it looks like someone * else kicked us off the lock. */ - if (deadlock_state != DS_HARD_DEADLOCK) + if (deadlock_state != DS_HARD_DEADLOCK && deadlock_state != DS_DISTRIBUTED_DEADLOCK) ereport(LOG, (errmsg("process %d failed to acquire %s on %s after %ld.%03d ms", MyProcPid, modename, buf.data, msecs, usecs), @@ -1636,7 +1652,7 @@ CheckDeadLock(void) /* Run the deadlock check, and set deadlock_state for use by ProcSleep */ deadlock_state = DeadLockCheck(MyProc); - if (deadlock_state == DS_HARD_DEADLOCK) + if (deadlock_state == DS_HARD_DEADLOCK || deadlock_state == DS_DISTRIBUTED_DEADLOCK) { /* * Oops. We have a deadlock. diff --git a/src/backend/utils/time/tqual.c b/src/backend/utils/time/tqual.c index 465933d..7ca0e06 100644 --- a/src/backend/utils/time/tqual.c +++ b/src/backend/utils/time/tqual.c @@ -66,6 +66,7 @@ #include "access/subtrans.h" #include "access/transam.h" #include "access/xact.h" +#include "access/xtm.h" #include "access/xlog.h" #include "storage/bufmgr.h" #include "storage/procarray.h" @@ -1454,6 +1455,12 @@ HeapTupleIsSurelyDead(HeapTuple htup, TransactionId OldestXmin) return TransactionIdPrecedes(HeapTupleHeaderGetRawXmax(tuple), OldestXmin); } +bool +XidInMVCCSnapshot(TransactionId xid, Snapshot snapshot) +{ + return TM->IsInSnapshot(xid, snapshot); +} + /* * XidInMVCCSnapshot * Is the given XID still-in-progress according to the snapshot? @@ -1464,8 +1471,8 @@ HeapTupleIsSurelyDead(HeapTuple htup, TransactionId OldestXmin) * TransactionIdIsCurrentTransactionId first, except for known-committed * XIDs which could not be ours anyway. */ -static bool -XidInMVCCSnapshot(TransactionId xid, Snapshot snapshot) +bool +PgXidInMVCCSnapshot(TransactionId xid, Snapshot snapshot) { uint32 i; diff --git a/src/include/access/clog.h b/src/include/access/clog.h index 06c069a..3adf6b1 100644 --- a/src/include/access/clog.h +++ b/src/include/access/clog.h @@ -27,6 +27,7 @@ typedef int XidStatus; #define TRANSACTION_STATUS_COMMITTED 0x01 #define TRANSACTION_STATUS_ABORTED 0x02 #define TRANSACTION_STATUS_SUB_COMMITTED 0x03 +#define TRANSACTION_STATUS_UNKNOWN 0x03 extern void TransactionIdSetTreeStatus(TransactionId xid, int nsubxids, diff --git a/src/include/access/twophase.h b/src/include/access/twophase.h index b7ce0c6..05d6aef 100644 --- a/src/include/access/twophase.h +++ b/src/include/access/twophase.h @@ -56,4 +56,6 @@ extern void CheckPointTwoPhase(XLogRecPtr redo_horizon); extern void FinishPreparedTransaction(const char *gid, bool isCommit); +extern const char *GetLockedGlobalTransactionId(void); + #endif /* TWOPHASE_H */ diff --git a/src/include/access/xact.h b/src/include/access/xact.h index ebeb582..4a59047 100644 --- a/src/include/access/xact.h +++ b/src/include/access/xact.h @@ -77,6 +77,7 @@ extern bool MyXactAccessedTempRel; */ typedef enum { + XACT_EVENT_START, XACT_EVENT_COMMIT, XACT_EVENT_PARALLEL_COMMIT, XACT_EVENT_ABORT, @@ -84,7 +85,9 @@ typedef enum XACT_EVENT_PREPARE, XACT_EVENT_PRE_COMMIT, XACT_EVENT_PARALLEL_PRE_COMMIT, - XACT_EVENT_PRE_PREPARE + XACT_EVENT_PRE_PREPARE, + XACT_EVENT_COMMIT_PREPARED, + XACT_EVENT_ABORT_PREPARED } XactEvent; typedef void (*XactCallback) (XactEvent event, void *arg); @@ -100,6 +103,8 @@ typedef enum typedef void (*SubXactCallback) (SubXactEvent event, SubTransactionId mySubid, SubTransactionId parentSubid, void *arg); +void CallXactCallbacks(XactEvent event); + /* ---------------- * transaction-related XLOG entries @@ -380,4 +385,6 @@ extern void EnterParallelMode(void); extern void ExitParallelMode(void); extern bool IsInParallelMode(void); +extern void MarkAsAborted(void); + #endif /* XACT_H */ diff --git a/src/include/access/xtm.h b/src/include/access/xtm.h new file mode 100644 index 0000000..08fa259 --- /dev/null +++ b/src/include/access/xtm.h @@ -0,0 +1,103 @@ +/* + * xtm.h + * + * PostgreSQL transaction-commit-log manager + * + * Portions Copyright (c) 1996-2015, PostgreSQL Global Development Group + * Portions Copyright (c) 1994, Regents of the University of California + * + * src/include/access/xtm.h + */ +#ifndef XTM_H +#define XTM_H + +#include "storage/proc.h" +#include "access/clog.h" +#include "utils/snapmgr.h" +#include "utils/relcache.h" + +typedef struct +{ + /* + * Get current transaction status (encapsulation of TransactionIdGetStatus + * in clog.c) + */ + XidStatus (*GetTransactionStatus) (TransactionId xid, XLogRecPtr *lsn); + + /* + * Set current transaction status (encapsulation of + * TransactionIdSetTreeStatus in clog.c) + */ + void (*SetTransactionStatus) (TransactionId xid, int nsubxids, TransactionId *subxids, XidStatus status, XLogRecPtr lsn); + + /* + * Get current transaction snaphot (encapsulation of GetSnapshotData in + * procarray.c) + */ + Snapshot (*GetSnapshot) (Snapshot snapshot); + + /* + * Assign new Xid to transaction (encapsulation of GetNewTransactionId in + * varsup.c) + */ + TransactionId (*GetNewTransactionId) (bool isSubXact); + + /* + * Get oldest transaction Xid that was running when any current + * transaction was started (encapsulation of GetOldestXmin in procarray.c) + */ + TransactionId (*GetOldestXmin) (Relation rel, bool ignoreVacuum); + + /* + * Check if current transaction is not yet completed (encapsulation of + * TransactionIdIsInProgress in procarray.c) + */ + bool (*IsInProgress) (TransactionId xid); + + /* + * Get global transaction XID: returns XID of current transaction if it is + * global, InvalidTransactionId otherwise + */ + TransactionId (*GetGlobalTransactionId) (void); + + /* + * Is the given XID still-in-progress according to the snapshot + * (encapsulation of XidInMVCCSnapshot in tqual.c) + */ + bool (*IsInSnapshot) (TransactionId xid, Snapshot snapshot); + + /* Detect distributed deadlock */ + bool (*DetectGlobalDeadLock) (PGPROC *proc); + + char const *(*GetName) (void); +} TransactionManager; + +/* Get pointer to transaction manager: actually returns content of TM variable */ +TransactionManager *GetTransactionManager(void); + +extern TransactionManager *TM; /* Current transaction manager (can be + * substituted by extensions) */ +extern TransactionManager PgTM; /* Standard PostgreSQL transaction manager */ + +/* Standard PostgreSQL function implementing TM interface */ +extern bool PgXidInMVCCSnapshot(TransactionId xid, Snapshot snapshot); + +extern void PgTransactionIdSetTreeStatus(TransactionId xid, int nsubxids, + TransactionId *subxids, XidStatus status, XLogRecPtr lsn); +extern XidStatus PgTransactionIdGetStatus(TransactionId xid, XLogRecPtr *lsn); + +extern Snapshot PgGetSnapshotData(Snapshot snapshot); + +extern TransactionId PgGetOldestXmin(Relation rel, bool ignoreVacuum); + +extern bool PgTransactionIdIsInProgress(TransactionId xid); + +extern TransactionId PgGetGlobalTransactionId(void); + +extern TransactionId PgGetNewTransactionId(bool isSubXact); + +extern bool PgDetectGlobalDeadLock(PGPROC *proc); + +extern char const *PgGetTransactionManagerName(void); + +#endif diff --git a/src/include/storage/lock.h b/src/include/storage/lock.h index 6b4e365..0d45894 100644 --- a/src/include/storage/lock.h +++ b/src/include/storage/lock.h @@ -454,8 +454,9 @@ typedef enum DS_NO_DEADLOCK, /* no deadlock detected */ DS_SOFT_DEADLOCK, /* deadlock avoided by queue rearrangement */ DS_HARD_DEADLOCK, /* deadlock, no way out but ERROR */ - DS_BLOCKED_BY_AUTOVACUUM /* no deadlock; queue blocked by autovacuum + DS_BLOCKED_BY_AUTOVACUUM, /* no deadlock; queue blocked by autovacuum * worker */ + DS_DISTRIBUTED_DEADLOCK /* distributed deadlock detected by DTM */ } DeadLockState; /* @@ -547,6 +548,10 @@ extern void DumpLocks(PGPROC *proc); extern void DumpAllLocks(void); #endif +typedef void (*LockIterator) (PROCLOCK *lock, void *arg); + +extern void EnumerateLocks(LockIterator iterator, void *arg); + /* Lock a VXID (used to wait for a transaction to finish) */ extern void VirtualXactLockTableInsert(VirtualTransactionId vxid); extern void VirtualXactLockTableCleanup(void);