Annotation of libaitsched/src/tasks.c, revision 1.10.2.5

1.1       misho       1: /*************************************************************************
                      2: * (C) 2011 AITNET ltd - Sofia/Bulgaria - <misho@aitbg.com>
                      3: *  by Michael Pounov <misho@openbsd-bg.org>
                      4: *
                      5: * $Author: misho $
1.10.2.5! misho       6: * $Id: tasks.c,v 1.10.2.4 2012/08/01 16:47:06 misho Exp $
1.1       misho       7: *
                      8: **************************************************************************
                      9: The ELWIX and AITNET software is distributed under the following
                     10: terms:
                     11: 
                     12: All of the documentation and software included in the ELWIX and AITNET
                     13: Releases is copyrighted by ELWIX - Sofia/Bulgaria <info@elwix.org>
                     14: 
1.6       misho      15: Copyright 2004, 2005, 2006, 2007, 2008, 2009, 2010, 2011, 2012
1.1       misho      16:        by Michael Pounov <misho@elwix.org>.  All rights reserved.
                     17: 
                     18: Redistribution and use in source and binary forms, with or without
                     19: modification, are permitted provided that the following conditions
                     20: are met:
                     21: 1. Redistributions of source code must retain the above copyright
                     22:    notice, this list of conditions and the following disclaimer.
                     23: 2. Redistributions in binary form must reproduce the above copyright
                     24:    notice, this list of conditions and the following disclaimer in the
                     25:    documentation and/or other materials provided with the distribution.
                     26: 3. All advertising materials mentioning features or use of this software
                     27:    must display the following acknowledgement:
                     28: This product includes software developed by Michael Pounov <misho@elwix.org>
                     29: ELWIX - Embedded LightWeight unIX and its contributors.
                     30: 4. Neither the name of AITNET nor the names of its contributors
                     31:    may be used to endorse or promote products derived from this software
                     32:    without specific prior written permission.
                     33: 
                     34: THIS SOFTWARE IS PROVIDED BY AITNET AND CONTRIBUTORS ``AS IS'' AND
                     35: ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
                     36: IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
                     37: ARE DISCLAIMED.  IN NO EVENT SHALL THE REGENTS OR CONTRIBUTORS BE LIABLE
                     38: FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
                     39: DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS
                     40: OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION)
                     41: HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
                     42: LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY
                     43: OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF
                     44: SUCH DAMAGE.
                     45: */
                     46: #include "global.h"
                     47: 
                     48: 
1.4       misho      49: #pragma GCC visibility push(hidden)
                     50: 
                     51: inline sched_task_t *
                     52: _sched_useTask(sched_root_task_t * __restrict root)
                     53: {
1.7       misho      54:        sched_task_t *task, *tmp;
1.4       misho      55: 
1.7       misho      56:        TAILQ_FOREACH_SAFE(task, &root->root_unuse, task_node, tmp) {
1.4       misho      57:                if (!TASK_ISLOCKED(task)) {
1.5       misho      58: #ifdef HAVE_LIBPTHREAD
                     59:                        pthread_mutex_lock(&root->root_mtx[taskUNUSE]);
                     60: #endif
1.4       misho      61:                        TAILQ_REMOVE(&root->root_unuse, task, task_node);
1.5       misho      62: #ifdef HAVE_LIBPTHREAD
                     63:                        pthread_mutex_unlock(&root->root_mtx[taskUNUSE]);
                     64: #endif
1.4       misho      65:                        break;
                     66:                }
                     67:        }
                     68: 
                     69:        if (!task) {
                     70:                task = malloc(sizeof(sched_task_t));
                     71:                if (!task) {
                     72:                        LOGERR;
                     73:                        return NULL;
                     74:                }
                     75:        }
                     76: 
1.9       misho      77:        memset(task, 0, sizeof(sched_task_t));
                     78:        task->task_id = (uintptr_t) task;
1.4       misho      79:        return task;
                     80: }
                     81: 
                     82: inline sched_task_t *
                     83: _sched_unuseTask(sched_task_t * __restrict task)
                     84: {
                     85:        TASK_UNLOCK(task);
1.5       misho      86:        TASK_TYPE(task) = taskUNUSE;
                     87: #ifdef HAVE_LIBPTHREAD
                     88:        pthread_mutex_lock(&TASK_ROOT(task)->root_mtx[taskUNUSE]);
                     89: #endif
1.9       misho      90:        TAILQ_INSERT_TAIL(&TASK_ROOT(task)->root_unuse, TASK_ID(task), task_node);
1.5       misho      91: #ifdef HAVE_LIBPTHREAD
                     92:        pthread_mutex_unlock(&TASK_ROOT(task)->root_mtx[taskUNUSE]);
                     93: #endif
1.4       misho      94:        task = NULL;
                     95: 
                     96:        return task;
                     97: }
                     98: 
                     99: #pragma GCC visibility pop
                    100: 
                    101: 
