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