File:  [ELWIX - Embedded LightWeight unIX -] / libaitsched / src / tasks.c
Revision 1.10.2.5: download - view: text, annotated - select for diffs - revision graph
Wed Aug 1 22:12:39 2012 UTC (11 years, 11 months ago) by misho
Branches: sched2_6
Diff to: branchpoint 1.10: preferred, unified
ultra dummy bug

    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.10.2.5 2012/08/01 22:12:39 misho Exp $
    7: *
    8: **************************************************************************
    9: The ELWIX and AITNET software is distributed under the following
   10: terms:
   11: 
   12: All of the documentation and software included in the ELWIX and AITNET
   13: Releases is copyrighted by ELWIX - Sofia/Bulgaria <info@elwix.org>
   14: 
   15: Copyright 2004, 2005, 2006, 2007, 2008, 2009, 2010, 2011, 2012
   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: #pragma GCC visibility push(hidden)
   50: 
   51: inline sched_task_t *
   52: _sched_useTask(sched_root_task_t * __restrict root)
   53: {
   54: 	sched_task_t *task, *tmp;
   55: 
   56: 	TAILQ_FOREACH_SAFE(task, &root->root_unuse, task_node, tmp) {
   57: 		if (!TASK_ISLOCKED(task)) {
   58: #ifdef HAVE_LIBPTHREAD
   59: 			pthread_mutex_lock(&root->root_mtx[taskUNUSE]);
   60: #endif
   61: 			TAILQ_REMOVE(&root->root_unuse, task, task_node);
   62: #ifdef HAVE_LIBPTHREAD
   63: 			pthread_mutex_unlock(&root->root_mtx[taskUNUSE]);
   64: #endif
   65: 			break;
   66: 		}
   67: 	}
   68: 
   69: 	if (!task) {
   70: 		task = malloc(sizeof(sched_task_t));
   71: 		if (!task) {
   72: 			LOGERR;
   73: 			return NULL;
   74: 		}
   75: 	}
   76: 
   77: 	memset(task, 0, sizeof(sched_task_t));
   78: 	task->task_id = (uintptr_t) task;
   79: 	return task;
   80: }
   81: 
   82: inline sched_task_t *
   83: _sched_unuseTask(sched_task_t * __restrict task)
   84: {
   85: 	TASK_UNLOCK(task);
   86: 	TASK_TYPE(task) = taskUNUSE;
   87: #ifdef HAVE_LIBPTHREAD
   88: 	pthread_mutex_lock(&TASK_ROOT(task)->root_mtx[taskUNUSE]);
   89: #endif
   90: 	TAILQ_INSERT_TAIL(&TASK_ROOT(task)->root_unuse, TASK_ID(task), task_node);
   91: #ifdef HAVE_LIBPTHREAD
   92: 	pthread_mutex_unlock(&TASK_ROOT(task)->root_mtx[taskUNUSE]);
   93: #endif
   94: 	task = NULL;
   95: 
   96: 	return task;
   97: }
   98: 
   99: #pragma GCC visibility pop
  100: 
  101: 
  102: /*
  103:  * schedRead() - Add READ I/O task to scheduler queue
  104:  *
  105:  * @root = root task
  106:  * @func = task execution function
  107:  * @arg = 1st func argument
  108:  * @fd = fd handle
  109:  * @opt_data = Optional data
  110:  * @opt_dlen = Optional data length
  111:  * return: NULL error or !=NULL new queued task
  112:  */
  113: sched_task_t *
  114: schedRead(sched_root_task_t * __restrict root, sched_task_func_t func, void *arg, int fd, 
  115: 		void *opt_data, size_t opt_dlen)
  116: {
  117: 	sched_task_t *task;
  118: 	void *ptr;
  119: 
  120: 	if (!root || !func)
  121: 		return NULL;
  122: 
  123: 	/* get new task */
  124: 	if (!(task = _sched_useTask(root)))
  125: 		return NULL;
  126: 
  127: 	task->task_func = func;
  128: 	TASK_TYPE(task) = taskREAD;
  129: 	TASK_ROOT(task) = root;
  130: 
  131: 	TASK_ARG(task) = arg;
  132: 	TASK_FD(task) = fd;
  133: 
  134: 	TASK_DATA(task) = opt_data;
  135: 	TASK_DATLEN(task) = opt_dlen;
  136: 
  137: 	if (root->root_hooks.hook_add.read)
  138: 		ptr = root->root_hooks.hook_add.read(task, NULL);
  139: 	else
  140: 		ptr = NULL;
  141: 
  142: 	if (!ptr) {
  143: #ifdef HAVE_LIBPTHREAD
  144: 		pthread_mutex_lock(&root->root_mtx[taskREAD]);
  145: #endif
  146: 		TAILQ_INSERT_TAIL(&root->root_read, TASK_ID(task), task_node);
  147: #ifdef HAVE_LIBPTHREAD
  148: 		pthread_mutex_unlock(&root->root_mtx[taskREAD]);
  149: #endif
  150: 	} else
  151: 		task = _sched_unuseTask(task);
  152: 
  153: 	return task;
  154: }
  155: 
  156: /*
  157:  * schedWrite() - Add WRITE I/O task to scheduler queue
  158:  *
  159:  * @root = root task
  160:  * @func = task execution function
  161:  * @arg = 1st func argument
  162:  * @fd = fd handle
  163:  * @opt_data = Optional data
  164:  * @opt_dlen = Optional data length
  165:  * return: NULL error or !=NULL new queued task
  166:  */
  167: sched_task_t *
  168: schedWrite(sched_root_task_t * __restrict root, sched_task_func_t func, void *arg, int fd, 
  169: 		void *opt_data, size_t opt_dlen)
  170: {
  171: 	sched_task_t *task;
  172: 	void *ptr;
  173: 
  174: 	if (!root || !func)
  175: 		return NULL;
  176: 
  177: 	/* get new task */
  178: 	if (!(task = _sched_useTask(root)))
  179: 		return NULL;
  180: 
  181: 	task->task_func = func;
  182: 	TASK_TYPE(task) = taskWRITE;
  183: 	TASK_ROOT(task) = root;
  184: 
  185: 	TASK_ARG(task) = arg;
  186: 	TASK_FD(task) = fd;
  187: 
  188: 	TASK_DATA(task) = opt_data;
  189: 	TASK_DATLEN(task) = opt_dlen;
  190: 
  191: 	if (root->root_hooks.hook_add.write)
  192: 		ptr = root->root_hooks.hook_add.write(task, NULL);
  193: 	else
  194: 		ptr = NULL;
  195: 
  196: 	if (!ptr) {
  197: #ifdef HAVE_LIBPTHREAD
  198: 		pthread_mutex_lock(&root->root_mtx[taskWRITE]);
  199: #endif
  200: 		TAILQ_INSERT_TAIL(&root->root_write, TASK_ID(task), task_node);
  201: #ifdef HAVE_LIBPTHREAD
  202: 		pthread_mutex_unlock(&root->root_mtx[taskWRITE]);
  203: #endif
  204: 	} else
  205: 		task = _sched_unuseTask(task);
  206: 
  207: 	return task;
  208: }
  209: 
  210: /*
  211:  * schedNode() - Add NODE task to scheduler queue
  212:  *
  213:  * @root = root task
  214:  * @func = task execution function
  215:  * @arg = 1st func argument
  216:  * @fd = fd handle
  217:  * @opt_data = Optional data
  218:  * @opt_dlen = Optional data length
  219:  * return: NULL error or !=NULL new queued task
  220:  */
  221: sched_task_t *
  222: schedNode(sched_root_task_t * __restrict root, sched_task_func_t func, void *arg, int fd, 
  223: 		void *opt_data, size_t opt_dlen)
  224: {
  225: 	sched_task_t *task;
  226: 	void *ptr;
  227: 
  228: 	if (!root || !func)
  229: 		return NULL;
  230: 
  231: 	/* get new task */
  232: 	if (!(task = _sched_useTask(root)))
  233: 		return NULL;
  234: 
  235: 	task->task_func = func;
  236: 	TASK_TYPE(task) = taskNODE;
  237: 	TASK_ROOT(task) = root;
  238: 
  239: 	TASK_ARG(task) = arg;
  240: 	TASK_FD(task) = fd;
  241: 
  242: 	TASK_DATA(task) = opt_data;
  243: 	TASK_DATLEN(task) = opt_dlen;
  244: 
  245: 	if (root->root_hooks.hook_add.node)
  246: 		ptr = root->root_hooks.hook_add.node(task, NULL);
  247: 	else
  248: 		ptr = NULL;
  249: 
  250: 	if (!ptr) {
  251: #ifdef HAVE_LIBPTHREAD
  252: 		pthread_mutex_lock(&root->root_mtx[taskNODE]);
  253: #endif
  254: 		TAILQ_INSERT_TAIL(&root->root_node, TASK_ID(task), task_node);
  255: #ifdef HAVE_LIBPTHREAD
  256: 		pthread_mutex_unlock(&root->root_mtx[taskNODE]);
  257: #endif
  258: 	} else
  259: 		task = _sched_unuseTask(task);
  260: 
  261: 	return task;
  262: }
  263: 
  264: /*
  265:  * schedProc() - Add PROC task to scheduler queue
  266:  *
  267:  * @root = root task
  268:  * @func = task execution function
  269:  * @arg = 1st func argument
  270:  * @pid = PID
  271:  * @opt_data = Optional data
  272:  * @opt_dlen = Optional data length
  273:  * return: NULL error or !=NULL new queued task
  274:  */
  275: sched_task_t *
  276: schedProc(sched_root_task_t * __restrict root, sched_task_func_t func, void *arg, u_long pid, 
  277: 		void *opt_data, size_t opt_dlen)
  278: {
  279: 	sched_task_t *task;
  280: 	void *ptr;
  281: 
  282: 	if (!root || !func)
  283: 		return NULL;
  284: 
  285: 	/* get new task */
  286: 	if (!(task = _sched_useTask(root)))
  287: 		return NULL;
  288: 
  289: 	task->task_func = func;
  290: 	TASK_TYPE(task) = taskPROC;
  291: 	TASK_ROOT(task) = root;
  292: 
  293: 	TASK_ARG(task) = arg;
  294: 	TASK_VAL(task) = pid;
  295: 
  296: 	TASK_DATA(task) = opt_data;
  297: 	TASK_DATLEN(task) = opt_dlen;
  298: 
  299: 	if (root->root_hooks.hook_add.proc)
  300: 		ptr = root->root_hooks.hook_add.proc(task, NULL);
  301: 	else
  302: 		ptr = NULL;
  303: 
  304: 	if (!ptr) {
  305: #ifdef HAVE_LIBPTHREAD
  306: 		pthread_mutex_lock(&root->root_mtx[taskPROC]);
  307: #endif
  308: 		TAILQ_INSERT_TAIL(&root->root_proc, TASK_ID(task), task_node);
  309: #ifdef HAVE_LIBPTHREAD
  310: 		pthread_mutex_unlock(&root->root_mtx[taskPROC]);
  311: #endif
  312: 	} else
  313: 		task = _sched_unuseTask(task);
  314: 
  315: 	return task;
  316: }
  317: 
  318: /*
  319:  * schedUser() - Add trigger USER task to scheduler queue
  320:  *
  321:  * @root = root task
  322:  * @func = task execution function
  323:  * @arg = 1st func argument
  324:  * @id = Trigger ID
  325:  * @opt_data = Optional data
  326:  * @opt_dlen = Optional user's trigger flags
  327:  * return: NULL error or !=NULL new queued task
  328:  */
  329: sched_task_t *
  330: schedUser(sched_root_task_t * __restrict root, sched_task_func_t func, void *arg, u_long id, 
  331: 		void *opt_data, size_t opt_dlen)
  332: {
  333: #ifndef EVFILT_USER
  334: 	sched_SetErr(ENOTSUP, "Not supported kevent() filter");
  335: 	return NULL;
  336: #else
  337: 	sched_task_t *task;
  338: 	void *ptr;
  339: 
  340: 	if (!root || !func)
  341: 		return NULL;
  342: 
  343: 	/* get new task */
  344: 	if (!(task = _sched_useTask(root)))
  345: 		return NULL;
  346: 
  347: 	task->task_func = func;
  348: 	TASK_TYPE(task) = taskUSER;
  349: 	TASK_ROOT(task) = root;
  350: 
  351: 	TASK_ARG(task) = arg;
  352: 	TASK_VAL(task) = id;
  353: 
  354: 	TASK_DATA(task) = opt_data;
  355: 	TASK_DATLEN(task) = opt_dlen;
  356: 
  357: 	if (root->root_hooks.hook_add.user)
  358: 		ptr = root->root_hooks.hook_add.user(task, NULL);
  359: 	else
  360: 		ptr = NULL;
  361: 
  362: 	if (!ptr) {
  363: #ifdef HAVE_LIBPTHREAD
  364: 		pthread_mutex_lock(&root->root_mtx[taskUSER]);
  365: #endif
  366: 		TAILQ_INSERT_TAIL(&root->root_user, TASK_ID(task), task_node);
  367: #ifdef HAVE_LIBPTHREAD
  368: 		pthread_mutex_unlock(&root->root_mtx[taskUSER]);
  369: #endif
  370: 	} else
  371: 		task = _sched_unuseTask(task);
  372: 
  373: 	return task;
  374: #endif
  375: }
  376: 
  377: /*
  378:  * schedSignal() - Add SIGNAL task to scheduler queue
  379:  *
  380:  * @root = root task
  381:  * @func = task execution function
  382:  * @arg = 1st func argument
  383:  * @sig = Signal
  384:  * @opt_data = Optional data
  385:  * @opt_dlen = Optional data length
  386:  * return: NULL error or !=NULL new queued task
  387:  */
  388: sched_task_t *
  389: schedSignal(sched_root_task_t * __restrict root, sched_task_func_t func, void *arg, u_long sig, 
  390: 		void *opt_data, size_t opt_dlen)
  391: {
  392: 	sched_task_t *task;
  393: 	void *ptr;
  394: 
  395: 	if (!root || !func)
  396: 		return NULL;
  397: 
  398: 	/* get new task */
  399: 	if (!(task = _sched_useTask(root)))
  400: 		return NULL;
  401: 
  402: 	task->task_func = func;
  403: 	TASK_TYPE(task) = taskSIGNAL;
  404: 	TASK_ROOT(task) = root;
  405: 
  406: 	TASK_ARG(task) = arg;
  407: 	TASK_VAL(task) = sig;
  408: 
  409: 	TASK_DATA(task) = opt_data;
  410: 	TASK_DATLEN(task) = opt_dlen;
  411: 
  412: 	if (root->root_hooks.hook_add.signal)
  413: 		ptr = root->root_hooks.hook_add.signal(task, NULL);
  414: 	else
  415: 		ptr = NULL;
  416: 
  417: 	if (!ptr) {
  418: #ifdef HAVE_LIBPTHREAD
  419: 		pthread_mutex_lock(&root->root_mtx[taskSIGNAL]);
  420: #endif
  421: 		TAILQ_INSERT_TAIL(&root->root_signal, TASK_ID(task), task_node);
  422: #ifdef HAVE_LIBPTHREAD
  423: 		pthread_mutex_unlock(&root->root_mtx[taskSIGNAL]);
  424: #endif
  425: 	} else
  426: 		task = _sched_unuseTask(task);
  427: 
  428: 	return task;
  429: }
  430: 
  431: /*
  432:  * schedAlarm() - Add ALARM task to scheduler queue
  433:  *
  434:  * @root = root task
  435:  * @func = task execution function
  436:  * @arg = 1st func argument
  437:  * @ts = timeout argument structure, minimum alarm timer resolution is 1msec!
  438:  * @opt_data = Optional data
  439:  * @opt_dlen = Optional data length
  440:  * return: NULL error or !=NULL new queued task
  441:  */
  442: sched_task_t *
  443: schedAlarm(sched_root_task_t * __restrict root, sched_task_func_t func, void *arg, struct timespec ts, 
  444: 		void *opt_data, size_t opt_dlen)
  445: {
  446: 	sched_task_t *task;
  447: 	void *ptr;
  448: 
  449: 	if (!root || !func)
  450: 		return NULL;
  451: 
  452: 	/* get new task */
  453: 	if (!(task = _sched_useTask(root)))
  454: 		return NULL;
  455: 
  456: 	task->task_func = func;
  457: 	TASK_TYPE(task) = taskALARM;
  458: 	TASK_ROOT(task) = root;
  459: 
  460: 	TASK_ARG(task) = arg;
  461: 	TASK_TS(task) = ts;
  462: 
  463: 	TASK_DATA(task) = opt_data;
  464: 	TASK_DATLEN(task) = opt_dlen;
  465: 
  466: 	if (root->root_hooks.hook_add.alarm)
  467: 		ptr = root->root_hooks.hook_add.alarm(task, NULL);
  468: 	else
  469: 		ptr = NULL;
  470: 
  471: 	if (!ptr) {
  472: #ifdef HAVE_LIBPTHREAD
  473: 		pthread_mutex_lock(&root->root_mtx[taskALARM]);
  474: #endif
  475: 		TAILQ_INSERT_TAIL(&root->root_alarm, TASK_ID(task), task_node);
  476: #ifdef HAVE_LIBPTHREAD
  477: 		pthread_mutex_unlock(&root->root_mtx[taskALARM]);
  478: #endif
  479: 	} else
  480: 		task = _sched_unuseTask(task);
  481: 
  482: 	return task;
  483: }
  484: 
  485: /*
  486:  * schedAIO() - Add AIO task to scheduler queue
  487:  *
  488:  * @root = root task
  489:  * @func = task execution function
  490:  * @arg = 1st func argument
  491:  * @acb = AIO cb structure address
  492:  * @opt_data = Optional data
  493:  * @opt_dlen = Optional data length
  494:  * return: NULL error or !=NULL new queued task
  495:  */
  496: sched_task_t *
  497: schedAIO(sched_root_task_t * __restrict root, sched_task_func_t func, void *arg, 
  498: 		struct aiocb * __restrict acb, void *opt_data, size_t opt_dlen)
  499: {
  500: #ifndef EVFILT_AIO
  501: 	sched_SetErr(ENOTSUP, "Not supported kevent() filter");
  502: 	return NULL;
  503: #else
  504: 	sched_task_t *task;
  505: 	void *ptr;
  506: 
  507: 	if (!root || !func)
  508: 		return NULL;
  509: 
  510: 	/* get new task */
  511: 	if (!(task = _sched_useTask(root)))
  512: 		return NULL;
  513: 
  514: 	task->task_func = func;
  515: 	TASK_TYPE(task) = taskAIO;
  516: 	TASK_ROOT(task) = root;
  517: 
  518: 	TASK_ARG(task) = arg;
  519: 	TASK_VAL(task) = (u_long) acb;
  520: 
  521: 	TASK_DATA(task) = opt_data;
  522: 	TASK_DATLEN(task) = opt_dlen;
  523: 
  524: 	if (root->root_hooks.hook_add.aio)
  525: 		ptr = root->root_hooks.hook_add.aio(task, NULL);
  526: 	else
  527: 		ptr = NULL;
  528: 
  529: 	if (!ptr) {
  530: #ifdef HAVE_LIBPTHREAD
  531: 		pthread_mutex_lock(&root->root_mtx[taskAIO]);
  532: #endif
  533: 		TAILQ_INSERT_TAIL(&root->root_aio, TASK_ID(task), task_node);
  534: #ifdef HAVE_LIBPTHREAD
  535: 		pthread_mutex_unlock(&root->root_mtx[taskAIO]);
  536: #endif
  537: 	} else
  538: 		task = _sched_unuseTask(task);
  539: 
  540: 	return task;
  541: #endif
  542: }
  543: 
  544: /*
  545:  * schedAIORead() - Add AIO read task to scheduler queue
  546:  *
  547:  * @root = root task
  548:  * @func = task execution function
  549:  * @arg = 1st func argument
  550:  * @fd = file descriptor
  551:  * @buffer = Buffer
  552:  * @buflen = Buffer length
  553:  * return: NULL error or !=NULL new queued task
  554:  */
  555: inline sched_task_t *
  556: schedAIORead(sched_root_task_t * __restrict root, sched_task_func_t func, void *arg, int fd, 
  557: 		void *buffer, size_t buflen)
  558: {
  559: #ifndef EVFILT_AIO
  560: 	sched_SetErr(ENOTSUP, "Not supported kevent() filter");
  561: 	return NULL;
  562: #else
  563: 	struct aiocb *acb;
  564: 	off_t off = 0;
  565: 
  566: 	if (!root || !func)
  567: 		return NULL;
  568: 	else
  569: 		memset(buffer, 0, buflen);
  570: 
  571: 	if (!(acb = malloc(sizeof(struct aiocb)))) {
  572: 		LOGERR;
  573: 		return NULL;
  574: 	} else
  575: 		memset(acb, 0, sizeof(struct aiocb));
  576: 
  577: 	acb->aio_fildes = fd;
  578: 	acb->aio_nbytes = buflen;
  579: 	acb->aio_buf = buffer;
  580: 	off = lseek(fd, 0, SEEK_CUR);
  581: 	if (off == -1) {
  582: 		LOGERR;
  583: 		free(acb);
  584: 		return NULL;
  585: 	} else
  586: 		acb->aio_offset = off;
  587: 	acb->aio_sigevent.sigev_notify = SIGEV_KEVENT;
  588: 	acb->aio_sigevent.sigev_notify_kqueue = root->root_kq;
  589: 	acb->aio_sigevent.sigev_value.sival_ptr = acb;
  590: 
  591: 	if (aio_read(acb)) {
  592: 		LOGERR;
  593: 		free(acb);
  594: 		return NULL;
  595: 	}
  596: 
  597: 	return schedAIO(root, func, arg, acb, buffer, buflen);
  598: #endif
  599: }
  600: 
  601: /*
  602:  * schedAIOWrite() - Add AIO write task to scheduler queue
  603:  *
  604:  * @root = root task
  605:  * @func = task execution function
  606:  * @arg = 1st func argument
  607:  * @fd = file descriptor
  608:  * @buffer = Buffer
  609:  * @buflen = Buffer length
  610:  * return: NULL error or !=NULL new queued task
  611:  */
  612: inline sched_task_t *
  613: schedAIOWrite(sched_root_task_t * __restrict root, sched_task_func_t func, void *arg, int fd, 
  614: 		void *buffer, size_t buflen)
  615: {
  616: #ifndef EVFILT_AIO
  617: 	sched_SetErr(ENOTSUP, "Not supported kevent() filter");
  618: 	return NULL;
  619: #else
  620: 	struct aiocb *acb;
  621: 	off_t off = 0;
  622: 
  623: 	if (!root || !func)
  624: 		return NULL;
  625: 
  626: 	if (!(acb = malloc(sizeof(struct aiocb)))) {
  627: 		LOGERR;
  628: 		return NULL;
  629: 	} else
  630: 		memset(acb, 0, sizeof(struct aiocb));
  631: 
  632: 	acb->aio_fildes = fd;
  633: 	acb->aio_nbytes = buflen;
  634: 	acb->aio_buf = buffer;
  635: 	off = lseek(fd, 0, SEEK_CUR);
  636: 	if (off == -1) {
  637: 		LOGERR;
  638: 		free(acb);
  639: 		return NULL;
  640: 	} else
  641: 		acb->aio_offset = off;
  642: 	acb->aio_sigevent.sigev_notify = SIGEV_KEVENT;
  643: 	acb->aio_sigevent.sigev_notify_kqueue = root->root_kq;
  644: 	acb->aio_sigevent.sigev_value.sival_ptr = acb;
  645: 
  646: 	if (aio_write(acb)) {
  647: 		LOGERR;
  648: 		free(acb);
  649: 		return NULL;
  650: 	}
  651: 
  652: 	return schedAIO(root, func, arg, acb, buffer, buflen);
  653: #endif
  654: }
  655: 
  656: /*
  657:  * schedTimer() - Add TIMER task to scheduler queue
  658:  *
  659:  * @root = root task
  660:  * @func = task execution function
  661:  * @arg = 1st func argument
  662:  * @ts = timeout argument structure
  663:  * @opt_data = Optional data
  664:  * @opt_dlen = Optional data length
  665:  * return: NULL error or !=NULL new queued task
  666:  */
  667: sched_task_t *
  668: schedTimer(sched_root_task_t * __restrict root, sched_task_func_t func, void *arg, struct timespec ts, 
  669: 		void *opt_data, size_t opt_dlen)
  670: {
  671: 	sched_task_t *task, *tmp, *t = NULL;
  672: 	void *ptr;
  673: 	struct timespec now;
  674: 
  675: 	if (!root || !func)
  676: 		return NULL;
  677: 
  678: 	/* get new task */
  679: 	if (!(task = _sched_useTask(root)))
  680: 		return NULL;
  681: 
  682: 	task->task_func = func;
  683: 	TASK_TYPE(task) = taskTIMER;
  684: 	TASK_ROOT(task) = root;
  685: 
  686: 	TASK_ARG(task) = arg;
  687: 
  688: 	TASK_DATA(task) = opt_data;
  689: 	TASK_DATLEN(task) = opt_dlen;
  690: 
  691: 	/* calculate timeval structure */
  692: 	clock_gettime(CLOCK_MONOTONIC, &now);
  693: 	now.tv_sec += ts.tv_sec;
  694: 	now.tv_nsec += ts.tv_nsec;
  695: 	if (now.tv_nsec >= 1000000000L) {
  696: 		now.tv_sec++;
  697: 		now.tv_nsec -= 1000000000L;
  698: 	} else if (now.tv_nsec < 0) {
  699: 		now.tv_sec--;
  700: 		now.tv_nsec += 1000000000L;
  701: 	}
  702: 	TASK_TS(task) = now;
  703: 
  704: 	if (root->root_hooks.hook_add.timer)
  705: 		ptr = root->root_hooks.hook_add.timer(task, NULL);
  706: 	else
  707: 		ptr = NULL;
  708: 
  709: 	if (!ptr) {
  710: #ifdef HAVE_LIBPTHREAD
  711: 		pthread_mutex_lock(&root->root_mtx[taskTIMER]);
  712: #endif
  713: #ifdef TIMER_WITHOUT_SORT
  714: 		TAILQ_INSERT_TAIL(&root->root_timer, TASK_ID(task), task_node);
  715: #else
  716: 		TAILQ_FOREACH_SAFE(t, &root->root_timer, task_node, tmp)
  717: 			if (sched_timespeccmp(&TASK_TS(task), &TASK_TS(t), -) < 1)
  718: 				break;
  719: 		if (!t)
  720: 			TAILQ_INSERT_TAIL(&root->root_timer, TASK_ID(task), task_node);
  721: 		else
  722: 			TAILQ_INSERT_BEFORE(t, TASK_ID(task), task_node);
  723: #endif
  724: #ifdef HAVE_LIBPTHREAD
  725: 		pthread_mutex_unlock(&root->root_mtx[taskTIMER]);
  726: #endif
  727: 	} else
  728: 		task = _sched_unuseTask(task);
  729: 
  730: 	return task;
  731: }
  732: 
  733: /*
  734:  * schedEvent() - Add EVENT task to scheduler queue
  735:  *
  736:  * @root = root task
  737:  * @func = task execution function
  738:  * @arg = 1st func argument
  739:  * @val = additional func argument
  740:  * @opt_data = Optional data
  741:  * @opt_dlen = Optional data length
  742:  * return: NULL error or !=NULL new queued task
  743:  */
  744: sched_task_t *
  745: schedEvent(sched_root_task_t * __restrict root, sched_task_func_t func, void *arg, u_long val, 
  746: 		void *opt_data, size_t opt_dlen)
  747: {
  748: 	sched_task_t *task;
  749: 	void *ptr;
  750: 
  751: 	if (!root || !func)
  752: 		return NULL;
  753: 
  754: 	/* get new task */
  755: 	if (!(task = _sched_useTask(root)))
  756: 		return NULL;
  757: 
  758: 	task->task_func = func;
  759: 	TASK_TYPE(task) = taskEVENT;
  760: 	TASK_ROOT(task) = root;
  761: 
  762: 	TASK_ARG(task) = arg;
  763: 	TASK_VAL(task) = val;
  764: 
  765: 	TASK_DATA(task) = opt_data;
  766: 	TASK_DATLEN(task) = opt_dlen;
  767: 
  768: 	if (root->root_hooks.hook_add.event)
  769: 		ptr = root->root_hooks.hook_add.event(task, NULL);
  770: 	else
  771: 		ptr = NULL;
  772: 
  773: 	if (!ptr) {
  774: #ifdef HAVE_LIBPTHREAD
  775: 		pthread_mutex_lock(&root->root_mtx[taskEVENT]);
  776: #endif
  777: 		TAILQ_INSERT_TAIL(&root->root_event, TASK_ID(task), task_node);
  778: #ifdef HAVE_LIBPTHREAD
  779: 		pthread_mutex_unlock(&root->root_mtx[taskEVENT]);
  780: #endif
  781: 	} else
  782: 		task = _sched_unuseTask(task);
  783: 
  784: 	return task;
  785: }
  786: 
  787: 
  788: /*
  789:  * schedEventLo() - Add EVENT_Lo task to scheduler queue
  790:  *
  791:  * @root = root task
  792:  * @func = task execution function
  793:  * @arg = 1st func argument
  794:  * @val = additional func argument
  795:  * @opt_data = Optional data
  796:  * @opt_dlen = Optional data length
  797:  * return: NULL error or !=NULL new queued task
  798:  */
  799: sched_task_t *
  800: schedEventLo(sched_root_task_t * __restrict root, sched_task_func_t func, void *arg, u_long val, 
  801: 		void *opt_data, size_t opt_dlen)
  802: {
  803: 	sched_task_t *task;
  804: 	void *ptr;
  805: 
  806: 	if (!root || !func)
  807: 		return NULL;
  808: 
  809: 	/* get new task */
  810: 	if (!(task = _sched_useTask(root)))
  811: 		return NULL;
  812: 
  813: 	task->task_func = func;
  814: 	TASK_TYPE(task) = taskEVENT;
  815: 	TASK_ROOT(task) = root;
  816: 
  817: 	TASK_ARG(task) = arg;
  818: 	TASK_VAL(task) = val;
  819: 
  820: 	TASK_DATA(task) = opt_data;
  821: 	TASK_DATLEN(task) = opt_dlen;
  822: 
  823: 	if (root->root_hooks.hook_add.eventlo)
  824: 		ptr = root->root_hooks.hook_add.eventlo(task, NULL);
  825: 	else
  826: 		ptr = NULL;
  827: 
  828: 	if (!ptr) {
  829: #ifdef HAVE_LIBPTHREAD
  830: 		pthread_mutex_lock(&root->root_mtx[taskEVENTLO]);
  831: #endif
  832: 		TAILQ_INSERT_TAIL(&root->root_eventlo, TASK_ID(task), task_node);
  833: #ifdef HAVE_LIBPTHREAD
  834: 		pthread_mutex_unlock(&root->root_mtx[taskEVENTLO]);
  835: #endif
  836: 	} else
  837: 		task = _sched_unuseTask(task);
  838: 
  839: 	return task;
  840: }
  841: 
  842: /*
  843:  * schedSuspend() - Add Suspended task to scheduler queue
  844:  *
  845:  * @root = root task
  846:  * @func = task execution function
  847:  * @arg = 1st func argument
  848:  * @id = Trigger ID
  849:  * @opt_data = Optional data
  850:  * @opt_dlen = Optional data length
  851:  * return: NULL error or !=NULL new queued task
  852:  */
  853: sched_task_t *
  854: schedSuspend(sched_root_task_t * __restrict root, sched_task_func_t func, void *arg, u_long id, 
  855: 		void *opt_data, size_t opt_dlen)
  856: {
  857: 	sched_task_t *task;
  858: 	void *ptr;
  859: 
  860: 	if (!root || !func)
  861: 		return NULL;
  862: 
  863: 	/* get new task */
  864: 	if (!(task = _sched_useTask(root)))
  865: 		return NULL;
  866: 
  867: 	task->task_func = func;
  868: 	TASK_TYPE(task) = taskSUSPEND;
  869: 	TASK_ROOT(task) = root;
  870: 
  871: 	TASK_ARG(task) = arg;
  872: 	TASK_VAL(task) = id;
  873: 
  874: 	TASK_DATA(task) = opt_data;
  875: 	TASK_DATLEN(task) = opt_dlen;
  876: 
  877: 	if (root->root_hooks.hook_add.suspend)
  878: 		ptr = root->root_hooks.hook_add.suspend(task, NULL);
  879: 	else
  880: 		ptr = NULL;
  881: 
  882: 	if (!ptr) {
  883: #ifdef HAVE_LIBPTHREAD
  884: 		pthread_mutex_lock(&root->root_mtx[taskSUSPEND]);
  885: #endif
  886: 		TAILQ_INSERT_TAIL(&root->root_suspend, TASK_ID(task), task_node);
  887: #ifdef HAVE_LIBPTHREAD
  888: 		pthread_mutex_unlock(&root->root_mtx[taskSUSPEND]);
  889: #endif
  890: 	} else
  891: 		task = _sched_unuseTask(task);
  892: 
  893: 	return task;
  894: }
  895: 
  896: /*
  897:  * schedCallOnce() - Call once from scheduler
  898:  *
  899:  * @root = root task
  900:  * @func = task execution function
  901:  * @arg = 1st func argument
  902:  * @val = additional func argument
  903:  * @opt_data = Optional data
  904:  * @opt_dlen = Optional data length
  905:  * return: return value from called func
  906:  */
  907: sched_task_t *
  908: schedCallOnce(sched_root_task_t * __restrict root, sched_task_func_t func, void *arg, u_long val, 
  909: 		void *opt_data, size_t opt_dlen)
  910: {
  911: 	sched_task_t *task;
  912: 	void *ret;
  913: 
  914: 	if (!root || !func)
  915: 		return NULL;
  916: 
  917: 	/* get new task */
  918: 	if (!(task = _sched_useTask(root)))
  919: 		return NULL;
  920: 
  921: 	task->task_func = func;
  922: 	TASK_TYPE(task) = taskEVENT;
  923: 	TASK_ROOT(task) = root;
  924: 
  925: 	TASK_ARG(task) = arg;
  926: 	TASK_VAL(task) = val;
  927: 
  928: 	TASK_DATA(task) = opt_data;
  929: 	TASK_DATLEN(task) = opt_dlen;
  930: 
  931: 	ret = schedCall(task);
  932: 
  933: 	_sched_unuseTask(task);
  934: 	return ret;
  935: }

FreeBSD-CVSweb <freebsd-cvsweb@FreeBSD.org>