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