From 4b995e942cd8eda0694c96012d9f07ce843cf32e Mon Sep 17 00:00:00 2001 From: MuJinqiang <1165845907@qq.com> Date: Thu, 4 Feb 2021 22:24:03 +0800 Subject: [PATCH 1/2] =?UTF-8?q?=E5=A4=87=E6=9C=BAio+autovacuum=E4=BC=98?= =?UTF-8?q?=E5=8C=96?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../process/postmaster/bgwriter.cpp | 34 ++- .../process/postmaster/pagewriter.cpp | 155 ++++++++--- .../storage/access/transam/double_write.cpp | 8 +- .../storage/access/transam/xlog.cpp | 259 +++++++++++++++--- src/gausskernel/storage/buffer/freelist.cpp | 19 +- src/gausskernel/storage/ipc/ipci.cpp | 15 + src/gausskernel/storage/lmgr/lwlock.cpp | 4 + src/include/knl/knl_instance.h | 1 + src/include/knl/knl_thread.h | 4 +- src/include/postmaster/bgwriter.h | 1 + src/include/postmaster/pagewriter.h | 17 ++ src/include/storage/lock/lwlock.h | 1 + 12 files changed, 412 insertions(+), 106 deletions(-) diff --git a/src/gausskernel/process/postmaster/bgwriter.cpp b/src/gausskernel/process/postmaster/bgwriter.cpp index bc58ec2cf..765a9c927 100755 --- a/src/gausskernel/process/postmaster/bgwriter.cpp +++ b/src/gausskernel/process/postmaster/bgwriter.cpp @@ -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++) { diff --git a/src/gausskernel/process/postmaster/pagewriter.cpp b/src/gausskernel/process/postmaster/pagewriter.cpp index d2ab827ae..c385dfa80 100644 --- a/src/gausskernel/process/postmaster/pagewriter.cpp +++ b/src/gausskernel/process/postmaster/pagewriter.cpp @@ -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 */ diff --git a/src/gausskernel/storage/access/transam/double_write.cpp b/src/gausskernel/storage/access/transam/double_write.cpp index a909051fa..909fd1f74 100644 --- a/src/gausskernel/storage/access/transam/double_write.cpp +++ b/src/gausskernel/storage/access/transam/double_write.cpp @@ -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; } diff --git a/src/gausskernel/storage/access/transam/xlog.cpp b/src/gausskernel/storage/access/transam/xlog.cpp index 305b3120c..b90a9b9ae 100755 --- a/src/gausskernel/storage/access/transam/xlog.cpp +++ b/src/gausskernel/storage/access/transam/xlog.cpp @@ -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,7 @@ 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->ckpt_redo_state.recovery_queue_lock = NULL; ShutdownCLOG(); ShutdownCSNLOG(); @@ -10284,7 +10292,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 +10384,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 +10419,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 +10782,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 +10914,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 +10956,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(recordReadRecPtr >> shitRightLength), static_cast(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(xlogctl->lastCheckPoint) = const_cast(checkPoint); @@ -10993,8 +11023,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 +11035,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(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 +11167,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 +11179,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(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 +11192,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(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 +11292,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 +11300,21 @@ 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; + g_instance.ckpt_cxt_ctl->full_ckpt_expected_flush_loc = get_loc_for_lsn(lastCheckPoint.redo); + 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 +11326,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 +11426,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 +11720,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 +12561,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 +14255,7 @@ void SetXLogReplayRecPtr(XLogRecPtr readRecPtr, XLogRecPtr endRecPtr) if (isUpdated && !IsExtremeRedo()) { RedoSpeedDiag(readRecPtr, endRecPtr); } + update_dirty_page_queue_rec_lsn(readRecPtr); } void DumpXlogCtl() @@ -16595,7 +16772,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 +16817,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 +16875,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 +16907,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) diff --git a/src/gausskernel/storage/buffer/freelist.cpp b/src/gausskernel/storage/buffer/freelist.cpp index c7d1d332a..1cae56524 100644 --- a/src/gausskernel/storage/buffer/freelist.cpp +++ b/src/gausskernel/storage/buffer/freelist.cpp @@ -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 diff --git a/src/gausskernel/storage/ipc/ipci.cpp b/src/gausskernel/storage/ipc/ipci.cpp index 4892dc6e4..913937307 100755 --- a/src/gausskernel/storage/ipc/ipci.cpp +++ b/src/gausskernel/storage/ipc/ipci.cpp @@ -383,6 +383,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 */ diff --git a/src/gausskernel/storage/lmgr/lwlock.cpp b/src/gausskernel/storage/lmgr/lwlock.cpp index d841c1bd4..b8c24b1e2 100755 --- a/src/gausskernel/storage/lmgr/lwlock.cpp +++ b/src/gausskernel/storage/lmgr/lwlock.cpp @@ -157,6 +157,7 @@ static const char *BuiltinTrancheNames[] = { "DoubleWriteLock", "DWSingleFlushPosLock", "DWSingleFlushWriteLock", + "RestartPointQueueLock", "LWTRANCHE_ACCOUNT_TABLE", "GeneralExtendedLock", "MPFLLOCK", @@ -374,6 +375,9 @@ int NumLWLocks(void) /* for materialized view */ numLocks += 1; + /* for recovery state queue */ + numLocks += 1; + /* * Add any requested by loadable modules; for backwards-compatibility * reasons, allocate at least NUM_USER_DEFINED_LWLOCKS of them even if diff --git a/src/include/knl/knl_instance.h b/src/include/knl/knl_instance.h index 736179ede..645de3dc3 100644 --- a/src/include/knl/knl_instance.h +++ b/src/include/knl/knl_instance.h @@ -423,6 +423,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; diff --git a/src/include/knl/knl_thread.h b/src/include/knl/knl_thread.h index 2bbba0eeb..ad933636f 100644 --- a/src/include/knl/knl_thread.h +++ b/src/include/knl/knl_thread.h @@ -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 diff --git a/src/include/postmaster/bgwriter.h b/src/include/postmaster/bgwriter.h index b411f5c38..de1a2de05 100755 --- a/src/include/postmaster/bgwriter.h +++ b/src/include/postmaster/bgwriter.h @@ -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; diff --git a/src/include/postmaster/pagewriter.h b/src/include/postmaster/pagewriter.h index 309489aa6..06fbf2ad4 100644 --- a/src/include/postmaster/pagewriter.h +++ b/src/include/postmaster/pagewriter.h @@ -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; diff --git a/src/include/storage/lock/lwlock.h b/src/include/storage/lock/lwlock.h index cef909b11..c6cf9adb7 100644 --- a/src/include/storage/lock/lwlock.h +++ b/src/include/storage/lock/lwlock.h @@ -177,6 +177,7 @@ enum BuiltinTrancheIds LWTRANCHE_DOUBLE_WRITE, LWTRANCHE_DW_SINGLE_POS, LWTRANCHE_DW_SINGLE_WRITE, + LWTRANCHE_REDO_POINT_QUEUE, LWTRANCHE_ACCOUNT_TABLE, LWTRANCHE_EXTEND, // For general 3rd plugin LWTRANCHE_MPFL, From 72d4ada21a6c4ad939690fbefcf39b4c930a70f8 Mon Sep 17 00:00:00 2001 From: MuJinqiang <1165845907@qq.com> Date: Sat, 6 Feb 2021 21:58:19 +0800 Subject: [PATCH 2/2] fix core restartpoint --- src/gausskernel/process/postmaster/pagewriter.cpp | 4 +++- src/gausskernel/storage/access/transam/xlog.cpp | 3 +++ src/gausskernel/storage/ipc/ipci.cpp | 4 ++++ src/gausskernel/storage/lmgr/lwlock.cpp | 4 ++++ src/include/knl/knl_instance.h | 1 + src/include/storage/lock/lwlock.h | 1 + 6 files changed, 16 insertions(+), 1 deletion(-) diff --git a/src/gausskernel/process/postmaster/pagewriter.cpp b/src/gausskernel/process/postmaster/pagewriter.cpp index c385dfa80..f3a21162b 100644 --- a/src/gausskernel/process/postmaster/pagewriter.cpp +++ b/src/gausskernel/process/postmaster/pagewriter.cpp @@ -1275,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; } @@ -1342,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)); } @@ -1381,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); diff --git a/src/gausskernel/storage/access/transam/xlog.cpp b/src/gausskernel/storage/access/transam/xlog.cpp index b90a9b9ae..517f1b1cd 100755 --- a/src/gausskernel/storage/access/transam/xlog.cpp +++ b/src/gausskernel/storage/access/transam/xlog.cpp @@ -10142,6 +10142,7 @@ 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(); @@ -11302,7 +11303,9 @@ bool CreateRestartPoint(int flags) 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); diff --git a/src/gausskernel/storage/ipc/ipci.cpp b/src/gausskernel/storage/ipc/ipci.cpp index 913937307..de6db9320 100755 --- a/src/gausskernel/storage/ipc/ipci.cpp +++ b/src/gausskernel/storage/ipc/ipci.cpp @@ -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 = diff --git a/src/gausskernel/storage/lmgr/lwlock.cpp b/src/gausskernel/storage/lmgr/lwlock.cpp index b8c24b1e2..b3a338856 100755 --- a/src/gausskernel/storage/lmgr/lwlock.cpp +++ b/src/gausskernel/storage/lmgr/lwlock.cpp @@ -158,6 +158,7 @@ static const char *BuiltinTrancheNames[] = { "DWSingleFlushPosLock", "DWSingleFlushWriteLock", "RestartPointQueueLock", + "PruneDirtyQueueLock", "LWTRANCHE_ACCOUNT_TABLE", "GeneralExtendedLock", "MPFLLOCK", @@ -378,6 +379,9 @@ int NumLWLocks(void) /* 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 diff --git a/src/include/knl/knl_instance.h b/src/include/knl/knl_instance.h index 645de3dc3..87ba38847 100644 --- a/src/include/knl/knl_instance.h +++ b/src/include/knl/knl_instance.h @@ -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; diff --git a/src/include/storage/lock/lwlock.h b/src/include/storage/lock/lwlock.h index c6cf9adb7..acc347f5c 100644 --- a/src/include/storage/lock/lwlock.h +++ b/src/include/storage/lock/lwlock.h @@ -178,6 +178,7 @@ enum BuiltinTrancheIds 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,