Logo Search packages:      
Sourcecode: postgresql-8.4 version File versions

twophase.c

/*-------------------------------------------------------------------------
 *
 * twophase.c
 *          Two-phase commit support functions.
 *
 * Portions Copyright (c) 1996-2009, PostgreSQL Global Development Group
 * Portions Copyright (c) 1994, Regents of the University of California
 *
 * IDENTIFICATION
 *          $PostgreSQL: pgsql/src/backend/access/transam/twophase.c,v 1.52 2009/04/23 00:23:45 tgl Exp $
 *
 * NOTES
 *          Each global transaction is associated with a global transaction
 *          identifier (GID). The client assigns a GID to a postgres
 *          transaction with the PREPARE TRANSACTION command.
 *
 *          We keep all active global transactions in a shared memory array.
 *          When the PREPARE TRANSACTION command is issued, the GID is
 *          reserved for the transaction in the array. This is done before
 *          a WAL entry is made, because the reservation checks for duplicate
 *          GIDs and aborts the transaction if there already is a global
 *          transaction in prepared state with the same GID.
 *
 *          A global transaction (gxact) also has a dummy PGPROC that is entered
 *          into the ProcArray array; this is what keeps the XID considered
 *          running by TransactionIdIsInProgress.  It is also convenient as a
 *          PGPROC to hook the gxact's locks to.
 *
 *          In order to survive crashes and shutdowns, all prepared
 *          transactions must be stored in permanent storage. This includes
 *          locking information, pending notifications etc. All that state
 *          information is written to the per-transaction state file in
 *          the pg_twophase directory.
 *
 *-------------------------------------------------------------------------
 */
#include "postgres.h"

#include <fcntl.h>
#include <sys/stat.h>
#include <sys/types.h>
#include <time.h>
#include <unistd.h>

#include "access/htup.h"
#include "access/subtrans.h"
#include "access/transam.h"
#include "access/twophase.h"
#include "access/twophase_rmgr.h"
#include "access/xact.h"
#include "access/xlogutils.h"
#include "catalog/pg_type.h"
#include "catalog/storage.h"
#include "funcapi.h"
#include "miscadmin.h"
#include "pg_trace.h"
#include "pgstat.h"
#include "storage/fd.h"
#include "storage/procarray.h"
#include "storage/smgr.h"
#include "utils/builtins.h"
#include "utils/memutils.h"


/*
 * Directory where Two-phase commit files reside within PGDATA
 */
#define TWOPHASE_DIR "pg_twophase"

/* GUC variable, can't be changed after startup */
int               max_prepared_xacts = 0;

/*
 * This struct describes one global transaction that is in prepared state
 * or attempting to become prepared.
 *
 * The first component of the struct is a dummy PGPROC that is inserted
 * into the global ProcArray so that the transaction appears to still be
 * running and holding locks.  It must be first because we cast pointers
 * to PGPROC and pointers to GlobalTransactionData back and forth.
 *
 * The lifecycle of a global transaction is:
 *
 * 1. After checking that the requested GID is not in use, set up an
 * entry in the TwoPhaseState->prepXacts array with the correct XID and GID,
 * with locking_xid = my own XID and valid = false.
 *
 * 2. After successfully completing prepare, set valid = true and enter the
 * contained PGPROC into the global ProcArray.
 *
 * 3. To begin COMMIT PREPARED or ROLLBACK PREPARED, check that the entry
 * is valid and its locking_xid is no longer active, then store my current
 * XID into locking_xid.  This prevents concurrent attempts to commit or
 * rollback the same prepared xact.
 *
 * 4. On completion of COMMIT PREPARED or ROLLBACK PREPARED, remove the entry
 * from the ProcArray and the TwoPhaseState->prepXacts array and return it to
 * the freelist.
 *
 * Note that if the preparing transaction fails between steps 1 and 2, the
 * entry will remain in prepXacts until recycled.  We can detect recyclable
 * entries by checking for valid = false and locking_xid no longer active.
 *
 * typedef struct GlobalTransactionData *GlobalTransaction appears in
 * twophase.h
 */
#define GIDSIZE 200

typedef struct GlobalTransactionData
{
      PGPROC            proc;             /* dummy proc */
      TimestampTz prepared_at;      /* time of preparation */
      XLogRecPtr  prepare_lsn;      /* XLOG offset of prepare record */
      Oid               owner;                  /* ID of user that executed the xact */
      TransactionId locking_xid;    /* top-level XID of backend working on xact */
      bool        valid;                  /* TRUE if fully prepared */
      char        gid[GIDSIZE];     /* The GID assigned to the prepared xact */
} GlobalTransactionData;

/*
 * Two Phase Commit shared state.  Access to this struct is protected
 * by TwoPhaseStateLock.
 */
typedef struct TwoPhaseStateData
{
      /* Head of linked list of free GlobalTransactionData structs */
      GlobalTransaction freeGXacts;

      /* Number of valid prepXacts entries. */
      int               numPrepXacts;

      /*
       * There are max_prepared_xacts items in this array, but C wants a
       * fixed-size array.
       */
      GlobalTransaction prepXacts[1];           /* VARIABLE LENGTH ARRAY */
} TwoPhaseStateData;                /* VARIABLE LENGTH STRUCT */

static TwoPhaseStateData *TwoPhaseState;


static void RecordTransactionCommitPrepared(TransactionId xid,
                                                int nchildren,
                                                TransactionId *children,
                                                int nrels,
                                                RelFileNode *rels);
static void RecordTransactionAbortPrepared(TransactionId xid,
                                             int nchildren,
                                             TransactionId *children,
                                             int nrels,
                                             RelFileNode *rels);
static void ProcessRecords(char *bufptr, TransactionId xid,
                     const TwoPhaseCallback callbacks[]);


/*
 * Initialization of shared memory
 */
Size
TwoPhaseShmemSize(void)
{
      Size        size;

      /* Need the fixed struct, the array of pointers, and the GTD structs */
      size = offsetof(TwoPhaseStateData, prepXacts);
      size = add_size(size, mul_size(max_prepared_xacts,
                                                   sizeof(GlobalTransaction)));
      size = MAXALIGN(size);
      size = add_size(size, mul_size(max_prepared_xacts,
                                                   sizeof(GlobalTransactionData)));

      return size;
}

void
TwoPhaseShmemInit(void)
{
      bool        found;

      TwoPhaseState = ShmemInitStruct("Prepared Transaction Table",
                                                      TwoPhaseShmemSize(),
                                                      &found);
      if (!IsUnderPostmaster)
      {
            GlobalTransaction gxacts;
            int               i;

            Assert(!found);
            TwoPhaseState->freeGXacts = NULL;
            TwoPhaseState->numPrepXacts = 0;

            /*
             * Initialize the linked list of free GlobalTransactionData structs
             */
            gxacts = (GlobalTransaction)
                  ((char *) TwoPhaseState +
                   MAXALIGN(offsetof(TwoPhaseStateData, prepXacts) +
                                sizeof(GlobalTransaction) * max_prepared_xacts));
            for (i = 0; i < max_prepared_xacts; i++)
            {
                  gxacts[i].proc.links.next = (SHM_QUEUE *) TwoPhaseState->freeGXacts;
                  TwoPhaseState->freeGXacts = &gxacts[i];
            }
      }
      else
            Assert(found);
}


/*
 * MarkAsPreparing
 *          Reserve the GID for the given transaction.
 *
 * Internally, this creates a gxact struct and puts it into the active array.
 * NOTE: this is also used when reloading a gxact after a crash; so avoid
 * assuming that we can use very much backend context.
 */
