public inbox for [email protected]
help / color / mirror / Atom feed[PATCH 05/20] Use cf* abstraction in archiver and tar
3+ messages / 3 participants
[nested] [flat]
* [PATCH 05/20] Use cf* abstraction in archiver and tar
@ 2020-12-25 06:34 Justin Pryzby <[email protected]>
0 siblings, 0 replies; 3+ messages in thread
From: Justin Pryzby @ 2020-12-25 06:34 UTC (permalink / raw)
..rather than direct, conditional calls to gzopen/fopen.
See also: bf9aa490db24b2334b3595ee33653bf2fe39208c
---
src/bin/pg_dump/compress_io.c | 53 +++++++++++++
src/bin/pg_dump/compress_io.h | 5 ++
src/bin/pg_dump/pg_backup_archiver.c | 109 +++++++++------------------
src/bin/pg_dump/pg_backup_archiver.h | 16 ++--
src/bin/pg_dump/pg_backup_tar.c | 85 +++++----------------
5 files changed, 117 insertions(+), 151 deletions(-)
diff --git a/src/bin/pg_dump/compress_io.c b/src/bin/pg_dump/compress_io.c
index 21957d68f3..d66d6f60f5 100644
--- a/src/bin/pg_dump/compress_io.c
+++ b/src/bin/pg_dump/compress_io.c
@@ -540,6 +540,58 @@ cfopen(const char *path, const char *mode, Compress *compression)
}
}
+/*
+ * Open a file descriptor, with specified compression.
+ * Returns an opaque cfp object.
+ */
+cfp *
+cfdopen(int fd, const char *mode, Compress *compression)
+{
+ cfp *fp = pg_malloc0(sizeof(cfp));
+
+ switch (compression->alg)
+ {
+#ifdef HAVE_LIBZ
+ case COMPR_ALG_LIBZ:
+ if (compression->level != Z_DEFAULT_COMPRESSION)
+ {
+ /* user has specified a compression level, so tell zlib to use it */
+ char mode_compression[32];
+
+ snprintf(mode_compression, sizeof(mode_compression), "%s%d",
+ mode, compression->level);
+ fp->compressedfp = gzdopen(fd, mode_compression);
+ }
+ else
+ {
+ /* don't specify a level, just use the zlib default */
+ fp->compressedfp = gzdopen(fd, mode);
+ }
+
+ if (fp->compressedfp == NULL)
+ {
+ free_keep_errno(fp);
+ fp = NULL;
+ }
+ return fp;
+#endif
+
+ case COMPR_ALG_NONE:
+ fp->uncompressedfp = fdopen(fd, mode);
+ if (fp->uncompressedfp == NULL)
+ {
+ free_keep_errno(fp);
+ fp = NULL;
+ }
+ else
+ setvbuf(fp->uncompressedfp, NULL, _IONBF, 0);
+ return fp;
+
+ default:
+ /* Should not happen */
+ fatal("requested compression not available in this installation");
+ }
+}
int
cfread(void *ptr, int size, cfp *fp)
@@ -616,6 +668,7 @@ cfgets(cfp *fp, char *buf, int len)
return fgets(buf, len, fp->uncompressedfp);
}
+/* Close the given compressed or uncompressed stream; return 0 on success. */
int
cfclose(cfp *fp)
{
diff --git a/src/bin/pg_dump/compress_io.h b/src/bin/pg_dump/compress_io.h
index fb9d659acc..318a6b5340 100644
--- a/src/bin/pg_dump/compress_io.h
+++ b/src/bin/pg_dump/compress_io.h
@@ -21,6 +21,10 @@
#define ZLIB_OUT_SIZE 4096
#define ZLIB_IN_SIZE 4096
+/* Forward declaration */
+struct ArchiveHandle;
+typedef struct _archiveHandle ArchiveHandle;
+
/* Prototype for callback function to WriteDataToArchive() */
typedef void (*WriteFunc) (ArchiveHandle *AH, const char *buf, size_t len);
@@ -58,6 +62,7 @@ extern const struct compressLibs compresslibs[];
typedef struct cfp cfp;
extern cfp *cfopen(const char *path, const char *mode, Compress *compression);
+extern cfp *cfdopen(int fd, const char *mode, Compress *compression);
extern cfp *cfopen_read(const char *path, const char *mode, Compress *compression);
extern cfp *cfopen_write(const char *path, const char *mode, Compress *compression);
extern int cfread(void *ptr, int size, cfp *fp);
diff --git a/src/bin/pg_dump/pg_backup_archiver.c b/src/bin/pg_dump/pg_backup_archiver.c
index 3eb6c55600..bd06fbb787 100644
--- a/src/bin/pg_dump/pg_backup_archiver.c
+++ b/src/bin/pg_dump/pg_backup_archiver.c
@@ -39,17 +39,11 @@
#include "pg_backup_archiver.h"
#include "pg_backup_db.h"
#include "pg_backup_utils.h"
+#include "compress_io.h"
#define TEXT_DUMP_HEADER "--\n-- PostgreSQL database dump\n--\n\n"
#define TEXT_DUMPALL_HEADER "--\n-- PostgreSQL database cluster dump\n--\n\n"
-/* state needed to save/restore an archive's output target */
-typedef struct _outputContext
-{
- void *OF;
- int gzOut;
-} OutputContext;
-
/*
* State for tracking TocEntrys that are ready to process during a parallel
* restore. (This used to be a list, and we still call it that, though now
@@ -99,8 +93,8 @@ static int RestoringToDB(ArchiveHandle *AH);
static void dump_lo_buf(ArchiveHandle *AH);
static void dumpTimestamp(ArchiveHandle *AH, const char *msg, time_t tim);
static void SetOutput(ArchiveHandle *AH, const char *filename, Compress *compress);
-static OutputContext SaveOutput(ArchiveHandle *AH);
-static void RestoreOutput(ArchiveHandle *AH, OutputContext savedContext);
+static cfp *SaveOutput(ArchiveHandle *AH);
+static void RestoreOutput(ArchiveHandle *AH, cfp *fp);
static int restore_toc_entry(ArchiveHandle *AH, TocEntry *te, bool is_parallel);
static void restore_toc_entries_prefork(ArchiveHandle *AH,
@@ -270,10 +264,8 @@ CloseArchive(Archive *AHX)
AH->ClosePtr(AH);
/* Close the output */
- if (AH->gzOut)
- res = GZCLOSE(AH->OF);
- else if (AH->OF != stdout)
- res = fclose(AH->OF);
+ if ((FILE *)AH->OF != stdout)
+ res = cfclose(AH->OF);
if (res != 0)
fatal("could not close output file: %m");
@@ -355,7 +347,7 @@ RestoreArchive(Archive *AHX)
RestoreOptions *ropt = AH->public.ropt;
bool parallel_mode;
TocEntry *te;
- OutputContext sav;
+ cfp *sav;
AH->stage = STAGE_INITIALIZING;
@@ -1120,7 +1112,7 @@ PrintTOCSummary(Archive *AHX)
RestoreOptions *ropt = AH->public.ropt;
TocEntry *te;
teSection curSection;
- OutputContext sav;
+ cfp *sav;
const char *fmtName;
char stamp_str[64];
@@ -1492,6 +1484,7 @@ archprintf(Archive *AH, const char *fmt,...)
static void
SetOutput(ArchiveHandle *AH, const char *filename, Compress *compression)
{
+ char fmode[14];
int fn;
if (filename)
@@ -1511,38 +1504,22 @@ SetOutput(ArchiveHandle *AH, const char *filename, Compress *compression)
else
fn = fileno(stdout);
- /* If compression explicitly requested, use gzopen */
-#ifdef HAVE_LIBZ
- if (compression->alg != COMPR_ALG_NONE)
+ if (fn >= 0)
{
- char fmode[14];
+ /* Handle output to stdout */
+ sprintf(fmode, "%sb%d",
+ AH->mode == archModeAppend ? PG_BINARY_A : PG_BINARY_W,
+ compression->level);
- /* Don't use PG_BINARY_x since this is zlib */
- sprintf(fmode, "wb%d", compression->level);
- if (fn >= 0)
- AH->OF = gzdopen(dup(fn), fmode);
- else
- AH->OF = gzopen(filename, fmode);
- AH->gzOut = 1;
+ AH->OF = cfdopen(dup(fn), fmode, compression);
}
else
-#endif
- { /* Use fopen */
- if (AH->mode == archModeAppend)
- {
- if (fn >= 0)
- AH->OF = fdopen(dup(fn), PG_BINARY_A);
- else
- AH->OF = fopen(filename, PG_BINARY_A);
- }
- else
- {
- if (fn >= 0)
- AH->OF = fdopen(dup(fn), PG_BINARY_W);
- else
- AH->OF = fopen(filename, PG_BINARY_W);
- }
- AH->gzOut = 0;
+ {
+ Assert(filename != NULL);
+ sprintf(fmode, "%cb%d",
+ AH->mode == archModeAppend ? 'a' : 'w',
+ compression->level);
+ AH->OF = cfopen(filename, fmode, compression);
}
if (!AH->OF)
@@ -1554,32 +1531,22 @@ SetOutput(ArchiveHandle *AH, const char *filename, Compress *compression)
}
}
-static OutputContext
+/* Return a pointer to the old FP */
+static cfp *
SaveOutput(ArchiveHandle *AH)
{
- OutputContext sav;
-
- sav.OF = AH->OF;
- sav.gzOut = AH->gzOut;
-
- return sav;
+ return AH->OF;
}
static void
-RestoreOutput(ArchiveHandle *AH, OutputContext savedContext)
+RestoreOutput(ArchiveHandle *AH, cfp *savedContext)
{
int res;
-
- if (AH->gzOut)
- res = GZCLOSE(AH->OF);
- else
- res = fclose(AH->OF);
-
+ res = cfclose(AH->OF);
if (res != 0)
fatal("could not close output file: %m");
- AH->gzOut = savedContext.gzOut;
- AH->OF = savedContext.OF;
+ AH->OF = savedContext;
}
@@ -1703,22 +1670,14 @@ ahwrite(const void *ptr, size_t size, size_t nmemb, ArchiveHandle *AH)
bytes_written = size * nmemb;
}
- else if (AH->gzOut)
- bytes_written = GZWRITE(ptr, size, nmemb, AH->OF);
else if (AH->CustomOutPtr)
bytes_written = AH->CustomOutPtr(AH, ptr, size * nmemb);
-
+ else if (RestoringToDB(AH))
+ /* If we're doing a restore, and it's direct to DB, and we're
+ * connected then send it to the DB. */
+ bytes_written = ExecuteSqlCommandBuf(&AH->public, (const char *) ptr, size * nmemb);
else
- {
- /*
- * If we're doing a restore, and it's direct to DB, and we're
- * connected then send it to the DB.
- */
- if (RestoringToDB(AH))
- bytes_written = ExecuteSqlCommandBuf(&AH->public, (const char *) ptr, size * nmemb);
- else
- bytes_written = fwrite(ptr, size, nmemb, AH->OF) * size;
- }
+ bytes_written = cfwrite(ptr, size * nmemb, AH->OF);
if (bytes_written != size * nmemb)
WRITE_ERROR_EXIT;
@@ -2127,6 +2086,7 @@ _discoverArchiveFormat(ArchiveHandle *AH)
fh = stdin;
if (!fh)
fatal("could not open input file: %m");
+ setvbuf(fh, NULL, _IONBF, 0);
}
if ((cnt = fread(sig, 1, 5, fh)) != 5)
@@ -2266,6 +2226,7 @@ _allocAH(const char *FileSpec, const ArchiveFormat fmt,
SetupWorkerPtrType setupWorkerPtr)
{
ArchiveHandle *AH;
+ Compress nocompression = {0};
pg_log_debug("allocating AH for %s, format %d",
FileSpec ? FileSpec : "(stdio)", fmt);
@@ -2319,8 +2280,8 @@ _allocAH(const char *FileSpec, const ArchiveFormat fmt,
memset(&(AH->sqlparse), 0, sizeof(AH->sqlparse));
/* Open stdout with no compression for AH output handle */
- AH->gzOut = 0;
- AH->OF = stdout;
+ AH->OF = cfdopen(fileno(stdout), "w", &nocompression);
+ // AH->OF = cfdopen(STDOUT_FILENO, "w", compression); // XXX
/*
* On Windows, we need to use binary mode to read/write non-text files,
diff --git a/src/bin/pg_dump/pg_backup_archiver.h b/src/bin/pg_dump/pg_backup_archiver.h
index 6e033d040e..9f511b49b9 100644
--- a/src/bin/pg_dump/pg_backup_archiver.h
+++ b/src/bin/pg_dump/pg_backup_archiver.h
@@ -30,6 +30,9 @@
#include "pg_backup.h"
#include "pqexpbuffer.h"
+/* Forward declaration XXX: CIRCULAR */
+typedef struct cfp cfp;
+
#define LOBBUFSIZE 16384
/*
@@ -38,19 +41,11 @@
*/
#ifdef HAVE_LIBZ
#include <zlib.h>
-#define GZCLOSE(fh) gzclose(fh)
-#define GZWRITE(p, s, n, fh) gzwrite(fh, p, (n) * (s))
-#define GZREAD(p, s, n, fh) gzread(fh, p, (n) * (s))
-#define GZEOF(fh) gzeof(fh)
#else
-#define GZCLOSE(fh) fclose(fh)
-#define GZWRITE(p, s, n, fh) (fwrite(p, s, n, fh) * (s))
-#define GZREAD(p, s, n, fh) fread(p, s, n, fh)
-#define GZEOF(fh) feof(fh)
+
/* this is just the redefinition of a libz constant, in case zlib isn't
* available */
#define Z_DEFAULT_COMPRESSION (-1)
-
typedef struct _z_stream
{
void *next_in;
@@ -318,8 +313,7 @@ struct _archiveHandle
char *fSpec; /* Archive File Spec */
FILE *FH; /* General purpose file handle */
- void *OF;
- int gzOut; /* Output file */
+ cfp *OF; /* Output file (compressed or not) */
struct _tocEntry *toc; /* Header of circular list of TOC entries */
int tocCount; /* Number of TOC entries */
diff --git a/src/bin/pg_dump/pg_backup_tar.c b/src/bin/pg_dump/pg_backup_tar.c
index 4ba79ab924..16f4e0792a 100644
--- a/src/bin/pg_dump/pg_backup_tar.c
+++ b/src/bin/pg_dump/pg_backup_tar.c
@@ -66,12 +66,7 @@ static void _EndBlobs(ArchiveHandle *AH, TocEntry *te);
typedef struct
{
-#ifdef HAVE_LIBZ
- gzFile zFH;
-#else
- FILE *zFH;
-#endif
- FILE *nFH;
+ cfp *FH;
FILE *tarFH;
FILE *tmpFH;
char *targetFile;
@@ -191,7 +186,7 @@ InitArchiveFmt_Tar(ArchiveHandle *AH)
* Make unbuffered since we will dup() it, and the buffers screw each
* other
*/
- /* setvbuf(ctx->tarFH, NULL, _IONBF, 0); */
+ // setvbuf(ctx->tarFH, NULL, _IONBF, 0);
ctx->hasSeek = checkSeek(ctx->tarFH);
@@ -223,7 +218,7 @@ InitArchiveFmt_Tar(ArchiveHandle *AH)
* Make unbuffered since we will dup() it, and the buffers screw each
* other
*/
- /* setvbuf(ctx->tarFH, NULL, _IONBF, 0); */
+ setvbuf(ctx->tarFH, NULL, _IONBF, 0);
ctx->tarFHpos = 0;
@@ -321,10 +316,6 @@ tarOpen(ArchiveHandle *AH, const char *filename, char mode)
lclContext *ctx = (lclContext *) AH->formatData;
TAR_MEMBER *tm;
-#ifdef HAVE_LIBZ
- char fmode[14];
-#endif
-
if (mode == 'r')
{
tm = _tarPositionTo(AH, filename);
@@ -345,16 +336,10 @@ tarOpen(ArchiveHandle *AH, const char *filename, char mode)
}
}
-#ifdef HAVE_LIBZ
-
if (AH->compression.alg == COMPR_ALG_NONE)
- tm->nFH = ctx->tarFH;
+ tm->FH = cfdopen(dup(fileno(ctx->tarFH)), "rb", &AH->compression);
else
fatal("compression is not supported by tar archive format");
- /* tm->zFH = gzdopen(dup(fileno(ctx->tarFH)), "rb"); */
-#else
- tm->nFH = ctx->tarFH;
-#endif
}
else
{
@@ -406,21 +391,11 @@ tarOpen(ArchiveHandle *AH, const char *filename, char mode)
umask(old_umask);
-#ifdef HAVE_LIBZ
-
- if (AH->compression.alg != COMPR_ALG_NONE)
- {
- sprintf(fmode, "wb%d", AH->compression.level);
- tm->zFH = gzdopen(dup(fileno(tm->tmpFH)), fmode);
- if (tm->zFH == NULL)
- fatal("could not open temporary file");
- }
- else
- tm->nFH = tm->tmpFH;
-#else
-
- tm->nFH = tm->tmpFH;
-#endif
+ tm->FH = cfdopen(dup(fileno(tm->tmpFH)),
+ mode == 'r' ? "r" : "w",
+ &AH->compression);
+ if (tm->FH == NULL)
+ fatal("could not open temporary file");
tm->AH = AH;
tm->targetFile = pg_strdup(filename);
@@ -435,12 +410,14 @@ tarOpen(ArchiveHandle *AH, const char *filename, char mode)
static void
tarClose(ArchiveHandle *AH, TAR_MEMBER *th)
{
+ int res;
+
/*
* Close the GZ file since we dup'd. This will flush the buffers.
*/
- if (AH->compression.alg != COMPR_ALG_NONE)
- if (GZCLOSE(th->zFH) != 0)
- fatal("could not close tar member");
+ res = cfclose(th->FH);
+ if (res != 0)
+ fatal("could not close tar member");
if (th->mode == 'w')
_tarAddFile(AH, th); /* This will close the temp file */
@@ -453,8 +430,7 @@ tarClose(ArchiveHandle *AH, TAR_MEMBER *th)
if (th->targetFile)
free(th->targetFile);
- th->nFH = NULL;
- th->zFH = NULL;
+ th->FH = NULL;
}
#ifdef __NOT_USED__
@@ -540,29 +516,9 @@ _tarReadRaw(ArchiveHandle *AH, void *buf, size_t len, TAR_MEMBER *th, FILE *fh)
}
else if (th)
{
- if (th->zFH)
- {
- res = GZREAD(&((char *) buf)[used], 1, len, th->zFH);
- if (res != len && !GZEOF(th->zFH))
- {
-#ifdef HAVE_LIBZ
- int errnum;
- const char *errmsg = gzerror(th->zFH, &errnum);
-
- fatal("could not read from input file: %s",
- errnum == Z_ERRNO ? strerror(errno) : errmsg);
-#else
- fatal("could not read from input file: %s",
- strerror(errno));
-#endif
- }
- }
- else
- {
- res = fread(&((char *) buf)[used], 1, len, th->nFH);
- if (res != len && !feof(th->nFH))
- READ_ERROR_EXIT(th->nFH);
- }
+ res = cfread(&((char *) buf)[used], len, th->FH);
+ if (res != len && !cfeof(th->FH))
+ fatal("could not read from input file: %m");
}
}
@@ -594,10 +550,7 @@ tarWrite(const void *buf, size_t len, TAR_MEMBER *th)
{
size_t res;
- if (th->zFH != NULL)
- res = GZWRITE(buf, 1, len, th->zFH);
- else
- res = fwrite(buf, 1, len, th->nFH);
+ res = cfwrite(buf, len, th->FH);
th->pos += res;
return res;
--
2.17.0
--GRPZ8SYKNexpdSJ7
Content-Type: text/x-diff; charset=us-ascii
Content-Disposition: attachment;
filename="0006-pg_dump-zstd-compression.patch"
^ permalink raw reply [nested|flat] 3+ messages in thread
* What's our minimum supported Python version?
@ 2025-04-19 16:17 Tom Lane <[email protected]>
2025-04-22 08:13 ` Re: What's our minimum supported Python version? Daniel Gustafsson <[email protected]>
0 siblings, 1 reply; 3+ messages in thread
From: Tom Lane @ 2025-04-19 16:17 UTC (permalink / raw)
To: [email protected]
Our fine manual claims the answer to $SUBJECT is 3.2.
However, there is room to doubt that our code still actually
works with 3.2, because the oldest version I can find being
tested by the buildfarm is topminnow's 3.4.2. The next oldest
is shelduck's 3.4.10, and everything else has 3.5.2 or newer.
topminnow hasn't reported in a couple months, so it may
actually be dead.
The reason I bring this up is that I found out the hard way
that src/test/modules/oauth_validator fails on RHEL8, because
its oauth_server.py script is not compatible with the 3.6.8
version of Python supplied by this distro. (There are 22
buildfarm animals running 3.6.8, presumably mostly also Red
Hat-derived platforms, so I'm not going to apologize for
my workstation being a little long in the tooth.)
I think we need to do some combination of moving our
minimum-supported-version goalposts forward, making sure that
whatever we claim is the minimum Python version is actually
being tested in the buildfarm, and fixing oauth_server.py
so that it works on that version.
I am not familiar enough with the Python landscape to
have an informed opinion about what the minimum supported
version should be. But I'll present this data scraped
from the buildfarm about how many animals are running what
(counting animals that have reported since 2025-01-01):
Count Version
1 3.4.2
1 3.4.10
2 3.5.2
3 3.5.3
2 3.6.5
22 3.6.8
3 3.6.9
5 3.6.15
1 3.7.1
2 3.7.3
1 3.7.10
2 3.7.16
1 3.8.8
4 3.8.10
15 3.9.2
6 3.9.16
6 3.9.18
1 3.9.19
2 3.9.20
13 3.9.21
6 3.10.x
18 3.11.x
21 3.12.x
22 3.13.x
160 total
Thoughts?
regards, tom lane
^ permalink raw reply [nested|flat] 3+ messages in thread
* Re: What's our minimum supported Python version?
2025-04-19 16:17 What's our minimum supported Python version? Tom Lane <[email protected]>
@ 2025-04-22 08:13 ` Daniel Gustafsson <[email protected]>
0 siblings, 0 replies; 3+ messages in thread
From: Daniel Gustafsson @ 2025-04-22 08:13 UTC (permalink / raw)
To: Tom Lane <[email protected]>; +Cc: [email protected]
> On 19 Apr 2025, at 18:17, Tom Lane <[email protected]> wrote:
> The reason I bring this up is that I found out the hard way
> that src/test/modules/oauth_validator fails on RHEL8, because
> its oauth_server.py script is not compatible with the 3.6.8
> version of Python supplied by this distro.
Do you have the error message/log for the failure handy?
--
Daniel Gustafsson
^ permalink raw reply [nested|flat] 3+ messages in thread
end of thread, other threads:[~2025-04-22 08:13 UTC | newest]
Thread overview: 3+ messages (download: mbox mbox.gz follow: Atom feed)
-- links below jump to the message on this page --
2020-12-25 06:34 [PATCH 05/20] Use cf* abstraction in archiver and tar Justin Pryzby <[email protected]>
2025-04-19 16:17 What's our minimum supported Python version? Tom Lane <[email protected]>
2025-04-22 08:13 ` Re: What's our minimum supported Python version? Daniel Gustafsson <[email protected]>
This inbox is served by agora; see mirroring instructions
for how to clone and mirror all data and code used for this inbox