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

!754 备机IO写放大问题合入 

Merge pull request !754 from 熊小军/1.1.0_tmp_new
This commit is contained in:
opengauss-bot 2021-03-03 16:35:41 +08:00 committed by Gitee
parent e130d900b3 72d4ada21a
commit 98f79ce2f5
12 changed files with 428 additions and 107 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -100,7 +100,6 @@ const int MILLISECOND_TO_MICROSECOND = 1000;
static void candidate_buf_push(int buf_id, int thread_id);
static int64 get_thread_candidate_nums(int thread_id);
static uint32 get_curr_candidate_nums(void);
static uint32 get_candidate_buf(bool *contain_hashbucket);
static uint32 get_buf_form_dirty_queue(bool *contain_hashbucket);
@ -732,10 +731,9 @@ static void incre_ckpt_bgwriter_kill(int code, Datum arg)
static int64 get_bgwriter_sleep_time()
{
pg_time_t now;
uint64 now;
int64 time_diff;
int thread_id = t_thrd.bgwriter_cxt.thread_id;
BgWriterProc *bgwriter = &g_instance.bgwriter_cxt.bgwriter_procs[thread_id];
/* If primary instance do full checkpoint and not the first bgwriter thread, can scan the dirty
* page queue, help the pagewriter thread finish the dirty page flush.
@ -744,17 +742,11 @@ static int64 get_bgwriter_sleep_time()
return 0;
}
now = (pg_time_t) time(NULL);
now = get_time_ms();
if (t_thrd.bgwriter_cxt.next_flush_time > now) {
time_diff = t_thrd.bgwriter_cxt.next_flush_time - now;
} else {
time_diff = 0;
if (now - t_thrd.bgwriter_cxt.next_flush_time > u_sess->attr.attr_storage.BgWriterDelay * 3) {
ereport(WARNING, (errmodule(MOD_INCRE_BG),
errmsg("bgwriter took %ld ms to flush %d pages, please check the max_io_capacity and bgwriter_delay",
u_sess->attr.attr_storage.BgWriterDelay + now - t_thrd.bgwriter_cxt.next_flush_time,
bgwriter->thread_last_flush)));
}
}
return time_diff;
@ -771,6 +763,7 @@ void incre_ckpt_background_writer_main(void)
WritebackContext wb_context;
int thread_id = t_thrd.bgwriter_cxt.thread_id;
BgWriterProc *bgwriter = &g_instance.bgwriter_cxt.bgwriter_procs[thread_id];
uint64 now;
t_thrd.role = BGWRITER;
@ -821,7 +814,7 @@ void incre_ckpt_background_writer_main(void)
t_thrd.xlog_cxt.ThisTimeLineID = GetRecoveryTargetTLI();
}
pg_time_t now = (pg_time_t) time(NULL);
now = get_time_ms();
t_thrd.bgwriter_cxt.next_flush_time = now + u_sess->attr.attr_storage.BgWriterDelay;
pgstat_report_appname("IncrBgWriter");
@ -871,9 +864,12 @@ void incre_ckpt_background_writer_main(void)
pgstat_report_activity(STATE_RUNNING, NULL);
now = (pg_time_t) time(NULL);
now = get_time_ms();
t_thrd.bgwriter_cxt.next_flush_time = now + u_sess->attr.attr_storage.BgWriterDelay;
if (get_curr_candidate_nums() == (uint32)g_instance.attr.attr_storage.NBuffers) {
continue;
}
/*
* When the primary instance do full checkpoint, the first thread remain scan the
* buffer pool to maintain the candidate buffer list, other threads scan the dirty
@ -892,7 +888,7 @@ void incre_ckpt_background_writer_main(void)
}
incre_ckpt_bgwriter_flush_page_batch(wb_context, need_flush_num, contain_hashbucket);
}
}
}
int get_bgwriter_thread_id(void)
@ -943,6 +939,11 @@ static uint32 get_bgwriter_flush_num()
high_water_mark = buffer_num * (percent_target + GAP_PERCENT);
cur_candidate_num = get_curr_candidate_nums();
/* If the slots are sufficient, the standby DN does not need to flush too many pages. */
if (RecoveryInProgress() && cur_candidate_num >= total_target / 2) {
max_io = max_io / 2;
}
/* max_io need greater than one batch flush num, and need less than the dirty list size */
max_io = MAX(max_io / g_instance.bgwriter_cxt.bgwriter_num, DW_DIRTY_PAGE_MAX_FOR_NOHBK);
max_io = MIN(max_io, dirty_list_size);
@ -985,9 +986,6 @@ static uint32 get_candidate_buf(bool *contain_hashbucket)
int start = MAX(bgwriter->buf_id_start, bgwriter->next_scan_loc);
int end = bgwriter->buf_id_start + bgwriter->cand_list_size;
if (RecoveryInProgress()) {
end = bgwriter->buf_id_start + bgwriter->cand_list_size * u_sess->attr.attr_storage.shared_buffers_fraction;
}
end = MIN(start + batch_scan_num, end);
for (int buf_id = start; buf_id < end; buf_id++) {
@ -1175,7 +1173,7 @@ static int64 get_thread_candidate_nums(int thread_id)
/**
* @Description: Return a rough estimate of the current number of buffers in the candidate list.
*/
static uint32 get_curr_candidate_nums(void)
uint32 get_curr_candidate_nums(void)
{
uint32 currCandidates = 0;
for (int i = 0; i < g_instance.bgwriter_cxt.bgwriter_num; i++) {

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

@ -214,6 +214,16 @@ const incre_ckpt_view_col g_ckpt_view_col[INCRE_CKPT_VIEW_COL_NUM] = {{"node_nam
{"ckpt_predicate_flush_num", INT8OID, ckpt_view_get_predicate_flush_num},
{"ckpt_twophase_flush_num", INT8OID, ckpt_view_get_twophase_flush_num}};
uint64 get_time_ms()
{
struct timeval tv;
uint64 time_ms;
(void)gettimeofday(&tv, NULL);
time_ms = (int64)tv.tv_sec * 1000 + (int64)tv.tv_usec / 1000;
return time_ms;
}
bool IsPagewriterProcess(void)
{
return (t_thrd.role == PAGEWRITER_THREAD);
@ -416,7 +426,7 @@ static uint32 ckpt_get_expected_flush_num()
* @out Offset to the new head
* @return Actual number of dirty pages need to flush
*/
static uint32 ckpt_qsort_dirty_page_for_flush(uint32 expected_flush_num, bool *contain_hashbucket)
static uint32 ckpt_qsort_dirty_page_for_flush(bool *contain_hashbucket)
{
uint32 num_to_flush = 0;
bool retry = false;
@ -498,12 +508,6 @@ try_get_buf:
goto try_get_buf;
}
qsort(g_instance.ckpt_cxt_ctl->CkptBufferIds, num_to_flush, sizeof(CkptSortItem), ckpt_buforder_comparator);
if (u_sess->attr.attr_storage.log_pagewriter) {
ereport(LOG,
(errmodule(MOD_INCRE_CKPT),
errmsg("expected_flush_num is %u, requested_flush_num is %u",
expected_flush_num, num_to_flush)));
}
return num_to_flush;
}
@ -602,7 +606,7 @@ static void ckpt_move_queue_head_after_flush()
/* We flushed some buffers, so update the statistics */
if (actual_flushed > 0) {
g_instance.ckpt_cxt_ctl->page_writer_actual_flush += actual_flushed;
g_instance.ckpt_cxt_ctl->page_writer_last_flush = actual_flushed;
g_instance.ckpt_cxt_ctl->page_writer_last_flush += actual_flushed;
}
if (u_sess->attr.attr_storage.log_pagewriter) {
@ -622,7 +626,7 @@ static void ckpt_pagewriter_main_thread_flush_dirty_page()
WritebackContext wb_context;
uint32 requested_flush_num;
int thread_id = t_thrd.pagewriter_cxt.pagewriter_id;
uint32 expected_flush_num;
int32 expected_flush_num;
bool contain_hashbucket = false;
WritebackContextInit(&wb_context, &t_thrd.pagewriter_cxt.page_writer_after);
@ -633,8 +637,10 @@ static void ckpt_pagewriter_main_thread_flush_dirty_page()
return;
}
g_instance.ckpt_cxt_ctl->page_writer_last_flush = 0;
while (expected_flush_num > 0) {
requested_flush_num = ckpt_qsort_dirty_page_for_flush(expected_flush_num, &contain_hashbucket);
requested_flush_num = ckpt_qsort_dirty_page_for_flush(&contain_hashbucket);
if (SECUREC_UNLIKELY(requested_flush_num == 0)) {
break;
@ -659,39 +665,78 @@ static void ckpt_pagewriter_main_thread_flush_dirty_page()
* If request flush num less than the batch max, break this loop,
* It indicates that there are not many dirty pages.
*/
if (requested_flush_num < GET_DW_DIRTY_PAGE_MAX(contain_hashbucket) && !FULL_CKPT) {
if (expected_flush_num < (int32)GET_DW_DIRTY_PAGE_MAX(contain_hashbucket) && !FULL_CKPT) {
break;
}
}
return;
}
static int64 get_pagewriter_sleep_time()
{
pg_time_t now;
uint64 now;
int64 time_diff;
if (FULL_CKPT) {
return 0;
}
now = (pg_time_t) time(NULL);
now = get_time_ms();
if (t_thrd.pagewriter_cxt.next_flush_time > now) {
time_diff = MAX(t_thrd.pagewriter_cxt.next_flush_time - now, 1);
} else {
time_diff = 0;
if (now - t_thrd.pagewriter_cxt.next_flush_time > u_sess->attr.attr_storage.pageWriterSleep * 3) {
ereport(WARNING, (errmodule(MOD_INCRE_BG),
errmsg("pagewriter took %ld ms to flush %u pages, "
"please check the max_io_capacity and pagewriter_sleep",
u_sess->attr.attr_storage.pageWriterSleep + now - t_thrd.pagewriter_cxt.next_flush_time,
g_instance.ckpt_cxt_ctl->page_writer_last_flush)));
}
}
return time_diff;
}
static uint32 get_page_num_for_lsn(XLogRecPtr target_lsn)
uint32 get_loc_for_lsn(XLogRecPtr target_lsn)
{
uint64 last_loc = 0;
XLogRecPtr page_rec_lsn = InvalidXLogRecPtr;
uint64 queue_loc = pg_atomic_read_u64(&g_instance.ckpt_cxt_ctl->dirty_page_queue_head);
if (get_dirty_page_num() == 0) {
return get_dirty_page_queue_tail();
}
while (queue_loc < get_dirty_page_queue_tail()) {
Buffer buffer;
BufferDesc *buf_desc = NULL;
uint64 temp_loc = queue_loc % g_instance.ckpt_cxt_ctl->dirty_page_queue_size;
volatile DirtyPageQueueSlot *slot = &g_instance.ckpt_cxt_ctl->dirty_page_queue[temp_loc];
/* slot location is pre-occupied, but the buffer not set finish, need wait and retry. */
if (!(pg_atomic_read_u32(&slot->slot_state) & SLOT_VALID)) {
pg_usleep(1);
queue_loc = pg_atomic_read_u64(&g_instance.ckpt_cxt_ctl->dirty_page_queue_head);
continue;
}
queue_loc++;
pg_memory_barrier();
buffer = slot->buffer;
/* slot state is vaild, buffer is invalid, the slot buffer set 0 when BufferAlloc or InvalidateBuffer */
if (BufferIsInvalid(buffer)) {
continue;
}
buf_desc = GetBufferDescriptor(buffer - 1);
page_rec_lsn = pg_atomic_read_u64(&buf_desc->rec_lsn);
if (!BufferIsInvalid(slot->buffer) && XLByteLE(target_lsn, page_rec_lsn)) {
last_loc = queue_loc - 1;
break;
}
}
if (last_loc == 0) {
return get_dirty_page_queue_tail();
}
return last_loc;
}
static uint32 get_page_num_for_lsn(XLogRecPtr target_lsn, uint32 max_num)
{
uint32 i;
uint32 num_for_lsn = 0;
@ -717,10 +762,13 @@ static uint32 get_page_num_for_lsn(XLogRecPtr target_lsn)
}
buf_desc = GetBufferDescriptor(buffer - 1);
page_rec_lsn = pg_atomic_read_u64(&buf_desc->rec_lsn);
if (XLByteLT(target_lsn, page_rec_lsn)) {
if (!BufferIsInvalid(slot->buffer) && XLByteLE(target_lsn, page_rec_lsn)) {
break;
}
num_for_lsn++;
if (num_for_lsn >= max_num) {
break;
}
}
return num_for_lsn;
}
@ -746,7 +794,7 @@ uint32 calculate_thread_max_flush_num(bool is_pagewriter)
}
const int AVG_CALCULATE_NUM = 30;
const int LSN_SCAN_FASTOR = 3;
const float HIGH_WATER = 0.75;
static uint32 calculate_pagewriter_flush_num()
{
static XLogRecPtr prev_lsn = InvalidXLogRecPtr;
@ -754,18 +802,22 @@ static uint32 calculate_pagewriter_flush_num()
static pg_time_t prev_time = 0;
static int64 total_flush_num = 0;
static uint32 avg_flush_num = 0;
static uint32 prev_lsn_num = 0;
static int counter = 0;
XLogRecPtr target_lsn;
XLogRecPtr cur_lsn;
XLogRecPtr min_lsn;
uint32 flush_num = 0;
pg_time_t now;
double time_diff;
uint64 now;
int64 time_diff;
float dirty_page_pct;
float dirty_slot_pct;
uint32 num_for_dirty;
uint32 num_for_lsn;
uint32 min_io = DW_DIRTY_PAGE_MAX_FOR_NOHBK;
uint32 max_io = calculate_thread_max_flush_num(true);
uint32 num_for_lsn_max;
float dirty_percent;
/* primary get the xlog insert loc, standby get the replay loc */
if (RecoveryInProgress()) {
@ -776,16 +828,13 @@ static uint32 calculate_pagewriter_flush_num()
if (XLogRecPtrIsInvalid(prev_lsn)) {
prev_lsn = cur_lsn;
prev_time = (pg_time_t) time(NULL);
goto DEFAULT;
}
if (XLByteEQ(prev_lsn, cur_lsn)) {
prev_time = get_time_ms();
avg_flush_num = min_io;
goto DEFAULT;
}
total_flush_num += g_instance.ckpt_cxt_ctl->page_writer_last_flush;
now = (pg_time_t) time(NULL);
now = get_time_ms();
time_diff = now - prev_time;
/*
@ -796,8 +845,10 @@ static uint32 calculate_pagewriter_flush_num()
time_diff > AVG_CALCULATE_NUM * u_sess->attr.attr_storage.pageWriterSleep) {
time_diff = MAX(1, time_diff);
avg_flush_num = (uint32)((((double)total_flush_num) / time_diff + avg_flush_num) / 2);
avg_lsn_rate = ((double)(cur_lsn - prev_lsn) / time_diff + avg_lsn_rate) / 2;
avg_flush_num = (uint32)((((double)total_flush_num) / time_diff * u_sess->attr.attr_storage.pageWriterSleep
+ avg_flush_num) / 2);
avg_lsn_rate = ((double)(cur_lsn - prev_lsn) / time_diff * u_sess->attr.attr_storage.pageWriterSleep
+ avg_lsn_rate) / 2;
/* reset our variables */
prev_lsn = cur_lsn;
@ -808,11 +859,33 @@ static uint32 calculate_pagewriter_flush_num()
dirty_page_pct = g_instance.ckpt_cxt_ctl->actual_dirty_page_num / (float)(g_instance.attr.attr_storage.NBuffers);
dirty_slot_pct = get_dirty_page_num() / (float)(g_instance.ckpt_cxt_ctl->dirty_page_queue_size);
num_for_dirty = MAX(dirty_page_pct, dirty_slot_pct) /
u_sess->attr.attr_storage.dirty_page_percent_max * min_io * 2;
target_lsn = prev_lsn + avg_lsn_rate * LSN_SCAN_FASTOR;
num_for_lsn = get_page_num_for_lsn(target_lsn);
num_for_lsn = MIN(max_io * 2, num_for_lsn / LSN_SCAN_FASTOR);
dirty_percent = MAX(dirty_page_pct, dirty_slot_pct) / u_sess->attr.attr_storage.dirty_page_percent_max;
if (RecoveryInProgress()) {
max_io = max_io * 0.9;
}
if (dirty_percent < HIGH_WATER) {
num_for_dirty = min_io;
num_for_lsn_max = max_io;
} else if (dirty_percent <= 1) {
num_for_dirty = min_io + (float)(dirty_percent - HIGH_WATER) / (float)(1 - HIGH_WATER) * (max_io - min_io);
num_for_lsn_max = max_io + (float)(dirty_percent - HIGH_WATER) / (float)(1 - HIGH_WATER) * (max_io);
} else {
num_for_dirty = max_io;
num_for_lsn_max = max_io * 2;
}
min_lsn = ckpt_get_min_rec_lsn();
if (XLogRecPtrIsInvalid(min_lsn)) {
min_lsn = get_dirty_page_queue_rec_lsn();
}
target_lsn = min_lsn + avg_lsn_rate;
num_for_lsn = get_page_num_for_lsn(target_lsn, num_for_lsn_max);
num_for_lsn = (num_for_lsn + prev_lsn_num) / 2;
prev_lsn_num = num_for_lsn;
flush_num = (avg_flush_num + num_for_dirty + num_for_lsn) / 3;
DEFAULT:
@ -829,7 +902,7 @@ DEFAULT:
static void ckpt_pagewriter_main_thread_loop(void)
{
uint32 rc = 0;
pg_time_t now;
uint64 now;
int64 sleep_time;
if (t_thrd.pagewriter_cxt.got_SIGHUP) {
@ -876,7 +949,7 @@ static void ckpt_pagewriter_main_thread_loop(void)
ckpt_try_skip_invalid_elem_in_queue_head();
ckpt_try_prune_dirty_page_queue();
/* Full checkpoint, don't sleep; the num of dirty page greater than max_dirty_page_num, don't sleep */
/* Full checkpoint, don't sleep */
sleep_time = get_pagewriter_sleep_time();
while (sleep_time > 0 && !t_thrd.pagewriter_cxt.shutdown_requested && !FULL_CKPT) {
/* sleep 1ms check whether a full checkpoint is triggered */
@ -885,7 +958,7 @@ static void ckpt_pagewriter_main_thread_loop(void)
}
/* Calculate next flush time before flush this batch dirty page */
now = (pg_time_t) time(NULL);
now = get_time_ms();
t_thrd.pagewriter_cxt.next_flush_time = now + u_sess->attr.attr_storage.pageWriterSleep;
/* pagewriter thread flush dirty page */
@ -1202,7 +1275,7 @@ static bool ckpt_found_valid_and_invalid_buffer_loc(
dirty_page_num = get_dirty_page_num();
if (dirty_page_num < g_instance.ckpt_cxt_ctl->dirty_page_queue_size * NEED_PRUNE_DIRTY_QUEUE_SLOT) {
if (dirty_page_num < g_instance.ckpt_cxt_ctl->dirty_page_queue_size * NEED_PRUNE_DIRTY_QUEUE_SLOT || FULL_CKPT) {
return false;
}
@ -1269,6 +1342,7 @@ static void ckpt_try_prune_dirty_page_queue()
* pages are moved to a new position after slot 100 due to this prune queue. than
* 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);
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));
}
@ -1308,6 +1382,7 @@ static void ckpt_try_prune_dirty_page_queue()
last_invalid_slot--;
}
LWLockRelease(g_instance.ckpt_cxt_ctl->prune_queue_lock);
if (u_sess->attr.attr_storage.log_pagewriter) {
print_dirty_page_queue_info(true);

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

@ -1075,7 +1075,7 @@ static XLogRecPtr dw_copy_page(ThrdDwCxt* thrd_dw_cxt, int buf_desc_id, bool* is
uint16 page_num;
uint32 buf_state;
errno_t rc;
*is_skipped = false;
*is_skipped = true;
buf_desc = GetBufferDescriptor(buf_desc_id);
buf_state = LockBufHdr(buf_desc);
@ -1100,9 +1100,9 @@ static XLogRecPtr dw_copy_page(ThrdDwCxt* thrd_dw_cxt, int buf_desc_id, bool* is
*/
if (!LWLockConditionalAcquire(buf_desc->content_lock, LW_SHARED)) {
UnpinBuffer(buf_desc, true);
*is_skipped = true;
return page_lsn;
}
*is_skipped = false;
thrd_dw_cxt->write_pos++;
if (thrd_dw_cxt->write_pos <= GET_DW_BATCH_DATA_PAGE_MAX(thrd_dw_cxt->contain_hashbucket)) {
batch = (dw_batch_t*)thrd_dw_cxt->dw_buf;
@ -1205,13 +1205,13 @@ static void dw_batch_flush(knl_g_dw_context* dw_cxt, XLogRecPtr latest_lsn, Thrd
dw_file_head_t* file_head = NULL;
errno_t rc;
(void)LWLockAcquire(dw_cxt->flush_lock, LW_EXCLUSIVE);
if (!XLogRecPtrIsInvalid(latest_lsn)) {
XLogFlush(latest_lsn);
g_instance.ckpt_cxt_ctl->page_writer_xlog_flush_loc = latest_lsn;
}
(void)LWLockAcquire(dw_cxt->flush_lock, LW_EXCLUSIVE);
if (thrd_dw_cxt->contain_hashbucket) {
dw_cxt->contain_hashbucket = true;
}

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

@ -142,6 +142,8 @@
#define STANDBY_SIGNAL_FILE "standby"
#define XLOG_SWITCH_HISTORY_FILE "switch.history"
#define MAX_PATH_LEN 1024
#define MAX(A, B) ((B) > (A) ? (B) : (A))
#define ENABLE_INCRE_CKPT g_instance.attr.attr_storage.enableIncrementalCheckpoint
#define RecoveryFromDummyStandby() (t_thrd.postmaster_cxt.ReplConnArray[2] != NULL && IS_DN_DUMMY_STANDYS_MODE())
@ -6706,7 +6708,7 @@ void BootStrapXLOG(void)
* segment with logid=0 logseg=1. The very first WAL segment, 0/0, is not
* used, so that we can use 0/0 to mean "before any valid WAL segment".
*/
if (g_instance.attr.attr_storage.enableIncrementalCheckpoint) {
if (ENABLE_INCRE_CKPT) {
u_sess->attr.attr_storage.fullPageWrites = false;
}
checkPoint.redo = XLogSegSize + SizeOfXLogLongPHD;
@ -8497,7 +8499,7 @@ void StartupXLOG(void)
t_thrd.xlog_cxt.RedoRecPtr = t_thrd.shemem_ptr_cxt.XLogCtl->RedoRecPtr =
t_thrd.shemem_ptr_cxt.XLogCtl->Insert.RedoRecPtr = checkPoint.redo;
if (g_instance.attr.attr_storage.enableIncrementalCheckpoint) {
if (ENABLE_INCRE_CKPT) {
t_thrd.xlog_cxt.doPageWrites = false;
} else {
t_thrd.xlog_cxt.doPageWrites = t_thrd.xlog_cxt.lastFullPageWrites;
@ -8771,7 +8773,7 @@ void StartupXLOG(void)
PublishStartupProcessInformation();
SetForwardFsyncRequests();
SendPostmasterSignal(PMSIGNAL_RECOVERY_STARTED);
if (g_instance.attr.attr_storage.enableIncrementalCheckpoint) {
if (ENABLE_INCRE_CKPT) {
t_thrd.xlog_cxt.pagewriter_launched = true;
} else {
t_thrd.xlog_cxt.bgwriterLaunched = true;
@ -9199,7 +9201,7 @@ void StartupXLOG(void)
* record before resource manager writes cleanup WAL records or checkpoint
* record is written.
*/
if (g_instance.attr.attr_storage.enableIncrementalCheckpoint) {
if (ENABLE_INCRE_CKPT) {
Insert->fullPageWrites = false;
} else {
Insert->fullPageWrites = t_thrd.xlog_cxt.lastFullPageWrites;
@ -9376,6 +9378,12 @@ void StartupXLOG(void)
xlogctl->SharedRecoveryInProgress = false;
xlogctl->IsRecoveryDone = true;
SpinLockRelease(&xlogctl->info_lck);
if (ENABLE_INCRE_CKPT) {
RecoveryQueueState *state = &g_instance.ckpt_cxt_ctl->ckpt_redo_state;
(void)LWLockAcquire(state->recovery_queue_lock, LW_EXCLUSIVE);
state->start = state->end;
(void)LWLockRelease(state->recovery_queue_lock);
}
}
NextXidAfterReovery = t_thrd.xact_cxt.ShmemVariableCache->nextXid;
@ -9967,7 +9975,7 @@ void InitXLOGAccess(void)
(void)GetRedoRecPtr();
/* Also update our copy of doPageWrites. */
if (g_instance.attr.attr_storage.enableIncrementalCheckpoint) {
if (ENABLE_INCRE_CKPT) {
t_thrd.xlog_cxt.doPageWrites = false;
} else {
t_thrd.xlog_cxt.doPageWrites = (Insert->fullPageWrites || Insert->forcePageWrites);
@ -10014,11 +10022,10 @@ XLogRecPtr GetRedoRecPtr(void)
*/
void GetFullPageWriteInfo(XLogFPWInfo *fpwInfo_p)
{
bool incremental = g_instance.attr.attr_storage.enableIncrementalCheckpoint;
fpwInfo_p->redoRecPtr = t_thrd.xlog_cxt.RedoRecPtr;
fpwInfo_p->doPageWrites = t_thrd.xlog_cxt.doPageWrites && !incremental;
fpwInfo_p->doPageWrites = t_thrd.xlog_cxt.doPageWrites && !ENABLE_INCRE_CKPT;
fpwInfo_p->forcePageWrites = t_thrd.shemem_ptr_cxt.XLogCtl->FpwBeforeFirstCkpt && !IsInitdb && !incremental;
fpwInfo_p->forcePageWrites = t_thrd.shemem_ptr_cxt.XLogCtl->FpwBeforeFirstCkpt && !IsInitdb && !ENABLE_INCRE_CKPT;
}
/*
@ -10135,6 +10142,8 @@ void ShutdownXLOG(int code, Datum arg)
ckpt_shutdown_pagewriter();
free(g_instance.ckpt_cxt_ctl->dirty_page_queue);
g_instance.ckpt_cxt_ctl->dirty_page_queue = NULL;
g_instance.ckpt_cxt_ctl->prune_queue_lock = NULL;
g_instance.ckpt_cxt_ctl->ckpt_redo_state.recovery_queue_lock = NULL;
ShutdownCLOG();
ShutdownCSNLOG();
@ -10284,7 +10293,7 @@ void CreateCheckPoint(int flags)
int nvxids = 0;
errno_t errorno = EOK;
XLogRecPtr curMinRecLSN = InvalidXLogRecPtr;
bool doFullCheckpoint = !g_instance.attr.attr_storage.enableIncrementalCheckpoint;
bool doFullCheckpoint = !ENABLE_INCRE_CKPT;
TransactionId oldest_active_xid = InvalidTransactionId;
TransactionId globalXmin = InvalidTransactionId;
@ -10376,8 +10385,7 @@ void CreateCheckPoint(int flags)
curInsert = XLogBytePosToRecPtr(Insert->CurrBytePos);
if ((g_instance.attr.attr_storage.enableIncrementalCheckpoint && (flags & CHECKPOINT_CAUSE_TIME)) ||
doFullCheckpoint) {
if ((ENABLE_INCRE_CKPT && (flags & CHECKPOINT_CAUSE_TIME)) || doFullCheckpoint) {
update_dirty_page_queue_rec_lsn(curInsert, true);
}
@ -10412,7 +10420,7 @@ void CreateCheckPoint(int flags)
gstrace_exit(GS_TRC_ID_CreateCheckPoint);
return;
}
} else if (g_instance.attr.attr_storage.enableIncrementalCheckpoint && doFullCheckpoint) {
} else if (ENABLE_INCRE_CKPT && doFullCheckpoint) {
/*
* enableIncrementalCheckpoint guc is on, but some conditions shuld do
* full checkpoint.
@ -10775,9 +10783,9 @@ void CreateCheckPoint(int flags)
}
}
if (doFullCheckpoint && g_instance.attr.attr_storage.enableIncrementalCheckpoint) {
if (doFullCheckpoint && ENABLE_INCRE_CKPT) {
XLogRecPtr MinRecLSN = ckpt_get_min_rec_lsn();
if (!XLogRecPtrIsInvalid(curMinRecLSN) && XLByteLT(MinRecLSN, t_thrd.xlog_cxt.RedoRecPtr)) {
if (!XLogRecPtrIsInvalid(MinRecLSN) && XLByteLT(MinRecLSN, t_thrd.xlog_cxt.RedoRecPtr)) {
ereport(PANIC, (errmsg("current dirty page list head recLSN %08X/%08X smaller than redo lsn %08X/%08X",
(uint32)(MinRecLSN >> XLOG_LSN_SWAP), (uint32)MinRecLSN,
(uint32)(t_thrd.xlog_cxt.RedoRecPtr >> XLOG_LSN_SWAP),
@ -10907,6 +10915,26 @@ static void CheckPointGuts(XLogRecPtr checkPointRedo, int flags, bool doFullChec
gstrace_exit(GS_TRC_ID_CheckPointGuts);
}
void PushRestartPointToQueue(XLogRecPtr recordReadRecPtr, const CheckPoint checkPoint)
{
RecoveryQueueState *state = &g_instance.ckpt_cxt_ctl->ckpt_redo_state;
uint loc = 0;
if (!ENABLE_INCRE_CKPT) {
return;
}
(void)LWLockAcquire(state->recovery_queue_lock, LW_EXCLUSIVE);
if (state->end - state->start + 1 >= RESTART_POINT_QUEUE_LEN) {
state->start++;
}
loc = state->end % RESTART_POINT_QUEUE_LEN;
state->ckpt_rec_queue[loc].CkptLSN = recordReadRecPtr;
state->ckpt_rec_queue[loc].checkpoint = checkPoint;
state->end++;
LWLockRelease(state->recovery_queue_lock);
}
/*
* Save a checkpoint for recovery restart if appropriate
*
@ -10929,21 +10957,24 @@ static void RecoveryRestartPoint(const CheckPoint checkPoint, XLogRecPtr recordR
if (IsExtremeRedo()) {
XLogRecPtr safeCheckPoint = extreme_rto::GetSafeMinCheckPoint();
if (XLByteEQ(safeCheckPoint, MAX_XLOG_REC_PTR) || XLByteLT(safeCheckPoint, recordReadRecPtr)) {
ereport(WARNING, (errmsg("RecoveryRestartPoint is false at %X/%X,last safe point is %X/%X",
ereport(WARNING, (errmsg("RecoveryRestartPoint is false at %X/%X,last safe point is %X/%X",
(uint32)(recordReadRecPtr >> 32), (uint32)(recordReadRecPtr), (uint32)(safeCheckPoint >> 32),
(uint32)(safeCheckPoint))));
return;
}
} else if (!parallel_recovery::IsRecoveryRestartPointSafeForWorkers(recordReadRecPtr)) {
ereport(WARNING, (errmsg("RecoveryRestartPointSafe is false at %X/%X",
ereport(WARNING, (errmsg("RecoveryRestartPointSafe is false at %X/%X",
static_cast<uint32>(recordReadRecPtr >> shitRightLength), static_cast<uint32>(recordReadRecPtr))));
return;
}
update_dirty_page_queue_rec_lsn(recordReadRecPtr, true);
pg_write_barrier();
PushRestartPointToQueue(recordReadRecPtr, checkPoint);
/*
* Copy the checkpoint record to shared memory, so that checkpointer can
* work out the next time it wants to perform a restartpoint.
*/
SpinLockAcquire(&xlogctl->info_lck);
xlogctl->lastCheckPointRecPtr = recordReadRecPtr;
const_cast<CheckPoint &>(xlogctl->lastCheckPoint) = const_cast<CheckPoint &>(checkPoint);
@ -10993,8 +11024,7 @@ bool IsRestartPointSafe(const XLogRecPtr checkPoint)
void wait_all_dirty_page_flush(int flags, XLogRecPtr redo)
{
/* need wait all dirty page finish flush */
if (g_instance.attr.attr_storage.enableIncrementalCheckpoint) {
update_dirty_page_queue_rec_lsn(redo, true);
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();
@ -11006,6 +11036,119 @@ void wait_all_dirty_page_flush(int flags, XLogRecPtr redo)
}
return;
}
bool RecoveryQueueIsEmpty()
{
RecoveryQueueState *state = &g_instance.ckpt_cxt_ctl->ckpt_redo_state;
int num;
(void)LWLockAcquire(state->recovery_queue_lock, LW_EXCLUSIVE);
num = state->end - state->start;
(void)LWLockRelease(state->recovery_queue_lock);
if (num == 0) {
return true;
} else {
return false;
}
}
/*
* In the recovery phase, don't push the redo point to the latest position, avoid the I/O peak.
* calculate the redo point based on the page flushing speed and max_redo_log_size.
*/
const int BYTE_PER_KB = 1024;
XLogRecPtr GetRestartPointInRecovery(CheckPoint *restartCheckPoint)
{
volatile XLogCtlData *xlogctl = t_thrd.shemem_ptr_cxt.XLogCtl;
XLogRecPtr restartRecPtr = InvalidXLogRecPtr;
XLogRecPtr curMinRecLSN = ckpt_get_min_rec_lsn();
RecoveryQueueState *state = &g_instance.ckpt_cxt_ctl->ckpt_redo_state;
if (XLogRecPtrIsInvalid(curMinRecLSN)) {
/* The dirty page queue is empty, so the redo point can be updated to the latest position. */
(void)LWLockAcquire(state->recovery_queue_lock, LW_EXCLUSIVE);
state->start = state->end > 0 ? state->end - 1 : state->end;
(void)LWLockRelease(state->recovery_queue_lock);
SpinLockAcquire(&xlogctl->info_lck);
restartRecPtr = xlogctl->lastCheckPointRecPtr;
*restartCheckPoint = const_cast<CheckPoint &>(xlogctl->lastCheckPoint);
SpinLockRelease(&xlogctl->info_lck);
Assert(XLByteLE(restartRecPtr, get_dirty_page_queue_rec_lsn()));
} else {
int num = 0;
int loc = 0;
int i = 0;
XLogRecPtr replayLastLSN;
XLogRecPtr targetLSN;
if (RecoveryQueueIsEmpty()) {
return restartRecPtr;
}
SpinLockAcquire(&xlogctl->info_lck);
replayLastLSN = xlogctl->lastCheckPointRecPtr;
SpinLockRelease(&xlogctl->info_lck);
targetLSN = replayLastLSN - u_sess->attr.attr_storage.max_redo_log_size * BYTE_PER_KB;
(void)LWLockAcquire(state->recovery_queue_lock, LW_EXCLUSIVE);
num = state->end - state->start;
/*
* If the pagewriter flush dirty page to disk quickly, push the redo point to
* the position closest to curMinRecLSN .
*/
if (XLByteLE(targetLSN, curMinRecLSN)) {
for (i = 0; i < num; i++) {
loc = (state->start + i) % RESTART_POINT_QUEUE_LEN;
if (state->ckpt_rec_queue[loc].checkpoint.redo > curMinRecLSN) {
if (i > 0) {
loc = (state->start + i - 1) % RESTART_POINT_QUEUE_LEN;
state->start = state->start + i - 1;
}
restartRecPtr = state->ckpt_rec_queue[loc].CkptLSN;
*restartCheckPoint = state->ckpt_rec_queue[loc].checkpoint;
break;
}
}
} else {
/* In other cases, push the checkpoint loc to the position closest to targetLSN. */
for (i = 0; i < num; i++) {
loc = (state->start + i) % RESTART_POINT_QUEUE_LEN;
if (state->ckpt_rec_queue[loc].CkptLSN > targetLSN) {
if (i > 0) {
uint64 gap = state->ckpt_rec_queue[loc].CkptLSN - targetLSN;
int prevLoc = (state->start + i - 1) % RESTART_POINT_QUEUE_LEN;
if (targetLSN - state->ckpt_rec_queue[prevLoc].CkptLSN < gap) {
restartRecPtr = state->ckpt_rec_queue[prevLoc].CkptLSN;
*restartCheckPoint = state->ckpt_rec_queue[prevLoc].checkpoint;
state->start = state->start + i - 1;
} else {
restartRecPtr = state->ckpt_rec_queue[loc].CkptLSN;
*restartCheckPoint = state->ckpt_rec_queue[loc].checkpoint;
state->start = state->start + i;
}
} else {
restartRecPtr = state->ckpt_rec_queue[loc].CkptLSN;
*restartCheckPoint = state->ckpt_rec_queue[loc].checkpoint;
}
break;
}
}
}
if (XLogRecPtrIsInvalid(restartRecPtr) && num > 0) {
loc = (state->end - 1) % RESTART_POINT_QUEUE_LEN;
restartRecPtr = state->ckpt_rec_queue[loc].CkptLSN;
*restartCheckPoint = state->ckpt_rec_queue[loc].checkpoint;
state->start = state->end - 1;
}
(void)LWLockRelease(state->recovery_queue_lock);
}
return restartRecPtr;
}
/*
* Establish a restartpoint if possible.
*
@ -11025,6 +11168,7 @@ bool CreateRestartPoint(int flags)
TimestampTz xtime;
errno_t errorno = EOK;
bool recoveryInProgress = true;
bool doFullCkpt = !ENABLE_INCRE_CKPT;
/* use volatile pointer to prevent code rearrangement */
volatile XLogCtlData *xlogctl = t_thrd.shemem_ptr_cxt.XLogCtl;
@ -11036,12 +11180,6 @@ bool CreateRestartPoint(int flags)
gstrace_entry(GS_TRC_ID_CreateRestartPoint);
LWLockAcquire(CheckpointLock, LW_EXCLUSIVE);
/* Get a local copy of the last safe checkpoint record. */
SpinLockAcquire(&xlogctl->info_lck);
lastCheckPointRecPtr = xlogctl->lastCheckPointRecPtr;
lastCheckPoint = const_cast<CheckPoint &>(xlogctl->lastCheckPoint);
SpinLockRelease(&xlogctl->info_lck);
recoveryInProgress = RecoveryInProgress();
/*
* Check that we're still in recovery mode. It's ok if we exit recovery
@ -11055,6 +11193,23 @@ bool CreateRestartPoint(int flags)
return false;
}
if (doFullCkpt ||
((unsigned int)flags & (CHECKPOINT_IS_SHUTDOWN | CHECKPOINT_END_OF_RECOVERY | CHECKPOINT_FORCE))) {
doFullCkpt = true;
if (ENABLE_INCRE_CKPT) {
RecoveryQueueState *state = &g_instance.ckpt_cxt_ctl->ckpt_redo_state;
(void)LWLockAcquire(state->recovery_queue_lock, LW_EXCLUSIVE);
state->start = state->end > 0 ? state->end - 1 : state->end;
(void)LWLockRelease(state->recovery_queue_lock);
}
SpinLockAcquire(&xlogctl->info_lck);
lastCheckPointRecPtr = xlogctl->lastCheckPointRecPtr;
lastCheckPoint = const_cast<CheckPoint &>(xlogctl->lastCheckPoint);
SpinLockRelease(&xlogctl->info_lck);
} else {
lastCheckPointRecPtr = GetRestartPointInRecovery(&lastCheckPoint);
}
/*
* If the last checkpoint record we've replayed is already our last
* restartpoint, we can't perform a new restart point. We still update
@ -11138,8 +11293,7 @@ bool CreateRestartPoint(int flags)
LogCheckpointStart((unsigned int)flags, true);
}
if (g_instance.attr.attr_storage.enableIncrementalCheckpoint) {
update_dirty_page_queue_rec_lsn(lastCheckPoint.redo, true);
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();
@ -11147,7 +11301,23 @@ bool CreateRestartPoint(int flags)
g_instance.ckpt_cxt_ctl->flush_all_dirty_page = true;
}
ereport(LOG, (errmsg("CreateRestartPoint, need flush %ld pages.", get_dirty_page_num())));
} 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);
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 ?
g_instance.ckpt_cxt_ctl->full_ckpt_expected_flush_loc - head : 0;
if (need_flush_num > 0) {
g_instance.ckpt_cxt_ctl->flush_all_dirty_page = true;
}
ereport(LOG, (errmsg("CreateRestartPoint, need flush %ld pages", need_flush_num)));
}
CheckPointGuts(lastCheckPoint.redo, flags, true);
#ifdef ENABLE_MOT
@ -11159,6 +11329,15 @@ bool CreateRestartPoint(int flags)
* prior checkpoint's earliest info.
*/
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)) {
ereport(PANIC, (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)));
}
}
/*
* Update pg_control, using current time. Check that it still shows
@ -11250,7 +11429,7 @@ bool CreateRestartPoint(int flags)
* Reduce the frequency of trucate CSN log to avoid the probability of lock contention.
* Incremental chekpoint does not require frequent truncate of csnlog.
*/
if (!g_instance.attr.attr_storage.enableIncrementalCheckpoint ||
if (!ENABLE_INCRE_CKPT ||
elapsed_secs >= u_sess->attr.attr_storage.fullCheckPointTimeout) {
TransactionId globalXmin = InvalidTransactionId;
(void)GetOldestActiveTransactionId(&globalXmin);
@ -11544,7 +11723,7 @@ void UpdateFullPageWrites(void)
* because we assume that there is no concurrently running process which
* can update it.
*/
if (g_instance.attr.attr_storage.enableIncrementalCheckpoint) {
if (ENABLE_INCRE_CKPT) {
u_sess->attr.attr_storage.fullPageWrites = false;
}
if (u_sess->attr.attr_storage.fullPageWrites == Insert->fullPageWrites) {
@ -12385,7 +12564,7 @@ char** tblspcmapfile, List** tablespaces, bool infotbssize, bool needtblspcmapfi
} else {
t_thrd.shemem_ptr_cxt.XLogCtl->Insert.nonExclusiveBackups++;
}
if (g_instance.attr.attr_storage.enableIncrementalCheckpoint) {
if (ENABLE_INCRE_CKPT) {
t_thrd.shemem_ptr_cxt.XLogCtl->Insert.forcePageWrites = false;
} else {
t_thrd.shemem_ptr_cxt.XLogCtl->Insert.forcePageWrites = true;
@ -14079,6 +14258,7 @@ void SetXLogReplayRecPtr(XLogRecPtr readRecPtr, XLogRecPtr endRecPtr)
if (isUpdated && !IsExtremeRedo()) {
RedoSpeedDiag(readRecPtr, endRecPtr);
}
update_dirty_page_queue_rec_lsn(readRecPtr);
}
void DumpXlogCtl()
@ -16595,7 +16775,7 @@ bool IsRoachRestore(void)
strncmp(t_thrd.xlog_cxt.recoveryTargetBarrierId, ROACH_BACKUP_PREFIX, strlen(ROACH_BACKUP_PREFIX)) == 0);
}
const int UPDATE_REC_XLOG_NUM = 10;
const uint UPDATE_REC_XLOG_NUM = 4;
#if defined(__x86_64__) || defined(__aarch64__)
bool atomic_update_dirty_page_queue_rec_lsn(XLogRecPtr current_insert_lsn, bool need_immediately_update)
{
@ -16640,7 +16820,7 @@ void update_dirty_page_queue_rec_lsn(XLogRecPtr current_insert_lsn, bool need_im
bool is_update = false;
uint32 freespace;
if (!g_instance.attr.attr_storage.enableIncrementalCheckpoint) {
if (!ENABLE_INCRE_CKPT) {
return;
}
@ -16698,21 +16878,20 @@ uint64 get_dirty_page_queue_rec_lsn()
XLogRecPtr ckpt_get_min_rec_lsn(void)
{
uint64 queue_loc;
XLogRecPtr dirty_queue_min_lsn = InvalidXLogRecPtr;
uint64 dirty_page_queue_tail;
XLogRecPtr min_rec_lsn = InvalidXLogRecPtr;
/*
* If head recLSN is Invalid, then add head, get next buffer recLSN, if head equal tail,
* return InvalidXLogRecPtr.
*/
queue_loc = pg_atomic_read_u64(&g_instance.ckpt_cxt_ctl->dirty_page_queue_head);
dirty_page_queue_tail = get_dirty_page_queue_tail();
if (dirty_page_queue_tail - queue_loc == 0) {
if (get_dirty_page_num() == 0) {
return InvalidXLogRecPtr;
}
while (XLogRecPtrIsInvalid(dirty_queue_min_lsn) && (queue_loc < get_dirty_page_queue_tail())) {
queue_loc = pg_atomic_read_u64(&g_instance.ckpt_cxt_ctl->dirty_page_queue_head);
while (XLogRecPtrIsInvalid(min_rec_lsn) && (queue_loc < get_dirty_page_queue_tail())) {
Buffer buffer;
BufferDesc *buf_desc = NULL;
XLogRecPtr page_rec_lsn = InvalidXLogRecPtr;
uint64 temp_loc = queue_loc % g_instance.ckpt_cxt_ctl->dirty_page_queue_size;
volatile DirtyPageQueueSlot *slot = &g_instance.ckpt_cxt_ctl->dirty_page_queue[temp_loc];
@ -16731,9 +16910,12 @@ XLogRecPtr ckpt_get_min_rec_lsn(void)
continue;
}
buf_desc = GetBufferDescriptor(buffer - 1);
dirty_queue_min_lsn = pg_atomic_read_u64(&buf_desc->rec_lsn);
page_rec_lsn = pg_atomic_read_u64(&buf_desc->rec_lsn);
if (!BufferIsInvalid(slot->buffer)) {
min_rec_lsn = page_rec_lsn;
}
}
return dirty_queue_min_lsn;
return min_rec_lsn;
}
void WaitCheckpointSync(void)

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

@ -22,6 +22,7 @@
#include "storage/buf/bufmgr.h"
#include "storage/proc.h"
#include "postmaster/aiocompleter.h" /* this is for the function AioCompltrIsReady() */
#include "postmaster/bgwriter.h"
#include "postmaster/pagewriter.h"
#include "postmaster/postmaster.h"
#include "access/double_write.h"
@ -467,7 +468,8 @@ BufferAccessStrategy GetAccessStrategy(BufferAccessStrategyType btype)
ring_size = (u_sess->attr.attr_storage.bulk_write_ring_size / BLCKSZ) * 1024;
break;
case BAS_VACUUM:
ring_size = 256 * 1024 / BLCKSZ;
ring_size = g_instance.attr.attr_storage.NBuffers / 32 /
Max(g_instance.attr.attr_storage.autovacuum_max_workers, 1);
break;
default:
@ -508,6 +510,8 @@ void FreeAccessStrategy(BufferAccessStrategy strategy)
}
}
const int MAX_RETRY_RING_TIMES = 100;
const float MAX_RETRY_RING_PCT = 0.1;
/*
* GetBufferFromRing -- returns a buffer from the ring, or NULL if the
* ring is empty.
@ -519,10 +523,13 @@ static BufferDesc *GetBufferFromRing(BufferAccessStrategy strategy, uint32 *buf_
BufferDesc *buf = NULL;
Buffer buf_num;
uint32 local_buf_state; /* to avoid repeated (de-)referencing */
uint16 retry_times = 0;
RETRY:
/* Advance to next ring slot */
if (++strategy->current >= strategy->ring_size)
strategy->current = 0;
retry_times++;
ADIO_RUN()
{
@ -575,6 +582,15 @@ static BufferDesc *GetBufferFromRing(BufferAccessStrategy strategy, uint32 *buf_
* shouldn't re-use it.
*/
buf = GetBufferDescriptor(buf_num - 1);
if ((pg_atomic_read_u32(&buf->state) & BM_DIRTY) &&
retry_times < Min(MAX_RETRY_RING_TIMES, strategy->ring_size * MAX_RETRY_RING_PCT)) {
goto RETRY;
} else if (get_curr_candidate_nums() >= (uint32)g_instance.attr.attr_storage.NBuffers *
u_sess->attr.attr_storage.candidate_buf_percent_target) {
strategy->current_was_in_ring = false;
return NULL;
}
local_buf_state = LockBufHdr(buf);
if (BUF_STATE_GET_REFCOUNT(local_buf_state) == 0 && BUF_STATE_GET_USAGECOUNT(local_buf_state) <= 1 &&
(backend_can_flush_dirty_page() || !(local_buf_state & BM_DIRTY))) {
@ -582,6 +598,7 @@ static BufferDesc *GetBufferFromRing(BufferAccessStrategy strategy, uint32 *buf_
*buf_state = local_buf_state;
return buf;
}
UnlockBufHdr(buf, local_buf_state);
/*
* Tell caller to allocate a new buffer with the normal allocation

19
src/gausskernel/storage/ipc/ipci.cpp
View File

@ -363,6 +363,10 @@ void CreateSharedMemoryAndSemaphores(bool makePrivate, int port)
LsnXlogFlushChkShmInit();
if (g_instance.ckpt_cxt_ctl->prune_queue_lock == NULL) {
g_instance.ckpt_cxt_ctl->prune_queue_lock = LWLockAssign(LWTRANCHE_PRUNE_DIRTY_QUEUE);
}
if (g_instance.pid_cxt.PageWriterPID == NULL) {
MemoryContext oldcontext = MemoryContextSwitchTo(g_instance.increCheckPoint_context);
g_instance.pid_cxt.PageWriterPID =
@ -383,6 +387,21 @@ void CreateSharedMemoryAndSemaphores(bool makePrivate, int port)
incre_ckpt_bgwriter_cxt_init();
}
if (g_instance.attr.attr_storage.enableIncrementalCheckpoint &&
g_instance.ckpt_cxt_ctl->ckpt_redo_state.ckpt_rec_queue == NULL) {
MemoryContext oldcontext = MemoryContextSwitchTo(g_instance.increCheckPoint_context);
g_instance.ckpt_cxt_ctl->ckpt_redo_state.recovery_queue_lock = LWLockAssign(LWTRANCHE_REDO_POINT_QUEUE);
g_instance.ckpt_cxt_ctl->ckpt_redo_state.ckpt_rec_queue =
(CheckPointItem*)palloc0(sizeof(CheckPointItem) * RESTART_POINT_QUEUE_LEN);
g_instance.ckpt_cxt_ctl->ckpt_redo_state.start = 0;
g_instance.ckpt_cxt_ctl->ckpt_redo_state.end = 0;
(void)MemoryContextSwitchTo(oldcontext);
}
if (g_instance.ckpt_cxt_ctl->ckpt_redo_state.recovery_queue_lock == NULL) {
g_instance.ckpt_cxt_ctl->ckpt_redo_state.recovery_queue_lock = LWLockAssign(LWTRANCHE_REDO_POINT_QUEUE);
}
/*
* Now give loadable modules a chance to set up their shmem allocations
*/

8
src/gausskernel/storage/lmgr/lwlock.cpp
View File

@ -157,6 +157,8 @@ static const char *BuiltinTrancheNames[] = {
"DoubleWriteLock",
"DWSingleFlushPosLock",
"DWSingleFlushWriteLock",
"RestartPointQueueLock",
"PruneDirtyQueueLock",
"LWTRANCHE_ACCOUNT_TABLE",
"GeneralExtendedLock",
"MPFLLOCK",
@ -374,6 +376,12 @@ int NumLWLocks(void)
/* for materialized view */
numLocks += 1;
/* for recovery state queue */
numLocks += 1;
/* for prune dirty queue */
numLocks += 1;
/*
* Add any requested by loadable modules; for backwards-compatibility
* reasons, allocate at least NUM_USER_DEFINED_LWLOCKS of them even if

2
src/include/knl/knl_instance.h
View File

@ -399,6 +399,7 @@ typedef struct knl_g_ckpt_context {
pg_atomic_uint64 dirty_page_queue_head;
pg_atomic_uint32 actual_dirty_page_num;
slock_t queue_lock;
struct LWLock* prune_queue_lock;
/* pagewriter thread */
PageWriterProcs page_writer_procs;
@ -423,6 +424,7 @@ typedef struct knl_g_ckpt_context {
int64 ckpt_predicate_flush_num;
int64 ckpt_twophase_flush_num;
volatile XLogRecPtr ckpt_current_redo_point;
RecoveryQueueState ckpt_redo_state;
#ifdef ENABLE_MOT
struct CheckpointCallbackItem* ckptCallback;

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

@ -1407,7 +1407,7 @@ typedef struct knl_t_bgwriter_context {
volatile sig_atomic_t got_SIGHUP;
volatile sig_atomic_t shutdown_requested;
int thread_id;
pg_time_t next_flush_time;
uint64 next_flush_time;
} knl_t_bgwriter_context;
typedef struct knl_t_pagewriter_context {
@ -1415,7 +1415,7 @@ typedef struct knl_t_pagewriter_context {
volatile sig_atomic_t shutdown_requested;
int page_writer_after;
int pagewriter_id;
pg_time_t next_flush_time;
uint64 next_flush_time;
} knl_t_pagewriter_context;
#define MAX_SEQ_SCANS 100

1
src/include/postmaster/bgwriter.h
View File

@ -59,6 +59,7 @@ extern void incre_ckpt_background_writer_main(void);
extern void ckpt_shutdown_bgwriter();
extern int get_bgwriter_thread_id(void);
extern bool candidate_buf_pop(int *bufId, int threadId);
extern uint32 get_curr_candidate_nums(void);
typedef struct BgWriterProc {
PGPROC *proc;

17
src/include/postmaster/pagewriter.h
View File

@ -25,6 +25,7 @@
#ifndef _PAGEWRITER_H
#define _PAGEWRITER_H
#include "storage/buf/buf.h"
#include "catalog/pg_control.h"
typedef struct PGPROC PGPROC;
typedef struct BufferDesc BufferDesc;
@ -66,6 +67,20 @@ typedef struct incre_ckpt_view_col {
incre_ckpt_view_get_data_func get_val;
} incre_ckpt_view_col;
const int RESTART_POINT_QUEUE_LEN = 20;
/* recovery checkpoint queue */
typedef struct CheckPointItem {
XLogRecPtr CkptLSN;
CheckPoint checkpoint;
} CheckPointItem;
typedef struct RecoveryQueueState {
uint64 start;
uint64 end;
CheckPointItem *ckpt_rec_queue;
LWLock *recovery_queue_lock;
} RecoveryQueueState;
/*
* The slot location is pre-occupied. When the slot buffer is set, the state will set
* to valid. when remove dirty page form queue, don't change the state, only when move
@ -89,6 +104,8 @@ extern void ckpt_shutdown_pagewriter();
extern uint64 get_dirty_page_queue_rec_lsn();
extern XLogRecPtr ckpt_get_min_rec_lsn(void);
extern uint32 calculate_thread_max_flush_num(bool is_pagewriter);
extern uint32 get_loc_for_lsn(XLogRecPtr target_lsn);
extern uint64 get_time_ms();
const int PAGEWRITER_VIEW_COL_NUM = 8;
const int INCRE_CKPT_VIEW_COL_NUM = 7;

2
src/include/storage/lock/lwlock.h
View File

@ -177,6 +177,8 @@ enum BuiltinTrancheIds
LWTRANCHE_DOUBLE_WRITE,
LWTRANCHE_DW_SINGLE_POS,
LWTRANCHE_DW_SINGLE_WRITE,
LWTRANCHE_REDO_POINT_QUEUE,
LWTRANCHE_PRUNE_DIRTY_QUEUE,
LWTRANCHE_ACCOUNT_TABLE,
LWTRANCHE_EXTEND, // For general 3rd plugin
LWTRANCHE_MPFL,