1.1       misho     102: /*
1.2       misho     103:  * schedRead() - Add READ I/O task to scheduler queue
1.6       misho     104:  *
1.1       misho     105:  * @root = root task
                    106:  * @func = task execution function
                    107:  * @arg = 1st func argument
1.2       misho     108:  * @fd = fd handle
1.5       misho     109:  * @opt_data = Optional data
                    110:  * @opt_dlen = Optional data length
1.1       misho     111:  * return: NULL error or !=NULL new queued task
                    112:  */
                    113: sched_task_t *
1.5       misho     114: schedRead(sched_root_task_t * __restrict root, sched_task_func_t func, void *arg, int fd, 
                    115:                void *opt_data, size_t opt_dlen)
1.1       misho     116: {
                    117:        sched_task_t *task;
                    118:        void *ptr;
                    119: 
                    120:        if (!root || !func)
                    121:                return NULL;
                    122: 
                    123:        /* get new task */
1.4       misho     124:        if (!(task = _sched_useTask(root)))
                    125:                return NULL;
1.1       misho     126: 
                    127:        task->task_func = func;
1.5       misho     128:        TASK_TYPE(task) = taskREAD;
                    129:        TASK_ROOT(task) = root;
1.1       misho     130: 
                    131:        TASK_ARG(task) = arg;
1.2       misho     132:        TASK_FD(task) = fd;
1.1       misho     133: 
1.5       misho     134:        TASK_DATA(task) = opt_data;
                    135:        TASK_DATLEN(task) = opt_dlen;
                    136: 
1.1       misho     137:        if (root->root_hooks.hook_add.read)
                    138:                ptr = root->root_hooks.hook_add.read(task, NULL);
                    139:        else
                    140:                ptr = NULL;
                    141: 
1.5       misho     142:        if (!ptr) {
                    143: #ifdef HAVE_LIBPTHREAD
                    144:                pthread_mutex_lock(&root->root_mtx[taskREAD]);
                    145: #endif
1.9       misho     146:                TAILQ_INSERT_TAIL(&root->root_read, TASK_ID(task), task_node);
1.5       misho     147: #ifdef HAVE_LIBPTHREAD
                    148:                pthread_mutex_unlock(&root->root_mtx[taskREAD]);
                    149: #endif
                    150:        } else
1.4       misho     151:                task = _sched_unuseTask(task);
1.1       misho     152: 
                    153:        return task;
                    154: }
                    155: 
                    156: /*
1.2       misho     157:  * schedWrite() - Add WRITE I/O task to scheduler queue
1.6       misho     158:  *
1.1       misho     159:  * @root = root task
                    160:  * @func = task execution function
                    161:  * @arg = 1st func argument
1.2       misho     162:  * @fd = fd handle
1.5       misho     163:  * @opt_data = Optional data
                    164:  * @opt_dlen = Optional data length
1.1       misho     165:  * return: NULL error or !=NULL new queued task
                    166:  */
                    167: sched_task_t *
1.5       misho     168: schedWrite(sched_root_task_t * __restrict root, sched_task_func_t func, void *arg, int fd, 
                    169:                void *opt_data, size_t opt_dlen)
