MDEV-32861 InnoDB hangs when running out of I/O slots

When the constant OS_AIO_N_PENDING_IOS_PER_THREAD is changed from 256 to 1
and the server is run with the minimum parameters
innodb_read_io_threads=1 and innodb_write_io_threads=2, two hangs
were observed.

tpool::cache<T>::put(T*): Ensure that get() in io_slots::acquire()
will be woken up when the cache previously was empty.

buf_pool_t::io_buf_t::reserve(): Schedule a possibly partial doublewrite
batch so that os_aio_wait_until_no_pending_writes() has a chance of
returning. Add a Boolean parameter and pass wait_for_reads=false inside
buf_page_decrypt_after_read(), because those calls will be executed
inside a read completion callback, and therefore
os_aio_wait_until_no_pending_reads() would block indefinitely.

storage/innobase/buf/buf0buf.cc

@@ -414,7 +414,7 @@ static bool buf_page_decrypt_after_read(buf_page_t *bpage,
 
 	if (node.space->purpose == FIL_TYPE_TEMPORARY
 	    && innodb_encrypt_temporary_tables) {
-		slot = buf_pool.io_buf_reserve();
+		slot = buf_pool.io_buf_reserve(false);
 		ut_a(slot);
 		slot->allocate();
 
@@ -444,7 +444,7 @@ static bool buf_page_decrypt_after_read(buf_page_t *bpage,
 			return false;
 		}
 
-		slot = buf_pool.io_buf_reserve();
+		slot = buf_pool.io_buf_reserve(false);
 		slot->allocate();
 
 decompress_with_slot:
@@ -471,7 +471,7 @@ static bool buf_page_decrypt_after_read(buf_page_t *bpage,
 			return false;
 		}
 
-		slot = buf_pool.io_buf_reserve();
+		slot = buf_pool.io_buf_reserve(false);
 		slot->allocate();
 		ut_d(fil_page_type_validate(node.space, dst_frame));
 
@@ -1513,14 +1513,17 @@ void buf_pool_t::io_buf_t::close()
   n_slots= 0;
 }
 
-buf_tmp_buffer_t *buf_pool_t::io_buf_t::reserve()
+buf_tmp_buffer_t *buf_pool_t::io_buf_t::reserve(bool wait_for_reads)
 {
   for (;;)
   {
     for (buf_tmp_buffer_t *s= slots, *e= slots + n_slots; s != e; s++)
       if (s->acquire())
         return s;
+    buf_dblwr.flush_buffered_writes();
     os_aio_wait_until_no_pending_writes();
+    if (!wait_for_reads)
+      continue;
     for (buf_tmp_buffer_t *s= slots, *e= slots + n_slots; s != e; s++)
       if (s->acquire())
         return s;

storage/innobase/buf/buf0flu.cc

@@ -702,7 +702,7 @@ static byte *buf_page_encrypt(fil_space_t* space, buf_page_t* bpage, byte* s,
 
   ut_ad(!bpage->zip_size() || !page_compressed);
   /* Find free slot from temporary memory array */
-  buf_tmp_buffer_t *slot= buf_pool.io_buf_reserve();
+  buf_tmp_buffer_t *slot= buf_pool.io_buf_reserve(true);
   ut_a(slot);
   slot->allocate();
   slot->out_buf= NULL;

storage/innobase/include/buf0buf.h

@@ -2030,7 +2030,8 @@ class buf_pool_t
   a delete-buffering operation is pending. Protected by mutex. */
   buf_page_t watch[innodb_purge_threads_MAX + 1];
   /** Reserve a buffer. */
-  buf_tmp_buffer_t *io_buf_reserve() { return io_buf.reserve(); }
+  buf_tmp_buffer_t *io_buf_reserve(bool wait_for_reads)
+  { return io_buf.reserve(wait_for_reads); }
 
   /** @return whether any I/O is pending */
   bool any_io_pending()
@@ -2083,7 +2084,7 @@ class buf_pool_t
     void close();
 
     /** Reserve a buffer */
-    buf_tmp_buffer_t *reserve();
+    buf_tmp_buffer_t *reserve(bool wait_for_reads);
   } io_buf;
 
   /** whether resize() is in the critical path */

tpool/tpool_structs.h

@@ -138,12 +138,13 @@ template<typename T> class cache
   {
     std::unique_lock<std::mutex> lk(m_mtx);
     assert(!is_full());
+    const bool was_empty= is_empty();
     // put element to the logical end of the array
     m_cache[--m_pos] = ele;
 
     /* Notify waiters  when the cache becomes
      not empty, or when it becomes full */
-    if (m_pos == 1 || (m_waiters && is_full()))
+    if (was_empty || (is_full() && m_waiters))
       m_cv.notify_all();
   }