aio: Try repeatedly to give batched IOs to workers.

Previously, if the first of a batch of IOs didn't fit in a batch we'd
run all of them synchronously.  Andres rightly pointed out that we
should really try again between synchronous IOs, since the workers might
have made progress.

Suggested-by: Andres Freund <andres@anarazel.de>

Author: macdice

src/backend/storage/aio/method_worker.c

@@ -280,12 +280,36 @@ pgaio_worker_submit_internal(int num_staged_ios, PgAioHandle **staged_ios)
 		SetLatch(wakeup);
 
 	/* Run whatever is left synchronously. */
-	if (nsync > 0)
+	for (int i = 0; i < nsync; ++i)
 	{
-		for (int i = 0; i < nsync; ++i)
+		wakeup = NULL;
+
+		/*
+		 * Between synchronous IO operations, try again to enqueue as many as
+		 * we can.
+		 */
+		if (i > 0)
 		{
-			pgaio_io_perform_synchronously(synchronous_ios[i]);
+			wakeup = NULL;
+
+			LWLockAcquire(AioWorkerSubmissionQueueLock, LW_EXCLUSIVE);
+			while (i < nsync &&
+				   pgaio_worker_submission_queue_insert(synchronous_ios[i]))
+			{
+				if (wakeup == NULL && (worker = pgaio_worker_choose_idle()) >= 0)
+					wakeup = io_worker_control->workers[worker].latch;
+				i++;
+			}
+			LWLockRelease(AioWorkerSubmissionQueueLock);
+
+			if (wakeup)
+				SetLatch(wakeup);
+
+			if (i == nsync)
+				break;
 		}
+
+		pgaio_io_perform_synchronously(synchronous_ios[i]);
 	}
 }