--- libaitsched/src/hooks.c 2011/10/04 13:29:00 1.2.2.1 +++ libaitsched/src/hooks.c 2012/01/08 03:28:26 1.3.4.1 @@ -3,7 +3,7 @@ * by Michael Pounov * * $Author: misho $ -* $Id: hooks.c,v 1.2.2.1 2011/10/04 13:29:00 misho Exp $ +* $Id: hooks.c,v 1.3.4.1 2012/01/08 03:28:26 misho Exp $ * ************************************************************************** The ELWIX and AITNET software is distributed under the following @@ -121,19 +121,19 @@ sched_hook_cancel(void *task, void *arg __unused) struct kevent chg[1]; struct timespec timeout = { 0, 0 }; - if (!t || !t->task_root || !ROOT_DATA(t->task_root) || !ROOT_DATLEN(t->task_root)) + if (!t || !TASK_ROOT(t) || !ROOT_DATA(t->task_root) || !ROOT_DATLEN(t->task_root)) return (void*) -1; else io = ROOT_DATA(t->task_root); - switch (t->task_type) { + switch (TASK_TYPE(t)) { case taskREAD: #ifdef __NetBSD__ EV_SET(&chg[0], TASK_FD(t), EVFILT_READ, EV_DELETE, 0, 0, (intptr_t) TASK_FD(t)); #else EV_SET(&chg[0], TASK_FD(t), EVFILT_READ, EV_DELETE, 0, 0, (void*) TASK_FD(t)); #endif - kevent(t->task_root->root_kq, chg, 1, NULL, 0, &timeout); + kevent(TASK_ROOT(t)->root_kq, chg, 1, NULL, 0, &timeout); FD_CLR(TASK_FD(t), &io->rfd); break; case taskWRITE: @@ -142,7 +142,7 @@ sched_hook_cancel(void *task, void *arg __unused) #else EV_SET(&chg[0], TASK_FD(t), EVFILT_WRITE, EV_DELETE, 0, 0, (void*) TASK_FD(t)); #endif - kevent(t->task_root->root_kq, chg, 1, NULL, 0, &timeout); + kevent(TASK_ROOT(t)->root_kq, chg, 1, NULL, 0, &timeout); FD_CLR(TASK_FD(t), &io->wfd); break; default: @@ -166,7 +166,7 @@ sched_hook_read(void *task, void *arg __unused) struct kevent chg[1]; struct timespec timeout = { 0, 0 }; - if (!t || !t->task_root || !ROOT_DATA(t->task_root) || !ROOT_DATLEN(t->task_root)) + if (!t || !TASK_ROOT(t) || !ROOT_DATA(t->task_root) || !ROOT_DATLEN(t->task_root)) return (void*) -1; else io = ROOT_DATA(t->task_root); @@ -181,9 +181,11 @@ sched_hook_read(void *task, void *arg __unused) #else EV_SET(&chg[0], TASK_FD(t), EVFILT_READ, EV_ADD, 0, 0, (void*) TASK_FD(t)); #endif - if (kevent(t->task_root->root_kq, chg, 1, NULL, 0, &timeout) == -1) { - if (t->task_root->root_hooks.hook_exec.exception) - t->task_root->root_hooks.hook_exec.exception(t->task_root, NULL); + if (kevent(TASK_ROOT(t)->root_kq, chg, 1, NULL, 0, &timeout) == -1) { + if (TASK_ROOT(t)->root_hooks.hook_exec.exception) + TASK_ROOT(t)->root_hooks.hook_exec.exception(TASK_ROOT(t), NULL); + else + LOGERR; return (void*) -1; } @@ -204,7 +206,7 @@ sched_hook_write(void *task, void *arg __unused) struct kevent chg[1]; struct timespec timeout = { 0, 0 }; - if (!t || !t->task_root || !ROOT_DATA(t->task_root) || !ROOT_DATLEN(t->task_root)) + if (!t || !TASK_ROOT(t) || !ROOT_DATA(t->task_root) || !ROOT_DATLEN(t->task_root)) return (void*) -1; else io = ROOT_DATA(t->task_root); @@ -219,9 +221,11 @@ sched_hook_write(void *task, void *arg __unused) #else EV_SET(&chg[0], TASK_FD(t), EVFILT_WRITE, EV_ADD, 0, 0, (void*) TASK_FD(t)); #endif - if (kevent(t->task_root->root_kq, chg, 1, NULL, 0, &timeout) == -1) { - if (t->task_root->root_hooks.hook_exec.exception) - t->task_root->root_hooks.hook_exec.exception(t->task_root, NULL); + if (kevent(TASK_ROOT(t)->root_kq, chg, 1, NULL, 0, &timeout) == -1) { + if (TASK_ROOT(t)->root_hooks.hook_exec.exception) + TASK_ROOT(t)->root_hooks.hook_exec.exception(TASK_ROOT(t), NULL); + else + LOGERR; return (void*) -1; } @@ -252,15 +256,39 @@ sched_hook_fetch(void *root, void *arg __unused) /* get new task by queue priority */ retry: while ((task = TAILQ_FIRST(&r->root_event))) { +#ifdef HAVE_LIBPTHREAD + pthread_mutex_lock(&r->root_mtx[taskEVENT]); +#endif TAILQ_REMOVE(&r->root_event, task, task_node); +#ifdef HAVE_LIBPTHREAD + pthread_mutex_unlock(&r->root_mtx[taskEVENT]); +#endif task->task_type = taskUNUSE; +#ifdef HAVE_LIBPTHREAD + pthread_mutex_lock(&r->root_mtx[taskUNUSE]); +#endif TAILQ_INSERT_TAIL(&r->root_unuse, task, task_node); +#ifdef HAVE_LIBPTHREAD + pthread_mutex_unlock(&r->root_mtx[taskUNUSE]); +#endif return task; } while ((task = TAILQ_FIRST(&r->root_ready))) { +#ifdef HAVE_LIBPTHREAD + pthread_mutex_lock(&r->root_mtx[taskREADY]); +#endif TAILQ_REMOVE(&r->root_ready, task, task_node); +#ifdef HAVE_LIBPTHREAD + pthread_mutex_unlock(&r->root_mtx[taskREADY]); +#endif task->task_type = taskUNUSE; +#ifdef HAVE_LIBPTHREAD + pthread_mutex_lock(&r->root_mtx[taskUNUSE]); +#endif TAILQ_INSERT_TAIL(&r->root_unuse, task, task_node); +#ifdef HAVE_LIBPTHREAD + pthread_mutex_unlock(&r->root_mtx[taskUNUSE]); +#endif return task; } @@ -310,9 +338,11 @@ retry: } else /* wait INFTIM */ timeout = NULL; if ((en = kevent(r->root_kq, NULL, 0, res, KQ_EVENTS, timeout)) == -1) { - if (r->root_hooks.hook_exec.exception) + if (r->root_hooks.hook_exec.exception) { if (r->root_hooks.hook_exec.exception(r, NULL)) - return NULL; /* exit from scheduler */ + return NULL; + } else + LOGERR; #ifdef NDEBUG /* kevent no exit by error, if non-debug version */ goto retry; @@ -331,57 +361,108 @@ retry: switch (res[i].filter) { case EVFILT_READ: TAILQ_FOREACH(task, &r->root_read, task_node) { - if (TASK_FD(task) != ((int) res[i].udata)) + if (TASK_FD(task) != ((intptr_t) res[i].udata)) continue; /* remove read handle */ io = ROOT_DATA(task->task_root); FD_CLR(TASK_FD(task), &io->rfd); +#ifdef HAVE_LIBPTHREAD + pthread_mutex_lock(&r->root_mtx[taskREAD]); +#endif TAILQ_REMOVE(&r->root_read, task, task_node); +#ifdef HAVE_LIBPTHREAD + pthread_mutex_unlock(&r->root_mtx[taskREAD]); +#endif if (r->root_hooks.hook_exec.exception && res[i].flags & EV_EOF) { if (r->root_hooks.hook_exec.exception(r, (void*) EV_EOF)) { task->task_type = taskUNUSE; +#ifdef HAVE_LIBPTHREAD + pthread_mutex_lock(&r->root_mtx[taskUNUSE]); +#endif TAILQ_INSERT_TAIL(&r->root_unuse, task, task_node); +#ifdef HAVE_LIBPTHREAD + pthread_mutex_unlock(&r->root_mtx[taskUNUSE]); +#endif } else { task->task_type = taskREADY; +#ifdef HAVE_LIBPTHREAD + pthread_mutex_lock(&r->root_mtx[taskREADY]); +#endif TAILQ_INSERT_TAIL(&r->root_ready, task, task_node); +#ifdef HAVE_LIBPTHREAD + pthread_mutex_unlock(&r->root_mtx[taskREADY]); +#endif } } else { task->task_type = taskREADY; +#ifdef HAVE_LIBPTHREAD + pthread_mutex_lock(&r->root_mtx[taskREADY]); +#endif TAILQ_INSERT_TAIL(&r->root_ready, task, task_node); +#ifdef HAVE_LIBPTHREAD + pthread_mutex_unlock(&r->root_mtx[taskREADY]); +#endif } break; } break; case EVFILT_WRITE: TAILQ_FOREACH(task, &r->root_write, task_node) { - if (TASK_FD(task) != ((int) res[i].udata)) + if (TASK_FD(task) != ((intptr_t) res[i].udata)) continue; /* remove write handle */ io = ROOT_DATA(task->task_root); FD_CLR(TASK_FD(task), &io->wfd); +#ifdef HAVE_LIBPTHREAD + pthread_mutex_lock(&r->root_mtx[taskWRITE]); +#endif TAILQ_REMOVE(&r->root_write, task, task_node); +#ifdef HAVE_LIBPTHREAD + pthread_mutex_unlock(&r->root_mtx[taskWRITE]); +#endif if (r->root_hooks.hook_exec.exception && res[i].flags & EV_EOF) { if (r->root_hooks.hook_exec.exception(r, (void*) EV_EOF)) { task->task_type = taskUNUSE; +#ifdef HAVE_LIBPTHREAD + pthread_mutex_lock(&r->root_mtx[taskUNUSE]); +#endif TAILQ_INSERT_TAIL(&r->root_unuse, task, task_node); +#ifdef HAVE_LIBPTHREAD + pthread_mutex_unlock(&r->root_mtx[taskUNUSE]); +#endif } else { task->task_type = taskREADY; +#ifdef HAVE_LIBPTHREAD + pthread_mutex_lock(&r->root_mtx[taskREADY]); +#endif TAILQ_INSERT_TAIL(&r->root_ready, task, task_node); +#ifdef HAVE_LIBPTHREAD + pthread_mutex_unlock(&r->root_mtx[taskREADY]); +#endif } } else { task->task_type = taskREADY; +#ifdef HAVE_LIBPTHREAD + pthread_mutex_lock(&r->root_mtx[taskREADY]); +#endif TAILQ_INSERT_TAIL(&r->root_ready, task, task_node); +#ifdef HAVE_LIBPTHREAD + pthread_mutex_unlock(&r->root_mtx[taskREADY]); +#endif } break; } break; } - if (kevent(r->root_kq, evt, 1, NULL, 0, &nw) == -1) - if (r->root_hooks.hook_exec.exception) + if (kevent(r->root_kq, evt, 1, NULL, 0, &nw) == -1) { + if (r->root_hooks.hook_exec.exception) { if (r->root_hooks.hook_exec.exception(r, NULL)) - return NULL; /* exit from scheduler */ + return NULL; + } else + LOGERR; + } } /* timer update & put in ready queue */ @@ -391,9 +472,21 @@ retry: TAILQ_FOREACH(task, &r->root_timer, task_node) if (timercmp(&now, &TASK_TV(task), -) >= 0) { +#ifdef HAVE_LIBPTHREAD + pthread_mutex_lock(&r->root_mtx[taskTIMER]); +#endif TAILQ_REMOVE(&r->root_timer, task, task_node); +#ifdef HAVE_LIBPTHREAD + pthread_mutex_unlock(&r->root_mtx[taskTIMER]); +#endif task->task_type = taskREADY; +#ifdef HAVE_LIBPTHREAD + pthread_mutex_lock(&r->root_mtx[taskREADY]); +#endif TAILQ_INSERT_TAIL(&r->root_ready, task, task_node); +#ifdef HAVE_LIBPTHREAD + pthread_mutex_unlock(&r->root_mtx[taskREADY]); +#endif } /* put eventlo priority task to ready queue, if there is no ready task or @@ -402,9 +495,21 @@ retry: if (!TAILQ_FIRST(&r->root_ready) || r->root_eventlo_miss > MAX_EVENTLO_MISS) { r->root_eventlo_miss = 0; +#ifdef HAVE_LIBPTHREAD + pthread_mutex_lock(&r->root_mtx[taskEVENTLO]); +#endif TAILQ_REMOVE(&r->root_eventlo, task, task_node); +#ifdef HAVE_LIBPTHREAD + pthread_mutex_unlock(&r->root_mtx[taskEVENTLO]); +#endif task->task_type = taskREADY; +#ifdef HAVE_LIBPTHREAD + pthread_mutex_lock(&r->root_mtx[taskREADY]); +#endif TAILQ_INSERT_TAIL(&r->root_ready, task, task_node); +#ifdef HAVE_LIBPTHREAD + pthread_mutex_unlock(&r->root_mtx[taskREADY]); +#endif } else r->root_eventlo_miss++; } else @@ -413,16 +518,28 @@ retry: /* OK, lets get ready task !!! */ if (!(task = TAILQ_FIRST(&r->root_ready))) goto retry; +#ifdef HAVE_LIBPTHREAD + pthread_mutex_lock(&r->root_mtx[taskREADY]); +#endif TAILQ_REMOVE(&r->root_ready, task, task_node); +#ifdef HAVE_LIBPTHREAD + pthread_mutex_unlock(&r->root_mtx[taskREADY]); +#endif task->task_type = taskUNUSE; +#ifdef HAVE_LIBPTHREAD + pthread_mutex_lock(&r->root_mtx[taskUNUSE]); +#endif TAILQ_INSERT_TAIL(&r->root_unuse, task, task_node); +#ifdef HAVE_LIBPTHREAD + pthread_mutex_unlock(&r->root_mtx[taskUNUSE]); +#endif return task; } /* * sched_hook_exception() - Default EXCEPTION hook * @root = root task - * @arg = may be EV_EOF + * @arg = custom handling: if arg == EV_EOF or other value; default: arg == NULL log errno * return: <0 errors and 0 ok */ void * @@ -433,15 +550,18 @@ sched_hook_exception(void *root, void *arg) if (!r || !ROOT_DATA(r) || !ROOT_DATLEN(r)) return NULL; + /* custom exception handling ... */ if (arg) { if (arg == (void*) EV_EOF) return NULL; - return (void*) -1; /* raise error */ - } else - LOGERR; + return (void*) -1; /* raise scheduler error!!! */ + } - if (errno == EINTR) - return (void*) -1; /* raise error */ + /* if error hook exists */ + if (r->root_hooks.hook_root.error) + return (r->root_hooks.hook_root.error(root, (void*) ((intptr_t) errno))); + /* default case! */ + LOGERR; return NULL; }