optimize recovery

Author: trdtnguyen

build_mysql.sh

@@ -2,8 +2,8 @@
 MYSQL_HOME=/home/vldb/mysql-plnvm
 
 #debug mode
-#IS_DEBUG=0
-IS_DEBUG=1
+IS_DEBUG=0
+#IS_DEBUG=1
 
 #BUILD_NAME="-DUNIV_TRACE_RECOVERY_TIME"
 #BUILD_NAME="-DUNIV_TRACE_FLUSH_TIME -DUNIV_SKIPLOG"

storage/innobase/buf/buf0flu.cc

@@ -65,6 +65,8 @@ static const int buf_flush_page_cleaner_priority = -20;
 extern ulint gb_flush_time;
 #endif
 #if defined(UNIV_PMEMOBJ_BUF) || defined (UNIV_PMEMOBJ_PART_PL)
+#include <sys/syscall.h>
+#include <sys/types.h> //for gettid()
 #include "my_pmemobj.h"
 #include <libpmemobj.h>
 extern PMEM_WRAPPER* gb_pmw;
@@ -4337,6 +4339,13 @@ DECLARE_THREAD(pm_log_redoer_worker)(
 			os_thread_create */
 {
 	ulint i;
+	pid_t thread_id;
+	ulint idx;
+	ulint lines_per_thread;
+	//int dist_mode = 2;
+	int dist_mode = 1;
+
+	ulint start_time, end_time, e_time;
 
 	PMEM_LOG_REDOER* redoer = gb_pmw->ppl->redoer;
 
@@ -4346,9 +4355,18 @@ DECLARE_THREAD(pm_log_redoer_worker)(
 	my_thread_init();
 
 	mutex_enter(&redoer->mutex);
+	idx = redoer->n_workers;
 	redoer->n_workers++;
 	os_event_reset(redoer->is_log_all_closed);
 	mutex_exit(&redoer->mutex);
+	
+	//thread_id = os_thread_pf(os_thread_get_curr_id());
+	lines_per_thread = redoer->size / (srv_ppl_n_redoer_threads - 1);
+
+	//thread_id = syscall(SYS_gettid);
+	//idx = thread_id % srv_ppl_n_redoer_threads;
+
+	printf("Redoers thread %zu idx %zu created\n",thread_id, idx);
 
 	while (true) {
 		//worker thread wait until there is is_requested signal 
@@ -4361,23 +4379,44 @@ DECLARE_THREAD(pm_log_redoer_worker)(
 			//do nothing
 			break;
 		}
-
-		for (i = 0; i < redoer->size; i++) {
-			mutex_enter(&redoer->mutex);
+		/*Method 1: sequential distribute*/
+		//for (i = 0; i < redoer->size; i++) 
+		/*Method 2: segment distribute*/
+		for (i = idx * lines_per_thread;
+			   	i < (idx + 1) * lines_per_thread &&
+			   	i < redoer->size
+				; i++) 
+		/*Method 3: evently distribute*/
+		//for (i = idx ;
+		//	   	i < redoer->size
+		//		; i+= srv_ppl_n_redoer_threads) 
+		{
+			if (dist_mode ==1)
+				mutex_enter(&redoer->mutex);
 
 			pline = redoer->hashed_line_arr[i];
 
 			if (pline != NULL && !pline->is_redoing)
 			{
 				pline->is_redoing = true;
+				recv_line = pline->recv_line;
 				//do not hold the mutex during REDOing
-				mutex_exit(&redoer->mutex);
+				if (dist_mode ==1)
+					mutex_exit(&redoer->mutex);
 
 				/***this call REDOing for a line ***/
 				if (redoer->phase == PMEM_REDO_PHASE1){
 					//printf("PMEM_REDO: start REDO_PHASE1 (scan and parse) line %zu ...\n", pline->hashed_id);
 
+
+					//start_time = ut_time_us(NULL);
 					bool is_err = pm_ppl_redo_line(gb_pmw->pop, gb_pmw->ppl, pline);
+					//end_time = ut_time_us(NULL);
+
+					//recv_line->redo1_thread_id = idx; 	
+					//recv_line->redo1_start_time = start_time;
+					//recv_line->redo1_end_time = end_time;
+					//recv_line->redo1_elapse_time = (end_time - start_time);
 
 					if (is_err){
 						printf("PMEM_REDO: error redoing line %zu \n", pline->hashed_id);
@@ -4389,27 +4428,37 @@ DECLARE_THREAD(pm_log_redoer_worker)(
 #if defined (UNIV_PMEMOBJ_PART_PL_DEBUG)
 					printf("PMEM_REDO: start REDO_PHASE2 (applying) line %zu ...\n", pline->hashed_id);
 #endif
+					//start_time = ut_time_us(NULL);
 					pm_ppl_recv_apply_hashed_line(
 							gb_pmw->pop, gb_pmw->ppl,
 							pline, pline->recv_line->is_ibuf_avail);
+					//end_time = ut_time_us(NULL);
 
+					//recv_line->redo2_thread_id = idx; 	
+					//recv_line->redo2_start_time = start_time;
+					//recv_line->redo2_end_time = end_time;
+					//recv_line->redo2_elapse_time = (end_time - start_time);
 #if defined (UNIV_PMEMOBJ_PART_PL_DEBUG)
 					printf("PMEM_REDO: end REDO_PHASE2 (applying) line %zu\n", pline->hashed_id);
 #endif
 				}
 
-				mutex_enter(&redoer->mutex);
+				if (dist_mode ==1)
+					mutex_enter(&redoer->mutex);
+
 				redoer->hashed_line_arr[i] = NULL;
 				//redoer->n_requested--;
 				redoer->n_remains--;
 
 				if (redoer->n_remains == 0){
 					//this is the last REDO
-					mutex_exit(&redoer->mutex);
+					if (dist_mode ==1)
+						mutex_exit(&redoer->mutex);
 					break;
 				}
 			}
-			mutex_exit(&redoer->mutex);
+			if (dist_mode ==1)
+				mutex_exit(&redoer->mutex);
 		} //end for
 
 		// after this for loop, all lines are either done REDO or REDOing by other threads, this thread has nothing to do
@@ -4419,7 +4468,8 @@ DECLARE_THREAD(pm_log_redoer_worker)(
 	mutex_enter(&redoer->mutex);
 	redoer->n_workers--;
 	if (redoer->n_workers == 0) {
-		printf("The last log redoer is closing\n");
+		printf("The last log redoer is closing. Redo phase %zu redoer->n_remains %zu ppl->n_redoing_lines %zu\n",
+				redoer->phase, redoer->n_remains, gb_pmw->ppl->n_redoing_lines);
 		//trigger the coordinator (the pm_ppl_redo) to wakeup
 		os_event_set(redoer->is_log_all_finished);
 	}

storage/innobase/handler/ha_innodb.cc

@@ -19691,7 +19691,7 @@ static MYSQL_SYSVAR_DOUBLE(ppl_log_buf_flush_pct, srv_ppl_log_buf_flush_pct,
 static MYSQL_SYSVAR_DOUBLE(ppl_ckpt_threshold, srv_ppl_ckpt_threshold,
   PLUGIN_VAR_RQCMDARG | PLUGIN_VAR_READONLY,
   "Percentage of log file fill to trigger the checkpoint, default is 0.7",
-  NULL, NULL, 0.7, 0.1, 1, 0);
+  NULL, NULL, 0.7, 0.05, 1000, 0);
 
 static MYSQL_SYSVAR_ULONG(ppl_log_flusher_wake_threshold, srv_ppl_log_flusher_wake_threshold,
   PLUGIN_VAR_RQCMDARG | PLUGIN_VAR_READONLY,

storage/innobase/include/my_pmemobj.h

@@ -479,7 +479,8 @@ struct __pmem_dpt {
  * */
 struct __pmem_page_part_log {
 	//we only use this lock to protect below values
-	PMEMrwlock		lock;
+	PMEMrwlock		ckpt_lock;
+	PMEMrwlock		recv_lock;
 	uint64_t	max_oldest_lsn;
 	uint64_t	min_oldest_lsn;
 	uint64_t	ckpt_lsn;
@@ -577,6 +578,8 @@ struct __pmem_page_log_hashed_line {
 	TOID_ARRAY(TOID(PMEM_PAGE_LOG_BLOCK)) arr;
 	uint64_t n_blocks; //the current non-free blocks
 	uint64_t max_blocks; //the total log block in bucket
+	long long* bit_arr; //bit array to manage free slots
+	uint16_t n_bit_blocks; //number of block in bit_arr
 
 	/*Hash table*/
 	hash_table_t* addr_hash; //hash the log block in this line
@@ -641,6 +644,7 @@ struct __pmem_page_log_block {
 
 	bool			is_free; //flag
 	uint64_t		bid; //block id
+	uint32_t		id; //id on the line
 	uint64_t		key; //fold id
 
 	int32_t			count; //number of active tx
@@ -693,9 +697,24 @@ struct __pmem_recv_line {
 	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 */
+	ulint			n_addrs;/*!< number of not processed pages in the hash table */
+	ulint			n_addrs_done;/*!< number of not processed pages in the hash table */
+	ulint			n_read_reqs; /*number of read request in phase 2 REDO */
+	ulint			n_read_done; /*number of read request in phase 2 REDO */
+
 	recv_dblwr_t	dblwr;
 	encryption_list_t* encryption_list;
+
+	/*statistic info*/
+	ulint		redo1_thread_id;
+	ulint		redo1_start_time;
+	ulint		redo1_end_time;
+	ulint		redo1_elapse_time;
+
+	ulint		redo2_thread_id;
+	ulint		redo2_start_time;
+	ulint		redo2_end_time;
+	ulint		redo2_elapse_time;
 };
 
 ////////////////     FLUSHER         /////////////////
@@ -1093,6 +1112,15 @@ pm_ppl_hash_get(
 
 plog_hash_t*
 pm_ppl_hash_add(
+		PMEMobjpool*		pop,
+		PMEM_PAGE_PART_LOG*		ppl,
+		PMEM_PAGE_LOG_HASHED_LINE* pline,
+		PMEM_PAGE_LOG_BLOCK*	plog_block,
+		uint32_t				idx
+		);
+
+plog_hash_t*
+pm_ppl_hash_check_and_add(
 		PMEMobjpool*		pop,
 		PMEM_PAGE_PART_LOG*		ppl,
 		PMEM_PAGE_LOG_HASHED_LINE* pline,
@@ -1347,7 +1375,7 @@ uint64_t
 pm_ppl_recv_parse_log_rec(
 	PMEMobjpool*		pop,
 	PMEM_PAGE_PART_LOG*	ppl,
-	PMEM_RECV_LINE* recv_line,
+	PMEM_PAGE_LOG_HASHED_LINE* pline,
 	mlog_id_t*	type,
 	byte*		ptr,
 	byte*		end_ptr,
@@ -1493,6 +1521,32 @@ void
 pm_ppl_remove_fil_spaces();
 
 ////////////// UTILITY////////////
+
+//////////// BIT ARRAY /////////
+void
+pm_bit_set(
+		long long*	arr,
+		size_t		block_size,
+		uint64_t	bit_i);
+
+void
+pm_bit_clear(
+		long long*	arr,
+		size_t		block_size,
+		uint64_t	bit_i);
+
+int32_t 
+pm_search_first_free_slot(
+		long long*	bit_arr,
+		uint16_t	n_bit_blocks,
+		uint16_t	block_size);
+
+#define PM_BIT_SET(A, bs, i) ( A[i / bs] |= 1 << (i % bs) )
+
+#define PM_BIT_SET(A, bs, i) ( A[i / bs] &= ~(1 << (i % bs)) )
+
+/////////// END BIT ARRAY //////
+
 int64_t
 __update_tt_entry_on_write_log(
 		PMEMobjpool*		pop,

storage/innobase/log/log0log.cc

@@ -2113,9 +2113,9 @@ pm_ppl_checkpoint(
 	//}
 
 	/* (5) update the global ckpt_lsn*/
-	pmemobj_rwlock_wrlock(pop, &ppl->lock);
+	pmemobj_rwlock_wrlock(pop, &ppl->ckpt_lock);
 	ppl->ckpt_lsn = new_oldest;
-	pmemobj_rwlock_unlock(pop, &ppl->lock);
+	pmemobj_rwlock_unlock(pop, &ppl->ckpt_lock);
 }
 #endif //UNIV_PMEMOBJ_PART_PL