GlobalTransaction
MarkAsPreparing(TransactionId xid, const char *gid,
                        TimestampTz prepared_at, Oid owner, Oid databaseid)
{
      GlobalTransaction gxact;
      int               i;

      if (strlen(gid) >= GIDSIZE)
            ereport(ERROR,
                        (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
                         errmsg("transaction identifier \"%s\" is too long",
                                    gid)));

      /* fail immediately if feature is disabled */
      if (max_prepared_xacts == 0)
            ereport(ERROR,
                        (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
                         errmsg("prepared transactions are disabled"),
                         errhint("Set max_prepared_transactions to a nonzero value.")));

      LWLockAcquire(TwoPhaseStateLock, LW_EXCLUSIVE);

      /*
       * First, find and recycle any gxacts that failed during prepare. We do
       * this partly to ensure we don't mistakenly say their GIDs are still
       * reserved, and partly so we don't fail on out-of-slots unnecessarily.
       */
      for (i = 0; i < TwoPhaseState->numPrepXacts; i++)
      {
            gxact = TwoPhaseState->prepXacts[i];
            if (!gxact->valid && !TransactionIdIsActive(gxact->locking_xid))
            {
                  /* It's dead Jim ... remove from the active array */
                  TwoPhaseState->numPrepXacts--;
                  TwoPhaseState->prepXacts[i] = TwoPhaseState->prepXacts[TwoPhaseState->numPrepXacts];
                  /* and put it back in the freelist */
                  gxact->proc.links.next = (SHM_QUEUE *) TwoPhaseState->freeGXacts;
                  TwoPhaseState->freeGXacts = gxact;
                  /* Back up index count too, so we don't miss scanning one */
                  i--;
            }
      }

      /* Check for conflicting GID */
      for (i = 0; i < TwoPhaseState->numPrepXacts; i++)
      {
            gxact = TwoPhaseState->prepXacts[i];
            if (strcmp(gxact->gid, gid) == 0)
            {
                  ereport(ERROR,
                              (errcode(ERRCODE_DUPLICATE_OBJECT),
                               errmsg("transaction identifier \"%s\" is already in use",
                                          gid)));
            }
      }

      /* Get a free gxact from the freelist */
      if (TwoPhaseState->freeGXacts == NULL)
            ereport(ERROR,
                        (errcode(ERRCODE_OUT_OF_MEMORY),
                         errmsg("maximum number of prepared transactions reached"),
                         errhint("Increase max_prepared_transactions (currently %d).",
                                     max_prepared_xacts)));
      gxact = TwoPhaseState->freeGXacts;
      TwoPhaseState->freeGXacts = (GlobalTransaction) gxact->proc.links.next;

      /* Initialize it */
      MemSet(&gxact->proc, 0, sizeof(PGPROC));
      SHMQueueElemInit(&(gxact->proc.links));
      gxact->proc.waitStatus = STATUS_OK;
      /* We set up the gxact's VXID as InvalidBackendId/XID */
      gxact->proc.lxid = (LocalTransactionId) xid;
      gxact->proc.xid = xid;
      gxact->proc.xmin = InvalidTransactionId;
      gxact->proc.pid = 0;
      gxact->proc.backendId = InvalidBackendId;
      gxact->proc.databaseId = databaseid;
      gxact->proc.roleId = owner;
      gxact->proc.inCommit = false;
      gxact->proc.vacuumFlags = 0;
      gxact->proc.lwWaiting = false;
      gxact->proc.lwExclusive = false;
      gxact->proc.lwWaitLink = NULL;
      gxact->proc.waitLock = NULL;
      gxact->proc.waitProcLock = NULL;
      for (i = 0; i < NUM_LOCK_PARTITIONS; i++)
            SHMQueueInit(&(gxact->proc.myProcLocks[i]));
      /* subxid data must be filled later by GXactLoadSubxactData */
      gxact->proc.subxids.overflowed = false;
      gxact->proc.subxids.nxids = 0;

      gxact->prepared_at = prepared_at;
      /* initialize LSN to 0 (start of WAL) */
      gxact->prepare_lsn.xlogid = 0;
      gxact->prepare_lsn.xrecoff = 0;
      gxact->owner = owner;
      gxact->locking_xid = xid;
      gxact->valid = false;
      strcpy(gxact->gid, gid);

      /* And insert it into the active array */
      Assert(TwoPhaseState->numPrepXacts < max_prepared_xacts);
      TwoPhaseState->prepXacts[TwoPhaseState->numPrepXacts++] = gxact;

      LWLockRelease(TwoPhaseStateLock);

      return gxact;
}

/*
 * GXactLoadSubxactData
 *
 * If the transaction being persisted had any subtransactions, this must
 * be called before MarkAsPrepared() to load information into the dummy
 * PGPROC.
 */
static void
GXactLoadSubxactData(GlobalTransaction gxact, int nsubxacts,
                               TransactionId *children)
{
      /* We need no extra lock since the GXACT isn't valid yet */
      if (nsubxacts > PGPROC_MAX_CACHED_SUBXIDS)
      {
            gxact->proc.subxids.overflowed = true;
            nsubxacts = PGPROC_MAX_CACHED_SUBXIDS;
      }
      if (nsubxacts > 0)
      {
            memcpy(gxact->proc.subxids.xids, children,
                     nsubxacts * sizeof(TransactionId));
            gxact->proc.subxids.nxids = nsubxacts;
      }
}

/*
 * MarkAsPrepared
 *          Mark the GXACT as fully valid, and enter it into the global ProcArray.
 */
static void
MarkAsPrepared(GlobalTransaction gxact)
{
      /* Lock here may be overkill, but I'm not convinced of that ... */
      LWLockAcquire(TwoPhaseStateLock, LW_EXCLUSIVE);
      Assert(!gxact->valid);
      gxact->valid = true;
      LWLockRelease(TwoPhaseStateLock);

      /*
       * Put it into the global ProcArray so TransactionIdIsInProgress considers
       * the XID as still running.
       */
      ProcArrayAdd(&gxact->proc);
}

/*
 * LockGXact
 *          Locate the prepared transaction and mark it busy for COMMIT or PREPARE.
 */
static GlobalTransaction
LockGXact(const char *gid, Oid user)
{
      int               i;

      LWLockAcquire(TwoPhaseStateLock, LW_EXCLUSIVE);

      for (i = 0; i < TwoPhaseState->numPrepXacts; i++)
      {
            GlobalTransaction gxact = TwoPhaseState->prepXacts[i];

            /* Ignore not-yet-valid GIDs */
            if (!gxact->valid)
                  continue;
            if (strcmp(gxact->gid, gid) != 0)
                  continue;

            /* Found it, but has someone else got it locked? */
            if (TransactionIdIsValid(gxact->locking_xid))
            {
                  if (TransactionIdIsActive(gxact->locking_xid))
                        ereport(ERROR,
                                    (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
                        errmsg("prepared transaction with identifier \"%s\" is busy",
                                 gid)));
                  gxact->locking_xid = InvalidTransactionId;
            }

            if (user != gxact->owner && !superuser_arg(user))
                  ereport(ERROR,
                              (errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
                          errmsg("permission denied to finish prepared transaction"),
                               errhint("Must be superuser or the user that prepared the transaction.")));

            /*
             * Note: it probably would be possible to allow committing from
             * another database; but at the moment NOTIFY is known not to work and
             * there may be some other issues as well.      Hence disallow until
             * someone gets motivated to make it work.
             */
            if (MyDatabaseId != gxact->proc.databaseId)
                  ereport(ERROR,
                              (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
                          errmsg("prepared transaction belongs to another database"),
                               errhint("Connect to the database where the transaction was prepared to finish it.")));

            /* OK for me to lock it */
            gxact->locking_xid = GetTopTransactionId();

            LWLockRelease(TwoPhaseStateLock);

            return gxact;
      }

      LWLockRelease(TwoPhaseStateLock);

      ereport(ERROR,
                  (errcode(ERRCODE_UNDEFINED_OBJECT),
             errmsg("prepared transaction with identifier \"%s\" does not exist",
                        gid)));

      /* NOTREACHED */
      return NULL;
}

