--- libaitsched/src/aitsched.c 2013/11/14 21:37:27 1.22 +++ libaitsched/src/aitsched.c 2023/08/31 15:25:59 1.34.2.1 @@ -3,7 +3,7 @@ * by Michael Pounov * * $Author: misho $ -* $Id: aitsched.c,v 1.22 2013/11/14 21:37:27 misho Exp $ +* $Id: aitsched.c,v 1.34.2.1 2023/08/31 15:25:59 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, 2012, 2013 +Copyright 2004 - 2023 by Michael Pounov . All rights reserved. Redistribution and use in source and binary forms, with or without @@ -82,6 +82,79 @@ 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 */ /* @@ -101,19 +174,21 @@ 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; - root->root_hooks.hook_add.alarm = sched_hook_alarm; -#if defined(HAVE_TIMER_CREATE) && defined(HAVE_TIMER_SETTIME) +#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; - root->root_hooks.hook_add.signal = sched_hook_signal; #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; @@ -140,7 +215,7 @@ schedInit(void ** __restrict data, size_t datlen) register int i; #endif - root = malloc(sizeof(sched_root_task_t)); + root = e_malloc(sizeof(sched_root_task_t)); if (!root) { LOGERR; } else { @@ -153,12 +228,18 @@ schedInit(void ** __restrict data, size_t datlen) 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]); - free(root); + e_free(root); return NULL; } @@ -187,6 +268,10 @@ schedInit(void ** __restrict data, size_t datlen) #ifdef HAVE_LIBPTHREAD for (i = 0; i < taskMAX; i++) pthread_mutex_unlock(&root->root_mtx[i]); + + pthread_sigmask(SIG_BLOCK, NULL, &root->root_oldset); +#else + sigprocmask(SIG_BLOCK, NULL, &root->root_oldset); #endif if (data && *data) { @@ -224,7 +309,45 @@ schedEnd(sched_root_task_t ** __restrict root) if (!root || !*root) return -1; +#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_SAFE(task, &(*root)->root_write, task_node, tmp) schedCancel(task); @@ -257,26 +380,39 @@ schedEnd(sched_root_task_t ** __restrict root) TAILQ_FOREACH_SAFE(task, &(*root)->root_task, task_node, tmp) schedCancel(task); -#ifdef HAVE_LIBPTHREAD - pthread_mutex_lock(&(*root)->root_mtx[taskUNUSE]); -#endif + SCHED_QLOCK((*root), taskUNUSE); TAILQ_FOREACH_SAFE(task, &(*root)->root_unuse, task_node, tmp) { TAILQ_REMOVE(&(*root)->root_unuse, task, task_node); - free(task); + e_free(task); } -#ifdef HAVE_LIBPTHREAD - pthread_mutex_unlock(&(*root)->root_mtx[taskUNUSE]); -#endif + SCHED_QUNLOCK((*root), taskUNUSE); if ((*root)->root_hooks.hook_root.fini) (*root)->root_hooks.hook_root.fini(*root, NULL); #ifdef HAVE_LIBPTHREAD - for (i = 0; i < taskMAX; i++) + for (i = 0; i < taskMAX; i++) { + SCHED_QTRYLOCK(*root, i); + SCHED_QUNLOCK(*root, i); pthread_mutex_destroy(&(*root)->root_mtx[i]); + } + + 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; } @@ -335,6 +471,10 @@ schedFetch(sched_root_task_t * __restrict root) 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; @@ -357,6 +497,7 @@ schedTrigger(sched_task_t * __restrict task) return 0; #endif +#endif /* KQ_SUPPORT */ } /* @@ -441,55 +582,54 @@ schedQuery(sched_task_t * __restrict 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_ID|CRITERIA_TS|CRITERIA_DATA|CRITERIA_DATLEN ] * @param = search parameter - * return: -1 error, 0 found or 1 not found + * return: NULL not found or !=NULL task */ -int +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; - register int flg = 0; if (!root) - return -1; + return NULL; /* if type == taskMAX check in all queues */ if (type == taskMAX) { - if ((flg = schedQueryby(root, taskREAD, criteria, param)) < 1) - return flg; - if ((flg = schedQueryby(root, taskWRITE, criteria, param)) < 1) - return flg; - if ((flg = schedQueryby(root, taskTIMER, criteria, param)) < 1) - return flg; - if ((flg = schedQueryby(root, taskALARM, criteria, param)) < 1) - return flg; - if ((flg = schedQueryby(root, taskRTC, criteria, param)) < 1) - return flg; - if ((flg = schedQueryby(root, taskNODE, criteria, param)) < 1) - return flg; - if ((flg = schedQueryby(root, taskPROC, criteria, param)) < 1) - return flg; - if ((flg = schedQueryby(root, taskSIGNAL, criteria, param)) < 1) - return flg; - if ((flg = schedQueryby(root, taskAIO, criteria, param)) < 1) - return flg; - if ((flg = schedQueryby(root, taskLIO, criteria, param)) < 1) - return flg; - if ((flg = schedQueryby(root, taskUSER, criteria, param)) < 1) - return flg; - if ((flg = schedQueryby(root, taskEVENT, criteria, param)) < 1) - return flg; - if ((flg = schedQueryby(root, taskTASK, criteria, param)) < 1) - return flg; - if ((flg = schedQueryby(root, taskSUSPEND, criteria, param)) < 1) - return flg; - if ((flg = schedQueryby(root, taskREADY, criteria, param)) < 1) - return flg; - if ((flg = schedQueryby(root, taskTHREAD, criteria, param)) < 1) - return flg; - return 1; /* not found */ + 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) { @@ -542,47 +682,51 @@ schedQueryby(sched_root_task_t * __restrict root, sche queue = &root->root_thread; break; default: - return 1; /* not found */ + return NULL; /* not found */ } TAILQ_FOREACH(task, queue, task_node) { switch (criteria) { case CRITERIA_ANY: - return 0; /* found */ + return task; /* found */ case CRITERIA_CALL: if (TASK_FUNC(task) == (sched_task_func_t) param) - return 0; /* found */ + return task; /* found */ break; case CRITERIA_ARG: if (TASK_ARG(task) == param) - return 0; /* found */ + return task; /* found */ break; case CRITERIA_FD: if (TASK_FD(task) == (intptr_t) param) - return 0; /* found */ + return task; /* found */ break; case CRITERIA_ID: case CRITERIA_VAL: if (TASK_VAL(task) == (u_long) param) - return 0; /* found */ + return task; /* found */ break; case CRITERIA_TS: if (!sched_timespeccmp(&TASK_TS(task), (struct timespec*) param, -)) - return 0; /* found */ + return task; /* found */ break; case CRITERIA_DATA: if (TASK_DATA(task) == param) - return 0; /* found */ + 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 1; /* not found */ + return NULL; /* not found */ } } - return 1; /* not found */ + return NULL; /* not found */ } /* @@ -655,15 +799,8 @@ schedCancel(sched_task_t * __restrict task) default: queue = NULL; } - if (queue) { -#ifdef HAVE_LIBPTHREAD - pthread_mutex_lock(&TASK_ROOT(task)->root_mtx[TASK_TYPE(task)]); -#endif - TAILQ_REMOVE(queue, TASK_ID(task), task_node); -#ifdef HAVE_LIBPTHREAD - pthread_mutex_unlock(&TASK_ROOT(task)->root_mtx[TASK_TYPE(task)]); -#endif - } + if (queue) + remove_task_from(task, queue); if (TASK_TYPE(task) != taskUNUSE) sched_unuseTask(task); @@ -677,7 +814,7 @@ schedCancel(sched_task_t * __restrict task) * @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_ID|CRITERIA_TS|CRITERIA_DATA|CRITERIA_DATLEN ] * @param = search parameter * @hook = custom cleanup hook function, may be NULL * return: -1 error, -2 error in sub-stage cancel execution, -3 error from custom hook or 0 ok @@ -782,9 +919,7 @@ schedCancelby(sched_root_task_t * __restrict root, sch return 0; } -#ifdef HAVE_LIBPTHREAD - pthread_mutex_lock(&root->root_mtx[type]); -#endif + SCHED_QLOCK(root, type); TAILQ_FOREACH_SAFE(task, queue, task_node, tmp) { flg ^= flg; switch (criteria) { @@ -816,6 +951,10 @@ schedCancelby(sched_root_task_t * __restrict root, sch 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; @@ -843,9 +982,8 @@ schedCancelby(sched_root_task_t * __restrict root, sch flg ^= flg; /* ok */ } } -#ifdef HAVE_LIBPTHREAD - pthread_mutex_unlock(&root->root_mtx[type]); -#endif + SCHED_QUNLOCK(root, type); + return flg; } @@ -926,12 +1064,12 @@ schedPolling(sched_root_task_t * __restrict root, stru * return: -1 error or 0 ok */ int -schedTermCondition(sched_root_task_t * __restrict root, intptr_t condValue) +schedTermCondition(sched_root_task_t * __restrict root, intptr_t * __restrict condValue) { - if (!root) + if (!root && !condValue) return -1; - root->root_cond = condValue; + *root->root_cond = *condValue; root->root_hooks.hook_exec.condition = sched_hook_condition; return 0; } @@ -941,7 +1079,7 @@ schedTermCondition(sched_root_task_t * __restrict root * * @root = root task * @criteria = find task by criteria - * [CRITERIA_ANY|CRITERIA_ID|CRITERIA_DATA] + * [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 */ @@ -954,9 +1092,7 @@ schedResumeby(sched_root_task_t * __restrict root, u_c if (!root) return -1; -#ifdef HAVE_LIBPTHREAD - pthread_mutex_lock(&root->root_mtx[taskSUSPEND]); -#endif + SCHED_QLOCK(root, taskSUSPEND); TAILQ_FOREACH_SAFE(task, &root->root_suspend, task_node, tmp) { flg ^= flg; switch (criteria) { @@ -964,6 +1100,7 @@ schedResumeby(sched_root_task_t * __restrict root, u_c flg = 1; break; case CRITERIA_ID: + case CRITERIA_VAL: if (TASK_VAL(task) == (u_long) param) flg = 1; break; @@ -988,20 +1125,116 @@ schedResumeby(sched_root_task_t * __restrict root, u_c TAILQ_REMOVE(&root->root_suspend, task, task_node); task->task_type = taskREADY; -#ifdef HAVE_LIBPTHREAD - pthread_mutex_lock(&root->root_mtx[taskREADY]); -#endif - TAILQ_INSERT_TAIL(&root->root_ready, task, task_node); -#ifdef HAVE_LIBPTHREAD - pthread_mutex_unlock(&root->root_mtx[taskREADY]); -#endif + insert_task_to(task, &root->root_ready); flg ^= flg; /* ok */ } } -#ifdef HAVE_LIBPTHREAD - pthread_mutex_unlock(&root->root_mtx[taskSUSPEND]); -#endif + 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; }