--- libaitio/src/aio.c	2012/05/14 12:49:21	1.5
+++ libaitio/src/aio.c	2012/08/02 00:38:24	1.5.14.3
@@ -3,7 +3,7 @@
 *  by Michael Pounov <misho@elwix.org>
 *
 * $Author: misho $
-* $Id: aio.c,v 1.5 2012/05/14 12:49:21 misho Exp $
+* $Id: aio.c,v 1.5.14.3 2012/08/02 00:38:24 misho Exp $
 *
 **************************************************************************
 The ELWIX and AITNET software is distributed under the following
@@ -46,6 +46,10 @@ SUCH DAMAGE.
 #include "global.h"
 
 
+/* default AIO completion request timeout */
+struct timespec aiots = { 10, 0 };
+
+
 /*
  * io_rread() - Raw VFS read function
  *
@@ -62,7 +66,7 @@ io_rread(int fd, void * __restrict buf, size_t nbytes,
 	int ret;
 #ifdef AIO_OPS
 	off_t old = 0;
-	struct aiocb acb;
+	struct aiocb acb, *racb;
 #endif
 
 	if (!buf) {
@@ -102,13 +106,15 @@ io_rread(int fd, void * __restrict buf, size_t nbytes,
 		return -1;
 	}
 
+#if 0
 	while (aio_error(&acb) == EINPROGRESS);
 	ret = aio_return(&acb);
+#endif
+	ret = aio_waitcomplete(&racb, &aiots);
 	if (ret == -1) {
 		LOGERR;
 		return -1;
-	} else
-		ret = acb.aio_nbytes;
+	}
 #else
 	ret = pread(fd, buf, nbytes, offset);
 	if (ret == -1) {
@@ -131,14 +137,14 @@ io_rread(int fd, void * __restrict buf, size_t nbytes,
  * @nbytes = Write bytes from buffer
  * @offset = Write at position, if -1 write nbytes from current position
  * @update = Update file handle position !0
- * return: -1 error or !=-1 writed bytes
+ * return: -1 error or !=-1 written bytes
  */
 inline int
 io_rwrite(int fd, void * __restrict buf, size_t nbytes, off_t offset, int update)
 {
 	int ret;
 #ifdef AIO_OPS
-	struct aiocb acb;
+	struct aiocb acb, *racb;
 #endif
 
 	if (!buf) {
@@ -167,13 +173,15 @@ io_rwrite(int fd, void * __restrict buf, size_t nbytes
 		return -1;
 	}
 
+#if 0
 	while (aio_error(&acb) == EINPROGRESS);
 	ret = aio_return(&acb);
+#endif
+	ret = aio_waitcomplete(&racb, &aiots);
 	if (ret == -1) {
 		LOGERR;
 		return -1;
-	} else
-		ret = acb.aio_nbytes;
+	}
 #else
 	ret = pwrite(fd, buf, nbytes, offset);
 	if (ret == -1) {
@@ -185,6 +193,227 @@ io_rwrite(int fd, void * __restrict buf, size_t nbytes
 	if (update)
 		lseek(fd, offset + ret, SEEK_SET);
 
+
+	return ret;
+}
+
+#ifdef AIO_OPS
+/*
+ * io_aiobulk() - AIO bulk R/W function
+ *
+ * @mode = Bulk wait mode
+ * @acbs = List of aiocb structures
+ * @nacb = Number of aiocb in list
+ * @sig = Event for completed operations, may be =NULL
+ * return: -1 error or 0 ok
+ */
+inline int
+io_aiobulk(int mode, struct aiocb ** __restrict acbs, int nacb, struct sigevent *sig)
+{
+	int ret;
+
+	if (!acbs) {
+		io_SetErr(EINVAL, "Invalid arguments");
+		return -1;
+	}
+	if (!nacb)
+		return 0;
+
+	if ((ret = lio_listio(mode, acbs, nacb, sig)) == -1)
+		LOGERR;
+
+	return ret;
+}
+#endif
+
+/*
+ * io_rreadv() - Raw VFS bulk read function
+ *
+ * @fd = File handle
+ * @bufs = Read buffers
+ * @nbufs = Number of read buffers
+ * @offset = Read from position, if -1 read nbytes from current position
+ * @update = Update file handle position !0
+ * return: -1 error or !=-1 readed bytes
+ */
+int
+io_rreadv(int fd, struct iovec * __restrict bufs, int nbufs, off_t offset, int update)
+{
+	int ret;
+#ifdef AIO_OPS
+	register int i;
+	off_t off, old = 0;
+	struct aiocb **acb;
+#endif
+
+	if (!bufs) {
+		io_SetErr(EINVAL, "Invalid arguments");
+		return -1;
+	}
+	if (!nbufs)
+		return 0;
+	if (offset == -1) {
+		offset = lseek(fd, 0, SEEK_CUR);
+		if (offset == -1) {
+			LOGERR;
+			return -1;
+		}
+	}
+
+#ifdef AIO_OPS
+	/* This made for generate EOF for file */
+	if (!update && (old = lseek(fd, 0, SEEK_CUR)) == -1) {
+		LOGERR;
+		return -1;
+	}
+	if (offset >= lseek(fd, 0, SEEK_END))
+		return 0;
+	if (!update)
+		lseek(fd, old, SEEK_SET);
+
+	acb = io_calloc(sizeof(void*), nbufs);
+	if (!acb) {
+		LOGERR;
+		return -1;
+	}
+	memset(acb, 0, sizeof(void*) * nbufs);
+	for (i = 0, off = offset; i < nbufs; off += bufs[i++].iov_len) {
+		acb[i] = io_malloc(sizeof(struct aiocb));
+		if (!acb[i]) {
+			LOGERR;
+			for (i = 0; i < nbufs; i++)
+				if (acb[i])
+					io_free(acb[i]);
+			io_free(acb);
+			return -1;
+		} else
+			memset(acb[i], 0, sizeof(struct aiocb));
+		acb[i]->aio_fildes = fd;
+		acb[i]->aio_nbytes = bufs[i].iov_len;
+		acb[i]->aio_buf = bufs[i].iov_base;
+		acb[i]->aio_offset = off;
+		acb[i]->aio_lio_opcode = LIO_READ;
+	}
+
+	ret = io_aiobulk(LIO_WAIT, acb, nbufs, NULL);
+	if (ret == -1) {
+		LOGERR;
+		for (i = 0; i < nbufs; i++)
+			if (acb[i])
+				io_free(acb[i]);
+		io_free(acb);
+		return -1;
+	}
+
+	for (i = 0; i < nbufs; i++) {
+		bufs[i].iov_len = aio_return(acb[i]);
+		offset += bufs[i].iov_len;
+		if (acb[i])
+			io_free(acb[i]);
+	}
+	io_free(acb);
+#else
+	ret = preadv(fd, bufs, nbufs, offset);
+	if (ret == -1) {
+		LOGERR;
+		return -1;
+	} else
+		offset += ret;
+#endif
+
+	if (update)
+		lseek(fd, offset, SEEK_SET);
+
+	return ret;
+}
+
+/*
+ * io_rwritev() - Raw VFS bulk write function
+ *
+ * @fd = File handle
+ * @bufs = Write buffers
+ * @nbufs = Number of write buffers
+ * @offset = Write to position, if -1 write nbytes to current position
+ * @update = Update file handle position !0
+ * return: -1 error or !=-1 written bytes
+ */
+int
+io_rwritev(int fd, struct iovec * __restrict bufs, int nbufs, off_t offset, int update)
+{
+	int ret;
+#ifdef AIO_OPS
+	register int i;
+	struct aiocb **acb;
+	off_t off;
+#endif
+
+	if (!bufs) {
+		io_SetErr(EINVAL, "Invalid arguments");
+		return -1;
+	}
+	if (!nbufs)
+		return 0;
+	if (offset == -1) {
+		offset = lseek(fd, 0, SEEK_CUR);
+		if (offset == -1) {
+			LOGERR;
+			return -1;
+		}
+	}
+
+#ifdef AIO_OPS
+	acb = io_calloc(sizeof(void*), nbufs);
+	if (!acb) {
+		LOGERR;
+		return -1;
+	} else
+		memset(acb, 0, sizeof(void*) * nbufs);
+	for (i = 0, off = offset; i < nbufs; off += bufs[i++].iov_len) {
+		acb[i] = io_malloc(sizeof(struct aiocb));
+		if (!acb[i]) {
+			LOGERR;
+			for (i = 0; i < nbufs; i++)
+				if (acb[i])
+					io_free(acb[i]);
+			io_free(acb);
+			return -1;
+		} else
+			memset(acb[i], 0, sizeof(struct aiocb));
+		acb[i]->aio_fildes = fd;
+		acb[i]->aio_nbytes = bufs[i].iov_len;
+		acb[i]->aio_buf = bufs[i].iov_base;
+		acb[i]->aio_offset = off;
+		acb[i]->aio_lio_opcode = LIO_WRITE;
+	}
+
+	ret = io_aiobulk(LIO_WAIT, acb, nbufs, NULL);
+	if (ret == -1) {
+		LOGERR;
+		for (i = 0; i < nbufs; i++)
+			if (acb[i])
+				io_free(acb[i]);
+		io_free(acb);
+		return -1;
+	}
+
+	for (i = 0; i < nbufs; i++) {
+		bufs[i].iov_len = aio_return(acb[i]);
+		offset += bufs[i].iov_len;
+		if (acb[i])
+			io_free(acb[i]);
+	}
+	io_free(acb);
+#else
+	ret = pwritev(fd, bufs, nbufs, offset);
+	if (ret == -1) {
+		LOGERR;
+		return -1;
+	} else
+		offset += ret;
+#endif
+
+	if (update)
+		lseek(fd, offset, SEEK_SET);
 
 	return ret;
 }