/*
 * RemoveGXact
 *          Remove the prepared transaction from the shared memory array.
 *
 * NB: caller should have already removed it from ProcArray
 */
static void
RemoveGXact(GlobalTransaction gxact)
{
      int               i;

      LWLockAcquire(TwoPhaseStateLock, LW_EXCLUSIVE);

      for (i = 0; i < TwoPhaseState->numPrepXacts; i++)
      {
            if (gxact == TwoPhaseState->prepXacts[i])
            {
                  /* remove from the active array */
                  TwoPhaseState->numPrepXacts--;
                  TwoPhaseState->prepXacts[i] = TwoPhaseState->prepXacts[TwoPhaseState->numPrepXacts];

                  /* and put it back in the freelist */
                  gxact->proc.links.next = (SHM_QUEUE *) TwoPhaseState->freeGXacts;
                  TwoPhaseState->freeGXacts = gxact;

                  LWLockRelease(TwoPhaseStateLock);

                  return;
            }
      }

      LWLockRelease(TwoPhaseStateLock);

      elog(ERROR, "failed to find %p in GlobalTransaction array", gxact);
}

/*
 * TransactionIdIsPrepared
 *          True iff transaction associated with the identifier is prepared
 *          for two-phase commit
 *
 * Note: only gxacts marked "valid" are considered; but notice we do not
 * check the locking status.
 *
 * This is not currently exported, because it is only needed internally.
 */
static bool
TransactionIdIsPrepared(TransactionId xid)
{
      bool        result = false;
      int               i;

      LWLockAcquire(TwoPhaseStateLock, LW_SHARED);

      for (i = 0; i < TwoPhaseState->numPrepXacts; i++)
      {
            GlobalTransaction gxact = TwoPhaseState->prepXacts[i];

            if (gxact->valid && gxact->proc.xid == xid)
            {
                  result = true;
                  break;
            }
      }

      LWLockRelease(TwoPhaseStateLock);

      return result;
}

/*
 * Returns an array of all prepared transactions for the user-level
 * function pg_prepared_xact.
 *
 * The returned array and all its elements are copies of internal data
 * structures, to minimize the time we need to hold the TwoPhaseStateLock.
 *
 * WARNING -- we return even those transactions that are not fully prepared
 * yet.  The caller should filter them out if he doesn't want them.
 *
 * The returned array is palloc'd.
 */
static int
GetPreparedTransactionList(GlobalTransaction *gxacts)
{
      GlobalTransaction array;
      int               num;
      int               i;

      LWLockAcquire(TwoPhaseStateLock, LW_SHARED);

      if (TwoPhaseState->numPrepXacts == 0)
      {
            LWLockRelease(TwoPhaseStateLock);

            *gxacts = NULL;
            return 0;
      }

      num = TwoPhaseState->numPrepXacts;
      array = (GlobalTransaction) palloc(sizeof(GlobalTransactionData) * num);
      *gxacts = array;
      for (i = 0; i < num; i++)
            memcpy(array + i, TwoPhaseState->prepXacts[i],
                     sizeof(GlobalTransactionData));

      LWLockRelease(TwoPhaseStateLock);

      return num;
}


/* Working status for pg_prepared_xact */
typedef struct
{
      GlobalTransaction array;
      int               ngxacts;
      int               currIdx;
} Working_State;

/*
 * pg_prepared_xact
 *          Produce a view with one row per prepared transaction.
 *
 * This function is here so we don't have to export the
 * GlobalTransactionData struct definition.
 */
Datum
pg_prepared_xact(PG_FUNCTION_ARGS)
{
      FuncCallContext *funcctx;
      Working_State *status;

      if (SRF_IS_FIRSTCALL())
      {
            TupleDesc   tupdesc;
            MemoryContext oldcontext;

            /* create a function context for cross-call persistence */
            funcctx = SRF_FIRSTCALL_INIT();

            /*
             * Switch to memory context appropriate for multiple function calls
             */
            oldcontext = MemoryContextSwitchTo(funcctx->multi_call_memory_ctx);

            /* build tupdesc for result tuples */
            /* this had better match pg_prepared_xacts view in system_views.sql */
            tupdesc = CreateTemplateTupleDesc(5, false);
            TupleDescInitEntry(tupdesc, (AttrNumber) 1, "transaction",
                                       XIDOID, -1, 0);
            TupleDescInitEntry(tupdesc, (AttrNumber) 2, "gid",
                                       TEXTOID, -1, 0);
            TupleDescInitEntry(tupdesc, (AttrNumber) 3, "prepared",
                                       TIMESTAMPTZOID, -1, 0);
            TupleDescInitEntry(tupdesc, (AttrNumber) 4, "ownerid",
                                       OIDOID, -1, 0);
            TupleDescInitEntry(tupdesc, (AttrNumber) 5, "dbid",
                                       OIDOID, -1, 0);

            funcctx->tuple_desc = BlessTupleDesc(tupdesc);

            /*
             * Collect all the 2PC status information that we will format and send
             * out as a result set.
             */
            status = (Working_State *) palloc(sizeof(Working_State));
            funcctx->user_fctx = (void *) status;

            status->ngxacts = GetPreparedTransactionList(&status->array);
            status->currIdx = 0;

            MemoryContextSwitchTo(oldcontext);
      }

      funcctx = SRF_PERCALL_SETUP();
      status = (Working_State *) funcctx->user_fctx;

      while (status->array != NULL && status->currIdx < status->ngxacts)
      {
            GlobalTransaction gxact = &status->array[status->currIdx++];
            Datum       values[5];
            bool        nulls[5];
            HeapTuple   tuple;
            Datum       result;

            if (!gxact->valid)
                  continue;

            /*
             * Form tuple with appropriate data.
             */
            MemSet(values, 0, sizeof(values));
            MemSet(nulls, 0, sizeof(nulls));

            values[0] = TransactionIdGetDatum(gxact->proc.xid);
            values[1] = CStringGetTextDatum(gxact->gid);
            values[2] = TimestampTzGetDatum(gxact->prepared_at);
            values[3] = ObjectIdGetDatum(gxact->owner);
            values[4] = ObjectIdGetDatum(gxact->proc.databaseId);

            tuple = heap_form_tuple(funcctx->tuple_desc, values, nulls);
            result = HeapTupleGetDatum(tuple);
            SRF_RETURN_NEXT(funcctx, result);
      }

      SRF_RETURN_DONE(funcctx);
}

/*
 * TwoPhaseGetDummyProc
 *          Get the PGPROC that represents a prepared transaction specified by XID
 */
PGPROC *
TwoPhaseGetDummyProc(TransactionId xid)
{
      PGPROC         *result = NULL;
      int               i;

      static TransactionId cached_xid = InvalidTransactionId;
      static PGPROC *cached_proc = NULL;

      /*
       * During a recovery, COMMIT PREPARED, or ABORT PREPARED, we'll be called
       * repeatedly for the same XID.  We can save work with a simple cache.
       */
      if (xid == cached_xid)
            return cached_proc;

      LWLockAcquire(TwoPhaseStateLock, LW_SHARED);

      for (i = 0; i < TwoPhaseState->numPrepXacts; i++)
      {
            GlobalTransaction gxact = TwoPhaseState->prepXacts[i];

            if (gxact->proc.xid == xid)
            {
                  result = &gxact->proc;
                  break;
            }
      }

      LWLockRelease(TwoPhaseStateLock);

      if (result == NULL)                 /* should not happen */
            elog(ERROR, "failed to find dummy PGPROC for xid %u", xid);

      cached_xid = xid;
      cached_proc = result;

      return result;
}

