1: /*************************************************************************
2: * (C) 2011 AITNET ltd - Sofia/Bulgaria - <misho@aitbg.com>
3: * by Michael Pounov <misho@openbsd-bg.org>
4: *
5: * $Author: misho $
6: * $Id: tasks.c,v 1.21 2013/08/26 18:48:23 misho Exp $
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:
15: Copyright 2004, 2005, 2006, 2007, 2008, 2009, 2010, 2011, 2012, 2013
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:
49: /*
50: * sched_useTask() - Get and init new task
51: *
52: * @root = root task
53: * return: NULL error or !=NULL prepared task
54: */
55: sched_task_t *
56: sched_useTask(sched_root_task_t * __restrict root)
57: {
58: sched_task_t *task, *tmp;
59:
60: #ifdef HAVE_LIBPTHREAD
61: pthread_mutex_lock(&root->root_mtx[taskUNUSE]);
62: #endif
63: TAILQ_FOREACH_SAFE(task, &root->root_unuse, task_node, tmp) {
64: if (!TASK_ISLOCKED(task)) {
65: TAILQ_REMOVE(&root->root_unuse, task, task_node);
66: break;
67: }
68: }
69: #ifdef HAVE_LIBPTHREAD
70: pthread_mutex_unlock(&root->root_mtx[taskUNUSE]);
71: #endif
72:
73: if (!task) {
74: task = malloc(sizeof(sched_task_t));
75: if (!task) {
76: LOGERR;
77: return NULL;
78: }
79: }
80:
81: memset(task, 0, sizeof(sched_task_t));
82: task->task_id = (uintptr_t) task;
83: return task;
84: }
85:
86: /*
87: * sched_unuseTask() - Unlock and put task to unuse queue
88: *
89: * @task = task
90: * return: always is NULL
91: */
92: sched_task_t *
93: sched_unuseTask(sched_task_t * __restrict task)
94: {
95: TASK_UNLOCK(task);
96: TASK_TYPE(task) = taskUNUSE;
97: #ifdef HAVE_LIBPTHREAD
98: pthread_mutex_lock(&TASK_ROOT(task)->root_mtx[taskUNUSE]);
99: #endif
100: TAILQ_INSERT_TAIL(&TASK_ROOT(task)->root_unuse, TASK_ID(task), task_node);
101: #ifdef HAVE_LIBPTHREAD
102: pthread_mutex_unlock(&TASK_ROOT(task)->root_mtx[taskUNUSE]);
103: #endif
104: task = NULL;
105:
106: return task;
107: }
108:
109: #pragma GCC visibility push(hidden)
110:
111: #ifdef HAVE_LIBPTHREAD
112: static void
113: _sched_threadCleanup(sched_task_t *t)
114: {
115: if (!t || !TASK_ROOT(t))
116: return;
117:
118: pthread_mutex_lock(&TASK_ROOT(t)->root_mtx[taskTHREAD]);
119: TAILQ_REMOVE(&TASK_ROOT(t)->root_thread, t, task_node);
120: pthread_mutex_unlock(&TASK_ROOT(t)->root_mtx[taskTHREAD]);
121: sched_unuseTask(t);
122: }
123: void *
124: _sched_threadWrapper(sched_task_t *t)
125: {
126: void *ret = NULL;
127:
128: pthread_cleanup_push((void (*)(void*)) _sched_threadCleanup, t);
129:
130: if (!t || !TASK_ROOT(t))
131: pthread_exit(ret);
132:
133: pthread_setcancelstate(PTHREAD_CANCEL_ENABLE, NULL);
134: /*
135: pthread_setcanceltype(PTHREAD_CANCEL_ASYNCHRONOUS, NULL);
136: */
137:
138: /* notify parent, thread is ready for execution */
139: pthread_testcancel();
140:
141: ret = TASK_FUNC(t)(t);
142:
143: pthread_cleanup_pop(42);
144: TASK_ROOT(t)->root_ret = ret;
145: pthread_exit(ret);
146: }
147: #endif
148:
149: #if defined(HAVE_TIMER_CREATE) && defined(HAVE_TIMER_SETTIME)
150: void *
151: _sched_rtcWrapper(sched_task_t *t)
152: {
153: sched_task_func_t func;
154: sched_task_t *task;
155: sched_root_task_t *r;
156:
157: if (!t || !TASK_ROOT(t) || !TASK_DATA(t))
158: return NULL;
159: else {
160: r = TASK_ROOT(t);
161: task = (sched_task_t*) TASK_DATA(t);
162: func = TASK_FUNC(task);
163: }
164:
165: #ifdef HAVE_LIBPTHREAD
166: pthread_mutex_lock(&r->root_mtx[taskRTC]);
167: #endif
168: TAILQ_REMOVE(&r->root_rtc, task, task_node);
169: #ifdef HAVE_LIBPTHREAD
170: pthread_mutex_unlock(&r->root_mtx[taskRTC]);
171: #endif
172: sched_unuseTask(task);
173:
174: timer_delete((timer_t) TASK_DATLEN(t));
175:
176: return func(task);
177: }
178: #endif
179:
180: #pragma GCC visibility pop
181:
182: /*
183: * sched_taskExit() - Exit routine for scheduler task, explicit required for thread tasks
184: *
185: * @task = current task
186: * @retcode = return code
187: * return: return code
188: */
189: void *
190: sched_taskExit(sched_task_t *task, intptr_t retcode)
191: {
192: if (!task || !TASK_ROOT(task))
193: return (void*) -1;
194:
195: if (TASK_ROOT(task)->root_hooks.hook_exec.exit)
196: TASK_ROOT(task)->root_hooks.hook_exec.exit(task, (void*) retcode);
197:
198: TASK_ROOT(task)->root_ret = (void*) retcode;
199: return (void*) retcode;
200: }
201:
202:
203: /*
204: * schedRead() - Add READ I/O task to scheduler queue
205: *
206: * @root = root task
207: * @func = task execution function
208: * @arg = 1st func argument
209: * @fd = fd handle
210: * @opt_data = Optional data
211: * @opt_dlen = Optional data length
212: * return: NULL error or !=NULL new queued task
213: */
214: sched_task_t *
215: schedRead(sched_root_task_t * __restrict root, sched_task_func_t func, void *arg, int fd,
216: void *opt_data, size_t opt_dlen)
217: {
218: sched_task_t *task;
219: void *ptr;
220:
221: if (!root || !func)
222: return NULL;
223:
224: /* get new task */
225: if (!(task = sched_useTask(root)))
226: return NULL;
227:
228: task->task_func = func;
229: TASK_TYPE(task) = taskREAD;
230: TASK_ROOT(task) = root;
231:
232: TASK_ARG(task) = arg;
233: TASK_FD(task) = fd;
234:
235: TASK_DATA(task) = opt_data;
236: TASK_DATLEN(task) = opt_dlen;
237:
238: if (root->root_hooks.hook_add.read)
239: ptr = root->root_hooks.hook_add.read(task, NULL);
240: else
241: ptr = NULL;
242:
243: if (!ptr) {
244: #ifdef HAVE_LIBPTHREAD
245: pthread_mutex_lock(&root->root_mtx[taskREAD]);
246: #endif
247: TAILQ_INSERT_TAIL(&root->root_read, TASK_ID(task), task_node);
248: #ifdef HAVE_LIBPTHREAD
249: pthread_mutex_unlock(&root->root_mtx[taskREAD]);
250: #endif
251: } else
252: task = sched_unuseTask(task);
253:
254: return task;
255: }
256:
257: /*
258: * schedWrite() - Add WRITE I/O task to scheduler queue
259: *
260: * @root = root task
261: * @func = task execution function
262: * @arg = 1st func argument
263: * @fd = fd handle
264: * @opt_data = Optional data
265: * @opt_dlen = Optional data length
266: * return: NULL error or !=NULL new queued task
267: */
268: sched_task_t *
269: schedWrite(sched_root_task_t * __restrict root, sched_task_func_t func, void *arg, int fd,
270: void *opt_data, size_t opt_dlen)
271: {
272: sched_task_t *task;
273: void *ptr;
274:
275: if (!root || !func)
276: return NULL;
277:
278: /* get new task */
279: if (!(task = sched_useTask(root)))
280: return NULL;
281:
282: task->task_func = func;
283: TASK_TYPE(task) = taskWRITE;
284: TASK_ROOT(task) = root;
285:
286: TASK_ARG(task) = arg;
287: TASK_FD(task) = fd;
288:
289: TASK_DATA(task) = opt_data;
290: TASK_DATLEN(task) = opt_dlen;
291:
292: if (root->root_hooks.hook_add.write)
293: ptr = root->root_hooks.hook_add.write(task, NULL);
294: else
295: ptr = NULL;
296:
297: if (!ptr) {
298: #ifdef HAVE_LIBPTHREAD
299: pthread_mutex_lock(&root->root_mtx[taskWRITE]);
300: #endif
301: TAILQ_INSERT_TAIL(&root->root_write, TASK_ID(task), task_node);
302: #ifdef HAVE_LIBPTHREAD
303: pthread_mutex_unlock(&root->root_mtx[taskWRITE]);
304: #endif
305: } else
306: task = sched_unuseTask(task);
307:
308: return task;
309: }
310:
311: /*
312: * schedNode() - Add NODE task to scheduler queue
313: *
314: * @root = root task
315: * @func = task execution function
316: * @arg = 1st func argument
317: * @fd = fd handle
318: * @opt_data = Optional data
319: * @opt_dlen = Optional data length
320: * return: NULL error or !=NULL new queued task
321: */
322: sched_task_t *
323: schedNode(sched_root_task_t * __restrict root, sched_task_func_t func, void *arg, int fd,
324: void *opt_data, size_t opt_dlen)
325: {
326: sched_task_t *task;
327: void *ptr;
328:
329: if (!root || !func)
330: return NULL;
331:
332: /* get new task */
333: if (!(task = sched_useTask(root)))
334: return NULL;
335:
336: task->task_func = func;
337: TASK_TYPE(task) = taskNODE;
338: TASK_ROOT(task) = root;
339:
340: TASK_ARG(task) = arg;
341: TASK_FD(task) = fd;
342:
343: TASK_DATA(task) = opt_data;
344: TASK_DATLEN(task) = opt_dlen;
345:
346: if (root->root_hooks.hook_add.node)
347: ptr = root->root_hooks.hook_add.node(task, NULL);
348: else
349: ptr = NULL;
350:
351: if (!ptr) {
352: #ifdef HAVE_LIBPTHREAD
353: pthread_mutex_lock(&root->root_mtx[taskNODE]);
354: #endif
355: TAILQ_INSERT_TAIL(&root->root_node, TASK_ID(task), task_node);
356: #ifdef HAVE_LIBPTHREAD
357: pthread_mutex_unlock(&root->root_mtx[taskNODE]);
358: #endif
359: } else
360: task = sched_unuseTask(task);
361:
362: return task;
363: }
364:
365: /*
366: * schedProc() - Add PROC task to scheduler queue
367: *
368: * @root = root task
369: * @func = task execution function
370: * @arg = 1st func argument
371: * @pid = PID
372: * @opt_data = Optional data
373: * @opt_dlen = Optional data length
374: * return: NULL error or !=NULL new queued task
375: */
376: sched_task_t *
377: schedProc(sched_root_task_t * __restrict root, sched_task_func_t func, void *arg, u_long pid,
378: void *opt_data, size_t opt_dlen)
379: {
380: sched_task_t *task;
381: void *ptr;
382:
383: if (!root || !func)
384: return NULL;
385:
386: /* get new task */
387: if (!(task = sched_useTask(root)))
388: return NULL;
389:
390: task->task_func = func;
391: TASK_TYPE(task) = taskPROC;
392: TASK_ROOT(task) = root;
393:
394: TASK_ARG(task) = arg;
395: TASK_VAL(task) = pid;
396:
397: TASK_DATA(task) = opt_data;
398: TASK_DATLEN(task) = opt_dlen;
399:
400: if (root->root_hooks.hook_add.proc)
401: ptr = root->root_hooks.hook_add.proc(task, NULL);
402: else
403: ptr = NULL;
404:
405: if (!ptr) {
406: #ifdef HAVE_LIBPTHREAD
407: pthread_mutex_lock(&root->root_mtx[taskPROC]);
408: #endif
409: TAILQ_INSERT_TAIL(&root->root_proc, TASK_ID(task), task_node);
410: #ifdef HAVE_LIBPTHREAD
411: pthread_mutex_unlock(&root->root_mtx[taskPROC]);
412: #endif
413: } else
414: task = sched_unuseTask(task);
415:
416: return task;
417: }
418:
419: /*
420: * schedUser() - Add trigger USER task to scheduler queue
421: *
422: * @root = root task
423: * @func = task execution function
424: * @arg = 1st func argument
425: * @id = Trigger ID
426: * @opt_data = Optional data
427: * @opt_dlen = Optional user's trigger flags
428: * return: NULL error or !=NULL new queued task
429: */
430: sched_task_t *
431: schedUser(sched_root_task_t * __restrict root, sched_task_func_t func, void *arg, u_long id,
432: void *opt_data, size_t opt_dlen)
433: {
434: #ifndef EVFILT_USER
435: sched_SetErr(ENOTSUP, "Not supported kevent() filter");
436: return NULL;
437: #else
438: sched_task_t *task;
439: void *ptr;
440:
441: if (!root || !func)
442: return NULL;
443:
444: /* get new task */
445: if (!(task = sched_useTask(root)))
446: return NULL;
447:
448: task->task_func = func;
449: TASK_TYPE(task) = taskUSER;
450: TASK_ROOT(task) = root;
451:
452: TASK_ARG(task) = arg;
453: TASK_VAL(task) = id;
454:
455: TASK_DATA(task) = opt_data;
456: TASK_DATLEN(task) = opt_dlen;
457:
458: if (root->root_hooks.hook_add.user)
459: ptr = root->root_hooks.hook_add.user(task, NULL);
460: else
461: ptr = NULL;
462:
463: if (!ptr) {
464: #ifdef HAVE_LIBPTHREAD
465: pthread_mutex_lock(&root->root_mtx[taskUSER]);
466: #endif
467: TAILQ_INSERT_TAIL(&root->root_user, TASK_ID(task), task_node);
468: #ifdef HAVE_LIBPTHREAD
469: pthread_mutex_unlock(&root->root_mtx[taskUSER]);
470: #endif
471: } else
472: task = sched_unuseTask(task);
473:
474: return task;
475: #endif
476: }
477:
478: /*
479: * schedSignal() - Add SIGNAL task to scheduler queue
480: *
481: * @root = root task
482: * @func = task execution function
483: * @arg = 1st func argument
484: * @sig = Signal
485: * @opt_data = Optional data
486: * @opt_dlen = Optional data length
487: * return: NULL error or !=NULL new queued task
488: */
489: sched_task_t *
490: schedSignal(sched_root_task_t * __restrict root, sched_task_func_t func, void *arg, u_long sig,
491: void *opt_data, size_t opt_dlen)
492: {
493: sched_task_t *task;
494: void *ptr;
495:
496: if (!root || !func)
497: return NULL;
498:
499: /* get new task */
500: if (!(task = sched_useTask(root)))
501: return NULL;
502:
503: task->task_func = func;
504: TASK_TYPE(task) = taskSIGNAL;
505: TASK_ROOT(task) = root;
506:
507: TASK_ARG(task) = arg;
508: TASK_VAL(task) = sig;
509:
510: TASK_DATA(task) = opt_data;
511: TASK_DATLEN(task) = opt_dlen;
512:
513: if (root->root_hooks.hook_add.signal)
514: ptr = root->root_hooks.hook_add.signal(task, NULL);
515: else
516: ptr = NULL;
517:
518: if (!ptr) {
519: #ifdef HAVE_LIBPTHREAD
520: pthread_mutex_lock(&root->root_mtx[taskSIGNAL]);
521: #endif
522: TAILQ_INSERT_TAIL(&root->root_signal, TASK_ID(task), task_node);
523: #ifdef HAVE_LIBPTHREAD
524: pthread_mutex_unlock(&root->root_mtx[taskSIGNAL]);
525: #endif
526: } else
527: task = sched_unuseTask(task);
528:
529: return task;
530: }
531:
532: /*
533: * schedAlarm() - Add ALARM task to scheduler queue
534: *
535: * @root = root task
536: * @func = task execution function
537: * @arg = 1st func argument
538: * @ts = timeout argument structure, minimum alarm timer resolution is 1msec!
539: * @opt_data = Alarm timer ID
540: * @opt_dlen = Optional data length
541: * return: NULL error or !=NULL new queued task
542: */
543: sched_task_t *
544: schedAlarm(sched_root_task_t * __restrict root, sched_task_func_t func, void *arg, struct timespec ts,
545: void *opt_data, size_t opt_dlen)
546: {
547: sched_task_t *task;
548: void *ptr;
549:
550: if (!root || !func)
551: return NULL;
552:
553: /* get new task */
554: if (!(task = sched_useTask(root)))
555: return NULL;
556:
557: task->task_func = func;
558: TASK_TYPE(task) = taskALARM;
559: TASK_ROOT(task) = root;
560:
561: TASK_ARG(task) = arg;
562: TASK_TS(task) = ts;
563:
564: TASK_DATA(task) = opt_data;
565: TASK_DATLEN(task) = opt_dlen;
566:
567: if (root->root_hooks.hook_add.alarm)
568: ptr = root->root_hooks.hook_add.alarm(task, NULL);
569: else
570: ptr = NULL;
571:
572: if (!ptr) {
573: #ifdef HAVE_LIBPTHREAD
574: pthread_mutex_lock(&root->root_mtx[taskALARM]);
575: #endif
576: TAILQ_INSERT_TAIL(&root->root_alarm, TASK_ID(task), task_node);
577: #ifdef HAVE_LIBPTHREAD
578: pthread_mutex_unlock(&root->root_mtx[taskALARM]);
579: #endif
580: } else
581: task = sched_unuseTask(task);
582:
583: return task;
584: }
585:
586: #ifdef AIO_SUPPORT
587: /*
588: * schedAIO() - Add AIO task to scheduler queue
589: *
590: * @root = root task
591: * @func = task execution function
592: * @arg = 1st func argument
593: * @acb = AIO cb structure address
594: * @opt_data = Optional data
595: * @opt_dlen = Optional data length
596: * return: NULL error or !=NULL new queued task
597: */
598: sched_task_t *
599: schedAIO(sched_root_task_t * __restrict root, sched_task_func_t func, void *arg,
600: struct aiocb * __restrict acb, void *opt_data, size_t opt_dlen)
601: {
602: sched_task_t *task;
603: void *ptr;
604:
605: if (!root || !func || !acb || !opt_dlen)
606: return NULL;
607:
608: /* get new task */
609: if (!(task = sched_useTask(root)))
610: return NULL;
611:
612: task->task_func = func;
613: TASK_TYPE(task) = taskAIO;
614: TASK_ROOT(task) = root;
615:
616: TASK_ARG(task) = arg;
617: TASK_VAL(task) = (u_long) acb;
618:
619: TASK_DATA(task) = opt_data;
620: TASK_DATLEN(task) = opt_dlen;
621:
622: if (root->root_hooks.hook_add.aio)
623: ptr = root->root_hooks.hook_add.aio(task, NULL);
624: else
625: ptr = NULL;
626:
627: if (!ptr) {
628: #ifdef HAVE_LIBPTHREAD
629: pthread_mutex_lock(&root->root_mtx[taskAIO]);
630: #endif
631: TAILQ_INSERT_TAIL(&root->root_aio, TASK_ID(task), task_node);
632: #ifdef HAVE_LIBPTHREAD
633: pthread_mutex_unlock(&root->root_mtx[taskAIO]);
634: #endif
635: } else
636: task = sched_unuseTask(task);
637:
638: return task;
639: }
640:
641: /*
642: * schedAIORead() - Add AIO read task to scheduler queue
643: *
644: * @root = root task
645: * @func = task execution function
646: * @arg = 1st func argument
647: * @fd = file descriptor
648: * @buffer = Buffer
649: * @buflen = Buffer length
650: * @offset = Offset from start of file, if =-1 from current position
651: * return: NULL error or !=NULL new queued task
652: */
653: sched_task_t *
654: schedAIORead(sched_root_task_t * __restrict root, sched_task_func_t func, void *arg, int fd,
655: void *buffer, size_t buflen, off_t offset)
656: {
657: struct aiocb *acb;
658: off_t off;
659:
660: if (!root || !func || !buffer || !buflen)
661: return NULL;
662:
663: if (offset == (off_t) -1) {
664: off = lseek(fd, 0, SEEK_CUR);
665: if (off == -1) {
666: LOGERR;
667: return NULL;
668: }
669: } else
670: off = offset;
671:
672: if (!(acb = malloc(sizeof(struct aiocb)))) {
673: LOGERR;
674: return NULL;
675: } else
676: memset(acb, 0, sizeof(struct aiocb));
677:
678: acb->aio_fildes = fd;
679: acb->aio_nbytes = buflen;
680: acb->aio_buf = buffer;
681: acb->aio_offset = off;
682: acb->aio_sigevent.sigev_notify = SIGEV_KEVENT;
683: acb->aio_sigevent.sigev_notify_kqueue = root->root_kq;
684: acb->aio_sigevent.sigev_value.sival_ptr = acb;
685:
686: if (aio_read(acb)) {
687: LOGERR;
688: free(acb);
689: return NULL;
690: }
691:
692: return schedAIO(root, func, arg, acb, buffer, buflen);
693: }
694:
695: /*
696: * schedAIOWrite() - Add AIO write task to scheduler queue
697: *
698: * @root = root task
699: * @func = task execution function
700: * @arg = 1st func argument
701: * @fd = file descriptor
702: * @buffer = Buffer
703: * @buflen = Buffer length
704: * @offset = Offset from start of file, if =-1 from current position
705: * return: NULL error or !=NULL new queued task
706: */
707: sched_task_t *
708: schedAIOWrite(sched_root_task_t * __restrict root, sched_task_func_t func, void *arg, int fd,
709: void *buffer, size_t buflen, off_t offset)
710: {
711: struct aiocb *acb;
712: off_t off;
713:
714: if (!root || !func || !buffer || !buflen)
715: return NULL;
716:
717: if (offset == (off_t) -1) {
718: off = lseek(fd, 0, SEEK_CUR);
719: if (off == -1) {
720: LOGERR;
721: return NULL;
722: }
723: } else
724: off = offset;
725:
726: if (!(acb = malloc(sizeof(struct aiocb)))) {
727: LOGERR;
728: return NULL;
729: } else
730: memset(acb, 0, sizeof(struct aiocb));
731:
732: acb->aio_fildes = fd;
733: acb->aio_nbytes = buflen;
734: acb->aio_buf = buffer;
735: acb->aio_offset = off;
736: acb->aio_sigevent.sigev_notify = SIGEV_KEVENT;
737: acb->aio_sigevent.sigev_notify_kqueue = root->root_kq;
738: acb->aio_sigevent.sigev_value.sival_ptr = acb;
739:
740: if (aio_write(acb)) {
741: LOGERR;
742: free(acb);
743: return NULL;
744: }
745:
746: return schedAIO(root, func, arg, acb, buffer, buflen);
747: }
748:
749: #ifdef EVFILT_LIO
750: /*
751: * schedLIO() - Add AIO bulk tasks to scheduler queue
752: *
753: * @root = root task
754: * @func = task execution function
755: * @arg = 1st func argument
756: * @acbs = AIO cb structure addresses
757: * @opt_data = Optional data
758: * @opt_dlen = Optional data length
759: * return: NULL error or !=NULL new queued task
760: */
761: sched_task_t *
762: schedLIO(sched_root_task_t * __restrict root, sched_task_func_t func, void *arg,
763: struct aiocb ** __restrict acbs, void *opt_data, size_t opt_dlen)
764: {
765: sched_task_t *task;
766: void *ptr;
767:
768: if (!root || !func || !acbs || !opt_dlen)
769: return NULL;
770:
771: /* get new task */
772: if (!(task = sched_useTask(root)))
773: return NULL;
774:
775: task->task_func = func;
776: TASK_TYPE(task) = taskLIO;
777: TASK_ROOT(task) = root;
778:
779: TASK_ARG(task) = arg;
780: TASK_VAL(task) = (u_long) acbs;
781:
782: TASK_DATA(task) = opt_data;
783: TASK_DATLEN(task) = opt_dlen;
784:
785: if (root->root_hooks.hook_add.lio)
786: ptr = root->root_hooks.hook_add.lio(task, NULL);
787: else
788: ptr = NULL;
789:
790: if (!ptr) {
791: #ifdef HAVE_LIBPTHREAD
792: pthread_mutex_lock(&root->root_mtx[taskLIO]);
793: #endif
794: TAILQ_INSERT_TAIL(&root->root_lio, TASK_ID(task), task_node);
795: #ifdef HAVE_LIBPTHREAD
796: pthread_mutex_unlock(&root->root_mtx[taskLIO]);
797: #endif
798: } else
799: task = sched_unuseTask(task);
800:
801: return task;
802: }
803:
804: /*
805: * schedLIORead() - Add list of AIO read tasks to scheduler queue
806: *
807: * @root = root task
808: * @func = task execution function
809: * @arg = 1st func argument
810: * @fd = file descriptor
811: * @bufs = Buffer's list
812: * @nbufs = Number of Buffers
813: * @offset = Offset from start of file, if =-1 from current position
814: * return: NULL error or !=NULL new queued task
815: */
816: sched_task_t *
817: schedLIORead(sched_root_task_t * __restrict root, sched_task_func_t func, void *arg, int fd,
818: struct iovec *bufs, size_t nbufs, off_t offset)
819: {
820: struct sigevent sig;
821: struct aiocb **acb;
822: off_t off;
823: register int i;
824:
825: if (!root || !func || !bufs || !nbufs)
826: return NULL;
827:
828: if (offset == (off_t) -1) {
829: off = lseek(fd, 0, SEEK_CUR);
830: if (off == -1) {
831: LOGERR;
832: return NULL;
833: }
834: } else
835: off = offset;
836:
837: if (!(acb = calloc(sizeof(void*), nbufs))) {
838: LOGERR;
839: return NULL;
840: } else
841: memset(acb, 0, sizeof(void*) * nbufs);
842: for (i = 0; i < nbufs; off += bufs[i++].iov_len) {
843: acb[i] = malloc(sizeof(struct aiocb));
844: if (!acb[i]) {
845: LOGERR;
846: for (i = 0; i < nbufs; i++)
847: if (acb[i])
848: free(acb[i]);
849: free(acb);
850: return NULL;
851: } else
852: memset(acb[i], 0, sizeof(struct aiocb));
853: acb[i]->aio_fildes = fd;
854: acb[i]->aio_nbytes = bufs[i].iov_len;
855: acb[i]->aio_buf = bufs[i].iov_base;
856: acb[i]->aio_offset = off;
857: acb[i]->aio_lio_opcode = LIO_READ;
858: }
859: memset(&sig, 0, sizeof sig);
860: sig.sigev_notify = SIGEV_KEVENT;
861: sig.sigev_notify_kqueue = root->root_kq;
862: sig.sigev_value.sival_ptr = acb;
863:
864: if (lio_listio(LIO_NOWAIT, acb, nbufs, &sig)) {
865: LOGERR;
866: for (i = 0; i < nbufs; i++)
867: if (acb[i])
868: free(acb[i]);
869: free(acb);
870: return NULL;
871: }
872:
873: return schedLIO(root, func, arg, (void*) acb, bufs, nbufs);
874: }
875:
876: /*
877: * schedLIOWrite() - Add list of AIO write tasks to scheduler queue
878: *
879: * @root = root task
880: * @func = task execution function
881: * @arg = 1st func argument
882: * @fd = file descriptor
883: * @bufs = Buffer's list
884: * @nbufs = Number of Buffers
885: * @offset = Offset from start of file, if =-1 from current position
886: * return: NULL error or !=NULL new queued task
887: */
888: sched_task_t *
889: schedLIOWrite(sched_root_task_t * __restrict root, sched_task_func_t func, void *arg, int fd,
890: struct iovec *bufs, size_t nbufs, off_t offset)
891: {
892: struct sigevent sig;
893: struct aiocb **acb;
894: off_t off;
895: register int i;
896:
897: if (!root || !func || !bufs || !nbufs)
898: return NULL;
899:
900: if (offset == (off_t) -1) {
901: off = lseek(fd, 0, SEEK_CUR);
902: if (off == -1) {
903: LOGERR;
904: return NULL;
905: }
906: } else
907: off = offset;
908:
909: if (!(acb = calloc(sizeof(void*), nbufs))) {
910: LOGERR;
911: return NULL;
912: } else
913: memset(acb, 0, sizeof(void*) * nbufs);
914: for (i = 0; i < nbufs; off += bufs[i++].iov_len) {
915: acb[i] = malloc(sizeof(struct aiocb));
916: if (!acb[i]) {
917: LOGERR;
918: for (i = 0; i < nbufs; i++)
919: if (acb[i])
920: free(acb[i]);
921: free(acb);
922: return NULL;
923: } else
924: memset(acb[i], 0, sizeof(struct aiocb));
925: acb[i]->aio_fildes = fd;
926: acb[i]->aio_nbytes = bufs[i].iov_len;
927: acb[i]->aio_buf = bufs[i].iov_base;
928: acb[i]->aio_offset = off;
929: acb[i]->aio_lio_opcode = LIO_WRITE;
930: }
931: memset(&sig, 0, sizeof sig);
932: sig.sigev_notify = SIGEV_KEVENT;
933: sig.sigev_notify_kqueue = root->root_kq;
934: sig.sigev_value.sival_ptr = acb;
935:
936: if (lio_listio(LIO_NOWAIT, acb, nbufs, &sig)) {
937: LOGERR;
938: for (i = 0; i < nbufs; i++)
939: if (acb[i])
940: free(acb[i]);
941: free(acb);
942: return NULL;
943: }
944:
945: return schedLIO(root, func, arg, (void*) acb, bufs, nbufs);
946: }
947: #endif /* EVFILT_LIO */
948: #endif /* AIO_SUPPORT */
949:
950: /*
951: * schedTimer() - Add TIMER task to scheduler queue
952: *
953: * @root = root task
954: * @func = task execution function
955: * @arg = 1st func argument
956: * @ts = timeout argument structure
957: * @opt_data = Optional data
958: * @opt_dlen = Optional data length
959: * return: NULL error or !=NULL new queued task
960: */
961: sched_task_t *
962: schedTimer(sched_root_task_t * __restrict root, sched_task_func_t func, void *arg, struct timespec ts,
963: void *opt_data, size_t opt_dlen)
964: {
965: sched_task_t *task, *tmp, *t = NULL;
966: void *ptr;
967: struct timespec now;
968:
969: if (!root || !func)
970: return NULL;
971:
972: /* get new task */
973: if (!(task = sched_useTask(root)))
974: return NULL;
975:
976: task->task_func = func;
977: TASK_TYPE(task) = taskTIMER;
978: TASK_ROOT(task) = root;
979:
980: TASK_ARG(task) = arg;
981:
982: TASK_DATA(task) = opt_data;
983: TASK_DATLEN(task) = opt_dlen;
984:
985: /* calculate timeval structure */
986: clock_gettime(CLOCK_MONOTONIC, &now);
987: now.tv_sec += ts.tv_sec;
988: now.tv_nsec += ts.tv_nsec;
989: if (now.tv_nsec >= 1000000000L) {
990: now.tv_sec++;
991: now.tv_nsec -= 1000000000L;
992: } else if (now.tv_nsec < 0) {
993: now.tv_sec--;
994: now.tv_nsec += 1000000000L;
995: }
996: TASK_TS(task) = now;
997:
998: if (root->root_hooks.hook_add.timer)
999: ptr = root->root_hooks.hook_add.timer(task, NULL);
1000: else
1001: ptr = NULL;
1002:
1003: if (!ptr) {
1004: #ifdef HAVE_LIBPTHREAD
1005: pthread_mutex_lock(&root->root_mtx[taskTIMER]);
1006: #endif
1007: #ifdef TIMER_WITHOUT_SORT
1008: TAILQ_INSERT_TAIL(&root->root_timer, TASK_ID(task), task_node);
1009: #else
1010: TAILQ_FOREACH_SAFE(t, &root->root_timer, task_node, tmp)
1011: if (sched_timespeccmp(&TASK_TS(task), &TASK_TS(t), -) < 1)
1012: break;
1013: if (!t)
1014: TAILQ_INSERT_TAIL(&root->root_timer, TASK_ID(task), task_node);
1015: else
1016: TAILQ_INSERT_BEFORE(t, TASK_ID(task), task_node);
1017: #endif
1018: #ifdef HAVE_LIBPTHREAD
1019: pthread_mutex_unlock(&root->root_mtx[taskTIMER]);
1020: #endif
1021: } else
1022: task = sched_unuseTask(task);
1023:
1024: return task;
1025: }
1026:
1027: /*
1028: * schedEvent() - Add EVENT task to scheduler queue
1029: *
1030: * @root = root task
1031: * @func = task execution function
1032: * @arg = 1st func argument
1033: * @val = additional func argument
1034: * @opt_data = Optional data
1035: * @opt_dlen = Optional data length
1036: * return: NULL error or !=NULL new queued task
1037: */
1038: sched_task_t *
1039: schedEvent(sched_root_task_t * __restrict root, sched_task_func_t func, void *arg, u_long val,
1040: void *opt_data, size_t opt_dlen)
1041: {
1042: sched_task_t *task;
1043: void *ptr;
1044:
1045: if (!root || !func)
1046: return NULL;
1047:
1048: /* get new task */
1049: if (!(task = sched_useTask(root)))
1050: return NULL;
1051:
1052: task->task_func = func;
1053: TASK_TYPE(task) = taskEVENT;
1054: TASK_ROOT(task) = root;
1055:
1056: TASK_ARG(task) = arg;
1057: TASK_VAL(task) = val;
1058:
1059: TASK_DATA(task) = opt_data;
1060: TASK_DATLEN(task) = opt_dlen;
1061:
1062: if (root->root_hooks.hook_add.event)
1063: ptr = root->root_hooks.hook_add.event(task, NULL);
1064: else
1065: ptr = NULL;
1066:
1067: if (!ptr) {
1068: #ifdef HAVE_LIBPTHREAD
1069: pthread_mutex_lock(&root->root_mtx[taskEVENT]);
1070: #endif
1071: TAILQ_INSERT_TAIL(&root->root_event, TASK_ID(task), task_node);
1072: #ifdef HAVE_LIBPTHREAD
1073: pthread_mutex_unlock(&root->root_mtx[taskEVENT]);
1074: #endif
1075: } else
1076: task = sched_unuseTask(task);
1077:
1078: return task;
1079: }
1080:
1081:
1082: /*
1083: * schedTask() - Add regular task to scheduler queue
1084: *
1085: * @root = root task
1086: * @func = task execution function
1087: * @arg = 1st func argument
1088: * @prio = regular task priority, 0 is hi priority for regular tasks
1089: * @opt_data = Optional data
1090: * @opt_dlen = Optional data length
1091: * return: NULL error or !=NULL new queued task
1092: */
1093: sched_task_t *
1094: schedTask(sched_root_task_t * __restrict root, sched_task_func_t func, void *arg, u_long prio,
1095: void *opt_data, size_t opt_dlen)
1096: {
1097: sched_task_t *task, *tmp, *t = NULL;
1098: void *ptr;
1099:
1100: if (!root || !func)
1101: return NULL;
1102:
1103: /* get new task */
1104: if (!(task = sched_useTask(root)))
1105: return NULL;
1106:
1107: task->task_func = func;
1108: TASK_TYPE(task) = taskTASK;
1109: TASK_ROOT(task) = root;
1110:
1111: TASK_ARG(task) = arg;
1112: TASK_VAL(task) = prio;
1113:
1114: TASK_DATA(task) = opt_data;
1115: TASK_DATLEN(task) = opt_dlen;
1116:
1117: if (root->root_hooks.hook_add.task)
1118: ptr = root->root_hooks.hook_add.task(task, NULL);
1119: else
1120: ptr = NULL;
1121:
1122: if (!ptr) {
1123: #ifdef HAVE_LIBPTHREAD
1124: pthread_mutex_lock(&root->root_mtx[taskTASK]);
1125: #endif
1126: TAILQ_FOREACH_SAFE(t, &root->root_task, task_node, tmp)
1127: if (TASK_VAL(task) < TASK_VAL(t))
1128: break;
1129: if (!t)
1130: TAILQ_INSERT_TAIL(&root->root_task, TASK_ID(task), task_node);
1131: else
1132: TAILQ_INSERT_BEFORE(t, TASK_ID(task), task_node);
1133: #ifdef HAVE_LIBPTHREAD
1134: pthread_mutex_unlock(&root->root_mtx[taskTASK]);
1135: #endif
1136: } else
1137: task = sched_unuseTask(task);
1138:
1139: return task;
1140: }
1141:
1142: /*
1143: * schedSuspend() - Add Suspended task to scheduler queue
1144: *
1145: * @root = root task
1146: * @func = task execution function
1147: * @arg = 1st func argument
1148: * @id = Trigger ID
1149: * @opt_data = Optional data
1150: * @opt_dlen = Optional data length
1151: * return: NULL error or !=NULL new queued task
1152: */
1153: sched_task_t *
1154: schedSuspend(sched_root_task_t * __restrict root, sched_task_func_t func, void *arg, u_long id,
1155: void *opt_data, size_t opt_dlen)
1156: {
1157: sched_task_t *task;
1158: void *ptr;
1159:
1160: if (!root || !func)
1161: return NULL;
1162:
1163: /* get new task */
1164: if (!(task = sched_useTask(root)))
1165: return NULL;
1166:
1167: task->task_func = func;
1168: TASK_TYPE(task) = taskSUSPEND;
1169: TASK_ROOT(task) = root;
1170:
1171: TASK_ARG(task) = arg;
1172: TASK_VAL(task) = id;
1173:
1174: TASK_DATA(task) = opt_data;
1175: TASK_DATLEN(task) = opt_dlen;
1176:
1177: if (root->root_hooks.hook_add.suspend)
1178: ptr = root->root_hooks.hook_add.suspend(task, NULL);
1179: else
1180: ptr = NULL;
1181:
1182: if (!ptr) {
1183: #ifdef HAVE_LIBPTHREAD
1184: pthread_mutex_lock(&root->root_mtx[taskSUSPEND]);
1185: #endif
1186: TAILQ_INSERT_TAIL(&root->root_suspend, TASK_ID(task), task_node);
1187: #ifdef HAVE_LIBPTHREAD
1188: pthread_mutex_unlock(&root->root_mtx[taskSUSPEND]);
1189: #endif
1190: } else
1191: task = sched_unuseTask(task);
1192:
1193: return task;
1194: }
1195:
1196: /*
1197: * schedCallOnce() - Call once from scheduler
1198: *
1199: * @root = root task
1200: * @func = task execution function
1201: * @arg = 1st func argument
1202: * @val = additional func argument
1203: * @opt_data = Optional data
1204: * @opt_dlen = Optional data length
1205: * return: return value from called func
1206: */
1207: sched_task_t *
1208: schedCallOnce(sched_root_task_t * __restrict root, sched_task_func_t func, void *arg, u_long val,
1209: void *opt_data, size_t opt_dlen)
1210: {
1211: sched_task_t *task;
1212: void *ret;
1213:
1214: if (!root || !func)
1215: return NULL;
1216:
1217: /* get new task */
1218: if (!(task = sched_useTask(root)))
1219: return NULL;
1220:
1221: task->task_func = func;
1222: TASK_TYPE(task) = taskEVENT;
1223: TASK_ROOT(task) = root;
1224:
1225: TASK_ARG(task) = arg;
1226: TASK_VAL(task) = val;
1227:
1228: TASK_DATA(task) = opt_data;
1229: TASK_DATLEN(task) = opt_dlen;
1230:
1231: ret = schedCall(task);
1232:
1233: sched_unuseTask(task);
1234: return ret;
1235: }
1236:
1237: /*
1238: * schedThread() - Add thread task to scheduler queue
1239: *
1240: * @root = root task
1241: * @func = task execution function
1242: * @arg = 1st func argument
1243: * @ss = stack size
1244: * @opt_data = Optional data
1245: * @opt_dlen = Optional data length
1246: * return: NULL error or !=NULL new queued task
1247: */
1248: sched_task_t *
1249: schedThread(sched_root_task_t * __restrict root, sched_task_func_t func, void *arg,
1250: size_t ss, void *opt_data, size_t opt_dlen)
1251: {
1252: #ifndef HAVE_LIBPTHREAD
1253: sched_SetErr(ENOTSUP, "Not supported thread tasks");
1254: return NULL;
1255: #endif
1256: sched_task_t *task;
1257: pthread_attr_t attr;
1258:
1259: if (!root || !func)
1260: return NULL;
1261:
1262: /* get new task */
1263: if (!(task = sched_useTask(root))) {
1264:
1265: return NULL;
1266: }
1267:
1268: task->task_func = func;
1269: TASK_TYPE(task) = taskTHREAD;
1270: TASK_ROOT(task) = root;
1271:
1272: TASK_ARG(task) = arg;
1273:
1274: TASK_DATA(task) = opt_data;
1275: TASK_DATLEN(task) = opt_dlen;
1276:
1277: pthread_attr_init(&attr);
1278: pthread_attr_setdetachstate(&attr, PTHREAD_DETACHED);
1279: if (ss && (errno = pthread_attr_setstacksize(&attr, ss))) {
1280: LOGERR;
1281: pthread_attr_destroy(&attr);
1282: return sched_unuseTask(task);
1283: }
1284: if ((errno = pthread_attr_getstacksize(&attr, &ss))) {
1285: LOGERR;
1286: pthread_attr_destroy(&attr);
1287: return sched_unuseTask(task);
1288: } else
1289: TASK_FLAG(task) = ss;
1290: if ((errno = pthread_attr_setguardsize(&attr, ss))) {
1291: LOGERR;
1292: pthread_attr_destroy(&attr);
1293: return sched_unuseTask(task);
1294: }
1295: #ifdef SCHED_RR
1296: pthread_attr_setschedpolicy(&attr, SCHED_RR);
1297: #else
1298: pthread_attr_setschedpolicy(&attr, SCHED_OTHER);
1299: #endif
1300:
1301: pthread_mutex_lock(&root->root_mtx[taskTHREAD]);
1302: TAILQ_INSERT_TAIL(&root->root_thread, TASK_ID(task), task_node);
1303: pthread_mutex_unlock(&root->root_mtx[taskTHREAD]);
1304:
1305: if (root->root_hooks.hook_add.thread)
1306: if (root->root_hooks.hook_add.thread(task, &attr)) {
1307: schedCancel(task);
1308: task = NULL;
1309: }
1310: pthread_attr_destroy(&attr);
1311: return task;
1312: }
1313:
1314: /*
1315: * schedRTC() - Add RTC task to scheduler queue
1316: *
1317: * @root = root task
1318: * @func = task execution function
1319: * @arg = 1st func argument
1320: * @ts = timeout argument structure, minimum alarm timer resolution is 1msec!
1321: * @opt_data = Optional RTC ID
1322: * @opt_dlen = Optional data length
1323: * return: NULL error or !=NULL new queued task
1324: */
1325: sched_task_t *
1326: schedRTC(sched_root_task_t * __restrict root, sched_task_func_t func, void *arg, struct timespec ts,
1327: void *opt_data, size_t opt_dlen)
1328: {
1329: #if defined(HAVE_TIMER_CREATE) && defined(HAVE_TIMER_SETTIME)
1330: sched_task_t *task;
1331: void *ptr;
1332:
1333: if (!root || !func)
1334: return NULL;
1335:
1336: /* get new task */
1337: if (!(task = sched_useTask(root)))
1338: return NULL;
1339:
1340: task->task_func = func;
1341: TASK_TYPE(task) = taskRTC;
1342: TASK_ROOT(task) = root;
1343:
1344: TASK_ARG(task) = arg;
1345: TASK_TS(task) = ts;
1346:
1347: TASK_DATA(task) = opt_data;
1348: TASK_DATLEN(task) = opt_dlen;
1349:
1350: if (root->root_hooks.hook_add.rtc)
1351: ptr = root->root_hooks.hook_add.rtc(task, NULL);
1352: else
1353: ptr = NULL;
1354:
1355: if (!ptr) {
1356: #ifdef HAVE_LIBPTHREAD
1357: pthread_mutex_lock(&root->root_mtx[taskRTC]);
1358: #endif
1359: TAILQ_INSERT_TAIL(&root->root_rtc, TASK_ID(task), task_node);
1360: #ifdef HAVE_LIBPTHREAD
1361: pthread_mutex_unlock(&root->root_mtx[taskRTC]);
1362: #endif
1363: } else
1364: task = sched_unuseTask(task);
1365:
1366: return task;
1367: #else
1368: sched_SetErr(ENOTSUP, "Not supported realtime clock extensions");
1369: return NULL;
1370: #endif
1371: }
FreeBSD-CVSweb <freebsd-cvsweb@FreeBSD.org>
![[BACK]](/icons/cvsweb/back.gif){kind=icon}
Return to tasks.c CVS log ![[TXT]](/icons/cvsweb/text.gif){kind=icon}
![[DIR]](/icons/cvsweb/dir.gif){kind=icon}
Up to [ELWIX - Embedded LightWeight unIX -] / libaitsched / src