From ed8347d725dd89341c9ff5b6f07b761b42f48f12 Mon Sep 17 00:00:00 2001
From: Nathan Bossart <bossartn@amazon.com>
Date: Wed, 25 Aug 2021 19:14:38 +0000
Subject: [PATCH 1/1] Improve performance of pgarch_readyXlog() with many
 status files.

Presently, the archive_status directory is scanned for each file to
archive.  When there are many status files (e.g., archive_command
has been failing for a long time), these directory scans can take
much longer, which significantly slows the rate of archival.  With
this change, the archiver stores several files to archive in each
directory scan, which improves the rate of archival when there are
many files in archive_status.  To ensure timeline history files are
archived as quickly as possible, XLogArchiveNotify() forces the
archiver to do a new directory scan as soon as the .ready file for
one is created.
---
 src/backend/access/transam/xlogarchive.c |   4 +
 src/backend/postmaster/pgarch.c          | 190 ++++++++++++++++++++++++++-----
 src/include/postmaster/pgarch.h          |   1 +
 3 files changed, 169 insertions(+), 26 deletions(-)

diff --git a/src/backend/access/transam/xlogarchive.c b/src/backend/access/transam/xlogarchive.c
index b9c19b2085..eea7b9b049 100644
--- a/src/backend/access/transam/xlogarchive.c
+++ b/src/backend/access/transam/xlogarchive.c
@@ -492,6 +492,10 @@ XLogArchiveNotify(const char *xlog, bool nudge)
 		return;
 	}
 
+	/* Archive timeline history files as soon as possible. */
+	if (IsTLHistoryFileName(xlog))
+		PgArchForceDirScan();
+
 	/* If caller requested, let archiver know it's got work to do */
 	if (nudge)
 		PgArchWakeup();
diff --git a/src/backend/postmaster/pgarch.c b/src/backend/postmaster/pgarch.c
index 74a7d7c4d0..251d24e741 100644
--- a/src/backend/postmaster/pgarch.c
+++ b/src/backend/postmaster/pgarch.c
@@ -35,6 +35,7 @@
 
 #include "access/xlog.h"
 #include "access/xlog_internal.h"
+#include "lib/binaryheap.h"
 #include "libpq/pqsignal.h"
 #include "miscadmin.h"
 #include "pgstat.h"
@@ -47,6 +48,7 @@
 #include "storage/proc.h"
 #include "storage/procsignal.h"
 #include "storage/shmem.h"
+#include "storage/spin.h"
 #include "utils/guc.h"
 #include "utils/ps_status.h"
 
@@ -72,10 +74,23 @@
  */
 #define NUM_ORPHAN_CLEANUP_RETRIES 3
 
+/*
+ * Maximum number of .ready files to gather per directory scan.
+ */
+#define NUM_FILES_PER_DIRECTORY_SCAN 64
+
 /* Shared memory area for archiver process */
 typedef struct PgArchData
 {
 	int			pgprocno;		/* pgprocno of archiver process */
+
+	/*
+	 * Forces a directory scan in pgarch_readyXlog().  Protected by
+	 * arch_lck.
+	 */
+	bool		force_dir_scan;
+
+	slock_t		arch_lck;
 } PgArchData;
 
 
@@ -86,6 +101,22 @@ typedef struct PgArchData
 static time_t last_sigterm_time = 0;
 static PgArchData *PgArch = NULL;
 
+/*
+ * Stuff for tracking multiple files to archive for each directory scan.
+ * Minimizing the number of directory scans when there are many files to
+ * archive can significantly improve archival rate.
+ *
+ * arch_heap is a max-heap that is used during the directory scan to track
+ * the highest-priority files to archive.  After the directory scan
+ * completes, the file names are stored in ascending order of priority in
+ * arch_files.  pgarch_readyXlog() returns files from arch_files until it
+ * is empty, at which point another directory scan must be performed.
+ */
+static binaryheap *arch_heap = NULL;
+static char arch_filenames[NUM_FILES_PER_DIRECTORY_SCAN][MAX_XFN_CHARS];
+static char *arch_files[NUM_FILES_PER_DIRECTORY_SCAN];
+static int arch_files_size = 0;
+
 /*
  * Flags set by interrupt handlers for later service in the main loop.
  */