/************************************************************************/
/* State file support                                                                           */
/************************************************************************/

#define TwoPhaseFilePath(path, xid) \
      snprintf(path, MAXPGPATH, TWOPHASE_DIR "/%08X", xid)

/*
 * 2PC state file format:
 *
 *    1. TwoPhaseFileHeader
 *    2. TransactionId[] (subtransactions)
 *    3. RelFileNode[] (files to be deleted at commit)
 *    4. RelFileNode[] (files to be deleted at abort)
 *    5. TwoPhaseRecordOnDisk
 *    6. ...
 *    7. TwoPhaseRecordOnDisk (end sentinel, rmid == TWOPHASE_RM_END_ID)
 *    8. CRC32
 *
 * Each segment except the final CRC32 is MAXALIGN'd.
 */

/*
 * Header for a 2PC state file
 */
#define TWOPHASE_MAGIC  0x57F94531        /* format identifier */

typedef struct TwoPhaseFileHeader
{
      uint32            magic;                  /* format identifier */
      uint32            total_len;        /* actual file length */
      TransactionId xid;                  /* original transaction XID */
      Oid               database;         /* OID of database it was in */
      TimestampTz prepared_at;      /* time of preparation */
      Oid               owner;                  /* user running the transaction */
      int32       nsubxacts;        /* number of following subxact XIDs */
      int32       ncommitrels;      /* number of delete-on-commit rels */
      int32       nabortrels;       /* number of delete-on-abort rels */
      char        gid[GIDSIZE];     /* GID for transaction */
} TwoPhaseFileHeader;

/*
 * Header for each record in a state file
 *
 * NOTE: len counts only the rmgr data, not the TwoPhaseRecordOnDisk header.
 * The rmgr data will be stored starting on a MAXALIGN boundary.
 */
typedef struct TwoPhaseRecordOnDisk
{
      uint32            len;              /* length of rmgr data */
      TwoPhaseRmgrId rmid;          /* resource manager for this record */
      uint16            info;             /* flag bits for use by rmgr */
} TwoPhaseRecordOnDisk;

/*
 * During prepare, the state file is assembled in memory before writing it
 * to WAL and the actual state file.  We use a chain of XLogRecData blocks
 * so that we will be able to pass the state file contents directly to
 * XLogInsert.
 */
static struct xllist
{
      XLogRecData *head;                  /* first data block in the chain */
      XLogRecData *tail;                  /* last block in chain */
      uint32            bytes_free;       /* free bytes left in tail block */
      uint32            total_len;        /* total data bytes in chain */
}     records;


/*
 * Append a block of data to records data structure.
 *
 * NB: each block is padded to a MAXALIGN multiple.  This must be
 * accounted for when the file is later read!
 *
 * The data is copied, so the caller is free to modify it afterwards.
 */
static void
save_state_data(const void *data, uint32 len)
{
      uint32            padlen = MAXALIGN(len);

      if (padlen > records.bytes_free)
      {
            records.tail->next = palloc0(sizeof(XLogRecData));
            records.tail = records.tail->next;
            records.tail->buffer = InvalidBuffer;
            records.tail->len = 0;
            records.tail->next = NULL;

            records.bytes_free = Max(padlen, 512);
            records.tail->data = palloc(records.bytes_free);
      }

      memcpy(((char *) records.tail->data) + records.tail->len, data, len);
      records.tail->len += padlen;
      records.bytes_free -= padlen;
      records.total_len += padlen;
}

/*
 * Start preparing a state file.
 *
 * Initializes data structure and inserts the 2PC file header record.
 */
void
StartPrepare(GlobalTransaction gxact)
{
      TransactionId xid = gxact->proc.xid;
      TwoPhaseFileHeader hdr;
      TransactionId *children;
      RelFileNode *commitrels;
      RelFileNode *abortrels;

      /* Initialize linked list */
      records.head = palloc0(sizeof(XLogRecData));
      records.head->buffer = InvalidBuffer;
      records.head->len = 0;
      records.head->next = NULL;

      records.bytes_free = Max(sizeof(TwoPhaseFileHeader), 512);
      records.head->data = palloc(records.bytes_free);

      records.tail = records.head;

      records.total_len = 0;

      /* Create header */
      hdr.magic = TWOPHASE_MAGIC;
      hdr.total_len = 0;                  /* EndPrepare will fill this in */
      hdr.xid = xid;
      hdr.database = gxact->proc.databaseId;
      hdr.prepared_at = gxact->prepared_at;
      hdr.owner = gxact->owner;
      hdr.nsubxacts = xactGetCommittedChildren(&children);
      hdr.ncommitrels = smgrGetPendingDeletes(true, &commitrels, NULL);
      hdr.nabortrels = smgrGetPendingDeletes(false, &abortrels, NULL);
      StrNCpy(hdr.gid, gxact->gid, GIDSIZE);

      save_state_data(&hdr, sizeof(TwoPhaseFileHeader));

      /* Add the additional info about subxacts and deletable files */
      if (hdr.nsubxacts > 0)
      {
            save_state_data(children, hdr.nsubxacts * sizeof(TransactionId));
            /* While we have the child-xact data, stuff it in the gxact too */
            GXactLoadSubxactData(gxact, hdr.nsubxacts, children);
      }
      if (hdr.ncommitrels > 0)
      {
            save_state_data(commitrels, hdr.ncommitrels * sizeof(RelFileNode));
            pfree(commitrels);
      }
      if (hdr.nabortrels > 0)
      {
            save_state_data(abortrels, hdr.nabortrels * sizeof(RelFileNode));
            pfree(abortrels);
      }
}

/*
 * Finish preparing state file.
 *
 * Calculates CRC and writes state file to WAL and in pg_twophase directory.
 */
