src/backend/storage/lmgr/lmgr.c

   1 /*-------------------------------------------------------------------------
   2  *
   3  * lmgr.c
   4  *        POSTGRES lock manager code
   5  *
   6  * Portions Copyright (c) 1996-2025, PostgreSQL Global Development Group
   7  * Portions Copyright (c) 1994, Regents of the University of California
   8  *
   9  *
  10  * IDENTIFICATION
  11  *        src/backend/storage/lmgr/lmgr.c
  12  *
  13  *-------------------------------------------------------------------------
  14  */
  15
  16 #include "postgres.h"
  17
  18 #include "access/subtrans.h"
  19 #include "access/xact.h"
  20 #include "catalog/catalog.h"
  21 #include "commands/progress.h"
  22 #include "miscadmin.h"
  23 #include "pgstat.h"
  24 #include "storage/lmgr.h"
  25 #include "storage/proc.h"
  26 #include "storage/procarray.h"
  27 #include "utils/inval.h"
  28
  29
  30 /*
  31  * Per-backend counter for generating speculative insertion tokens.
  32  *
  33  * This may wrap around, but that's OK as it's only used for the short
  34  * duration between inserting a tuple and checking that there are no (unique)
  35  * constraint violations.  It's theoretically possible that a backend sees a
  36  * tuple that was speculatively inserted by another backend, but before it has
  37  * started waiting on the token, the other backend completes its insertion,
  38  * and then performs 2^32 unrelated insertions.  And after all that, the
  39  * first backend finally calls SpeculativeInsertionLockAcquire(), with the
  40  * intention of waiting for the first insertion to complete, but ends up
  41  * waiting for the latest unrelated insertion instead.  Even then, nothing
  42  * particularly bad happens: in the worst case they deadlock, causing one of
  43  * the transactions to abort.
  44  */
  45 static uint32 speculativeInsertionToken = 0;
  46
  47
  48 /*
  49  * Struct to hold context info for transaction lock waits.
  50  *
  51  * 'oper' is the operation that needs to wait for the other transaction; 'rel'
  52  * and 'ctid' specify the address of the tuple being waited for.
  53  */
  54 typedef struct XactLockTableWaitInfo
  55 {
  56         XLTW_Oper       oper;
  57         Relation        rel;
  58         ItemPointer ctid;
  59 } XactLockTableWaitInfo;
  60
  61 static void XactLockTableWaitErrorCb(void *arg);
  62
  63 /*
  64  * RelationInitLockInfo
  65  *              Initializes the lock information in a relation descriptor.
  66  *
  67  *              relcache.c must call this during creation of any reldesc.
  68  */
  69 void
  70 RelationInitLockInfo(Relation relation)
  71 {
  72         Assert(RelationIsValid(relation));
  73         Assert(OidIsValid(RelationGetRelid(relation)));
  74
  75         relation->rd_lockInfo.lockRelId.relId = RelationGetRelid(relation);
  76
  77         if (relation->rd_rel->relisshared)
  78                 relation->rd_lockInfo.lockRelId.dbId = InvalidOid;
  79         else
  80                 relation->rd_lockInfo.lockRelId.dbId = MyDatabaseId;
  81 }
  82
  83 /*
  84  * SetLocktagRelationOid
  85  *              Set up a locktag for a relation, given only relation OID
  86  */
  87 static inline void
  88 SetLocktagRelationOid(LOCKTAG *tag, Oid relid)
  89 {
  90         Oid                     dbid;
  91
  92         if (IsSharedRelation(relid))
  93                 dbid = InvalidOid;
  94         else
  95                 dbid = MyDatabaseId;
  96
  97         SET_LOCKTAG_RELATION(*tag, dbid, relid);
  98 }
  99
 100 /*
 101  *              LockRelationOid
 102  *
 103  * Lock a relation given only its OID.  This should generally be used
 104  * before attempting to open the relation's relcache entry.
 105  */
 106 void
 107 LockRelationOid(Oid relid, LOCKMODE lockmode)
 108 {
 109         LOCKTAG         tag;
 110         LOCALLOCK  *locallock;
 111         LockAcquireResult res;
 112
 113         SetLocktagRelationOid(&tag, relid);
 114
 115         res = LockAcquireExtended(&tag, lockmode, false, false, true, &locallock);
 116
 117         /*
 118          * Now that we have the lock, check for invalidation messages, so that we
 119          * will update or flush any stale relcache entry before we try to use it.
 120          * RangeVarGetRelid() specifically relies on us for this.  We can skip
 121          * this in the not-uncommon case that we already had the same type of lock
 122          * being requested, since then no one else could have modified the
 123          * relcache entry in an undesirable way.  (In the case where our own xact
 124          * modifies the rel, the relcache update happens via
 125          * CommandCounterIncrement, not here.)
 126          *
 127          * However, in corner cases where code acts on tables (usually catalogs)
 128          * recursively, we might get here while still processing invalidation
 129          * messages in some outer execution of this function or a sibling.  The
 130          * "cleared" status of the lock tells us whether we really are done
 131          * absorbing relevant inval messages.
 132          */
 133         if (res != LOCKACQUIRE_ALREADY_CLEAR)
 134         {
 135                 AcceptInvalidationMessages();
 136                 MarkLockClear(locallock);
 137         }
 138 }
 139
 140 /*
 141  *              ConditionalLockRelationOid
 142  *
 143  * As above, but only lock if we can get the lock without blocking.
 144  * Returns true iff the lock was acquired.
 145  *
 146  * NOTE: we do not currently need conditional versions of all the
 147  * LockXXX routines in this file, but they could easily be added if needed.
 148  */
 149 bool
 150 ConditionalLockRelationOid(Oid relid, LOCKMODE lockmode)
 151 {
 152         LOCKTAG         tag;
 153         LOCALLOCK  *locallock;
 154         LockAcquireResult res;
 155
 156         SetLocktagRelationOid(&tag, relid);
 157
 158         res = LockAcquireExtended(&tag, lockmode, false, true, true, &locallock);
 159
 160         if (res == LOCKACQUIRE_NOT_AVAIL)
 161                 return false;
 162
 163         /*
 164          * Now that we have the lock, check for invalidation messages; see notes
 165          * in LockRelationOid.
 166          */
 167         if (res != LOCKACQUIRE_ALREADY_CLEAR)
 168         {
 169                 AcceptInvalidationMessages();
 170                 MarkLockClear(locallock);
 171         }
 172
 173         return true;
 174 }
 175
 176 /*
 177  *              LockRelationId
 178  *
 179  * Lock, given a LockRelId.  Same as LockRelationOid but take LockRelId as an
 180  * input.
 181  */
 182 void
 183 LockRelationId(LockRelId *relid, LOCKMODE lockmode)
 184 {
 185         LOCKTAG         tag;
 186         LOCALLOCK  *locallock;
 187         LockAcquireResult res;
 188
 189         SET_LOCKTAG_RELATION(tag, relid->dbId, relid->relId);
 190
 191         res = LockAcquireExtended(&tag, lockmode, false, false, true, &locallock);
 192
 193         /*
 194          * Now that we have the lock, check for invalidation messages; see notes
 195          * in LockRelationOid.
 196          */
 197         if (res != LOCKACQUIRE_ALREADY_CLEAR)
 198         {
 199                 AcceptInvalidationMessages();
 200                 MarkLockClear(locallock);
 201         }
 202 }
 203
 204 /*
 205  *              UnlockRelationId
 206  *
 207  * Unlock, given a LockRelId.  This is preferred over UnlockRelationOid
 208  * for speed reasons.
 209  */
 210 void
 211 UnlockRelationId(LockRelId *relid, LOCKMODE lockmode)
 212 {
 213         LOCKTAG         tag;
 214
 215         SET_LOCKTAG_RELATION(tag, relid->dbId, relid->relId);
 216
 217         LockRelease(&tag, lockmode, false);
 218 }
 219
 220 /*
 221  *              UnlockRelationOid
 222  *
 223  * Unlock, given only a relation Oid.  Use UnlockRelationId if you can.
 224  */
 225 void
 226 UnlockRelationOid(Oid relid, LOCKMODE lockmode)
 227 {
 228         LOCKTAG         tag;
 229
 230         SetLocktagRelationOid(&tag, relid);
 231
 232         LockRelease(&tag, lockmode, false);
 233 }
 234
 235 /*
 236  *              LockRelation
 237  *
 238  * This is a convenience routine for acquiring an additional lock on an
 239  * already-open relation.  Never try to do "relation_open(foo, NoLock)"
 240  * and then lock with this.
 241  */
 242 void
 243 LockRelation(Relation relation, LOCKMODE lockmode)
 244 {
 245         LOCKTAG         tag;
 246         LOCALLOCK  *locallock;
 247         LockAcquireResult res;
 248
 249         SET_LOCKTAG_RELATION(tag,
 250                                                  relation->rd_lockInfo.lockRelId.dbId,
 251                                                  relation->rd_lockInfo.lockRelId.relId);
 252
 253         res = LockAcquireExtended(&tag, lockmode, false, false, true, &locallock);
 254
 255         /*
 256          * Now that we have the lock, check for invalidation messages; see notes
 257          * in LockRelationOid.
 258          */
 259         if (res != LOCKACQUIRE_ALREADY_CLEAR)
 260         {
 261                 AcceptInvalidationMessages();
 262                 MarkLockClear(locallock);
 263         }
 264 }
 265
 266 /*
 267  *              ConditionalLockRelation
 268  *
 269  * This is a convenience routine for acquiring an additional lock on an
 270  * already-open relation.  Never try to do "relation_open(foo, NoLock)"
 271  * and then lock with this.
 272  */
 273 bool
 274 ConditionalLockRelation(Relation relation, LOCKMODE lockmode)
 275 {
 276         LOCKTAG         tag;
 277         LOCALLOCK  *locallock;
 278         LockAcquireResult res;
 279
 280         SET_LOCKTAG_RELATION(tag,
 281                                                  relation->rd_lockInfo.lockRelId.dbId,
 282                                                  relation->rd_lockInfo.lockRelId.relId);
 283
 284         res = LockAcquireExtended(&tag, lockmode, false, true, true, &locallock);
 285
 286         if (res == LOCKACQUIRE_NOT_AVAIL)
 287                 return false;
 288
 289         /*
 290          * Now that we have the lock, check for invalidation messages; see notes
 291          * in LockRelationOid.
 292          */
 293         if (res != LOCKACQUIRE_ALREADY_CLEAR)
 294         {
 295                 AcceptInvalidationMessages();
 296                 MarkLockClear(locallock);
 297         }
 298
 299         return true;
 300 }
 301
 302 /*
 303  *              UnlockRelation
 304  *
 305  * This is a convenience routine for unlocking a relation without also
 306  * closing it.
 307  */
 308 void
 309 UnlockRelation(Relation relation, LOCKMODE lockmode)
 310 {
 311         LOCKTAG         tag;
 312
 313         SET_LOCKTAG_RELATION(tag,
 314                                                  relation->rd_lockInfo.lockRelId.dbId,
 315                                                  relation->rd_lockInfo.lockRelId.relId);
 316
 317         LockRelease(&tag, lockmode, false);
 318 }
 319
 320 /*
 321  *              CheckRelationLockedByMe
 322  *
 323  * Returns true if current transaction holds a lock on 'relation' of mode
 324  * 'lockmode'.  If 'orstronger' is true, a stronger lockmode is also OK.
 325  * ("Stronger" is defined as "numerically higher", which is a bit
 326  * semantically dubious but is OK for the purposes we use this for.)
 327  */
 328 bool
 329 CheckRelationLockedByMe(Relation relation, LOCKMODE lockmode, bool orstronger)
 330 {
 331         LOCKTAG         tag;
 332
 333         SET_LOCKTAG_RELATION(tag,
 334                                                  relation->rd_lockInfo.lockRelId.dbId,
 335                                                  relation->rd_lockInfo.lockRelId.relId);
 336
 337         return LockHeldByMe(&tag, lockmode, orstronger);
 338 }
 339
 340 /*
 341  *              CheckRelationOidLockedByMe
 342  *
 343  * Like the above, but takes an OID as argument.
 344  */
 345 bool
 346 CheckRelationOidLockedByMe(Oid relid, LOCKMODE lockmode, bool orstronger)
 347 {
 348         LOCKTAG         tag;
 349
 350         SetLocktagRelationOid(&tag, relid);
 351
 352         return LockHeldByMe(&tag, lockmode, orstronger);
 353 }
 354
 355 /*
 356  *              LockHasWaitersRelation
 357  *
 358  * This is a function to check whether someone else is waiting for a
 359  * lock which we are currently holding.
 360  */
 361 bool
 362 LockHasWaitersRelation(Relation relation, LOCKMODE lockmode)
 363 {
 364         LOCKTAG         tag;
 365
 366         SET_LOCKTAG_RELATION(tag,
 367                                                  relation->rd_lockInfo.lockRelId.dbId,
 368                                                  relation->rd_lockInfo.lockRelId.relId);
 369
 370         return LockHasWaiters(&tag, lockmode, false);
 371 }
 372
 373 /*
 374  *              LockRelationIdForSession
 375  *
 376  * This routine grabs a session-level lock on the target relation.  The
 377  * session lock persists across transaction boundaries.  It will be removed
 378  * when UnlockRelationIdForSession() is called, or if an ereport(ERROR) occurs,
 379  * or if the backend exits.
 380  *
 381  * Note that one should also grab a transaction-level lock on the rel
 382  * in any transaction that actually uses the rel, to ensure that the
 383  * relcache entry is up to date.
 384  */
 385 void
 386 LockRelationIdForSession(LockRelId *relid, LOCKMODE lockmode)
 387 {
 388         LOCKTAG         tag;
 389
 390         SET_LOCKTAG_RELATION(tag, relid->dbId, relid->relId);
 391
 392         (void) LockAcquire(&tag, lockmode, true, false);
 393 }
 394
 395 /*
 396  *              UnlockRelationIdForSession
 397  */
 398 void
 399 UnlockRelationIdForSession(LockRelId *relid, LOCKMODE lockmode)
 400 {
 401         LOCKTAG         tag;
 402
 403         SET_LOCKTAG_RELATION(tag, relid->dbId, relid->relId);
 404
 405         LockRelease(&tag, lockmode, true);
 406 }
 407
 408 /*
 409  *              LockRelationForExtension
 410  *
 411  * This lock tag is used to interlock addition of pages to relations.
 412  * We need such locking because bufmgr/smgr definition of P_NEW is not
 413  * race-condition-proof.
 414  *
 415  * We assume the caller is already holding some type of regular lock on
 416  * the relation, so no AcceptInvalidationMessages call is needed here.
 417  */
 418 void
 419 LockRelationForExtension(Relation relation, LOCKMODE lockmode)
 420 {
 421         LOCKTAG         tag;
 422
 423         SET_LOCKTAG_RELATION_EXTEND(tag,
 424                                                                 relation->rd_lockInfo.lockRelId.dbId,
 425                                                                 relation->rd_lockInfo.lockRelId.relId);
 426
 427         (void) LockAcquire(&tag, lockmode, false, false);
 428 }
 429
 430 /*
 431  *              ConditionalLockRelationForExtension
 432  *
 433  * As above, but only lock if we can get the lock without blocking.
 434  * Returns true iff the lock was acquired.
 435  */
 436 bool
 437 ConditionalLockRelationForExtension(Relation relation, LOCKMODE lockmode)
 438 {
 439         LOCKTAG         tag;
 440
 441         SET_LOCKTAG_RELATION_EXTEND(tag,
 442                                                                 relation->rd_lockInfo.lockRelId.dbId,
 443                                                                 relation->rd_lockInfo.lockRelId.relId);
 444
 445         return (LockAcquire(&tag, lockmode, false, true) != LOCKACQUIRE_NOT_AVAIL);
 446 }
 447
 448 /*
 449  *              RelationExtensionLockWaiterCount
 450  *
 451  * Count the number of processes waiting for the given relation extension lock.
 452  */
 453 int
 454 RelationExtensionLockWaiterCount(Relation relation)
 455 {
 456         LOCKTAG         tag;
 457
 458         SET_LOCKTAG_RELATION_EXTEND(tag,
 459                                                                 relation->rd_lockInfo.lockRelId.dbId,
 460                                                                 relation->rd_lockInfo.lockRelId.relId);
 461
 462         return LockWaiterCount(&tag);
 463 }
 464
 465 /*
 466  *              UnlockRelationForExtension
 467  */
 468 void
 469 UnlockRelationForExtension(Relation relation, LOCKMODE lockmode)
 470 {
 471         LOCKTAG         tag;
 472
 473         SET_LOCKTAG_RELATION_EXTEND(tag,
 474                                                                 relation->rd_lockInfo.lockRelId.dbId,
 475                                                                 relation->rd_lockInfo.lockRelId.relId);
 476
 477         LockRelease(&tag, lockmode, false);
 478 }
 479
 480 /*
 481  *              LockDatabaseFrozenIds
 482  *
 483  * This allows one backend per database to execute vac_update_datfrozenxid().
 484  */
 485 void
 486 LockDatabaseFrozenIds(LOCKMODE lockmode)
 487 {
 488         LOCKTAG         tag;
 489
 490         SET_LOCKTAG_DATABASE_FROZEN_IDS(tag, MyDatabaseId);
 491
 492         (void) LockAcquire(&tag, lockmode, false, false);
 493 }
 494
 495 /*
 496  *              LockPage
 497  *
 498  * Obtain a page-level lock.  This is currently used by some index access
 499  * methods to lock individual index pages.
 500  */
 501 void
 502 LockPage(Relation relation, BlockNumber blkno, LOCKMODE lockmode)
 503 {
 504         LOCKTAG         tag;
 505
 506         SET_LOCKTAG_PAGE(tag,
 507                                          relation->rd_lockInfo.lockRelId.dbId,
 508                                          relation->rd_lockInfo.lockRelId.relId,
 509                                          blkno);
 510
 511         (void) LockAcquire(&tag, lockmode, false, false);
 512 }
 513
 514 /*
 515  *              ConditionalLockPage
 516  *
 517  * As above, but only lock if we can get the lock without blocking.
 518  * Returns true iff the lock was acquired.
 519  */
 520 bool
 521 ConditionalLockPage(Relation relation, BlockNumber blkno, LOCKMODE lockmode)
 522 {
 523         LOCKTAG         tag;
 524
 525         SET_LOCKTAG_PAGE(tag,
 526                                          relation->rd_lockInfo.lockRelId.dbId,
 527                                          relation->rd_lockInfo.lockRelId.relId,
 528                                          blkno);
 529
 530         return (LockAcquire(&tag, lockmode, false, true) != LOCKACQUIRE_NOT_AVAIL);
 531 }
 532
 533 /*
 534  *              UnlockPage
 535  */
 536 void
 537 UnlockPage(Relation relation, BlockNumber blkno, LOCKMODE lockmode)
 538 {
 539         LOCKTAG         tag;
 540
 541         SET_LOCKTAG_PAGE(tag,
 542                                          relation->rd_lockInfo.lockRelId.dbId,
 543                                          relation->rd_lockInfo.lockRelId.relId,
 544                                          blkno);
 545
 546         LockRelease(&tag, lockmode, false);
 547 }
 548
 549 /*
 550  *              LockTuple
 551  *
 552  * Obtain a tuple-level lock.  This is used in a less-than-intuitive fashion
 553  * because we can't afford to keep a separate lock in shared memory for every
 554  * tuple.  See heap_lock_tuple before using this!
 555  */
 556 void
 557 LockTuple(Relation relation, ItemPointer tid, LOCKMODE lockmode)
 558 {
 559         LOCKTAG         tag;
 560
 561         SET_LOCKTAG_TUPLE(tag,
 562                                           relation->rd_lockInfo.lockRelId.dbId,
 563                                           relation->rd_lockInfo.lockRelId.relId,
 564                                           ItemPointerGetBlockNumber(tid),
 565                                           ItemPointerGetOffsetNumber(tid));
 566
 567         (void) LockAcquire(&tag, lockmode, false, false);
 568 }
 569
 570 /*
 571  *              ConditionalLockTuple
 572  *
 573  * As above, but only lock if we can get the lock without blocking.
 574  * Returns true iff the lock was acquired.
 575  */
 576 bool
 577 ConditionalLockTuple(Relation relation, ItemPointer tid, LOCKMODE lockmode)
 578 {
 579         LOCKTAG         tag;
 580
 581         SET_LOCKTAG_TUPLE(tag,
 582                                           relation->rd_lockInfo.lockRelId.dbId,
 583                                           relation->rd_lockInfo.lockRelId.relId,
 584                                           ItemPointerGetBlockNumber(tid),
 585                                           ItemPointerGetOffsetNumber(tid));
 586
 587         return (LockAcquire(&tag, lockmode, false, true) != LOCKACQUIRE_NOT_AVAIL);
 588 }
 589
 590 /*
 591  *              UnlockTuple
 592  */
 593 void
 594 UnlockTuple(Relation relation, ItemPointer tid, LOCKMODE lockmode)
 595 {
 596         LOCKTAG         tag;
 597
 598         SET_LOCKTAG_TUPLE(tag,
 599                                           relation->rd_lockInfo.lockRelId.dbId,
 600                                           relation->rd_lockInfo.lockRelId.relId,
 601                                           ItemPointerGetBlockNumber(tid),
 602                                           ItemPointerGetOffsetNumber(tid));
 603
 604         LockRelease(&tag, lockmode, false);
 605 }
 606
 607 /*
 608  *              XactLockTableInsert
 609  *
 610  * Insert a lock showing that the given transaction ID is running ---
 611  * this is done when an XID is acquired by a transaction or subtransaction.
 612  * The lock can then be used to wait for the transaction to finish.
 613  */
 614 void
 615 XactLockTableInsert(TransactionId xid)
 616 {
 617         LOCKTAG         tag;
 618
 619         SET_LOCKTAG_TRANSACTION(tag, xid);
 620
 621         (void) LockAcquire(&tag, ExclusiveLock, false, false);
 622 }
 623
 624 /*
 625  *              XactLockTableDelete
 626  *
 627  * Delete the lock showing that the given transaction ID is running.
 628  * (This is never used for main transaction IDs; those locks are only
 629  * released implicitly at transaction end.  But we do use it for subtrans IDs.)
 630  */
 631 void
 632 XactLockTableDelete(TransactionId xid)
 633 {
 634         LOCKTAG         tag;
 635
 636         SET_LOCKTAG_TRANSACTION(tag, xid);
 637
 638         LockRelease(&tag, ExclusiveLock, false);
 639 }
 640
 641 /*
 642  *              XactLockTableWait
 643  *
 644  * Wait for the specified transaction to commit or abort.  If an operation
 645  * is specified, an error context callback is set up.  If 'oper' is passed as
 646  * None, no error context callback is set up.
 647  *
 648  * Note that this does the right thing for subtransactions: if we wait on a
 649  * subtransaction, we will exit as soon as it aborts or its top parent commits.
 650  * It takes some extra work to ensure this, because to save on shared memory
 651  * the XID lock of a subtransaction is released when it ends, whether
 652  * successfully or unsuccessfully.  So we have to check if it's "still running"
 653  * and if so wait for its parent.
 654  */
 655 void
 656 XactLockTableWait(TransactionId xid, Relation rel, ItemPointer ctid,
 657                                   XLTW_Oper oper)
 658 {
 659         LOCKTAG         tag;
 660         XactLockTableWaitInfo info;
 661         ErrorContextCallback callback;
 662         bool            first = true;
 663
 664         /*
 665          * If an operation is specified, set up our verbose error context
 666          * callback.
 667          */
 668         if (oper != XLTW_None)
 669         {
 670                 Assert(RelationIsValid(rel));
 671                 Assert(ItemPointerIsValid(ctid));
 672
 673                 info.rel = rel;
 674                 info.ctid = ctid;
 675                 info.oper = oper;
 676
 677                 callback.callback = XactLockTableWaitErrorCb;
 678                 callback.arg = &info;
 679                 callback.previous = error_context_stack;
 680                 error_context_stack = &callback;
 681         }
 682
 683         for (;;)
 684         {
 685                 Assert(TransactionIdIsValid(xid));
 686                 Assert(!TransactionIdEquals(xid, GetTopTransactionIdIfAny()));
 687
 688                 SET_LOCKTAG_TRANSACTION(tag, xid);
 689
 690                 (void) LockAcquire(&tag, ShareLock, false, false);
 691
 692                 LockRelease(&tag, ShareLock, false);
 693
 694                 if (!TransactionIdIsInProgress(xid))
 695                         break;
 696
 697                 /*
 698                  * If the Xid belonged to a subtransaction, then the lock would have
 699                  * gone away as soon as it was finished; for correct tuple visibility,
 700                  * the right action is to wait on its parent transaction to go away.
 701                  * But instead of going levels up one by one, we can just wait for the
 702                  * topmost transaction to finish with the same end result, which also
 703                  * incurs less locktable traffic.
 704                  *
 705                  * Some uses of this function don't involve tuple visibility -- such
 706                  * as when building snapshots for logical decoding.  It is possible to
 707                  * see a transaction in ProcArray before it registers itself in the
 708                  * locktable.  The topmost transaction in that case is the same xid,
 709                  * so we try again after a short sleep.  (Don't sleep the first time
 710                  * through, to avoid slowing down the normal case.)
 711                  */
 712                 if (!first)
 713                         pg_usleep(1000L);
 714                 first = false;
 715                 xid = SubTransGetTopmostTransaction(xid);
 716         }
 717
 718         if (oper != XLTW_None)
 719                 error_context_stack = callback.previous;
 720 }
 721
 722 /*
 723  *              ConditionalXactLockTableWait
 724  *
 725  * As above, but only lock if we can get the lock without blocking.
 726  * Returns true if the lock was acquired.
 727  */
 728 bool
 729 ConditionalXactLockTableWait(TransactionId xid)
 730 {
 731         LOCKTAG         tag;
 732         bool            first = true;
 733
 734         for (;;)
 735         {
 736                 Assert(TransactionIdIsValid(xid));
 737                 Assert(!TransactionIdEquals(xid, GetTopTransactionIdIfAny()));
 738
 739                 SET_LOCKTAG_TRANSACTION(tag, xid);
 740
 741                 if (LockAcquire(&tag, ShareLock, false, true) == LOCKACQUIRE_NOT_AVAIL)
 742                         return false;
 743
 744                 LockRelease(&tag, ShareLock, false);
 745
 746                 if (!TransactionIdIsInProgress(xid))
 747                         break;
 748
 749                 /* See XactLockTableWait about this case */
 750                 if (!first)
 751                         pg_usleep(1000L);
 752                 first = false;
 753                 xid = SubTransGetTopmostTransaction(xid);
 754         }
 755
 756         return true;
 757 }
 758
 759 /*
 760  *              SpeculativeInsertionLockAcquire
 761  *
 762  * Insert a lock showing that the given transaction ID is inserting a tuple,
 763  * but hasn't yet decided whether it's going to keep it.  The lock can then be
 764  * used to wait for the decision to go ahead with the insertion, or aborting
 765  * it.
 766  *
 767  * The token is used to distinguish multiple insertions by the same
 768  * transaction.  It is returned to caller.
 769  */
 770 uint32
 771 SpeculativeInsertionLockAcquire(TransactionId xid)
 772 {
 773         LOCKTAG         tag;
 774
 775         speculativeInsertionToken++;
 776
 777         /*
 778          * Check for wrap-around. Zero means no token is held, so don't use that.
 779          */
 780         if (speculativeInsertionToken == 0)
 781                 speculativeInsertionToken = 1;
 782
 783         SET_LOCKTAG_SPECULATIVE_INSERTION(tag, xid, speculativeInsertionToken);
 784
 785         (void) LockAcquire(&tag, ExclusiveLock, false, false);
 786
 787         return speculativeInsertionToken;
 788 }
 789
 790 /*
 791  *              SpeculativeInsertionLockRelease
 792  *
 793  * Delete the lock showing that the given transaction is speculatively
 794  * inserting a tuple.
 795  */
 796 void
 797 SpeculativeInsertionLockRelease(TransactionId xid)
 798 {
 799         LOCKTAG         tag;
 800
 801         SET_LOCKTAG_SPECULATIVE_INSERTION(tag, xid, speculativeInsertionToken);
 802
 803         LockRelease(&tag, ExclusiveLock, false);
 804 }
 805
 806 /*
 807  *              SpeculativeInsertionWait
 808  *
 809  * Wait for the specified transaction to finish or abort the insertion of a
 810  * tuple.
 811  */
 812 void
 813 SpeculativeInsertionWait(TransactionId xid, uint32 token)
 814 {
 815         LOCKTAG         tag;
 816
 817         SET_LOCKTAG_SPECULATIVE_INSERTION(tag, xid, token);
 818
 819         Assert(TransactionIdIsValid(xid));
 820         Assert(token != 0);
 821
 822         (void) LockAcquire(&tag, ShareLock, false, false);
 823         LockRelease(&tag, ShareLock, false);
 824 }
 825
 826 /*
 827  * XactLockTableWaitErrorCb
 828  *              Error context callback for transaction lock waits.
 829  */
 830 static void
 831 XactLockTableWaitErrorCb(void *arg)
 832 {
 833         XactLockTableWaitInfo *info = (XactLockTableWaitInfo *) arg;
 834
 835         /*
 836          * We would like to print schema name too, but that would require a
 837          * syscache lookup.
 838          */
 839         if (info->oper != XLTW_None &&
 840                 ItemPointerIsValid(info->ctid) && RelationIsValid(info->rel))
 841         {
 842                 const char *cxt;
 843
 844                 switch (info->oper)
 845                 {
 846                         case XLTW_Update:
 847                                 cxt = gettext_noop("while updating tuple (%u,%u) in relation \"%s\"");
 848                                 break;
 849                         case XLTW_Delete:
 850                                 cxt = gettext_noop("while deleting tuple (%u,%u) in relation \"%s\"");
 851                                 break;
 852                         case XLTW_Lock:
 853                                 cxt = gettext_noop("while locking tuple (%u,%u) in relation \"%s\"");
 854                                 break;
 855                         case XLTW_LockUpdated:
 856                                 cxt = gettext_noop("while locking updated version (%u,%u) of tuple in relation \"%s\"");
 857                                 break;
 858                         case XLTW_InsertIndex:
 859                                 cxt = gettext_noop("while inserting index tuple (%u,%u) in relation \"%s\"");
 860                                 break;
 861                         case XLTW_InsertIndexUnique:
 862                                 cxt = gettext_noop("while checking uniqueness of tuple (%u,%u) in relation \"%s\"");
 863                                 break;
 864                         case XLTW_FetchUpdated:
 865                                 cxt = gettext_noop("while rechecking updated tuple (%u,%u) in relation \"%s\"");
 866                                 break;
 867                         case XLTW_RecheckExclusionConstr:
 868                                 cxt = gettext_noop("while checking exclusion constraint on tuple (%u,%u) in relation \"%s\"");
 869                                 break;
 870
 871                         default:
 872                                 return;
 873                 }
 874
 875                 errcontext(cxt,
 876                                    ItemPointerGetBlockNumber(info->ctid),
 877                                    ItemPointerGetOffsetNumber(info->ctid),
 878                                    RelationGetRelationName(info->rel));
 879         }
 880 }
 881
 882 /*
 883  * WaitForLockersMultiple
 884  *              Wait until no transaction holds locks that conflict with the given
 885  *              locktags at the given lockmode.
 886  *
 887  * To do this, obtain the current list of lockers, and wait on their VXIDs
 888  * until they are finished.
 889  *
 890  * Note we don't try to acquire the locks on the given locktags, only the
 891  * VXIDs and XIDs of their lock holders; if somebody grabs a conflicting lock
 892  * on the objects after we obtained our initial list of lockers, we will not
 893  * wait for them.
 894  */
 895 void
 896 WaitForLockersMultiple(List *locktags, LOCKMODE lockmode, bool progress)
 897 {
 898         List       *holders = NIL;
 899         ListCell   *lc;
 900         int                     total = 0;
 901         int                     done = 0;
 902
 903         /* Done if no locks to wait for */
 904         if (locktags == NIL)
 905                 return;
 906
 907         /* Collect the transactions we need to wait on */
 908         foreach(lc, locktags)
 909         {
 910                 LOCKTAG    *locktag = lfirst(lc);
 911                 int                     count;
 912
 913                 holders = lappend(holders,
 914                                                   GetLockConflicts(locktag, lockmode,
 915                                                                                    progress ? &count : NULL));
 916                 if (progress)
 917                         total += count;
 918         }
 919
 920         if (progress)
 921                 pgstat_progress_update_param(PROGRESS_WAITFOR_TOTAL, total);
 922
 923         /*
 924          * Note: GetLockConflicts() never reports our own xid, hence we need not
 925          * check for that.  Also, prepared xacts are reported and awaited.
 926          */
 927
 928         /* Finally wait for each such transaction to complete */
 929         foreach(lc, holders)
 930         {
 931                 VirtualTransactionId *lockholders = lfirst(lc);
 932
 933                 while (VirtualTransactionIdIsValid(*lockholders))
 934                 {
 935                         /* If requested, publish who we're going to wait for. */
 936                         if (progress)
 937                         {
 938                                 PGPROC     *holder = ProcNumberGetProc(lockholders->procNumber);
 939
 940                                 if (holder)
 941                                         pgstat_progress_update_param(PROGRESS_WAITFOR_CURRENT_PID,
 942                                                                                                  holder->pid);
 943                         }
 944                         VirtualXactLock(*lockholders, true);
 945                         lockholders++;
 946
 947                         if (progress)
 948                                 pgstat_progress_update_param(PROGRESS_WAITFOR_DONE, ++done);
 949                 }
 950         }
 951         if (progress)
 952         {
 953                 const int       index[] = {
 954                         PROGRESS_WAITFOR_TOTAL,
 955                         PROGRESS_WAITFOR_DONE,
 956                         PROGRESS_WAITFOR_CURRENT_PID
 957                 };
 958                 const int64 values[] = {
 959                         0, 0, 0
 960                 };
 961
 962                 pgstat_progress_update_multi_param(3, index, values);
 963         }
 964
 965         list_free_deep(holders);
 966 }
 967
 968 /*
 969  * WaitForLockers
 970  *
 971  * Same as WaitForLockersMultiple, for a single lock tag.
 972  */
 973 void
 974 WaitForLockers(LOCKTAG heaplocktag, LOCKMODE lockmode, bool progress)
 975 {
 976         List       *l;
 977
 978         l = list_make1(&heaplocktag);
 979         WaitForLockersMultiple(l, lockmode, progress);
 980         list_free(l);
 981 }
 982
 983
 984 /*
 985  *              LockDatabaseObject
 986  *
 987  * Obtain a lock on a general object of the current database.  Don't use
 988  * this for shared objects (such as tablespaces).  It's unwise to apply it
 989  * to relations, also, since a lock taken this way will NOT conflict with
 990  * locks taken via LockRelation and friends.
 991  */
 992 void
 993 LockDatabaseObject(Oid classid, Oid objid, uint16 objsubid,
 994                                    LOCKMODE lockmode)
 995 {
 996         LOCKTAG         tag;
 997
 998         SET_LOCKTAG_OBJECT(tag,
 999                                            MyDatabaseId,
1000                                            classid,
1001                                            objid,
1002                                            objsubid);
1003
1004         (void) LockAcquire(&tag, lockmode, false, false);
1005
1006         /* Make sure syscaches are up-to-date with any changes we waited for */
1007         AcceptInvalidationMessages();
1008 }
1009
1010 /*
1011  *              ConditionalLockDatabaseObject
1012  *
1013  * As above, but only lock if we can get the lock without blocking.
1014  * Returns true iff the lock was acquired.
1015  */
1016 bool
1017 ConditionalLockDatabaseObject(Oid classid, Oid objid, uint16 objsubid,
1018                                                           LOCKMODE lockmode)
1019 {
1020         LOCKTAG         tag;
1021         LOCALLOCK  *locallock;
1022         LockAcquireResult res;
1023
1024         SET_LOCKTAG_OBJECT(tag,
1025                                            MyDatabaseId,
1026                                            classid,
1027                                            objid,
1028                                            objsubid);
1029
1030         res = LockAcquireExtended(&tag, lockmode, false, true, true, &locallock);
1031
1032         if (res == LOCKACQUIRE_NOT_AVAIL)
1033                 return false;
1034
1035         /*
1036          * Now that we have the lock, check for invalidation messages; see notes
1037          * in LockRelationOid.
1038          */
1039         if (res != LOCKACQUIRE_ALREADY_CLEAR)
1040         {
1041                 AcceptInvalidationMessages();
1042                 MarkLockClear(locallock);
1043         }
1044
1045         return true;
1046 }
1047
1048 /*
1049  *              UnlockDatabaseObject
1050  */
1051 void
1052 UnlockDatabaseObject(Oid classid, Oid objid, uint16 objsubid,
1053                                          LOCKMODE lockmode)
1054 {
1055         LOCKTAG         tag;
1056
1057         SET_LOCKTAG_OBJECT(tag,
1058                                            MyDatabaseId,
1059                                            classid,
1060                                            objid,
1061                                            objsubid);
1062
1063         LockRelease(&tag, lockmode, false);
1064 }
1065
1066 /*
1067  *              LockSharedObject
1068  *
1069  * Obtain a lock on a shared-across-databases object.
1070  */
1071 void
1072 LockSharedObject(Oid classid, Oid objid, uint16 objsubid,
1073                                  LOCKMODE lockmode)
1074 {
1075         LOCKTAG         tag;
1076
1077         SET_LOCKTAG_OBJECT(tag,
1078                                            InvalidOid,
1079                                            classid,
1080                                            objid,
1081                                            objsubid);
1082
1083         (void) LockAcquire(&tag, lockmode, false, false);
1084
1085         /* Make sure syscaches are up-to-date with any changes we waited for */
1086         AcceptInvalidationMessages();
1087 }
1088
1089 /*
1090  *              ConditionalLockSharedObject
1091  *
1092  * As above, but only lock if we can get the lock without blocking.
1093  * Returns true iff the lock was acquired.
1094  */
1095 bool
1096 ConditionalLockSharedObject(Oid classid, Oid objid, uint16 objsubid,
1097                                                         LOCKMODE lockmode)
1098 {
1099         LOCKTAG         tag;
1100         LOCALLOCK  *locallock;
1101         LockAcquireResult res;
1102
1103         SET_LOCKTAG_OBJECT(tag,
1104                                            InvalidOid,
1105                                            classid,
1106                                            objid,
1107                                            objsubid);
1108
1109         res = LockAcquireExtended(&tag, lockmode, false, true, true, &locallock);
1110
1111         if (res == LOCKACQUIRE_NOT_AVAIL)
1112                 return false;
1113
1114         /*
1115          * Now that we have the lock, check for invalidation messages; see notes
1116          * in LockRelationOid.
1117          */
1118         if (res != LOCKACQUIRE_ALREADY_CLEAR)
1119         {
1120                 AcceptInvalidationMessages();
1121                 MarkLockClear(locallock);
1122         }
1123
1124         return true;
1125 }
1126
1127 /*
1128  *              UnlockSharedObject
1129  */
1130 void
1131 UnlockSharedObject(Oid classid, Oid objid, uint16 objsubid,
1132                                    LOCKMODE lockmode)
1133 {
1134         LOCKTAG         tag;
1135
1136         SET_LOCKTAG_OBJECT(tag,
1137                                            InvalidOid,
1138                                            classid,
1139                                            objid,
1140                                            objsubid);
1141
1142         LockRelease(&tag, lockmode, false);
1143 }
1144
1145 /*
1146  *              LockSharedObjectForSession
1147  *
1148  * Obtain a session-level lock on a shared-across-databases object.
1149  * See LockRelationIdForSession for notes about session-level locks.
1150  */
1151 void
1152 LockSharedObjectForSession(Oid classid, Oid objid, uint16 objsubid,
1153                                                    LOCKMODE lockmode)
1154 {
1155         LOCKTAG         tag;
1156
1157         SET_LOCKTAG_OBJECT(tag,
1158                                            InvalidOid,
1159                                            classid,
1160                                            objid,
1161                                            objsubid);
1162
1163         (void) LockAcquire(&tag, lockmode, true, false);
1164 }
1165
1166 /*
1167  *              UnlockSharedObjectForSession
1168  */
1169 void
1170 UnlockSharedObjectForSession(Oid classid, Oid objid, uint16 objsubid,
1171                                                          LOCKMODE lockmode)
1172 {
1173         LOCKTAG         tag;
1174
1175         SET_LOCKTAG_OBJECT(tag,
1176                                            InvalidOid,
1177                                            classid,
1178                                            objid,
1179                                            objsubid);
1180
1181         LockRelease(&tag, lockmode, true);
1182 }
1183
1184 /*
1185  *              LockApplyTransactionForSession
1186  *
1187  * Obtain a session-level lock on a transaction being applied on a logical
1188  * replication subscriber. See LockRelationIdForSession for notes about
1189  * session-level locks.
1190  */
1191 void
1192 LockApplyTransactionForSession(Oid suboid, TransactionId xid, uint16 objid,
1193                                                            LOCKMODE lockmode)
1194 {
1195         LOCKTAG         tag;
1196
1197         SET_LOCKTAG_APPLY_TRANSACTION(tag,
1198                                                                   MyDatabaseId,
1199                                                                   suboid,
1200                                                                   xid,
1201                                                                   objid);
1202
1203         (void) LockAcquire(&tag, lockmode, true, false);
1204 }
1205
1206 /*
1207  *              UnlockApplyTransactionForSession
1208  */
1209 void
1210 UnlockApplyTransactionForSession(Oid suboid, TransactionId xid, uint16 objid,
1211                                                                  LOCKMODE lockmode)
1212 {
1213         LOCKTAG         tag;
1214
1215         SET_LOCKTAG_APPLY_TRANSACTION(tag,
1216                                                                   MyDatabaseId,
1217                                                                   suboid,
1218                                                                   xid,
1219                                                                   objid);
1220
1221         LockRelease(&tag, lockmode, true);
1222 }
1223
1224 /*
1225  * Append a description of a lockable object to buf.
1226  *
1227  * Ideally we would print names for the numeric values, but that requires
1228  * getting locks on system tables, which might cause problems since this is
1229  * typically used to report deadlock situations.
1230  */
1231 void
1232 DescribeLockTag(StringInfo buf, const LOCKTAG *tag)
1233 {
1234         switch ((LockTagType) tag->locktag_type)
1235         {
1236                 case LOCKTAG_RELATION:
1237                         appendStringInfo(buf,
1238                                                          _("relation %u of database %u"),
1239                                                          tag->locktag_field2,
1240                                                          tag->locktag_field1);
1241                         break;
1242                 case LOCKTAG_RELATION_EXTEND:
1243                         appendStringInfo(buf,
1244                                                          _("extension of relation %u of database %u"),
1245                                                          tag->locktag_field2,
1246                                                          tag->locktag_field1);
1247                         break;
1248                 case LOCKTAG_DATABASE_FROZEN_IDS:
1249                         appendStringInfo(buf,
1250                                                          _("pg_database.datfrozenxid of database %u"),
1251                                                          tag->locktag_field1);
1252                         break;
1253                 case LOCKTAG_PAGE:
1254                         appendStringInfo(buf,
1255                                                          _("page %u of relation %u of database %u"),
1256                                                          tag->locktag_field3,
1257                                                          tag->locktag_field2,
1258                                                          tag->locktag_field1);
1259                         break;
1260                 case LOCKTAG_TUPLE:
1261                         appendStringInfo(buf,
1262                                                          _("tuple (%u,%u) of relation %u of database %u"),
1263                                                          tag->locktag_field3,
1264                                                          tag->locktag_field4,
1265                                                          tag->locktag_field2,
1266                                                          tag->locktag_field1);
1267                         break;
1268                 case LOCKTAG_TRANSACTION:
1269                         appendStringInfo(buf,
1270                                                          _("transaction %u"),
1271                                                          tag->locktag_field1);
1272                         break;
1273                 case LOCKTAG_VIRTUALTRANSACTION:
1274                         appendStringInfo(buf,
1275                                                          _("virtual transaction %d/%u"),
1276                                                          tag->locktag_field1,
1277                                                          tag->locktag_field2);
1278                         break;
1279                 case LOCKTAG_SPECULATIVE_TOKEN:
1280                         appendStringInfo(buf,
1281                                                          _("speculative token %u of transaction %u"),
1282                                                          tag->locktag_field2,
1283                                                          tag->locktag_field1);
1284                         break;
1285                 case LOCKTAG_OBJECT:
1286                         appendStringInfo(buf,
1287                                                          _("object %u of class %u of database %u"),
1288                                                          tag->locktag_field3,
1289                                                          tag->locktag_field2,
1290                                                          tag->locktag_field1);
1291                         break;
1292                 case LOCKTAG_USERLOCK:
1293                         /* reserved for old contrib code, now on pgfoundry */
1294                         appendStringInfo(buf,
1295                                                          _("user lock [%u,%u,%u]"),
1296                                                          tag->locktag_field1,
1297                                                          tag->locktag_field2,
1298                                                          tag->locktag_field3);
1299                         break;
1300                 case LOCKTAG_ADVISORY:
1301                         appendStringInfo(buf,
1302                                                          _("advisory lock [%u,%u,%u,%u]"),
1303                                                          tag->locktag_field1,
1304                                                          tag->locktag_field2,
1305                                                          tag->locktag_field3,
1306                                                          tag->locktag_field4);
1307                         break;
1308                 case LOCKTAG_APPLY_TRANSACTION:
1309                         appendStringInfo(buf,
1310                                                          _("remote transaction %u of subscription %u of database %u"),
1311                                                          tag->locktag_field3,
1312                                                          tag->locktag_field2,
1313                                                          tag->locktag_field1);
1314                         break;
1315                 default:
1316                         appendStringInfo(buf,
1317                                                          _("unrecognized locktag type %d"),
1318                                                          (int) tag->locktag_type);
1319                         break;
1320         }
1321 }
1322
1323 /*
1324  * GetLockNameFromTagType
1325  *
1326  *      Given locktag type, return the corresponding lock name.
1327  */
1328 const char *
1329 GetLockNameFromTagType(uint16 locktag_type)
1330 {
1331         if (locktag_type > LOCKTAG_LAST_TYPE)
1332                 return "???";
1333         return LockTagTypeNames[locktag_type];
1334 }