Submit the rest of a short Linux AIO batch

pkg/aio/aio_linux_test.go

@@ -18,7 +18,11 @@
 package aio
 
 import (
+	"errors"
+	"os"
 	"testing"
+
+	"golang.org/x/sys/unix"
 )
 
 func TestLinuxQueue(t *testing.T) {
@@ -27,3 +31,26 @@ func TestLinuxQueue(t *testing.T) {
 		return q, err
 	})
 }
+
+func TestLinuxQueueShortSubmit(t *testing.T) {
+	f, err := os.Open(os.DevNull)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer f.Close()
+	q, err := NewLinuxQueue(2)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer q.Destroy()
+
+	// Linux accepts the read, then rejects the write to the same read-only
+	// descriptor. Waiting for one completion lets a lost write fail the test
+	// without blocking on a completion the kernel will never produce.
+	fd := int32(f.Fd())
+	Read(q, 0, fd, 0, nil)
+	Write(q, 1, fd, 0, nil)
+	if _, err := q.Wait(nil, 1); !errors.Is(err, unix.EBADF) {
+		t.Errorf("submitting a write to a read-only descriptor: got %v, want %v", err, unix.EBADF)
+	}
+}

pkg/aio/aio_linux_unsafe.go

@@ -112,13 +112,16 @@ func (q *LinuxQueue) Add(r Request) {
 
 // Wait implements Queue.Wait.
 func (q *LinuxQueue) Wait(cs []Completion, minCompletions int) ([]Completion, error) {
-	if len(q.iocbs) != 0 {
-		_, _, e := unix.Syscall(unix.SYS_IO_SUBMIT, q.ctxID, uintptr(len(q.iocbs)), uintptr(unsafe.Pointer(unsafe.SliceData(q.iocbPtrs))))
-		q.iocbs = q.iocbs[:0]
+	// io_submit returns the number of accepted requests, even if a later
+	// request fails. Submit the remainder before waiting for its completions.
+	for pending := q.iocbPtrs[:len(q.iocbs)]; len(pending) > 0; {
+		n, _, e := unix.Syscall(unix.SYS_IO_SUBMIT, q.ctxID, uintptr(len(pending)), uintptr(unsafe.Pointer(unsafe.SliceData(pending))))
 		if e != 0 {
 			return cs, e
 		}
+		pending = pending[n:]
 	}
+	q.iocbs = q.iocbs[:0]
 	for {
 		if ring := q.ring(); ring.Magic == linux.AIO_RING_MAGIC {
 			nr := ring.Nr