diff --git a/contrib/pg_buffercache/pg_buffercache_pages.cpp b/contrib/pg_buffercache/pg_buffercache_pages.cpp index 09f4a11d..214d1105 100755 --- a/contrib/pg_buffercache/pg_buffercache_pages.cpp +++ b/contrib/pg_buffercache/pg_buffercache_pages.cpp @@ -86,8 +86,8 @@ Datum pg_buffercache_pages(PG_FUNCTION_ARGS) fctx->tupdesc = BlessTupleDesc(tupledesc); /* Allocate g_instance.attr.attr_storage.NBuffers worth of BufferCachePagesRec records. */ - fctx->record = - (BufferCachePagesRec*)palloc(sizeof(BufferCachePagesRec) * g_instance.attr.attr_storage.NBuffers); + fctx->record = (BufferCachePagesRec *)palloc_huge(CurrentMemoryContext, + sizeof(BufferCachePagesRec) * g_instance.attr.attr_storage.NBuffers); /* Set max calls and remember the user function context. */ funcctx->max_calls = g_instance.attr.attr_storage.NBuffers; diff --git a/src/bin/pg_basebackup/pg_basebackup.cpp b/src/bin/pg_basebackup/pg_basebackup.cpp index 82a9cae5..489680db 100644 --- a/src/bin/pg_basebackup/pg_basebackup.cpp +++ b/src/bin/pg_basebackup/pg_basebackup.cpp @@ -708,6 +708,7 @@ static void ReceiveTarFile(PGconn *conn, PGresult *res, int rownum) } disconnect_and_exit(1); } + PQclear(res); while (true) { if (copybuf != NULL) { @@ -886,6 +887,7 @@ static void ReceiveAndUnpackTarFile(PGconn *conn, PGresult *res, int rownum) * Get the COPY data */ res = backup_get_result(conn); + PQclear(res); while (1) { int r; diff --git a/src/bin/pg_ctl/backup.cpp b/src/bin/pg_ctl/backup.cpp index 52268fd9..a9a91a03 100644 --- a/src/bin/pg_ctl/backup.cpp +++ b/src/bin/pg_ctl/backup.cpp @@ -132,6 +132,8 @@ static int replace_node_name(char* sSrc, const char* sMatchStr, const char* sRep static void show_full_build_process(const char* errmg); static void backup_dw_file(const char* target_dir); void get_xlog_location(char (&xlog_location)[MAXPGPATH]); +static void DeleteAlreadyDropedFile(const char* path, bool is_table_space); +static int DeleteUnusedFile(const char* path, unsigned int SegNo, unsigned int fileNode); /* * tblspaceDirectory is used for saving the table space directory created by @@ -1027,6 +1029,8 @@ static void BaseBackup(const char* dirname, uint32 term) char nodetablespacepath[MAXPGPATH] = {0}; char nodetablespaceparentpath[MAXPGPATH] = {0}; char escaped_label[MAXPGPATH] = {0}; + char basePath[MAXPGPATH] = {0}; + char tblspcPath[MAXPGPATH] = {0}; int i; char xlogstart[MAXFNAMELEN] = {0}; char xlogend[MAXFNAMELEN] = {0}; @@ -1436,6 +1440,13 @@ static void BaseBackup(const char* dirname, uint32 term) PQfinish(streamConn); streamConn = NULL; + /* fsync all data come from source */ + if (!no_need_fsync) { + show_full_build_process("starting fsync all files come from source."); + (void) fsync_pgdata(basedir); + show_full_build_process("finish fsync all files."); + } + /* delete dw file if exists, recreate it and write a page of zero */ backup_dw_file(dirname); show_full_build_process("build dummy dw file success"); @@ -1444,6 +1455,14 @@ static void BaseBackup(const char* dirname, uint32 term) RENAME_BUILD_FILE(buildstart_file, builddone_file); show_full_build_process("rename build status file success"); + + nRet = snprintf_s(basePath, MAXPGPATH, MAXPGPATH, "%s/base", dirname); + securec_check_ss_c(nRet, "\0", "\0"); + DeleteAlreadyDropedFile(basePath, false); + + nRet = snprintf_s(tblspcPath, MAXPGPATH, MAXPGPATH, "%s/pg_tblspc", dirname); + securec_check_ss_c(nRet, "\0", "\0"); + DeleteAlreadyDropedFile(tblspcPath, true); } /* @@ -1949,3 +1968,110 @@ void get_xlog_location(char (&xlog_location)[MAXPGPATH]) } xlog_location[MAXPGPATH - 1] = '\0'; } + +static void DeleteAlreadyDropedFile(const char* path, bool is_table_space) +{ + char* fileName = NULL; + char pathbuf[MAXPGPATH] = {0}; + unsigned int fileNode = 0; + unsigned int spaceNode = 0; + unsigned int SegNo = 0; + unsigned int dbNode = 0; + struct stat statbuf; + struct dirent *de = NULL; + int nmatch = 0; + int res = -1; + int rc = 0; + + DIR *dir = opendir(path); + + while ((de = readdir(dir)) != NULL) { + /* skip entries point current dir or parent dir */ + if (strcmp(de->d_name, ".") == 0 || strcmp(de->d_name, "..") == 0) + continue; + rc = snprintf_s(pathbuf, MAXPGPATH, MAXPGPATH - 1, "%s/%s", path, de->d_name); + securec_check_ss_c(rc, "\0", "\0"); + if (lstat(pathbuf, &statbuf) != 0) { + if (errno != ENOENT) { + pg_log(PG_WARNING, _("could not lstat file or directory : %s!\n"), de->d_name); + continue; + } + } + if (S_ISDIR(statbuf.st_mode)) { + DeleteAlreadyDropedFile(pathbuf, is_table_space); + } else if (S_ISREG(statbuf.st_mode)) { + if (is_table_space) { + if ((fileName = strstr(pathbuf, "pg_tblspc/")) != NULL) { + nmatch = sscanf_s(fileName, "pg_tblspc/%u/%*[^/]/%u/%u.%u", &spaceNode, + &dbNode, &fileNode, &SegNo); + if (nmatch == 4) { + res = DeleteUnusedFile(path, SegNo, fileNode); + if (res < 0) { + (void)closedir(dir); + disconnect_and_exit(1); + } + } + } + } else { + if ((fileName = strstr(pathbuf, "base/")) != NULL) { + nmatch = sscanf_s(fileName, "base/%u/%u.%u", &dbNode, &fileNode, &SegNo); + if (nmatch == 3) { + res = DeleteUnusedFile(path, SegNo, fileNode); + if (res < 0) { + (void)closedir(dir); + disconnect_and_exit(1); + } + } + } + } + } + } + (void)closedir(dir); +} + +static int DeleteUnusedFile(const char* path, unsigned int SegNo, unsigned int fileNode) +{ + char firstFileName[MAXPGPATH] = {0}; + char beforeFileName[MAXPGPATH] = {0}; + char currentFileName[MAXPGPATH] = {0}; + struct stat statbuf; + struct stat tmpStatBuf; + int rc = 0; + + rc = snprintf_s(currentFileName, MAXPGPATH, MAXPGPATH - 1, "%s/%u.%u", path, fileNode, SegNo); + securec_check_ss_c(rc, "\0", "\0"); + rc = snprintf_s(firstFileName, MAXPGPATH, MAXPGPATH - 1, "%s/%u", path, fileNode); + securec_check_ss_c(rc, "\0", "\0"); + if (lstat(firstFileName, &statbuf) != 0) { + if (errno != ENOENT) { + pg_log(PG_WARNING, _("could not lstat file: %s!\n"), firstFileName); + return -1; + } else { + while (SegNo >= 1) { + rc = snprintf_s(currentFileName, MAXPGPATH, MAXPGPATH - 1, "%s/%u.%u", path, fileNode, SegNo); + securec_check_ss_c(rc, "\0", "\0"); + if (lstat(currentFileName, &tmpStatBuf) == 0) { + pg_log(PG_DEBUG, _("the file %s should be unlink without origin file\n"), currentFileName); + unlink(currentFileName); + } + SegNo--; + } + return 0; + } + } + if (statbuf.st_size == 0) { + while (SegNo > 1) { + SegNo -= 1; + rc = snprintf_s(beforeFileName, MAXPGPATH, MAXPGPATH - 1, "%s/%u.%u", path, fileNode, SegNo); + securec_check_ss_c(rc, "\0", "\0"); + if (lstat(beforeFileName, &tmpStatBuf) != 0) { + if (errno == ENOENT) { + pg_log(PG_DEBUG, _("the file %s before file does not exist\n"), currentFileName); + unlink(currentFileName); + break; + } + } + } + } + return 0; +} diff --git a/src/bin/pg_ctl/backup.h b/src/bin/pg_ctl/backup.h index 101f5b2d..8cc0075d 100644 --- a/src/bin/pg_ctl/backup.h +++ b/src/bin/pg_ctl/backup.h @@ -9,6 +9,7 @@ extern int standby_connect_timeout; extern int standby_message_timeout; extern char* conn_str; +extern bool no_need_fsync; extern pid_t process_id; extern char* basedir; extern int bgpipe[2]; diff --git a/src/bin/pg_ctl/pg_build.cpp b/src/bin/pg_ctl/pg_build.cpp index 99b16388..f955f4e2 100644 --- a/src/bin/pg_ctl/pg_build.cpp +++ b/src/bin/pg_ctl/pg_build.cpp @@ -57,6 +57,8 @@ int g_replication_type = -1; #define RT_WITH_DUMMY_STANDBY 0 #define RT_WITH_MULTI_STANDBY 1 +static void walkdir(const char *path, int (*action) (const char *fname, bool isdir), bool process_symlinks); + int32 pg_atoi(const char* s, int size, int c) { long l; @@ -1452,3 +1454,178 @@ bool libpqRotateCbmFile(PGconn* connObj, XLogRecPtr lsn) return ec; } +/* + * Issue fsync recursively on PGDATA and all its contents. + * + * We fsync regular files and directories wherever they are, but we follow + * symlinks only for pg_wal (or pg_xlog) and immediately under pg_tblspc. + * Other symlinks are presumed to point at files we're not responsible for + * fsyncing, and might not have privileges to write at all. + * + */ +void fsync_pgdata(const char *pg_data) +{ + bool xlog_is_symlink = false; + char pg_xlog[MAXPGPATH] = {0}; + char pg_tblspc[MAXPGPATH] = {0}; + errno_t errorno = EOK; + + errorno = snprintf_s(pg_xlog, MAXPGPATH, MAXPGPATH - 1, "%s/pg_xlog", pg_data); + securec_check_ss_c(errorno, "\0", "\0"); + errorno = snprintf_s(pg_tblspc, MAXPGPATH, MAXPGPATH - 1, "%s/pg_tblspc", pg_data); + securec_check_ss_c(errorno, "\0", "\0"); + +#ifndef WIN32 + { + struct stat st; + + if (lstat(pg_xlog, &st) < 0) { + pg_log(PG_WARNING, _("could not stat file \"%s\": %m\n"), pg_xlog); + exit(1); + } + else if (S_ISLNK(st.st_mode)) + xlog_is_symlink = true; + } +#else + if (pgwin32_is_junction(pg_xlog)) + xlog_is_symlink = true; +#endif + + /* + * Now we do the fsync()s in the same order. + * + * The main call ignores symlinks, so in addition to specially processing + * pg_wal if it's a symlink, pg_tblspc has to be visited separately with + * process_symlinks = true. Note that if there are any plain directories + * in pg_tblspc, they'll get fsync'd twice. That's not an expected case + * so we don't worry about optimizing it. + */ + walkdir(pg_data, fsync_fname, false); + if (xlog_is_symlink) + walkdir(pg_xlog, fsync_fname, false); + walkdir(pg_tblspc, fsync_fname, true); +} + +/* + * walkdir: recursively walk a directory, applying the action to each + * regular file and directory (including the named directory itself). + * + * If process_symlinks is true, the action and recursion are also applied + * to regular files and directories that are pointed to by symlinks in the + * given directory; otherwise symlinks are ignored. Symlinks are always + * ignored in subdirectories, ie we intentionally don't pass down the + * process_symlinks flag to recursive calls. + * + * Errors are reported but not considered fatal. + * + * See also walkdir in fd.cpp, which is a backend version of this logic. + */ +static void walkdir(const char *path, int (*action) (const char *fname, bool isdir), bool process_symlinks) +{ + DIR *dir; + struct dirent *de = NULL; + errno_t errorno = EOK; + + dir = opendir(path); + if (dir == NULL) { + pg_log(PG_WARNING, _("could not open directory \"%s\": %m\n"), path); + return; + } + + while (errno = 0, (de = readdir(dir)) != NULL) { + char subpath[MAXPGPATH * 2] = {0}; + struct stat fst; + int sret; + + if (strcmp(de->d_name, ".") == 0 || strcmp(de->d_name, "..") == 0) + continue; + + if (strcmp(de->d_name, "pg_ctl.lock") == 0) { + continue; + } + errorno = snprintf_s(subpath, sizeof(subpath), sizeof(subpath) - 1, "%s/%s", path, de->d_name); + securec_check_ss_c(errorno, "\0", "\0"); + + if (process_symlinks) + sret = stat(subpath, &fst); + else + sret = lstat(subpath, &fst); + if (sret < 0) { + pg_log(PG_WARNING, _("could not stat file \"%s\": %m\n"), subpath); + continue; + } + + if (S_ISREG(fst.st_mode)) + (*action) (subpath, false); + else if (S_ISDIR(fst.st_mode)) + walkdir(subpath, action, false); + } + + if (errno) + pg_log(PG_WARNING, _("could not read directory \"%s\": %m\n"), path); + + (void)closedir(dir); + + /* + * It's important to fsync the destination directory itself as individual + * file fsyncs don't guarantee that the directory entry for the file is + * synced. Recent versions of ext4 have made the window much wider but + * it's been an issue for ext3 and other filesystems in the past. + */ + (*action) (path, true); +} + +/* + * fsync_fname -- Try to fsync a file or directory + * + * Ignores errors trying to open unreadable files, or trying to fsync + * directories on systems where that isn't allowed/required. All other errors + * are fatal. + */ +int fsync_fname(const char *fname, bool isdir) +{ + int fd = -1; + int flags; + int returncode; + + /* + * Some OSs require directories to be opened read-only whereas other + * systems don't allow us to fsync files opened read-only; so we need both + * cases here. Using O_RDWR will cause us to fail to fsync files that are + * not writable by our userid, but we assume that's OK. + */ + flags = PG_BINARY; + if (!isdir) + flags |= O_RDWR; + else + flags |= O_RDONLY; + + /* + * Open the file, silently ignoring errors about unreadable files (or + * unsupported operations, e.g. opening a directory under Windows), and + * logging others. + */ + fd = open(fname, flags, 0); + if (fd < 0) { + if (errno == EACCES || (isdir && errno == EISDIR)) + return 0; + pg_log(PG_WARNING, _("could not open file \"%s\": %m\n"), fname); + return -1; + } + + returncode = fsync(fd); + + /* + * Some OSes don't allow us to fsync directories at all, so we can ignore + * those errors. Anything else needs to be reported. + */ + if (returncode != 0 && !(isdir && (errno == EBADF || errno == EINVAL))) { + pg_log(PG_WARNING, _("could not fsync file \"%s\": %m\n"), fname); + (void) close(fd); + exit(EXIT_FAILURE); + } + + (void) close(fd); + return 0; +} + diff --git a/src/bin/pg_ctl/pg_build.h b/src/bin/pg_ctl/pg_build.h index e8770414..40a67319 100644 --- a/src/bin/pg_ctl/pg_build.h +++ b/src/bin/pg_ctl/pg_build.h @@ -60,5 +60,7 @@ extern char* pg_strdup(const char* in); extern void pg_free(void* ptr); extern void get_slot_name(char* slotname, size_t len); extern bool libpqRotateCbmFile(PGconn* connObj, XLogRecPtr lsn); +extern int fsync_fname(const char *fname, bool isdir); +extern void fsync_pgdata(const char *pg_data); #endif /* PG_BUILD_H */ diff --git a/src/bin/pg_ctl/pg_ctl.cpp b/src/bin/pg_ctl/pg_ctl.cpp index 4d7ff06e..49d0fc0f 100644 --- a/src/bin/pg_ctl/pg_ctl.cpp +++ b/src/bin/pg_ctl/pg_ctl.cpp @@ -184,6 +184,7 @@ char gaussdb_state_file[MAXPGPATH] = {0}; static char postport_lock_file[MAXPGPATH]; static PGconn* dbConn = NULL; +bool no_need_fsync = false; pid_t process_id = 0; const int g_length_stop_char = 2; @@ -3980,7 +3981,7 @@ static void do_incremental_build(uint32 term) /* Concate connection str to primary host for performing rewind. */ errorno = sprintf_s(connstrSource, sizeof(connstrSource), - "host=%s port=%s dbname=postgres application_name=gs_rewind connect_timeout=5", + "host=%s port=%s dbname=postgres application_name=gs_rewind connect_timeout=5 rw_timeout=600", (streamConn->pghost != NULL) ? streamConn->pghost : streamConn->pghostaddr, streamConn->pgport); securec_check_ss_c(errorno, "\0", "\0"); @@ -4637,6 +4638,7 @@ int main(int argc, char** argv) {"connect-string", required_argument, NULL, 'C'}, {"remove-backup", no_argument, NULL, 1}, {"action", required_argument, NULL, 'a'}, + {"no-fsync", no_argument, NULL, 3}, {NULL, 0, NULL, 0}}; int option_index; @@ -4923,6 +4925,9 @@ int main(int argc, char** argv) case 1: clear_backup_dir = true; break; + case 3: + no_need_fsync = true; + break; default: /* getopt_long already issued a suitable error message */ do_advice(); diff --git a/src/bin/pg_probackup/pg_probackup.cpp b/src/bin/pg_probackup/pg_probackup.cpp index 58445b25..37d28c2e 100644 --- a/src/bin/pg_probackup/pg_probackup.cpp +++ b/src/bin/pg_probackup/pg_probackup.cpp @@ -804,6 +804,8 @@ int main(int argc, char *argv[]) */ parse_backup_option_to_params(command, command_name); + pfree(command_name); + compress_init(); /* do actual operation */ diff --git a/src/bin/pg_probackup/pgut.cpp b/src/bin/pg_probackup/pgut.cpp index 502c0f8a..b513c97c 100644 --- a/src/bin/pg_probackup/pgut.cpp +++ b/src/bin/pg_probackup/pgut.cpp @@ -613,6 +613,7 @@ PGconn* pgut_connect_replication(const char *host, const char *port, const char **values; errno_t rc = EOK; char rwtimeoutStr[12] = {0}; + const char *malloc_port = NULL; if (interrupted && !in_cleanup) elog(ERROR, "interrupted"); @@ -659,6 +660,7 @@ PGconn* pgut_connect_replication(const char *host, const char *port, { keywords[i] = "port"; values[i] = inc_dbport(port); + malloc_port = values[i]; i++; } @@ -683,6 +685,10 @@ PGconn* pgut_connect_replication(const char *host, const char *port, { free(values); free(keywords); + if (malloc_port) + { + free((void *)malloc_port); + } return tmpconn; } @@ -700,6 +706,10 @@ PGconn* pgut_connect_replication(const char *host, const char *port, PQfinish(tmpconn); free(values); free(keywords); + if (malloc_port) + { + free((void *)malloc_port); + } return NULL; } } diff --git a/src/bin/pg_probackup/restore.cpp b/src/bin/pg_probackup/restore.cpp index f6b0ab11..6b0aa5b1 100644 --- a/src/bin/pg_probackup/restore.cpp +++ b/src/bin/pg_probackup/restore.cpp @@ -19,6 +19,8 @@ #include "thread.h" #include "common/fe_memutils.h" +#define RESTORE_ARRAY_LEN 100 + typedef struct { parray *pgdata_files; @@ -1379,7 +1381,15 @@ create_recovery_conf(time_t backup_id, /* construct restore_command */ if (pitr_requested) { + char *timestamp = NULL; + const char *oldtime = NULL; + timestamp = (char *)pg_malloc(RESTORE_ARRAY_LEN); + time2iso(timestamp, RESTORE_ARRAY_LEN, backup->end_time); + oldtime = rt->time_string; + rt->time_string = timestamp; construct_restore_cmd(fp, rt, restore_command_provided, target_immediate); + rt->time_string = oldtime; + free(timestamp); } if (fio_fflush(fp) != 0 || diff --git a/src/bin/pg_rewind/filemap.cpp b/src/bin/pg_rewind/filemap.cpp index 0b98a2f8..605bedca 100644 --- a/src/bin/pg_rewind/filemap.cpp +++ b/src/bin/pg_rewind/filemap.cpp @@ -63,7 +63,7 @@ const char *excludeFiles[] = { "postgresql.conf.bak", "postgresql.conf.old", "pg_ctl.lock", - "build_completed.start" + "build_completed.start", "backup_label", "client.crt", "client.key", diff --git a/src/common/backend/utils/adt/geo_ops.cpp b/src/common/backend/utils/adt/geo_ops.cpp index 148695e6..4bbe9f1b 100644 --- a/src/common/backend/utils/adt/geo_ops.cpp +++ b/src/common/backend/utils/adt/geo_ops.cpp @@ -83,9 +83,14 @@ static Point* lseg_interpt_internal(LSEG* l1, LSEG* l2); #define LDELIM_C '<' #define RDELIM_C '>' -/* Maximum number of characters printed by pair_encode() */ -/* ...+3+7 : 3 accounts for extra_float_digits max value */ -#define P_MAXLEN (2 * (DBL_DIG + 3 + 7) + 1) +/* Maximum number of characters printed by pair_encode(). + * The value range of float8 is -1.79E+308 ~ 1.79E+308. + * For point(-1.79E+308,-1.79E+308), + * (2 * (DBL_DIG + 3 + 7) + 1 + 1) : 3 accounts for extra_float_digits max value, + * 7 accounts for "-.E+308", first number 1 accounts for comma in the middle of numbers, + * last number 1 accounts for string terminator. + */ +#define P_MAXLEN (2 * (DBL_DIG + 3 + 7) + 1 + 1) /* * Geometric data types are composed of points. diff --git a/src/common/backend/utils/adt/pgstatfuncs.cpp b/src/common/backend/utils/adt/pgstatfuncs.cpp index a28f9bca..d43c175a 100644 --- a/src/common/backend/utils/adt/pgstatfuncs.cpp +++ b/src/common/backend/utils/adt/pgstatfuncs.cpp @@ -7663,8 +7663,8 @@ Datum pg_buffercache_pages(PG_FUNCTION_ARGS) fctx->tupdesc = BlessTupleDesc(tupledesc); /* Allocate g_instance.attr.attr_storage.NBuffers worth of BufferCachePagesRec records. */ - fctx->record = - (BufferCachePagesRec*)palloc(sizeof(BufferCachePagesRec) * g_instance.attr.attr_storage.NBuffers); + fctx->record = (BufferCachePagesRec *)palloc_huge(CurrentMemoryContext, + sizeof(BufferCachePagesRec) * g_instance.attr.attr_storage.NBuffers); /* Set max calls and remember the user function context. */ funcctx->max_calls = g_instance.attr.attr_storage.NBuffers; diff --git a/src/common/backend/utils/error/elog.cpp b/src/common/backend/utils/error/elog.cpp index bdff1e3c..aef980f0 100644 --- a/src/common/backend/utils/error/elog.cpp +++ b/src/common/backend/utils/error/elog.cpp @@ -514,7 +514,10 @@ void errfinish(int dummy, ...) if (edata->elevel >= u_sess->attr.attr_common.backtrace_min_messages) { StringInfoData buf; initStringInfo(&buf); + + HOLD_INTERRUPTS(); int ret = output_backtrace_to_log(&buf); + RESUME_INTERRUPTS(); if (0 == ret) { edata->backtrace_log = pstrdup(buf.data); diff --git a/src/common/backend/utils/time/combocid.cpp b/src/common/backend/utils/time/combocid.cpp index 53cdbdcf..5ee17084 100644 --- a/src/common/backend/utils/time/combocid.cpp +++ b/src/common/backend/utils/time/combocid.cpp @@ -256,7 +256,9 @@ static CommandId GetComboCommandId(CommandId cmin, CommandId cmax) u_sess->utils_cxt.comboHash = hash_create("Combo CIDs", CCID_HASH_SIZE, &hash_ctl, HASH_ELEM | HASH_FUNCTION | HASH_CONTEXT); + } + if (u_sess->utils_cxt.comboCids == NULL) { u_sess->utils_cxt.comboCids = (ComboCidKeyData*)MemoryContextAlloc( u_sess->top_transaction_mem_cxt, sizeof(ComboCidKeyData) * CCID_ARRAY_SIZE); u_sess->utils_cxt.sizeComboCids = CCID_ARRAY_SIZE; diff --git a/src/common/port/Makefile b/src/common/port/Makefile index c4181ed8..efe7741b 100644 --- a/src/common/port/Makefile +++ b/src/common/port/Makefile @@ -32,8 +32,8 @@ VERSION = 1 override CPPFLAGS := -I$(top_builddir)/src/common/port -DFRONTEND $(CPPFLAGS) $(CFLAGS_SSE42) LIBS += $(PTHREAD_LIBS) -override CPPFLAGS := $(filter-out -fPIE, $(CPPFLAGS)) -fPIC -override CFLAGS := $(filter-out -fPIE, $(CFLAGS)) -fPIC +override CPPFLAGS := $(filter-out -fPIE, $(CPPFLAGS)) -fPIC -fstack-protector-all +override CFLAGS := $(filter-out -fPIE, $(CFLAGS)) -fPIC -fstack-protector-all override CPPSources=$(shell find -name "*.cpp" ! -name "path.cpp" | sort) ifneq "$(MAKECMDGOALS)" "clean" diff --git a/src/gausskernel/cbb/instruments/ash/ash.cpp b/src/gausskernel/cbb/instruments/ash/ash.cpp index 6818f5df..50e34677 100644 --- a/src/gausskernel/cbb/instruments/ash/ash.cpp +++ b/src/gausskernel/cbb/instruments/ash/ash.cpp @@ -64,6 +64,7 @@ #define NUM_UNIQUE_SQL_PARTITIONS 64 #define UINT32_ACCESS_ONCE(var) ((uint32)(*((volatile uint32*)&(var)))) #define UNIQUE_SQL_MAX_LEN (g_instance.attr.attr_common.pgstat_track_activity_query_size + 1) +const int ATTR_NUM = 27; /* unique SQL max hash table size */ const int UNIQUE_SQL_MAX_HASH_SIZE = 1000; extern Datum hash_uint32(uint32 k); @@ -1144,34 +1145,37 @@ static void InitTupleAttr(FuncCallContext** funcctx) { MemoryContext oldcontext; TupleDesc tupdesc = NULL; + int i = 0; oldcontext = MemoryContextSwitchTo((*funcctx)->multi_call_memory_ctx); - tupdesc = CreateTemplateTupleDesc(26, false); - TupleDescInitEntry(tupdesc, (AttrNumber)1, "sampleid", INT8OID, -1, 0); - TupleDescInitEntry(tupdesc, (AttrNumber)2, "sample_time", TIMESTAMPTZOID, -1, 0); - TupleDescInitEntry(tupdesc, (AttrNumber)3, "need_flush_sample", BOOLOID, -1, 0); - TupleDescInitEntry(tupdesc, (AttrNumber)4, "databaseid", OIDOID, -1, 0); - TupleDescInitEntry(tupdesc, (AttrNumber)5, "thread_id", INT8OID, -1, 0); - TupleDescInitEntry(tupdesc, (AttrNumber)6, "sessionid", INT8OID, -1, 0); - TupleDescInitEntry(tupdesc, (AttrNumber)7, "start_time", TIMESTAMPTZOID, -1, 0); - TupleDescInitEntry(tupdesc, (AttrNumber)8, "event", TEXTOID, -1, 0); - TupleDescInitEntry(tupdesc, (AttrNumber)9, "lwtid", INT4OID, -1, 0); - TupleDescInitEntry(tupdesc, (AttrNumber)10, "psessionid", INT8OID, -1, 0); - TupleDescInitEntry(tupdesc, (AttrNumber)11, "tlevel", INT4OID, -1, 0); - TupleDescInitEntry(tupdesc, (AttrNumber)12, "smpid", INT4OID, -1, 0); - TupleDescInitEntry(tupdesc, (AttrNumber)13, "userid", OIDOID, -1, 0); - TupleDescInitEntry(tupdesc, (AttrNumber)14, "application_name", TEXTOID, -1, 0); - TupleDescInitEntry(tupdesc, (AttrNumber)15, "client_addr", INETOID, -1, 0); - TupleDescInitEntry(tupdesc, (AttrNumber)16, "client_hostname", TEXTOID, -1, 0); - TupleDescInitEntry(tupdesc, (AttrNumber)17, "client_port", INT4OID, -1, 0); - TupleDescInitEntry(tupdesc, (AttrNumber)18, "query_id", INT8OID, -1, 0); - TupleDescInitEntry(tupdesc, (AttrNumber)19, "unique_query_id", INT8OID, -1, 0); - TupleDescInitEntry(tupdesc, (AttrNumber)20, "user_id", OIDOID, -1, 0); - TupleDescInitEntry(tupdesc, (AttrNumber)21, "cn_id", INT4OID, -1, 0); - TupleDescInitEntry(tupdesc, (AttrNumber)22, "unique_query", TEXTOID, -1, 0); - TupleDescInitEntry(tupdesc, (AttrNumber)23, "locktag", TEXTOID, -1, 0); - TupleDescInitEntry(tupdesc, (AttrNumber)24, "lockmode", TEXTOID, -1, 0); - TupleDescInitEntry(tupdesc, (AttrNumber)25, "block_sessionid", INT8OID, -1, 0); - TupleDescInitEntry(tupdesc, (AttrNumber)26, "wait_status", TEXTOID, -1, 0); + tupdesc = CreateTemplateTupleDesc(ATTR_NUM, false); + TupleDescInitEntry(tupdesc, (AttrNumber)++i, "sampleid", INT8OID, -1, 0); + TupleDescInitEntry(tupdesc, (AttrNumber)++i, "sample_time", TIMESTAMPTZOID, -1, 0); + TupleDescInitEntry(tupdesc, (AttrNumber)++i, "need_flush_sample", BOOLOID, -1, 0); + TupleDescInitEntry(tupdesc, (AttrNumber)++i, "databaseid", OIDOID, -1, 0); + TupleDescInitEntry(tupdesc, (AttrNumber)++i, "thread_id", INT8OID, -1, 0); + TupleDescInitEntry(tupdesc, (AttrNumber)++i, "sessionid", INT8OID, -1, 0); + TupleDescInitEntry(tupdesc, (AttrNumber)++i, "start_time", TIMESTAMPTZOID, -1, 0); + TupleDescInitEntry(tupdesc, (AttrNumber)++i, "event", TEXTOID, -1, 0); + TupleDescInitEntry(tupdesc, (AttrNumber)++i, "lwtid", INT4OID, -1, 0); + TupleDescInitEntry(tupdesc, (AttrNumber)++i, "psessionid", INT8OID, -1, 0); + TupleDescInitEntry(tupdesc, (AttrNumber)++i, "tlevel", INT4OID, -1, 0); + TupleDescInitEntry(tupdesc, (AttrNumber)++i, "smpid", INT4OID, -1, 0); + TupleDescInitEntry(tupdesc, (AttrNumber)++i, "userid", OIDOID, -1, 0); + TupleDescInitEntry(tupdesc, (AttrNumber)++i, "application_name", TEXTOID, -1, 0); + TupleDescInitEntry(tupdesc, (AttrNumber)++i, "client_addr", INETOID, -1, 0); + TupleDescInitEntry(tupdesc, (AttrNumber)++i, "client_hostname", TEXTOID, -1, 0); + TupleDescInitEntry(tupdesc, (AttrNumber)++i, "client_port", INT4OID, -1, 0); + TupleDescInitEntry(tupdesc, (AttrNumber)++i, "query_id", INT8OID, -1, 0); + TupleDescInitEntry(tupdesc, (AttrNumber)++i, "unique_query_id", INT8OID, -1, 0); + TupleDescInitEntry(tupdesc, (AttrNumber)++i, "user_id", OIDOID, -1, 0); + TupleDescInitEntry(tupdesc, (AttrNumber)++i, "cn_id", INT4OID, -1, 0); + TupleDescInitEntry(tupdesc, (AttrNumber)++i, "unique_query", TEXTOID, -1, 0); + TupleDescInitEntry(tupdesc, (AttrNumber)++i, "locktag", TEXTOID, -1, 0); + TupleDescInitEntry(tupdesc, (AttrNumber)++i, "lockmode", TEXTOID, -1, 0); + TupleDescInitEntry(tupdesc, (AttrNumber)++i, "block_sessionid", INT8OID, -1, 0); + TupleDescInitEntry(tupdesc, (AttrNumber)++i, "wait_status", TEXTOID, -1, 0); + TupleDescInitEntry(tupdesc, (AttrNumber)++i, "global_sessionid", TEXTOID, -1, 0); + Assert(i == ATTR_NUM); (*funcctx)->tuple_desc = BlessTupleDesc(tupdesc); (*funcctx)->user_fctx = palloc0(sizeof(int)); (*funcctx)->max_calls = g_instance.stat_cxt.active_sess_hist_arrary->curr_index; @@ -1194,8 +1198,8 @@ Datum get_local_active_session(PG_FUNCTION_ARGS) if (funcctx->call_cntr < funcctx->max_calls) { /* for each row */ - Datum values[26]; - bool nulls[26] = {false}; + Datum values[ATTR_NUM]; + bool nulls[ATTR_NUM] = {false}; HeapTuple tuple = NULL; SessionHistEntry *beentry = NULL; errno_t rc = memset_s(values, sizeof(values), 0, sizeof(values)); @@ -1210,7 +1214,7 @@ Datum get_local_active_session(PG_FUNCTION_ARGS) GetTuple(values, Natts_gs_asp, nulls, Natts_gs_asp, beentry); } else { /* No permissions to view data about this session */ - for (uint32 i = 0; i < 26; i++) { + for (uint32 i = 0; i < ATTR_NUM; i++) { nulls[i] = true; } } diff --git a/src/gausskernel/cbb/instruments/percentile/percentile.cpp b/src/gausskernel/cbb/instruments/percentile/percentile.cpp index 2fa92fb8..13e56c2f 100644 --- a/src/gausskernel/cbb/instruments/percentile/percentile.cpp +++ b/src/gausskernel/cbb/instruments/percentile/percentile.cpp @@ -353,7 +353,7 @@ void PercentileSpace::SubPercentileMain(void) t_thrd.percentile_cxt.need_reset_timer = true; g_instance.stat_cxt.force_process = false; } - pg_usleep(SLEEP_INTERVAL); // CCN check if need force process percentile + pg_usleep(SLEEP_INTERVAL * 1000L); // CCN check if need force process percentile } /* end of loop */ } diff --git a/src/gausskernel/cbb/instruments/statement/instr_statement.cpp b/src/gausskernel/cbb/instruments/statement/instr_statement.cpp index 9fd435c2..5cf61aef 100644 --- a/src/gausskernel/cbb/instruments/statement/instr_statement.cpp +++ b/src/gausskernel/cbb/instruments/statement/instr_statement.cpp @@ -129,8 +129,8 @@ bool check_statement_stat_level(char** newval, void** extra, GucSource source) List *l = split_levels_into_list(*newval); if (list_length(l) != STATEMENT_SQL_KIND) { - list_free_deep(l); GUC_check_errdetail("attr num:%d is error,track_stmt_stat_level attr is 2", l->length); + list_free_deep(l); return false; } @@ -370,7 +370,9 @@ static HeapTuple GetStatementTuple(Relation rel, StatementStatContext* statement /* is slow sql */ values[i++] = BoolGetDatum( - (statementInfo->finish_time - statementInfo->start_time >= statementInfo->slow_query_threshold) ? true : false); + (statementInfo->finish_time - statementInfo->start_time >= statementInfo->slow_query_threshold && + statementInfo->slow_query_threshold >= 0) ? true : false); + return heap_form_tuple(RelationGetDescr(rel), values, nulls); } diff --git a/src/gausskernel/optimizer/path/indxpath.cpp b/src/gausskernel/optimizer/path/indxpath.cpp index afefa180..53aa5e03 100644 --- a/src/gausskernel/optimizer/path/indxpath.cpp +++ b/src/gausskernel/optimizer/path/indxpath.cpp @@ -1015,6 +1015,16 @@ static List* build_paths_for_OR( continue; } + /* + * Build paths with global indexes only for un-bounded partition tables. + * The partition bounded tables should be handled by partition iterator + * or local indexes. + */ + RangeTblEntry* rte = planner_rt_fetch(rel->relid, root); + if (index->isGlobal && rte && OidIsValid(rte->partitionOid)) { + continue; + } + /* * Ignore partial indexes that do not match the query. If a partial * index is marked predOK then we know it's OK. Otherwise, we have to diff --git a/src/gausskernel/process/postmaster/autovacuum.cpp b/src/gausskernel/process/postmaster/autovacuum.cpp index 7c72157b..faeca642 100644 --- a/src/gausskernel/process/postmaster/autovacuum.cpp +++ b/src/gausskernel/process/postmaster/autovacuum.cpp @@ -2514,8 +2514,21 @@ static void do_autovacuum(void) if (!worker->wi_sharedrel && worker->wi_dboid != u_sess->proc_cxt.MyDatabaseId) goto next_worker; - /* we can not identify it only by oid. */ - if (worker->wi_tableoid == relid && worker->wi_parentoid == parentid) { + /* + * we can not identify it only by oid. + * check the main table: + * 1. other worker handle the main table, need check the worker's tableoid not equal the relid; + * 2. other worker handle the part table, need check the worker's parentoid not equal the relid; + * check the part table: + * 1. other worker handle the main table, need check the worker's tableoid not equal the parentid; + * 2. other worker handle the part table, need check the worker's parentoid not equal the parentid; + */ + if (parentid == InvalidOid && (worker->wi_tableoid == relid || worker->wi_parentoid == relid)) { + AUTOVAC_LOG(LOG, "parentoid = %u, tableoid = %u is is on autovac, just skip it", parentid, relid); + skipit = true; + break; + } + if (parentid != InvalidOid && (worker->wi_tableoid == parentid || worker->wi_parentoid == parentid)) { AUTOVAC_LOG(LOG, "parentoid = %u, tableoid = %u is is on autovac, just skip it", parentid, relid); skipit = true; break; diff --git a/src/gausskernel/process/postmaster/bgwriter.cpp b/src/gausskernel/process/postmaster/bgwriter.cpp index f32be1d9..b50a2575 100644 --- a/src/gausskernel/process/postmaster/bgwriter.cpp +++ b/src/gausskernel/process/postmaster/bgwriter.cpp @@ -654,13 +654,9 @@ void candidate_buf_init(void) if (found_candidate_buf || found_candidate_fm) { Assert(found_candidate_buf && found_candidate_fm); } else { - errno_t rc; - rc = memset_s(g_instance.bgwriter_cxt.candidate_buffers, buffer_num * sizeof(Buffer), - -1, buffer_num * sizeof(Buffer)); - rc = memset_s(g_instance.bgwriter_cxt.candidate_free_map, buffer_num * sizeof(bool), - false, buffer_num * sizeof(bool)); - securec_check(rc, "", ""); - + MemsetHugeMem((char *)g_instance.bgwriter_cxt.candidate_buffers, buffer_num * sizeof(Buffer), -1); + MemsetHugeMem((char *)g_instance.bgwriter_cxt.candidate_free_map, buffer_num * sizeof(bool)); + if (g_instance.bgwriter_cxt.bgwriter_procs != NULL) { int thread_num = g_instance.bgwriter_cxt.bgwriter_num; int avg_num = g_instance.attr.attr_storage.NBuffers / thread_num; @@ -1121,7 +1117,9 @@ static void candidate_buf_push(int buf_id, int thread_id) uint32 list_size = bgwriter->cand_list_size; uint32 tail_loc; + pg_memory_barrier(); volatile uint64 head = pg_atomic_read_u64(&bgwriter->head); + pg_memory_barrier(); volatile uint64 tail = pg_atomic_read_u64(&bgwriter->tail); if (unlikely(tail - head >= list_size)) { @@ -1130,8 +1128,8 @@ static void candidate_buf_push(int buf_id, int thread_id) } tail_loc = tail % list_size; bgwriter->cand_buf_list[tail_loc] = buf_id; - pg_write_barrier(); (void)pg_atomic_fetch_add_u64(&bgwriter->tail, 1); + pg_memory_barrier(); } /** @@ -1146,17 +1144,19 @@ bool candidate_buf_pop(int *buf_id, int thread_id) uint32 head_loc; while (true) { + pg_memory_barrier(); uint64 head = pg_atomic_read_u64(&bgwriter->head); + pg_memory_barrier(); volatile uint64 tail = pg_atomic_read_u64(&bgwriter->tail); if (unlikely(head >= tail)) { return false; /* candidate list is empty */ } - pg_write_barrier(); head_loc = head % list_size; *buf_id = bgwriter->cand_buf_list[head_loc]; if (pg_atomic_compare_exchange_u64(&bgwriter->head, &head, head + 1)) { + pg_memory_barrier(); return true; } } @@ -1166,6 +1166,7 @@ static int64 get_thread_candidate_nums(int thread_id) { BgWriterProc *bgwriter = &g_instance.bgwriter_cxt.bgwriter_procs[thread_id]; volatile uint64 head = pg_atomic_read_u64(&bgwriter->head); + pg_memory_barrier(); volatile uint64 tail = pg_atomic_read_u64(&bgwriter->tail); int64 curr_cand_num = tail - head; Assert(curr_cand_num >= 0); diff --git a/src/gausskernel/process/postmaster/checkpointer.cpp b/src/gausskernel/process/postmaster/checkpointer.cpp index df23e504..1beea4aa 100644 --- a/src/gausskernel/process/postmaster/checkpointer.cpp +++ b/src/gausskernel/process/postmaster/checkpointer.cpp @@ -940,8 +940,7 @@ void CheckpointerShmemInit(void) * requests array; this is so that CompactCheckpointerRequestQueue * can assume that any pad bytes in the request structs are zeroes. */ - errno_t ret = memset_s(t_thrd.checkpoint_cxt.CheckpointerShmem, size, 0, size); - securec_check(ret, "\0", "\0"); + MemsetHugeMem((char*)t_thrd.checkpoint_cxt.CheckpointerShmem, size); SpinLockInit(&t_thrd.checkpoint_cxt.CheckpointerShmem->ckpt_lck); t_thrd.checkpoint_cxt.CheckpointerShmem->max_requests = g_instance.attr.attr_storage.NBuffers; } diff --git a/src/gausskernel/process/postmaster/pagewriter.cpp b/src/gausskernel/process/postmaster/pagewriter.cpp index d80fc146..7d4c2656 100644 --- a/src/gausskernel/process/postmaster/pagewriter.cpp +++ b/src/gausskernel/process/postmaster/pagewriter.cpp @@ -338,7 +338,7 @@ bool push_pending_flush_queue(Buffer buffer) actual_loc = new_tail_loc % g_instance.ckpt_cxt_ctl->dirty_page_queue_size; buf_desc->dirty_queue_loc = actual_loc; g_instance.ckpt_cxt_ctl->dirty_page_queue[actual_loc].buffer = buffer; - pg_write_barrier(); + pg_memory_barrier(); pg_atomic_write_u32(&g_instance.ckpt_cxt_ctl->dirty_page_queue[actual_loc].slot_state, (SLOT_VALID)); (void)pg_atomic_fetch_add_u32(&g_instance.ckpt_cxt_ctl->actual_dirty_page_num, 1); return true; @@ -463,7 +463,7 @@ try_get_buf: if (!(pg_atomic_read_u32(&slot->slot_state) & SLOT_VALID)) { break; } - pg_read_barrier(); + pg_memory_barrier(); buffer = slot->buffer; /* slot state is valid, buffer is invalid, the slot buffer set 0 when BufferAlloc or InvalidateBuffer */ if (BufferIsInvalid(buffer)) { @@ -1344,6 +1344,7 @@ static void ckpt_try_prune_dirty_page_queue() * the redo point will be wrong, because some page not flush to disk. */ (void)LWLockAcquire(g_instance.ckpt_cxt_ctl->prune_queue_lock, LW_EXCLUSIVE); + pg_memory_barrier(); if (last_invalid_slot > pg_atomic_read_u64(&g_instance.ckpt_cxt_ctl->full_ckpt_expected_flush_loc)) { pg_atomic_write_u64(&g_instance.ckpt_cxt_ctl->full_ckpt_expected_flush_loc, (last_invalid_slot + 1)); } diff --git a/src/gausskernel/process/postmaster/postmaster.cpp b/src/gausskernel/process/postmaster/postmaster.cpp index d58ac6da..cb1ed43f 100755 --- a/src/gausskernel/process/postmaster/postmaster.cpp +++ b/src/gausskernel/process/postmaster/postmaster.cpp @@ -4708,9 +4708,6 @@ static void ProcessDemoteRequest(void) if (g_instance.pid_cxt.DataReceiverPID != 0) signal_child(g_instance.pid_cxt.DataReceiverPID, SIGTERM); - if (g_instance.pid_cxt.HeartbeatPID != 0) - signal_child(g_instance.pid_cxt.HeartbeatPID, SIGTERM); - if (g_instance.pid_cxt.TwoPhaseCleanerPID != 0) signal_child(g_instance.pid_cxt.TwoPhaseCleanerPID, SIGTERM); diff --git a/src/gausskernel/process/stream/streamMain.cpp b/src/gausskernel/process/stream/streamMain.cpp index 85bd8247..23bbce55 100644 --- a/src/gausskernel/process/stream/streamMain.cpp +++ b/src/gausskernel/process/stream/streamMain.cpp @@ -86,7 +86,10 @@ int StreamMain() int curTryCounter; int* oldTryCounter = NULL; if (sigsetjmp(local_sigjmp_buf, 1) != 0) { - t_thrd.int_cxt.ignoreBackendSignal = false; + t_thrd.int_cxt.ignoreBackendSignal = false; + if (g_threadPoolControler) { + g_threadPoolControler->GetSessionCtrl()->releaseLockIfNecessary(); + } /* reset STP thread local valueables */ stp_reset_opt_values(); diff --git a/src/gausskernel/process/tcop/postgres.cpp b/src/gausskernel/process/tcop/postgres.cpp index 950774bc..8f82d03d 100644 --- a/src/gausskernel/process/tcop/postgres.cpp +++ b/src/gausskernel/process/tcop/postgres.cpp @@ -7317,7 +7317,10 @@ int PostgresMain(int argc, char* argv[], const char* dbname, const char* usernam int curTryCounter; int* oldTryCounter = NULL; if (sigsetjmp(local_sigjmp_buf, 1) != 0) { - t_thrd.int_cxt.ignoreBackendSignal = false; + t_thrd.int_cxt.ignoreBackendSignal = false; + if (g_threadPoolControler) { + g_threadPoolControler->GetSessionCtrl()->releaseLockIfNecessary(); + } gstrace_tryblock_exit(true, oldTryCounter); Assert(t_thrd.proc->dw_pos == -1); diff --git a/src/gausskernel/process/threadpool/knl_thread.cpp b/src/gausskernel/process/threadpool/knl_thread.cpp index 3405d333..411316e5 100644 --- a/src/gausskernel/process/threadpool/knl_thread.cpp +++ b/src/gausskernel/process/threadpool/knl_thread.cpp @@ -700,6 +700,8 @@ static void knl_t_sig_init(knl_t_sig_context* sig_cxt) { sig_cxt->signal_handle_cnt = 0; sig_cxt->gs_sigale_check_type = SIGNAL_CHECK_NONE; + sig_cxt->session_id = 0; + sig_cxt->cur_ctrl_index = 0; } static void knl_t_slot_init(knl_t_slot_context* slot_cxt) diff --git a/src/gausskernel/process/threadpool/threadpool_controler.cpp b/src/gausskernel/process/threadpool/threadpool_controler.cpp index 43802e05..4eb39463 100644 --- a/src/gausskernel/process/threadpool/threadpool_controler.cpp +++ b/src/gausskernel/process/threadpool/threadpool_controler.cpp @@ -558,11 +558,10 @@ void ThreadPoolControler::ConstrainThreadNum() { /* Thread pool size should not be larger than max_connections. */ if (MAX_THREAD_POOL_SIZE > g_instance.attr.attr_network.MaxConnections) { - m_maxPoolSize = g_instance.attr.attr_network.MaxConnections; ereport(LOG, (errcode(ERRCODE_OPERATE_INVALID_PARAM), - errmsg("Thread pool size %d should not be larger than max_connections %d, " - "so reduce thread pool size to max_connections", - m_threadNum, g_instance.attr.attr_network.MaxConnections))); + errmsg("Max thread pool size %d should not be larger than max_connections %d, " + "so reduce max thread pool size to max_connections", + MAX_THREAD_POOL_SIZE, g_instance.attr.attr_network.MaxConnections))); } m_maxPoolSize = Min(MAX_THREAD_POOL_SIZE, g_instance.attr.attr_network.MaxConnections); diff --git a/src/gausskernel/process/threadpool/threadpool_listener.cpp b/src/gausskernel/process/threadpool/threadpool_listener.cpp index 53cb8fdf..818fbcbf 100644 --- a/src/gausskernel/process/threadpool/threadpool_listener.cpp +++ b/src/gausskernel/process/threadpool/threadpool_listener.cpp @@ -229,7 +229,15 @@ void ThreadPoolListener::ReaperAllSession() (errmsg("No thread pool worker left while waiting for session close. " "This is a very rare case when all thread pool workers happen to" " encounter FATAL problems before session close."))); - ExitPostmaster(1); + abort(); + } + /* m_sessionCount should be sum of the list length of m_idleSessionList and m_readySessionList + and worker's attached session */ + pg_memory_barrier(); + if (m_idleSessionList->IsEmpty() && m_readySessionList->IsEmpty() && + m_group->m_workerNum - m_group->m_idleWorkerNum == 0) { + ereport(WARNING, (errmsg("SessionCount should be zero when no session in this group."))); + m_group->m_sessionCount = 0; } elem = m_idleSessionList->RemoveHead(); diff --git a/src/gausskernel/process/threadpool/threadpool_sessctl.cpp b/src/gausskernel/process/threadpool/threadpool_sessctl.cpp index 1e532a25..908b9115 100644 --- a/src/gausskernel/process/threadpool/threadpool_sessctl.cpp +++ b/src/gausskernel/process/threadpool/threadpool_sessctl.cpp @@ -194,7 +194,7 @@ void ThreadPoolSessControl::MarkAllSessionClose() alock.unLock(); } -void ThreadPoolSessControl::CheckPermissionForSendSignal(knl_session_context* sess) +void ThreadPoolSessControl::CheckPermissionForSendSignal(knl_session_context* sess, sig_atomic_t* lock) { /* User id is invalid only when sometimes dealing with cancel signal. Because that permission is ensured by random cancel key, so we don't have to check the permission again. */ @@ -204,6 +204,7 @@ void ThreadPoolSessControl::CheckPermissionForSendSignal(knl_session_context* se /* Only superuser , DB owner and user himself have the permission to send singal. */ if (!superuser() && !pg_database_ownercheck(sess->proc_cxt.MyDatabaseId, u_sess->misc_cxt.CurrentUserId)) { if (sess->proc_cxt.MyRoleId != GetUserId()) { + *lock = 0; ereport(ERROR, (errcode(ERRCODE_INSUFFICIENT_PRIVILEGE), (errmsg("must be system admin, db owner or have the same role to terminate other backend")))); @@ -211,6 +212,18 @@ void ThreadPoolSessControl::CheckPermissionForSendSignal(knl_session_context* se } } +void ThreadPoolSessControl::releaseLockIfNecessary() +{ + if (unlikely(t_thrd.sig_cxt.cur_ctrl_index != 0)) { + knl_sess_control* ctrl = &m_base[t_thrd.sig_cxt.cur_ctrl_index - m_maxReserveSessionCount]; + volatile sig_atomic_t plock = ctrl->lock; + if (plock != 0) { + plock = 0; + } + t_thrd.sig_cxt.cur_ctrl_index = 0; + } +} + int ThreadPoolSessControl::SendSignal(int ctrl_index, int signal) { Assert(signal != SIGHUP); @@ -221,6 +234,7 @@ int ThreadPoolSessControl::SendSignal(int ctrl_index, int signal) } knl_sess_control* ctrl = &m_base[ctrl_index - m_maxReserveSessionCount]; + t_thrd.sig_cxt.cur_ctrl_index = ctrl_index; volatile sig_atomic_t* plock = &ctrl->lock; sig_atomic_t val; do { @@ -232,13 +246,13 @@ int ThreadPoolSessControl::SendSignal(int ctrl_index, int signal) /* Session may be NULL when the session exits during the clean connection process. We do nothing if the session is NULL */ if (sess == NULL) { - /* restore the value */ + /* restore the value */ ctrl->lock = 0; status = ESRCH; break; } /* Check user permission, and we dont have user id for cancel request. */ - CheckPermissionForSendSignal(sess); + CheckPermissionForSendSignal(sess, (sig_atomic_t*)plock); if (sess->status == KNL_SESS_ATTACH) { t_thrd.sig_cxt.gs_sigale_check_type = SIGNAL_CHECK_SESS_KEY; t_thrd.sig_cxt.session_id = sess->session_id; @@ -265,6 +279,7 @@ int ThreadPoolSessControl::SendSignal(int ctrl_index, int signal) } pg_usleep(100); } while (true); + t_thrd.sig_cxt.cur_ctrl_index = 0; return status; } diff --git a/src/gausskernel/process/threadpool/threadpool_stream.cpp b/src/gausskernel/process/threadpool/threadpool_stream.cpp index 281c0ede..7afad7a3 100644 --- a/src/gausskernel/process/threadpool/threadpool_stream.cpp +++ b/src/gausskernel/process/threadpool/threadpool_stream.cpp @@ -130,6 +130,8 @@ void ThreadPoolStream::InitStream() SetStreamWorkerInfo(m_producer); ExtractProduerInfo(); + SetProcessingMode(InitProcessing); + /* Init GUC option for this session. */ InitializeGUCOptions(); /* Read in remaining GUC variables */ @@ -142,7 +144,9 @@ void ThreadPoolStream::InitStream() t_thrd.proc_cxt.PostInit->SetDatabaseAndUser( u_sess->stream_cxt.producer_obj->getDbName(), InvalidOid, u_sess->stream_cxt.producer_obj->getUserName()); t_thrd.proc_cxt.PostInit->InitStreamSession(); - + + SetProcessingMode(NormalProcessing); + repair_guc_variables(); RestoreStreamSyncParam(&m_producer->m_syncParam); diff --git a/src/gausskernel/process/threadpool/threadpool_worker.cpp b/src/gausskernel/process/threadpool/threadpool_worker.cpp index 001f052d..69ee93de 100644 --- a/src/gausskernel/process/threadpool/threadpool_worker.cpp +++ b/src/gausskernel/process/threadpool/threadpool_worker.cpp @@ -396,6 +396,13 @@ void ThreadPoolWorker::ShutDownIfNecessary() RestoreThreadVariable(); proc_exit(0); } + /* there is time window which the cancle signal has arrived but ignored by prevent signal called before, + * so we rebuild the signal status here in case that happens. */ + if (unlikely(m_currentSession != NULL && m_currentSession->status == KNL_SESS_CLOSE)) { + ereport(LOG, (errmodule(MOD_THREAD_POOL), + errmsg("Cancle signal has arrived but ignored by prevent signal called before, rebuild it."))); + t_thrd.int_cxt.ClientConnectionLost = true; + } } void ThreadPoolWorker::CleanThread() @@ -425,6 +432,8 @@ void ThreadPoolWorker::CleanThread() } InterruptPending = false; + t_thrd.int_cxt.QueryCancelPending = false; + t_thrd.int_cxt.PoolValidateCancelPending = false; t_thrd.libpq_cxt.PqSendStart = 0; t_thrd.libpq_cxt.PqSendPointer = 0; t_thrd.libpq_cxt.PqRecvLength = 0; @@ -494,8 +503,16 @@ bool ThreadPoolWorker::AttachSessionToThread() * Since thread pool worker may start earlier than startup finishing recovery, * init xlog access if necessary. */ - (void)RecoveryInProgress(); - + PG_TRY(); + { + (void)RecoveryInProgress(); + } + PG_CATCH(); + { + /* if init xlog has error, should throw fatal this thread */ + ereport(FATAL, (errmsg("init xlog failed, throw fatal for this thread"))); + } + PG_END_TRY(); #ifdef ENABLE_QUNIT set_qunit_case_number_hook(u_sess->utils_cxt.qunit_case_number, NULL); #endif @@ -660,6 +677,9 @@ static void init_session_share_memory() static bool InitSession(knl_session_context* session) { + /* non't send ereport to client now */ + t_thrd.postgres_cxt.whereToSendOutput = DestNone; + /* Switch context to Session context. */ AutoContextSwitch memSwitch(session->mcxt_group->GetMemCxtGroup(MEMORY_CONTEXT_DEFAULT)); @@ -681,6 +701,12 @@ static bool InitSession(knl_session_context* session) /* Read in remaining GUC variables */ read_nondefault_variables(); + + /* now safe to ereport to client */ + t_thrd.postgres_cxt.whereToSendOutput = DestRemote; + + /* now safe to ereport to client */ + t_thrd.postgres_cxt.whereToSendOutput = DestRemote; /* Init port and connection. */ if (!InitPort(session->proc_cxt.MyProcPort)) { diff --git a/src/gausskernel/runtime/executor/execMain.cpp b/src/gausskernel/runtime/executor/execMain.cpp index 07533abd..710822c6 100644 --- a/src/gausskernel/runtime/executor/execMain.cpp +++ b/src/gausskernel/runtime/executor/execMain.cpp @@ -170,31 +170,16 @@ static void report_iud_time(QueryDesc *query) if (OidIsValid(rid) == false || rid < FirstNormalObjectId) { continue; } - MemoryContext current_ctx = CurrentMemoryContext; + Relation rel = NULL; - PG_TRY(); - { - rel = heap_open(rid, AccessShareLock); - if (rel->rd_rel->relkind == RELKIND_RELATION) { - if (rel->rd_rel->relpersistence == RELPERSISTENCE_PERMANENT || - rel->rd_rel->relpersistence == RELPERSISTENCE_UNLOGGED) { - pgstat_report_data_changed(rid, STATFLG_RELATION, rel->rd_rel->relisshared); - } - } - heap_close(rel, AccessShareLock); - } - PG_CATCH(); - { - (void)MemoryContextSwitchTo(current_ctx); - ErrorData *edata = CopyErrorData(); - ereport(DEBUG1, (errmsg("Failed to send data changed time, cause: %s", edata->message))); - FlushErrorState(); - FreeErrorData(edata); - if (rel != NULL) { - heap_close(rel, AccessShareLock); + rel = heap_open(rid, AccessShareLock); + if (rel->rd_rel->relkind == RELKIND_RELATION) { + if (rel->rd_rel->relpersistence == RELPERSISTENCE_PERMANENT || + rel->rd_rel->relpersistence == RELPERSISTENCE_UNLOGGED) { + pgstat_report_data_changed(rid, STATFLG_RELATION, rel->rd_rel->relisshared); } } - PG_END_TRY(); + heap_close(rel, AccessShareLock); } } diff --git a/src/gausskernel/runtime/executor/nodeModifyTable.cpp b/src/gausskernel/runtime/executor/nodeModifyTable.cpp index 21189002..f92b61be 100644 --- a/src/gausskernel/runtime/executor/nodeModifyTable.cpp +++ b/src/gausskernel/runtime/executor/nodeModifyTable.cpp @@ -362,6 +362,7 @@ checktest: errmsg("unexpected self-updated tuple"))); break; case TM_Updated: + case TM_Deleted: ReleaseBuffer(buffer); if (IsolationUsesXactSnapshot()) { ereport(ERROR, (errcode(ERRCODE_T_R_SERIALIZATION_FAILURE), diff --git a/src/gausskernel/storage/access/transam/double_write.cpp b/src/gausskernel/storage/access/transam/double_write.cpp index c18c0234..f9945b2a 100644 --- a/src/gausskernel/storage/access/transam/double_write.cpp +++ b/src/gausskernel/storage/access/transam/double_write.cpp @@ -1806,15 +1806,14 @@ bool dw_verify_item(const dw_single_flush_item* item, uint16 dwn) if (item->dwn != dwn) { return false; } - - if (item->buf_tag.forkNum == InvalidForkNumber || item->buf_tag.blockNum == InvalidBlockNumber || + if (item->buf_tag.forkNum == InvalidForkNumber || item->buf_tag.blockNum == InvalidBlockNumber || item->buf_tag.rnode.relNode == InvalidOid) { - ereport(DEBUG1, + ereport(WARNING, (errmsg("dw recovery, find invalid item [page_idx %hu dwn %hu] skip this item," - "buf_tag[rel %u/%u/%u blk %u fork %d]", item->data_page_idx, item->dwn, - item->buf_tag.rnode.spcNode, item->buf_tag.rnode.dbNode, item->buf_tag.rnode.relNode, + "buf_tag[rel %u/%u/%u blk %u fork %d]", item->data_page_idx, item->dwn, + item->buf_tag.rnode.spcNode, item->buf_tag.rnode.dbNode, item->buf_tag.rnode.relNode, item->buf_tag.blockNum, item->buf_tag.forkNum))); - return false; + return false; } pg_crc32c crc; /* Contents are protected with a CRC */ diff --git a/src/gausskernel/storage/access/transam/extreme_rto/dispatcher.cpp b/src/gausskernel/storage/access/transam/extreme_rto/dispatcher.cpp index 462d6894..d7f21bc4 100644 --- a/src/gausskernel/storage/access/transam/extreme_rto/dispatcher.cpp +++ b/src/gausskernel/storage/access/transam/extreme_rto/dispatcher.cpp @@ -860,23 +860,10 @@ static bool DispatchRelMapRecord(XLogReaderState *record, List *expectedTLIs, Ti static bool DispatchXactRecord(XLogReaderState *record, List *expectedTLIs, TimestampTz recordXTime) { if (XactWillRemoveRelFiles(record)) { - /* for parallel performance */ - if (SUPPORT_FPAGE_DISPATCH) { - int nrels = 0; - ColFileNodeRel *xnodes = NULL; - XactGetRelFiles(record, &xnodes, &nrels); - for (int i = 0; ((i < nrels) && (xnodes != NULL)); ++i) { - ColFileNode node; - ColFileNodeRel *nodeRel = xnodes + i; - ColFileNodeCopy(&node, nodeRel); - uint32 id = GetSlotId(node.filenode, 0, 0, GetBatchCount()); - AddSlotToPLSet(id); - } - } else { - for (uint32 i = 0; i < g_dispatcher->pageLineNum; i++) { - AddSlotToPLSet(i); - } + for (uint32 i = 0; i < g_dispatcher->pageLineNum; i++) { + AddSlotToPLSet(i); } + /* sync with trxn thread */ /* trx execute drop action, pageworker forger invalid page, * pageworker first exe and update lastcomplateLSN diff --git a/src/gausskernel/storage/access/transam/parallel_recovery/dispatcher.cpp b/src/gausskernel/storage/access/transam/parallel_recovery/dispatcher.cpp index 935dec33..284ba410 100644 --- a/src/gausskernel/storage/access/transam/parallel_recovery/dispatcher.cpp +++ b/src/gausskernel/storage/access/transam/parallel_recovery/dispatcher.cpp @@ -722,23 +722,10 @@ static bool DispatchRelMapRecord(XLogReaderState *record, List *expectedTLIs, Ti static bool DispatchXactRecord(XLogReaderState *record, List *expectedTLIs, TimestampTz recordXTime) { if (XactWillRemoveRelFiles(record)) { - /* for parallel performance */ - if (SUPPORT_FPAGE_DISPATCH) { - int nrels = 0; - ColFileNodeRel *xnodes = NULL; - XactGetRelFiles(record, &xnodes, &nrels); - for (int i = 0; ((i < nrels) && (xnodes != NULL)); ++i) { - ColFileNode node; - ColFileNodeRel *nodeRel = xnodes + i; - ColFileNodeCopy(&node, nodeRel); - uint32 id = GetWorkerId(node.filenode, 0, 0); - AddWorkerToSet(id); - } - } else { - for (uint32 i = 0; i < g_dispatcher->pageWorkerCount; i++) { - AddWorkerToSet(i); - } + for (uint32 i = 0; i < g_dispatcher->pageWorkerCount; i++) { + AddWorkerToSet(i); } + /* sync with trxn thread */ /* trx execute drop action, pageworker forger invalid page, * pageworker first exe and update lastcomplateLSN diff --git a/src/gausskernel/storage/access/transam/xlog.cpp b/src/gausskernel/storage/access/transam/xlog.cpp index b3ea96ac..13e1c40d 100644 --- a/src/gausskernel/storage/access/transam/xlog.cpp +++ b/src/gausskernel/storage/access/transam/xlog.cpp @@ -539,6 +539,7 @@ static bool XLogArchiveIsBusy(const char *xlog); static bool XLogArchiveIsReady(const char *xlog); static void XLogArchiveCleanup(const char *xlog); static void readRecoveryCommandFile(void); +static XLogSegNo GetOldestXLOGSegNo(const char *workingPath); static void exitArchiveRecovery(TimeLineID endTLI, XLogSegNo endSegNo); static bool recoveryStopsHere(XLogReaderState *record, bool *includeThis); static void recoveryPausesHere(void); @@ -4875,11 +4876,11 @@ void CheckXLogRemoved(XLogSegNo segno, TimeLineID tli) * NB: the result can be out of date arbitrarily fast, the caller has to deal * with that. */ -XLogRecPtr XLogGetLastRemovedSegno(void) +XLogSegNo XLogGetLastRemovedSegno(void) { /* use volatile pointer to prevent code rearrangement */ volatile XLogCtlData *xlogctl = t_thrd.shemem_ptr_cxt.XLogCtl; - XLogRecPtr lastRemovedSegNo; + XLogSegNo lastRemovedSegNo; SpinLockAcquire(&xlogctl->info_lck); lastRemovedSegNo = xlogctl->lastRemovedSegNo; @@ -6784,10 +6785,7 @@ void XLOGShmemInit(void) */ allocptr = (char *)TYPEALIGN(XLOG_BLCKSZ, allocptr); t_thrd.shemem_ptr_cxt.XLogCtl->pages = allocptr; - errorno = memset_s(t_thrd.shemem_ptr_cxt.XLogCtl->pages, - (Size)XLOG_BLCKSZ * g_instance.attr.attr_storage.XLOGbuffers, 0, - (Size)XLOG_BLCKSZ * g_instance.attr.attr_storage.XLOGbuffers); - securec_check(errorno, "", ""); + MemsetHugeMem(t_thrd.shemem_ptr_cxt.XLogCtl->pages, (Size)XLOG_BLCKSZ * g_instance.attr.attr_storage.XLOGbuffers); if (BBOX_BLACKLIST_XLOG_BUFFER) { bbox_blacklist_add(XLOG_BUFFER, t_thrd.shemem_ptr_cxt.XLogCtl->pages, @@ -6803,6 +6801,9 @@ void XLOGShmemInit(void) t_thrd.shemem_ptr_cxt.XLogCtl->IsRecoveryDone = false; t_thrd.shemem_ptr_cxt.XLogCtl->SharedHotStandbyActive = false; t_thrd.shemem_ptr_cxt.XLogCtl->WalWriterSleeping = false; + if (!IsInitdb) { + t_thrd.shemem_ptr_cxt.XLogCtl->lastRemovedSegNo = GetOldestXLOGSegNo(t_thrd.proc_cxt.DataDir); + } #if (!defined __x86_64__) && (!defined __aarch64__) SpinLockInit(&t_thrd.shemem_ptr_cxt.XLogCtl->Insert.insertpos_lck); @@ -6821,6 +6822,46 @@ void XLOGShmemInit(void) } } +static XLogSegNo GetOldestXLOGSegNo(const char *workingPath) +{ +#define XLOGFILENAMELEN 24 + DIR *xlogDir = NULL; + struct dirent *dirEnt = NULL; + char xlogDirStr[MAXPGPATH] = {0}; + char oldestXLogFileName[MAXPGPATH] = {0}; + TimeLineID tli = 0; + uint32 xlogReadLogid = -1; + uint32 xlogReadLogSeg = -1; + XLogSegNo segno; + errno_t rc = EOK; + + rc = snprintf_s(xlogDirStr, MAXPGPATH, MAXPGPATH - 1, "%s/%s", workingPath, XLOGDIR); + securec_check_ss(rc, "", ""); + xlogDir = opendir(xlogDirStr); + if (!xlogDir) { + ereport(ERROR, (errcode_for_file_access(), errmsg("could not open xlog dir in GetOldestXLOGSegNo."))); + } + while ((dirEnt = readdir(xlogDir)) != NULL) { + if (strlen(dirEnt->d_name) == XLOGFILENAMELEN && + strspn(dirEnt->d_name, "0123456789ABCDEF") == XLOGFILENAMELEN) { + if (strlen(oldestXLogFileName) == 0 || strcmp(dirEnt->d_name, oldestXLogFileName) < 0) { + rc = strncpy_s(oldestXLogFileName, MAXPGPATH - 1, dirEnt->d_name, strlen(dirEnt->d_name) + 1); + securec_check_ss(rc, "", ""); + oldestXLogFileName[strlen(dirEnt->d_name)] = '\0'; + } + } + } + + (void)closedir(xlogDir); + + if (sscanf_s(oldestXLogFileName, "%08X%08X%08X", &tli, &xlogReadLogid, &xlogReadLogSeg) != 3) { + ereport(ERROR, (errcode_for_file_access(), errmsg("failed to translate name to xlog in GetOldestXLOGSegNo."))); + } + segno = (uint64)xlogReadLogid * XLogSegmentsPerXLogId + xlogReadLogSeg - 1; + + return segno; +} + static uint64 GetMACAddr(void) { macaddr mac; @@ -10895,7 +10936,7 @@ void CreateCheckPoint(int flags) */ g_instance.ckpt_cxt_ctl->full_ckpt_expected_flush_loc = get_dirty_page_queue_tail(); g_instance.ckpt_cxt_ctl->full_ckpt_redo_ptr = curInsert; - pg_write_barrier(); + pg_memory_barrier(); if (get_dirty_page_num() > 0) { g_instance.ckpt_cxt_ctl->flush_all_dirty_page = true; } @@ -11496,7 +11537,7 @@ void wait_all_dirty_page_flush(int flags, XLogRecPtr redo) if (ENABLE_INCRE_CKPT) { g_instance.ckpt_cxt_ctl->full_ckpt_redo_ptr = redo; g_instance.ckpt_cxt_ctl->full_ckpt_expected_flush_loc = get_dirty_page_queue_tail(); - pg_write_barrier(); + pg_memory_barrier(); if (get_dirty_page_num() > 0) { g_instance.ckpt_cxt_ctl->flush_all_dirty_page = true; ereport(LOG, (errmsg("CreateRestartPoint, need flush %ld pages.", get_dirty_page_num()))); @@ -11765,7 +11806,7 @@ bool CreateRestartPoint(int flags) if (ENABLE_INCRE_CKPT && doFullCkpt) { g_instance.ckpt_cxt_ctl->full_ckpt_redo_ptr = lastCheckPoint.redo; g_instance.ckpt_cxt_ctl->full_ckpt_expected_flush_loc = get_dirty_page_queue_tail(); - pg_write_barrier(); + pg_memory_barrier(); if (get_dirty_page_num() > 0) { g_instance.ckpt_cxt_ctl->flush_all_dirty_page = true; } @@ -11773,9 +11814,9 @@ bool CreateRestartPoint(int flags) } else if (ENABLE_INCRE_CKPT) { g_instance.ckpt_cxt_ctl->full_ckpt_redo_ptr = lastCheckPoint.redo; (void)LWLockAcquire(g_instance.ckpt_cxt_ctl->prune_queue_lock, LW_EXCLUSIVE); - g_instance.ckpt_cxt_ctl->full_ckpt_expected_flush_loc = get_loc_for_lsn(lastCheckPoint.redo); + g_instance.ckpt_cxt_ctl->full_ckpt_expected_flush_loc = get_loc_for_lsn(lastCheckPointRecPtr); + pg_memory_barrier(); LWLockRelease(g_instance.ckpt_cxt_ctl->prune_queue_lock); - pg_write_barrier(); uint64 head = pg_atomic_read_u64(&g_instance.ckpt_cxt_ctl->dirty_page_queue_head); int64 need_flush_num = g_instance.ckpt_cxt_ctl->full_ckpt_expected_flush_loc > head ? @@ -11800,11 +11841,11 @@ bool CreateRestartPoint(int flags) XLByteToSeg(t_thrd.shemem_ptr_cxt.ControlFile->checkPointCopy.redo, _logSegNo); if (ENABLE_INCRE_CKPT) { XLogRecPtr MinRecLSN = ckpt_get_min_rec_lsn(); - if (!XLogRecPtrIsInvalid(MinRecLSN) && XLByteLT(MinRecLSN, lastCheckPoint.redo)) { + if (!XLogRecPtrIsInvalid(MinRecLSN) && XLByteLT(MinRecLSN, lastCheckPointRecPtr)) { ereport(WARNING, (errmsg("current dirty page list head recLSN %08X/%08X smaller than redo lsn %08X/%08X", (uint32)(MinRecLSN >> XLOG_LSN_SWAP), (uint32)MinRecLSN, - (uint32)(lastCheckPoint.redo >> XLOG_LSN_SWAP), - (uint32)lastCheckPoint.redo))); + (uint32)(lastCheckPointRecPtr >> XLOG_LSN_SWAP), + (uint32)lastCheckPointRecPtr))); LWLockRelease(CheckpointLock); smgrsync_with_absorption(); gstrace_exit(GS_TRC_ID_CreateRestartPoint); @@ -17692,4 +17733,4 @@ extern bool IsValidArchiverStandby(WalSnd* walsnd) } else { return false; } -} \ No newline at end of file +} diff --git a/src/gausskernel/storage/buffer/buf_init.cpp b/src/gausskernel/storage/buffer/buf_init.cpp index ec407ffe..3f1249c3 100644 --- a/src/gausskernel/storage/buffer/buf_init.cpp +++ b/src/gausskernel/storage/buffer/buf_init.cpp @@ -28,16 +28,16 @@ const int PAGE_QUEUE_SLOT_MULTI_NBUFFERS = 5; -static void MemsetPageQueue(char *buffer, Size len) +void MemsetHugeMem(char *buffer, Size len, int num) { int rc; while (len > 0) { if (len < SECUREC_MEM_MAX_LEN) { - rc = memset_s(buffer, len, 0, len); + rc = memset_s(buffer, len, num, len); securec_check(rc, "", ""); return; } else { - rc = memset_s(buffer, SECUREC_MEM_MAX_LEN, 0, SECUREC_MEM_MAX_LEN); + rc = memset_s(buffer, SECUREC_MEM_MAX_LEN, num, SECUREC_MEM_MAX_LEN); securec_check(rc, "", ""); len -= SECUREC_MEM_MAX_LEN; buffer += SECUREC_MEM_MAX_LEN; @@ -133,7 +133,7 @@ void InitBufferPool(void) ereport(ERROR, (errmodule(MOD_INCRE_CKPT), errmsg("Memory allocation failed.\n"))); } - MemsetPageQueue((char*)g_instance.ckpt_cxt_ctl->dirty_page_queue, queue_mem_size); + MemsetHugeMem((char *)g_instance.ckpt_cxt_ctl->dirty_page_queue, queue_mem_size); (void)MemoryContextSwitchTo(oldcontext); } diff --git a/src/gausskernel/storage/buffer/bufmgr.cpp b/src/gausskernel/storage/buffer/bufmgr.cpp index c317fab8..4f0bc1c4 100644 --- a/src/gausskernel/storage/buffer/bufmgr.cpp +++ b/src/gausskernel/storage/buffer/bufmgr.cpp @@ -1764,8 +1764,16 @@ Buffer ReadBuffer_common_for_localbuf(RelFileNode rnode, char relpersistence, Fo * should return that the tuple does not exist without error reporting. */ else if (RecoveryInProgress()) { - if (blockNum >= smgrnblocks(smgr, forkNum)) + BlockNumber totalBlkNum = smgrnblocks_cached(smgr, forkNum); + + /* Update cached blocks */ + if (totalBlkNum == InvalidBlockNumber || blockNum >= totalBlkNum) { + totalBlkNum = smgrnblocks(smgr, forkNum); + } + + if (blockNum >= totalBlkNum) { return InvalidBuffer; + } } #endif @@ -3882,6 +3890,7 @@ void CheckPointBuffers(int flags, bool doFullCheckpoint) * dirty page num. */ for (;;) { + pg_memory_barrier(); if ((pg_atomic_read_u64(&g_instance.ckpt_cxt_ctl->dirty_page_queue_head) >= pg_atomic_read_u64(&g_instance.ckpt_cxt_ctl->full_ckpt_expected_flush_loc)) || get_dirty_page_num() == 0) { diff --git a/src/gausskernel/storage/buffer/freelist.cpp b/src/gausskernel/storage/buffer/freelist.cpp index 1cae5652..df4507a8 100644 --- a/src/gausskernel/storage/buffer/freelist.cpp +++ b/src/gausskernel/storage/buffer/freelist.cpp @@ -478,6 +478,9 @@ BufferAccessStrategy GetAccessStrategy(BufferAccessStrategyType btype) return NULL; /* keep compiler quiet */ } + /* If the shared buffers is too small, make sure ring size not equal zero. */ + ring_size = Max(ring_size, 4); + /* Make sure ring isn't an undue fraction of shared buffers */ if (btype != BAS_BULKWRITE && btype != BAS_BULKREAD) ring_size = Min(g_instance.attr.attr_storage.NBuffers / 8, ring_size); diff --git a/src/gausskernel/storage/ipc/ipc.cpp b/src/gausskernel/storage/ipc/ipc.cpp index 9608d648..1d4367fa 100644 --- a/src/gausskernel/storage/ipc/ipc.cpp +++ b/src/gausskernel/storage/ipc/ipc.cpp @@ -666,6 +666,7 @@ void PreventInterrupt() InterruptPending = false; t_thrd.int_cxt.ProcDiePending = false; t_thrd.int_cxt.QueryCancelPending = false; + t_thrd.int_cxt.PoolValidateCancelPending = false; /* And le's just make *sure* we'tre not interrupted ... */ t_thrd.int_cxt.ImmediateInterruptOK = false; t_thrd.int_cxt.CritSectionCount = 0; diff --git a/src/gausskernel/storage/lmgr/proc.cpp b/src/gausskernel/storage/lmgr/proc.cpp index be18fb02..0992111c 100644 --- a/src/gausskernel/storage/lmgr/proc.cpp +++ b/src/gausskernel/storage/lmgr/proc.cpp @@ -1112,7 +1112,9 @@ static void ProcKill(int code, Datum arg) (errcode(ERRCODE_DATA_CORRUPTED), errmsg("there remain unreleased locks when process exists."))); } #endif - + if (g_threadPoolControler) { + g_threadPoolControler->GetSessionCtrl()->releaseLockIfNecessary(); + } /* * Release any LW locks I am holding. There really shouldn't be any, but * it's cheap to check again before we cut the knees off the LWLock diff --git a/src/gausskernel/storage/remote/remote_adapter.cpp b/src/gausskernel/storage/remote/remote_adapter.cpp index c693b00c..4a066983 100644 --- a/src/gausskernel/storage/remote/remote_adapter.cpp +++ b/src/gausskernel/storage/remote/remote_adapter.cpp @@ -191,6 +191,10 @@ int StandbyReadPageforPrimary(uint32 spcnode, uint32 dbnode, uint32 relnode, int /* read page, if PageIsVerified failed will long jump to PG_CATCH() */ Buffer buf = ReadBufferForRemote(relfilenode, forknum, blocknum, RBM_FOR_REMOTE, NULL, &hit); + if (BufferIsInvalid(buf)) { + ereport(ERROR, (errmodule(MOD_REMOTE), errmsg("standby page buffer is invalid!"))); + return REMOTE_READ_BLCKSZ_NOT_SAME; + } LockBuffer(buf, BUFFER_LOCK_SHARE); Block block = BufferGetBlock(buf); diff --git a/src/include/access/xlog.h b/src/include/access/xlog.h index 845f691b..5c4c220f 100644 --- a/src/include/access/xlog.h +++ b/src/include/access/xlog.h @@ -326,7 +326,7 @@ extern void XLogSetReplicationSlotMaximumLSN(XLogRecPtr lsn); extern XLogRecPtr XLogGetReplicationSlotMaximumLSN(void); extern XLogRecPtr XLogGetReplicationSlotMinimumLSNByOther(void); -extern XLogRecPtr XLogGetLastRemovedSegno(void); +extern XLogSegNo XLogGetLastRemovedSegno(void); extern void xlog_redo(XLogReaderState* record); extern void xlog_desc(StringInfo buf, XLogReaderState* record); diff --git a/src/include/knl/knl_thread.h b/src/include/knl/knl_thread.h index 1b37b4c4..48ef97ba 100644 --- a/src/include/knl/knl_thread.h +++ b/src/include/knl/knl_thread.h @@ -1969,6 +1969,7 @@ typedef struct knl_t_sig_context { unsigned long signal_handle_cnt; GsSignalCheckType gs_sigale_check_type; uint64 session_id; + int cur_ctrl_index; } knl_t_sig_context; typedef struct knl_t_slot_context { diff --git a/src/include/nodes/relation.h b/src/include/nodes/relation.h index 313a369b..bb24a0e1 100644 --- a/src/include/nodes/relation.h +++ b/src/include/nodes/relation.h @@ -18,6 +18,7 @@ #include "lib/stringinfo.h" #include "nodes/params.h" #include "nodes/parsenodes.h" +#include "parser/parsetree.h" #include "storage/buf/block.h" #include "utils/partitionmap.h" #include "utils/partitionmap_gs.h" diff --git a/src/include/storage/buf/bufmgr.h b/src/include/storage/buf/bufmgr.h index 442129e9..bd1fad7e 100644 --- a/src/include/storage/buf/bufmgr.h +++ b/src/include/storage/buf/bufmgr.h @@ -306,5 +306,6 @@ extern Buffer ReadBuffer_common_for_localbuf(RelFileNode rnode, char relpersiste BlockNumber blockNum, ReadBufferMode mode, BufferAccessStrategy strategy, bool *hit); extern void DropRelFileNodeShareBuffers(RelFileNode node, ForkNumber forkNum, BlockNumber firstDelBlock); extern int GetThreadBufferLeakNum(void); +extern void MemsetHugeMem(char *buffer, Size len, int num = 0); #endif diff --git a/src/include/storage/smgr.h b/src/include/storage/smgr.h index 88b7f9ef..db2b73cd 100644 --- a/src/include/storage/smgr.h +++ b/src/include/storage/smgr.h @@ -60,7 +60,7 @@ typedef struct SMgrRelationData { BlockNumber smgr_targblock; /* current insertion target block */ BlockNumber smgr_fsm_nblocks; /* last known size of fsm fork */ BlockNumber smgr_vm_nblocks; /* last known size of vm fork */ - BlockNumber smgr_cached_nblocks; /* last known size of main fork*/ + BlockNumber smgr_cached_nblocks; /* last known size of main fork */ int smgr_bcmarry_size; BlockNumber* smgr_bcm_nblocks; /* last known size of bcm fork */ diff --git a/src/include/threadpool/threadpool_sessctl.h b/src/include/threadpool/threadpool_sessctl.h index 91f9d91d..0e8e13dc 100644 --- a/src/include/threadpool/threadpool_sessctl.h +++ b/src/include/threadpool/threadpool_sessctl.h @@ -60,13 +60,13 @@ public: void SigHupHandler(); void HandlePoolerReload(); void CheckSessionTimeout(); - void CheckPermissionForSendSignal(knl_session_context* sess); + void CheckPermissionForSendSignal(knl_session_context* sess, sig_atomic_t* lock); void getSessionMemoryDetail(Tuplestorestate* tupStore, TupleDesc tupDesc, knl_sess_control** sess); knl_session_context* GetSessionByIdx(int idx); int FindCtrlIdxBySessId(uint64 id); TransactionId ListAllSessionGttFrozenxids(int maxSize, ThreadId *pids, TransactionId *xids, int *n); bool IsActiveListEmpty(); - + void releaseLockIfNecessary(); inline int GetActiveSessionCount() { return m_activeSessionCount;