MDEV-31095 tpool - restrict threadpool concurrency during bufferpool load

Add threadpool functionality to restrict concurrency during "batch"
periods (where tasks are added in rapid succession).
This will throttle thread creation more agressively than usual, while
keeping performance at least on-par.

One of these cases is bufferpool load, where async read IOs are executed
without any throttling. There can be as much as 650K read IOs for
loading 10GB buffer pool.

Why there are more threads than we expect?
Worker threads are not be recognized as idle, until they return to the
standby list, and to return to that list, they need to acquire
mutex currently held in the submit_task(). In those cases, submit_task()
has no worker to wake, and would create threads until default concurrency
level (2*ncpus) is satisfied. Only after that throttling would happen.

Normally, this is fine, and the extraneous threads would go away when they
are idle for 60sec, yet for specific cases such as DB startup we can lower
concurrency level, if this makes someone happier.

4 files changed, 55 insertions, 9 deletions

diff --git a/storage/innobase/buf/buf0dump.cc b/storage/innobase/buf/buf0dump.cc
index 03876666f1d..9f440b25b46 100644
--- a/storage/innobase/buf/buf0dump.cc
+++ b/storage/innobase/buf/buf0dump.cc
@@ -682,6 +682,9 @@ void buf_load_abort()
   buf_load_abort_flag= true;
 }
 
+
+extern tpool::task_group *os_get_read_slots_task_group();
+
 /*****************************************************************//**
 This is the main task for buffer pool dump/load. when scheduled
 either performs a dump or load, depending on server state, state of the variables etc- */
@@ -694,7 +697,20 @@ static void buf_dump_load_func(void *)
 #ifdef WITH_WSREP
 		if (!get_wsrep_recovery()) {
 #endif /* WITH_WSREP */
+			/* One thread is enough for loading buffer
+			that gives the fastest result (at least for native aio)*/
+			auto group= os_get_read_slots_task_group();
+			//Use 1 thread for read completion handling*/
+			group->set_max_tasks(1);
+			//Don't let thread pool create too many threads
+			//in the following batch
+			srv_thread_pool->set_concurrency(1);
+
 			buf_load();
+
+			/* Restore concurrency for IO and threadpool*/
+			group->set_max_tasks(srv_n_read_io_threads);
+			srv_thread_pool->set_concurrency();
 #ifdef WITH_WSREP
 		}
 #endif /* WITH_WSREP */
diff --git a/storage/innobase/os/os0file.cc b/storage/innobase/os/os0file.cc
index 3b81dc7ee07..e7adc74b220 100644
--- a/storage/innobase/os/os0file.cc
+++ b/storage/innobase/os/os0file.cc
@@ -141,6 +141,11 @@ public:
 static io_slots *read_slots;
 static io_slots *write_slots;
 
+tpool::task_group* os_get_read_slots_task_group()
+{
+  return read_slots->get_task_group();
+}
+
 /** Number of retries for partial I/O's */
 constexpr ulint NUM_RETRIES_ON_PARTIAL_IO = 10;
 
diff --git a/tpool/tpool.h b/tpool/tpool.h
index b6680e98acc..5ee7e620206 100644
--- a/tpool/tpool.h
+++ b/tpool/tpool.h
@@ -256,6 +256,24 @@ public:
   {
     m_aio.reset();
   }
+
+  /**
+  Tweaks how fast worker threads are created, or how often they are signaled.
+
+  @param threads - desired number of concurrently active threads
+  Special value 0 means default. Not the same as max number of threads
+  in the pool - oversubscription is allowed and stalls are still detected
+
+  @note
+  It is designed to use with "batch" operations, where huge number
+  of tasks is submitted in rapid succession. In this case, it is
+  better to temporarily restrict concurrency, which will make thread
+  creation throttling more aggressive.
+  Once the batch is over, restore default concurrency
+  by calling set_concurrency(0).
+  */
+  virtual void set_concurrency(unsigned int threads=0){}
+
   int bind(native_file_handle &fd) { return m_aio->bind(fd); }
   void unbind(const native_file_handle &fd) { if (m_aio) m_aio->unbind(fd); }
   int submit_io(aiocb *cb) { return m_aio->submit_io(cb); }
diff --git a/tpool/tpool_generic.cc b/tpool/tpool_generic.cc
index 5d99783e8b9..c2d3f172817 100644
--- a/tpool/tpool_generic.cc
+++ b/tpool/tpool_generic.cc
@@ -445,6 +445,7 @@ public:
   {
     return new timer_generic(func, data, this);
   }
+  void set_concurrency(unsigned int concurrency=0) override;
 };
 
 void thread_pool_generic::cancel_pending(task* t)
@@ -796,7 +797,6 @@ thread_pool_generic::thread_pool_generic(int min_threads, int max_threads) :
   m_tasks_dequeued(),
   m_wakeups(),
   m_spurious_wakeups(),
-  m_concurrency(std::thread::hardware_concurrency()*2),
   m_in_shutdown(),
   m_timestamp(),
   m_long_tasks_count(),
@@ -808,14 +808,7 @@ thread_pool_generic::thread_pool_generic(int min_threads, int max_threads) :
   m_last_activity(),
   m_maintenance_timer(thread_pool_generic::maintenance_func, this, nullptr)
 {
-
-  if (m_max_threads < m_concurrency)
-    m_concurrency = m_max_threads;
-  if (m_min_threads > m_concurrency)
-    m_concurrency = min_threads;
-  if (!m_concurrency)
-    m_concurrency = 1;
-
+  set_concurrency();
   // start the timer
   m_maintenance_timer.set_time(0, (int)m_timer_interval.count());
 }
@@ -844,6 +837,20 @@ bool thread_pool_generic::too_many_active_threads()
     m_concurrency* OVERSUBSCRIBE_FACTOR;
 }
 
+void thread_pool_generic::set_concurrency(unsigned int concurrency)
+{
+  std::unique_lock<std::mutex> lk(m_mtx);
+  if (concurrency == 0)
+    concurrency= 2 * std::thread::hardware_concurrency();
+  m_concurrency = concurrency;
+  if (m_concurrency > m_max_threads)
+    m_concurrency = m_max_threads;
+  if (m_concurrency < m_min_threads)
+    m_concurrency = m_min_threads;
+  if (m_concurrency < 1)
+    m_concurrency = 1;
+}
+
 /** Submit a new task*/
 void thread_pool_generic::submit_task(task* task)
 {