* Copyright (c) 2020 Huawei Technologies Co.,Ltd.
* Portions Copyright (c) 2021, openGauss Contributors
*
* openGauss is licensed under Mulan PSL v2.
* You can use this software according to the terms and conditions of the Mulan PSL v2.
* You may obtain a copy of Mulan PSL v2 at:
*
* http://license.coscl.org.cn/MulanPSL2
*
* THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
* EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
* MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
* See the Mulan PSL v2 for more details.
* ---------------------------------------------------------------------------------------
*
* double_write.cpp
* Before flush dirty pages to data file, flush them to double write file,
* in case of half-flushed pages. Recover those half-flushed data file pages
* before replaying xlog when starting.
*
* IDENTIFICATION
* src/gausskernel/storage/access/transam/seg_double_write.cpp
*
* ---------------------------------------------------------------------------------------
*/
#include <unistd.h>
#include "utils/elog.h"
#include "access/double_write.h"
#include "utils/palloc.h"
#ifdef ENABLE_UT
#define static
#endif
static void dw_empty_buftag_page_old(uint16 org_start, uint16 max_idx)
{
errno_t rc;
uint16 batch = org_start / SINGLE_BLOCK_TAG_NUM;
uint16 batch_begin = org_start;
uint32 start_offset;
uint32 set_size;
uint16 max_batch_num = DW_SINGLE_DIRTY_PAGE_NUM / SINGLE_BLOCK_TAG_NUM;
knl_g_dw_context* dw_single_cxt = &g_instance.dw_single_cxt;
MemoryContext old_mem_cxt = MemoryContextSwitchTo(dw_single_cxt->mem_cxt);
char *unaligned_buf = (char *)palloc0(BLCKSZ + BLCKSZ);
char *buf = (char *)TYPEALIGN(BLCKSZ, unaligned_buf);
for (int i = org_start; i < max_idx - org_start; i++) {
if (i >= SINGLE_BLOCK_TAG_NUM * (batch + 1)) {
uint32 tag_offset = (batch + 1) * BLCKSZ;
dw_pread_file(dw_single_cxt->fd, buf, BLCKSZ, tag_offset);
start_offset = (batch_begin % SINGLE_BLOCK_TAG_NUM) * sizeof(dw_single_flush_item);
set_size = (i - batch_begin) * sizeof(dw_single_flush_item);
rc = memset_s(buf + start_offset, set_size, 0, set_size);
securec_check(rc, "\0", "\0");
dw_pwrite_file(dw_single_cxt->fd, buf, BLCKSZ, tag_offset, SINGLE_DW_FILE_NAME);
batch++;
batch_begin = i;
}
if (batch >= max_batch_num - 1) {
uint32 tag_offset = (batch + 1) * BLCKSZ;
dw_pread_file(dw_single_cxt->fd, buf, BLCKSZ, tag_offset);
start_offset = (batch_begin % SINGLE_BLOCK_TAG_NUM) * sizeof(dw_single_flush_item);
set_size = (max_idx - batch_begin) * sizeof(dw_single_flush_item);
rc = memset_s(buf + start_offset, set_size, 0, set_size);
securec_check(rc, "\0", "\0");
dw_pwrite_file(dw_single_cxt->fd, buf, BLCKSZ, tag_offset, SINGLE_DW_FILE_NAME);
break;
}
}
pfree(unaligned_buf);
MemoryContextSwitchTo(old_mem_cxt);
return;
}
static uint16 get_max_single_write_pos_old()
{
uint16 max_idx = 0;
uint16 i = 0;
knl_g_dw_context* single_cxt = &g_instance.dw_single_cxt;
dw_file_head_t *file_head = single_cxt->recovery_buf.file_head;
for (i = file_head->start; i < single_cxt->recovery_buf.write_pos; i++) {
if (single_cxt->recovery_buf.single_flush_state[i] == false) {
break;
}
}
max_idx = i;
return max_idx;
}
void wait_all_single_dw_finish_flush_old()
{
knl_g_dw_context* dw_single_cxt = &g_instance.dw_single_cxt;
dw_file_head_t *file_head = dw_single_cxt->recovery_buf.file_head;
for (int i = file_head->start; i < dw_single_cxt->recovery_buf.write_pos; ) {
if (dw_single_cxt->recovery_buf.single_flush_state[i]) {
i++;
continue;
} else {
(void)sched_yield();
}
}
return;
}
bool dw_single_file_recycle_old(bool trunc_file)
{
bool file_full = false;
uint16 max_idx = 0;
knl_g_dw_context* single_cxt = &g_instance.dw_single_cxt;
dw_file_head_t *file_head = single_cxt->recovery_buf.file_head;
volatile uint16 org_start = file_head->start;
volatile uint16 org_dwn = file_head->head.dwn;
uint16 end = single_cxt->recovery_buf.write_pos;
errno_t rc;
Assert(g_instance.dw_single_cxt.dw_version == 0);
if (file_head->start != 0) {
if (end < file_head->start) {
file_full = end + 1 >= file_head->start;
} else {
file_full = false;
}
} else {
file_full = end + 1 >= DW_SINGLE_DIRTY_PAGE_NUM;
}
if (!file_full && !trunc_file) {
return true;
}
if (trunc_file) {
max_idx = get_max_single_write_pos_old();
if (max_idx == file_head->start) {
LWLockRelease(single_cxt->second_flush_lock);
return false;
}
LWLockRelease(single_cxt->second_flush_lock);
} else {
wait_all_single_dw_finish_flush_old();
file_head->start = 0;
single_cxt->recovery_buf.write_pos = 0;
rc = memset_s(single_cxt->recovery_buf.single_flush_state, sizeof(bool) * DW_SINGLE_DIRTY_PAGE_NUM,
0, sizeof(bool) * DW_SINGLE_DIRTY_PAGE_NUM);
securec_check(rc, "\0", "\0");
}
PageWriterSync();
if (trunc_file) {
if (!LWLockConditionalAcquire(single_cxt->second_flush_lock, LW_EXCLUSIVE)) {
ereport(LOG, (errmodule(MOD_DW),
errmsg("[single flush] can not get dw flush lock and skip dw truncate after sync for this time")));
return false;
} else if (pg_atomic_read_u32(&g_instance.dw_single_cxt.dw_version) != 0 ||
org_start != file_head->start || org_dwn != file_head->head.dwn) {
ereport(DW_LOG_LEVEL, (errmodule(MOD_DW),
errmsg("[single flush] Skip dw truncate after sync due to concurrent dw truncate/reset, or dw upgrade "
"original[dwn %hu, start %hu], current[dwn %hu, start %hu]",
org_dwn, org_start, file_head->head.dwn, file_head->start)));
return true;
}
}
* if truncate file and flush_page is not 0, the dwn can not plus,
* otherwise verify will failed when recovery the data form dw file.
*/
if (trunc_file) {
dw_empty_buftag_page_old(org_start, max_idx);
file_head->start = max_idx;
dw_prepare_file_head((char *)file_head, file_head->start, file_head->head.dwn);
} else {
dw_prepare_file_head((char *)file_head, file_head->start, file_head->head.dwn + 1);
}
Assert(file_head->head.dwn == file_head->tail.dwn);
dw_pwrite_file(single_cxt->fd, file_head, BLCKSZ, 0, SINGLE_DW_FILE_NAME);
ereport(LOG, (errmodule(MOD_DW), errmsg("Reset single flush DW file: file_head[dwn %hu, start %hu], "
"writer pos is %hu", file_head->head.dwn, file_head->start, single_cxt->write_pos)));
if (trunc_file) {
(void)pg_atomic_add_fetch_u64(&single_cxt->single_stat_info.file_trunc_num, 1);
} else {
(void)pg_atomic_add_fetch_u64(&single_cxt->single_stat_info.file_reset_num, 1);
}
return true;
}
void dw_single_old_file_truncate()
{
knl_g_dw_context *cxt = &g_instance.dw_single_cxt;
* If we can grab dw flush lock, truncate dw file for faster recovery.
*
* Note: This is only for recovery optimization. we can not block on
* dw flush lock, because, if we are checkpointer, pagewriter may be
* waiting for us to finish smgrsync before it can do a full recycle of dw file.
*/
if (!LWLockConditionalAcquire(cxt->second_flush_lock, LW_EXCLUSIVE)) {
ereport(LOG, (errmodule(MOD_DW),
errmsg("[single flush] Can not get dw flush lock and skip dw truncate for this time")));
return;
}
if (pg_atomic_read_u32(&g_instance.dw_single_cxt.dw_version) == 0) {
if (dw_single_file_recycle_old(true)) {
LWLockRelease(cxt->second_flush_lock);
}
} else {
LWLockRelease(cxt->second_flush_lock);
}
}
uint16 dw_single_flush_internal_old(BufferTag tag, Block block, XLogRecPtr page_lsn,
BufferTag phy_tag, bool *dw_flush)
{
errno_t rc;
uint16 actual_pos;
uint32 page_write_offset;
uint32 tag_write_offset;
uint16 block_offset;
dw_single_flush_item item;
knl_g_dw_context* dw_single_cxt = &g_instance.dw_single_cxt;
dw_file_head_t *file_head = dw_single_cxt->recovery_buf.file_head;
char *buf = t_thrd.proc->dw_buf;
rc = memcpy_s(buf, BLCKSZ, block, BLCKSZ);
securec_check(rc, "\0", "\0");
XLogWaitFlush(page_lsn);
(void)LWLockAcquire(dw_single_cxt->second_flush_lock, LW_EXCLUSIVE);
if (pg_atomic_read_u32(&g_instance.dw_single_cxt.dw_version) == DW_SUPPORT_NEW_SINGLE_FLUSH) {
LWLockRelease(dw_single_cxt->second_flush_lock);
*dw_flush = false;
return 0;
}
Assert(g_instance.dw_single_cxt.dw_version == 0);
(void)dw_single_file_recycle_old(false);
actual_pos = dw_single_cxt->recovery_buf.write_pos;
dw_single_cxt->recovery_buf.write_pos = (dw_single_cxt->recovery_buf.write_pos + 1) % DW_SINGLE_DIRTY_PAGE_NUM;
page_write_offset = (actual_pos + DW_SINGLE_BUFTAG_PAGE_NUM + 1) * BLCKSZ;
item.data_page_idx = actual_pos;
item.dwn = file_head->head.dwn;
item.buf_tag = phy_tag;
dw_set_pg_checksum(buf, item.buf_tag.blockNum);
INIT_CRC32C(item.crc);
COMP_CRC32C(item.crc, (char*)&item, offsetof(dw_single_flush_item, crc));
FIN_CRC32C(item.crc);
dw_pwrite_file(dw_single_cxt->fd, buf, BLCKSZ, page_write_offset, SINGLE_DW_FILE_NAME);
(void)pg_atomic_add_fetch_u64(&dw_single_cxt->single_stat_info.total_writes, 1);
tag_write_offset = BLCKSZ + (actual_pos / SINGLE_BLOCK_TAG_NUM) * BLCKSZ;
block_offset = (actual_pos % SINGLE_BLOCK_TAG_NUM) * sizeof(dw_single_flush_item);
Assert(block_offset <= BLCKSZ - sizeof(dw_single_flush_item));
dw_pread_file(dw_single_cxt->fd, buf, BLCKSZ, tag_write_offset);
rc = memcpy_s(buf + block_offset, BLCKSZ - block_offset, &item, sizeof(dw_single_flush_item));
securec_check(rc, "\0", "\0");
dw_pwrite_file(dw_single_cxt->fd, buf, BLCKSZ, tag_write_offset, SINGLE_DW_FILE_NAME);
LWLockRelease(dw_single_cxt->second_flush_lock);
(void)pg_atomic_add_fetch_u64(&dw_single_cxt->single_stat_info.total_writes, 1);
*dw_flush = true;
return actual_pos;
}
* Current, this function is only used by segment-page, which does not support TDE. If we want to support TDE,
* we could add additional parameter.
*/
uint16 seg_dw_single_flush_without_buffer(BufferTag tag, Block block, bool* flush_old_file)
{
bool dw_flush = false;
uint16 pos = 0;
if (pg_atomic_read_u32(&g_instance.dw_single_cxt.dw_version) == 0) {
pos = dw_single_flush_internal_old(tag, block, PageGetLSN(block), tag, &dw_flush);
*flush_old_file = true;
}
if (!dw_flush) {
*flush_old_file = false;
pos = second_version_dw_single_flush(tag, block, PageGetLSN(block), false, tag);
}
return pos;
}
uint16 seg_dw_single_flush(BufferDesc *buf_desc, bool* flush_old_file)
{
bool dw_flush = false;
uint16 pos = 0;
BufferTag phy_tag = buf_desc->tag;
Block block = BufHdrGetBlock(buf_desc);
uint64 buf_state = LockBufHdr(buf_desc);
XLogRecPtr page_lsn = BufferGetLSN(buf_desc);
UnlockBufHdr(buf_desc, buf_state);
dw_transfer_phybuffer_addr(buf_desc, &phy_tag);
if (pg_atomic_read_u32(&g_instance.dw_single_cxt.dw_version) == 0) {
pos = dw_single_flush_internal_old(buf_desc->tag, block, page_lsn, phy_tag, &dw_flush);
*flush_old_file = true;
}
if (!dw_flush) {
*flush_old_file = false;
pos = second_version_dw_single_flush(buf_desc->tag, block, page_lsn, buf_desc->extra->encrypt, phy_tag);
}
return pos;
}