MDEV-26865 fts_optimize_thread cannot keep up with workload

fts_cache_t::total_size_at_sync: New field, to sample total_size.

fts_add_doc_by_id(): Invoke sync if total_size has grown too much
since the previous sync request. (Maintain cache->total_size_at_sync.)

ib_wqueue_t::length: Caches ib_list_len(*items).

ib_wqueue_len(): Removed. We will refer to fts_optimize_wq->length
directly.

Based on mysql/mysql-server@bc9c46bf2894673d0df17cd0ee872d0d99663121

storage/innobase/buf/buf0mtflu.cc

@@ -367,14 +367,6 @@ DECLARE_THREAD(mtflush_io_thread)(void* arg)
 	mutex_exit(&(mtflush_io->thread_global_mtx));
 
 	while (TRUE) {
-
-#ifdef UNIV_MTFLUSH_DEBUG
- 		fprintf(stderr, "InnoDB: Note. Thread %lu work queue len %lu return queue len %lu\n",
- 					os_thread_get_curr_id(),
- 					ib_wqueue_len(mtflush_io->wq),
- 					ib_wqueue_len(mtflush_io->wr_cq));
-#endif /* UNIV_MTFLUSH_DEBUG */
-
 		mtflush_service_io(mtflush_io, this_thread_data);
 
 

storage/innobase/fts/fts0fts.cc

@@ -580,6 +580,7 @@ fts_cache_init(
 	cache->sync_heap->arg = mem_heap_create(1024);
 
 	cache->total_size = 0;
+	cache->total_size_at_sync = 0;
 
 	mutex_enter((ib_mutex_t*) &cache->deleted_lock);
 	cache->deleted_doc_ids = ib_vector_create(
@@ -3571,11 +3572,14 @@ fts_add_doc_by_id(
 					get_doc->index_cache,
 					doc_id, doc.tokens);
 
-				bool	need_sync = false;
-				if ((cache->total_size > fts_max_cache_size / 10
-				     || fts_need_sync)
-				    && !cache->sync->in_progress) {
-					need_sync = true;
+				bool	need_sync = !cache->sync->in_progress
+					&& (fts_need_sync
+					    || (cache->total_size
+						- cache->total_size_at_sync)
+					    > fts_max_cache_size / 10);
+				if (need_sync) {
+					cache->total_size_at_sync =
+						cache->total_size;
 				}
 
 				rw_lock_x_unlock(&table->fts->cache->lock);

storage/innobase/include/fts0types.h

 /*****************************************************************************
 
 Copyright (c) 2007, 2016, Oracle and/or its affiliates. All Rights Reserved.
-Copyright (c) 2017, 2020, MariaDB Corporation.
+Copyright (c) 2017, 2021, MariaDB Corporation.
 
 This program is free software; you can redistribute it and/or modify it under
 the terms of the GNU General Public License as published by the Free Software
@@ -150,6 +150,9 @@ struct fts_cache_t {
 	size_t		total_size;	/*!< total size consumed by the ilist
 					field of all nodes. SYNC is run
 					whenever this gets too big */
+	/** total_size at the time of the previous SYNC request */
+	size_t		total_size_at_sync;
+
 	fts_sync_t*	sync;		/*!< sync structure to sync data to
 					disk */
 	ib_alloc_t*	sync_heap;	/*!< The heap allocator, for indexes

storage/innobase/include/ut0wqueue.h

 /*****************************************************************************
 
 Copyright (c) 2006, 2014, Oracle and/or its affiliates. All Rights Reserved.
-Copyright (c) 2017, 2019, MariaDB Corporation.
+Copyright (c) 2017, 2021, MariaDB Corporation.
 
 This program is free software; you can redistribute it and/or modify it under
 the terms of the GNU General Public License as published by the Free Software
@@ -46,6 +46,8 @@ struct ib_wqueue_t
 	ib_mutex_t	mutex;
 	/** Work item list */
 	ib_list_t*	items;
+	/** ib_list_len(*items) */
+	size_t		length;
 	/** event we use to signal additions to list;
 	os_event_set() and os_event_reset() are protected by the mutex */
 	os_event_t	event;
@@ -103,12 +105,5 @@ void*
 ib_wqueue_nowait(
 /*=============*/
 	ib_wqueue_t*	wq);		/*<! in: work queue */
-/********************************************************************
-Get number of items on queue.
-@return number of items on queue */
-ulint
-ib_wqueue_len(
-/*==========*/
-	ib_wqueue_t*	wq);		/*<! in: work queue */
 
 #endif /* IB_WORK_QUEUE_H */

storage/innobase/ut/ut0wqueue.cc

 /*****************************************************************************
 
 Copyright (c) 2006, 2015, Oracle and/or its affiliates. All Rights Reserved.
-Copyright (c) 2019, MariaDB Corporation.
+Copyright (c) 2019, 2021, MariaDB Corporation.
 
 This program is free software; you can redistribute it and/or modify it under
 the terms of the GNU General Public License as published by the Free Software
@@ -45,6 +45,7 @@ ib_wqueue_create(void)
 
 	wq->items = ib_list_create();
 	wq->event = os_event_create(0);
+	wq->length = 0;
 
 	return(wq);
 }
@@ -76,6 +77,8 @@ ib_wqueue_add(ib_wqueue_t* wq, void* item, mem_heap_t* heap, bool wq_locked)
 	}
 
 	ib_list_add_last(wq->items, item, heap);
+	wq->length++;
+	ut_ad(wq->length == ib_list_len(wq->items));
 	os_event_set(wq->event);
 
 	if (!wq_locked) {
@@ -102,12 +105,12 @@ ib_wqueue_wait(
 
 		if (node) {
 			ib_list_remove(wq->items, node);
-
-			if (!ib_list_get_first(wq->items)) {
+			if (!--wq->length) {
 				/* We must reset the event when the list
 				gets emptied. */
 				os_event_reset(wq->event);
 			}
+			ut_ad(wq->length == ib_list_len(wq->items));
 
 			break;
 		}
@@ -142,7 +145,8 @@ ib_wqueue_timedwait(
 
 		if (node) {
 			ib_list_remove(wq->items, node);
-
+			wq->length--;
+			ut_ad(wq->length == ib_list_len(wq->items));
 			mutex_exit(&wq->mutex);
 			break;
 		}
@@ -204,20 +208,3 @@ bool ib_wqueue_is_empty(ib_wqueue_t* wq)
 	mutex_exit(&wq->mutex);
 	return is_empty;
 }
-
-/********************************************************************
-Get number of items on queue.
-@return number of items on queue */
-ulint
-ib_wqueue_len(
-/*==========*/
-	ib_wqueue_t*	wq)		/*<! in: work queue */
-{
-	ulint len = 0;
-
-	mutex_enter(&wq->mutex);
-	len = ib_list_len(wq->items);
-	mutex_exit(&wq->mutex);
-
-        return(len);
-}