Annotation of libaitsched/src/tasks.c, revision 1.10.2.4
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.4! misho 6: * $Id: tasks.c,v 1.10.2.3 2012/08/01 14:11:43 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: else
626: memset(buffer, 0, buflen);
627:
628: if (!(acb = malloc(sizeof(struct aiocb)))) {
629: LOGERR;
630: return NULL;
631: } else
632: memset(acb, 0, sizeof(struct aiocb));
633:
634: acb->aio_fildes = fd;
635: acb->aio_nbytes = buflen;
636: acb->aio_buf = buffer;
637: off = lseek(fd, 0, SEEK_CUR);
638: if (off == -1) {
639: LOGERR;
640: free(acb);
641: return NULL;
642: } else
643: acb->aio_offset = off;
644: acb->aio_sigevent.sigev_notify = SIGEV_KEVENT;
645: acb->aio_sigevent.sigev_notify_kqueue = root->root_kq;
1.10.2.4! misho 646: acb->aio_sigevent.sigev_value.sival_ptr = acb;
1.10.2.2 misho 647:
648: if (aio_write(acb)) {
649: LOGERR;
650: free(acb);
651: return NULL;
652: }
653:
654: return schedAIO(root, func, arg, acb, buffer, buflen);
655: #endif
656: }
657:
658: /*
1.1 misho 659: * schedTimer() - Add TIMER task to scheduler queue
1.6 misho 660: *
1.1 misho 661: * @root = root task
662: * @func = task execution function
663: * @arg = 1st func argument
1.5 misho 664: * @ts = timeout argument structure
665: * @opt_data = Optional data
666: * @opt_dlen = Optional data length
1.1 misho 667: * return: NULL error or !=NULL new queued task
668: */
669: sched_task_t *
1.5 misho 670: schedTimer(sched_root_task_t * __restrict root, sched_task_func_t func, void *arg, struct timespec ts,
671: void *opt_data, size_t opt_dlen)
1.1 misho 672: {
1.9 misho 673: sched_task_t *task, *tmp, *t = NULL;
1.1 misho 674: void *ptr;
1.5 misho 675: struct timespec now;
1.1 misho 676:
677: if (!root || !func)
678: return NULL;
679:
680: /* get new task */
1.4 misho 681: if (!(task = _sched_useTask(root)))
682: return NULL;
1.1 misho 683:
684: task->task_func = func;
1.5 misho 685: TASK_TYPE(task) = taskTIMER;
686: TASK_ROOT(task) = root;
1.1 misho 687:
688: TASK_ARG(task) = arg;
689:
1.5 misho 690: TASK_DATA(task) = opt_data;
691: TASK_DATLEN(task) = opt_dlen;
692:
1.1 misho 693: /* calculate timeval structure */
1.5 misho 694: clock_gettime(CLOCK_MONOTONIC, &now);
695: now.tv_sec += ts.tv_sec;
696: now.tv_nsec += ts.tv_nsec;
697: if (now.tv_nsec >= 1000000000L) {
1.1 misho 698: now.tv_sec++;
1.5 misho 699: now.tv_nsec -= 1000000000L;
700: } else if (now.tv_nsec < 0) {
1.1 misho 701: now.tv_sec--;
1.5 misho 702: now.tv_nsec += 1000000000L;
1.1 misho 703: }
1.5 misho 704: TASK_TS(task) = now;
1.1 misho 705:
706: if (root->root_hooks.hook_add.timer)
707: ptr = root->root_hooks.hook_add.timer(task, NULL);
708: else
709: ptr = NULL;
710:
711: if (!ptr) {
1.5 misho 712: #ifdef HAVE_LIBPTHREAD
713: pthread_mutex_lock(&root->root_mtx[taskTIMER]);
714: #endif
1.1 misho 715: #ifdef TIMER_WITHOUT_SORT
1.9 misho 716: TAILQ_INSERT_TAIL(&root->root_timer, TASK_ID(task), task_node);
1.1 misho 717: #else
1.9 misho 718: TAILQ_FOREACH_SAFE(t, &root->root_timer, task_node, tmp)
1.5 misho 719: if (sched_timespeccmp(&TASK_TS(task), &TASK_TS(t), -) < 1)
1.1 misho 720: break;
721: if (!t)
1.9 misho 722: TAILQ_INSERT_TAIL(&root->root_timer, TASK_ID(task), task_node);
1.1 misho 723: else
1.9 misho 724: TAILQ_INSERT_BEFORE(t, TASK_ID(task), task_node);
1.1 misho 725: #endif
1.5 misho 726: #ifdef HAVE_LIBPTHREAD
727: pthread_mutex_unlock(&root->root_mtx[taskTIMER]);
728: #endif
1.4 misho 729: } else
730: task = _sched_unuseTask(task);
1.1 misho 731:
732: return task;
733: }
734:
735: /*
736: * schedEvent() - Add EVENT task to scheduler queue
1.6 misho 737: *
1.1 misho 738: * @root = root task
739: * @func = task execution function
740: * @arg = 1st func argument
1.2 misho 741: * @val = additional func argument
1.5 misho 742: * @opt_data = Optional data
743: * @opt_dlen = Optional data length
1.1 misho 744: * return: NULL error or !=NULL new queued task
745: */
746: sched_task_t *
1.5 misho 747: schedEvent(sched_root_task_t * __restrict root, sched_task_func_t func, void *arg, u_long val,
748: void *opt_data, size_t opt_dlen)
1.1 misho 749: {
750: sched_task_t *task;
751: void *ptr;
752:
753: if (!root || !func)
754: return NULL;
755:
756: /* get new task */
1.4 misho 757: if (!(task = _sched_useTask(root)))
758: return NULL;
1.1 misho 759:
760: task->task_func = func;
1.5 misho 761: TASK_TYPE(task) = taskEVENT;
762: TASK_ROOT(task) = root;
1.1 misho 763:
764: TASK_ARG(task) = arg;
1.2 misho 765: TASK_VAL(task) = val;
1.1 misho 766:
1.5 misho 767: TASK_DATA(task) = opt_data;
768: TASK_DATLEN(task) = opt_dlen;
769:
1.1 misho 770: if (root->root_hooks.hook_add.event)
771: ptr = root->root_hooks.hook_add.event(task, NULL);
772: else
773: ptr = NULL;
774:
1.5 misho 775: if (!ptr) {
776: #ifdef HAVE_LIBPTHREAD
777: pthread_mutex_lock(&root->root_mtx[taskEVENT]);
778: #endif
1.9 misho 779: TAILQ_INSERT_TAIL(&root->root_event, TASK_ID(task), task_node);
1.5 misho 780: #ifdef HAVE_LIBPTHREAD
781: pthread_mutex_unlock(&root->root_mtx[taskEVENT]);
782: #endif
783: } else
1.4 misho 784: task = _sched_unuseTask(task);
1.1 misho 785:
786: return task;
787: }
788:
789:
790: /*
791: * schedEventLo() - Add EVENT_Lo task to scheduler queue
1.6 misho 792: *
1.1 misho 793: * @root = root task
794: * @func = task execution function
795: * @arg = 1st func argument
1.2 misho 796: * @val = additional func argument
1.5 misho 797: * @opt_data = Optional data
798: * @opt_dlen = Optional data length
1.1 misho 799: * return: NULL error or !=NULL new queued task
800: */
801: sched_task_t *
1.5 misho 802: schedEventLo(sched_root_task_t * __restrict root, sched_task_func_t func, void *arg, u_long val,
803: void *opt_data, size_t opt_dlen)
1.1 misho 804: {
805: sched_task_t *task;
806: void *ptr;
807:
808: if (!root || !func)
809: return NULL;
810:
811: /* get new task */
1.4 misho 812: if (!(task = _sched_useTask(root)))
813: return NULL;
1.1 misho 814:
815: task->task_func = func;
1.5 misho 816: TASK_TYPE(task) = taskEVENT;
817: TASK_ROOT(task) = root;
1.1 misho 818:
819: TASK_ARG(task) = arg;
1.2 misho 820: TASK_VAL(task) = val;
1.1 misho 821:
1.5 misho 822: TASK_DATA(task) = opt_data;
823: TASK_DATLEN(task) = opt_dlen;
824:
1.1 misho 825: if (root->root_hooks.hook_add.eventlo)
826: ptr = root->root_hooks.hook_add.eventlo(task, NULL);
827: else
828: ptr = NULL;
829:
1.5 misho 830: if (!ptr) {
831: #ifdef HAVE_LIBPTHREAD
832: pthread_mutex_lock(&root->root_mtx[taskEVENTLO]);
833: #endif
1.9 misho 834: TAILQ_INSERT_TAIL(&root->root_eventlo, TASK_ID(task), task_node);
1.5 misho 835: #ifdef HAVE_LIBPTHREAD
836: pthread_mutex_unlock(&root->root_mtx[taskEVENTLO]);
837: #endif
838: } else
1.4 misho 839: task = _sched_unuseTask(task);
1.1 misho 840:
841: return task;
842: }
843:
844: /*
1.10 misho 845: * schedSuspend() - Add Suspended task to scheduler queue
846: *
847: * @root = root task
848: * @func = task execution function
849: * @arg = 1st func argument
850: * @id = Trigger ID
851: * @opt_data = Optional data
852: * @opt_dlen = Optional data length
853: * return: NULL error or !=NULL new queued task
854: */
855: sched_task_t *
856: schedSuspend(sched_root_task_t * __restrict root, sched_task_func_t func, void *arg, u_long id,
857: void *opt_data, size_t opt_dlen)
858: {
859: sched_task_t *task;
860: void *ptr;
861:
862: if (!root || !func)
863: return NULL;
864:
865: /* get new task */
866: if (!(task = _sched_useTask(root)))
867: return NULL;
868:
869: task->task_func = func;
870: TASK_TYPE(task) = taskSUSPEND;
871: TASK_ROOT(task) = root;
872:
873: TASK_ARG(task) = arg;
874: TASK_VAL(task) = id;
875:
876: TASK_DATA(task) = opt_data;
877: TASK_DATLEN(task) = opt_dlen;
878:
879: if (root->root_hooks.hook_add.suspend)
880: ptr = root->root_hooks.hook_add.suspend(task, NULL);
881: else
882: ptr = NULL;
883:
884: if (!ptr) {
885: #ifdef HAVE_LIBPTHREAD
886: pthread_mutex_lock(&root->root_mtx[taskSUSPEND]);
887: #endif
888: TAILQ_INSERT_TAIL(&root->root_suspend, TASK_ID(task), task_node);
889: #ifdef HAVE_LIBPTHREAD
890: pthread_mutex_unlock(&root->root_mtx[taskSUSPEND]);
891: #endif
892: } else
893: task = _sched_unuseTask(task);
894:
895: return task;
896: }
897:
898: /*
1.1 misho 899: * schedCallOnce() - Call once from scheduler
1.6 misho 900: *
1.1 misho 901: * @root = root task
902: * @func = task execution function
903: * @arg = 1st func argument
1.2 misho 904: * @val = additional func argument
1.5 misho 905: * @opt_data = Optional data
906: * @opt_dlen = Optional data length
1.2 misho 907: * return: return value from called func
1.1 misho 908: */
909: sched_task_t *
1.5 misho 910: schedCallOnce(sched_root_task_t * __restrict root, sched_task_func_t func, void *arg, u_long val,
911: void *opt_data, size_t opt_dlen)
1.1 misho 912: {
913: sched_task_t *task;
1.2 misho 914: void *ret;
1.1 misho 915:
916: if (!root || !func)
917: return NULL;
918:
919: /* get new task */
1.4 misho 920: if (!(task = _sched_useTask(root)))
921: return NULL;
1.1 misho 922:
923: task->task_func = func;
1.5 misho 924: TASK_TYPE(task) = taskEVENT;
925: TASK_ROOT(task) = root;
1.1 misho 926:
927: TASK_ARG(task) = arg;
1.2 misho 928: TASK_VAL(task) = val;
1.1 misho 929:
1.5 misho 930: TASK_DATA(task) = opt_data;
931: TASK_DATLEN(task) = opt_dlen;
932:
1.2 misho 933: ret = schedCall(task);
1.1 misho 934:
1.4 misho 935: _sched_unuseTask(task);
1.2 misho 936: return ret;
1.1 misho 937: }
FreeBSD-CVSweb <freebsd-cvsweb@FreeBSD.org>