!3041 社区dss/dms commit点更新

Merge pull request !3041 from 陈志凯/5.0.0

src/gausskernel/ddes/adapter/ss_dms_bufmgr.cpp

@@ -30,6 +30,7 @@
 #include "ddes/dms/ss_dms_bufmgr.h"
 #include "securec_check.h"
 #include "miscadmin.h"
+#include "access/double_write.h"
 
 void InitDmsBufCtrl(void)
 {
@@ -765,3 +766,98 @@ bool DmsCheckBufAccessible()
     }
     return false;
 }
+
+bool SSTryFlushBuffer(BufferDesc *buf)
+{
+    //copy from BufferAlloc
+    if (!backend_can_flush_dirty_page()) {
+        return false;
+    }
+
+    if (LWLockConditionalAcquire(buf->content_lock, LW_SHARED)) {
+        if (dw_enabled() && pg_atomic_read_u32(&g_instance.ckpt_cxt_ctl->current_page_writer_count) > 0) {
+            if (!free_space_enough(buf->buf_id)) {
+                LWLockRelease(buf->content_lock);
+                return false;
+            }
+            uint32 pos = 0;
+            pos = first_version_dw_single_flush(buf);
+            t_thrd.proc->dw_pos = pos;
+            FlushBuffer(buf, NULL);
+            g_instance.dw_single_cxt.single_flush_state[pos] = true;
+            t_thrd.proc->dw_pos = -1;
+        } else {
+            FlushBuffer(buf, NULL);
+        }
+        LWLockRelease(buf->content_lock);
+        ScheduleBufferTagForWriteback(t_thrd.storage_cxt.BackendWritebackContext, &buf->tag);
+        return true;    
+    }
+    return false;
+}
+
+bool SSTrySegFlushBuffer(BufferDesc* buf)
+{
+    //copy from SegBufferAlloc
+    if (!backend_can_flush_dirty_page()) {
+        return false;
+    }
+
+    if (LWLockConditionalAcquire(buf->content_lock, LW_SHARED)) {
+        FlushOneSegmentBuffer(buf->buf_id + 1);
+        LWLockRelease(buf->content_lock);
+        ScheduleBufferTagForWriteback(t_thrd.storage_cxt.BackendWritebackContext, &buf->tag);
+        return true;
+    } 
+    return false;
+}
+
+/** true :1)this buffer dms think need flush, and flush success
+*         2) no need flush 
+*   false: this flush dms think need flush, but cannot flush 
+*/ 
+bool SSHelpFlushBufferIfNeed(BufferDesc* buf_desc)
+{
+    if (!ENABLE_DMS) {
+        return true;
+    }
+
+    if (IsInitdb) {
+        return true;
+    }
+
+    dms_buf_ctrl_t *buf_ctrl = GetDmsBufCtrl(buf_desc->buf_id);
+    if (buf_ctrl->state & BUF_DIRTY_NEED_FLUSH) {
+        // wait dw_init finish
+        while (!g_instance.dms_cxt.dw_init) {
+            pg_usleep(1000L);
+        }
+
+        XLogRecPtr pagelsn = BufferGetLSN(buf_desc);
+        if (!SS_IN_REFORM) {
+            ereport(PANIC,
+                (errmsg("[SS] this buffer should not exist with BUF_DIRTY_NEED_FLUSH but not in reform, "
+                "spc/db/rel/bucket fork-block: %u/%u/%u/%d %d-%u, page lsn (0x%llx), seg info:%u-%u",
+                buf_desc->tag.rnode.spcNode, buf_desc->tag.rnode.dbNode, buf_desc->tag.rnode.relNode, 
+                buf_desc->tag.rnode.bucketNode, buf_desc->tag.forkNum, buf_desc->tag.blockNum, 
+                (unsigned long long)pagelsn, (unsigned int)buf_desc->extra->seg_fileno,
+                buf_desc->extra->seg_blockno)));
+        }
+        bool in_flush_copy = SS_IN_FLUSHCOPY;
+        bool in_recovery = !g_instance.dms_cxt.SSRecoveryInfo.recovery_pause_flag;
+        ereport(LOG,
+            (errmsg("[SS flush copy] ready to flush buffer with need flush, "
+            "spc/db/rel/bucket fork-block: %u/%u/%u/%d %d-%u, page lsn (0x%llx), seg info:%u-%u, reform phase "
+            "is in flush_copy:%d, in recovery:%d",
+            buf_desc->tag.rnode.spcNode, buf_desc->tag.rnode.dbNode, buf_desc->tag.rnode.relNode, 
+            buf_desc->tag.rnode.bucketNode, buf_desc->tag.forkNum, buf_desc->tag.blockNum, 
+            (unsigned long long)pagelsn, (unsigned int)buf_desc->extra->seg_fileno, buf_desc->extra->seg_blockno,
+            in_flush_copy, in_recovery)));
+        if (IsSegmentBufferID(buf_desc->buf_id)) {
+            return SSTrySegFlushBuffer(buf_desc);
+        } else {
+            return SSTryFlushBuffer(buf_desc);
+        }
+    }
+    return true;
+}

