walmethods.c/h: Make WalWriteMethod more object-oriented.
Normally when we use object-oriented programming techniques, we provide a pointer to an object and then some way of looking up the associated table of callbacks, but walmethods.c/h took the alternative approach of providing only a pointer to the table of callbacks and thus imposed the artificial restriction that there could only ever be one object of each type, so that the callbacks could find it via a global variable. That doesn't seem like the right idea, so revise the approach. Each callback which does not already have an argument of type Walfile * now takes a pointer to the relevant WalWriteMethod * so that these callbacks need not rely on there being only one object of each type. Freeing a WalWriteMethod is now performed via a callback provided for that purpose rather than requiring the caller to know which WAL method they want to free. Discussion: http://postgr.es/m/CA+TgmoZS0Kw98fOoAcGz8B9iDhdqB4Be4e=vDZaJZ5A-xMYBqA@mail.gmail.com
This commit is contained in:
Robert Haas
parent
c35ba141de
commit
ebfb814f7c
|
|
@ -570,7 +570,7 @@ LogStreamerMain(logstreamer_param *param)
|
|||
return 1;
|
||||
}
|
||||
|
||||
if (!stream.walmethod->finish())
|
||||
if (!stream.walmethod->ops->finish(stream.walmethod))
|
||||
{
|
||||
pg_log_error("could not finish writing WAL files: %m");
|
||||
#ifdef WIN32
|
||||
|
|
@ -581,11 +581,7 @@ LogStreamerMain(logstreamer_param *param)
|
|||
|
||||
PQfinish(param->bgconn);
|
||||
|
||||
if (format == 'p')
|
||||
FreeWalDirectoryMethod();
|
||||
else
|
||||
FreeWalTarMethod();
|
||||
pg_free(stream.walmethod);
|
||||
stream.walmethod->ops->free(stream.walmethod);
|
||||
|
||||
return 0;
|
||||
}
|
||||
|
|
|
|||
|
|
@ -658,7 +658,7 @@ StreamLog(void)
|
|||
|
||||
ReceiveXlogStream(conn, &stream);
|
||||
|
||||
if (!stream.walmethod->finish())
|
||||
if (!stream.walmethod->ops->finish(stream.walmethod))
|
||||
{
|
||||
pg_log_info("could not finish writing WAL files: %m");
|
||||
return;
|
||||
|
|
@ -667,9 +667,7 @@ StreamLog(void)
|
|||
PQfinish(conn);
|
||||
conn = NULL;
|
||||
|
||||
FreeWalDirectoryMethod();
|
||||
pg_free(stream.walmethod);
|
||||
pg_free(stream.sysidentifier);
|
||||
stream.walmethod->ops->free(stream.walmethod);
|
||||
}
|
||||
|
||||
/*
|
||||
|
|
|
|||
|
|
@ -59,18 +59,19 @@ mark_file_as_archived(StreamCtl *stream, const char *fname)
|
|||
snprintf(tmppath, sizeof(tmppath), "archive_status/%s.done",
|
||||
fname);
|
||||
|
||||
f = stream->walmethod->open_for_write(tmppath, NULL, 0);
|
||||
f = stream->walmethod->ops->open_for_write(stream->walmethod, tmppath,
|
||||
NULL, 0);
|
||||
if (f == NULL)
|
||||
{
|
||||
pg_log_error("could not create archive status file \"%s\": %s",
|
||||
tmppath, stream->walmethod->getlasterror());
|
||||
tmppath, GetLastWalMethodError(stream->walmethod));
|
||||
return false;
|
||||
}
|
||||
|
||||
if (stream->walmethod->close(f, CLOSE_NORMAL) != 0)
|
||||
if (stream->walmethod->ops->close(f, CLOSE_NORMAL) != 0)
|
||||
{
|
||||
pg_log_error("could not close archive status file \"%s\": %s",
|
||||
tmppath, stream->walmethod->getlasterror());
|
||||
tmppath, GetLastWalMethodError(stream->walmethod));
|
||||
return false;
|
||||
}
|
||||
|
||||
|
|
@ -98,8 +99,9 @@ open_walfile(StreamCtl *stream, XLogRecPtr startpoint)
|
|||
XLogFileName(walfile_name, stream->timeline, segno, WalSegSz);
|
||||
|
||||
/* Note that this considers the compression used if necessary */
|
||||
fn = stream->walmethod->get_file_name(walfile_name,
|
||||
stream->partial_suffix);
|
||||
fn = stream->walmethod->ops->get_file_name(stream->walmethod,
|
||||
walfile_name,
|
||||
stream->partial_suffix);
|
||||
|
||||
/*
|
||||
* When streaming to files, if an existing file exists we verify that it's
|
||||
|
|
@ -111,35 +113,35 @@ open_walfile(StreamCtl *stream, XLogRecPtr startpoint)
|
|||
* When streaming to tar, no file with this name will exist before, so we
|
||||
* never have to verify a size.
|
||||
*/
|
||||
if (stream->walmethod->compression_algorithm() == PG_COMPRESSION_NONE &&
|
||||
stream->walmethod->existsfile(fn))
|
||||
if (stream->walmethod->compression_algorithm == PG_COMPRESSION_NONE &&
|
||||
stream->walmethod->ops->existsfile(stream->walmethod, fn))
|
||||
{
|
||||
size = stream->walmethod->get_file_size(fn);
|
||||
size = stream->walmethod->ops->get_file_size(stream->walmethod, fn);
|
||||
if (size < 0)
|
||||
{
|
||||
pg_log_error("could not get size of write-ahead log file \"%s\": %s",
|
||||
fn, stream->walmethod->getlasterror());
|
||||
fn, GetLastWalMethodError(stream->walmethod));
|
||||
pg_free(fn);
|
||||
return false;
|
||||
}
|
||||
if (size == WalSegSz)
|
||||
{
|
||||
/* Already padded file. Open it for use */
|
||||
f = stream->walmethod->open_for_write(walfile_name, stream->partial_suffix, 0);
|
||||
f = stream->walmethod->ops->open_for_write(stream->walmethod, walfile_name, stream->partial_suffix, 0);
|
||||
if (f == NULL)
|
||||
{
|
||||
pg_log_error("could not open existing write-ahead log file \"%s\": %s",
|
||||
fn, stream->walmethod->getlasterror());
|
||||
fn, GetLastWalMethodError(stream->walmethod));
|
||||
pg_free(fn);
|
||||
return false;
|
||||
}
|
||||
|
||||
/* fsync file in case of a previous crash */
|
||||
if (stream->walmethod->sync(f) != 0)
|
||||
if (stream->walmethod->ops->sync(f) != 0)
|
||||
{
|
||||
pg_log_error("could not fsync existing write-ahead log file \"%s\": %s",
|
||||
fn, stream->walmethod->getlasterror());
|
||||
stream->walmethod->close(f, CLOSE_UNLINK);
|
||||
fn, GetLastWalMethodError(stream->walmethod));
|
||||
stream->walmethod->ops->close(f, CLOSE_UNLINK);
|
||||
exit(1);
|
||||
}
|
||||
|
||||
|
|
@ -164,12 +166,14 @@ open_walfile(StreamCtl *stream, XLogRecPtr startpoint)
|
|||
|
||||
/* No file existed, so create one */
|
||||
|
||||
f = stream->walmethod->open_for_write(walfile_name,
|
||||
stream->partial_suffix, WalSegSz);
|
||||
f = stream->walmethod->ops->open_for_write(stream->walmethod,
|
||||
walfile_name,
|
||||
stream->partial_suffix,
|
||||
WalSegSz);
|
||||
if (f == NULL)
|
||||
{
|
||||
pg_log_error("could not open write-ahead log file \"%s\": %s",
|
||||
fn, stream->walmethod->getlasterror());
|
||||
fn, GetLastWalMethodError(stream->walmethod));
|
||||
pg_free(fn);
|
||||
return false;
|
||||
}
|
||||
|
|
@ -199,28 +203,29 @@ close_walfile(StreamCtl *stream, XLogRecPtr pos)
|
|||
currpos = walfile->currpos;
|
||||
|
||||
/* Note that this considers the compression used if necessary */
|
||||
fn = stream->walmethod->get_file_name(walfile_name,
|
||||
stream->partial_suffix);
|
||||
fn = stream->walmethod->ops->get_file_name(stream->walmethod,
|
||||
walfile_name,
|
||||
stream->partial_suffix);
|
||||
|
||||
if (stream->partial_suffix)
|
||||
{
|
||||
if (currpos == WalSegSz)
|
||||
r = stream->walmethod->close(walfile, CLOSE_NORMAL);
|
||||
r = stream->walmethod->ops->close(walfile, CLOSE_NORMAL);
|
||||
else
|
||||
{
|
||||
pg_log_info("not renaming \"%s\", segment is not complete", fn);
|
||||
r = stream->walmethod->close(walfile, CLOSE_NO_RENAME);
|
||||
r = stream->walmethod->ops->close(walfile, CLOSE_NO_RENAME);
|
||||
}
|
||||
}
|
||||
else
|
||||
r = stream->walmethod->close(walfile, CLOSE_NORMAL);
|
||||
r = stream->walmethod->ops->close(walfile, CLOSE_NORMAL);
|
||||
|
||||
walfile = NULL;
|
||||
|
||||
if (r != 0)
|
||||
{
|
||||
pg_log_error("could not close file \"%s\": %s",
|
||||
fn, stream->walmethod->getlasterror());
|
||||
fn, GetLastWalMethodError(stream->walmethod));
|
||||
|
||||
pg_free(fn);
|
||||
return false;
|
||||
|
|
@ -263,7 +268,7 @@ existsTimeLineHistoryFile(StreamCtl *stream)
|
|||
|
||||
TLHistoryFileName(histfname, stream->timeline);
|
||||
|
||||
return stream->walmethod->existsfile(histfname);
|
||||
return stream->walmethod->ops->existsfile(stream->walmethod, histfname);
|
||||
}
|
||||
|
||||
static bool
|
||||
|
|
@ -285,31 +290,32 @@ writeTimeLineHistoryFile(StreamCtl *stream, char *filename, char *content)
|
|||
return false;
|
||||
}
|
||||
|
||||
f = stream->walmethod->open_for_write(histfname, ".tmp", 0);
|
||||
f = stream->walmethod->ops->open_for_write(stream->walmethod,
|
||||
histfname, ".tmp", 0);
|
||||
if (f == NULL)
|
||||
{
|
||||
pg_log_error("could not create timeline history file \"%s\": %s",
|
||||
histfname, stream->walmethod->getlasterror());
|
||||
histfname, GetLastWalMethodError(stream->walmethod));
|
||||
return false;
|
||||
}
|
||||
|
||||
if ((int) stream->walmethod->write(f, content, size) != size)
|
||||
if ((int) stream->walmethod->ops->write(f, content, size) != size)
|
||||
{
|
||||
pg_log_error("could not write timeline history file \"%s\": %s",
|
||||
histfname, stream->walmethod->getlasterror());
|
||||
histfname, GetLastWalMethodError(stream->walmethod));
|
||||
|
||||
/*
|
||||
* If we fail to make the file, delete it to release disk space
|
||||
*/
|
||||
stream->walmethod->close(f, CLOSE_UNLINK);
|
||||
stream->walmethod->ops->close(f, CLOSE_UNLINK);
|
||||
|
||||
return false;
|
||||
}
|
||||
|
||||
if (stream->walmethod->close(f, CLOSE_NORMAL) != 0)
|
||||
if (stream->walmethod->ops->close(f, CLOSE_NORMAL) != 0)
|
||||
{
|
||||
pg_log_error("could not close file \"%s\": %s",
|
||||
histfname, stream->walmethod->getlasterror());
|
||||
histfname, GetLastWalMethodError(stream->walmethod));
|
||||
return false;
|
||||
}
|
||||
|
||||
|
|
@ -678,9 +684,9 @@ ReceiveXlogStream(PGconn *conn, StreamCtl *stream)
|
|||
}
|
||||
|
||||
error:
|
||||
if (walfile != NULL && stream->walmethod->close(walfile, CLOSE_NO_RENAME) != 0)
|
||||
if (walfile != NULL && stream->walmethod->ops->close(walfile, CLOSE_NO_RENAME) != 0)
|
||||
pg_log_error("could not close file \"%s\": %s",
|
||||
walfile->pathname, stream->walmethod->getlasterror());
|
||||
walfile->pathname, GetLastWalMethodError(stream->walmethod));
|
||||
walfile = NULL;
|
||||
return false;
|
||||
}
|
||||
|
|
@ -765,9 +771,9 @@ HandleCopyStream(PGconn *conn, StreamCtl *stream,
|
|||
*/
|
||||
if (stream->synchronous && lastFlushPosition < blockpos && walfile != NULL)
|
||||
{
|
||||
if (stream->walmethod->sync(walfile) != 0)
|
||||
if (stream->walmethod->ops->sync(walfile) != 0)
|
||||
pg_fatal("could not fsync file \"%s\": %s",
|
||||
walfile->pathname, stream->walmethod->getlasterror());
|
||||
walfile->pathname, GetLastWalMethodError(stream->walmethod));
|
||||
lastFlushPosition = blockpos;
|
||||
|
||||
/*
|
||||
|
|
@ -1012,9 +1018,9 @@ ProcessKeepaliveMsg(PGconn *conn, StreamCtl *stream, char *copybuf, int len,
|
|||
* data has been successfully replicated or not, at the normal
|
||||
* shutdown of the server.
|
||||
*/
|
||||
if (stream->walmethod->sync(walfile) != 0)
|
||||
if (stream->walmethod->ops->sync(walfile) != 0)
|
||||
pg_fatal("could not fsync file \"%s\": %s",
|
||||
walfile->pathname, stream->walmethod->getlasterror());
|
||||
walfile->pathname, GetLastWalMethodError(stream->walmethod));
|
||||
lastFlushPosition = blockpos;
|
||||
}
|
||||
|
||||
|
|
@ -1115,12 +1121,13 @@ ProcessXLogDataMsg(PGconn *conn, StreamCtl *stream, char *copybuf, int len,
|
|||
}
|
||||
}
|
||||
|
||||
if (stream->walmethod->write(walfile, copybuf + hdr_len + bytes_written,
|
||||
bytes_to_write) != bytes_to_write)
|
||||
if (stream->walmethod->ops->write(walfile,
|
||||
copybuf + hdr_len + bytes_written,
|
||||
bytes_to_write) != bytes_to_write)
|
||||
{
|
||||
pg_log_error("could not write %d bytes to WAL file \"%s\": %s",
|
||||
bytes_to_write, walfile->pathname,
|
||||
stream->walmethod->getlasterror());
|
||||
GetLastWalMethodError(stream->walmethod));
|
||||
return false;
|
||||
}
|
||||
|
||||
|
|
|
|||
File diff suppressed because it is too large
Load Diff
|
|
@ -16,6 +16,7 @@ typedef struct WalWriteMethod WalWriteMethod;
|
|||
|
||||
typedef struct
|
||||
{
|
||||
WalWriteMethod *wwmethod;
|
||||
off_t currpos;
|
||||
char *pathname;
|
||||
/*
|
||||
|
|
@ -34,16 +35,9 @@ typedef enum
|
|||
} WalCloseMethod;
|
||||
|
||||
/*
|
||||
* A WalWriteMethod structure represents the different methods used
|
||||
* to write the streaming WAL as it's received.
|
||||
*
|
||||
* All methods that have a failure return indicator will set state
|
||||
* allowing the getlasterror() method to return a suitable message.
|
||||
* Commonly, errno is this state (or part of it); so callers must take
|
||||
* care not to clobber errno between a failed method call and use of
|
||||
* getlasterror() to retrieve the message.
|
||||
* Table of callbacks for a WalWriteMethod.
|
||||
*/
|
||||
struct WalWriteMethod
|
||||
typedef struct WalWriteMethodOps
|
||||
{
|
||||
/*
|
||||
* Open a target file. Returns Walfile, or NULL if open failed. If a temp
|
||||
|
|
@ -51,7 +45,7 @@ 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) (WalWriteMethod *wwmethod, const char *pathname, const char *temp_suffix, size_t pad_to_size);
|
||||
|
||||
/*
|
||||
* Close an open Walfile, using one or more methods for handling automatic
|
||||
|
|
@ -60,19 +54,16 @@ struct WalWriteMethod
|
|||
int (*close) (Walfile *f, WalCloseMethod method);
|
||||
|
||||
/* Check if a file exist */
|
||||
bool (*existsfile) (const char *pathname);
|
||||
bool (*existsfile) (WalWriteMethod *wwmethod, const char *pathname);
|
||||
|
||||
/* Return the size of a file, or -1 on failure. */
|
||||
ssize_t (*get_file_size) (const char *pathname);
|
||||
ssize_t (*get_file_size) (WalWriteMethod *wwmethod, const char *pathname);
|
||||
|
||||
/*
|
||||
* Return the name of the current file to work on in pg_malloc()'d string,
|
||||
* without the base directory. This is useful for logging.
|
||||
*/
|
||||
char *(*get_file_name) (const char *pathname, const char *temp_suffix);
|
||||
|
||||
/* Returns the compression method */
|
||||
pg_compress_algorithm (*compression_algorithm) (void);
|
||||
char *(*get_file_name) (WalWriteMethod *wwmethod, const char *pathname, const char *temp_suffix);
|
||||
|
||||
/*
|
||||
* Write count number of bytes to the file, and return the number of bytes
|
||||
|
|
@ -91,10 +82,37 @@ struct WalWriteMethod
|
|||
* close/write/sync of shared resources succeeded, otherwise returns false
|
||||
* (but the resources are still closed).
|
||||
*/
|
||||
bool (*finish) (void);
|
||||
bool (*finish) (WalWriteMethod *wwmethod);
|
||||
|
||||
/* Return a text for the last error in this Walfile */
|
||||
const char *(*getlasterror) (void);
|
||||
/*
|
||||
* Free subsidiary data associated with the WalWriteMethod, and the
|
||||
* WalWriteMethod itself.
|
||||
*/
|
||||
void (*free) (WalWriteMethod *wwmethod);
|
||||
} WalWriteMethodOps;
|
||||
|
||||
/*
|
||||
* A WalWriteMethod structure represents a way of writing streaming WAL as
|
||||
* it's received.
|
||||
*
|
||||
* All methods that have a failure return indicator will set lasterrstring
|
||||
* or lasterrno (the former takes precedence) so that the caller can signal
|
||||
* a suitable error.
|
||||
*/
|
||||
struct WalWriteMethod
|
||||
{
|
||||
const WalWriteMethodOps *ops;
|
||||
pg_compress_algorithm compression_algorithm;
|
||||
int compression_level;
|
||||
bool sync;
|
||||
const char *lasterrstring; /* if set, takes precedence over lasterrno */
|
||||
int lasterrno;
|
||||
/*
|
||||
* 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.
|
||||
*/
|
||||
};
|
||||
|
||||
/*
|
||||
|
|
@ -111,6 +129,4 @@ WalWriteMethod *CreateWalTarMethod(const char *tarbase,
|
|||
pg_compress_algorithm compression_algo,
|
||||
int compression, bool sync);
|
||||
|
||||
/* Cleanup routines for previously-created methods */
|
||||
void FreeWalDirectoryMethod(void);
|
||||
void FreeWalTarMethod(void);
|
||||
const char *GetLastWalMethodError(WalWriteMethod *wwmethod);
|
||||
|
|
|
|||
Loading…
Reference in New Issue