void
EndPrepare(GlobalTransaction gxact)
{
      TransactionId xid = gxact->proc.xid;
      TwoPhaseFileHeader *hdr;
      char        path[MAXPGPATH];
      XLogRecData *record;
      pg_crc32    statefile_crc;
      pg_crc32    bogus_crc;
      int               fd;

      /* Add the end sentinel to the list of 2PC records */
      RegisterTwoPhaseRecord(TWOPHASE_RM_END_ID, 0,
                                       NULL, 0);

      /* Go back and fill in total_len in the file header record */
      hdr = (TwoPhaseFileHeader *) records.head->data;
      Assert(hdr->magic == TWOPHASE_MAGIC);
      hdr->total_len = records.total_len + sizeof(pg_crc32);

      /*
       * If the file size exceeds MaxAllocSize, we won't be able to read it in
       * ReadTwoPhaseFile. Check for that now, rather than fail at commit time.
       */
      if (hdr->total_len > MaxAllocSize)
            ereport(ERROR,
                        (errcode(ERRCODE_PROGRAM_LIMIT_EXCEEDED),
                         errmsg("two-phase state file maximum length exceeded")));

      /*
       * Create the 2PC state file.
       *
       * Note: because we use BasicOpenFile(), we are responsible for ensuring
       * the FD gets closed in any error exit path.  Once we get into the
       * critical section, though, it doesn't matter since any failure causes
       * PANIC anyway.
       */
      TwoPhaseFilePath(path, xid);

      fd = BasicOpenFile(path,
                                 O_CREAT | O_EXCL | O_WRONLY | PG_BINARY,
                                 S_IRUSR | S_IWUSR);
      if (fd < 0)
            ereport(ERROR,
                        (errcode_for_file_access(),
                         errmsg("could not create two-phase state file \"%s\": %m",
                                    path)));

      /* Write data to file, and calculate CRC as we pass over it */
      INIT_CRC32(statefile_crc);

      for (record = records.head; record != NULL; record = record->next)
      {
            COMP_CRC32(statefile_crc, record->data, record->len);
            if ((write(fd, record->data, record->len)) != record->len)
            {
                  close(fd);
                  ereport(ERROR,
                              (errcode_for_file_access(),
                               errmsg("could not write two-phase state file: %m")));
            }
      }

      FIN_CRC32(statefile_crc);

      /*
       * Write a deliberately bogus CRC to the state file; this is just paranoia
       * to catch the case where four more bytes will run us out of disk space.
       */
      bogus_crc = ~statefile_crc;

      if ((write(fd, &bogus_crc, sizeof(pg_crc32))) != sizeof(pg_crc32))
      {
            close(fd);
            ereport(ERROR,
                        (errcode_for_file_access(),
                         errmsg("could not write two-phase state file: %m")));
      }

      /* Back up to prepare for rewriting the CRC */
      if (lseek(fd, -((off_t) sizeof(pg_crc32)), SEEK_CUR) < 0)
      {
            close(fd);
            ereport(ERROR,
                        (errcode_for_file_access(),
                         errmsg("could not seek in two-phase state file: %m")));
      }

      /*
       * The state file isn't valid yet, because we haven't written the correct
       * CRC yet.  Before we do that, insert entry in WAL and flush it to disk.
       *
       * Between the time we have written the WAL entry and the time we write
       * out the correct state file CRC, we have an inconsistency: the xact is
       * prepared according to WAL but not according to our on-disk state. We
       * use a critical section to force a PANIC if we are unable to complete
       * the write --- then, WAL replay should repair the inconsistency.      The
       * odds of a PANIC actually occurring should be very tiny given that we
       * were able to write the bogus CRC above.
       *
       * We have to set inCommit here, too; otherwise a checkpoint starting
       * immediately after the WAL record is inserted could complete without
       * fsync'ing our state file.  (This is essentially the same kind of race
       * condition as the COMMIT-to-clog-write case that RecordTransactionCommit
       * uses inCommit for; see notes there.)
       *
       * We save the PREPARE record's location in the gxact for later use by
       * CheckPointTwoPhase.
       */
      START_CRIT_SECTION();

      MyProc->inCommit = true;

      gxact->prepare_lsn = XLogInsert(RM_XACT_ID, XLOG_XACT_PREPARE,
                                                      records.head);
      XLogFlush(gxact->prepare_lsn);

      /* If we crash now, we have prepared: WAL replay will fix things */

      /* write correct CRC and close file */
      if ((write(fd, &statefile_crc, sizeof(pg_crc32))) != sizeof(pg_crc32))
      {
            close(fd);
            ereport(ERROR,
                        (errcode_for_file_access(),
                         errmsg("could not write two-phase state file: %m")));
      }

      if (close(fd) != 0)
            ereport(ERROR,
                        (errcode_for_file_access(),
                         errmsg("could not close two-phase state file: %m")));

      /*
       * Mark the prepared transaction as valid.      As soon as xact.c marks MyProc
       * as not running our XID (which it will do immediately after this
       * function returns), others can commit/rollback the xact.
       *
       * NB: a side effect of this is to make a dummy ProcArray entry for the
       * prepared XID.  This must happen before we clear the XID from MyProc,
       * else there is a window where the XID is not running according to
       * TransactionIdIsInProgress, and onlookers would be entitled to assume
       * the xact crashed.  Instead we have a window where the same XID appears
       * twice in ProcArray, which is OK.
       */
      MarkAsPrepared(gxact);

      /*
       * Now we can mark ourselves as out of the commit critical section: a
       * checkpoint starting after this will certainly see the gxact as a
       * candidate for fsyncing.
       */
      MyProc->inCommit = false;

      END_CRIT_SECTION();

      records.tail = records.head = NULL;
}

/*
 * Register a 2PC record to be written to state file.
 */
void
RegisterTwoPhaseRecord(TwoPhaseRmgrId rmid, uint16 info,
                                 const void *data, uint32 len)
{
      TwoPhaseRecordOnDisk record;

      record.rmid = rmid;
      record.info = info;
      record.len = len;
      save_state_data(&record, sizeof(TwoPhaseRecordOnDisk));
      if (len > 0)
            save_state_data(data, len);
}


/*
 * Read and validate the state file for xid.
 *
 * If it looks OK (has a valid magic number and CRC), return the palloc'd
 * contents of the file.  Otherwise return NULL.
 */
static char *
ReadTwoPhaseFile(TransactionId xid)
{
      char        path[MAXPGPATH];
      char     *buf;
      TwoPhaseFileHeader *hdr;
      int               fd;
      struct stat stat;
      uint32            crc_offset;
      pg_crc32    calc_crc,
                        file_crc;

      TwoPhaseFilePath(path, xid);

      fd = BasicOpenFile(path, O_RDONLY | PG_BINARY, 0);
      if (fd < 0)
      {
            ereport(WARNING,
                        (errcode_for_file_access(),
                         errmsg("could not open two-phase state file \"%s\": %m",
                                    path)));
            return NULL;
      }

      /*
       * Check file length.  We can determine a lower bound pretty easily. We
       * set an upper bound to avoid palloc() failure on a corrupt file, though
       * we can't guarantee that we won't get an out of memory error anyway,
       * even on a valid file.
       */
      if (fstat(fd, &stat))
      {
            close(fd);
            ereport(WARNING,
                        (errcode_for_file_access(),
                         errmsg("could not stat two-phase state file \"%s\": %m",
                                    path)));
            return NULL;
      }

      if (stat.st_size < (MAXALIGN(sizeof(TwoPhaseFileHeader)) +
                                    MAXALIGN(sizeof(TwoPhaseRecordOnDisk)) +
                                    sizeof(pg_crc32)) ||
            stat.st_size > MaxAllocSize)
      {
            close(fd);
            return NULL;
      }

      crc_offset = stat.st_size - sizeof(pg_crc32);
      if (crc_offset != MAXALIGN(crc_offset))
      {
            close(fd);
            return NULL;
      }

      /*
       * OK, slurp in the file.
       */
      buf = (char *) palloc(stat.st_size);

      if (read(fd, buf, stat.st_size) != stat.st_size)
      {
            close(fd);
            ereport(WARNING,
                        (errcode_for_file_access(),
                         errmsg("could not read two-phase state file \"%s\": %m",
                                    path)));
            pfree(buf);
            return NULL;
      }

      close(fd);

      hdr = (TwoPhaseFileHeader *) buf;
      if (hdr->magic != TWOPHASE_MAGIC || hdr->total_len != stat.st_size)
      {
            pfree(buf);
            return NULL;
      }

      INIT_CRC32(calc_crc);
      COMP_CRC32(calc_crc, buf, crc_offset);
      FIN_CRC32(calc_crc);

      file_crc = *((pg_crc32 *) (buf + crc_offset));

      if (!EQ_CRC32(calc_crc, file_crc))
      {
            pfree(buf);
            return NULL;
      }

      return buf;
}


/*
 * FinishPreparedTransaction: execute COMMIT PREPARED or ROLLBACK PREPARED
 */
