granicus.if.org Git - postgresql/blob - src/backend/utils/adt/txid.c

   1 /*-------------------------------------------------------------------------
   2  * txid.c
   3  *
   4  *      Export internal transaction IDs to user level.
   5  *
   6  * Note that only top-level transaction IDs are ever converted to TXID.
   7  * This is important because TXIDs frequently persist beyond the global
   8  * xmin horizon, or may even be shipped to other machines, so we cannot
   9  * rely on being able to correlate subtransaction IDs with their parents
  10  * via functions such as SubTransGetTopmostTransaction().
  11  *
  12  *
  13  *      Copyright (c) 2003-2009, PostgreSQL Global Development Group
  14  *      Author: Jan Wieck, Afilias USA INC.
  15  *      64-bit txids: Marko Kreen, Skype Technologies
  16  *
  17  *      $PostgreSQL: pgsql/src/backend/utils/adt/txid.c,v 1.9 2009/12/19 01:32:36 sriggs Exp $
  18  *
  19  *-------------------------------------------------------------------------
  20  */
  21
  22 #include "postgres.h"
  23
  24 #include "access/transam.h"
  25 #include "access/xact.h"
  26 #include "funcapi.h"
  27 #include "miscadmin.h"
  28 #include "libpq/pqformat.h"
  29 #include "utils/builtins.h"
  30 #include "utils/snapmgr.h"
  31
  32
  33 #ifndef INT64_IS_BUSTED
  34 /* txid will be signed int8 in database, so must limit to 63 bits */
  35 #define MAX_TXID   UINT64CONST(0x7FFFFFFFFFFFFFFF)
  36 #else
  37 /* we only really have 32 bits to work with :-( */
  38 #define MAX_TXID   UINT64CONST(0x7FFFFFFF)
  39 #endif
  40
  41 /* Use unsigned variant internally */
  42 typedef uint64 txid;
  43
  44 /* sprintf format code for uint64 */
  45 #define TXID_FMT UINT64_FORMAT
  46
  47 /*
  48  * If defined, use bsearch() function for searching for txids in snapshots
  49  * that have more than the specified number of values.
  50  */
  51 #define USE_BSEARCH_IF_NXIP_GREATER 30
  52
  53
  54 /*
  55  * Snapshot containing 8byte txids.
  56  */
  57 typedef struct
  58 {
  59         /*
  60          * 4-byte length hdr, should not be touched directly.
  61          *
  62          * Explicit embedding is ok as we want always correct alignment anyway.
  63          */
  64         int32           __varsz;
  65
  66         uint32          nxip;                   /* number of txids in xip array */
  67         txid            xmin;
  68         txid            xmax;
  69         txid            xip[1];                 /* in-progress txids, xmin <= xip[i] < xmax */
  70 } TxidSnapshot;
  71
  72 #define TXID_SNAPSHOT_SIZE(nxip) \
  73         (offsetof(TxidSnapshot, xip) + sizeof(txid) * (nxip))
  74
  75 /*
  76  * Epoch values from xact.c
  77  */
  78 typedef struct
  79 {
  80         TransactionId last_xid;
  81         uint32          epoch;
  82 } TxidEpoch;
  83
  84
  85 /*
  86  * Fetch epoch data from xact.c.
  87  */
  88 static void
  89 load_xid_epoch(TxidEpoch *state)
  90 {
  91         GetNextXidAndEpoch(&state->last_xid, &state->epoch);
  92 }
  93
  94 /*
  95  * do a TransactionId -> txid conversion for an XID near the given epoch
  96  */
  97 static txid
  98 convert_xid(TransactionId xid, const TxidEpoch *state)
  99 {
 100 #ifndef INT64_IS_BUSTED
 101         uint64          epoch;
 102
 103         /* return special xid's as-is */
 104         if (!TransactionIdIsNormal(xid))
 105                 return (txid) xid;
 106
 107         /* xid can be on either side when near wrap-around */
 108         epoch = (uint64) state->epoch;
 109         if (xid > state->last_xid &&
 110                 TransactionIdPrecedes(xid, state->last_xid))
 111                 epoch--;
 112         else if (xid < state->last_xid &&
 113                          TransactionIdFollows(xid, state->last_xid))
 114                 epoch++;
 115
 116         return (epoch << 32) | xid;
 117 #else                                                   /* INT64_IS_BUSTED */
 118         /* we can't do anything with the epoch, so ignore it */
 119         return (txid) xid & MAX_TXID;
 120 #endif   /* INT64_IS_BUSTED */
 121 }
 122
 123 /*
 124  * txid comparator for qsort/bsearch
 125  */
 126 static int
 127 cmp_txid(const void *aa, const void *bb)
 128 {
 129         txid            a = *(const txid *) aa;
 130         txid            b = *(const txid *) bb;
 131
 132         if (a < b)
 133                 return -1;
 134         if (a > b)
 135                 return 1;
 136         return 0;
 137 }
 138
 139 /*
 140  * sort a snapshot's txids, so we can use bsearch() later.
 141  *
 142  * For consistency of on-disk representation, we always sort even if bsearch
 143  * will not be used.
 144  */
 145 static void
 146 sort_snapshot(TxidSnapshot *snap)
 147 {
 148         if (snap->nxip > 1)
 149                 qsort(snap->xip, snap->nxip, sizeof(txid), cmp_txid);
 150 }
 151
 152 /*
 153  * check txid visibility.
 154  */
 155 static bool
 156 is_visible_txid(txid value, const TxidSnapshot *snap)
 157 {
 158         if (value < snap->xmin)
 159                 return true;
 160         else if (value >= snap->xmax)
 161                 return false;
 162 #ifdef USE_BSEARCH_IF_NXIP_GREATER
 163         else if (snap->nxip > USE_BSEARCH_IF_NXIP_GREATER)
 164         {
 165                 void       *res;
 166
 167                 res = bsearch(&value, snap->xip, snap->nxip, sizeof(txid), cmp_txid);
 168                 /* if found, transaction is still in progress */
 169                 return (res) ? false : true;
 170         }
 171 #endif
 172         else
 173         {
 174                 uint32          i;
 175
 176                 for (i = 0; i < snap->nxip; i++)
 177                 {
 178                         if (value == snap->xip[i])
 179                                 return false;
 180                 }
 181                 return true;
 182         }
 183 }
 184
 185 /*
 186  * helper functions to use StringInfo for TxidSnapshot creation.
 187  */
 188
 189 static StringInfo
 190 buf_init(txid xmin, txid xmax)
 191 {
 192         TxidSnapshot snap;
 193         StringInfo      buf;
 194
 195         snap.xmin = xmin;
 196         snap.xmax = xmax;
 197         snap.nxip = 0;
 198
 199         buf = makeStringInfo();
 200         appendBinaryStringInfo(buf, (char *) &snap, TXID_SNAPSHOT_SIZE(0));
 201         return buf;
 202 }
 203
 204 static void
 205 buf_add_txid(StringInfo buf, txid xid)
 206 {
 207         TxidSnapshot *snap = (TxidSnapshot *) buf->data;
 208
 209         /* do this before possible realloc */
 210         snap->nxip++;
 211
 212         appendBinaryStringInfo(buf, (char *) &xid, sizeof(xid));
 213 }
 214
 215 static TxidSnapshot *
 216 buf_finalize(StringInfo buf)
 217 {
 218         TxidSnapshot *snap = (TxidSnapshot *) buf->data;
 219
 220         SET_VARSIZE(snap, buf->len);
 221
 222         /* buf is not needed anymore */
 223         buf->data = NULL;
 224         pfree(buf);
 225
 226         return snap;
 227 }
 228
 229 /*
 230  * simple number parser.
 231  *
 232  * We return 0 on error, which is invalid value for txid.
 233  */
 234 static txid
 235 str2txid(const char *s, const char **endp)
 236 {
 237         txid            val = 0;
 238         txid            cutoff = MAX_TXID / 10;
 239         txid            cutlim = MAX_TXID % 10;
 240
 241         for (; *s; s++)
 242         {
 243                 unsigned        d;
 244
 245                 if (*s < '0' || *s > '9')
 246                         break;
 247                 d = *s - '0';
 248
 249                 /*
 250                  * check for overflow
 251                  */
 252                 if (val > cutoff || (val == cutoff && d > cutlim))
 253                 {
 254                         val = 0;
 255                         break;
 256                 }
 257
 258                 val = val * 10 + d;
 259         }
 260         if (endp)
 261                 *endp = s;
 262         return val;
 263 }
 264
 265 /*
 266  * parse snapshot from cstring
 267  */
 268 static TxidSnapshot *
 269 parse_snapshot(const char *str)
 270 {
 271         txid            xmin;
 272         txid            xmax;
 273         txid            last_val = 0,
 274                                 val;
 275         const char *str_start = str;
 276         const char *endp;
 277         StringInfo      buf;
 278
 279         xmin = str2txid(str, &endp);
 280         if (*endp != ':')
 281                 goto bad_format;
 282         str = endp + 1;
 283
 284         xmax = str2txid(str, &endp);
 285         if (*endp != ':')
 286                 goto bad_format;
 287         str = endp + 1;
 288
 289         /* it should look sane */
 290         if (xmin == 0 || xmax == 0 || xmin > xmax)
 291                 goto bad_format;
 292
 293         /* allocate buffer */
 294         buf = buf_init(xmin, xmax);
 295
 296         /* loop over values */
 297         while (*str != '\0')
 298         {
 299                 /* read next value */
 300                 val = str2txid(str, &endp);
 301                 str = endp;
 302
 303                 /* require the input to be in order */
 304                 if (val < xmin || val >= xmax || val <= last_val)
 305                         goto bad_format;
 306
 307                 buf_add_txid(buf, val);
 308                 last_val = val;
 309
 310                 if (*str == ',')
 311                         str++;
 312                 else if (*str != '\0')
 313                         goto bad_format;
 314         }
 315
 316         return buf_finalize(buf);
 317
 318 bad_format:
 319         elog(ERROR, "invalid input for txid_snapshot: \"%s\"", str_start);
 320         return NULL;
 321 }
 322
 323 /*
 324  * Public functions.
 325  *
 326  * txid_current() and txid_current_snapshot() are the only ones that
 327  * communicate with core xid machinery.  All the others work on data
 328  * returned by them.
 329  */
 330
 331 /*
 332  * txid_current() returns int8
 333  *
 334  *              Return the current toplevel transaction ID as TXID
 335  */
 336 Datum
 337 txid_current(PG_FUNCTION_ARGS)
 338 {
 339         txid            val;
 340         TxidEpoch       state;
 341
 342         /*
 343          * Must prevent during recovery because if an xid is
 344          * not assigned we try to assign one, which would fail.
 345          * Programs already rely on this function to always
 346          * return a valid current xid, so we should not change
 347          * this to return NULL or similar invalid xid.
 348          */
 349         PreventCommandDuringRecovery();
 350
 351         load_xid_epoch(&state);
 352
 353         val = convert_xid(GetTopTransactionId(), &state);
 354
 355         PG_RETURN_INT64(val);
 356 }
 357
 358 /*
 359  * txid_current_snapshot() returns txid_snapshot
 360  *
 361  *              Return current snapshot in TXID format
 362  *
 363  * Note that only top-transaction XIDs are included in the snapshot.
 364  */
 365 Datum
 366 txid_current_snapshot(PG_FUNCTION_ARGS)
 367 {
 368         TxidSnapshot *snap;
 369         uint32          nxip,
 370                                 i,
 371                                 size;
 372         TxidEpoch       state;
 373         Snapshot        cur;
 374
 375         cur = GetActiveSnapshot();
 376         if (cur == NULL)
 377                 elog(ERROR, "no active snapshot set");
 378
 379         load_xid_epoch(&state);
 380
 381         /* allocate */
 382         nxip = cur->xcnt;
 383         size = TXID_SNAPSHOT_SIZE(nxip);
 384         snap = palloc(size);
 385         SET_VARSIZE(snap, size);
 386
 387         /* fill */
 388         snap->xmin = convert_xid(cur->xmin, &state);
 389         snap->xmax = convert_xid(cur->xmax, &state);
 390         snap->nxip = nxip;
 391         for (i = 0; i < nxip; i++)
 392                 snap->xip[i] = convert_xid(cur->xip[i], &state);
 393
 394         /* we want them guaranteed to be in ascending order */
 395         sort_snapshot(snap);
 396
 397         PG_RETURN_POINTER(snap);
 398 }
 399
 400 /*
 401  * txid_snapshot_in(cstring) returns txid_snapshot
 402  *
 403  *              input function for type txid_snapshot
 404  */
 405 Datum
 406 txid_snapshot_in(PG_FUNCTION_ARGS)
 407 {
 408         char       *str = PG_GETARG_CSTRING(0);
 409         TxidSnapshot *snap;
 410
 411         snap = parse_snapshot(str);
 412
 413         PG_RETURN_POINTER(snap);
 414 }
 415
 416 /*
 417  * txid_snapshot_out(txid_snapshot) returns cstring
 418  *
 419  *              output function for type txid_snapshot
 420  */
 421 Datum
 422 txid_snapshot_out(PG_FUNCTION_ARGS)
 423 {
 424         TxidSnapshot *snap = (TxidSnapshot *) PG_GETARG_VARLENA_P(0);
 425         StringInfoData str;
 426         uint32          i;
 427
 428         initStringInfo(&str);
 429
 430         appendStringInfo(&str, TXID_FMT ":", snap->xmin);
 431         appendStringInfo(&str, TXID_FMT ":", snap->xmax);
 432
 433         for (i = 0; i < snap->nxip; i++)
 434         {
 435                 if (i > 0)
 436                         appendStringInfoChar(&str, ',');
 437                 appendStringInfo(&str, TXID_FMT, snap->xip[i]);
 438         }
 439
 440         PG_RETURN_CSTRING(str.data);
 441 }
 442
 443 /*
 444  * txid_snapshot_recv(internal) returns txid_snapshot
 445  *
 446  *              binary input function for type txid_snapshot
 447  *
 448  *              format: int4 nxip, int8 xmin, int8 xmax, int8 xip
 449  */
 450 Datum
 451 txid_snapshot_recv(PG_FUNCTION_ARGS)
 452 {
 453         StringInfo      buf = (StringInfo) PG_GETARG_POINTER(0);
 454         TxidSnapshot *snap;
 455         txid            last = 0;
 456         int                     nxip;
 457         int                     i;
 458         int                     avail;
 459         int                     expect;
 460         txid            xmin,
 461                                 xmax;
 462
 463         /*
 464          * load nxip and check for nonsense.
 465          *
 466          * (nxip > avail) check is against int overflows in 'expect'.
 467          */
 468         nxip = pq_getmsgint(buf, 4);
 469         avail = buf->len - buf->cursor;
 470         expect = 8 + 8 + nxip * 8;
 471         if (nxip < 0 || nxip > avail || expect > avail)
 472                 goto bad_format;
 473
 474         xmin = pq_getmsgint64(buf);
 475         xmax = pq_getmsgint64(buf);
 476         if (xmin == 0 || xmax == 0 || xmin > xmax || xmax > MAX_TXID)
 477                 goto bad_format;
 478
 479         snap = palloc(TXID_SNAPSHOT_SIZE(nxip));
 480         snap->xmin = xmin;
 481         snap->xmax = xmax;
 482         snap->nxip = nxip;
 483         SET_VARSIZE(snap, TXID_SNAPSHOT_SIZE(nxip));
 484
 485         for (i = 0; i < nxip; i++)
 486         {
 487                 txid            cur = pq_getmsgint64(buf);
 488
 489                 if (cur <= last || cur < xmin || cur >= xmax)
 490                         goto bad_format;
 491                 snap->xip[i] = cur;
 492                 last = cur;
 493         }
 494         PG_RETURN_POINTER(snap);
 495
 496 bad_format:
 497         elog(ERROR, "invalid snapshot data");
 498         return (Datum) NULL;
 499 }
 500
 501 /*
 502  * txid_snapshot_send(txid_snapshot) returns bytea
 503  *
 504  *              binary output function for type txid_snapshot
 505  *
 506  *              format: int4 nxip, int8 xmin, int8 xmax, int8 xip
 507  */
 508 Datum
 509 txid_snapshot_send(PG_FUNCTION_ARGS)
 510 {
 511         TxidSnapshot *snap = (TxidSnapshot *) PG_GETARG_VARLENA_P(0);
 512         StringInfoData buf;
 513         uint32          i;
 514
 515         pq_begintypsend(&buf);
 516         pq_sendint(&buf, snap->nxip, 4);
 517         pq_sendint64(&buf, snap->xmin);
 518         pq_sendint64(&buf, snap->xmax);
 519         for (i = 0; i < snap->nxip; i++)
 520                 pq_sendint64(&buf, snap->xip[i]);
 521         PG_RETURN_BYTEA_P(pq_endtypsend(&buf));
 522 }
 523
 524 /*
 525  * txid_visible_in_snapshot(int8, txid_snapshot) returns bool
 526  *
 527  *              is txid visible in snapshot ?
 528  */
 529 Datum
 530 txid_visible_in_snapshot(PG_FUNCTION_ARGS)
 531 {
 532         txid            value = PG_GETARG_INT64(0);
 533         TxidSnapshot *snap = (TxidSnapshot *) PG_GETARG_VARLENA_P(1);
 534
 535         PG_RETURN_BOOL(is_visible_txid(value, snap));
 536 }
 537
 538 /*
 539  * txid_snapshot_xmin(txid_snapshot) returns int8
 540  *
 541  *              return snapshot's xmin
 542  */
 543 Datum
 544 txid_snapshot_xmin(PG_FUNCTION_ARGS)
 545 {
 546         TxidSnapshot *snap = (TxidSnapshot *) PG_GETARG_VARLENA_P(0);
 547
 548         PG_RETURN_INT64(snap->xmin);
 549 }
 550
 551 /*
 552  * txid_snapshot_xmax(txid_snapshot) returns int8
 553  *
 554  *              return snapshot's xmax
 555  */
 556 Datum
 557 txid_snapshot_xmax(PG_FUNCTION_ARGS)
 558 {
 559         TxidSnapshot *snap = (TxidSnapshot *) PG_GETARG_VARLENA_P(0);
 560
 561         PG_RETURN_INT64(snap->xmax);
 562 }
 563
 564 /*
 565  * txid_snapshot_xip(txid_snapshot) returns setof int8
 566  *
 567  *              return in-progress TXIDs in snapshot.
 568  */
 569 Datum
 570 txid_snapshot_xip(PG_FUNCTION_ARGS)
 571 {
 572         FuncCallContext *fctx;
 573         TxidSnapshot *snap;
 574         txid            value;
 575
 576         /* on first call initialize snap_state and get copy of snapshot */
 577         if (SRF_IS_FIRSTCALL())
 578         {
 579                 TxidSnapshot *arg = (TxidSnapshot *) PG_GETARG_VARLENA_P(0);
 580
 581                 fctx = SRF_FIRSTCALL_INIT();
 582
 583                 /* make a copy of user snapshot */
 584                 snap = MemoryContextAlloc(fctx->multi_call_memory_ctx, VARSIZE(arg));
 585                 memcpy(snap, arg, VARSIZE(arg));
 586
 587                 fctx->user_fctx = snap;
 588         }
 589
 590         /* return values one-by-one */
 591         fctx = SRF_PERCALL_SETUP();
 592         snap = fctx->user_fctx;
 593         if (fctx->call_cntr < snap->nxip)
 594         {
 595                 value = snap->xip[fctx->call_cntr];
 596                 SRF_RETURN_NEXT(fctx, Int64GetDatum(value));
 597         }
 598         else
 599         {
 600                 SRF_RETURN_DONE(fctx);
 601         }
 602 }