1.1       misho     170: {
                    171:        sched_task_t *task;
                    172:        void *ptr;
                    173: 
                    174:        if (!root || !func)
                    175:                return NULL;
                    176: 
                    177:        /* get new task */
1.4       misho     178:        if (!(task = _sched_useTask(root)))
                    179:                return NULL;
1.1       misho     180: 
                    181:        task->task_func = func;
1.5       misho     182:        TASK_TYPE(task) = taskWRITE;
                    183:        TASK_ROOT(task) = root;
1.1       misho     184: 
                    185:        TASK_ARG(task) = arg;
1.2       misho     186:        TASK_FD(task) = fd;
1.1       misho     187: 
1.5       misho     188:        TASK_DATA(task) = opt_data;
                    189:        TASK_DATLEN(task) = opt_dlen;
                    190: 
1.1       misho     191:        if (root->root_hooks.hook_add.write)
                    192:                ptr = root->root_hooks.hook_add.write(task, NULL);
                    193:        else
                    194:                ptr = NULL;
                    195: 
1.5       misho     196:        if (!ptr) {
                    197: #ifdef HAVE_LIBPTHREAD
                    198:                pthread_mutex_lock(&root->root_mtx[taskWRITE]);
                    199: #endif
1.9       misho     200:                TAILQ_INSERT_TAIL(&root->root_write, TASK_ID(task), task_node);
1.5       misho     201: #ifdef HAVE_LIBPTHREAD
                    202:                pthread_mutex_unlock(&root->root_mtx[taskWRITE]);
                    203: #endif
                    204:        } else
1.4       misho     205:                task = _sched_unuseTask(task);
1.1       misho     206: 
                    207:        return task;
                    208: }
                    209: 
                    210: /*
1.9       misho     211:  * schedNode() - Add NODE task to scheduler queue
                    212:  *
                    213:  * @root = root task
                    214:  * @func = task execution function
                    215:  * @arg = 1st func argument
                    216:  * @fd = fd handle
                    217:  * @opt_data = Optional data
                    218:  * @opt_dlen = Optional data length
                    219:  * return: NULL error or !=NULL new queued task
                    220:  */
                    221: sched_task_t *
                    222: schedNode(sched_root_task_t * __restrict root, sched_task_func_t func, void *arg, int fd, 
                    223:                void *opt_data, size_t opt_dlen)
                    224: {
                    225:        sched_task_t *task;
                    226:        void *ptr;
                    227: 
                    228:        if (!root || !func)
                    229:                return NULL;
                    230: 
                    231:        /* get new task */
                    232:        if (!(task = _sched_useTask(root)))
                    233:                return NULL;
                    234: 
                    235:        task->task_func = func;
                    236:        TASK_TYPE(task) = taskNODE;
                    237:        TASK_ROOT(task) = root;
                    238: 
                    239:        TASK_ARG(task) = arg;
                    240:        TASK_FD(task) = fd;
                    241: 
                    242:        TASK_DATA(task) = opt_data;
                    243:        TASK_DATLEN(task) = opt_dlen;
                    244: 
                    245:        if (root->root_hooks.hook_add.node)
                    246:                ptr = root->root_hooks.hook_add.node(task, NULL);
                    247:        else
                    248:                ptr = NULL;
                    249: 
                    250:        if (!ptr) {
                    251: #ifdef HAVE_LIBPTHREAD
                    252:                pthread_mutex_lock(&root->root_mtx[taskNODE]);
                    253: #endif
                    254:                TAILQ_INSERT_TAIL(&root->root_node, TASK_ID(task), task_node);
                    255: #ifdef HAVE_LIBPTHREAD
                    256:                pthread_mutex_unlock(&root->root_mtx[taskNODE]);
                    257: #endif
                    258:        } else
                    259:                task = _sched_unuseTask(task);
                    260: 
                    261:        return task;
                    262: }
                    263: 
                    264: /*
                    265:  * schedProc() - Add PROC task to scheduler queue
                    266:  *
                    267:  * @root = root task
                    268:  * @func = task execution function
                    269:  * @arg = 1st func argument
                    270:  * @pid = PID
                    271:  * @opt_data = Optional data
                    272:  * @opt_dlen = Optional data length
                    273:  * return: NULL error or !=NULL new queued task
                    274:  */
                    275: sched_task_t *
                    276: schedProc(sched_root_task_t * __restrict root, sched_task_func_t func, void *arg, u_long pid, 
                    277:                void *opt_data, size_t opt_dlen)
                    278: {
                    279:        sched_task_t *task;
                    280:        void *ptr;
                    281: 
                    282:        if (!root || !func)
                    283:                return NULL;
                    284: 
                    285:        /* get new task */
                    286:        if (!(task = _sched_useTask(root)))
                    287:                return NULL;
                    288: 
                    289:        task->task_func = func;
                    290:        TASK_TYPE(task) = taskPROC;
                    291:        TASK_ROOT(task) = root;
                    292: 
                    293:        TASK_ARG(task) = arg;
                    294:        TASK_VAL(task) = pid;
                    295: 
                    296:        TASK_DATA(task) = opt_data;
                    297:        TASK_DATLEN(task) = opt_dlen;
                    298: 
                    299:        if (root->root_hooks.hook_add.proc)
                    300:                ptr = root->root_hooks.hook_add.proc(task, NULL);
                    301:        else
                    302:                ptr = NULL;
                    303: 
                    304:        if (!ptr) {
                    305: #ifdef HAVE_LIBPTHREAD
                    306:                pthread_mutex_lock(&root->root_mtx[taskPROC]);
                    307: #endif
                    308:                TAILQ_INSERT_TAIL(&root->root_proc, TASK_ID(task), task_node);
                    309: #ifdef HAVE_LIBPTHREAD
                    310:                pthread_mutex_unlock(&root->root_mtx[taskPROC]);
                    311: #endif
                    312:        } else
                    313:                task = _sched_unuseTask(task);
                    314: 
                    315:        return task;
                    316: }
                    317: 
                    318: /*
                    319:  * schedUser() - Add trigger USER task to scheduler queue
                    320:  *
                    321:  * @root = root task
                    322:  * @func = task execution function
                    323:  * @arg = 1st func argument
                    324:  * @id = Trigger ID
                    325:  * @opt_data = Optional data
                    326:  * @opt_dlen = Optional user's trigger flags
                    327:  * return: NULL error or !=NULL new queued task
                    328:  */
                    329: sched_task_t *
                    330: schedUser(sched_root_task_t * __restrict root, sched_task_func_t func, void *arg, u_long id, 
                    331:                void *opt_data, size_t opt_dlen)
                    332: {
                    333: #ifndef EVFILT_USER
                    334:        sched_SetErr(ENOTSUP, "Not supported kevent() filter");
                    335:        return NULL;
                    336: #else
                    337:        sched_task_t *task;
                    338:        void *ptr;
                    339: 
                    340:        if (!root || !func)
                    341:                return NULL;
                    342: 
                    343:        /* get new task */
                    344:        if (!(task = _sched_useTask(root)))
                    345:                return NULL;
                    346: 
                    347:        task->task_func = func;
                    348:        TASK_TYPE(task) = taskUSER;
                    349:        TASK_ROOT(task) = root;
                    350: 
                    351:        TASK_ARG(task) = arg;
                    352:        TASK_VAL(task) = id;
                    353: 
                    354:        TASK_DATA(task) = opt_data;
                    355:        TASK_DATLEN(task) = opt_dlen;
                    356: 
                    357:        if (root->root_hooks.hook_add.user)
                    358:                ptr = root->root_hooks.hook_add.user(task, NULL);
                    359:        else
                    360:                ptr = NULL;
                    361: 
                    362:        if (!ptr) {
                    363: #ifdef HAVE_LIBPTHREAD
                    364:                pthread_mutex_lock(&root->root_mtx[taskUSER]);
                    365: #endif
                    366:                TAILQ_INSERT_TAIL(&root->root_user, TASK_ID(task), task_node);
                    367: #ifdef HAVE_LIBPTHREAD
                    368:                pthread_mutex_unlock(&root->root_mtx[taskUSER]);
                    369: #endif
                    370:        } else
                    371:                task = _sched_unuseTask(task);
                    372: 
                    373:        return task;
                    374: #endif
                    375: }
                    376: 
                    377: /*
                    378:  * schedSignal() - Add SIGNAL task to scheduler queue
                    379:  *
                    380:  * @root = root task
                    381:  * @func = task execution function
                    382:  * @arg = 1st func argument
                    383:  * @sig = Signal
                    384:  * @opt_data = Optional data
                    385:  * @opt_dlen = Optional data length
                    386:  * return: NULL error or !=NULL new queued task
                    387:  */
                    388: sched_task_t *
                    389: schedSignal(sched_root_task_t * __restrict root, sched_task_func_t func, void *arg, u_long sig, 
                    390:                void *opt_data, size_t opt_dlen)
                    391: {
                    392:        sched_task_t *task;
                    393:        void *ptr;
                    394: 
                    395:        if (!root || !func)
                    396:                return NULL;
                    397: 
                    398:        /* get new task */
                    399:        if (!(task = _sched_useTask(root)))
                    400:                return NULL;
                    401: 
                    402:        task->task_func = func;
                    403:        TASK_TYPE(task) = taskSIGNAL;
                    404:        TASK_ROOT(task) = root;
                    405: 
                    406:        TASK_ARG(task) = arg;
                    407:        TASK_VAL(task) = sig;
                    408: 
                    409:        TASK_DATA(task) = opt_data;
                    410:        TASK_DATLEN(task) = opt_dlen;
                    411: 
                    412:        if (root->root_hooks.hook_add.signal)
                    413:                ptr = root->root_hooks.hook_add.signal(task, NULL);
                    414:        else
                    415:                ptr = NULL;
                    416: 
                    417:        if (!ptr) {
                    418: #ifdef HAVE_LIBPTHREAD
                    419:                pthread_mutex_lock(&root->root_mtx[taskSIGNAL]);
                    420: #endif
                    421:                TAILQ_INSERT_TAIL(&root->root_signal, TASK_ID(task), task_node);
                    422: #ifdef HAVE_LIBPTHREAD
                    423:                pthread_mutex_unlock(&root->root_mtx[taskSIGNAL]);
                    424: #endif
                    425:        } else
                    426:                task = _sched_unuseTask(task);
                    427: 
                    428:        return task;
                    429: }
                    430: 
                    431: /*
1.8       misho     432:  * schedAlarm() - Add ALARM task to scheduler queue
                    433:  *
                    434:  * @root = root task
                    435:  * @func = task execution function
                    436:  * @arg = 1st func argument
                    437:  * @ts = timeout argument structure, minimum alarm timer resolution is 1msec!
                    438:  * @opt_data = Optional data
                    439:  * @opt_dlen = Optional data length
                    440:  * return: NULL error or !=NULL new queued task
                    441:  */
                    442: sched_task_t *
                    443: schedAlarm(sched_root_task_t * __restrict root, sched_task_func_t func, void *arg, struct timespec ts, 
                    444:                void *opt_data, size_t opt_dlen)
                    445: {
                    446:        sched_task_t *task;
                    447:        void *ptr;
                    448: 
                    449:        if (!root || !func)
                    450:                return NULL;
                    451: 
                    452:        /* get new task */
                    453:        if (!(task = _sched_useTask(root)))
                    454:                return NULL;
                    455: 
                    456:        task->task_func = func;
                    457:        TASK_TYPE(task) = taskALARM;
                    458:        TASK_ROOT(task) = root;
                    459: 
                    460:        TASK_ARG(task) = arg;
                    461:        TASK_TS(task) = ts;
                    462: 
                    463:        TASK_DATA(task) = opt_data;
                    464:        TASK_DATLEN(task) = opt_dlen;
                    465: 
                    466:        if (root->root_hooks.hook_add.alarm)
                    467:                ptr = root->root_hooks.hook_add.alarm(task, NULL);
                    468:        else
                    469:                ptr = NULL;
                    470: 
                    471:        if (!ptr) {
                    472: #ifdef HAVE_LIBPTHREAD
                    473:                pthread_mutex_lock(&root->root_mtx[taskALARM]);
                    474: #endif
1.9       misho     475:                TAILQ_INSERT_TAIL(&root->root_alarm, TASK_ID(task), task_node);
1.8       misho     476: #ifdef HAVE_LIBPTHREAD
                    477:                pthread_mutex_unlock(&root->root_mtx[taskALARM]);
                    478: #endif
                    479:        } else
                    480:                task = _sched_unuseTask(task);
                    481: 
                    482:        return task;
                    483: }
                    484: 
                    485: /*
1.10.2.1  misho     486:  * schedAIO() - Add AIO task to scheduler queue
                    487:  *
                    488:  * @root = root task
                    489:  * @func = task execution function
                    490:  * @arg = 1st func argument
                    491:  * @acb = AIO cb structure address
                    492:  * @opt_data = Optional data
                    493:  * @opt_dlen = Optional data length
                    494:  * return: NULL error or !=NULL new queued task
                    495:  */
                    496: sched_task_t *
