overhaul aio implementation for correctness

previously, aio operations were not tracked by file descriptor; each
operation was completely independent. this resulted in non-conforming
behavior for non-seekable/append-mode writes (which are required to be
ordered) and made it impossible to implement aio_cancel, which in turn
made closing file descriptors with outstanding aio operations unsafe.

the new implementation is significantly heavier (roughly twice the
size, and seems to be slightly slower) and presently aims mainly at
correctness, not performance.

most of the public interfaces have been moved into a single file,
aio.c, because there is little benefit to be had from splitting them.
whenever any aio functions are used, aio_cancel and the internal
queue lifetime management and fd-to-queue mapping code must be linked,
and these functions make up the bulk of the code size.

the close function's interaction with aio is implemented with weak
alias magic, to avoid pulling in heavy aio cancellation code in
programs that don't use aio, and the expensive cancellation path
(which includes signal blocking) is optimized out when there are no
active aio queues.

1 files changed, 46 insertions, 33 deletions

diff --git a/src/aio/aio_suspend.c b/src/aio/aio_suspend.c
index dcdf6019..2391d786 100644
--- a/src/aio/aio_suspend.c
+++ b/src/aio/aio_suspend.c
@@ -1,57 +1,70 @@
 #include <aio.h>
 #include <errno.h>
-#include "pthread_impl.h"
+#include <time.h>
+#include "atomic.h"
 #include "libc.h"
+#include "pthread_impl.h"
 
-/* Due to the requirement that aio_suspend be async-signal-safe, we cannot
- * use any locks, wait queues, etc. that would make it more efficient. The
- * only obviously-correct algorithm is to generate a wakeup every time any
- * aio operation finishes and have aio_suspend re-evaluate the completion
- * status of each aiocb it was waiting on. */
-
-static volatile int seq;
-
-void __aio_wake(void)
-{
-	a_inc(&seq);
-	__wake(&seq, -1, 1);
-}
+extern volatile int __aio_fut;
 
 int aio_suspend(const struct aiocb *const cbs[], int cnt, const struct timespec *ts)
 {
-	int i, last, first=1, ret=0;
+	int i, tid = 0, ret, expect = 0;
 	struct timespec at;
+	volatile int dummy_fut, *pfut;
+	int nzcnt = 0;
+	const struct aiocb *cb = 0;
 
 	if (cnt<0) {
 		errno = EINVAL;
 		return -1;
 	}
 
-	for (;;) {
-		last = seq;
+	for (i=0; i<cnt; i++) if (cbs[i]) {
+		if (aio_error(cbs[i]) != EINPROGRESS) return 0;
+		nzcnt++;
+		cb = cbs[i];
+	}
 
-		for (i=0; i<cnt; i++) {
-			if (cbs[i] && cbs[i]->__err != EINPROGRESS)
-				return 0;
+	if (ts) {
+		clock_gettime(CLOCK_MONOTONIC, &at);
+		at.tv_sec += ts->tv_sec;
+		if ((at.tv_nsec += ts->tv_nsec) >= 1000000000) {
+			at.tv_nsec -= 1000000000;
+			at.tv_sec++;
 		}
+	}
 
-		if (first && ts) {
-			clock_gettime(CLOCK_MONOTONIC, &at);
-			at.tv_sec += ts->tv_sec;
-			if ((at.tv_nsec += ts->tv_nsec) >= 1000000000) {
-				at.tv_nsec -= 1000000000;
-				at.tv_sec++;
-			}
-			first = 0;
-		}
+	for (;;) {
+		for (i=0; i<cnt; i++)
+			if (cbs[i] && aio_error(cbs[i]) != EINPROGRESS)
+				return 0;
 
-		ret = __timedwait(&seq, last, CLOCK_MONOTONIC,
-			ts ? &at : 0, 0, 0, 1);
+		switch (nzcnt) {
+		case 0:
+			pfut = &dummy_fut;
+			break;
+		case 1:
+			pfut = (void *)&cb->__err;
+			expect = EINPROGRESS | 0x80000000;
+			a_cas(pfut, EINPROGRESS, expect);
+			break;
+		default:
+			pfut = &__aio_fut;
+			if (!tid) tid = __pthread_self()->tid;
+			expect = a_cas(pfut, 0, tid);
+			if (!expect) expect = tid;
+			/* Need to recheck the predicate before waiting. */
+			for (i=0; i<cnt; i++)
+				if (cbs[i] && aio_error(cbs[i]) != EINPROGRESS)
+					return 0;
+			break;
+		}
 
-		if (ret == ETIMEDOUT) ret = EAGAIN;
+		ret = __timedwait(pfut, expect, CLOCK_MONOTONIC, ts?&at:0, 0, 0, 1);
 
 		if (ret) {
-			errno = ret;
+			errno = ret==ETIMEDOUT ? EAGAIN : ret;
 			return -1;
 		}
 	}