@@ -103,6 +134,7 @@ static bool pgarch_readyXlog(char *xlog);
 static void pgarch_archiveDone(char *xlog);
 static void pgarch_die(int code, Datum arg);
 static void HandlePgArchInterrupts(void);
+static int ready_file_comparator(Datum a, Datum b, void *arg);
 
 /* Report shared memory space needed by PgArchShmemInit */
 Size
@@ -129,6 +161,7 @@ PgArchShmemInit(void)
 		/* First time through, so initialize */
 		MemSet(PgArch, 0, PgArchShmemSize());
 		PgArch->pgprocno = INVALID_PGPROCNO;
+		SpinLockInit(&PgArch->arch_lck);
 	}
 }
 
@@ -198,6 +231,10 @@ PgArchiverMain(void)
 	 */
 	PgArch->pgprocno = MyProc->pgprocno;
 
+	/* Initialize our max-heap for prioritizing files to archive. */
+	arch_heap = binaryheap_allocate(NUM_FILES_PER_DIRECTORY_SCAN,
+									ready_file_comparator, NULL);
+
 	pgarch_MainLoop();
 
 	proc_exit(0);
@@ -325,6 +362,9 @@ pgarch_ArchiverCopyLoop(void)
 {
 	char		xlog[MAX_XFN_CHARS + 1];
 
+	/* force directory scan in the first call to pgarch_readyXlog() */
+	arch_files_size = 0;
+
 	/*
 	 * loop through all xlogs with archive_status of .ready and archive
 	 * them...mostly we expect this to be a single file, though it is possible
@@ -600,18 +640,54 @@ pgarch_archiveXlog(char *xlog)
 static bool
 pgarch_readyXlog(char *xlog)
 {
-	/*
-	 * open xlog status directory and read through list of xlogs that have the
-	 * .ready suffix, looking for earliest file. It is possible to optimise
-	 * this code, though only a single file is expected on the vast majority
-	 * of calls, so....
-	 */
 	char		XLogArchiveStatusDir[MAXPGPATH];
 	DIR		   *rldir;
 	struct dirent *rlde;
-	bool		found = false;
-	bool		historyFound = false;
+	bool		force_dir_scan;
 
+	/*
+	 * If a directory scan was requested, clear the stored file names and
+	 * proceed.
+	 */
+	SpinLockAcquire(&PgArch->arch_lck);
+	force_dir_scan = PgArch->force_dir_scan;
+	PgArch->force_dir_scan = false;
+	SpinLockRelease(&PgArch->arch_lck);
+
+	if (force_dir_scan)
+		arch_files_size = 0;
+
+	/*
+	 * If we still have stored file names from the previous directory scan,
+	 * try to return one of those.  We check to make sure the status file
+	 * is still present, as the archive_command for a previous file may
+	 * have already marked it done.
+	 */
+	while (arch_files_size > 0)
+	{
+		struct stat	st;
+		char		status_file[MAXPGPATH];
+		char	   *arch_file;
+
+		arch_files_size--;
+		arch_file = arch_files[arch_files_size];
+		StatusFilePath(status_file, arch_file, ".ready");
+
+		if (stat(status_file, &st) == 0)
+		{
+			strcpy(xlog, arch_file);
+			return true;
+		}
+		else if (errno != ENOENT)
+			ereport(ERROR,
+					(errcode_for_file_access(),
+					 errmsg("could not stat file \"%s\": %m", status_file)));
+	}
+
+	/*
+	 * Open the archive status directory and read through the list of files
+	 * with the .ready suffix, looking for the earliest files.
+	 */
 	snprintf(XLogArchiveStatusDir, MAXPGPATH, XLOGDIR "/archive_status");
 	rldir = AllocateDir(XLogArchiveStatusDir);
 
@@ -619,7 +695,7 @@ pgarch_readyXlog(char *xlog)
 	{
 		int			basenamelen = (int) strlen(rlde->d_name) - 6;
 		char		basename[MAX_XFN_CHARS + 1];
-		bool		ishistory;
+		char	   *arch_file;
 
 		/* Ignore entries with unexpected number of characters */
 		if (basenamelen < MIN_XFN_CHARS ||
@@ -638,32 +714,94 @@ pgarch_readyXlog(char *xlog)
 		memcpy(basename, rlde->d_name, basenamelen);
 		basename[basenamelen] = '\0';
 
-		/* Is this a history file? */
-		ishistory = IsTLHistoryFileName(basename);
-
 		/*
-		 * Consume the file to archive.  History files have the highest
-		 * priority.  If this is the first file or the first history file
-		 * ever, copy it.  In the presence of a history file already chosen as
-		 * target, ignore all other files except history files which have been
-		 * generated for an older timeline than what is already chosen as
-		 * target to archive.
+		 * Store the file in our max-heap if it has a high enough priority.
 		 */
-		if (!found || (ishistory && !historyFound))
+		if (arch_heap->bh_size < NUM_FILES_PER_DIRECTORY_SCAN)
 		{
-			strcpy(xlog, basename);
-			found = true;
-			historyFound = ishistory;
+			/* If the heap isn't full yet, quickly add it. */
+			arch_file = arch_filenames[arch_heap->bh_size];
+			strcpy(arch_file, basename);
+			binaryheap_add_unordered(arch_heap, CStringGetDatum(arch_file));
+
+			/* If we just filled the heap, make it a valid one. */
+			if (arch_heap->bh_size == NUM_FILES_PER_DIRECTORY_SCAN)
+				binaryheap_build(arch_heap);
 		}
-		else if (ishistory || !historyFound)
+		else if (ready_file_comparator(binaryheap_first(arch_heap),
+									   CStringGetDatum(basename), NULL) > 0)
 		{
-			if (strcmp(basename, xlog) < 0)
-				strcpy(xlog, basename);
+			/*
+			 * Remove the lowest priority file and add the current one to
+			 * the heap.
+			 */
+			arch_file = DatumGetCString(binaryheap_remove_first(arch_heap));
+			strcpy(arch_file, basename);
+			binaryheap_add(arch_heap, CStringGetDatum(arch_file));
 		}
 	}
 	FreeDir(rldir);
 
-	return found;
+	/* If no files were found, simply return. */
+	if (arch_heap->bh_size == 0)
+		return false;
+
+	/*
+	 * If we didn't fill the heap, we didn't make it a valid one.  Do that
+	 * now.
+	 */
+	if (arch_heap->bh_size < NUM_FILES_PER_DIRECTORY_SCAN)
+		binaryheap_build(arch_heap);
+
+	/*
+	 * Fill arch_files array with the files to archive in ascending order
+	 * of priority.
+	 */
+	arch_files_size = arch_heap->bh_size;
+	for (int i = 0; i < arch_files_size; i++)
+		arch_files[i] = DatumGetCString(binaryheap_remove_first(arch_heap));
+
+	/* Return the highest priority file. */
+	arch_files_size--;
+	strcpy(xlog, arch_files[arch_files_size]);
+
+	return true;
+}
+
+/*
+ * ready_file_comparator
+ *
+ * Compares the archival priority of the given files to archive.
+ */
+static int
+ready_file_comparator(Datum a, Datum b, void *arg)
+{
+	char *a_str = DatumGetCString(a);
+	char *b_str = DatumGetCString(b);
+	bool a_history = IsTLHistoryFileName(a_str);
+	bool b_history = IsTLHistoryFileName(b_str);
+
+	/* Timeline history files always have the highest priority. */
+	if (a_history != b_history)
+		return a_history ? -1 : 1;
+
+	/* Priority is given to older files. */
+	return strcmp(a_str, b_str);
+}
+
+/*
+ * PgArchForceDirScan
+ *
+ * When called, the next call to pgarch_readyXlog() will perform a
+ * directory scan.  This is useful for ensuring that important files such
+ * as timeline history files are archived as quickly as possible.
+ */
+void
+PgArchForceDirScan(void)
+{
+	SpinLockAcquire(&PgArch->arch_lck);
+	PgArch->force_dir_scan = true;
+	SpinLockRelease(&PgArch->arch_lck);
 }
 
 /*
diff --git a/src/include/postmaster/pgarch.h b/src/include/postmaster/pgarch.h
index 1e47a143e1..732615be57 100644
--- a/src/include/postmaster/pgarch.h
+++ b/src/include/postmaster/pgarch.h
@@ -31,5 +31,6 @@ extern void PgArchShmemInit(void);
 extern bool PgArchCanRestart(void);
 extern void PgArchiverMain(void) pg_attribute_noreturn();
 extern void PgArchWakeup(void);
+extern void PgArchForceDirScan(void);
 
 #endif							/* _PGARCH_H */
-- 
2.16.6