void
FinishPreparedTransaction(const char *gid, bool isCommit)
{
      GlobalTransaction gxact;
      TransactionId xid;
      char     *buf;
      char     *bufptr;
      TwoPhaseFileHeader *hdr;
      TransactionId latestXid;
      TransactionId *children;
      RelFileNode *commitrels;
      RelFileNode *abortrels;
      RelFileNode *delrels;
      int               ndelrels;
      int               i;

      /*
       * Validate the GID, and lock the GXACT to ensure that two backends do not
       * try to commit the same GID at once.
       */
      gxact = LockGXact(gid, GetUserId());
      xid = gxact->proc.xid;

      /*
       * Read and validate the state file
       */
      buf = ReadTwoPhaseFile(xid);
      if (buf == NULL)
            ereport(ERROR,
                        (errcode(ERRCODE_DATA_CORRUPTED),
                         errmsg("two-phase state file for transaction %u is corrupt",
                                    xid)));

      /*
       * Disassemble the header area
       */
      hdr = (TwoPhaseFileHeader *) buf;
      Assert(TransactionIdEquals(hdr->xid, xid));
      bufptr = buf + MAXALIGN(sizeof(TwoPhaseFileHeader));
      children = (TransactionId *) bufptr;
      bufptr += MAXALIGN(hdr->nsubxacts * sizeof(TransactionId));
      commitrels = (RelFileNode *) bufptr;
      bufptr += MAXALIGN(hdr->ncommitrels * sizeof(RelFileNode));
      abortrels = (RelFileNode *) bufptr;
      bufptr += MAXALIGN(hdr->nabortrels * sizeof(RelFileNode));

      /* compute latestXid among all children */
      latestXid = TransactionIdLatest(xid, hdr->nsubxacts, children);

      /*
       * The order of operations here is critical: make the XLOG entry for
       * commit or abort, then mark the transaction committed or aborted in
       * pg_clog, then remove its PGPROC from the global ProcArray (which means
       * TransactionIdIsInProgress will stop saying the prepared xact is in
       * progress), then run the post-commit or post-abort callbacks. The
       * callbacks will release the locks the transaction held.
       */
      if (isCommit)
            RecordTransactionCommitPrepared(xid,
                                                            hdr->nsubxacts, children,
                                                            hdr->ncommitrels, commitrels);
      else
            RecordTransactionAbortPrepared(xid,
                                                         hdr->nsubxacts, children,
                                                         hdr->nabortrels, abortrels);

      ProcArrayRemove(&gxact->proc, latestXid);

      /*
       * In case we fail while running the callbacks, mark the gxact invalid so
       * no one else will try to commit/rollback, and so it can be recycled
       * properly later.      It is still locked by our XID so it won't go away yet.
       *
       * (We assume it's safe to do this without taking TwoPhaseStateLock.)
       */
      gxact->valid = false;

      /*
       * We have to remove any files that were supposed to be dropped. For
       * consistency with the regular xact.c code paths, must do this before
       * releasing locks, so do it before running the callbacks.
       *
       * NB: this code knows that we couldn't be dropping any temp rels ...
       */
      if (isCommit)
      {
            delrels = commitrels;
            ndelrels = hdr->ncommitrels;
      }
      else
      {
            delrels = abortrels;
            ndelrels = hdr->nabortrels;
      }
      for (i = 0; i < ndelrels; i++)
      {
            SMgrRelation srel = smgropen(delrels[i]);
            ForkNumber  fork;

            for (fork = 0; fork <= MAX_FORKNUM; fork++)
            {
                  if (smgrexists(srel, fork))
                  {
                        XLogDropRelation(delrels[i], fork);
                        smgrdounlink(srel, fork, false, true);
                  }
            }
            smgrclose(srel);
      }

      /* And now do the callbacks */
      if (isCommit)
            ProcessRecords(bufptr, xid, twophase_postcommit_callbacks);
      else
            ProcessRecords(bufptr, xid, twophase_postabort_callbacks);

      /* Count the prepared xact as committed or aborted */
      AtEOXact_PgStat(isCommit);

      /*
       * And now we can clean up our mess.
       */
      RemoveTwoPhaseFile(xid, true);

      RemoveGXact(gxact);

      pfree(buf);
}

/*
 * Scan a 2PC state file (already read into memory by ReadTwoPhaseFile)
 * and call the indicated callbacks for each 2PC record.
 */
static void
ProcessRecords(char *bufptr, TransactionId xid,
                     const TwoPhaseCallback callbacks[])
{
      for (;;)
      {
            TwoPhaseRecordOnDisk *record = (TwoPhaseRecordOnDisk *) bufptr;

            Assert(record->rmid <= TWOPHASE_RM_MAX_ID);
            if (record->rmid == TWOPHASE_RM_END_ID)
                  break;

            bufptr += MAXALIGN(sizeof(TwoPhaseRecordOnDisk));

            if (callbacks[record->rmid] != NULL)
                  callbacks[record->rmid] (xid, record->info,
                                                       (void *) bufptr, record->len);

            bufptr += MAXALIGN(record->len);
      }
}

/*
 * Remove the 2PC file for the specified XID.
 *
 * If giveWarning is false, do not complain about file-not-present;
 * this is an expected case during WAL replay.
 */
void
RemoveTwoPhaseFile(TransactionId xid, bool giveWarning)
{
      char        path[MAXPGPATH];

      TwoPhaseFilePath(path, xid);
      if (unlink(path))
            if (errno != ENOENT || giveWarning)
                  ereport(WARNING,
                              (errcode_for_file_access(),
                           errmsg("could not remove two-phase state file \"%s\": %m",
                                      path)));
}

/*
 * Recreates a state file. This is used in WAL replay.
 *
 * Note: content and len don't include CRC.
 */
void
RecreateTwoPhaseFile(TransactionId xid, void *content, int len)
{
      char        path[MAXPGPATH];
      pg_crc32    statefile_crc;
      int               fd;

      /* Recompute CRC */
      INIT_CRC32(statefile_crc);
      COMP_CRC32(statefile_crc, content, len);
      FIN_CRC32(statefile_crc);

      TwoPhaseFilePath(path, xid);

      fd = BasicOpenFile(path,
                                 O_CREAT | O_TRUNC | O_WRONLY | PG_BINARY,
                                 S_IRUSR | S_IWUSR);
      if (fd < 0)
            ereport(ERROR,
                        (errcode_for_file_access(),
                         errmsg("could not recreate two-phase state file \"%s\": %m",
                                    path)));

      /* Write content and CRC */
      if (write(fd, content, len) != len)
      {
            close(fd);
            ereport(ERROR,
                        (errcode_for_file_access(),
                         errmsg("could not write two-phase state file: %m")));
      }
      if (write(fd, &statefile_crc, sizeof(pg_crc32)) != sizeof(pg_crc32))
      {
            close(fd);
            ereport(ERROR,
                        (errcode_for_file_access(),
                         errmsg("could not write two-phase state file: %m")));
      }

      /*
       * We must fsync the file because the end-of-replay checkpoint will not do
       * so, there being no GXACT in shared memory yet to tell it to.
       */
      if (pg_fsync(fd) != 0)
      {
            close(fd);
            ereport(ERROR,
                        (errcode_for_file_access(),
                         errmsg("could not fsync two-phase state file: %m")));
      }

      if (close(fd) != 0)
            ereport(ERROR,
                        (errcode_for_file_access(),
                         errmsg("could not close two-phase state file: %m")));
}

/*
 * CheckPointTwoPhase -- handle 2PC component of checkpointing.
 *
 * We must fsync the state file of any GXACT that is valid and has a PREPARE
 * LSN <= the checkpoint's redo horizon.  (If the gxact isn't valid yet or
 * has a later LSN, this checkpoint is not responsible for fsyncing it.)
 *
 * This is deliberately run as late as possible in the checkpoint sequence,
 * because GXACTs ordinarily have short lifespans, and so it is quite
 * possible that GXACTs that were valid at checkpoint start will no longer
 * exist if we wait a little bit.
 *
 * If a GXACT remains valid across multiple checkpoints, it'll be fsynced
 * each time.  This is considered unusual enough that we don't bother to
 * expend any extra code to avoid the redundant fsyncs.  (They should be
 * reasonably cheap anyway, since they won't cause I/O.)
 */
