v1-0001-walmethods.c-h-Make-Walfile-an-extensible-struct-.patch

From 16e23b431b08e2c6b2a89f10eb5817f5fc836d66 Mon Sep 17 00:00:00 2001
From: Robert Haas <rhaas@postgresql.org>
Date: Thu, 1 Sep 2022 13:05:25 -0400
Subject: [PATCH v1 1/2] walmethods.c/h: Make Walfile an extensible struct,
 instead of void *

---
 src/bin/pg_basebackup/receivelog.c | 44 ++++++---------
 src/bin/pg_basebackup/walmethods.c | 89 +++++++++++-------------------
 src/bin/pg_basebackup/walmethods.h | 27 ++++++---
 3 files changed, 67 insertions(+), 93 deletions(-)

diff --git a/src/bin/pg_basebackup/receivelog.c b/src/bin/pg_basebackup/receivelog.c
index 5f6fd3201f..c48090661b 100644
--- a/src/bin/pg_basebackup/receivelog.c
+++ b/src/bin/pg_basebackup/receivelog.c
@@ -25,9 +25,8 @@
 #include "receivelog.h"
 #include "streamutil.h"
 
-/* fd and filename for currently open WAL file */
+/* currently open WAL file */
 static Walfile *walfile = NULL;
-static char current_walfile_name[MAXPGPATH] = "";
 static bool reportFlushPosition = false;
 static XLogRecPtr lastFlushPosition = InvalidXLogRecPtr;
 
@@ -82,8 +81,7 @@ mark_file_as_archived(StreamCtl *stream, const char *fname)
  * Open a new WAL file in the specified directory.
  *
  * Returns true if OK; on failure, returns false after printing an error msg.
- * On success, 'walfile' is set to the FD for the file, and the base filename
- * (without partial_suffix) is stored in 'current_walfile_name'.
+ * On success, 'walfile' is set to the FD for the file.
  *
  * The file will be padded to 16Mb with zeroes.
  */
@@ -94,12 +92,13 @@ open_walfile(StreamCtl *stream, XLogRecPtr startpoint)
 	char	   *fn;
 	ssize_t		size;
 	XLogSegNo	segno;
+	char		walfile_name[MAXPGPATH];
 
 	XLByteToSeg(startpoint, segno, WalSegSz);
-	XLogFileName(current_walfile_name, stream->timeline, segno, WalSegSz);
+	XLogFileName(walfile_name, stream->timeline, segno, WalSegSz);
 
 	/* Note that this considers the compression used if necessary */
