Mingxiaoming
/
openGauss-server
forked from huawei/openGauss-server
0
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

!1203 [同步PR] 同步master分支中多个已合并的PR 到2.0分支 

Merge pull request !1203 from cchen676/late20
This commit is contained in:
opengauss-bot 2021-08-07 08:15:15 +00:00 committed by Gitee
parent 0399758403 52f0789fb8
commit d6ed737bc7
51 changed files with 614 additions and 160 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
contrib/pg_buffercache/pg_buffercache_pages.cpp
View File

@ -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;

2
src/bin/pg_basebackup/pg_basebackup.cpp
View File

@ -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;

126
src/bin/pg_ctl/backup.cpp
View File

@ -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;
}

1
src/bin/pg_ctl/backup.h
View File

@ -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];

177
src/bin/pg_ctl/pg_build.cpp
View File

@ -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;
}

2
src/bin/pg_ctl/pg_build.h
View File

@ -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 */

7
src/bin/pg_ctl/pg_ctl.cpp
View File

@ -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();

2
src/bin/pg_probackup/pg_probackup.cpp
View File

@ -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 */

10
src/bin/pg_probackup/pgut.cpp
View File

@ -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;
}
}

10
src/bin/pg_probackup/restore.cpp
View File

@ -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 ||

2
src/bin/pg_rewind/filemap.cpp
View File

@ -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",

11
src/common/backend/utils/adt/geo_ops.cpp
View File

@ -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.

4
src/common/backend/utils/adt/pgstatfuncs.cpp
View File

@ -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;

3
src/common/backend/utils/error/elog.cpp
View File

@ -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);

2
src/common/backend/utils/time/combocid.cpp
View File

@ -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;

4
src/common/port/Makefile
View File

@ -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"

64
src/gausskernel/cbb/instruments/ash/ash.cpp
View File

@ -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;
}
}

2
src/gausskernel/cbb/instruments/percentile/percentile.cpp
View File

@ -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 */
}

6
src/gausskernel/cbb/instruments/statement/instr_statement.cpp
View File

@ -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);
}

10
src/gausskernel/optimizer/path/indxpath.cpp
View File

@ -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

17
src/gausskernel/process/postmaster/autovacuum.cpp
View File

@ -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;

19
src/gausskernel/process/postmaster/bgwriter.cpp
View File

@ -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);

3
src/gausskernel/process/postmaster/checkpointer.cpp
View File

@ -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;
}

5
src/gausskernel/process/postmaster/pagewriter.cpp
View File

@ -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));
}

3
src/gausskernel/process/postmaster/postmaster.cpp
View File

@ -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);

5
src/gausskernel/process/stream/streamMain.cpp
View File

@ -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();

5
src/gausskernel/process/tcop/postgres.cpp
View File

@ -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);

2
src/gausskernel/process/threadpool/knl_thread.cpp
View File

@ -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)

7
src/gausskernel/process/threadpool/threadpool_controler.cpp
View File

@ -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);

10
src/gausskernel/process/threadpool/threadpool_listener.cpp
View File

@ -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();

21
src/gausskernel/process/threadpool/threadpool_sessctl.cpp
View File

@ -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;
}

6
src/gausskernel/process/threadpool/threadpool_stream.cpp
View File

@ -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);

30
src/gausskernel/process/threadpool/threadpool_worker.cpp
View File

@ -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)) {

29
src/gausskernel/runtime/executor/execMain.cpp
View File

@ -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);
}
}

1
src/gausskernel/runtime/executor/nodeModifyTable.cpp
View File

@ -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),

11
src/gausskernel/storage/access/transam/double_write.cpp
View File

@ -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 */

19
src/gausskernel/storage/access/transam/extreme_rto/dispatcher.cpp
View File

@ -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

19
src/gausskernel/storage/access/transam/parallel_recovery/dispatcher.cpp
View File

@ -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

71
src/gausskernel/storage/access/transam/xlog.cpp
View File

@ -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;
}
}
}

8
src/gausskernel/storage/buffer/buf_init.cpp
View File

@ -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);
}

11
src/gausskernel/storage/buffer/bufmgr.cpp
View File

@ -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) {

3
src/gausskernel/storage/buffer/freelist.cpp
View File

@ -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);

1
src/gausskernel/storage/ipc/ipc.cpp
View File

@ -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;

4
src/gausskernel/storage/lmgr/proc.cpp
View File

@ -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

4
src/gausskernel/storage/remote/remote_adapter.cpp
View File

@ -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);

2
src/include/access/xlog.h
View File

@ -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);

1
src/include/knl/knl_thread.h
View File

@ -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 {

1
src/include/nodes/relation.h
View File

@ -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"

1
src/include/storage/buf/bufmgr.h
View File

@ -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

2
src/include/storage/smgr.h
View File

@ -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 */

4
src/include/threadpool/threadpool_sessctl.h
View File

@ -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;