Skip to content

Commit 4f5849d

Browse files
knizhnikknizhnik
committed
Add entries to procarray
1 parent 89c8d4d commit 4f5849d

File tree

4 files changed

+140
-3
lines changed

4 files changed

+140
-3
lines changed

‎contrib/pg_xtm/pg_dtm.c

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -122,7 +122,7 @@ static void DtmSetTransactionStatus(TransactionId xid, int nsubxids, Transaction
122122
/* Already should be IN_PROGRESS */
123123
/* CLOGTransactionIdSetTreeStatus(xid, nsubxids, subxids, TRANSACTION_STATUS_IN_PROGRESS, lsn); */
124124
if (status == TRANSACTION_STATUS_COMMITTED) {
125-
ProcArrayAdd(&ProcGlobal->allProcs[MyProc->pgprocno]);
125+
ProcArrayAdd(&ProcGlobal->allProcs[AllocGXid(xid)]);
126126
}
127127
DtmHasSnapshot = false;
128128
DtmEnsureConnection();

‎src/backend/access/transam/twophase.c

Lines changed: 133 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -576,6 +576,139 @@ RemoveGXact(GlobalTransaction gxact)
576576
elog(ERROR, "failed to find %p in GlobalTransaction array", gxact);
577577
}
578578

579+
int
580+
AllocGXid(TransactionId xid)
581+
{
582+
GlobalTransaction gxact;
583+
PGPROC *proc;
584+
PGXACT *pgxact;
585+
int i;
586+
char gid[GIDSIZE];
587+
588+
sprintf(gid, "%u", xid);
589+
590+
/* fail immediately if feature is disabled */
591+
if (max_prepared_xacts == 0)
592+
ereport(ERROR,
593+
(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
594+
errmsg("prepared transactions are disabled"),
595+
errhint("Set max_prepared_transactions to a nonzero value.")));
596+
597+
/* on first call, register the exit hook */
598+
if (!twophaseExitRegistered)
599+
{
600+
before_shmem_exit(AtProcExit_Twophase, 0);
601+
twophaseExitRegistered = true;
602+
}
603+
604+
LWLockAcquire(TwoPhaseStateLock, LW_EXCLUSIVE);
605+
606+
/* Check for conflicting GID */
607+
for (i = 0; i < TwoPhaseState->numPrepXacts; i++)
608+
{
609+
gxact = TwoPhaseState->prepXacts[i];
610+
if (strcmp(gxact->gid, gid) == 0)
611+
{
612+
ereport(ERROR,
613+
(errcode(ERRCODE_DUPLICATE_OBJECT),
614+
errmsg("transaction identifier \"%s\" is already in use",
615+
gid)));
616+
}
617+
}
618+
619+
/* Get a free gxact from the freelist */
620+
if (TwoPhaseState->freeGXacts == NULL)
621+
ereport(ERROR,
622+
(errcode(ERRCODE_OUT_OF_MEMORY),
623+
errmsg("maximum number of prepared transactions reached"),
624+
errhint("Increase max_prepared_transactions (currently %d).",
625+
max_prepared_xacts)));
626+
gxact = TwoPhaseState->freeGXacts;
627+
TwoPhaseState->freeGXacts = gxact->next;
628+
629+
proc = &ProcGlobal->allProcs[gxact->pgprocno];
630+
pgxact = &ProcGlobal->allPgXact[gxact->pgprocno];
631+
632+
/* Initialize the PGPROC entry */
633+
MemSet(proc, 0, sizeof(PGPROC));
634+
proc->pgprocno = gxact->pgprocno;
635+
SHMQueueElemInit(&(proc->links));
636+
proc->waitStatus = STATUS_OK;
637+
/* We set up the gxact's VXID as InvalidBackendId/XID */
638+
proc->lxid = (LocalTransactionId) xid;
639+
pgxact->xid = xid;
640+
pgxact->xmin = InvalidTransactionId;
641+
pgxact->delayChkpt = false;
642+
pgxact->vacuumFlags = 0;
643+
proc->pid = 0;
644+
proc->backendId = InvalidBackendId;
645+
proc->databaseId = 0;
646+
proc->roleId = 0;
647+
proc->lwWaiting = false;
648+
proc->lwWaitMode = 0;
649+
proc->waitLock = NULL;
650+
proc->waitProcLock = NULL;
651+
for (i = 0; i < NUM_LOCK_PARTITIONS; i++)
652+
SHMQueueInit(&(proc->myProcLocks[i]));
653+
/* subxid data must be filled later by GXactLoadSubxactData */
654+
pgxact->overflowed = false;
655+
pgxact->nxids = 0;
656+
657+
gxact->prepared_at = 0;
658+
/* initialize LSN to 0 (start of WAL) */
659+
gxact->prepare_lsn = 0;
660+
gxact->owner = 0;
661+
gxact->locking_backend = MyBackendId;
662+
gxact->valid = false;
663+
strcpy(gxact->gid, gid);
664+
665+
/* And insert it into the active array */
666+
Assert(TwoPhaseState->numPrepXacts < max_prepared_xacts);
667+
TwoPhaseState->prepXacts[TwoPhaseState->numPrepXacts++] = gxact;
668+
669+
/*
670+
* Remember that we have this GlobalTransaction entry locked for us. If we
671+
* abort after this, we must release it.
672+
*/
673+
MyLockedGxact = gxact;
674+
675+
LWLockRelease(TwoPhaseStateLock);
676+
677+
return gxact->pgprocno;
678+
}
679+
680+
void
681+
RemoveGXid(TransactionId xid)
682+
{
683+
int i;
684+
char gid[GIDSIZE];
685+
686+
sprintf(gid, "%u", xid);
687+
688+
LWLockAcquire(TwoPhaseStateLock, LW_EXCLUSIVE);
689+
690+
for (i = 0; i < TwoPhaseState->numPrepXacts; i++)
691+
{
692+
if (strcmp(TwoPhaseState->prepXacts[i]->gid, gid) == 0)
693+
{
694+
/* remove from the active array */
695+
TwoPhaseState->numPrepXacts--;
696+
TwoPhaseState->prepXacts[i] = TwoPhaseState->prepXacts[TwoPhaseState->numPrepXacts];
697+
698+
/* and put it back in the freelist */
699+
TwoPhaseState->prepXacts[i]->next = TwoPhaseState->freeGXacts;
700+
TwoPhaseState->freeGXacts = TwoPhaseState->prepXacts[i];
701+
702+
LWLockRelease(TwoPhaseStateLock);
703+
704+
return;
705+
}
706+
}
707+
708+
LWLockRelease(TwoPhaseStateLock);
709+
710+
}
711+
579712
/*
580713
* TransactionIdIsPrepared
581714
* True iff transaction associated with the identifier is prepared

‎src/backend/storage/ipc/procarray.c

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -3902,16 +3902,17 @@ void VacuumProcArray(Snapshot snapshot)
39023902
continue;
39033903
}
39043904
if (TransactionIsStillInProgress(pxid, snapshot)) {
3905-
elog(WARNING, "ProcArray: %d is in progress\n", pxid);
3905+
//elog(WARNING, "procArray[%d]=%d is in progress\n", i, pxid);
39063906
nInProgress += 1;
39073907
continue;
39083908
}
3909+
RemoveGXid(pxid);
39093910
nCompleted += 1;
39103911
memmove(&arrayP->pgprocnos[i], &arrayP->pgprocnos[i + 1],
39113912
(arrayP->numProcs - i - 1) * sizeof(int));
39123913
arrayP->pgprocnos[arrayP->numProcs - 1] = -1; /* for debugging */
39133914
arrayP->numProcs--;
39143915
}
39153916
LWLockRelease(ProcArrayLock);
3916-
elog(WARNING, "ProcArray: %d in progress, %d completed, %d total\n", nInProgress, nCompleted, arrayP->numProcs);
3917+
elog(WARNING, "VacuumProcArray: %d in progress, %d completed, %d total\n", nInProgress, nCompleted, arrayP->numProcs);
39173918
}

‎src/include/access/twophase.h

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -58,4 +58,7 @@ extern void FinishPreparedTransaction(const char *gid, bool isCommit);
5858

5959
extern const char* GetLockedGlobalTransactionId(void);
6060

61+
extern int AllocGXid(TransactionId xid);
62+
extern void RemoveGXid(TransactionId xid);
63+
6164
#endif /* TWOPHASE_H */

0 commit comments

Comments
 (0)