1.10.2.2  misho     497: schedAIO(sched_root_task_t * __restrict root, sched_task_func_t func, void *arg, 
                    498:                struct aiocb * __restrict acb, void *opt_data, size_t opt_dlen)
1.10.2.1  misho     499: {
                    500: #ifndef EVFILT_AIO
                    501:        sched_SetErr(ENOTSUP, "Not supported kevent() filter");
                    502:        return NULL;
                    503: #else
                    504:        sched_task_t *task;
                    505:        void *ptr;
                    506: 
                    507:        if (!root || !func)
                    508:                return NULL;
                    509: 
                    510:        /* get new task */
                    511:        if (!(task = _sched_useTask(root)))
                    512:                return NULL;
                    513: 
                    514:        task->task_func = func;
                    515:        TASK_TYPE(task) = taskAIO;
                    516:        TASK_ROOT(task) = root;
                    517: 
                    518:        TASK_ARG(task) = arg;
1.10.2.2  misho     519:        TASK_VAL(task) = (u_long) acb;
1.10.2.1  misho     520: 
                    521:        TASK_DATA(task) = opt_data;
                    522:        TASK_DATLEN(task) = opt_dlen;
                    523: 
                    524:        if (root->root_hooks.hook_add.aio)
                    525:                ptr = root->root_hooks.hook_add.aio(task, NULL);
                    526:        else
                    527:                ptr = NULL;
                    528: 
                    529:        if (!ptr) {
                    530: #ifdef HAVE_LIBPTHREAD
                    531:                pthread_mutex_lock(&root->root_mtx[taskAIO]);
                    532: #endif
                    533:                TAILQ_INSERT_TAIL(&root->root_aio, TASK_ID(task), task_node);
                    534: #ifdef HAVE_LIBPTHREAD
                    535:                pthread_mutex_unlock(&root->root_mtx[taskAIO]);
                    536: #endif
                    537:        } else
                    538:                task = _sched_unuseTask(task);
                    539: 
                    540:        return task;
                    541: #endif
                    542: }
                    543: 
                    544: /*
1.10.2.2  misho     545:  * schedAIORead() - Add AIO read task to scheduler queue
                    546:  *
                    547:  * @root = root task
                    548:  * @func = task execution function
                    549:  * @arg = 1st func argument
                    550:  * @fd = file descriptor
                    551:  * @buffer = Buffer
                    552:  * @buflen = Buffer length
                    553:  * return: NULL error or !=NULL new queued task
                    554:  */
                    555: inline sched_task_t *
                    556: schedAIORead(sched_root_task_t * __restrict root, sched_task_func_t func, void *arg, int fd, 
                    557:                void *buffer, size_t buflen)
                    558: {
                    559: #ifndef EVFILT_AIO
                    560:        sched_SetErr(ENOTSUP, "Not supported kevent() filter");
                    561:        return NULL;
                    562: #else
                    563:        struct aiocb *acb;
                    564:        off_t off = 0;
                    565: 
                    566:        if (!root || !func)
                    567:                return NULL;
                    568:        else
                    569:                memset(buffer, 0, buflen);
                    570: 
                    571:        if (!(acb = malloc(sizeof(struct aiocb)))) {
                    572:                LOGERR;
                    573:                return NULL;
                    574:        } else
                    575:                memset(acb, 0, sizeof(struct aiocb));
                    576: 
                    577:        acb->aio_fildes = fd;
                    578:        acb->aio_nbytes = buflen;
                    579:        acb->aio_buf = buffer;
                    580:        off = lseek(fd, 0, SEEK_CUR);
                    581:        if (off == -1) {
                    582:                LOGERR;
                    583:                free(acb);
                    584:                return NULL;
                    585:        } else
                    586:                acb->aio_offset = off;
                    587:        acb->aio_sigevent.sigev_notify = SIGEV_KEVENT;
                    588:        acb->aio_sigevent.sigev_notify_kqueue = root->root_kq;
1.10.2.4  misho     589:        acb->aio_sigevent.sigev_value.sival_ptr = acb;
1.10.2.2  misho     590: 
                    591:        if (aio_read(acb)) {
                    592:                LOGERR;
                    593:                free(acb);
                    594:                return NULL;
                    595:        }
                    596: 
                    597:        return schedAIO(root, func, arg, acb, buffer, buflen);
                    598: #endif
                    599: }
                    600: 
                    601: /*
                    602:  * schedAIOWrite() - Add AIO write task to scheduler queue
                    603:  *
                    604:  * @root = root task
                    605:  * @func = task execution function
                    606:  * @arg = 1st func argument
                    607:  * @fd = file descriptor
                    608:  * @buffer = Buffer
                    609:  * @buflen = Buffer length
                    610:  * return: NULL error or !=NULL new queued task
                    611:  */
                    612: inline sched_task_t *
                    613: schedAIOWrite(sched_root_task_t * __restrict root, sched_task_func_t func, void *arg, int fd, 
                    614:                void *buffer, size_t buflen)
                    615: {
                    616: #ifndef EVFILT_AIO
                    617:        sched_SetErr(ENOTSUP, "Not supported kevent() filter");
                    618:        return NULL;
                    619: #else
                    620:        struct aiocb *acb;
                    621:        off_t off = 0;
                    622: 
                    623:        if (!root || !func)
                    624:                return NULL;
                    625: 
                    626:        if (!(acb = malloc(sizeof(struct aiocb)))) {
                    627:                LOGERR;
                    628:                return NULL;
                    629:        } else
                    630:                memset(acb, 0, sizeof(struct aiocb));
                    631: 
                    632:        acb->aio_fildes = fd;
                    633:        acb->aio_nbytes = buflen;
                    634:        acb->aio_buf = buffer;
                    635:        off = lseek(fd, 0, SEEK_CUR);
                    636:        if (off == -1) {
                    637:                LOGERR;
                    638:                free(acb);
                    639:                return NULL;
                    640:        } else
                    641:                acb->aio_offset = off;
                    642:        acb->aio_sigevent.sigev_notify = SIGEV_KEVENT;
                    643:        acb->aio_sigevent.sigev_notify_kqueue = root->root_kq;
1.10.2.4  misho     644:        acb->aio_sigevent.sigev_value.sival_ptr = acb;
1.10.2.2  misho     645: 
                    646:        if (aio_write(acb)) {
                    647:                LOGERR;
                    648:                free(acb);
                    649:                return NULL;
                    650:        }
                    651: 
                    652:        return schedAIO(root, func, arg, acb, buffer, buflen);
                    653: #endif
                    654: }
                    655: 
                    656: /*
1.1       misho     657:  * schedTimer() - Add TIMER task to scheduler queue
1.6       misho     658:  *
1.1       misho     659:  * @root = root task
                    660:  * @func = task execution function
                    661:  * @arg = 1st func argument
1.5       misho     662:  * @ts = timeout argument structure
                    663:  * @opt_data = Optional data
                    664:  * @opt_dlen = Optional data length
1.1       misho     665:  * return: NULL error or !=NULL new queued task
                    666:  */
                    667: sched_task_t *
