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