-	fn = stream->walmethod->get_file_name(current_walfile_name,
+	fn = stream->walmethod->get_file_name(walfile_name,
 										  stream->partial_suffix);
 
 	/*
@@ -126,7 +125,7 @@ open_walfile(StreamCtl *stream, XLogRecPtr startpoint)
 		if (size == WalSegSz)
 		{
 			/* Already padded file. Open it for use */
-			f = stream->walmethod->open_for_write(current_walfile_name, stream->partial_suffix, 0);
+			f = stream->walmethod->open_for_write(walfile_name, stream->partial_suffix, 0);
 			if (f == NULL)
 			{
 				pg_log_error("could not open existing write-ahead log file \"%s\": %s",
@@ -165,7 +164,7 @@ open_walfile(StreamCtl *stream, XLogRecPtr startpoint)
 
 	/* No file existed, so create one */
 
-	f = stream->walmethod->open_for_write(current_walfile_name,
+	f = stream->walmethod->open_for_write(walfile_name,
 										  stream->partial_suffix, WalSegSz);
 	if (f == NULL)
 	{
@@ -196,21 +195,10 @@ close_walfile(StreamCtl *stream, XLogRecPtr pos)
 		return true;
 
 	/* Note that this considers the compression used if necessary */
-	fn = stream->walmethod->get_file_name(current_walfile_name,
+	fn = stream->walmethod->get_file_name(walfile->pathname,
 										  stream->partial_suffix);
 
-	currpos = stream->walmethod->get_current_pos(walfile);
-
-	if (currpos == -1)
-	{
-		pg_log_error("could not determine seek position in file \"%s\": %s",
-					 fn, stream->walmethod->getlasterror());
-		stream->walmethod->close(walfile, CLOSE_UNLINK);
-		walfile = NULL;
-
-		pg_free(fn);
-		return false;
-	}
+	currpos = walfile->currpos;
 
 	if (stream->partial_suffix)
 	{
@@ -247,7 +235,7 @@ close_walfile(StreamCtl *stream, XLogRecPtr pos)
 	if (currpos == WalSegSz && stream->mark_done)
 	{
 		/* writes error message if failed */
-		if (!mark_file_as_archived(stream, current_walfile_name))
+		if (!mark_file_as_archived(stream, walfile->pathname))
 			return false;
 	}
 
@@ -690,7 +678,7 @@ ReceiveXlogStream(PGconn *conn, StreamCtl *stream)
 error:
 	if (walfile != NULL && stream->walmethod->close(walfile, CLOSE_NO_RENAME) != 0)
 		pg_log_error("could not close file \"%s\": %s",
-					 current_walfile_name, stream->walmethod->getlasterror());
+					 walfile->pathname, stream->walmethod->getlasterror());
 	walfile = NULL;
 	return false;
 }
@@ -777,7 +765,7 @@ HandleCopyStream(PGconn *conn, StreamCtl *stream,
 		{
 			if (stream->walmethod->sync(walfile) != 0)
 				pg_fatal("could not fsync file \"%s\": %s",
-						 current_walfile_name, stream->walmethod->getlasterror());
+						 walfile->pathname, stream->walmethod->getlasterror());
 			lastFlushPosition = blockpos;
 
 			/*
@@ -1024,7 +1012,7 @@ ProcessKeepaliveMsg(PGconn *conn, StreamCtl *stream, char *copybuf, int len,
 			 */
 			if (stream->walmethod->sync(walfile) != 0)
 				pg_fatal("could not fsync file \"%s\": %s",
-						 current_walfile_name, stream->walmethod->getlasterror());
+						 walfile->pathname, stream->walmethod->getlasterror());
 			lastFlushPosition = blockpos;
 		}
 
@@ -1092,10 +1080,10 @@ ProcessXLogDataMsg(PGconn *conn, StreamCtl *stream, char *copybuf, int len,
 	else
 	{
 		/* More data in existing segment */
-		if (stream->walmethod->get_current_pos(walfile) != xlogoff)
+		if (walfile->currpos != xlogoff)
 		{
 			pg_log_error("got WAL data offset %08x, expected %08x",
-						 xlogoff, (int) stream->walmethod->get_current_pos(walfile));
+						 xlogoff, (int) walfile->currpos);
 			return false;
 		}
 	}
@@ -1129,7 +1117,7 @@ ProcessXLogDataMsg(PGconn *conn, StreamCtl *stream, char *copybuf, int len,
 									 bytes_to_write) != bytes_to_write)
 		{
 			pg_log_error("could not write %d bytes to WAL file \"%s\": %s",
-						 bytes_to_write, current_walfile_name,
+						 bytes_to_write, walfile->pathname,
 						 stream->walmethod->getlasterror());
 			return false;
 		}
diff --git a/src/bin/pg_basebackup/walmethods.c b/src/bin/pg_basebackup/walmethods.c
index e90aa0ba37..096b8686e4 100644
--- a/src/bin/pg_basebackup/walmethods.c
+++ b/src/bin/pg_basebackup/walmethods.c
@@ -62,9 +62,8 @@ static DirectoryMethodData *dir_data = NULL;
  */
 typedef struct DirectoryMethodFile
 {
+	Walfile		base;
 	int			fd;
-	off_t		currpos;
-	char	   *pathname;
 	char	   *fullpath;
 	char	   *temp_suffix;
 #ifdef HAVE_LIBZ
@@ -104,7 +103,7 @@ dir_get_file_name(const char *pathname, const char *temp_suffix)
 	return filename;
 }
 
-static Walfile
+static Walfile *
 dir_open_for_write(const char *pathname, const char *temp_suffix, size_t pad_to_size)
 {
 	char		tmppath[MAXPGPATH];
@@ -279,18 +278,18 @@ dir_open_for_write(const char *pathname, const char *temp_suffix, size_t pad_to_
 	}
 #endif
 
+	f->base.currpos = 0;
+	f->base.pathname = pg_strdup(pathname);
 	f->fd = fd;
-	f->currpos = 0;
-	f->pathname = pg_strdup(pathname);
 	f->fullpath = pg_strdup(tmppath);
 	if (temp_suffix)
 		f->temp_suffix = pg_strdup(temp_suffix);
 
-	return f;
+	return &f->base;
 }
 
 static ssize_t
-dir_write(Walfile f, const void *buf, size_t count)
+dir_write(Walfile *f, const void *buf, size_t count)
 {
 	ssize_t		r;
 	DirectoryMethodFile *df = (DirectoryMethodFile *) f;
@@ -366,22 +365,12 @@ dir_write(Walfile f, const void *buf, size_t count)
 		}
 	}
 	if (r > 0)
-		df->currpos += r;
+		df->base.currpos += r;
 	return r;
 }
 
-static off_t
-dir_get_current_pos(Walfile f)
-{
-	Assert(f != NULL);
-	dir_clear_error();
-
-	/* Use a cached value to prevent lots of reseeks */
-	return ((DirectoryMethodFile *) f)->currpos;
-}
-
 static int
-dir_close(Walfile f, WalCloseMethod method)
+dir_close(Walfile *f, WalCloseMethod method)
 {
 	int			r;
 	DirectoryMethodFile *df = (DirectoryMethodFile *) f;
@@ -440,13 +429,13 @@ dir_close(Walfile f, WalCloseMethod method)
 			 * If we have a temp prefix, normal operation is to rename the
 			 * file.
 			 */
-			filename = dir_get_file_name(df->pathname, df->temp_suffix);
+			filename = dir_get_file_name(df->base.pathname, df->temp_suffix);
 			snprintf(tmppath, sizeof(tmppath), "%s/%s",
 					 dir_data->basedir, filename);
 			pg_free(filename);
 
 			/* permanent name, so no need for the prefix */
-			filename2 = dir_get_file_name(df->pathname, NULL);
+			filename2 = dir_get_file_name(df->base.pathname, NULL);
 			snprintf(tmppath2, sizeof(tmppath2), "%s/%s",
 					 dir_data->basedir, filename2);
 			pg_free(filename2);
@@ -467,7 +456,7 @@ dir_close(Walfile f, WalCloseMethod method)
 			char	   *filename;
 
 			/* Unlink the file once it's closed */
-			filename = dir_get_file_name(df->pathname, df->temp_suffix);
+			filename = dir_get_file_name(df->base.pathname, df->temp_suffix);
 			snprintf(tmppath, sizeof(tmppath), "%s/%s",
 					 dir_data->basedir, filename);
 			pg_free(filename);
@@ -498,7 +487,7 @@ dir_close(Walfile f, WalCloseMethod method)
 	LZ4F_freeCompressionContext(df->ctx);
 #endif
 
-	pg_free(df->pathname);
+	pg_free(df->base.pathname);
 	pg_free(df->fullpath);
 	pg_free(df->temp_suffix);
 	pg_free(df);
@@ -507,7 +496,7 @@ dir_close(Walfile f, WalCloseMethod method)
 }
 
 static int
-dir_sync(Walfile f)
+dir_sync(Walfile *f)
 {
 	int			r;
 
@@ -630,7 +619,6 @@ CreateWalDirectoryMethod(const char *basedir,
 	method = pg_malloc0(sizeof(WalWriteMethod));
 	method->open_for_write = dir_open_for_write;
 	method->write = dir_write;
-	method->get_current_pos = dir_get_current_pos;
 	method->get_file_size = dir_get_file_size;
 	method->get_file_name = dir_get_file_name;
 	method->compression_algorithm = dir_compression_algorithm;
@@ -665,10 +653,9 @@ FreeWalDirectoryMethod(void)
 
 typedef struct TarMethodFile
 {
+	Walfile		base;
 	off_t		ofs_start;		/* Where does the *header* for this file start */
-	off_t		currpos;
 	char		header[TAR_BLOCK_SIZE];
-	char	   *pathname;
 	size_t		pad_to_size;
 } TarMethodFile;
 
@@ -755,7 +742,7 @@ tar_write_compressed_data(void *buf, size_t count, bool flush)
 #endif
 
 static ssize_t
-tar_write(Walfile f, const void *buf, size_t count)
+tar_write(Walfile *f, const void *buf, size_t count)
 {
 	ssize_t		r;
 
@@ -773,7 +760,7 @@ tar_write(Walfile f, const void *buf, size_t count)
 			tar_data->lasterrno = errno ? errno : ENOSPC;
 			return -1;
 		}
-		((TarMethodFile *) f)->currpos += r;
+		f->currpos += r;
 		return r;
 	}
 #ifdef HAVE_LIBZ
@@ -781,7 +768,7 @@ tar_write(Walfile f, const void *buf, size_t count)
 	{
 		if (!tar_write_compressed_data(unconstify(void *, buf), count, false))
 			return -1;
-		((TarMethodFile *) f)->currpos += count;
+		f->currpos += count;
 		return count;
 	}
 #endif
@@ -803,7 +790,7 @@ tar_write_padding_data(TarMethodFile *f, size_t bytes)
 	while (bytesleft)
 	{
 		size_t		bytestowrite = Min(bytesleft, XLOG_BLCKSZ);
-		ssize_t		r = tar_write(f, zerobuf.data, bytestowrite);
+		ssize_t		r = tar_write(&f->base, zerobuf.data, bytestowrite);
 
 		if (r < 0)
 			return false;
@@ -824,7 +811,7 @@ tar_get_file_name(const char *pathname, const char *temp_suffix)
 	return filename;
 }
 
-static Walfile
+static Walfile *
 tar_open_for_write(const char *pathname, const char *temp_suffix, size_t pad_to_size)
 {
 	char	   *tmppath;
@@ -920,7 +907,7 @@ tar_open_for_write(const char *pathname, const char *temp_suffix, size_t pad_to_
 		tar_data->currentfile = NULL;
 		return NULL;
 	}
-	tar_data->currentfile->currpos = 0;
+	tar_data->currentfile->base.currpos = 0;
 
 	if (tar_data->compression_algorithm == PG_COMPRESSION_NONE)
 	{
@@ -957,7 +944,7 @@ tar_open_for_write(const char *pathname, const char *temp_suffix, size_t pad_to_
 		Assert(false);
 	}
 
-	tar_data->currentfile->pathname = pg_strdup(pathname);
+	tar_data->currentfile->base.pathname = pg_strdup(pathname);
 
 	/*
 	 * Uncompressed files are padded on creation, but for compression we can't
@@ -980,11 +967,11 @@ tar_open_for_write(const char *pathname, const char *temp_suffix, size_t pad_to_
 				return NULL;
 			}
 
-			tar_data->currentfile->currpos = 0;
+			tar_data->currentfile->base.currpos = 0;
 		}
 	}
 
-	return tar_data->currentfile;
+	return &tar_data->currentfile->base;
 }
 
 static ssize_t
@@ -1003,17 +990,8 @@ tar_compression_algorithm(void)
 	return tar_data->compression_algorithm;
 }
 
-static off_t
-tar_get_current_pos(Walfile f)
-{
-	Assert(f != NULL);
-	tar_clear_error();
-
-	return ((TarMethodFile *) f)->currpos;
-}
-
 static int
-tar_sync(Walfile f)
+tar_sync(Walfile *f)
 {
 	int			r;
 
@@ -1037,7 +1015,7 @@ tar_sync(Walfile f)
 }
 
 static int
-tar_close(Walfile f, WalCloseMethod method)
+tar_close(Walfile *f, WalCloseMethod method)
 {
 	ssize_t		filesize;
 	int			padding;
@@ -1065,7 +1043,7 @@ tar_close(Walfile f, WalCloseMethod method)
 			return -1;
 		}
 
-		pg_free(tf->pathname);
+		pg_free(tf->base.pathname);
 		pg_free(tf);
 		tar_data->currentfile = NULL;
 
@@ -1085,7 +1063,7 @@ tar_close(Walfile f, WalCloseMethod method)
 			 * A compressed tarfile is padded on close since we cannot know
 			 * the size of the compressed output until the end.
 			 */
-			size_t		sizeleft = tf->pad_to_size - tf->currpos;
+			size_t		sizeleft = tf->pad_to_size - tf->base.currpos;
 
 			if (sizeleft)
 			{
@@ -1099,7 +1077,7 @@ tar_close(Walfile f, WalCloseMethod method)
 			 * An uncompressed tarfile was padded on creation, so just adjust
 			 * the current position as if we seeked to the end.
 			 */
-			tf->currpos = tf->pad_to_size;
+			tf->base.currpos = tf->pad_to_size;
 		}
 	}
 
@@ -1107,7 +1085,7 @@ tar_close(Walfile f, WalCloseMethod method)
 	 * Get the size of the file, and pad out to a multiple of the tar block
 	 * size.
 	 */
-	filesize = tar_get_current_pos(f);
+	filesize = f->currpos;
 	padding = tarPaddingBytesRequired(filesize);
 	if (padding)
 	{
@@ -1140,7 +1118,7 @@ tar_close(Walfile f, WalCloseMethod method)
 		 * We overwrite it with what it was before if we have no tempname,
 		 * since we're going to write the buffer anyway.
 		 */
-		strlcpy(&(tf->header[0]), tf->pathname, 100);
+		strlcpy(&(tf->header[0]), tf->base.pathname, 100);
 
 	print_tar_number(&(tf->header[148]), 8, tarChecksum(((TarMethodFile *) f)->header));
 	if (lseek(tar_data->fd, tf->ofs_start, SEEK_SET) != ((TarMethodFile *) f)->ofs_start)
@@ -1199,11 +1177,11 @@ tar_close(Walfile f, WalCloseMethod method)
 	{
 		/* XXX this seems pretty bogus; why is only this case fatal? */
 		pg_fatal("could not fsync file \"%s\": %s",
-				 tf->pathname, tar_getlasterror());
+				 tf->base.pathname, tar_getlasterror());
 	}
 
 	/* Clean up and done */
-	pg_free(tf->pathname);
+	pg_free(tf->base.pathname);
 	pg_free(tf);
 	tar_data->currentfile = NULL;
 
@@ -1227,7 +1205,7 @@ tar_finish(void)
 
 	if (tar_data->currentfile)
 	{
-		if (tar_close(tar_data->currentfile, CLOSE_NORMAL) != 0)
+		if (tar_close(&tar_data->currentfile->base, CLOSE_NORMAL) != 0)
 			return false;
 	}
 
@@ -1343,7 +1321,6 @@ CreateWalTarMethod(const char *tarbase,
 	method = pg_malloc0(sizeof(WalWriteMethod));
 	method->open_for_write = tar_open_for_write;
 	method->write = tar_write;
-	method->get_current_pos = tar_get_current_pos;
 	method->get_file_size = tar_get_file_size;
 	method->get_file_name = tar_get_file_name;
 	method->compression_algorithm = tar_compression_algorithm;
diff --git a/src/bin/pg_basebackup/walmethods.h b/src/bin/pg_basebackup/walmethods.h
index 76530dc941..cf5ed87fbe 100644
--- a/src/bin/pg_basebackup/walmethods.h
+++ b/src/bin/pg_basebackup/walmethods.h
@@ -11,7 +11,20 @@
 
 #include "common/compression.h"
 
-typedef void *Walfile;
+struct WalWriteMethod;
+typedef struct WalWriteMethod WalWriteMethod;
+
+typedef struct
+{
+	off_t		currpos;
+	char	   *pathname;
+	/*
+	 * MORE DATA FOLLOWS AT END OF STRUCT
+	 *
+	 * Each WalWriteMethod is expected to embed this as the first member of
+	 * a larger struct with method-specific fields following.
+	 */
+} Walfile;
 
 typedef enum
 {
@@ -30,7 +43,6 @@ typedef enum
  * care not to clobber errno between a failed method call and use of
  * getlasterror() to retrieve the message.
  */
-typedef struct WalWriteMethod WalWriteMethod;
 struct WalWriteMethod
 {
 	/*
@@ -39,13 +51,13 @@ struct WalWriteMethod
 	 * automatically renamed in close(). If pad_to_size is specified, the file
 	 * will be padded with NUL up to that size, if supported by the Walmethod.
 	 */
-	Walfile		(*open_for_write) (const char *pathname, const char *temp_suffix, size_t pad_to_size);
+	Walfile	   *(*open_for_write) (const char *pathname, const char *temp_suffix, size_t pad_to_size);
 
 	/*
 	 * Close an open Walfile, using one or more methods for handling automatic
 	 * unlinking etc. Returns 0 on success, other values for error.
 	 */
-	int			(*close) (Walfile f, WalCloseMethod method);
+	int			(*close) (Walfile *f, WalCloseMethod method);
 
 	/* Check if a file exist */
 	bool		(*existsfile) (const char *pathname);
@@ -66,15 +78,12 @@ struct WalWriteMethod
 	 * Write count number of bytes to the file, and return the number of bytes
 	 * actually written or -1 for error.
 	 */
-	ssize_t		(*write) (Walfile f, const void *buf, size_t count);
-
-	/* Return the current position in a file or -1 on error */
-	off_t		(*get_current_pos) (Walfile f);
+	ssize_t		(*write) (Walfile *f, const void *buf, size_t count);
 
 	/*
 	 * fsync the contents of the specified file. Returns 0 on success.
 	 */
-	int			(*sync) (Walfile f);
+	int			(*sync) (Walfile *f);
 
 	/*
 	 * Clean up the Walmethod, closing any shared resources. For methods like
-- 
2.24.3 (Apple Git-128)