1.5       misho     668: schedTimer(sched_root_task_t * __restrict root, sched_task_func_t func, void *arg, struct timespec ts, 
                    669:                void *opt_data, size_t opt_dlen)
1.1       misho     670: {
1.9       misho     671:        sched_task_t *task, *tmp, *t = NULL;
1.1       misho     672:        void *ptr;
1.5       misho     673:        struct timespec now;
1.1       misho     674: 
                    675:        if (!root || !func)
                    676:                return NULL;
                    677: 
                    678:        /* get new task */
1.4       misho     679:        if (!(task = _sched_useTask(root)))
                    680:                return NULL;
1.1       misho     681: 
                    682:        task->task_func = func;
1.5       misho     683:        TASK_TYPE(task) = taskTIMER;
                    684:        TASK_ROOT(task) = root;
1.1       misho     685: 
                    686:        TASK_ARG(task) = arg;
                    687: 
1.5       misho     688:        TASK_DATA(task) = opt_data;
                    689:        TASK_DATLEN(task) = opt_dlen;
                    690: 
1.1       misho     691:        /* calculate timeval structure */
1.5       misho     692:        clock_gettime(CLOCK_MONOTONIC, &now);
                    693:        now.tv_sec += ts.tv_sec;
                    694:        now.tv_nsec += ts.tv_nsec;
                    695:        if (now.tv_nsec >= 1000000000L) {
1.1       misho     696:                now.tv_sec++;
1.5       misho     697:                now.tv_nsec -= 1000000000L;
                    698:        } else if (now.tv_nsec < 0) {
1.1       misho     699:                now.tv_sec--;
1.5       misho     700:                now.tv_nsec += 1000000000L;
1.1       misho     701:        }
1.5       misho     702:        TASK_TS(task) = now;
1.1       misho     703: 
                    704:        if (root->root_hooks.hook_add.timer)
                    705:                ptr = root->root_hooks.hook_add.timer(task, NULL);
                    706:        else
                    707:                ptr = NULL;
                    708: 
                    709:        if (!ptr) {
1.5       misho     710: #ifdef HAVE_LIBPTHREAD
                    711:                pthread_mutex_lock(&root->root_mtx[taskTIMER]);
                    712: #endif
1.1       misho     713: #ifdef TIMER_WITHOUT_SORT
1.9       misho     714:                TAILQ_INSERT_TAIL(&root->root_timer, TASK_ID(task), task_node);
1.1       misho     715: #else
1.9       misho     716:                TAILQ_FOREACH_SAFE(t, &root->root_timer, task_node, tmp)
1.5       misho     717:                        if (sched_timespeccmp(&TASK_TS(task), &TASK_TS(t), -) < 1)
1.1       misho     718:                                break;
                    719:                if (!t)
1.9       misho     720:                        TAILQ_INSERT_TAIL(&root->root_timer, TASK_ID(task), task_node);
1.1       misho     721:                else
1.9       misho     722:                        TAILQ_INSERT_BEFORE(t, TASK_ID(task), task_node);
1.1       misho     723: #endif
1.5       misho     724: #ifdef HAVE_LIBPTHREAD
                    725:                pthread_mutex_unlock(&root->root_mtx[taskTIMER]);
                    726: #endif
1.4       misho     727:        } else
                    728:                task = _sched_unuseTask(task);
1.1       misho     729: 
                    730:        return task;
                    731: }
                    732: 
                    733: /*
                    734:  * schedEvent() - Add EVENT task to scheduler queue
1.6       misho     735:  *
1.1       misho     736:  * @root = root task
                    737:  * @func = task execution function
                    738:  * @arg = 1st func argument
1.2       misho     739:  * @val = additional func argument
1.5       misho     740:  * @opt_data = Optional data
                    741:  * @opt_dlen = Optional data length
1.1       misho     742:  * return: NULL error or !=NULL new queued task
                    743:  */
                    744: sched_task_t *
