Merged latest mt-flush code to xtradb. Cleaned up thread statistic output

code.

Author: Jan Lindström

storage/innobase/buf/buf0mtflu.cc

@@ -116,18 +116,13 @@ typedef struct wrk_itm
 /* Thread syncronization data */
 typedef struct thread_sync
 {
+	ulint           n_threads;	/*!< Number of threads */
 	os_thread_id_t	wthread_id;	/*!< Identifier */
 	os_thread_t 	wthread;	/*!< Thread id */
 	ib_wqueue_t	*wq;		/*!< Work Queue */
 	ib_wqueue_t     *wr_cq;		/*!< Write Completion Queue */
 	ib_wqueue_t     *rd_cq;		/*!< Read Completion Queue */
 	wthr_status_t   wt_status;	/*!< Worker thread status */
-	ulint		stat_universal_num_processed;
-					/*!< Total number of pages
-					processed by this thread */
-	ulint		stat_cycle_num_processed;
-					/*!< Number of pages processed
-					on this cycle */
 	mem_heap_t*     wheap;		/*!< Work heap where memory
 					is allocated */
 	wrk_t*          work_item;      /*!< Work items to be processed */
@@ -231,6 +226,7 @@ buf_mtflu_flush_pool_instance(
                                     		work_item->wr.min,
 						work_item->wr.lsn_limit);
 
+
 	buf_flush_end(work_item->wr.buf_pool, work_item->wr.flush_type);
 	buf_flush_common(work_item->wr.flush_type, work_item->n_flushed);
 
@@ -239,28 +235,29 @@ buf_mtflu_flush_pool_instance(
 
 #ifdef UNIV_DEBUG
 /******************************************************************//**
-Output work item list status,
+Print flush statistics of work items.
 */
 static
 void
-mtflu_print_work_list(
-/*==================*/
-	wrk_t*	wi_list)	/*!< in: Work item list */
+mtflu_print_thread_stat(
+/*====================*/
+	wrk_t* work_item)	/*!< in: Work items */
 {
-	wrk_t* wi = wi_list;
+	ulint stat_tot=0;
 	ulint i=0;
 
-	if(!wi_list) {
-		fprintf(stderr, "list NULL\n");
-	}
+ 	for(i=0; i< MTFLUSH_MAX_WORKER; i++) {
+ 		stat_tot+=work_item[i].n_flushed;
 
-	while(wi) {
-		fprintf(stderr, "-\t[%p]\t[%s]\t[%lu] > %p\n",
-			wi, (wi->id_usr == -1)?"free":"Busy", wi->n_flushed, wi->next);
-		wi = wi->next;
-		i++;
-	}
-	fprintf(stderr, "list len: %d\n", i);
+ 		fprintf(stderr, "MTFLUSH: Thread[%lu] stat [%lu]\n",
+			work_item[i].id_usr,
+ 			work_item[i].n_flushed);
+
+		if (work_item[i].next == NULL) {
+			break; /* No more filled work items */
+		}
+ 	}
+ 	fprintf(stderr, "MTFLUSH: Stat-Total:%lu\n", stat_tot);
 }
 #endif /* UNIV_DEBUG */
 
@@ -282,10 +279,6 @@ mtflush_service_io(
    	mtflush_io->wt_status = WTHR_SIG_WAITING;
 	work_item = (wrk_t *)ib_wqueue_timedwait(mtflush_io->wq, max_wait_usecs);
 
-#ifdef UNIV_DEBUG
-	mtflu_print_work_list(mtflush_io->work_item);
-#endif
-
 	if (work_item) {
 		mtflush_io->wt_status = WTHR_RUNNING;
 	} else {
@@ -345,10 +338,28 @@ DECLARE_THREAD(mtflush_io_thread)(
 	void * arg)
 {
 	thread_sync_t *mtflush_io = ((thread_sync_t *)arg);
+#ifdef UNIV_DEBUG
+	ib_uint64_t   stat_universal_num_processed = 0;
+	ib_uint64_t   stat_cycle_num_processed = 0;
+	wrk_t*		work_item = mtflush_io[0].work_item;
+	ulint i;
+#endif
 
 	while (srv_shutdown_state != SRV_SHUTDOWN_EXIT_THREADS) {
 		mtflush_service_io(mtflush_io);
-		mtflush_io->stat_cycle_num_processed = 0;
+
+#ifdef UNIV_DEBUG
+		for(i=0; i < MTFLUSH_MAX_WORKER; i++) {
+			stat_cycle_num_processed+= work_item[i].n_flushed;
+		}
+
+		stat_universal_num_processed+=stat_cycle_num_processed;
+		stat_cycle_num_processed = 0;
+		fprintf(stderr, "MTFLUSH_IO_THREAD: total %lu cycle %lu\n",
+			stat_universal_num_processed,
+			stat_cycle_num_processed);
+		mtflu_print_thread_stat(work_item);
+#endif
 	}
 
 	/* This should make sure that all current work items are
@@ -458,13 +469,16 @@ buf_mtflu_handler_init(
 	work_items = (wrk_t*)mem_heap_alloc(mtflush_heap,
 					    MTFLUSH_MAX_WORKER * sizeof(wrk_t));
 	ut_a(work_items != NULL);
+	memset(work_items, 0, sizeof(wrk_t) * MTFLUSH_MAX_WORKER);
+	memset(mtflush_ctx, 0, sizeof(thread_sync_t) * MTFLUSH_MAX_WORKER);
 
 	/* Initialize work items */
-	mtflu_setup_work_items(work_items, MTFLUSH_MAX_WORKER);
+	mtflu_setup_work_items(work_items, n_threads);
 
 	/* Create threads for page-compression-flush */
 	for(i=0; i < n_threads; i++) {
 		os_thread_id_t new_thread_id;
+		mtflush_ctx[i].n_threads = n_threads;
 		mtflush_ctx[i].wq = mtflush_work_queue;
 		mtflush_ctx[i].wr_cq = mtflush_write_comp_queue;
 		mtflush_ctx[i].rd_cq = mtflush_read_comp_queue;
@@ -531,67 +545,23 @@ buf_mtflu_flush_work_items(
 				per_pool_pages_flushed[i] = done_wi->n_flushed;
 			}
 
-			if(done_wi->id_usr == -1 &&
+			if((int)done_wi->id_usr == -1 &&
 			   done_wi->wi_status == WRK_ITEM_SET ) {
 				fprintf(stderr,
-					"**Set/Unused work_item[%d] flush_type=%lu\n",
+					"**Set/Unused work_item[%lu] flush_type=%lu\n",
 					i,
 					done_wi->wr.flush_type);
 				ut_a(0);
 			}
 
 			n_flushed+= done_wi->n_flushed;
-			/* Reset for next round*/
-			mtflush_ctx->work_item[i].id_usr = -1;
-
 			i++;
 		}
 	}
 
 	return(n_flushed);
 }
 
-/*******************************************************************//**
-Flushes dirty blocks from the end of the LRU list and also
-puts replaceable clean pages from the end of the LRU list to the free
-list.
-NOTE: The calling thread is not allowed to own any latches on pages!
-@return true if a batch was queued successfully. false if another batch
-of same type was already running. */
-bool
-buf_mtflu_flush_LRU(
-/*================*/
-	buf_pool_t*	buf_pool,	/*!< in/out: buffer pool instance */
-	ulint		min_n,		/*!< in: wished minimum mumber of blocks
-					flushed (it is not guaranteed that the
-					actual number is that big, though) */
-	ulint*		n_processed)	/*!< out: the number of pages
-					which were processed is passed
-					back to caller. Ignored if NULL */
-{
-	ulint		page_count;
-
-	if (n_processed) {
-		*n_processed = 0;
-	}
-
-	if (!buf_flush_start(buf_pool, BUF_FLUSH_LRU)) {
-		return(false);
-	}
-
-	page_count = buf_flush_batch(buf_pool, BUF_FLUSH_LRU, min_n, 0);
-
-	buf_flush_end(buf_pool, BUF_FLUSH_LRU);
-
-	buf_flush_common(BUF_FLUSH_LRU, page_count);
-
-	if (n_processed) {
-		*n_processed = page_count;
-	}
-
-	return(true);
-}
-
 /*******************************************************************//**
 Multi-threaded version of buf_flush_list
 */

storage/xtradb/CMakeLists.txt

@@ -284,8 +284,7 @@ SET(INNOBASE_SOURCES
 	buf/buf0flu.cc
 	buf/buf0lru.cc
 	buf/buf0rea.cc
-# TODO: JAN uncomment
-#	buf/buf0mtflu.cc
+	buf/buf0mtflu.cc
 	data/data0data.cc
 	data/data0type.cc
 	dict/dict0boot.cc