src/gausskernel/ddes/adapter/ss_dms_callback.cpp

@@ -1283,6 +1283,11 @@ static int CBFlushCopy(void *db_handle, char *pageid)
         smgrcloseall();
     }
 
+    // only 1) primary restart 2) failover need flush_copy
+    if (SS_REFORM_REFORMER && g_instance.dms_cxt.dms_status == DMS_STATUS_IN && !SS_STANDBY_FAILOVER) {
+        return GS_SUCCESS;
+    }
+
     BufferTag* tag = (BufferTag*)pageid;
     Buffer buffer;
     SegSpace *spc = NULL;
@@ -1304,7 +1309,7 @@ static int CBFlushCopy(void *db_handle, char *pageid)
         ErrorData* edata = CopyErrorData();
         FlushErrorState();
         FreeErrorData(edata);
-        ereport(PANIC, (errmsg("[SS Flush Copy] Error happend, spc/db/rel/bucket fork-block: %u/%u/%u/%d %d-%u",
+        ereport(PANIC, (errmsg("[SS flush copy] Error happend, spc/db/rel/bucket fork-block: %u/%u/%u/%d %d-%u",
                         tag->rnode.spcNode, tag->rnode.dbNode, tag->rnode.relNode, tag->rnode.bucketNode,
                         tag->forkNum, tag->blockNum)));
     }
@@ -1417,6 +1422,7 @@ static void CBReformStartNotify(void *db_handle, dms_role_t role, unsigned char
     if (ss_reform_type == DMS_REFORM_TYPE_FOR_FAILOVER_OPENGAUSS) {
         g_instance.dms_cxt.SSRecoveryInfo.in_failover = true;
         if (role == DMS_ROLE_REFORMER) {
+            g_instance.dms_cxt.dw_init = false;
             // variable set order: SharedRecoveryInProgress -> failover_triggered -> dms_role
             volatile XLogCtlData *xlogctl = t_thrd.shemem_ptr_cxt.XLogCtl;
             SpinLockAcquire(&xlogctl->info_lck);

src/gausskernel/ddes/adapter/ss_dms_recovery.cpp

@@ -327,7 +327,6 @@ void ss_failover_dw_init()
         }
     }
     ckpt_shutdown_pagewriter();
-    g_instance.dms_cxt.SSRecoveryInfo.in_flushcopy = false;
     ss_failover_dw_init_internal();
     g_instance.dms_cxt.dw_init = true;
 }
@@ -346,5 +345,6 @@ void ss_switchover_promoting_dw_init()
     dw_exit(false);
     dw_ext_init();
     dw_init();
+    g_instance.dms_cxt.dw_init = true;
     ereport(LOG, (errmodule(MOD_DMS), errmsg("[SS switchover] dw init finished")));
 }

src/gausskernel/ddes/ddes_commit_id

@@ -1,2 +1,2 @@
-dms_commit_id=4dad7c5286ed4a1975f5e1e101b0c533cbb00880
-dss_commit_id=38705ea061f9f701c5dd930518edfb320c8bccce
+dms_commit_id=4680252c5c93314e0a8c8a8cabac5365b6ec2f66
+dss_commit_id=e3e57f04172da967232eb5ec593e5576ee1563a4

src/gausskernel/ddes/script/dss_contrl.sh

@@ -362,10 +362,10 @@ function Reg()
         log "can't find inst id. Aborting."
         exit 1
     fi
-    dsscmd reghl -i ${LOCAL_INSTANCE_ID} -D ${DSS_HOME} >> /dev/null 2>&1
+    dsscmd reghl -D ${DSS_HOME} >> /dev/null 2>&1
     if [[ $? != 0 ]]
     then
-        log "dsscmd reghl -i ${LOCAL_INSTANCE_ID} -D ${DSS_HOME} fail."
+        log "dsscmd reghl -D ${DSS_HOME} fail."
         exit 1
     fi
     log "register success."
@@ -381,7 +381,7 @@ function Unreg()
     fi
     if [[ ${LOCAL_INSTANCE_ID} == ${INSTANCE_ID} ]]
     then
-        dsscmd unreghl -i ${LOCAL_INSTANCE_ID} -D ${DSS_HOME} >> /dev/null 2>&1
+        dsscmd unreghl -D ${DSS_HOME} >> /dev/null 2>&1
     else
         pid=$(program_pid dssserver ${DSS_HOME})
         if [[ -z ${pid} ]]
@@ -394,7 +394,7 @@ function Unreg()
 
     if [[ $? != 0 ]]
     then
-        log "dsscmd kickh -i ${INSTANCE_ID} -D ${DSS_HOME} fail, or dsscmd unreghl -i ${LOCAL_INSTANCE_ID} -D ${DSS_HOME} fail."
+        log "dsscmd kickh -i ${INSTANCE_ID} -D ${DSS_HOME} fail, or dsscmd unreghl -D ${DSS_HOME} fail."
         exit 1
     fi
     log "unregister ${INSTANCE_ID} success."

src/gausskernel/storage/buffer/bufmgr.cpp

@@ -2746,7 +2746,7 @@ static BufferDesc *BufferAlloc(SMgrRelation smgr, char relpersistence, ForkNumbe
         /* Pin the buffer and then release the buffer spinlock */
         PinBuffer_Locked(buf);
 
-        if (!SSPageCheckIfCanEliminate(buf)) {
+        if (!SSHelpFlushBufferIfNeed(buf)) {
             // for dms this page cannot eliminate, get another one 
             UnpinBuffer(buf, true);
             continue;
@@ -6350,6 +6350,7 @@ void CheckIOState(volatile void *buf_desc)
 bool StartBufferIO(BufferDesc *buf, bool for_input)
 {
     uint32 buf_state;
+    bool dms_need_flush = false; // used in dms
 
     Assert(!t_thrd.storage_cxt.InProgressBuf);
 
@@ -6390,8 +6391,15 @@ bool StartBufferIO(BufferDesc *buf, bool for_input)
         WaitIO(buf);
     }
 
+    if (ENABLE_DMS) {
+        dms_buf_ctrl_t *buf_ctrl = GetDmsBufCtrl(buf->buf_id);
+        if (buf_ctrl->state & BUF_DIRTY_NEED_FLUSH) {
+            dms_need_flush = true;
+        }
+    }
+
     /* Once we get here, there is definitely no I/O active on this buffer */
-    if (for_input ? (buf_state & BM_VALID) : !(buf_state & BM_DIRTY)) {
+    if (for_input ? (buf_state & BM_VALID) : !(buf_state & BM_DIRTY) && !dms_need_flush) {
         /* someone else already did the I/O */
         UnlockBufHdr(buf, buf_state);
         LWLockRelease(buf->io_in_progress_lock);
@@ -6483,6 +6491,23 @@ static void TerminateBufferIO_common(BufferDesc *buf, bool clear_dirty, uint32 s
         if (ENABLE_INCRE_CKPT) {
             if (!XLogRecPtrIsInvalid(pg_atomic_read_u64(&buf->extra->rec_lsn))) {
                 remove_dirty_page_from_queue(buf);
+            } else if (ENABLE_DMS) {
+                dms_buf_ctrl_t *buf_ctrl = GetDmsBufCtrl(buf->buf_id);
+                if (!(buf_ctrl->state & BUF_DIRTY_NEED_FLUSH)) {
+                    ereport(PANIC, (errmodule(MOD_INCRE_CKPT), errcode(ERRCODE_INVALID_BUFFER),
+                        (errmsg("buffer is dirty but not in dirty page queue in TerminateBufferIO_common"))));
+                }
+                buf_ctrl->state &= ~BUF_DIRTY_NEED_FLUSH;
+                XLogRecPtr pagelsn = BufferGetLSN(buf);
+                bool in_flush_copy = SS_IN_FLUSHCOPY;
+                bool in_recovery = !g_instance.dms_cxt.SSRecoveryInfo.recovery_pause_flag;
+                ereport(LOG,
+                    (errmsg("[SS flush copy] finish flush buffer with need flush, "
+                    "spc/db/rel/bucket fork-block: %u/%u/%u/%d %d-%u, page lsn (0x%llx), seg info:%u-%u, reform phase "
+                    "is in flush_copy:%d, in recovery:%d",
+                    buf->tag.rnode.spcNode, buf->tag.rnode.dbNode, buf->tag.rnode.relNode, buf->tag.rnode.bucketNode,
+                    buf->tag.forkNum, buf->tag.blockNum, (unsigned long long)pagelsn, 
+                    (unsigned int)buf->extra->seg_fileno, buf->extra->seg_blockno, in_flush_copy, in_recovery)));
             } else {
                 ereport(PANIC, (errmodule(MOD_INCRE_CKPT), errcode(ERRCODE_INVALID_BUFFER),
                                 (errmsg("buffer is dirty but not in dirty page queue in TerminateBufferIO_common"))));

src/gausskernel/storage/smgr/segment/segbuffer.cpp

@@ -75,6 +75,7 @@ void AbortSegBufferIO(void)
 static bool SegStartBufferIO(BufferDesc *buf, bool forInput)
 {
     uint32 buf_state;
+    bool dms_need_flush = false; // used in dms
 
     SegmentCheck(!InProgressBuf);
 
@@ -98,7 +99,14 @@ static bool SegStartBufferIO(BufferDesc *buf, bool forInput)
         WaitIO(buf);
     }
 
-    if (forInput ? (buf_state & BM_VALID) : !(buf_state & BM_DIRTY)) {
+    if (ENABLE_DMS) {
+        dms_buf_ctrl_t *buf_ctrl = GetDmsBufCtrl(buf->buf_id);
+        if (buf_ctrl->state & BUF_DIRTY_NEED_FLUSH) {
+            dms_need_flush = true;
+        }
+    }
+
+    if (forInput ? (buf_state & BM_VALID) : !(buf_state & BM_DIRTY) && !dms_need_flush) {
         /* IO finished */
         UnlockBufHdr(buf, buf_state);
         LWLockRelease(buf->io_in_progress_lock);
@@ -128,6 +136,23 @@ void SegTerminateBufferIO(BufferDesc *buf, bool clear_dirty, uint32 set_flag_bit
         if (ENABLE_INCRE_CKPT) {
             if (!XLogRecPtrIsInvalid(pg_atomic_read_u64(&buf->extra->rec_lsn))) {
                 remove_dirty_page_from_queue(buf);
+            } else if (ENABLE_DMS) {
+                dms_buf_ctrl_t *buf_ctrl = GetDmsBufCtrl(buf->buf_id);
+                if (!(buf_ctrl->state & BUF_DIRTY_NEED_FLUSH)) {
+                    ereport(PANIC, (errmodule(MOD_INCRE_CKPT), errcode(ERRCODE_INVALID_BUFFER),
+                        (errmsg("buffer is dirty but not in dirty page queue in SegTerminateBufferIO"))));
+                }
+                buf_ctrl->state &= ~BUF_DIRTY_NEED_FLUSH;
+                XLogRecPtr pagelsn = BufferGetLSN(buf);
+                bool in_flush_copy = SS_IN_FLUSHCOPY;
+                bool in_recovery = !g_instance.dms_cxt.SSRecoveryInfo.recovery_pause_flag;
+                ereport(LOG,
+                    (errmsg("[SS flush copy] finish seg flush buffer with need flush, "
+                    "spc/db/rel/bucket fork-block: %u/%u/%u/%d %d-%u, page lsn (0x%llx), seg info:%u-%u, reform phase "
+                    "is in flush_copy:%d, in recovery:%d",
+                    buf->tag.rnode.spcNode, buf->tag.rnode.dbNode, buf->tag.rnode.relNode, buf->tag.rnode.bucketNode,
+                    buf->tag.forkNum, buf->tag.blockNum, (unsigned long long)pagelsn, 
+                    (unsigned int)buf->extra->seg_fileno, buf->extra->seg_blockno, in_flush_copy, in_recovery)));
             } else {
                 ereport(PANIC, (errmodule(MOD_INCRE_CKPT), errcode(ERRCODE_INVALID_BUFFER),
                                 (errmsg("buffer is dirty but not in dirty page queue in TerminateBufferIO_common"))));
@@ -704,7 +729,7 @@ BufferDesc *SegBufferAlloc(SegSpace *spc, RelFileNode rnode, ForkNumber forkNum,
 
         SegPinBufferLocked(buf, &new_tag);
 
-        if (!SSPageCheckIfCanEliminate(buf)) {
+        if (!SSHelpFlushBufferIfNeed(buf)) {
             SegUnpinBuffer(buf);
             continue;
         }

src/include/ddes/dms/dms_api.h

@@ -286,7 +286,13 @@ typedef struct st_dms_buf_ctrl {
     unsigned char seg_fileno;
     unsigned int seg_blockno;
 #endif
-}dms_buf_ctrl_t;
+} dms_buf_ctrl_t;
+
+typedef struct st_dms_ctrl_info {
+    dms_buf_ctrl_t      ctrl;
+    unsigned long long  lsn;
+    unsigned char       is_dirty;
+} dms_ctrl_info_t;
 
 typedef enum en_dms_page_latch_mode {
     DMS_PAGE_LATCH_MODE_S = 1,
@@ -301,7 +307,7 @@ typedef enum en_dms_page_latch_mode {
 #define DMS_ENTER_PAGE_TRY            (unsigned char)8    // try to read from buffer, don't read from disk
 #define DMS_ENTER_PAGE_LRU_STATS_SCAN (unsigned char)0x10 // add to stats LRU list
 #define DMS_ENTER_PAGE_LRU_HIGH_AGE   (unsigned char)0x20 // decrease possibility to be recycled of page
-#define DMS_ENTER_PAGE_LOCAL          (unsigned char)0x40 // check local page without redo log
+#define DMS_ENTER_PAGE_LOCAL          (unsigned char)0x40 // check local page without redo log, use carefully
 #define DMS_ENTER_PAGE_REMOTE         (unsigned char)0x80 // remote access mode
 
 // pack read page parameters together
@@ -486,6 +492,7 @@ typedef int(*dms_confirm_converting)(void *db_handle, char *pageid, unsigned cha
 typedef int(*dms_confirm_owner)(void *db_handle, char *pageid, unsigned char *lock_mode, unsigned char *is_edp,
     unsigned long long *lsn);
 typedef int(*dms_flush_copy)(void *db_handle, char *pageid);
+typedef int(*dms_need_flush)(void *db_handle, char *pageid);
 typedef int(*dms_edp_lsn)(void *db_handle, char *pageid, unsigned long long *lsn);
 typedef int(*dms_disk_lsn)(void *db_handle, char *pageid, unsigned long long *lsn);
 typedef int(*dms_recovery)(void *db_handle, void *recovery_list, int is_reformer);
@@ -573,7 +580,8 @@ typedef char *(*dms_display_pageid)(char *display_buf, unsigned int count, char
 typedef char *(*dms_display_xid)(char *display_buf, unsigned int count, char *xid);
 typedef char *(*dms_display_rowid)(char *display_buf, unsigned int count, char *rowid);
 typedef int (*dms_drc_buf_res_rebuild)(void *db_handle);
-typedef int (*dms_drc_buf_res_rebuild_parallel)(void *db_handle, unsigned char thread_index, unsigned char thread_num);
+typedef int (*dms_drc_buf_res_rebuild_parallel)(void *db_handle, unsigned char thread_index, unsigned char thread_num,
+    unsigned char for_rebuild);
 typedef unsigned char(*dms_ckpt_session)(void *db_handle);
 typedef void (*dms_check_if_build_complete)(void *db_handle, unsigned int *build_complete);
 typedef int (*dms_db_is_primary)(void *db_handle);
@@ -622,6 +630,7 @@ typedef struct st_dms_callback {
     dms_confirm_owner confirm_owner;
     dms_confirm_converting confirm_converting;
     dms_flush_copy flush_copy;
+    dms_need_flush need_flush;
     dms_edp_lsn edp_lsn;
     dms_disk_lsn disk_lsn;
     dms_recovery recovery;
@@ -803,7 +812,7 @@ typedef struct st_logger_param {
 #define DMS_LOCAL_MINOR_VER_WEIGHT  1000
 #define DMS_LOCAL_MAJOR_VERSION     0
 #define DMS_LOCAL_MINOR_VERSION     0
-#define DMS_LOCAL_VERSION           52
+#define DMS_LOCAL_VERSION           54
 
 #ifdef __cplusplus
 }

src/include/ddes/dms/ss_dms_bufmgr.h

@@ -77,5 +77,6 @@ void BufValidateDrc(BufferDesc *buf_desc);
 bool SSPageCheckIfCanEliminate(BufferDesc* buf_desc);
 bool SSSegRead(SMgrRelation reln, ForkNumber forknum, char *buffer);
 bool DmsCheckBufAccessible();
+bool SSHelpFlushBufferIfNeed(BufferDesc* buf_desc);
 
 #endif