void
CheckPointTwoPhase(XLogRecPtr redo_horizon)
{
      TransactionId *xids;
      int               nxids;
      char        path[MAXPGPATH];
      int               i;

      /*
       * We don't want to hold the TwoPhaseStateLock while doing I/O, so we grab
       * it just long enough to make a list of the XIDs that require fsyncing,
       * and then do the I/O afterwards.
       *
       * This approach creates a race condition: someone else could delete a
       * GXACT between the time we release TwoPhaseStateLock and the time we try
       * to open its state file.    We handle this by special-casing ENOENT
       * failures: if we see that, we verify that the GXACT is no longer valid,
       * and if so ignore the failure.
       */
      if (max_prepared_xacts <= 0)
            return;                             /* nothing to do */

      TRACE_POSTGRESQL_TWOPHASE_CHECKPOINT_START();

      xids = (TransactionId *) palloc(max_prepared_xacts * sizeof(TransactionId));
      nxids = 0;

      LWLockAcquire(TwoPhaseStateLock, LW_SHARED);

      for (i = 0; i < TwoPhaseState->numPrepXacts; i++)
      {
            GlobalTransaction gxact = TwoPhaseState->prepXacts[i];

            if (gxact->valid &&
                  XLByteLE(gxact->prepare_lsn, redo_horizon))
                  xids[nxids++] = gxact->proc.xid;
      }

      LWLockRelease(TwoPhaseStateLock);

      for (i = 0; i < nxids; i++)
      {
            TransactionId xid = xids[i];
            int               fd;

            TwoPhaseFilePath(path, xid);

            fd = BasicOpenFile(path, O_RDWR | PG_BINARY, 0);
            if (fd < 0)
            {
                  if (errno == ENOENT)
                  {
                        /* OK if gxact is no longer valid */
                        if (!TransactionIdIsPrepared(xid))
                              continue;
                        /* Restore errno in case it was changed */
                        errno = ENOENT;
                  }
                  ereport(ERROR,
                              (errcode_for_file_access(),
                               errmsg("could not open two-phase state file \"%s\": %m",
                                          path)));
            }

            if (pg_fsync(fd) != 0)
            {
                  close(fd);
                  ereport(ERROR,
                              (errcode_for_file_access(),
                               errmsg("could not fsync two-phase state file \"%s\": %m",
                                          path)));
            }

            if (close(fd) != 0)
                  ereport(ERROR,
                              (errcode_for_file_access(),
                               errmsg("could not close two-phase state file \"%s\": %m",
                                          path)));
      }

      pfree(xids);

      TRACE_POSTGRESQL_TWOPHASE_CHECKPOINT_DONE();
}

/*
 * PrescanPreparedTransactions
 *
 * Scan the pg_twophase directory and determine the range of valid XIDs
 * present.  This is run during database startup, after we have completed
 * reading WAL.  ShmemVariableCache->nextXid has been set to one more than
 * the highest XID for which evidence exists in WAL.
 *
 * We throw away any prepared xacts with main XID beyond nextXid --- if any
 * are present, it suggests that the DBA has done a PITR recovery to an
 * earlier point in time without cleaning out pg_twophase.  We dare not
 * try to recover such prepared xacts since they likely depend on database
 * state that doesn't exist now.
 *
 * However, we will advance nextXid beyond any subxact XIDs belonging to
 * valid prepared xacts.  We need to do this since subxact commit doesn't
 * write a WAL entry, and so there might be no evidence in WAL of those
 * subxact XIDs.
 *
 * Our other responsibility is to determine and return the oldest valid XID
 * among the prepared xacts (if none, return ShmemVariableCache->nextXid).
 * This is needed to synchronize pg_subtrans startup properly.
 */
TransactionId
PrescanPreparedTransactions(void)
{
      TransactionId origNextXid = ShmemVariableCache->nextXid;
      TransactionId result = origNextXid;
      DIR            *cldir;
      struct dirent *clde;

      cldir = AllocateDir(TWOPHASE_DIR);
      while ((clde = ReadDir(cldir, TWOPHASE_DIR)) != NULL)
      {
            if (strlen(clde->d_name) == 8 &&
                  strspn(clde->d_name, "0123456789ABCDEF") == 8)
            {
                  TransactionId xid;
                  char     *buf;
                  TwoPhaseFileHeader *hdr;
                  TransactionId *subxids;
                  int               i;

                  xid = (TransactionId) strtoul(clde->d_name, NULL, 16);

                  /* Reject XID if too new */
                  if (TransactionIdFollowsOrEquals(xid, origNextXid))
                  {
                        ereport(WARNING,
                                    (errmsg("removing future two-phase state file \"%s\"",
                                                clde->d_name)));
                        RemoveTwoPhaseFile(xid, true);
                        continue;
                  }

                  /*
                   * Note: we can't check if already processed because clog
                   * subsystem isn't up yet.
                   */

                  /* Read and validate file */
                  buf = ReadTwoPhaseFile(xid);
                  if (buf == NULL)
                  {
                        ereport(WARNING,
                                (errmsg("removing corrupt two-phase state file \"%s\"",
                                            clde->d_name)));
                        RemoveTwoPhaseFile(xid, true);
                        continue;
                  }

                  /* Deconstruct header */
                  hdr = (TwoPhaseFileHeader *) buf;
                  if (!TransactionIdEquals(hdr->xid, xid))
                  {
                        ereport(WARNING,
                                (errmsg("removing corrupt two-phase state file \"%s\"",
                                            clde->d_name)));
                        RemoveTwoPhaseFile(xid, true);
                        pfree(buf);
                        continue;
                  }

                  /*
                   * OK, we think this file is valid.  Incorporate xid into the
                   * running-minimum result.
                   */
                  if (TransactionIdPrecedes(xid, result))
                        result = xid;

                  /*
                   * Examine subtransaction XIDs ... they should all follow main
                   * XID, and they may force us to advance nextXid.
                   */
                  subxids = (TransactionId *)
                        (buf + MAXALIGN(sizeof(TwoPhaseFileHeader)));
                  for (i = 0; i < hdr->nsubxacts; i++)
                  {
                        TransactionId subxid = subxids[i];

                        Assert(TransactionIdFollows(subxid, xid));
                        if (TransactionIdFollowsOrEquals(subxid,
                                                                         ShmemVariableCache->nextXid))
                        {
                              ShmemVariableCache->nextXid = subxid;
                              TransactionIdAdvance(ShmemVariableCache->nextXid);
                        }
                  }

                  pfree(buf);
            }
      }
      FreeDir(cldir);

      return result;
}

/*
 * RecoverPreparedTransactions
 *
 * Scan the pg_twophase directory and reload shared-memory state for each
 * prepared transaction (reacquire locks, etc).  This is run during database
 * startup.
 */
