--- libaitsched/src/tasks.c 2012/08/01 22:12:39 1.10.2.5 +++ libaitsched/src/tasks.c 2012/08/02 09:19:31 1.10.2.6 @@ -3,7 +3,7 @@ * by Michael Pounov * * $Author: misho $ -* $Id: tasks.c,v 1.10.2.5 2012/08/01 22:12:39 misho Exp $ +* $Id: tasks.c,v 1.10.2.6 2012/08/02 09:19:31 misho Exp $ * ************************************************************************** The ELWIX and AITNET software is distributed under the following @@ -482,6 +482,7 @@ schedAlarm(sched_root_task_t * __restrict root, sched_ return task; } +#ifdef EVFILT_AIO /* * schedAIO() - Add AIO task to scheduler queue * @@ -497,14 +498,10 @@ sched_task_t * schedAIO(sched_root_task_t * __restrict root, sched_task_func_t func, void *arg, struct aiocb * __restrict acb, void *opt_data, size_t opt_dlen) { -#ifndef EVFILT_AIO - sched_SetErr(ENOTSUP, "Not supported kevent() filter"); - return NULL; -#else sched_task_t *task; void *ptr; - if (!root || !func) + if (!root || !func || !acb || !opt_dlen) return NULL; /* get new task */ @@ -538,7 +535,6 @@ schedAIO(sched_root_task_t * __restrict root, sched_ta task = _sched_unuseTask(task); return task; -#endif } /* @@ -556,14 +552,10 @@ inline sched_task_t * schedAIORead(sched_root_task_t * __restrict root, sched_task_func_t func, void *arg, int fd, void *buffer, size_t buflen) { -#ifndef EVFILT_AIO - sched_SetErr(ENOTSUP, "Not supported kevent() filter"); - return NULL; -#else struct aiocb *acb; off_t off = 0; - if (!root || !func) + if (!root || !func || !buffer || !buflen) return NULL; else memset(buffer, 0, buflen); @@ -595,7 +587,6 @@ schedAIORead(sched_root_task_t * __restrict root, sche } return schedAIO(root, func, arg, acb, buffer, buflen); -#endif } /* @@ -613,14 +604,10 @@ inline sched_task_t * schedAIOWrite(sched_root_task_t * __restrict root, sched_task_func_t func, void *arg, int fd, void *buffer, size_t buflen) { -#ifndef EVFILT_AIO - sched_SetErr(ENOTSUP, "Not supported kevent() filter"); - return NULL; -#else struct aiocb *acb; off_t off = 0; - if (!root || !func) + if (!root || !func || !buffer || !buflen) return NULL; if (!(acb = malloc(sizeof(struct aiocb)))) { @@ -650,8 +637,138 @@ schedAIOWrite(sched_root_task_t * __restrict root, sch } return schedAIO(root, func, arg, acb, buffer, buflen); -#endif } + +#ifdef EVFILT_LIO +/* + * schedLIORead() - Add list of AIO read tasks to scheduler queue + * + * @root = root task + * @func = task execution function + * @arg = 1st func argument + * @fd = file descriptor + * @bufs = Buffer's list + * @nbufs = Number of Buffers + * return: NULL error or !=NULL new queued task + */ +inline sched_task_t * +schedLIORead(sched_root_task_t * __restrict root, sched_task_func_t func, void *arg, int fd, + struct iovec *bufs, size_t nbufs) +{ + struct sigevent sig; + struct aiocb **acb; + off_t off; + register int i; + + if (!root || !func || !bufs || !nbufs) + return NULL; + + off = lseek(fd, 0, SEEK_CUR); + if (off == -1) { + LOGERR; + return NULL; + } + + if (!(acb = calloc(sizeof(void*), nbufs))) { + LOGERR; + return NULL; + } else + memset(acb, 0, sizeof(void*) * nbufs); + for (i = 0; i < nbufs; off += bufs[i++].iov_len) { + acb[i] = malloc(sizeof(struct aiocb)); + if (!acb[i]) { + LOGERR; + for (i = 0; i < nbufs; i++) + if (acb[i]) + free(acb[i]); + free(acb); + return NULL; + } 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; + } + memset(&sig, 0, sizeof sig); + sig.sigev_notify = SIGEV_KEVENT; + sig.sigev_notify_kqueue = root->root_kq; + sig.sigev_value.sival_ptr = acb; + + if (lio_listio(LIO_NOWAIT, acb, nbufs, &sig)) { + LOGERR; + return NULL; + } + + return schedAIO(root, func, arg, (void*) acb, bufs, nbufs); +} + +/* + * schedLIOWrite() - Add list of AIO write tasks to scheduler queue + * + * @root = root task + * @func = task execution function + * @arg = 1st func argument + * @fd = file descriptor + * @bufs = Buffer's list + * @nbufs = Number of Buffers + * return: NULL error or !=NULL new queued task + */ +inline sched_task_t * +schedLIOWrite(sched_root_task_t * __restrict root, sched_task_func_t func, void *arg, int fd, + struct iovec *bufs, size_t nbufs) +{ + struct sigevent sig; + struct aiocb **acb; + off_t off; + register int i; + + if (!root || !func || !bufs || !nbufs) + return NULL; + + off = lseek(fd, 0, SEEK_CUR); + if (off == -1) { + LOGERR; + return NULL; + } + + if (!(acb = calloc(sizeof(void*), nbufs))) { + LOGERR; + return NULL; + } else + memset(acb, 0, sizeof(void*) * nbufs); + for (i = 0; i < nbufs; off += bufs[i++].iov_len) { + acb[i] = malloc(sizeof(struct aiocb)); + if (!acb[i]) { + LOGERR; + for (i = 0; i < nbufs; i++) + if (acb[i]) + free(acb[i]); + free(acb); + return NULL; + } 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; + } + memset(&sig, 0, sizeof sig); + sig.sigev_notify = SIGEV_KEVENT; + sig.sigev_notify_kqueue = root->root_kq; + sig.sigev_value.sival_ptr = acb; + + if (lio_listio(LIO_NOWAIT, acb, nbufs, &sig)) { + LOGERR; + return NULL; + } + + return schedAIO(root, func, arg, (void*) acb, bufs, nbufs); +} +#endif /* EVFILT_LIO */ +#endif /* EVFILT_AIO */ /* * schedTimer() - Add TIMER task to scheduler queue