prepare WAL mode. Implement UNIV_PMEMOBJ_BUF_APPEND that append page in pm_buf_write and read the lasted in pm_buf_read

trdtnguyen · trdtnguyen · commit 9c4dd1b4560a · 2018-02-23T14:53:27.000+09:00
diff --git a/storage/innobase/buf/buf0flu.cc b/storage/innobase/buf/buf0flu.cc
@@ -1096,6 +1096,8 @@ buf_flush_write_block_low(
 	int ret = pm_buf_write_no_free_pool(gb_pmw->pop, gb_pmw->pbuf, bpage->id, bpage->size, frame, sync);
 #elif defined (UNIV_PMEMOBJ_BUF_FLUSHER)
 	int ret = pm_buf_write_with_flusher(gb_pmw->pop, gb_pmw->pbuf, bpage->id, bpage->size, frame, sync);
+#elif defined (UNIV_PMEMOBJ_BUF_APPEND)
+	int ret = pm_buf_write_with_flusher_append(gb_pmw->pop, gb_pmw->pbuf, bpage->id, bpage->size, frame, sync);
 #else
 	int ret = pm_buf_write(gb_pmw->pop, gb_pmw->pbuf, bpage->id, bpage->size, frame, sync);
 #endif
@@ -4084,6 +4086,7 @@ DECLARE_THREAD(pm_flusher_worker)(
 				plist = flusher->flush_list_arr[i];
 				if (plist)
 				{
+					//***this call aio_batch ***
 					pm_buf_flush_list(gb_pmw->pop, gb_pmw->pbuf, plist);
 					flusher->n_requested--;
 					os_event_set(flusher->is_req_full);
@@ -4095,7 +4098,6 @@ DECLARE_THREAD(pm_flusher_worker)(
 		} //end if flusher->n_requested > 0
 
 		if (flusher->n_requested == 0) {
-			//if (srv_shutdown_state != SRV_SHUTDOWN_NONE) {
 			if (buf_page_cleaner_is_active) {
 				//buf_page_cleaner is running, start waiting
 				os_event_reset(flusher->is_req_not_empty);
@@ -4160,6 +4162,14 @@ pm_handle_finished_block_with_flusher(
 		//Now all pages in this list are persistent in disk
 		//(0) flush spaces
 		pm_buf_flush_spaces_in_list(pop, buf, pflush_list);
+		// Reset the param_array
+		ulint arr_idx;
+		arr_idx = pflush_list->param_arr_index;
+		assert(arr_idx >= 0);
+
+		pmemobj_rwlock_wrlock(pop, &buf->param_lock);
+		buf->param_arrs[arr_idx].is_free = true;
+		pmemobj_rwlock_unlock(pop, &buf->param_lock);
 
 		//(1) Reset blocks in the list
 		ulint i;
diff --git a/storage/innobase/buf/buf0rea.cc b/storage/innobase/buf/buf0rea.cc
@@ -195,7 +195,11 @@ buf_read_page_low(
 	);
 #if defined (UNIV_PMEMOBJ_BUF)
 	//size_t read_bytes=  pm_buf_read(gb_pmw->pop, gb_pmw->pbuf, page_id, page_size, (byte*)dst, sync);
+#if defined (UNIV_PMEMOBJ_BUF_APPEND)
+	const PMEM_BUF_BLOCK* pblock=  pm_buf_read_lasted(gb_pmw->pop, gb_pmw->pbuf, page_id, page_size, (byte*)dst, sync);
+#else
 	const PMEM_BUF_BLOCK* pblock=  pm_buf_read(gb_pmw->pop, gb_pmw->pbuf, page_id, page_size, (byte*)dst, sync);
+#endif
 	//if (read_bytes > 0) {
 	if (pblock) {
 		if (sync) {
diff --git a/storage/innobase/fil/fil0fil.cc b/storage/innobase/fil/fil0fil.cc
@@ -5930,8 +5930,34 @@ pm_fil_io_batch(
 	plist->n_flush = 0;
 #endif 	
 	assert(plist->hashed_id != PMEM_ID_NONE);
+	
+	//find a free params to fill aio_batch info
+	//params = pmem_buf->params_arr[plist->hashed_id];
+	ulint cur_free = pmem_buf->cur_free_param;
+	ulint arr_size = pmem_buf->param_arr_size;
+
+	/*Note that this thread've acquired flusher->mutex, so we don't need another mutex for param_array*/
+	pmemobj_rwlock_wrlock(pop, &pmem_buf->param_lock);
+	for (i = 0; i < arr_size; i++) {
+		if (pmem_buf->param_arrs[cur_free].is_free) {
+			params = pmem_buf->param_arrs[cur_free].params;
+			pmem_buf->param_arrs[cur_free].is_free = false; //we set this true in io_complete
+			plist->param_arr_index = cur_free;
+			pmem_buf->cur_free_param = (cur_free + 1) % arr_size;
+			break;
+		}
+		else {
+			cur_free = (cur_free + 1) % arr_size;
+		}
+	}
 
-	params = pmem_buf->params_arr[plist->hashed_id];
+	if (i == arr_size) {
+		//There is no free params to assign
+		printf("PMEM_ERROR: there is no free params to assign");
+		assert(0);
+	}
+	pmemobj_rwlock_unlock(pop, &pmem_buf->param_lock);
+	//params = pmem_buf->params_arr[plist->hashed_id];
 
 	n_params = 0;
 
diff --git a/storage/innobase/fsp/fsp0file.cc b/storage/innobase/fsp/fsp0file.cc
@@ -339,8 +339,13 @@ Datafile::read_first_page(bool read_only_mode)
 		ulint	n_read = 0;
 #if defined (UNIV_PMEMOBJ_BUF)
 		 //size_t read_bytes=  pm_buf_read(gb_pmw->pop, gb_pmw->pbuf,
+#if defined (UNIV_PMEMOBJ_BUF_APPEND)
+		 const PMEM_BUF_BLOCK* pblock =  pm_buf_read_lasted(gb_pmw->pop, gb_pmw->pbuf,
+				 page_id_t(0,0), page_size_t(UNIV_PAGE_SIZE, UNIV_PAGE_SIZE, false), m_first_page, true);
+#else
 		 const PMEM_BUF_BLOCK* pblock =  pm_buf_read(gb_pmw->pop, gb_pmw->pbuf,
 				 page_id_t(0,0), page_size_t(UNIV_PAGE_SIZE, UNIV_PAGE_SIZE, false), m_first_page, true);
+#endif
 		 //if (read_bytes > 0) {
 		 if (pblock) {
 			 err = DB_SUCCESS;
diff --git a/storage/innobase/handler/ha_innodb.cc b/storage/innobase/handler/ha_innodb.cc
@@ -4448,7 +4448,7 @@ innobase_commit(
 
 		/* Now do a write + flush of logs. */
 		if (!read_only) {
-#if defined(UNIV_PMEMOBJ_LOG)
+#if defined(UNIV_PMEMOBJ_LOG) || defined (UNIV_PMEMOBJ_WAL)
 			/*Set this flag to true make the trx_commit_complete_for_mysql() do nothing
 			//Since the log buffer now is in NVDIMM, we can safely skip writting
 			//and flushing log records when the transaction commit
@@ -19522,7 +19522,7 @@ static MYSQL_SYSVAR_ULONG(pmem_page_per_bucket_bits, srv_pmem_page_per_bucket_bi
   "Number of bits present the maxmum number of pages per space in a bucket in partition algorithm, from 1 to log2(srv_pmem_buf_bucket_size), default is 10.",
   NULL, NULL, 10, 1, 32, 0);
 #endif 
-#if defined (UNIV_PMEMOBJ_BUF) || defined (UNIV_PMEMOBJ_DBW) || defined (UNIV_PMEMOBJ_LOG) 
+#if defined (UNIV_PMEMOBJ_BUF) || defined (UNIV_PMEMOBJ_DBW) || defined (UNIV_PMEMOBJ_LOG) || defined (UNIV_PMEMOBJ_WAL)
 static MYSQL_SYSVAR_STR(pmem_home_dir, srv_pmem_home_dir,
   PLUGIN_VAR_RQCMDARG | PLUGIN_VAR_READONLY,
   "Path to PMEM home dir.", NULL, NULL, NULL);
@@ -20364,7 +20364,7 @@ static struct st_mysql_sys_var* innobase_system_variables[]= {
   MYSQL_SYSVAR(pmem_n_space_bits),
   MYSQL_SYSVAR(pmem_page_per_bucket_bits),
 #endif
-#if defined (UNIV_PMEMOBJ_BUF) || defined (UNIV_PMEMOBJ_DBW) || defined (UNIV_PMEMOBJ_LOG) 
+#if defined (UNIV_PMEMOBJ_BUF) || defined (UNIV_PMEMOBJ_DBW) || defined (UNIV_PMEMOBJ_LOG) || defined (UNIV_PMEMOBJ_WAL)
   MYSQL_SYSVAR(pmem_home_dir),
   MYSQL_SYSVAR(pmem_pool_size),
   MYSQL_SYSVAR(pmem_buf_size),
diff --git a/storage/innobase/include/my_pmemobj.h b/storage/innobase/include/my_pmemobj.h
@@ -250,6 +250,7 @@ struct __pmem_buf_block_list_t {
 	int					check;
 	ulint				last_time;
 	
+	ulint				param_arr_index; //index of the param in the param_arrs used to carry the info of this list
 	//int					flush_worker_id;
 	//bool				is_worker_handling;
 	
@@ -286,13 +287,19 @@ struct __pmem_buf {
 	//Those varables are in DRAM
 	os_event_t*  flush_events; //N flush events for N buckets
 	os_event_t free_pool_event; //event for free_pool
-	PMEM_AIO_PARAM** params_arr;
+	
+
+	PMEMrwlock				param_lock;
+	PMEM_AIO_PARAM_ARRAY* param_arrs;//circular array of pointers
+	ulint			param_arr_size; //size of the array
+	ulint			cur_free_param; //circular index, where the next free params is
 	
 	PMEM_FLUSHER* flusher;	
 
 	PMEM_FILE_MAP* filemap;
 };
 
+
 // PARTITION //////////////
 /*Map space id to hashed_id, for partition purpose
  * */
@@ -418,6 +425,15 @@ pm_buf_write_with_flusher(
 		   	page_size_t		size,
 		   	byte*			src_data,
 		   	bool			sync);
+int
+pm_buf_write_with_flusher_append(
+			PMEMobjpool*	pop,
+		   	PMEM_BUF*		buf,
+		   	page_id_t		page_id,
+		   	page_size_t		size,
+		   	byte*			src_data,
+		   	bool			sync);
+
 
 const PMEM_BUF_BLOCK*
 pm_buf_read(
@@ -428,6 +444,15 @@ pm_buf_read(
 		   	byte*				data,
 		   	bool sync);
 
+const PMEM_BUF_BLOCK*
+pm_buf_read_lasted(
+			PMEMobjpool*		pop,
+		   	PMEM_BUF*			buf,
+		   	const page_id_t		page_id,
+		   	const page_size_t	size,
+		   	byte*				data,
+		   	bool sync);
+
 void
 pm_buf_flush_list(
 			PMEMobjpool*			pop,
diff --git a/storage/innobase/include/os0file.h b/storage/innobase/include/os0file.h
@@ -1505,6 +1505,8 @@ pfs_os_aio_func(
 #if defined (UNIV_PMEMOBJ_BUF)  
 struct __pmem_aio_param;
 typedef struct __pmem_aio_param PMEM_AIO_PARAM;
+struct __pmem_aio_param_arr;
+typedef struct __pmem_aio_param_arr PMEM_AIO_PARAM_ARRAY;
 
 struct __pmem_aio_param {
 	const char*			name;
@@ -1515,6 +1517,10 @@ struct __pmem_aio_param {
 	fil_node_t*			m1;
 	void*				m2;
 };
+struct __pmem_aio_param_arr {
+	bool	is_free;
+	PMEM_AIO_PARAM* params;
+};
 
 UNIV_INLINE
 dberr_t
diff --git a/storage/innobase/include/srv0srv.h b/storage/innobase/include/srv0srv.h
@@ -281,7 +281,7 @@ extern ulong	srv_pmem_n_flush_threads;
 extern ulong	srv_pmem_n_space_bits;
 extern ulong	srv_pmem_page_per_bucket_bits;
 #endif
-#if defined(UNIV_PMEMOBJ_BUF) || defined (UNIV_PMEMOBJ_DBW) || defined (UNIV_PMEMOBJ_LOG) 
+#if defined(UNIV_PMEMOBJ_BUF) || defined (UNIV_PMEMOBJ_DBW) || defined (UNIV_PMEMOBJ_LOG) || defined(UNIV_PMEMOBJ_WAL)
 extern char*	srv_pmem_home_dir;
 extern ulong	srv_pmem_pool_size;
 extern ulong	srv_pmem_buf_size;
diff --git a/storage/innobase/log/log0log.cc b/storage/innobase/log/log0log.cc
@@ -224,7 +224,7 @@ log_buffer_extend(
 
 	log_sys->buf_free -= move_start;
 	log_sys->buf_next_to_write -= move_start;
-#if defined (UNIV_PMEMOBJ_LOG)
+#if defined (UNIV_PMEMOBJ_LOG) || defined (UNIV_PMEMOBJ_WAL)
 	/* reallocate log buffer */
 	srv_log_buffer_size = len / UNIV_PAGE_SIZE + 1;
 	//do not free here, does it by our API
@@ -472,7 +472,7 @@ log_write_low(
 			- LOG_BLOCK_TRL_SIZE;
 	}
 
-#if defined (UNIV_PMEMOBJ_LOG)
+#if defined (UNIV_PMEMOBJ_LOG) || defined (UNIV_PMEMOBJ_WAL)
 	//copy the trasaction's log records to persistent memory
 	pmemobj_memcpy_persist(gb_pmw->pop, log->buf + log->buf_free, str, len);
 	//set the flag here
@@ -512,7 +512,7 @@ log_write_low(
 	if (str_len > 0) {
 		goto part_loop;
 	}
-#if defined(UNIV_PMEMOBJ_LOG)
+#if defined(UNIV_PMEMOBJ_LOG) || defined(UNIV_PMEMOBJ_WAL)
 	// update the lsn and buf_free
 	gb_pmw->plogbuf->lsn = log->lsn;
 	gb_pmw->plogbuf->buf_free = log->buf_free;	
@@ -860,7 +860,7 @@ log_init(void)
 	ut_a(LOG_BUFFER_SIZE >= 4 * UNIV_PAGE_SIZE);
 
 	log_sys->buf_size = LOG_BUFFER_SIZE;
-#if defined(UNIV_PMEMOBJ_LOG)
+#if defined(UNIV_PMEMOBJ_LOG) || defined (UNIV_PMEMOBJ_WAL)
 	//allocate the log buffer in persistent memory
 	if (!gb_pmw->plogbuf){
 		if ( pm_wrapper_logbuf_alloc(gb_pmw, 2 * LOG_BUFFER_SIZE + OS_FILE_LOG_BLOCK_SIZE)
@@ -929,7 +929,7 @@ log_init(void)
 
 	log_sys->buf_free = LOG_BLOCK_HDR_SIZE;
 	log_sys->lsn = LOG_START_LSN + LOG_BLOCK_HDR_SIZE;
-#if defined(UNIV_PMEMOBJ_LOG)
+#if defined(UNIV_PMEMOBJ_LOG) || defined (UNIV_PMEMOBJ_WAL)
 	//Update the lsn, buf_free in necessary
 	if(gb_pmw->plogbuf->lsn < log_sys->lsn)
 		gb_pmw->plogbuf->lsn = log_sys->lsn;
@@ -1002,7 +1002,7 @@ log_io_complete(
 /*============*/
 	log_group_t*	group)	/*!< in: log group or a dummy pointer */
 {
-#if defined (UNIV_PMEMOBJ_LOG)
+#if defined (UNIV_PMEMOBJ_LOG) || defined (UNIV_PMEMOBJ_WAL)
 	//In original: This function called by fil_aio_wait as async call from log_checkpoint
 	//In our PMEMOBJ_LOG, log_write_up_to use async too, so we need distinguish between two cases
 		
@@ -1223,7 +1223,7 @@ log_group_write_buf(
 
 	const ulint	page_no
 		= (ulint) (next_offset / univ_page_size.physical());
-#if defined (UNIV_PMEMOBJ_LOG)
+#if defined (UNIV_PMEMOBJ_LOG) || defined (UNIV_PMEMOBJ_WAL)
 	//We can write async when the log buffer is in PMEM
 	//We don't adjust group to group + 1 because we already handled that in log_io_complete
 	fil_io(IORequestLogWrite, false,
@@ -2631,7 +2631,7 @@ log_shutdown(void)
 /*==============*/
 {
 	log_group_close_all();
-#if defined(UNIV_PMEMOBJ_LOG)
+#if defined(UNIV_PMEMOBJ_LOG) || defined (UNIV_PMEMOBJ_WAL)
 	//The pmem free() is done later
 	//do nothing for log_sys->buf_ptr
 #else //original
diff --git a/storage/innobase/log/log0recv.cc b/storage/innobase/log/log0recv.cc
@@ -68,7 +68,7 @@ bool	recv_replay_file_ops	= true;
 #include "fut0lst.h"
 #endif /* !UNIV_HOTBACKUP */
 
-#if defined(UNIV_PMEMOBJ_LOG)
+#if defined(UNIV_PMEMOBJ_LOG) || defined (UNIV_PMEMOBJ_WAL)
 #include "my_pmemobj.h"
 extern PMEM_WRAPPER* gb_pmw;
 #endif
@@ -4293,7 +4293,7 @@ recv_recovery_from_checkpoint_start(
 
 	log_sys->next_checkpoint_no = checkpoint_no + 1;
 
-#if defined(UNIV_PMEMOBJ_LOG)
+#if defined(UNIV_PMEMOBJ_LOG) || defined (UNIV_PMEMOBJ_WAL)
 	if (	gb_pmw->plogbuf->buf_free > log_sys->buf_free &&
 			gb_pmw->plogbuf->lsn > log_sys->lsn ) {
 		uint64_t len = gb_pmw->plogbuf->buf_free - log_sys->buf_free;
diff --git a/storage/innobase/os/os0file.cc b/storage/innobase/os/os0file.cc
@@ -7371,7 +7371,8 @@ AIO::pm_process_batch(
 	wrapper = &m_seg_wrapper_arr[local_seg];
 	assert (wrapper->local_index == local_seg);
 	wrapper->io_pending = 0;
-
+	
+	//this loop does assign parameters from input params to io_cb
 	for (i = count, slot_i = local_seg * slots_per_seg;
 		   	i < n_params && slot_i < m_slots.size();
 		   	++slot_i, ++i) {
diff --git a/storage/innobase/pmem/pmem0buf.cc b/storage/innobase/pmem/pmem0buf.cc
diff --git a/storage/innobase/row/row0trunc.cc b/storage/innobase/row/row0trunc.cc
diff --git a/storage/innobase/srv/srv0srv.cc b/storage/innobase/srv/srv0srv.cc
diff --git a/storage/innobase/srv/srv0start.cc b/storage/innobase/srv/srv0start.cc
