--- libaitsched/src/aitsched.c 2011/08/13 17:28:17 1.1.1.1.2.4 +++ libaitsched/src/aitsched.c 2023/08/31 15:33:51 1.35 @@ -3,7 +3,7 @@ * by Michael Pounov * * $Author: misho $ -* $Id: aitsched.c,v 1.1.1.1.2.4 2011/08/13 17:28:17 misho Exp $ +* $Id: aitsched.c,v 1.35 2023/08/31 15:33:51 misho Exp $ * ************************************************************************** The ELWIX and AITNET software is distributed under the following @@ -12,7 +12,7 @@ terms: All of the documentation and software included in the ELWIX and AITNET Releases is copyrighted by ELWIX - Sofia/Bulgaria -Copyright 2004, 2005, 2006, 2007, 2008, 2009, 2010, 2011 +Copyright 2004 - 2023 by Michael Pounov . All rights reserved. Redistribution and use in source and binary forms, with or without @@ -56,21 +56,21 @@ char sched_Error[STRSIZ]; // sched_GetErrno() Get error code of last operation -inline int +int sched_GetErrno() { return sched_Errno; } // sched_GetError() Get error text of last operation -inline const char * +const char * sched_GetError() { return sched_Error; } // sched_SetErr() Set error to variables for internal use!!! -inline void +void sched_SetErr(int eno, char *estr, ...) { va_list lst; @@ -82,18 +82,91 @@ sched_SetErr(int eno, char *estr, ...) va_end(lst); } + +/* string support functions directly imported from OpenBSD */ + +#ifndef HAVE_STRLCAT +/* + * Appends src to string dst of size siz (unlike strncat, siz is the + * full size of dst, not space left). At most siz-1 characters + * will be copied. Always NUL terminates (unless siz <= strlen(dst)). + * Returns strlen(src) + MIN(siz, strlen(initial dst)). + * If retval >= siz, truncation occurred. + */ +size_t +strlcat(char * __restrict dst, const char * __restrict src, size_t siz) +{ + char *d = dst; + const char *s = src; + size_t n = siz; + size_t dlen; + + /* Find the end of dst and adjust bytes left but don't go past end */ + while (n-- != 0 && *d != '\0') + d++; + dlen = d - dst; + n = siz - dlen; + + if (n == 0) + return(dlen + strlen(s)); + while (*s != '\0') { + if (n != 1) { + *d++ = *s; + n--; + } + s++; + } + *d = '\0'; + + return(dlen + (s - src)); /* count does not include NUL */ +} +#endif +#ifndef HAVE_STRLCPY +/* + * Copy src to string dst of size siz. At most siz-1 characters + * will be copied. Always NUL terminates (unless siz == 0). + * Returns strlen(src); if retval >= siz, truncation occurred. + */ +size_t +strlcpy(char * __restrict dst, const char * __restrict src, size_t siz) +{ + char *d = dst; + const char *s = src; + size_t n = siz; + + /* Copy as many bytes as will fit */ + if (n != 0) { + while (--n != 0) { + if ((*d++ = *s++) == '\0') + break; + } + } + + /* Not enough room in dst, add NUL and traverse rest of src */ + if (n == 0) { + if (siz != 0) + *d = '\0'; /* NUL-terminate dst */ + while (*s++) + ; + } + + return(s - src - 1); /* count does not include NUL */ +} +#endif + + /* Init and prepare scheduler functions */ /* * schedRegisterHooks() - Register IO handles and bind tasks to it + * * @root = root task * return: -1 error or 0 ok */ int schedRegisterHooks(sched_root_task_t * __restrict root) { - if (!root || (root->root_data.iov_base && root->root_data.iov_len)) - return -1; + assert(root); if (root->root_hooks.hook_root.fini) root->root_hooks.hook_root.fini(root, NULL); @@ -101,9 +174,25 @@ schedRegisterHooks(sched_root_task_t * __restrict root root->root_hooks.hook_add.read = sched_hook_read; root->root_hooks.hook_add.write = sched_hook_write; +#if defined(HAVE_TIMER_CREATE) && defined(HAVE_TIMER_SETTIME) && defined(HAVE_TIMER_DELETE) + root->root_hooks.hook_add.rtc = sched_hook_rtc; +#endif +#if SUP_ENABLE == KQ_SUPPORT + root->root_hooks.hook_add.alarm = sched_hook_alarm; + root->root_hooks.hook_add.node = sched_hook_node; + root->root_hooks.hook_add.proc = sched_hook_proc; +#ifdef EVFILT_USER + root->root_hooks.hook_add.user = sched_hook_user; +#endif +#endif /* KQ_SUPPORT */ +#ifdef HAVE_LIBPTHREAD + root->root_hooks.hook_add.thread = sched_hook_thread; +#endif + root->root_hooks.hook_add.signal = sched_hook_signal; root->root_hooks.hook_exec.cancel = sched_hook_cancel; root->root_hooks.hook_exec.fetch = sched_hook_fetch; + root->root_hooks.hook_exec.exception = sched_hook_exception; root->root_hooks.hook_root.init = sched_hook_init; root->root_hooks.hook_root.fini = sched_hook_fini; @@ -112,6 +201,7 @@ schedRegisterHooks(sched_root_task_t * __restrict root /* * schedInit() - Init scheduler + * * @data = optional data if !=NULL * @datlen = data len if data is set * return: allocated root task if ok or NULL error @@ -121,25 +211,83 @@ schedInit(void ** __restrict data, size_t datlen) { sched_root_task_t *root = NULL; int (*func)(sched_root_task_t *); +#if SUP_ENABLE == KQ_SUPPORT + sigset_t ss; +#endif +#ifdef HAVE_LIBPTHREAD + register int i; +#endif - root = malloc(sizeof(sched_root_task_t)); + root = e_malloc(sizeof(sched_root_task_t)); if (!root) { LOGERR; } else { memset(root, 0, sizeof(sched_root_task_t)); + + /* set default maximum regular task hit misses */ + root->root_miss = MAX_TASK_MISS; + + /* INFINIT polling period by default */ + sched_timespecinf(&root->root_poll); + +#ifdef HAVE_LIBPTHREAD + if (pthread_mutex_init(&root->root_sigmtx, NULL)) { + LOGERR; + e_free(root); + return NULL; + } + + for (i = 0; i < taskMAX; i++) + if ((errno = pthread_mutex_init(&root->root_mtx[i], NULL))) { + LOGERR; + while (i) + pthread_mutex_destroy(&root->root_mtx[--i]); + e_free(root); + return NULL; + } + + for (i = 0; i < taskMAX; i++) + pthread_mutex_lock(&root->root_mtx[i]); +#endif + TAILQ_INIT(&root->root_read); TAILQ_INIT(&root->root_write); TAILQ_INIT(&root->root_timer); + TAILQ_INIT(&root->root_alarm); + TAILQ_INIT(&root->root_rtc); + TAILQ_INIT(&root->root_node); + TAILQ_INIT(&root->root_proc); + TAILQ_INIT(&root->root_signal); + TAILQ_INIT(&root->root_aio); + TAILQ_INIT(&root->root_lio); + TAILQ_INIT(&root->root_user); TAILQ_INIT(&root->root_event); - TAILQ_INIT(&root->root_eventlo); + TAILQ_INIT(&root->root_task); + TAILQ_INIT(&root->root_suspend); TAILQ_INIT(&root->root_ready); TAILQ_INIT(&root->root_unuse); + TAILQ_INIT(&root->root_thread); +#ifdef HAVE_LIBPTHREAD + for (i = 0; i < taskMAX; i++) + pthread_mutex_unlock(&root->root_mtx[i]); + +#if SUP_ENABLE == KQ_SUPPORT + sigfillset(&ss); + pthread_sigmask(SIG_BLOCK, &ss, &root->root_oldset); +#else + pthread_sigmask(SIG_BLOCK, NULL, &root->root_oldset); +#endif +#else + sigprocmask(SIG_BLOCK, NULL, &root->root_oldset); +#endif + if (data && *data) { if (datlen) { root->root_data.iov_base = *data; root->root_data.iov_len = datlen; - } else { + } else { /* if datlen == 0, switch to callbacks init mode */ + /* little hack :) for correct initialization of scheduler */ func = (int(*)(sched_root_task_t*)) data; func(root); } @@ -154,66 +302,159 @@ schedInit(void ** __restrict data, size_t datlen) /* * schedEnd() - End scheduler & free all resources + * * @root = root task * return: -1 error or 0 ok */ int -schedEnd(sched_root_task_t * __restrict root) +schedEnd(sched_root_task_t ** __restrict root) { - sched_task_t *task; + sched_task_t *task, *tmp; +#ifdef HAVE_LIBPTHREAD + register int i; +#endif - if (!root) + if (!root || !*root) return -1; - TAILQ_FOREACH(task, &root->root_read, task_node) { +#if 0 + TAILQ_FOREACH_SAFE(task, &(*root)->root_read, task_node, tmp) + printf("read=%p\n", task); + TAILQ_FOREACH_SAFE(task, &(*root)->root_write, task_node, tmp) + printf("write=%p\n", task); + TAILQ_FOREACH_SAFE(task, &(*root)->root_timer, task_node, tmp) + printf("timer=%p\n", task); + TAILQ_FOREACH_SAFE(task, &(*root)->root_alarm, task_node, tmp) + printf("alarm=%p\n", task); + TAILQ_FOREACH_SAFE(task, &(*root)->root_rtc, task_node, tmp) + printf("rtc=%p\n", task); + TAILQ_FOREACH_SAFE(task, &(*root)->root_node, task_node, tmp) + printf("node=%p\n", task); + TAILQ_FOREACH_SAFE(task, &(*root)->root_proc, task_node, tmp) + printf("proc=%p\n", task); + TAILQ_FOREACH_SAFE(task, &(*root)->root_signal, task_node, tmp) + printf("signal=%p\n", task); + TAILQ_FOREACH_SAFE(task, &(*root)->root_aio, task_node, tmp) + printf("aio=%p\n", task); + TAILQ_FOREACH_SAFE(task, &(*root)->root_lio, task_node, tmp) + printf("lio=%p\n", task); + TAILQ_FOREACH_SAFE(task, &(*root)->root_user, task_node, tmp) + printf("user=%p\n", task); + TAILQ_FOREACH_SAFE(task, &(*root)->root_event, task_node, tmp) + printf("event=%p\n", task); + TAILQ_FOREACH_SAFE(task, &(*root)->root_suspend, task_node, tmp) + printf("suspend=%p\n", task); + TAILQ_FOREACH_SAFE(task, &(*root)->root_ready, task_node, tmp) + printf("ready=%p\n", task); + TAILQ_FOREACH_SAFE(task, &(*root)->root_thread, task_node, tmp) + printf("thread=%p\n", task); + TAILQ_FOREACH_SAFE(task, &(*root)->root_task, task_node, tmp) + printf("task=%p\n", task); + TAILQ_FOREACH_SAFE(task, &(*root)->root_unuse, task_node, tmp) + printf("unuse=%p\n", task); + fflush(stdout); +#endif + + TAILQ_FOREACH_SAFE(task, &(*root)->root_read, task_node, tmp) schedCancel(task); - } - TAILQ_FOREACH(task, &root->root_write, task_node) { + TAILQ_FOREACH_SAFE(task, &(*root)->root_write, task_node, tmp) schedCancel(task); - } - TAILQ_FOREACH(task, &root->root_timer, task_node) { + TAILQ_FOREACH_SAFE(task, &(*root)->root_timer, task_node, tmp) schedCancel(task); - } - TAILQ_FOREACH(task, &root->root_event, task_node) { + TAILQ_FOREACH_SAFE(task, &(*root)->root_alarm, task_node, tmp) schedCancel(task); - } - TAILQ_FOREACH(task, &root->root_ready, task_node) { + TAILQ_FOREACH_SAFE(task, &(*root)->root_rtc, task_node, tmp) schedCancel(task); + TAILQ_FOREACH_SAFE(task, &(*root)->root_node, task_node, tmp) + schedCancel(task); + TAILQ_FOREACH_SAFE(task, &(*root)->root_proc, task_node, tmp) + schedCancel(task); + TAILQ_FOREACH_SAFE(task, &(*root)->root_signal, task_node, tmp) + schedCancel(task); + TAILQ_FOREACH_SAFE(task, &(*root)->root_aio, task_node, tmp) + schedCancel(task); + TAILQ_FOREACH_SAFE(task, &(*root)->root_lio, task_node, tmp) + schedCancel(task); + TAILQ_FOREACH_SAFE(task, &(*root)->root_user, task_node, tmp) + schedCancel(task); + TAILQ_FOREACH_SAFE(task, &(*root)->root_event, task_node, tmp) + schedCancel(task); + TAILQ_FOREACH_SAFE(task, &(*root)->root_suspend, task_node, tmp) + schedCancel(task); + TAILQ_FOREACH_SAFE(task, &(*root)->root_ready, task_node, tmp) + schedCancel(task); + TAILQ_FOREACH_SAFE(task, &(*root)->root_thread, task_node, tmp) + schedCancel(task); + TAILQ_FOREACH_SAFE(task, &(*root)->root_task, task_node, tmp) + schedCancel(task); + + SCHED_QLOCK((*root), taskUNUSE); + TAILQ_FOREACH_SAFE(task, &(*root)->root_unuse, task_node, tmp) { + TAILQ_REMOVE(&(*root)->root_unuse, task, task_node); + e_free(task); } + SCHED_QUNLOCK((*root), taskUNUSE); - while ((task = TAILQ_FIRST(&root->root_unuse))) { - TAILQ_REMOVE(&root->root_unuse, task, task_node); - free(task); + if ((*root)->root_hooks.hook_root.fini) + (*root)->root_hooks.hook_root.fini(*root, NULL); + +#ifdef HAVE_LIBPTHREAD + for (i = 0; i < taskMAX; i++) { + SCHED_QTRYLOCK(*root, i); + SCHED_QUNLOCK(*root, i); + pthread_mutex_destroy(&(*root)->root_mtx[i]); } - if (root->root_hooks.hook_root.fini) - root->root_hooks.hook_root.fini(root, NULL); + if ((*root)->root_sigthr) { + pthread_cancel((*root)->root_sigthr); +#ifdef __linux__ + (*root)->root_sigthr = 0L; +#else + (*root)->root_sigthr = NULL; +#endif + pthread_sigmask(SIG_SETMASK, &(*root)->root_oldset, NULL); + pthread_mutex_unlock(&(*root)->root_sigmtx); + } + pthread_mutex_destroy(&(*root)->root_sigmtx); +#else + sigprocmask(SIG_SETMASK, &(*root)->root_oldset, NULL); +#endif - free(root); + e_free(*root); + *root = NULL; return 0; } /* * schedCall() - Call task execution function + * * @task = current task * return: !=NULL error or =NULL ok */ -inline void * +void * schedCall(sched_task_t * __restrict task) { + void *ptr = (void*) -1; + if (!task) - return (void*) -1; + return ptr; - task->task_id++; - return task->task_func(task); + if (!TASK_ISLOCKED(task)) + TASK_LOCK(task); + + ptr = task->task_func(task); + + TASK_UNLOCK(task); + return ptr; } /* * schedFetch() - Fetch ready task + * * @root = root task * return: =NULL error or !=NULL ready task */ -inline void * +void * schedFetch(sched_root_task_t * __restrict root) { void *ptr; @@ -230,7 +471,275 @@ schedFetch(sched_root_task_t * __restrict root) } /* + * schedTrigger() - Triggering USER task + * + * @task = task + * return: -1 error or 0 ok + */ +int +schedTrigger(sched_task_t * __restrict task) +{ +#if SUP_ENABLE != KQ_SUPPORT + sched_SetErr(ENOTSUP, "disabled kqueue support"); + return -1; +#else +#ifndef EVFILT_USER + sched_SetErr(ENOTSUP, "Not supported kevent() filter"); + return -1; +#else + struct kevent chg[1]; + struct timespec timeout = { 0, 0 }; + + if (!task || !TASK_ROOT(task)) + return -1; + +#ifdef __NetBSD__ + EV_SET(chg, TASK_VAL(task), EVFILT_USER, 0, NOTE_TRIGGER, 0, (intptr_t) TASK_VAL(task)); +#else + EV_SET(chg, TASK_VAL(task), EVFILT_USER, 0, NOTE_TRIGGER, 0, (void*) TASK_VAL(task)); +#endif + if (kevent(TASK_ROOT(task)->root_kq, chg, 1, NULL, 0, &timeout) == -1) { + LOGERR; + return -1; + } + + return 0; +#endif +#endif /* KQ_SUPPORT */ +} + +/* + * schedQuery() - Query task in scheduler + * + * @task = task + * return: -1 error, 0 found and 1 not found + */ +int +schedQuery(sched_task_t * __restrict task) +{ + sched_queue_t *queue; + sched_task_t *t; + + if (!task || !TASK_ROOT(task)) + return -1; /* error */ + + switch (TASK_TYPE(task)) { + case taskREAD: + queue = &TASK_ROOT(task)->root_read; + break; + case taskWRITE: + queue = &TASK_ROOT(task)->root_write; + break; + case taskTIMER: + queue = &TASK_ROOT(task)->root_timer; + break; + case taskALARM: + queue = &TASK_ROOT(task)->root_alarm; + break; + case taskRTC: + queue = &TASK_ROOT(task)->root_rtc; + break; + case taskNODE: + queue = &TASK_ROOT(task)->root_node; + break; + case taskPROC: + queue = &TASK_ROOT(task)->root_proc; + break; + case taskSIGNAL: + queue = &TASK_ROOT(task)->root_signal; + break; + case taskAIO: + queue = &TASK_ROOT(task)->root_aio; + break; + case taskLIO: + queue = &TASK_ROOT(task)->root_lio; + break; + case taskUSER: + queue = &TASK_ROOT(task)->root_user; + break; + case taskEVENT: + queue = &TASK_ROOT(task)->root_event; + break; + case taskTASK: + queue = &TASK_ROOT(task)->root_task; + break; + case taskSUSPEND: + queue = &TASK_ROOT(task)->root_suspend; + break; + case taskREADY: + queue = &TASK_ROOT(task)->root_ready; + break; + case taskTHREAD: + queue = &TASK_ROOT(task)->root_thread; + break; + default: + return 1; /* not in queue */ + } + if (queue) + TAILQ_FOREACH(t, queue, task_node) + if (TASK_ID(t) == TASK_ID(task)) + return 0; /* found */ + + return 1; /* not in queue */ +} + +/* + * schedQueryby() - Query task in scheduler by criteria + * + * @root = root task + * @type = query from queue type, if =taskMAX query same task from all queues + * @criteria = find task by criteria + * [ CRITERIA_ANY|CRITERIA_CALL|CRITERIA_ARG|CRITERIA_FD|CRITERIA_VAL| + * CRITERIA_ID|CRITERIA_TS|CRITERIA_DATA|CRITERIA_DATLEN ] + * @param = search parameter + * return: NULL not found or !=NULL task + */ +sched_task_t * +schedQueryby(sched_root_task_t * __restrict root, sched_task_type_t type, + u_char criteria, void *param) +{ + sched_task_t *task; + sched_queue_t *queue; + + if (!root) + return NULL; + /* if type == taskMAX check in all queues */ + if (type == taskMAX) { + if ((task = schedQueryby(root, taskREAD, criteria, param))) + return task; + if ((task = schedQueryby(root, taskWRITE, criteria, param))) + return task; + if ((task = schedQueryby(root, taskTIMER, criteria, param))) + return task; + if ((task = schedQueryby(root, taskALARM, criteria, param))) + return task; + if ((task = schedQueryby(root, taskRTC, criteria, param))) + return task; + if ((task = schedQueryby(root, taskNODE, criteria, param))) + return task; + if ((task = schedQueryby(root, taskPROC, criteria, param))) + return task; + if ((task = schedQueryby(root, taskSIGNAL, criteria, param))) + return task; + if ((task = schedQueryby(root, taskAIO, criteria, param))) + return task; + if ((task = schedQueryby(root, taskLIO, criteria, param))) + return task; + if ((task = schedQueryby(root, taskUSER, criteria, param))) + return task; + if ((task = schedQueryby(root, taskEVENT, criteria, param))) + return task; + if ((task = schedQueryby(root, taskTASK, criteria, param))) + return task; + if ((task = schedQueryby(root, taskSUSPEND, criteria, param))) + return task; + if ((task = schedQueryby(root, taskREADY, criteria, param))) + return task; + if ((task = schedQueryby(root, taskTHREAD, criteria, param))) + return task; + return NULL; /* not found */ + } + /* choosen queue */ + switch (type) { + case taskREAD: + queue = &root->root_read; + break; + case taskWRITE: + queue = &root->root_write; + break; + case taskTIMER: + queue = &root->root_timer; + break; + case taskALARM: + queue = &root->root_alarm; + break; + case taskRTC: + queue = &root->root_rtc; + break; + case taskNODE: + queue = &root->root_node; + break; + case taskPROC: + queue = &root->root_proc; + break; + case taskSIGNAL: + queue = &root->root_signal; + break; + case taskAIO: + queue = &root->root_aio; + break; + case taskLIO: + queue = &root->root_lio; + break; + case taskUSER: + queue = &root->root_user; + break; + case taskEVENT: + queue = &root->root_event; + break; + case taskTASK: + queue = &root->root_task; + break; + case taskSUSPEND: + queue = &root->root_suspend; + break; + case taskREADY: + queue = &root->root_ready; + break; + case taskTHREAD: + queue = &root->root_thread; + break; + default: + return NULL; /* not found */ + } + + TAILQ_FOREACH(task, queue, task_node) { + switch (criteria) { + case CRITERIA_ANY: + return task; /* found */ + case CRITERIA_CALL: + if (TASK_FUNC(task) == (sched_task_func_t) param) + return task; /* found */ + break; + case CRITERIA_ARG: + if (TASK_ARG(task) == param) + return task; /* found */ + break; + case CRITERIA_FD: + if (TASK_FD(task) == (intptr_t) param) + return task; /* found */ + break; + case CRITERIA_ID: + case CRITERIA_VAL: + if (TASK_VAL(task) == (u_long) param) + return task; /* found */ + break; + case CRITERIA_TS: + if (!sched_timespeccmp(&TASK_TS(task), + (struct timespec*) param, -)) + return task; /* found */ + break; + case CRITERIA_DATA: + if (TASK_DATA(task) == param) + return task; /* found */ + break; + case CRITERIA_DATLEN: + if (TASK_DATLEN(task) == (size_t) param) + return task; /* found */ + break; + default: + sched_SetErr(EINVAL, "Invalid parameter criteria %d", + criteria); + return NULL; /* not found */ + } + } + + return NULL; /* not found */ +} + +/* * schedCancel() - Cancel task from scheduler + * * @task = task * return: -1 error or 0 ok */ @@ -239,132 +748,262 @@ schedCancel(sched_task_t * __restrict task) { sched_queue_t *queue; - if (!task || !task->task_root) + if (!task || !TASK_ROOT(task)) return -1; - if (task->task_root->root_hooks.hook_exec.cancel) - if (task->task_root->root_hooks.hook_exec.cancel(task, NULL)) + if (TASK_ROOT(task)->root_hooks.hook_exec.cancel) + if (TASK_ROOT(task)->root_hooks.hook_exec.cancel(task, NULL)) return -1; - switch (task->task_type) { + switch (TASK_TYPE(task)) { case taskREAD: - queue = &task->task_root->root_read; + queue = &TASK_ROOT(task)->root_read; break; case taskWRITE: - queue = &task->task_root->root_write; + queue = &TASK_ROOT(task)->root_write; break; case taskTIMER: - queue = &task->task_root->root_timer; + queue = &TASK_ROOT(task)->root_timer; break; + case taskALARM: + queue = &TASK_ROOT(task)->root_alarm; + break; + case taskRTC: + queue = &TASK_ROOT(task)->root_rtc; + break; + case taskNODE: + queue = &TASK_ROOT(task)->root_node; + break; + case taskPROC: + queue = &TASK_ROOT(task)->root_proc; + break; + case taskSIGNAL: + queue = &TASK_ROOT(task)->root_signal; + break; + case taskAIO: + queue = &TASK_ROOT(task)->root_aio; + break; + case taskLIO: + queue = &TASK_ROOT(task)->root_lio; + break; + case taskUSER: + queue = &TASK_ROOT(task)->root_user; + break; case taskEVENT: - queue = &task->task_root->root_event; + queue = &TASK_ROOT(task)->root_event; break; + case taskTASK: + queue = &TASK_ROOT(task)->root_task; + break; + case taskSUSPEND: + queue = &TASK_ROOT(task)->root_suspend; + break; case taskREADY: - queue = &task->task_root->root_ready; + queue = &TASK_ROOT(task)->root_ready; break; + case taskTHREAD: + queue = &TASK_ROOT(task)->root_thread; + break; default: queue = NULL; } if (queue) - TAILQ_REMOVE(queue, task, task_node); - if (task->task_type != taskUNUSE) { - task->task_type = taskUNUSE; - TAILQ_INSERT_TAIL(&task->task_root->root_unuse, task, task_node); - } + remove_task_from(task, queue); + if (TASK_TYPE(task) != taskUNUSE) + sched_unuseTask(task); return 0; } /* * schedCancelby() - Cancel task from scheduler by criteria + * * @root = root task - * @queue = cancel from queue, if =NULL cancel same task from all queues - * @criteria = find task by criteria [CRITERIA_CALL|CRITERIA_ARG|CRITERIA_FD|CRITERIA_VAL|CRITERIA_TV] + * @type = cancel from queue type, if =taskMAX cancel same task from all queues + * @criteria = find task by criteria + * [ CRITERIA_ANY|CRITERIA_CALL|CRITERIA_ARG|CRITERIA_FD|CRITERIA_VAL| + * CRITERIA_ID|CRITERIA_TS|CRITERIA_DATA|CRITERIA_DATLEN ] * @param = search parameter * @hook = custom cleanup hook function, may be NULL - * return: -1 error or 0 ok + * return: -1 error, -2 error in sub-stage cancel execution, -3 error from custom hook or 0 ok */ int -schedCancelby(sched_root_task_t * __restrict root, sched_queue_t * __restrict queue, +schedCancelby(sched_root_task_t * __restrict root, sched_task_type_t type, u_char criteria, void *param, sched_hook_func_t hook) { - sched_task_t *task; - int flg = 0; + sched_task_t *task, *tmp; + sched_queue_t *queue; + register int flg = 0; if (!root) return -1; - if (!queue) { - if (schedCancelby(root, &root->root_read, criteria, param, hook)) + /* if type == taskMAX check in all queues */ + if (type == taskMAX) { + if (schedCancelby(root, taskREAD, criteria, param, hook)) return -2; - if (schedCancelby(root, &root->root_write, criteria, param, hook)) + if (schedCancelby(root, taskWRITE, criteria, param, hook)) return -2; - if (schedCancelby(root, &root->root_timer, criteria, param, hook)) + if (schedCancelby(root, taskTIMER, criteria, param, hook)) return -2; - if (schedCancelby(root, &root->root_event, criteria, param, hook)) + if (schedCancelby(root, taskALARM, criteria, param, hook)) return -2; - if (schedCancelby(root, &root->root_ready, criteria, param, hook)) + if (schedCancelby(root, taskRTC, criteria, param, hook)) return -2; - if (schedCancelby(root, &root->root_read, criteria, param, hook)) + if (schedCancelby(root, taskNODE, criteria, param, hook)) return -2; + if (schedCancelby(root, taskPROC, criteria, param, hook)) + return -2; + if (schedCancelby(root, taskSIGNAL, criteria, param, hook)) + return -2; + if (schedCancelby(root, taskAIO, criteria, param, hook)) + return -2; + if (schedCancelby(root, taskLIO, criteria, param, hook)) + return -2; + if (schedCancelby(root, taskUSER, criteria, param, hook)) + return -2; + if (schedCancelby(root, taskEVENT, criteria, param, hook)) + return -2; + if (schedCancelby(root, taskTASK, criteria, param, hook)) + return -2; + if (schedCancelby(root, taskSUSPEND, criteria, param, hook)) + return -2; + if (schedCancelby(root, taskREADY, criteria, param, hook)) + return -2; + if (schedCancelby(root, taskTHREAD, criteria, param, hook)) + return -2; return 0; } + /* choosen queue */ + switch (type) { + case taskREAD: + queue = &root->root_read; + break; + case taskWRITE: + queue = &root->root_write; + break; + case taskTIMER: + queue = &root->root_timer; + break; + case taskALARM: + queue = &root->root_alarm; + break; + case taskRTC: + queue = &root->root_rtc; + break; + case taskNODE: + queue = &root->root_node; + break; + case taskPROC: + queue = &root->root_proc; + break; + case taskSIGNAL: + queue = &root->root_signal; + break; + case taskAIO: + queue = &root->root_aio; + break; + case taskLIO: + queue = &root->root_lio; + break; + case taskUSER: + queue = &root->root_user; + break; + case taskEVENT: + queue = &root->root_event; + break; + case taskTASK: + queue = &root->root_task; + break; + case taskSUSPEND: + queue = &root->root_suspend; + break; + case taskREADY: + queue = &root->root_ready; + break; + case taskTHREAD: + queue = &root->root_thread; + break; + default: + return 0; + } - TAILQ_FOREACH(task, queue, task_node) - if (criteria == CRITERIA_CALL) { - if (task->task_func == (sched_task_func_t) param) { - flg++; + SCHED_QLOCK(root, type); + TAILQ_FOREACH_SAFE(task, queue, task_node, tmp) { + flg ^= flg; + switch (criteria) { + case CRITERIA_ANY: + flg = 1; break; - } - } else if (criteria == CRITERIA_ARG) { - if (task->task_arg == param) { - flg++; + case CRITERIA_CALL: + if (TASK_FUNC(task) == (sched_task_func_t) param) + flg = 1; break; - } - } else if (criteria == CRITERIA_FD) { - if (TASK_FD(task) == (u_long) param) { - flg++; + case CRITERIA_ARG: + if (TASK_ARG(task) == param) + flg = 1; break; - } - } else if (criteria == CRITERIA_VAL) { - if (TASK_VAL(task) == (u_long) param) { - flg++; + case CRITERIA_FD: + if (TASK_FD(task) == (intptr_t) param) + flg = 1; break; - } - } else if (criteria == CRITERIA_TV) { - if (&TASK_TV(task) == (struct timeval*) param) { - flg++; + case CRITERIA_ID: + case CRITERIA_VAL: + if (TASK_VAL(task) == (u_long) param) + flg = 1; break; - } - } else { - sched_SetErr(EINVAL, "Invalid parameter criteria %d", criteria); - return -1; + case CRITERIA_TS: + if (!sched_timespeccmp(&TASK_TS(task), (struct timespec*) param, -)) + flg = 1; + break; + case CRITERIA_DATA: + if (TASK_DATA(task) == param) + flg = 1; + break; + case CRITERIA_DATLEN: + if (TASK_DATLEN(task) == (size_t) param) + flg = 1; + break; + default: + sched_SetErr(EINVAL, "Invalid parameter criteria %d", criteria); + flg = -1; } - if (!flg || !task) /* task not found */ - return 0; + if (flg < 0) /* error */ + break; + /* cancel choosen task */ + if (flg > 0) { + if (TASK_ROOT(task)->root_hooks.hook_exec.cancel) + if (TASK_ROOT(task)->root_hooks.hook_exec.cancel(task, NULL)) { + flg = -1; + break; + } + /* custom hook */ + if (hook) + if (hook(task, NULL)) { + flg = -3; + break; + } - if (task->task_root->root_hooks.hook_exec.cancel) - if (task->task_root->root_hooks.hook_exec.cancel(task, NULL)) - return -1; - if (hook) - if (hook(task, NULL)) - return -3; + TAILQ_REMOVE(queue, task, task_node); + if (TASK_TYPE(task) != taskUNUSE) + sched_unuseTask(task); - TAILQ_REMOVE(queue, task, task_node); - - if (task->task_type != taskUNUSE) { - task->task_type = taskUNUSE; - TAILQ_INSERT_TAIL(&task->task_root->root_unuse, task, task_node); + flg ^= flg; /* ok */ + } } - return 0; + SCHED_QUNLOCK(root, type); + + return flg; } /* * schedRun() - Scheduler *run loop* + * * @root = root task + * @killState = kill condition variable, if !=0 stop scheduler loop * return: -1 error or 0 ok */ int -schedRun(sched_root_task_t * __restrict root) +schedRun(sched_root_task_t *root, volatile intptr_t * __restrict killState) { sched_task_t *task; @@ -374,9 +1013,236 @@ schedRun(sched_root_task_t * __restrict root) if (root->root_hooks.hook_exec.run) if (root->root_hooks.hook_exec.run(root, NULL)) return -1; - if (root->root_hooks.hook_exec.fetch) - while ((task = root->root_hooks.hook_exec.fetch(root, NULL))) - schedCall(task); + + if (killState) { + if (root->root_hooks.hook_exec.condition) + /* condition scheduler loop */ + while (root && root->root_hooks.hook_exec.fetch && + root->root_hooks.hook_exec.condition && + root->root_hooks.hook_exec.condition(root, (void*) killState)) { + if ((task = root->root_hooks.hook_exec.fetch(root, NULL))) + root->root_ret = schedCall(task); + } + else + /* trigger scheduler loop */ + while (!*killState && root && root->root_hooks.hook_exec.fetch) { + if ((task = root->root_hooks.hook_exec.fetch(root, NULL))) + root->root_ret = schedCall(task); + } + } else + /* infinite scheduler loop */ + while (root && root->root_hooks.hook_exec.fetch) + if ((task = root->root_hooks.hook_exec.fetch(root, NULL))) + root->root_ret = schedCall(task); + + return 0; +} + +/* + * schedPolling() - Polling timeout period if no timer task is present + * + * @root = root task + * @ts = timeout polling period, if ==NULL INFINIT timeout + * @tsold = old timeout polling if !=NULL + * return: -1 error or 0 ok + */ +int +schedPolling(sched_root_task_t * __restrict root, struct timespec * __restrict ts, + struct timespec * __restrict tsold) +{ + if (!root) + return -1; + + if (tsold) + *tsold = root->root_poll; + + if (!ts) + sched_timespecinf(&root->root_poll); + else + root->root_poll = *ts; + + return 0; +} + +/* + * schedTermCondition() - Activate hook for scheduler condition kill + * + * @root = root task + * @condValue = condition value, kill schedRun() if condValue == killState + * return: -1 error or 0 ok + */ +int +schedTermCondition(sched_root_task_t * __restrict root, intptr_t * __restrict condValue) +{ + if (!root && !condValue) + return -1; + + *root->root_cond = *condValue; + root->root_hooks.hook_exec.condition = sched_hook_condition; + return 0; +} + +/* + * schedResumeby() - Resume suspended task + * + * @root = root task + * @criteria = find task by criteria + * [CRITERIA_ANY|CRITERIA_ID|CRITERIA_VAL|CRITERIA_DATA] + * @param = search parameter (sched_task_t *task| unsigned long id) + * return: -1 error or 0 resumed ok + */ +int +schedResumeby(sched_root_task_t * __restrict root, u_char criteria, void *param) +{ + sched_task_t *task, *tmp; + register int flg = 0; + + if (!root) + return -1; + + SCHED_QLOCK(root, taskSUSPEND); + TAILQ_FOREACH_SAFE(task, &root->root_suspend, task_node, tmp) { + flg ^= flg; + switch (criteria) { + case CRITERIA_ANY: + flg = 1; + break; + case CRITERIA_ID: + case CRITERIA_VAL: + if (TASK_VAL(task) == (u_long) param) + flg = 1; + break; + case CRITERIA_DATA: + if (TASK_ID(task) == (sched_task_t*) param) + flg = 1; + break; + default: + sched_SetErr(EINVAL, "Invalid parameter criteria %d", criteria); + flg = -1; + } + if (flg < 0) + break; + /* resume choosen task */ + if (flg > 0) { + if (root->root_hooks.hook_exec.resume) + if (root->root_hooks.hook_exec.resume(task, NULL)) { + flg = -1; + break; + } + + TAILQ_REMOVE(&root->root_suspend, task, task_node); + + task->task_type = taskREADY; + insert_task_to(task, &root->root_ready); + + flg ^= flg; /* ok */ + } + } + SCHED_QUNLOCK(root, taskSUSPEND); + + return flg; +} + +static void * +_sched_sigDisp(void *arg) +{ + sched_root_task_t *root = arg; + sched_task_t *task, *tmp; + int sig, flg; + sigset_t ss; + + sigfillset(&ss); + while (root->root_sigthr) { + if (sigwait(&ss, &sig)) + continue; + + pthread_mutex_lock(&root->root_sigmtx); + if (!sigismember(&root->root_sigset, sig)) { + pthread_mutex_unlock(&root->root_sigmtx); + continue; + } + + flg = 0; + TAILQ_FOREACH_SAFE(task, &root->root_signal, task_node, tmp) { + if (TASK_VAL(task) == (uintptr_t) sig) { + if (!flg) { + TASK_RET(task) = 0; + TASK_FLAG(task) = 0; + + /* remove signal handle */ + transit_task2ready(task, &root->root_signal); + } + flg++; + } + } + if (flg < 2) + sigdelset(&root->root_sigset, sig); + + pthread_mutex_unlock(&root->root_sigmtx); + } + + return NULL; +} + +/* + * schedSignalDispatch() - Activate or Deactivate signal dispatcher + * + * @root = root task + * @on = Activate or =0 deactivate + * return: -1 error or 0 ok + */ +int +schedSignalDispatch(sched_root_task_t * __restrict root, int on) +{ + sigset_t ss; + struct timespec ts = { .tv_sec = 1, .tv_nsec = 0 }; +#ifndef HAVE_LIBPTHREAD + sched_SetErr(ENOTSUP, "Library has not support pthreads"); + return -1; +#else + pthread_attr_t attr; +#endif + +#if SUP_ENABLE == KQ_SUPPORT + return 0; +#endif + + if (!on) { + pthread_cancel(root->root_sigthr); +#ifdef __linux__ + root->root_sigthr = 0L; +#else + root->root_sigthr = NULL; +#endif + return 0; + } + + /* if we have not using polling, then we must turn on */ + if (sched_timespecisinf(&root->root_poll)) + schedPolling(root, &ts, NULL); + + pthread_attr_init(&attr); + pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED); +#ifdef SCHED_RR + pthread_attr_setschedpolicy(&attr, SCHED_RR); +#else + pthread_attr_setschedpolicy(&attr, SCHED_OTHER); +#endif + + sigfillset(&ss); + pthread_sigmask(SIG_BLOCK, &ss, NULL); + + if (pthread_create(&root->root_sigthr, &attr, _sched_sigDisp, root)) { + sched_SetErr(errno, "pthread_create(SignalDispatch) #%d - %s", + errno, strerror(errno)); +#ifdef __linux__ + root->root_sigthr = 0L; +#else + root->root_sigthr = NULL; +#endif + pthread_sigmask(SIG_SETMASK, &root->root_oldset, NULL); + return -1; + } return 0; }