1.5       misho     745: schedEvent(sched_root_task_t * __restrict root, sched_task_func_t func, void *arg, u_long val, 
                    746:                void *opt_data, size_t opt_dlen)
1.1       misho     747: {
                    748:        sched_task_t *task;
                    749:        void *ptr;
                    750: 
                    751:        if (!root || !func)
                    752:                return NULL;
                    753: 
                    754:        /* get new task */
1.4       misho     755:        if (!(task = _sched_useTask(root)))
                    756:                return NULL;
1.1       misho     757: 
                    758:        task->task_func = func;
1.5       misho     759:        TASK_TYPE(task) = taskEVENT;
                    760:        TASK_ROOT(task) = root;
1.1       misho     761: 
                    762:        TASK_ARG(task) = arg;
1.2       misho     763:        TASK_VAL(task) = val;
1.1       misho     764: 
1.5       misho     765:        TASK_DATA(task) = opt_data;
                    766:        TASK_DATLEN(task) = opt_dlen;
                    767: 
1.1       misho     768:        if (root->root_hooks.hook_add.event)
                    769:                ptr = root->root_hooks.hook_add.event(task, NULL);
                    770:        else
                    771:                ptr = NULL;
                    772: 
1.5       misho     773:        if (!ptr) {
                    774: #ifdef HAVE_LIBPTHREAD
                    775:                pthread_mutex_lock(&root->root_mtx[taskEVENT]);
                    776: #endif
1.9       misho     777:                TAILQ_INSERT_TAIL(&root->root_event, TASK_ID(task), task_node);
1.5       misho     778: #ifdef HAVE_LIBPTHREAD
                    779:                pthread_mutex_unlock(&root->root_mtx[taskEVENT]);
                    780: #endif
                    781:        } else
1.4       misho     782:                task = _sched_unuseTask(task);
1.1       misho     783: 
                    784:        return task;
                    785: }
                    786: 
                    787: 
                    788: /*
                    789:  * schedEventLo() - Add EVENT_Lo task to scheduler queue
1.6       misho     790:  *
1.1       misho     791:  * @root = root task
                    792:  * @func = task execution function
                    793:  * @arg = 1st func argument
1.2       misho     794:  * @val = additional func argument
1.5       misho     795:  * @opt_data = Optional data
                    796:  * @opt_dlen = Optional data length
1.1       misho     797:  * return: NULL error or !=NULL new queued task
                    798:  */
                    799: sched_task_t *
