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