PostgreSQL Source Code  git master
 All Data Structures Namespaces Files Functions Variables Typedefs Enumerations Enumerator Macros
slot.c
Go to the documentation of this file.
1 /*-------------------------------------------------------------------------
2  *
3  * slot.c
4  * Replication slot management.
5  *
6  *
7  * Copyright (c) 2012-2017, PostgreSQL Global Development Group
8  *
9  *
10  * IDENTIFICATION
11  * src/backend/replication/slot.c
12  *
13  * NOTES
14  *
15  * Replication slots are used to keep state about replication streams
16  * originating from this cluster. Their primary purpose is to prevent the
17  * premature removal of WAL or of old tuple versions in a manner that would
18  * interfere with replication; they are also useful for monitoring purposes.
19  * Slots need to be permanent (to allow restarts), crash-safe, and allocatable
20  * on standbys (to support cascading setups). The requirement that slots be
21  * usable on standbys precludes storing them in the system catalogs.
22  *
23  * Each replication slot gets its own directory inside the $PGDATA/pg_replslot
24  * directory. Inside that directory the state file will contain the slot's
25  * own data. Additional data can be stored alongside that file if required.
26  * While the server is running, the state data is also cached in memory for
27  * efficiency.
28  *
29  * ReplicationSlotAllocationLock must be taken in exclusive mode to allocate
30  * or free a slot. ReplicationSlotControlLock must be taken in shared mode
31  * to iterate over the slots, and in exclusive mode to change the in_use flag
32  * of a slot. The remaining data in each slot is protected by its mutex.
33  *
34  *-------------------------------------------------------------------------
35  */
36 
37 #include "postgres.h"
38 
39 #include <unistd.h>
40 #include <sys/stat.h>
41 
42 #include "access/transam.h"
43 #include "access/xlog_internal.h"
44 #include "common/string.h"
45 #include "miscadmin.h"
46 #include "pgstat.h"
47 #include "replication/slot.h"
48 #include "storage/fd.h"
49 #include "storage/proc.h"
50 #include "storage/procarray.h"
51 #include "utils/builtins.h"
52 
53 /*
54  * Replication slot on-disk data structure.
55  */
56 typedef struct ReplicationSlotOnDisk
57 {
58  /* first part of this struct needs to be version independent */
59 
60  /* data not covered by checksum */
63 
64  /* data covered by checksum */
67 
68  /*
69  * The actual data in the slot that follows can differ based on the above
70  * 'version'.
71  */
72 
75 
76 /* size of version independent data */
77 #define ReplicationSlotOnDiskConstantSize \
78  offsetof(ReplicationSlotOnDisk, slotdata)
79 /* size of the part of the slot not covered by the checksum */
80 #define SnapBuildOnDiskNotChecksummedSize \
81  offsetof(ReplicationSlotOnDisk, version)
82 /* size of the part covered by the checksum */
83 #define SnapBuildOnDiskChecksummedSize \
84  sizeof(ReplicationSlotOnDisk) - SnapBuildOnDiskNotChecksummedSize
85 /* size of the slot data that is version dependent */
86 #define ReplicationSlotOnDiskV2Size \
87  sizeof(ReplicationSlotOnDisk) - ReplicationSlotOnDiskConstantSize
88 
89 #define SLOT_MAGIC 0x1051CA1 /* format identifier */
90 #define SLOT_VERSION 2 /* version for new files */
91 
92 /* Control array for replication slot management */
94 
95 /* My backend's replication slot in the shared memory array */
97 
98 /* GUCs */
99 int max_replication_slots = 0; /* the maximum number of replication
100  * slots */
101 
102 static void ReplicationSlotDropAcquired(void);
103 static void ReplicationSlotDropPtr(ReplicationSlot *slot);
104 
105 /* internal persistency functions */
106 static void RestoreSlotFromDisk(const char *name);
107 static void CreateSlotOnDisk(ReplicationSlot *slot);
108 static void SaveSlotToPath(ReplicationSlot *slot, const char *path, int elevel);
109 
110 /*
111  * Report shared-memory space needed by ReplicationSlotShmemInit.
112  */
113 Size
115 {
116  Size size = 0;
117 
118  if (max_replication_slots == 0)
119  return size;
120 
121  size = offsetof(ReplicationSlotCtlData, replication_slots);
122  size = add_size(size,
124 
125  return size;
126 }
127 
128 /*
129  * Allocate and initialize walsender-related shared memory.
130  */
131 void
133 {
134  bool found;
135 
136  if (max_replication_slots == 0)
137  return;
138 
139  ReplicationSlotCtl = (ReplicationSlotCtlData *)
140  ShmemInitStruct("ReplicationSlot Ctl", ReplicationSlotsShmemSize(),
141  &found);
142 
144  "replication_slot_io");
145 
146  if (!found)
147  {
148  int i;
149 
150  /* First time through, so initialize */
151  MemSet(ReplicationSlotCtl, 0, ReplicationSlotsShmemSize());
152 
153  for (i = 0; i < max_replication_slots; i++)
154  {
155  ReplicationSlot *slot = &ReplicationSlotCtl->replication_slots[i];
156 
157  /* everything else is zeroed by the memset above */
158  SpinLockInit(&slot->mutex);
160  }
161  }
162 }
163 
164 /*
165  * Check whether the passed slot name is valid and report errors at elevel.
166  *
167  * Slot names may consist out of [a-z0-9_]{1,NAMEDATALEN-1} which should allow
168  * the name to be used as a directory name on every supported OS.
169  *
170  * Returns whether the directory name is valid or not if elevel < ERROR.
171  */
172 bool
174 {
175  const char *cp;
176 
177  if (strlen(name) == 0)
178  {
179  ereport(elevel,
180  (errcode(ERRCODE_INVALID_NAME),
181  errmsg("replication slot name \"%s\" is too short",
182  name)));
183  return false;
184  }
185 
186  if (strlen(name) >= NAMEDATALEN)
187  {
188  ereport(elevel,
189  (errcode(ERRCODE_NAME_TOO_LONG),
190  errmsg("replication slot name \"%s\" is too long",
191  name)));
192  return false;
193  }
194 
195  for (cp = name; *cp; cp++)
196  {
197  if (!((*cp >= 'a' && *cp <= 'z')
198  || (*cp >= '0' && *cp <= '9')
199  || (*cp == '_')))
200  {
201  ereport(elevel,
202  (errcode(ERRCODE_INVALID_NAME),
203  errmsg("replication slot name \"%s\" contains invalid character",
204  name),
205  errhint("Replication slot names may only contain lower case letters, numbers, and the underscore character.")));
206  return false;
207  }
208  }
209  return true;
210 }
211 
212 /*
213  * Create a new replication slot and mark it as used by this backend.
214  *
215  * name: Name of the slot
216  * db_specific: logical decoding is db specific; if the slot is going to
217  * be used for that pass true, otherwise false.
218  */
219 void
220 ReplicationSlotCreate(const char *name, bool db_specific,
221  ReplicationSlotPersistency persistency)
222 {
223  ReplicationSlot *slot = NULL;
224  int i;
225 
226  Assert(MyReplicationSlot == NULL);
227 
229 
230  /*
231  * If some other backend ran this code concurrently with us, we'd likely
232  * both allocate the same slot, and that would be bad. We'd also be at
233  * risk of missing a name collision. Also, we don't want to try to create
234  * a new slot while somebody's busy cleaning up an old one, because we
235  * might both be monkeying with the same directory.
236  */
237  LWLockAcquire(ReplicationSlotAllocationLock, LW_EXCLUSIVE);
238 
239  /*
240  * Check for name collision, and identify an allocatable slot. We need to
241  * hold ReplicationSlotControlLock in shared mode for this, so that nobody
242  * else can change the in_use flags while we're looking at them.
243  */
244  LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
245  for (i = 0; i < max_replication_slots; i++)
246  {
247  ReplicationSlot *s = &ReplicationSlotCtl->replication_slots[i];
248 
249  if (s->in_use && strcmp(name, NameStr(s->data.name)) == 0)
250  ereport(ERROR,
252  errmsg("replication slot \"%s\" already exists", name)));
253  if (!s->in_use && slot == NULL)
254  slot = s;
255  }
256  LWLockRelease(ReplicationSlotControlLock);
257 
258  /* If all slots are in use, we're out of luck. */
259  if (slot == NULL)
260  ereport(ERROR,
261  (errcode(ERRCODE_CONFIGURATION_LIMIT_EXCEEDED),
262  errmsg("all replication slots are in use"),
263  errhint("Free one or increase max_replication_slots.")));
264 
265  /*
266  * Since this slot is not in use, nobody should be looking at any part of
267  * it other than the in_use field unless they're trying to allocate it.
268  * And since we hold ReplicationSlotAllocationLock, nobody except us can
269  * be doing that. So it's safe to initialize the slot.
270  */
271  Assert(!slot->in_use);
272  Assert(slot->active_pid == 0);
273 
274  /* first initialize persistent data */
275  memset(&slot->data, 0, sizeof(ReplicationSlotPersistentData));
276  StrNCpy(NameStr(slot->data.name), name, NAMEDATALEN);
277  slot->data.database = db_specific ? MyDatabaseId : InvalidOid;
278  slot->data.persistency = persistency;
279 
280  /* and then data only present in shared memory */
281  slot->just_dirtied = false;
282  slot->dirty = false;
289 
290  /*
291  * Create the slot on disk. We haven't actually marked the slot allocated
292  * yet, so no special cleanup is required if this errors out.
293  */
294  CreateSlotOnDisk(slot);
295 
296  /*
297  * We need to briefly prevent any other backend from iterating over the
298  * slots while we flip the in_use flag. We also need to set the active
299  * flag while holding the ControlLock as otherwise a concurrent
300  * SlotAcquire() could acquire the slot as well.
301  */
302  LWLockAcquire(ReplicationSlotControlLock, LW_EXCLUSIVE);
303 
304  slot->in_use = true;
305 
306  /* We can now mark the slot active, and that makes it our slot. */
307  SpinLockAcquire(&slot->mutex);
308  Assert(slot->active_pid == 0);
309  slot->active_pid = MyProcPid;
310  SpinLockRelease(&slot->mutex);
311  MyReplicationSlot = slot;
312 
313  LWLockRelease(ReplicationSlotControlLock);
314 
315  /*
316  * Now that the slot has been marked as in_use and in_active, it's safe to
317  * let somebody else try to allocate a slot.
318  */
319  LWLockRelease(ReplicationSlotAllocationLock);
320 }
321 
322 /*
323  * Find a previously created slot and mark it as used by this backend.
324  */
325 void
327 {
328  ReplicationSlot *slot = NULL;
329  int i;
330  int active_pid = 0; /* Keep compiler quiet */
331 
332  Assert(MyReplicationSlot == NULL);
333 
335 
336  /* Search for the named slot and mark it active if we find it. */
337  LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
338  for (i = 0; i < max_replication_slots; i++)
339  {
340  ReplicationSlot *s = &ReplicationSlotCtl->replication_slots[i];
341 
342  if (s->in_use && strcmp(name, NameStr(s->data.name)) == 0)
343  {
344  SpinLockAcquire(&s->mutex);
345  active_pid = s->active_pid;
346  if (active_pid == 0)
347  active_pid = s->active_pid = MyProcPid;
348  SpinLockRelease(&s->mutex);
349  slot = s;
350  break;
351  }
352  }
353  LWLockRelease(ReplicationSlotControlLock);
354 
355  /* If we did not find the slot or it was already active, error out. */
356  if (slot == NULL)
357  ereport(ERROR,
358  (errcode(ERRCODE_UNDEFINED_OBJECT),
359  errmsg("replication slot \"%s\" does not exist", name)));
360  if (active_pid != MyProcPid)
361  ereport(ERROR,
362  (errcode(ERRCODE_OBJECT_IN_USE),
363  errmsg("replication slot \"%s\" is active for PID %d",
364  name, active_pid)));
365 
366  /* We made this slot active, so it's ours now. */
367  MyReplicationSlot = slot;
368 }
369 
370 /*
371  * Release a replication slot, this or another backend can ReAcquire it
372  * later. Resources this slot requires will be preserved.
373  */
374 void
376 {
378 
379  Assert(slot != NULL && slot->active_pid != 0);
380 
381  if (slot->data.persistency == RS_EPHEMERAL)
382  {
383  /*
384  * Delete the slot. There is no !PANIC case where this is allowed to
385  * fail, all that may happen is an incomplete cleanup of the on-disk
386  * data.
387  */
389  }
390  else if (slot->data.persistency == RS_PERSISTENT)
391  {
392  /*
393  * Mark persistent slot inactive. We're not freeing it, just
394  * disconnecting.
395  */
396  SpinLockAcquire(&slot->mutex);
397  slot->active_pid = 0;
398  SpinLockRelease(&slot->mutex);
399  }
400 
401 
402  /*
403  * If slot needed to temporarily restrain both data and catalog xmin to
404  * create the catalog snapshot, remove that temporary constraint.
405  * Snapshots can only be exported while the initial snapshot is still
406  * acquired.
407  */
408  if (!TransactionIdIsValid(slot->data.xmin) &&
410  {
411  SpinLockAcquire(&slot->mutex);
413  SpinLockRelease(&slot->mutex);
415  }
416 
417  MyReplicationSlot = NULL;
418 
419  /* might not have been set when we've been a plain slot */
420  LWLockAcquire(ProcArrayLock, LW_EXCLUSIVE);
422  LWLockRelease(ProcArrayLock);
423 }
424 
425 /*
426  * Cleanup all temporary slots created in current session.
427  */
428 void
430 {
431  int i;
432 
433  Assert(MyReplicationSlot == NULL);
434 
435  /*
436  * No need for locking as we are only interested in slots active in
437  * current process and those are not touched by other processes.
438  */
439  for (i = 0; i < max_replication_slots; i++)
440  {
441  ReplicationSlot *s = &ReplicationSlotCtl->replication_slots[i];
442 
443  if (s->active_pid == MyProcPid)
444  {
446 
448  }
449  }
450 }
451 
452 /*
453  * Permanently drop replication slot identified by the passed in name.
454  */
455 void
457 {
458  Assert(MyReplicationSlot == NULL);
459 
461 
463 }
464 
465 /*
466  * Permanently drop the currently acquired replication slot.
467  */
468 static void
470 {
472 
473  Assert(MyReplicationSlot != NULL);
474 
475  /* slot isn't acquired anymore */
476  MyReplicationSlot = NULL;
477 
479 }
480 
481 /*
482  * Permanently drop the replication slot which will be released by the point
483  * this function returns.
484  */
485 static void
487 {
488  char path[MAXPGPATH];
489  char tmppath[MAXPGPATH];
490 
491  /*
492  * If some other backend ran this code concurrently with us, we might try
493  * to delete a slot with a certain name while someone else was trying to
494  * create a slot with the same name.
495  */
496  LWLockAcquire(ReplicationSlotAllocationLock, LW_EXCLUSIVE);
497 
498  /* Generate pathnames. */
499  sprintf(path, "pg_replslot/%s", NameStr(slot->data.name));
500  sprintf(tmppath, "pg_replslot/%s.tmp", NameStr(slot->data.name));
501 
502  /*
503  * Rename the slot directory on disk, so that we'll no longer recognize
504  * this as a valid slot. Note that if this fails, we've got to mark the
505  * slot inactive before bailing out. If we're dropping an ephemeral or a
506  * temporary slot, we better never fail hard as the caller won't expect
507  * the slot to survive and this might get called during error handling.
508  */
509  if (rename(path, tmppath) == 0)
510  {
511  /*
512  * We need to fsync() the directory we just renamed and its parent to
513  * make sure that our changes are on disk in a crash-safe fashion. If
514  * fsync() fails, we can't be sure whether the changes are on disk or
515  * not. For now, we handle that by panicking;
516  * StartupReplicationSlots() will try to straighten it out after
517  * restart.
518  */
520  fsync_fname(tmppath, true);
521  fsync_fname("pg_replslot", true);
523  }
524  else
525  {
526  bool fail_softly = slot->data.persistency != RS_PERSISTENT;
527 
528  SpinLockAcquire(&slot->mutex);
529  slot->active_pid = 0;
530  SpinLockRelease(&slot->mutex);
531 
532  ereport(fail_softly ? WARNING : ERROR,
534  errmsg("could not rename file \"%s\" to \"%s\": %m",
535  path, tmppath)));
536  }
537 
538  /*
539  * The slot is definitely gone. Lock out concurrent scans of the array
540  * long enough to kill it. It's OK to clear the active flag here without
541  * grabbing the mutex because nobody else can be scanning the array here,
542  * and nobody can be attached to this slot and thus access it without
543  * scanning the array.
544  */
545  LWLockAcquire(ReplicationSlotControlLock, LW_EXCLUSIVE);
546  slot->active_pid = 0;
547  slot->in_use = false;
548  LWLockRelease(ReplicationSlotControlLock);
549 
550  /*
551  * Slot is dead and doesn't prevent resource removal anymore, recompute
552  * limits.
553  */
556 
557  /*
558  * If removing the directory fails, the worst thing that will happen is
559  * that the user won't be able to create a new slot with the same name
560  * until the next server restart. We warn about it, but that's all.
561  */
562  if (!rmtree(tmppath, true))
565  errmsg("could not remove directory \"%s\"", tmppath)));
566 
567  /*
568  * We release this at the very end, so that nobody starts trying to create
569  * a slot while we're still cleaning up the detritus of the old one.
570  */
571  LWLockRelease(ReplicationSlotAllocationLock);
572 }
573 
574 /*
575  * Serialize the currently acquired slot's state from memory to disk, thereby
576  * guaranteeing the current state will survive a crash.
577  */
578 void
580 {
581  char path[MAXPGPATH];
582 
583  Assert(MyReplicationSlot != NULL);
584 
585  sprintf(path, "pg_replslot/%s", NameStr(MyReplicationSlot->data.name));
586  SaveSlotToPath(MyReplicationSlot, path, ERROR);
587 }
588 
589 /*
590  * Signal that it would be useful if the currently acquired slot would be
591  * flushed out to disk.
592  *
593  * Note that the actual flush to disk can be delayed for a long time, if
594  * required for correctness explicitly do a ReplicationSlotSave().
595  */
596 void
598 {
600 
601  Assert(MyReplicationSlot != NULL);
602 
603  SpinLockAcquire(&slot->mutex);
604  MyReplicationSlot->just_dirtied = true;
605  MyReplicationSlot->dirty = true;
606  SpinLockRelease(&slot->mutex);
607 }
608 
609 /*
610  * Convert a slot that's marked as RS_EPHEMERAL to a RS_PERSISTENT slot,
611  * guaranteeing it will be there after an eventual crash.
612  */
613 void
615 {
617 
618  Assert(slot != NULL);
620 
621  SpinLockAcquire(&slot->mutex);
623  SpinLockRelease(&slot->mutex);
624 
627 }
628 
629 /*
630  * Compute the oldest xmin across all slots and store it in the ProcArray.
631  *
632  * If already_locked is true, ProcArrayLock has already been acquired
633  * exclusively.
634  */
635 void
637 {
638  int i;
640  TransactionId agg_catalog_xmin = InvalidTransactionId;
641 
642  Assert(ReplicationSlotCtl != NULL);
643 
644  LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
645 
646  for (i = 0; i < max_replication_slots; i++)
647  {
648  ReplicationSlot *s = &ReplicationSlotCtl->replication_slots[i];
649  TransactionId effective_xmin;
650  TransactionId effective_catalog_xmin;
651 
652  if (!s->in_use)
653  continue;
654 
655  SpinLockAcquire(&s->mutex);
656  effective_xmin = s->effective_xmin;
657  effective_catalog_xmin = s->effective_catalog_xmin;
658  SpinLockRelease(&s->mutex);
659 
660  /* check the data xmin */
661  if (TransactionIdIsValid(effective_xmin) &&
662  (!TransactionIdIsValid(agg_xmin) ||
663  TransactionIdPrecedes(effective_xmin, agg_xmin)))
664  agg_xmin = effective_xmin;
665 
666  /* check the catalog xmin */
667  if (TransactionIdIsValid(effective_catalog_xmin) &&
668  (!TransactionIdIsValid(agg_catalog_xmin) ||
669  TransactionIdPrecedes(effective_catalog_xmin, agg_catalog_xmin)))
670  agg_catalog_xmin = effective_catalog_xmin;
671  }
672 
673  LWLockRelease(ReplicationSlotControlLock);
674 
675  ProcArraySetReplicationSlotXmin(agg_xmin, agg_catalog_xmin, already_locked);
676 }
677 
678 /*
679  * Compute the oldest restart LSN across all slots and inform xlog module.
680  */
681 void
683 {
684  int i;
685  XLogRecPtr min_required = InvalidXLogRecPtr;
686 
687  Assert(ReplicationSlotCtl != NULL);
688 
689  LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
690  for (i = 0; i < max_replication_slots; i++)
691  {
692  ReplicationSlot *s = &ReplicationSlotCtl->replication_slots[i];
693  XLogRecPtr restart_lsn;
694 
695  if (!s->in_use)
696  continue;
697 
698  SpinLockAcquire(&s->mutex);
699  restart_lsn = s->data.restart_lsn;
700  SpinLockRelease(&s->mutex);
701 
702  if (restart_lsn != InvalidXLogRecPtr &&
703  (min_required == InvalidXLogRecPtr ||
704  restart_lsn < min_required))
705  min_required = restart_lsn;
706  }
707  LWLockRelease(ReplicationSlotControlLock);
708 
709  XLogSetReplicationSlotMinimumLSN(min_required);
710 }
711 
712 /*
713  * Compute the oldest WAL LSN required by *logical* decoding slots..
714  *
715  * Returns InvalidXLogRecPtr if logical decoding is disabled or no logical
716  * slots exist.
717  *
718  * NB: this returns a value >= ReplicationSlotsComputeRequiredLSN(), since it
719  * ignores physical replication slots.
720  *
721  * The results aren't required frequently, so we don't maintain a precomputed
722  * value like we do for ComputeRequiredLSN() and ComputeRequiredXmin().
723  */
726 {
728  int i;
729 
730  if (max_replication_slots <= 0)
731  return InvalidXLogRecPtr;
732 
733  LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
734 
735  for (i = 0; i < max_replication_slots; i++)
736  {
737  ReplicationSlot *s;
738  XLogRecPtr restart_lsn;
739 
740  s = &ReplicationSlotCtl->replication_slots[i];
741 
742  /* cannot change while ReplicationSlotCtlLock is held */
743  if (!s->in_use)
744  continue;
745 
746  /* we're only interested in logical slots */
747  if (!SlotIsLogical(s))
748  continue;
749 
750  /* read once, it's ok if it increases while we're checking */
751  SpinLockAcquire(&s->mutex);
752  restart_lsn = s->data.restart_lsn;
753  SpinLockRelease(&s->mutex);
754 
755  if (result == InvalidXLogRecPtr ||
756  restart_lsn < result)
757  result = restart_lsn;
758  }
759 
760  LWLockRelease(ReplicationSlotControlLock);
761 
762  return result;
763 }
764 
765 /*
766  * ReplicationSlotsCountDBSlots -- count the number of slots that refer to the
767  * passed database oid.
768  *
769  * Returns true if there are any slots referencing the database. *nslots will
770  * be set to the absolute number of slots in the database, *nactive to ones
771  * currently active.
772  */
773 bool
774 ReplicationSlotsCountDBSlots(Oid dboid, int *nslots, int *nactive)
775 {
776  int i;
777 
778  *nslots = *nactive = 0;
779 
780  if (max_replication_slots <= 0)
781  return false;
782 
783  LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
784  for (i = 0; i < max_replication_slots; i++)
785  {
786  ReplicationSlot *s;
787 
788  s = &ReplicationSlotCtl->replication_slots[i];
789 
790  /* cannot change while ReplicationSlotCtlLock is held */
791  if (!s->in_use)
792  continue;
793 
794  /* only logical slots are database specific, skip */
795  if (!SlotIsLogical(s))
796  continue;
797 
798  /* not our database, skip */
799  if (s->data.database != dboid)
800  continue;
801 
802  /* count slots with spinlock held */
803  SpinLockAcquire(&s->mutex);
804  (*nslots)++;
805  if (s->active_pid != 0)
806  (*nactive)++;
807  SpinLockRelease(&s->mutex);
808  }
809  LWLockRelease(ReplicationSlotControlLock);
810 
811  if (*nslots > 0)
812  return true;
813  return false;
814 }
815 
816 /*
817  * ReplicationSlotsDropDBSlots -- Drop all db-specific slots relating to the
818  * passed database oid. The caller should hold an exclusive lock on the
819  * pg_database oid for the database to prevent creation of new slots on the db
820  * or replay from existing slots.
821  *
822  * Another session that concurrently acquires an existing slot on the target DB
823  * (most likely to drop it) may cause this function to ERROR. If that happens
824  * it may have dropped some but not all slots.
825  *
826  * This routine isn't as efficient as it could be - but we don't drop
827  * databases often, especially databases with lots of slots.
828  */
829 void
831 {
832  int i;
833 
834  if (max_replication_slots <= 0)
835  return;
836 
837 restart:
838  LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
839  for (i = 0; i < max_replication_slots; i++)
840  {
841  ReplicationSlot *s;
842  char *slotname;
843  int active_pid;
844 
845  s = &ReplicationSlotCtl->replication_slots[i];
846 
847  /* cannot change while ReplicationSlotCtlLock is held */
848  if (!s->in_use)
849  continue;
850 
851  /* only logical slots are database specific, skip */
852  if (!SlotIsLogical(s))
853  continue;
854 
855  /* not our database, skip */
856  if (s->data.database != dboid)
857  continue;
858 
859  /* acquire slot, so ReplicationSlotDropAcquired can be reused */
860  SpinLockAcquire(&s->mutex);
861  /* can't change while ReplicationSlotControlLock is held */
862  slotname = NameStr(s->data.name);
863  active_pid = s->active_pid;
864  if (active_pid == 0)
865  {
866  MyReplicationSlot = s;
867  s->active_pid = MyProcPid;
868  }
869  SpinLockRelease(&s->mutex);
870 
871  /*
872  * Even though we hold an exclusive lock on the database object a
873  * logical slot for that DB can still be active, e.g. if it's
874  * concurrently being dropped by a backend connected to another DB.
875  *
876  * That's fairly unlikely in practice, so we'll just bail out.
877  */
878  if (active_pid)
879  ereport(ERROR,
880  (errcode(ERRCODE_OBJECT_IN_USE),
881  errmsg("replication slot \"%s\" is active for PID %d",
882  slotname, active_pid)));
883 
884  /*
885  * To avoid duplicating ReplicationSlotDropAcquired() and to avoid
886  * holding ReplicationSlotControlLock over filesystem operations,
887  * release ReplicationSlotControlLock and use
888  * ReplicationSlotDropAcquired.
889  *
890  * As that means the set of slots could change, restart scan from the
891  * beginning each time we release the lock.
892  */
893  LWLockRelease(ReplicationSlotControlLock);
895  goto restart;
896  }
897  LWLockRelease(ReplicationSlotControlLock);
898 }
899 
900 
901 /*
902  * Check whether the server's configuration supports using replication
903  * slots.
904  */
905 void
907 {
908  if (max_replication_slots == 0)
909  ereport(ERROR,
910  (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
911  (errmsg("replication slots can only be used if max_replication_slots > 0"))));
912 
914  ereport(ERROR,
915  (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
916  errmsg("replication slots can only be used if wal_level >= replica")));
917 }
918 
919 /*
920  * Reserve WAL for the currently active slot.
921  *
922  * Compute and set restart_lsn in a manner that's appropriate for the type of
923  * the slot and concurrency safe.
924  */
925 void
927 {
929 
930  Assert(slot != NULL);
932 
933  /*
934  * The replication slot mechanism is used to prevent removal of required
935  * WAL. As there is no interlock between this routine and checkpoints, WAL
936  * segments could concurrently be removed when a now stale return value of
937  * ReplicationSlotsComputeRequiredLSN() is used. In the unlikely case that
938  * this happens we'll just retry.
939  */
940  while (true)
941  {
942  XLogSegNo segno;
943 
944  /*
945  * For logical slots log a standby snapshot and start logical decoding
946  * at exactly that position. That allows the slot to start up more
947  * quickly.
948  *
949  * That's not needed (or indeed helpful) for physical slots as they'll
950  * start replay at the last logged checkpoint anyway. Instead return
951  * the location of the last redo LSN. While that slightly increases
952  * the chance that we have to retry, it's where a base backup has to
953  * start replay at.
954  */
955  if (!RecoveryInProgress() && SlotIsLogical(slot))
956  {
957  XLogRecPtr flushptr;
958 
959  /* start at current insert position */
961 
962  /* make sure we have enough information to start */
963  flushptr = LogStandbySnapshot();
964 
965  /* and make sure it's fsynced to disk */
966  XLogFlush(flushptr);
967  }
968  else
969  {
970  slot->data.restart_lsn = GetRedoRecPtr();
971  }
972 
973  /* prevent WAL removal as fast as possible */
975 
976  /*
977  * If all required WAL is still there, great, otherwise retry. The
978  * slot should prevent further removal of WAL, unless there's a
979  * concurrent ReplicationSlotsComputeRequiredLSN() after we've written
980  * the new restart_lsn above, so normally we should never need to loop
981  * more than twice.
982  */
983  XLByteToSeg(slot->data.restart_lsn, segno);
984  if (XLogGetLastRemovedSegno() < segno)
985  break;
986  }
987 }
988 
989 /*
990  * Flush all replication slots to disk.
991  *
992  * This needn't actually be part of a checkpoint, but it's a convenient
993  * location.
994  */
995 void
997 {
998  int i;
999 
1000  elog(DEBUG1, "performing replication slot checkpoint");
1001 
1002  /*
1003  * Prevent any slot from being created/dropped while we're active. As we
1004  * explicitly do *not* want to block iterating over replication_slots or
1005  * acquiring a slot we cannot take the control lock - but that's OK,
1006  * because holding ReplicationSlotAllocationLock is strictly stronger, and
1007  * enough to guarantee that nobody can change the in_use bits on us.
1008  */
1009  LWLockAcquire(ReplicationSlotAllocationLock, LW_SHARED);
1010 
1011  for (i = 0; i < max_replication_slots; i++)
1012  {
1013  ReplicationSlot *s = &ReplicationSlotCtl->replication_slots[i];
1014  char path[MAXPGPATH];
1015 
1016  if (!s->in_use)
1017  continue;
1018 
1019  /* save the slot to disk, locking is handled in SaveSlotToPath() */
1020  sprintf(path, "pg_replslot/%s", NameStr(s->data.name));
1021  SaveSlotToPath(s, path, LOG);
1022  }
1023  LWLockRelease(ReplicationSlotAllocationLock);
1024 }
1025 
1026 /*
1027  * Load all replication slots from disk into memory at server startup. This
1028  * needs to be run before we start crash recovery.
1029  */
1030 void
1032 {
1033  DIR *replication_dir;
1034  struct dirent *replication_de;
1035 
1036  elog(DEBUG1, "starting up replication slots");
1037 
1038  /* restore all slots by iterating over all on-disk entries */
1039  replication_dir = AllocateDir("pg_replslot");
1040  while ((replication_de = ReadDir(replication_dir, "pg_replslot")) != NULL)
1041  {
1042  struct stat statbuf;
1043  char path[MAXPGPATH + 12];
1044 
1045  if (strcmp(replication_de->d_name, ".") == 0 ||
1046  strcmp(replication_de->d_name, "..") == 0)
1047  continue;
1048 
1049  snprintf(path, sizeof(path), "pg_replslot/%s", replication_de->d_name);
1050 
1051  /* we're only creating directories here, skip if it's not our's */
1052  if (lstat(path, &statbuf) == 0 && !S_ISDIR(statbuf.st_mode))
1053  continue;
1054 
1055  /* we crashed while a slot was being setup or deleted, clean up */
1056  if (pg_str_endswith(replication_de->d_name, ".tmp"))
1057  {
1058  if (!rmtree(path, true))
1059  {
1060  ereport(WARNING,
1062  errmsg("could not remove directory \"%s\"", path)));
1063  continue;
1064  }
1065  fsync_fname("pg_replslot", true);
1066  continue;
1067  }
1068 
1069  /* looks like a slot in a normal state, restore */
1070  RestoreSlotFromDisk(replication_de->d_name);
1071  }
1072  FreeDir(replication_dir);
1073 
1074  /* currently no slots exist, we're done. */
1075  if (max_replication_slots <= 0)
1076  return;
1077 
1078  /* Now that we have recovered all the data, compute replication xmin */
1081 }
1082 
1083 /* ----
1084  * Manipulation of on-disk state of replication slots
1085  *
1086  * NB: none of the routines below should take any notice whether a slot is the
1087  * current one or not, that's all handled a layer above.
1088  * ----
1089  */
1090 static void
1092 {
1093  char tmppath[MAXPGPATH];
1094  char path[MAXPGPATH];
1095  struct stat st;
1096 
1097  /*
1098  * No need to take out the io_in_progress_lock, nobody else can see this
1099  * slot yet, so nobody else will write. We're reusing SaveSlotToPath which
1100  * takes out the lock, if we'd take the lock here, we'd deadlock.
1101  */
1102 
1103  sprintf(path, "pg_replslot/%s", NameStr(slot->data.name));
1104  sprintf(tmppath, "pg_replslot/%s.tmp", NameStr(slot->data.name));
1105 
1106  /*
1107  * It's just barely possible that some previous effort to create or drop a
1108  * slot with this name left a temp directory lying around. If that seems
1109  * to be the case, try to remove it. If the rmtree() fails, we'll error
1110  * out at the mkdir() below, so we don't bother checking success.
1111  */
1112  if (stat(tmppath, &st) == 0 && S_ISDIR(st.st_mode))
1113  rmtree(tmppath, true);
1114 
1115  /* Create and fsync the temporary slot directory. */
1116  if (mkdir(tmppath, S_IRWXU) < 0)
1117  ereport(ERROR,
1119  errmsg("could not create directory \"%s\": %m",
1120  tmppath)));
1121  fsync_fname(tmppath, true);
1122 
1123  /* Write the actual state file. */
1124  slot->dirty = true; /* signal that we really need to write */
1125  SaveSlotToPath(slot, tmppath, ERROR);
1126 
1127  /* Rename the directory into place. */
1128  if (rename(tmppath, path) != 0)
1129  ereport(ERROR,
1131  errmsg("could not rename file \"%s\" to \"%s\": %m",
1132  tmppath, path)));
1133 
1134  /*
1135  * If we'd now fail - really unlikely - we wouldn't know whether this slot
1136  * would persist after an OS crash or not - so, force a restart. The
1137  * restart would try to fsync this again till it works.
1138  */
1140 
1141  fsync_fname(path, true);
1142  fsync_fname("pg_replslot", true);
1143 
1144  END_CRIT_SECTION();
1145 }
1146 
1147 /*
1148  * Shared functionality between saving and creating a replication slot.
1149  */
1150 static void
1151 SaveSlotToPath(ReplicationSlot *slot, const char *dir, int elevel)
1152 {
1153  char tmppath[MAXPGPATH];
1154  char path[MAXPGPATH];
1155  int fd;
1157  bool was_dirty;
1158 
1159  /* first check whether there's something to write out */
1160  SpinLockAcquire(&slot->mutex);
1161  was_dirty = slot->dirty;
1162  slot->just_dirtied = false;
1163  SpinLockRelease(&slot->mutex);
1164 
1165  /* and don't do anything if there's nothing to write */
1166  if (!was_dirty)
1167  return;
1168 
1170 
1171  /* silence valgrind :( */
1172  memset(&cp, 0, sizeof(ReplicationSlotOnDisk));
1173 
1174  sprintf(tmppath, "%s/state.tmp", dir);
1175  sprintf(path, "%s/state", dir);
1176 
1177  fd = OpenTransientFile(tmppath,
1178  O_CREAT | O_EXCL | O_WRONLY | PG_BINARY,
1179  S_IRUSR | S_IWUSR);
1180  if (fd < 0)
1181  {
1182  ereport(elevel,
1184  errmsg("could not create file \"%s\": %m",
1185  tmppath)));
1186  return;
1187  }
1188 
1189  cp.magic = SLOT_MAGIC;
1190  INIT_CRC32C(cp.checksum);
1191  cp.version = SLOT_VERSION;
1193 
1194  SpinLockAcquire(&slot->mutex);
1195 
1196  memcpy(&cp.slotdata, &slot->data, sizeof(ReplicationSlotPersistentData));
1197 
1198  SpinLockRelease(&slot->mutex);
1199 
1200  COMP_CRC32C(cp.checksum,
1201  (char *) (&cp) + SnapBuildOnDiskNotChecksummedSize,
1203  FIN_CRC32C(cp.checksum);
1204 
1206  if ((write(fd, &cp, sizeof(cp))) != sizeof(cp))
1207  {
1208  int save_errno = errno;
1209 
1211  CloseTransientFile(fd);
1212  errno = save_errno;
1213  ereport(elevel,
1215  errmsg("could not write to file \"%s\": %m",
1216  tmppath)));
1217  return;
1218  }
1220 
1221  /* fsync the temporary file */
1223  if (pg_fsync(fd) != 0)
1224  {
1225  int save_errno = errno;
1226 
1228  CloseTransientFile(fd);
1229  errno = save_errno;
1230  ereport(elevel,
1232  errmsg("could not fsync file \"%s\": %m",
1233  tmppath)));
1234  return;
1235  }
1237 
1238  CloseTransientFile(fd);
1239 
1240  /* rename to permanent file, fsync file and directory */
1241  if (rename(tmppath, path) != 0)
1242  {
1243  ereport(elevel,
1245  errmsg("could not rename file \"%s\" to \"%s\": %m",
1246  tmppath, path)));
1247  return;
1248  }
1249 
1250  /* Check CreateSlot() for the reasoning of using a crit. section. */
1252 
1253  fsync_fname(path, false);
1254  fsync_fname(dir, true);
1255  fsync_fname("pg_replslot", true);
1256 
1257  END_CRIT_SECTION();
1258 
1259  /*
1260  * Successfully wrote, unset dirty bit, unless somebody dirtied again
1261  * already.
1262  */
1263  SpinLockAcquire(&slot->mutex);
1264  if (!slot->just_dirtied)
1265  slot->dirty = false;
1266  SpinLockRelease(&slot->mutex);
1267 
1269 }
1270 
1271 /*
1272  * Load a single slot from disk into memory.
1273  */
1274 static void
1276 {
1278  int i;
1279  char path[MAXPGPATH + 22];
1280  int fd;
1281  bool restored = false;
1282  int readBytes;
1283  pg_crc32c checksum;
1284 
1285  /* no need to lock here, no concurrent access allowed yet */
1286 
1287  /* delete temp file if it exists */
1288  sprintf(path, "pg_replslot/%s/state.tmp", name);
1289  if (unlink(path) < 0 && errno != ENOENT)
1290  ereport(PANIC,
1292  errmsg("could not remove file \"%s\": %m", path)));
1293 
1294  sprintf(path, "pg_replslot/%s/state", name);
1295 
1296  elog(DEBUG1, "restoring replication slot from \"%s\"", path);
1297 
1298  fd = OpenTransientFile(path, O_RDWR | PG_BINARY, 0);
1299 
1300  /*
1301  * We do not need to handle this as we are rename()ing the directory into
1302  * place only after we fsync()ed the state file.
1303  */
1304  if (fd < 0)
1305  ereport(PANIC,
1307  errmsg("could not open file \"%s\": %m", path)));
1308 
1309  /*
1310  * Sync state file before we're reading from it. We might have crashed
1311  * while it wasn't synced yet and we shouldn't continue on that basis.
1312  */
1314  if (pg_fsync(fd) != 0)
1315  {
1316  CloseTransientFile(fd);
1317  ereport(PANIC,
1319  errmsg("could not fsync file \"%s\": %m",
1320  path)));
1321  }
1323 
1324  /* Also sync the parent directory */
1326  fsync_fname(path, true);
1327  END_CRIT_SECTION();
1328 
1329  /* read part of statefile that's guaranteed to be version independent */
1331  readBytes = read(fd, &cp, ReplicationSlotOnDiskConstantSize);
1333  if (readBytes != ReplicationSlotOnDiskConstantSize)
1334  {
1335  int saved_errno = errno;
1336 
1337  CloseTransientFile(fd);
1338  errno = saved_errno;
1339  ereport(PANIC,
1341  errmsg("could not read file \"%s\", read %d of %u: %m",
1342  path, readBytes,
1344  }
1345 
1346  /* verify magic */
1347  if (cp.magic != SLOT_MAGIC)
1348  ereport(PANIC,
1350  errmsg("replication slot file \"%s\" has wrong magic number: %u instead of %u",
1351  path, cp.magic, SLOT_MAGIC)));
1352 
1353  /* verify version */
1354  if (cp.version != SLOT_VERSION)
1355  ereport(PANIC,
1357  errmsg("replication slot file \"%s\" has unsupported version %u",
1358  path, cp.version)));
1359 
1360  /* boundary check on length */
1362  ereport(PANIC,
1364  errmsg("replication slot file \"%s\" has corrupted length %u",
1365  path, cp.length)));
1366 
1367  /* Now that we know the size, read the entire file */
1369  readBytes = read(fd,
1370  (char *) &cp + ReplicationSlotOnDiskConstantSize,
1371  cp.length);
1373  if (readBytes != cp.length)
1374  {
1375  int saved_errno = errno;
1376 
1377  CloseTransientFile(fd);
1378  errno = saved_errno;
1379  ereport(PANIC,
1381  errmsg("could not read file \"%s\", read %d of %u: %m",
1382  path, readBytes, cp.length)));
1383  }
1384 
1385  CloseTransientFile(fd);
1386 
1387  /* now verify the CRC */
1388  INIT_CRC32C(checksum);
1389  COMP_CRC32C(checksum,
1390  (char *) &cp + SnapBuildOnDiskNotChecksummedSize,
1392  FIN_CRC32C(checksum);
1393 
1394  if (!EQ_CRC32C(checksum, cp.checksum))
1395  ereport(PANIC,
1396  (errmsg("checksum mismatch for replication slot file \"%s\": is %u, should be %u",
1397  path, checksum, cp.checksum)));
1398 
1399  /*
1400  * If we crashed with an ephemeral slot active, don't restore but delete
1401  * it.
1402  */
1404  {
1405  sprintf(path, "pg_replslot/%s", name);
1406 
1407  if (!rmtree(path, true))
1408  {
1409  ereport(WARNING,
1411  errmsg("could not remove directory \"%s\"", path)));
1412  }
1413  fsync_fname("pg_replslot", true);
1414  return;
1415  }
1416 
1417  /* nothing can be active yet, don't lock anything */
1418  for (i = 0; i < max_replication_slots; i++)
1419  {
1420  ReplicationSlot *slot;
1421 
1422  slot = &ReplicationSlotCtl->replication_slots[i];
1423 
1424  if (slot->in_use)
1425  continue;
1426 
1427  /* restore the entire set of persistent data */
1428  memcpy(&slot->data, &cp.slotdata,
1430 
1431  /* initialize in memory state */
1432  slot->effective_xmin = cp.slotdata.xmin;
1434 
1439 
1440  slot->in_use = true;
1441  slot->active_pid = 0;
1442 
1443  restored = true;
1444  break;
1445  }
1446 
1447  if (!restored)
1448  ereport(PANIC,
1449  (errmsg("too many replication slots active before shutdown"),
1450  errhint("Increase max_replication_slots and try again.")));
1451 }
#define INIT_CRC32C(crc)
Definition: pg_crc32c.h:41
static void RestoreSlotFromDisk(const char *name)
Definition: slot.c:1275
ReplicationSlotCtlData * ReplicationSlotCtl
Definition: slot.c:93
void CheckSlotRequirements(void)
Definition: slot.c:906
TransactionId candidate_catalog_xmin
Definition: slot.h:127
#define InvalidXLogRecPtr
Definition: xlogdefs.h:28
#define DEBUG1
Definition: elog.h:25
int MyProcPid
Definition: globals.c:38
int errhint(const char *fmt,...)
Definition: elog.c:987
Size ReplicationSlotsShmemSize(void)
Definition: slot.c:114
#define PROC_IN_LOGICAL_DECODING
Definition: proc.h:56
uint32 TransactionId
Definition: c.h:397
bool pg_str_endswith(const char *str, const char *end)
Definition: string.c:31
#define write(a, b, c)
Definition: win32.h:14
#define SLOT_MAGIC
Definition: slot.c:89
uint32 pg_crc32c
Definition: pg_crc32c.h:38
#define mkdir(a, b)
Definition: win32.h:57
int wal_level
Definition: xlog.c:104
#define SpinLockInit(lock)
Definition: spin.h:60
#define END_CRIT_SECTION()
Definition: miscadmin.h:132
void fsync_fname(const char *fname, bool isdir)
Definition: fd.c:567
void ReplicationSlotCreate(const char *name, bool db_specific, ReplicationSlotPersistency persistency)
Definition: slot.c:220
ReplicationSlotPersistency persistency
Definition: slot.h:49
#define START_CRIT_SECTION()
Definition: miscadmin.h:130
int errcode(int sqlerrcode)
Definition: elog.c:575
#define MemSet(start, val, len)
Definition: c.h:857
return result
Definition: formatting.c:1632
int snprintf(char *str, size_t count, const char *fmt,...) pg_attribute_printf(3
void ReplicationSlotSave(void)
Definition: slot.c:579
#define SnapBuildOnDiskNotChecksummedSize
Definition: slot.c:80
static void ReplicationSlotDropPtr(ReplicationSlot *slot)
Definition: slot.c:486
ReplicationSlotPersistentData data
Definition: slot.h:115
#define LOG
Definition: elog.h:26
unsigned int Oid
Definition: postgres_ext.h:31
bool RecoveryInProgress(void)
Definition: xlog.c:7873
Definition: dirent.h:9
#define PANIC
Definition: elog.h:53
void XLogFlush(XLogRecPtr record)
Definition: xlog.c:2758
static int fd(const char *x, int i)
Definition: preproc-init.c:105
#define PG_BINARY
Definition: c.h:1038
static void CreateSlotOnDisk(ReplicationSlot *slot)
Definition: slot.c:1091
void ReplicationSlotsShmemInit(void)
Definition: slot.c:132
static void SaveSlotToPath(ReplicationSlot *slot, const char *path, int elevel)
Definition: slot.c:1151
void XLogSetReplicationSlotMinimumLSN(XLogRecPtr lsn)
Definition: xlog.c:2649
PGXACT * MyPgXact
Definition: proc.c:68
uint8 vacuumFlags
Definition: proc.h:218
void LWLockRelease(LWLock *lock)
Definition: lwlock.c:1715
#define NAMEDATALEN
bool ReplicationSlotValidateName(const char *name, int elevel)
Definition: slot.c:173
#define SpinLockAcquire(lock)
Definition: spin.h:62
XLogSegNo XLogGetLastRemovedSegno(void)
Definition: xlog.c:3794
void ReplicationSlotReserveWal(void)
Definition: slot.c:926
static void ReplicationSlotDropAcquired(void)
Definition: slot.c:469
void ReplicationSlotsComputeRequiredLSN(void)
Definition: slot.c:682
ReplicationSlotPersistentData slotdata
Definition: slot.c:73
XLogRecPtr LogStandbySnapshot(void)
Definition: standby.c:909
Definition: dirent.c:25
#define ERROR
Definition: elog.h:43
void * ShmemInitStruct(const char *name, Size size, bool *foundPtr)
Definition: shmem.c:372
XLogRecPtr GetXLogInsertRecPtr(void)
Definition: xlog.c:11103
#define MAXPGPATH
void ReplicationSlotPersist(void)
Definition: slot.c:614
TransactionId effective_xmin
Definition: slot.h:111
XLogRecPtr candidate_restart_valid
Definition: slot.h:129
void StartupReplicationSlots(void)
Definition: slot.c:1031
uint64 XLogSegNo
Definition: xlogdefs.h:34
int OpenTransientFile(FileName fileName, int fileFlags, int fileMode)
Definition: fd.c:2144
int errcode_for_file_access(void)
Definition: elog.c:598
XLogRecPtr ReplicationSlotsComputeLogicalRestartLSN(void)
Definition: slot.c:725
TransactionId catalog_xmin
Definition: slot.h:65
#define InvalidTransactionId
Definition: transam.h:31
unsigned int uint32
Definition: c.h:268
DIR * AllocateDir(const char *dirname)
Definition: fd.c:2335
void ReplicationSlotRelease(void)
Definition: slot.c:375
static void pgstat_report_wait_end(void)
Definition: pgstat.h:1232
TransactionId xmin
Definition: slot.h:57
#define EQ_CRC32C(c1, c2)
Definition: pg_crc32c.h:42
#define SlotIsLogical(slot)
Definition: slot.h:134
int unlink(const char *filename)
#define ereport(elevel, rest)
Definition: elog.h:122
bool TransactionIdPrecedes(TransactionId id1, TransactionId id2)
Definition: transam.c:300
pg_crc32c checksum
Definition: slot.c:62
struct ReplicationSlotOnDisk ReplicationSlotOnDisk
void LWLockInitialize(LWLock *lock, int tranche_id)
Definition: lwlock.c:667
int CloseTransientFile(int fd)
Definition: fd.c:2305
#define WARNING
Definition: elog.h:40
bool rmtree(const char *path, bool rmtopdir)
Definition: rmtree.c:36
bool in_use
Definition: slot.h:91
static int elevel
Definition: vacuumlazy.c:137
#define SpinLockRelease(lock)
Definition: spin.h:64
Size mul_size(Size s1, Size s2)
Definition: shmem.c:492
bool just_dirtied
Definition: slot.h:97
Size add_size(Size s1, Size s2)
Definition: shmem.c:475
TransactionId effective_catalog_xmin
Definition: slot.h:112
Oid MyDatabaseId
Definition: globals.c:76
#define SLOT_VERSION
Definition: slot.c:90
void ReplicationSlotDrop(const char *name)
Definition: slot.c:456
#define InvalidOid
Definition: postgres_ext.h:36
ReplicationSlot * MyReplicationSlot
Definition: slot.c:96
#define XLByteToSeg(xlrp, logSegNo)
int max_replication_slots
Definition: slot.c:99
#define NULL
Definition: c.h:229
#define ReplicationSlotOnDiskV2Size
Definition: slot.c:86
uint64 XLogRecPtr
Definition: xlogdefs.h:21
#define Assert(condition)
Definition: c.h:675
#define StrNCpy(dst, src, len)
Definition: c.h:830
struct dirent * ReadDir(DIR *dir, const char *dirname)
Definition: fd.c:2401
XLogRecPtr restart_lsn
Definition: slot.h:68
void ReplicationSlotAcquire(const char *name)
Definition: slot.c:326
#define ReplicationSlotOnDiskConstantSize
Definition: slot.c:77
size_t Size
Definition: c.h:356
uint32 version
Definition: slot.c:65
static void pgstat_report_wait_start(uint32 wait_event_info)
Definition: pgstat.h:1208
#define SnapBuildOnDiskChecksummedSize
Definition: slot.c:83
bool LWLockAcquire(LWLock *lock, LWLockMode mode)
Definition: lwlock.c:1111
void LWLockRegisterTranche(int tranche_id, char *tranche_name)
Definition: lwlock.c:592
XLogRecPtr GetRedoRecPtr(void)
Definition: xlog.c:8163
const char * name
Definition: encode.c:521
bool ReplicationSlotsCountDBSlots(Oid dboid, int *nslots, int *nactive)
Definition: slot.c:774
XLogRecPtr candidate_xmin_lsn
Definition: slot.h:128
ReplicationSlotPersistency
Definition: slot.h:28
int errmsg(const char *fmt,...)
Definition: elog.c:797
pid_t active_pid
Definition: slot.h:94
int i
#define NameStr(name)
Definition: c.h:499
void ProcArraySetReplicationSlotXmin(TransactionId xmin, TransactionId catalog_xmin, bool already_locked)
Definition: procarray.c:2960
void ReplicationSlotCleanup(void)
Definition: slot.c:429
int pg_fsync(int fd)
Definition: fd.c:333
ReplicationSlot replication_slots[1]
Definition: slot.h:145
char d_name[MAX_PATH]
Definition: dirent.h:14
#define elog
Definition: elog.h:219
slock_t mutex
Definition: slot.h:88
#define TransactionIdIsValid(xid)
Definition: transam.h:41
#define COMP_CRC32C(crc, data, len)
Definition: pg_crc32c.h:73
#define ERRCODE_DUPLICATE_OBJECT
Definition: streamutil.c:31
#define lstat(path, sb)
Definition: win32.h:262
void CheckPointReplicationSlots(void)
Definition: slot.c:996
#define FIN_CRC32C(crc)
Definition: pg_crc32c.h:78
void ReplicationSlotsDropDBSlots(Oid dboid)
Definition: slot.c:830
void ReplicationSlotsComputeRequiredXmin(bool already_locked)
Definition: slot.c:636
bool dirty
Definition: slot.h:98
#define read(a, b, c)
Definition: win32.h:13
int FreeDir(DIR *dir)
Definition: fd.c:2444
XLogRecPtr candidate_restart_lsn
Definition: slot.h:130
#define offsetof(type, field)
Definition: c.h:555
void ReplicationSlotMarkDirty(void)
Definition: slot.c:597
LWLock io_in_progress_lock
Definition: slot.h:118