1.5       misho     800: schedEventLo(sched_root_task_t * __restrict root, sched_task_func_t func, void *arg, u_long val, 
                    801:                void *opt_data, size_t opt_dlen)
1.1       misho     802: {
                    803:        sched_task_t *task;
                    804:        void *ptr;
                    805: 
                    806:        if (!root || !func)
                    807:                return NULL;
                    808: 
                    809:        /* get new task */
1.4       misho     810:        if (!(task = _sched_useTask(root)))
                    811:                return NULL;
1.1       misho     812: 
                    813:        task->task_func = func;
1.5       misho     814:        TASK_TYPE(task) = taskEVENT;
                    815:        TASK_ROOT(task) = root;
1.1       misho     816: 
                    817:        TASK_ARG(task) = arg;
1.2       misho     818:        TASK_VAL(task) = val;
1.1       misho     819: 
1.5       misho     820:        TASK_DATA(task) = opt_data;
                    821:        TASK_DATLEN(task) = opt_dlen;
                    822: 
1.1       misho     823:        if (root->root_hooks.hook_add.eventlo)
                    824:                ptr = root->root_hooks.hook_add.eventlo(task, NULL);
                    825:        else
                    826:                ptr = NULL;
                    827: 
1.5       misho     828:        if (!ptr) {
                    829: #ifdef HAVE_LIBPTHREAD
                    830:                pthread_mutex_lock(&root->root_mtx[taskEVENTLO]);
                    831: #endif
1.9       misho     832:                TAILQ_INSERT_TAIL(&root->root_eventlo, TASK_ID(task), task_node);
1.5       misho     833: #ifdef HAVE_LIBPTHREAD
                    834:                pthread_mutex_unlock(&root->root_mtx[taskEVENTLO]);
                    835: #endif
                    836:        } else
1.4       misho     837:                task = _sched_unuseTask(task);
1.1       misho     838: 
                    839:        return task;
                    840: }
                    841: 
                    842: /*
1.10      misho     843:  * schedSuspend() - Add Suspended task to scheduler queue
                    844:  *
                    845:  * @root = root task
                    846:  * @func = task execution function
                    847:  * @arg = 1st func argument
                    848:  * @id = Trigger ID
                    849:  * @opt_data = Optional data
                    850:  * @opt_dlen = Optional data length
                    851:  * return: NULL error or !=NULL new queued task
                    852:  */
                    853: sched_task_t *
                    854: schedSuspend(sched_root_task_t * __restrict root, sched_task_func_t func, void *arg, u_long id, 
                    855:                void *opt_data, size_t opt_dlen)
                    856: {
                    857:        sched_task_t *task;
                    858:        void *ptr;
                    859: 
                    860:        if (!root || !func)
                    861:                return NULL;
                    862: 
                    863:        /* get new task */
                    864:        if (!(task = _sched_useTask(root)))
                    865:                return NULL;
                    866: 
                    867:        task->task_func = func;
                    868:        TASK_TYPE(task) = taskSUSPEND;
                    869:        TASK_ROOT(task) = root;
                    870: 
                    871:        TASK_ARG(task) = arg;
                    872:        TASK_VAL(task) = id;
                    873: 
                    874:        TASK_DATA(task) = opt_data;
                    875:        TASK_DATLEN(task) = opt_dlen;
                    876: 
                    877:        if (root->root_hooks.hook_add.suspend)
                    878:                ptr = root->root_hooks.hook_add.suspend(task, NULL);
                    879:        else
                    880:                ptr = NULL;
                    881: 
                    882:        if (!ptr) {
                    883: #ifdef HAVE_LIBPTHREAD
                    884:                pthread_mutex_lock(&root->root_mtx[taskSUSPEND]);
                    885: #endif
                    886:                TAILQ_INSERT_TAIL(&root->root_suspend, TASK_ID(task), task_node);
                    887: #ifdef HAVE_LIBPTHREAD
                    888:                pthread_mutex_unlock(&root->root_mtx[taskSUSPEND]);
                    889: #endif
                    890:        } else
                    891:                task = _sched_unuseTask(task);
                    892: 
                    893:        return task;
                    894: }
                    895: 
                    896: /*
1.1       misho     897:  * schedCallOnce() - Call once from scheduler
1.6       misho     898:  *
1.1       misho     899:  * @root = root task
                    900:  * @func = task execution function
                    901:  * @arg = 1st func argument
1.2       misho     902:  * @val = additional func argument
1.5       misho     903:  * @opt_data = Optional data
                    904:  * @opt_dlen = Optional data length
1.2       misho     905:  * return: return value from called func
1.1       misho     906:  */
                    907: sched_task_t *
