MDEV-24449 cleanup: Remove a timeout

recv_sys_t::apply(): At the end of the last batch, wait for
pending reads to complete (read_slots->wait()), instead of
waiting for some time, and assert that buf_pool.n_pend_reads==0
after that wait.

io_callback(): Do not invoke read_slots->release()
before the callback function has returned, to ensure
the correct operation of recv_sys_t::apply().

storage/innobase/include/os0file.h

@@ -1157,6 +1157,8 @@ be other, synchronous, pending writes. */
 void
 os_aio_wait_until_no_pending_writes();
 
+/** Wait until all pending asynchronous reads have completed. */
+void os_aio_wait_until_no_pending_reads();
 
 /** Prints info of the aio arrays.
 @param[in/out]	file		file where to print */

storage/innobase/log/log0recv.cc

@@ -2754,8 +2754,11 @@ void recv_sys_t::apply(bool last_batch)
             mysql_cond_wait(&cond, &mutex);
           else
           {
-            set_timespec_nsec(abstime, 100000ULL); /* 0.1ms */
-            mysql_cond_timedwait(&cond, &mutex, &abstime);
+            mysql_mutex_unlock(&mutex);
+            os_aio_wait_until_no_pending_reads();
+            ut_ad(!buf_pool.n_pend_reads);
+            mysql_mutex_lock(&mutex);
+            ut_ad(pages.empty());
           }
         }
         else

storage/innobase/os/os0file.cc

@@ -3859,24 +3859,26 @@ os_file_get_status(
 
 extern void fil_aio_callback(const IORequest &request);
 
-static void io_callback(tpool::aiocb* cb)
+static void io_callback(tpool::aiocb *cb)
 {
   ut_a(cb->m_err == DB_SUCCESS);
-  const IORequest request(*static_cast<const IORequest*>
-                          (static_cast<const void*>(cb->m_userdata)));
+  const IORequest &request= *static_cast<const IORequest*>
+    (static_cast<const void*>(cb->m_userdata));
+
   /* Return cb back to cache*/
   if (cb->m_opcode == tpool::aio_opcode::AIO_PREAD)
   {
     ut_ad(read_slots->contains(cb));
+    fil_aio_callback(request);
     read_slots->release(cb);
   }
   else
   {
     ut_ad(write_slots->contains(cb));
+    const IORequest req{request};
     write_slots->release(cb);
+    fil_aio_callback(req);
   }
-
-  fil_aio_callback(request);
 }
 
 #ifdef LINUX_NATIVE_AIO
@@ -4072,6 +4074,12 @@ void os_aio_wait_until_no_pending_writes()
   buf_dblwr.wait_flush_buffered_writes();
 }
 
+/** Wait until all pending asynchronous reads have completed. */
+void os_aio_wait_until_no_pending_reads()
+{
+  read_slots->wait();
+}
+
 /** Request a read or write.
 @param type		I/O request
 @param buf		buffer