Add io_uring_cqe_seen()

There's a failure case where an application gets a cqe entry, but
the kernel can then overwrite it before the application is done
reading it. This can happen since the io_uring_{get,wait}_completion()
interface both returns a CQE pointer AND increments the ring index.
If the kernel reuses this entry before the applications is done reading
it, the contents may be corrupted.

Remove the CQ head increment from the CQE retrieval, and put it into
a separate helper, io_uring_cqe_seen(). The application must call this
helper when it got a new CQE entry through one of the above calls, and
it's now done reading it.

Signed-off-by: Jens Axboe <axboe@kernel.dk>

test/cq-full.c

@@ -67,6 +67,7 @@ int main(int argc, char *argv[])
 			printf("wait completion %d\n", ret);
 			goto err;
 		}
+		io_uring_cqe_seen(&ring, cqe);
 		if (!cqe)
 			break;
 		i++;

test/fsync.c

@@ -45,6 +45,7 @@ static int test_single_fsync(struct io_uring *ring)
 		goto err;
 	}
 
+	io_uring_cqe_seen(ring, cqe);
 	unlink(buf);
 	return 0;
 err:
@@ -122,6 +123,7 @@ static int test_barrier_fsync(struct io_uring *ring)
 				goto err;
 			}
 		}
+		io_uring_cqe_seen(ring, cqe);
 	}
 
 	unlink("testfile");

test/io_uring-cp.c

@@ -177,6 +177,7 @@ static int copy_file(struct io_uring *ring, off_t insize)
 			if (cqe->res < 0) {
 				if (cqe->res == -EAGAIN) {
 					queue_prepped(ring, data);
+					io_uring_cqe_seen(ring, cqe);
 					continue;
 				}
 				fprintf(stderr, "cqe failed: %s\n",
@@ -188,6 +189,7 @@ static int copy_file(struct io_uring *ring, off_t insize)
 				data->iov.iov_len -= cqe->res;
 				data->offset += cqe->res;
 				queue_prepped(ring, data);
+				io_uring_cqe_seen(ring, cqe);
 				continue;
 			}
 
@@ -204,6 +206,7 @@ static int copy_file(struct io_uring *ring, off_t insize)
 				free(data);
 				writes--;
 			}
+			io_uring_cqe_seen(ring, cqe);
 		}
 	}
 

test/io_uring-test.c

@@ -74,10 +74,14 @@ int main(int argc, char *argv[])
 		}
 
 		done++;
+		ret = 0;
 		if (cqe->res != 4096) {
 			fprintf(stderr, "ret=%d, wanted 4096\n", cqe->res);
-			return 1;
+			ret = 1;
 		}
+		io_uring_cqe_seen(&ring, cqe);
+		if (ret)
+			break;
 	}
 
 	printf("Submitted=%d, completed=%d\n", pending, done);

test/io_uring_enter.c

@@ -158,6 +158,7 @@ reap_events(struct io_uring *ring, unsigned nr)
 		free(iov->iov_base);
 		free(iov);
 		left--;
+		io_uring_cqe_seen(ring, cqe);
 
 		gettimeofday(&now, NULL);
 		timersub(&now, &start, &elapsed);

test/io_uring_register.c

@@ -428,13 +428,15 @@ ioring_poll(struct io_uring *ring, int fd, int fixed)
 		printf("io_uring_wait_completion failed with %d\n", ret);
 		return 1;
 	}
+	ret = 0;
 	if (cqe->res != POLLOUT) {
 		printf("io_uring_wait_completion: expected 0x%.8x, got 0x%.8x\n",
 		       POLLOUT, cqe->res);
-		return 1;
+		ret = 1;
 	}
 
-	return 0;
+	io_uring_cqe_seen(ring, cqe);
+	return ret;
 }
 
 int

test/nop.c

@@ -37,6 +37,7 @@ static int test_single_nop(struct io_uring *ring)
 		goto err;
 	}
 
+	io_uring_cqe_seen(ring, cqe);
 	return 0;
 err:
 	return 1;
@@ -75,6 +76,7 @@ static int test_barrier_nop(struct io_uring *ring)
 			printf("wait completion %d\n", ret);
 			goto err;
 		}
+		io_uring_cqe_seen(ring, cqe);
 	}
 
 	return 0;

test/poll-cancel.c

@@ -99,6 +99,7 @@ int main(int argc, char *argv[])
 							pd->is_cancel, (long) cqe->res);
 		return 1;
 	}
+	io_uring_cqe_seen(&ring, cqe);
 
 	ret = io_uring_wait_completion(&ring, &cqe);
 	if (ret < 0) {
@@ -113,5 +114,6 @@ int main(int argc, char *argv[])
 		return 1;
 	}
 
+	io_uring_cqe_seen(&ring, cqe);
 	return 0;
 }

test/poll.c

@@ -74,6 +74,7 @@ int main(int argc, char *argv[])
 				printf("child: wait completion %d\n", ret);
 				break;
 			}
+			io_uring_cqe_seen(&ring, cqe);
 		} while (ret != 0);
 
 		if (ret < 0) {