free bug implement DAL

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_FLUSH_TIME"
 #BUILD_NAME="-DUNIV_TRACE_FLUSH_TIME -DUNIV_SKIPLOG"

storage/innobase/buf/buf0buf.cc

@@ -5842,7 +5842,8 @@ buf_page_io_complete(
 		}
 
 #if defined (UNIV_PMEMOBJ_PART_PL)
-skip_recv_page:
+		//Add read page to hash table
+
 #endif
 		/* If space is being truncated then avoid ibuf operation.
 		During re-init we have already freed ibuf entries. */

storage/innobase/buf/buf0flu.cc

@@ -3565,7 +3565,6 @@ DECLARE_THREAD(buf_flush_page_cleaner_coordinator)(
 #if defined (UNIV_PMEMOBJ_PART_PL)
 	//wake up the sleeping threads to close them
 
-	os_event_set(gb_pmw->ppl->catcher->is_log_req_not_empty);
 	os_event_set(gb_pmw->ppl->flusher->is_log_req_not_empty);
 #endif
 	my_thread_end();
@@ -3986,11 +3985,6 @@ pm_log_flusher_init(
 	flusher->n_requested = 0;
 	flusher->is_running = false;
 
-	//partition array
-	flusher->part_list_arr = static_cast <PMEM_MINI_BUF**> (	calloc(size, sizeof(PMEM_MINI_BUF*)));
-	for (i = 0; i < size; i++) {
-		flusher->part_list_arr[i] = NULL;
-	}	
 
 	//flush array
 	flusher->flush_list_arr = static_cast <PMEM_PAGE_LOG_BUF**> (	calloc(size, sizeof(PMEM_PAGE_LOG_BUF*)));
@@ -4013,16 +4007,6 @@ pm_log_flusher_close(
 
 	switch (flusher->type){
 		case CATCHER_LOG_BUF:
-			//free partition list
-			for (i = 0; i < flusher->size; i++) {
-				if (flusher->part_list_arr[i]){
-					flusher->part_list_arr[i] = NULL;
-				}
-			}	
-			if (flusher->flush_list_arr){
-				free(flusher->flush_list_arr);
-				flusher->flush_list_arr = NULL;
-			}	
 			break;
 		case FLUSHER_LOG_BUF:
 		default:
@@ -4126,83 +4110,6 @@ DECLARE_THREAD(pm_log_flusher_coordinator)(
 	OS_THREAD_DUMMY_RETURN;
 }
 
-/*Worker thread of log catcher.
- * number of threads are equal to the number of cleaner threds from config
-@return a dummy parameter */
-extern "C"
-os_thread_ret_t
-DECLARE_THREAD(pm_log_catcher_worker)(
-/*==========================================*/
-	void*	arg MY_ATTRIBUTE((unused)))
-			/*!< in: a dummy parameter required by
-			os_thread_create */
-{
-	ulint i;
-
-	PMEM_LOG_FLUSHER* catcher = gb_pmw->ppl->catcher;
-
-	PMEM_MINI_BUF* pmbuf = NULL;
-
-	my_thread_init();
-
-	mutex_enter(&catcher->mutex);
-	catcher->n_workers++;
-	os_event_reset(catcher->is_log_all_closed);
-	mutex_exit(&catcher->mutex);
-
-	while (true) {
-		//worker thread wait until there is is_requested signal 
-retry:
-		os_event_wait(catcher->is_log_req_not_empty);
-		//looking for a full list in wait-list and flush it
-		mutex_enter(&catcher->mutex);
-		if (catcher->n_requested > 0) {
-			for (i = 0; i < catcher->size; i++) {
-				pmbuf = catcher->part_list_arr[i];
-				if (pmbuf != NULL)
-				{
-					/*call back function*/
-					//printf("\t [CATCHER] thread #%zu BEGIN partition catcher->size %zu catcher->n_reqs %zu pmbuf id %zu\n", i, catcher->size, catcher->n_requested, pmbuf->id);
-					pm_log_partition_mini_buf(gb_pmw->pop, gb_pmw->ppl, pmbuf);
-
-					catcher->n_requested--;
-					os_event_set(catcher->is_log_req_full);
-					//we can set the pointer to null after the pm_buf_flush_list finished
-					//printf("\t [CATCHER] thread #%zu END partition catcher->size %zu catcher->n_reqs %zu pmbuf id %zu\n", i, catcher->size, catcher->n_requested, pmbuf->id);
-					catcher->part_list_arr[i] = NULL;
-					break;
-				}
-			}
-
-		} //end if flusher->n_requested > 0
-
-		if (catcher->n_requested == 0) {
-			if (buf_page_cleaner_is_active) {
-				//buf_page_cleaner is running, start waiting
-				os_event_reset(catcher->is_log_req_not_empty);
-			}
-			else {
-				mutex_exit(&catcher->mutex);
-				break;
-			}
-		}
-		mutex_exit(&catcher->mutex);
-	} //end while thread
-
-	mutex_enter(&catcher->mutex);
-	catcher->n_workers--;
-	if (catcher->n_workers == 0) {
-		printf("The last catcher worker is closing\n");
-		//os_event_set(flusher->is_all_closed);
-	}
-	mutex_exit(&catcher->mutex);
-
-	my_thread_end();
-
-	os_thread_exit();
-
-	OS_THREAD_DUMMY_RETURN;
-}
 
 /*Worker thread of log flusher.
  * Managed by the coordinator thread

storage/innobase/buf/buf0rea.cc

@@ -40,7 +40,7 @@ Created 11/5/1995 Heikki Tuuri
 #include "srv0start.h"
 #include "srv0srv.h"
 
-#if defined (UNIV_PMEMOBJ_BUF)
+#if defined (UNIV_PMEMOBJ_BUF) || defined (UNIV_PMEMOBJ_PART_PL)
 #include "my_pmemobj.h"
 #include <libpmemobj.h>
 extern PMEM_WRAPPER* gb_pmw;
@@ -271,6 +271,12 @@ buf_read_page_low(
 	if (sync) {
 		/* The i/o is already completed when we arrive from
 		fil_read */
+#if defined (UNIV_PMEMOBJ_PART_PL)
+		pm_ppl_hash_add_at_page_read(
+				gb_pmw->pop,
+				gb_pmw->ppl,
+				bpage);
+#endif 
 		if (!buf_page_io_complete(bpage)) {
 			return(0);
 		}

storage/innobase/fil/fil0fil.cc

@@ -6864,6 +6864,12 @@ fil_aio_wait(
 			}
 #else //original
 			buf_page_io_complete(static_cast<buf_page_t*>(message));
+#if defined (UNIV_PMEMOBJ_PART_PL)
+			pm_ppl_hash_add_at_page_read(
+					gb_pmw->pop,
+					gb_pmw->ppl,
+					static_cast<buf_page_t*>(message));
+#endif //UNIV_PMEMOBJ_PART_PL
 #endif /* UNIV_PMEMOBJ_BUF*/
 		}
 		return;

storage/innobase/include/mtr0log.ic

@@ -200,6 +200,12 @@ mlog_write_initial_log_record_low(
 
 		mach_write_to_2(begin_log_ptr + prev_len_off, rec_size);
 		mtr->add_size_at(rec_size, n_recs - 1);
+
+		//new in DAL (Direct Add Log), add the previous log rec to PPL
+		uint64_t prev_key = mtr->get_key_at(n_recs - 1);
+		uint64_t rec_lsn = mtr->get_LSN_at(n_recs - 1);
+
+		mtr->add_rec_to_ppl(prev_key, begin_log_ptr + prev_off, rec_size, rec_lsn);
 	}
 	else {
 		//this is the first log rec in the mtr, do nothing
@@ -221,19 +227,20 @@ mlog_write_initial_log_record_low(
 	log_ptr += 2;
 
 	/*reserve 8-byte for LSN, we write it later when log recs are written to PPL partition log buffer */
-	//lsn = ut_time_us(NULL);
-	//mach_write_to_8(log_ptr, lsn);
+	lsn = ut_time_us(NULL);
+	mtr->add_LSN(lsn);
+	mach_write_to_8(log_ptr, lsn);
 	log_ptr += 8;
 
 	//add info 
 
 	//We compute the key as InnoDB 
 	key = (space_id << 20) + space_id + page_no;
 	mtr->add_key(key);
-	//mtr->add_LSN(lsn);
 	mtr->add_type(type);
 	mtr->add_space(space_id);
 	mtr->add_page(page_no);
+	
 
 	mtr->added_rec();
 	return(log_ptr);

storage/innobase/include/mtr0mtr.h

@@ -616,6 +616,12 @@ struct mtr_t {
 	}
 
 #if defined (UNIV_PMEMOBJ_PL)
+	void add_rec_to_ppl(
+			uint64_t key,
+		   	byte* src,
+		   	uint32_t size,
+			uint64_t rec_lsn);
+
 	byte* get_buf(){
 		return (m_impl.buf);
 	}
@@ -673,6 +679,11 @@ struct mtr_t {
 	void add_key(uint64_t key){
 		m_impl.key_arr[m_impl.m_n_log_recs] = key;
 	}
+	uint64_t get_key_at(uint32_t i){
+		assert(i >= 0 && i < m_impl.m_n_log_recs);
+		return (m_impl.key_arr[i]);
+	}
+
 	void add_LSN(uint64_t LSN){
 		m_impl.LSN_arr[m_impl.m_n_log_recs] = LSN;
 	}