void
RecoverPreparedTransactions(void)
{
      char        dir[MAXPGPATH];
      DIR            *cldir;
      struct dirent *clde;

      snprintf(dir, MAXPGPATH, "%s", TWOPHASE_DIR);

      cldir = AllocateDir(dir);
      while ((clde = ReadDir(cldir, dir)) != NULL)
      {
            if (strlen(clde->d_name) == 8 &&
                  strspn(clde->d_name, "0123456789ABCDEF") == 8)
            {
                  TransactionId xid;
                  char     *buf;
                  char     *bufptr;
                  TwoPhaseFileHeader *hdr;
                  TransactionId *subxids;
                  GlobalTransaction gxact;
                  int               i;

                  xid = (TransactionId) strtoul(clde->d_name, NULL, 16);

                  /* Already processed? */
                  if (TransactionIdDidCommit(xid) || TransactionIdDidAbort(xid))
                  {
                        ereport(WARNING,
                                    (errmsg("removing stale two-phase state file \"%s\"",
                                                clde->d_name)));
                        RemoveTwoPhaseFile(xid, true);
                        continue;
                  }

                  /* Read and validate file */
                  buf = ReadTwoPhaseFile(xid);
                  if (buf == NULL)
                  {
                        ereport(WARNING,
                                (errmsg("removing corrupt two-phase state file \"%s\"",
                                            clde->d_name)));
                        RemoveTwoPhaseFile(xid, true);
                        continue;
                  }

                  ereport(LOG,
                              (errmsg("recovering prepared transaction %u", xid)));

                  /* Deconstruct header */
                  hdr = (TwoPhaseFileHeader *) buf;
                  Assert(TransactionIdEquals(hdr->xid, xid));
                  bufptr = buf + MAXALIGN(sizeof(TwoPhaseFileHeader));
                  subxids = (TransactionId *) bufptr;
                  bufptr += MAXALIGN(hdr->nsubxacts * sizeof(TransactionId));
                  bufptr += MAXALIGN(hdr->ncommitrels * sizeof(RelFileNode));
                  bufptr += MAXALIGN(hdr->nabortrels * sizeof(RelFileNode));

                  /*
                   * Reconstruct subtrans state for the transaction --- needed
                   * because pg_subtrans is not preserved over a restart.  Note that
                   * we are linking all the subtransactions directly to the
                   * top-level XID; there may originally have been a more complex
                   * hierarchy, but there's no need to restore that exactly.
                   */
                  for (i = 0; i < hdr->nsubxacts; i++)
                        SubTransSetParent(subxids[i], xid);

                  /*
                   * Recreate its GXACT and dummy PGPROC
                   *
                   * Note: since we don't have the PREPARE record's WAL location at
                   * hand, we leave prepare_lsn zeroes.  This means the GXACT will
                   * be fsync'd on every future checkpoint.  We assume this
                   * situation is infrequent enough that the performance cost is
                   * negligible (especially since we know the state file has already
                   * been fsynced).
                   */
                  gxact = MarkAsPreparing(xid, hdr->gid,
                                                      hdr->prepared_at,
                                                      hdr->owner, hdr->database);
                  GXactLoadSubxactData(gxact, hdr->nsubxacts, subxids);
                  MarkAsPrepared(gxact);

                  /*
                   * Recover other state (notably locks) using resource managers
                   */
                  ProcessRecords(bufptr, xid, twophase_recover_callbacks);

                  pfree(buf);
            }
      }
      FreeDir(cldir);
}

/*
 *    RecordTransactionCommitPrepared
 *
 * This is basically the same as RecordTransactionCommit: in particular,
 * we must set the inCommit flag to avoid a race condition.
 *
 * We know the transaction made at least one XLOG entry (its PREPARE),
 * so it is never possible to optimize out the commit record.
 */
static void
RecordTransactionCommitPrepared(TransactionId xid,
                                                int nchildren,
                                                TransactionId *children,
                                                int nrels,
                                                RelFileNode *rels)
{
      XLogRecData rdata[3];
      int               lastrdata = 0;
      xl_xact_commit_prepared xlrec;
      XLogRecPtr  recptr;

      START_CRIT_SECTION();

      /* See notes in RecordTransactionCommit */
      MyProc->inCommit = true;

      /* Emit the XLOG commit record */
      xlrec.xid = xid;
      xlrec.crec.xact_time = GetCurrentTimestamp();
      xlrec.crec.nrels = nrels;
      xlrec.crec.nsubxacts = nchildren;
      rdata[0].data = (char *) (&xlrec);
      rdata[0].len = MinSizeOfXactCommitPrepared;
      rdata[0].buffer = InvalidBuffer;
      /* dump rels to delete */
      if (nrels > 0)
      {
            rdata[0].next = &(rdata[1]);
            rdata[1].data = (char *) rels;
            rdata[1].len = nrels * sizeof(RelFileNode);
            rdata[1].buffer = InvalidBuffer;
            lastrdata = 1;
      }
      /* dump committed child Xids */
      if (nchildren > 0)
      {
            rdata[lastrdata].next = &(rdata[2]);
            rdata[2].data = (char *) children;
            rdata[2].len = nchildren * sizeof(TransactionId);
            rdata[2].buffer = InvalidBuffer;
            lastrdata = 2;
      }
      rdata[lastrdata].next = NULL;

      recptr = XLogInsert(RM_XACT_ID, XLOG_XACT_COMMIT_PREPARED, rdata);

      /*
       * We don't currently try to sleep before flush here ... nor is there any
       * support for async commit of a prepared xact (the very idea is probably
       * a contradiction)
       */

      /* Flush XLOG to disk */
      XLogFlush(recptr);

      /* Mark the transaction committed in pg_clog */
      TransactionIdCommitTree(xid, nchildren, children);

      /* Checkpoint can proceed now */
      MyProc->inCommit = false;

      END_CRIT_SECTION();
}

/*
 *    RecordTransactionAbortPrepared
 *
 * This is basically the same as RecordTransactionAbort.
 *
 * We know the transaction made at least one XLOG entry (its PREPARE),
 * so it is never possible to optimize out the abort record.
 */
static void
RecordTransactionAbortPrepared(TransactionId xid,
                                             int nchildren,
                                             TransactionId *children,
                                             int nrels,
                                             RelFileNode *rels)
{
      XLogRecData rdata[3];
      int               lastrdata = 0;
      xl_xact_abort_prepared xlrec;
      XLogRecPtr  recptr;

      /*
       * Catch the scenario where we aborted partway through
       * RecordTransactionCommitPrepared ...
       */
      if (TransactionIdDidCommit(xid))
            elog(PANIC, "cannot abort transaction %u, it was already committed",
                   xid);

      START_CRIT_SECTION();

      /* Emit the XLOG abort record */
      xlrec.xid = xid;
      xlrec.arec.xact_time = GetCurrentTimestamp();
      xlrec.arec.nrels = nrels;
      xlrec.arec.nsubxacts = nchildren;
      rdata[0].data = (char *) (&xlrec);
      rdata[0].len = MinSizeOfXactAbortPrepared;
      rdata[0].buffer = InvalidBuffer;
      /* dump rels to delete */
      if (nrels > 0)
      {
            rdata[0].next = &(rdata[1]);
            rdata[1].data = (char *) rels;
            rdata[1].len = nrels * sizeof(RelFileNode);
            rdata[1].buffer = InvalidBuffer;
            lastrdata = 1;
      }
      /* dump committed child Xids */
      if (nchildren > 0)
      {
            rdata[lastrdata].next = &(rdata[2]);
            rdata[2].data = (char *) children;
            rdata[2].len = nchildren * sizeof(TransactionId);
            rdata[2].buffer = InvalidBuffer;
            lastrdata = 2;
      }
      rdata[lastrdata].next = NULL;

      recptr = XLogInsert(RM_XACT_ID, XLOG_XACT_ABORT_PREPARED, rdata);

      /* Always flush, since we're about to remove the 2PC state file */
      XLogFlush(recptr);

      /*
       * Mark the transaction aborted in clog.  This is not absolutely necessary
       * but we may as well do it while we are here.
       */
      TransactionIdAbortTree(xid, nchildren, children);

      END_CRIT_SECTION();
}

Generated by  Doxygen 1.6.0   Back to index