1. Add n_recs field to header of logbuf. 2. Handle multiple in-flushing logbuf by using doubled-linked list

Author: trdtnguyen

storage/innobase/btr/btr0cur.cc

@@ -350,6 +350,9 @@ btr_cur_latch_leaves(
 				page_id_t(page_id.space(), right_page_no),
 				page_size, RW_X_LATCH, cursor->index, mtr);
 			latch_leaves.blocks[2] = get_block;
+#if defined(UNIV_PMEMOBJ_PART_PL)
+			//PM_PPL skip those
+#else //original
 #ifdef UNIV_BTR_DEBUG
 			ut_a(page_is_comp(get_block->frame)
 			     == page_is_comp(page));
@@ -359,6 +362,8 @@ btr_cur_latch_leaves(
 			     == page_get_page_no(page));
 #endif //UNIV_PMEMOBJ_BUF
 #endif /* UNIV_BTR_DEBUG */
+
+#endif /*UNIV_PMEMOBJ_PART_PL*/
 			if (spatial) {
 				cursor->rtr_info->tree_blocks[
 					RTR_MAX_LEVELS + 2] = get_block;

storage/innobase/buf/buf0dblwr.cc

@@ -43,6 +43,11 @@ extern PMEM_WRAPPER* gb_pmw;
 extern pfs_os_file_t gb_dbw_file;
 #endif /* UNIV_PMEMOBJ_DBW */
 
+#if defined (UNIV_PMEMOBJ_PART_PL)
+#include "my_pmemobj.h"
+extern PMEM_WRAPPER* gb_pmw;
+#endif /*UNIV_PMEMOBJ_PART_PL*/
+
 #ifndef UNIV_HOTBACKUP
 
 /** The doublewrite buffer */
@@ -952,7 +957,21 @@ buf_dblwr_check_block(
 	if (block->skip_flush_check) {
 		return;
 	}
-
+//#if defined (UNIV_PMEMOBJ_PART_PL)
+//	//debug
+//	ulint type = fil_page_get_type(block->frame);
+//	PMEM_PAGE_LOG_BLOCK* plogblock;
+//	buf_page_t* bpage;
+//
+//	if (type == 0){
+//		plogblock = 
+//			pm_ppl_get_log_block_by_key(gb_pmw->pop, gb_pmw->ppl, block->page.id.fold());
+//
+//		bpage = (buf_page_t*) &(block->page);
+//
+//		assert(0);
+//	}
+//#endif //UNIV_PMEMOBJ_PART_PL
 	switch (fil_page_get_type(block->frame)) {
 	case FIL_PAGE_INDEX:
 	case FIL_PAGE_RTREE:

storage/innobase/buf/buf0flu.cc

@@ -810,23 +810,6 @@ buf_flush_write_complete(
 	buf_pool_t*	buf_pool = buf_pool_from_bpage(bpage);
 
 	ut_ad(bpage);
-#if defined (UNIV_PMEMOBJ_PART_PL)
-	//tdnguyen test
-	ulint page_no;
-	ulint space_id;
-	byte* frame = ((buf_block_t*) bpage)->frame;
-	page_no = mach_read_from_4(frame + FIL_PAGE_OFFSET);
-	space_id = mach_read_from_4(frame + FIL_PAGE_ARCH_LOG_NO_OR_SPACE_ID);
-
-	if ((bpage->id.space() != 0 
-			    && bpage->id.space() != space_id)
-			   || bpage->id.page_no() != page_no) {
-		printf("PMEM_ERROR in buf_flush_write_complete, input (space %zu, page_no %zu) differ write (space %zu, page_no %zu)\n", bpage->id.space(), bpage->id.page_no(), space_id, page_no);
-
-		assert(0);
-	}
-
-#endif
 	buf_flush_remove(bpage);
 
 	flush_type = buf_page_get_flush_type(bpage);
@@ -844,6 +827,18 @@ buf_flush_write_complete(
 #else //original
 	buf_dblwr_update(bpage, flush_type);
 #endif /* UNIV_PMEMOBJ_BUF */
+
+#if defined (UNIV_PMEMOBJ_PART_PL)
+	//we only call pm_ppl_flush_page when the flushed page is persist on storage
+	pm_ppl_flush_page(
+			gb_pmw->pop, gb_pmw->ppl,
+			bpage,
+			bpage->id.space(),
+			bpage->id.page_no(),
+			bpage->id.fold(),
+			bpage->newest_modification);
+
+#endif //UNIV_PMEMBOJ_PART_PL
 }
 #endif /* !UNIV_HOTBACKUP */
 
@@ -1136,27 +1131,9 @@ buf_flush_write_block_low(
 	//skip_pm_write:
 #endif /*UNIV_PMEMOBJ_BUF*/
 
-#if defined (UNIV_PMEMOBJ_PART_PL) 
-
-	ulint read_space_id = mach_read_from_4(frame + FIL_PAGE_ARCH_LOG_NO_OR_SPACE_ID);
-	ulint read_page_no = mach_read_from_4(frame + FIL_PAGE_OFFSET);
-
-	if(read_space_id != bpage->id.space() || read_page_no != bpage->id.page_no()){
-		printf("PMEM_ERROR before pm_ppl_flush_page(), input (space %zu, page_no %zu) differ read (space %zu, page_no %zu)\n", bpage->id.space(), bpage->id.page_no(), read_space_id, read_page_no);
-		assert(0);
-	}
-	else{
-		//printf("PMEM_INFO pm_ppl_flush_page(), input (space %zu, page_no %zu)\n", bpage->id.space(), bpage->id.page_no());
-	}
-
-		// PL-NVM without PB-NVM
-		pm_ppl_flush_page(
-				gb_pmw->pop, gb_pmw->ppl,
-			   	bpage->id.space(),
-			    bpage->id.page_no(),
-			    bpage->id.fold(),
-				bpage->newest_modification);
-#endif /*UNIV_PMEMOBJ_BUF*/
+#if defined (UNIV_PMEMOBJ_PART_PL)
+		pm_ppl_set_flush_state(gb_pmw->pop, gb_pmw->ppl, bpage);
+#endif /*UNIV_PMEMOBJ_PART_PL*/
 
 	/* Disable use of double-write buffer for temporary tablespace.
 	Given the nature and load of temporary tablespace doublewrite buffer

storage/innobase/include/my_pmem_common.h

@@ -64,7 +64,8 @@ static const size_t PMEM_GROUP_PARTITION_SIZE= 512;
 #define PMEM_MAX_LISTS_PER_BUCKET 2
 //#define PMEM_BUF_THRESHOLD 0.8
 
-#define PMEM_LOG_BUF_HEADER_SIZE 4
+//#define PMEM_LOG_BUF_HEADER_SIZE 4
+#define PMEM_LOG_BUF_HEADER_SIZE 8 /*4-byte real_len, 4-byte n_recs*/
 
 enum {
 	PMEM_READ = 1,
@@ -103,6 +104,12 @@ enum PMEM_REDO_PHASE{
 	PMEM_REDO_PHASE2 = 2,
 };
 
+enum PMEM_PARSE_RESULT {
+	PMEM_PARSE_NEED = 1,
+	PMEM_PARSE_BLOCK_NOT_EXISTED = 2,
+	PMEM_PARSE_LSN_OLD = 3,
+};
+
 enum PMEM_LOG_BLOCK_STATE {
     PMEM_FREE_LOG_BLOCK = 1, //the log block is free, a transaction can write its log records
     PMEM_COMMIT_LOG_BLOCK = 2, // the transaction is either commit or abort

storage/innobase/include/my_pmemobj.h

@@ -555,8 +555,8 @@ struct __pmem_page_log_hashed_line {
 	PMEMrwlock		lock;
 
 	int hashed_id;
-	TOID(PMEM_PAGE_LOG_BUF)	logbuf;	//pointer to current log buffer
-	TOID(PMEM_PAGE_LOG_BUF)	flush_logbuf;	//pointer to flushing log buffer
+	TOID(PMEM_PAGE_LOG_BUF)	logbuf;	//pointer to current log buffer (head)
+	TOID(PMEM_PAGE_LOG_BUF)	tail_logbuf;	//pointer to the oldest log buffer (tail)
 
 	uint64_t		diskaddr; //log file offset, update when flush log, reset when purging file
 	uint64_t		write_diskaddr; //diskaddr that log recs are durable write write_diskaddr < diskaddr
@@ -614,21 +614,29 @@ struct __pmem_page_log_buf {
 	PMEM_LOG_BUF_STATE		state;
 	uint64_t				size;
 	uint64_t				cur_off; //the current offset (0 - log buf size), reset when switching log_buf 
+	uint64_t				n_recs;
 
 	int						check; //for AIO
 
 	uint64_t				diskaddr; //write address assigned when the buffer is full
 
+	TOID(PMEM_PAGE_LOG_BUF) prev;
+	TOID(PMEM_PAGE_LOG_BUF) next;
+
 	//link with the list free pool
 	POBJ_LIST_ENTRY(PMEM_PAGE_LOG_BUF) list_entries;
 };
 
 /*
  * One log block per-page
- * Follow the implementation of PMEM_BUF_BLOCK
+ * Metadata of a page during forward processing
+ * Persistent after the system crash
  * */
 struct __pmem_page_log_block {
 	PMEMrwlock		lock;
+
+	PMEM_BLOCK_STATE state;
+
 	bool			is_free; //flag
 	uint64_t		bid; //block id
 	uint64_t		key; //fold id
@@ -637,10 +645,16 @@ struct __pmem_page_log_block {
 
 	/*LSN */
 	uint64_t		pageLSN; // pageLSN of the NVM-page
-	uint64_t		firstLSN; // LSN of the first log record
 	uint64_t		lastLSN; // LSN of the last log record
+
+	uint64_t		firstLSN; // LSN of the first log record
 	uint32_t		start_off;// offset of the first log rec of this page on log buffer/ disk
 	uint32_t		start_diskaddr;// diskaddr when the first log rec is written 
+
+	//this value is used for testing recovery
+	bool			first_rec_found;
+	uint32_t		first_rec_size;
+	mlog_id_t		first_rec_type;
 };
 
 /*
@@ -662,6 +676,7 @@ struct __pmem_recv_line {
 	ulint		scanned_checkpoint_no;
 
 	ulint		recovered_offset;
+	ulint		recovered_addr;
 	lsn_t		recovered_lsn;
 
 	bool		found_corrupt_log;
@@ -672,7 +687,7 @@ struct __pmem_recv_line {
 	lsn_t		mlog_checkpoint_lsn;
 
 	mem_heap_t*	heap;	/*!< memory heap of log records and file addresses */
-
+	ulint		alloc_hash_size; //allocated heap size
 	hash_table_t*	addr_hash;/*!< hash table of file addresses of pages */
 	ulint		n_addrs;/*!< number of not processed hashed file addresses in the hash table */
 	recv_dblwr_t	dblwr;
@@ -1072,12 +1087,6 @@ pm_ppl_hash_get(
 		PMEM_PAGE_LOG_HASHED_LINE* pline,
 		uint64_t			key	);
 
-
-void pm_ppl_hash_add_at_page_read(
-		PMEMobjpool*		pop,
-		PMEM_PAGE_PART_LOG*	ppl,
-		buf_page_t* bpage);
-
 plog_hash_t*
 pm_ppl_hash_add(
 		PMEMobjpool*		pop,
@@ -1247,10 +1256,17 @@ pm_ppl_commit(
 		uint64_t				tid,
 		uint64_t				eid);
 
+void
+pm_ppl_set_flush_state(
+		PMEMobjpool*		pop,
+		PMEM_PAGE_PART_LOG*	ppl,
+		buf_page_t*			bpage);
+
 void 
 pm_ppl_flush_page(
 		PMEMobjpool*		pop,
 		PMEM_PAGE_PART_LOG*	ppl,
+		buf_page_t*			bpage,
 		uint64_t			space,
 		uint64_t			page_no,
 		uint64_t			key,
@@ -1317,6 +1333,8 @@ pm_ppl_parse_recs(
 		PMEM_PAGE_LOG_HASHED_LINE* pline,
 		byte* recv_buf,
 		uint64_t len,
+		uint64_t* n_skip1_recs,
+		uint64_t* n_skip2_recs,
 		uint64_t* n_need_recs
 		);
 
@@ -1333,6 +1351,7 @@ pm_ppl_recv_parse_log_rec(
 	bool		apply,
 	uint64_t*   rec_lsn,
 	byte**		body,
+	PMEM_PARSE_RESULT*	parse_res,
 	bool*		is_need);