1.5       misho     908: schedCallOnce(sched_root_task_t * __restrict root, sched_task_func_t func, void *arg, u_long val, 
                    909:                void *opt_data, size_t opt_dlen)
1.1       misho     910: {
                    911:        sched_task_t *task;
1.2       misho     912:        void *ret;
1.1       misho     913: 
                    914:        if (!root || !func)
                    915:                return NULL;
                    916: 
                    917:        /* get new task */
1.4       misho     918:        if (!(task = _sched_useTask(root)))
                    919:                return NULL;
1.1       misho     920: 
                    921:        task->task_func = func;
1.5       misho     922:        TASK_TYPE(task) = taskEVENT;
                    923:        TASK_ROOT(task) = root;
1.1       misho     924: 
                    925:        TASK_ARG(task) = arg;
1.2       misho     926:        TASK_VAL(task) = val;
1.1       misho     927: 
1.5       misho     928:        TASK_DATA(task) = opt_data;
                    929:        TASK_DATLEN(task) = opt_dlen;
                    930: 
1.2       misho     931:        ret = schedCall(task);
1.1       misho     932: 
1.4       misho     933:        _sched_unuseTask(task);
1.2       misho     934:        return ret;
1.1       misho     935: }

FreeBSD-CVSweb <freebsd-cvsweb@FreeBSD.org>