File:  [ELWIX - Embedded LightWeight unIX -] / libaitsched / src / tasks.c
Revision 1.10.2.3: download - view: text, annotated - select for diffs - revision graph
Wed Aug 1 14:11:43 2012 UTC (11 years, 11 months ago) by misho
Branches: sched2_6
Diff to: branchpoint 1.10: preferred, unified
go to test ;)

    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.3 2012/08/01 14:11:43 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_kevent_flags = EV_CLEAR | EV_DISPATCH;
  589: 	acb->aio_sigevent.sigev_notify_kqueue = root->root_kq;
  590: 	acb->aio_sigevent.sigev_value.sival_ptr = (void*) O_RDONLY;
  591: 
  592: 	if (aio_read(acb)) {
  593: 		LOGERR;
  594: 		free(acb);
  595: 		return NULL;
  596: 	}
  597: 
  598: 	return schedAIO(root, func, arg, acb, buffer, buflen);
  599: #endif
  600: }
  601: 
  602: /*
  603:  * schedAIOWrite() - Add AIO write task to scheduler queue
  604:  *
  605:  * @root = root task
  606:  * @func = task execution function
  607:  * @arg = 1st func argument
  608:  * @fd = file descriptor
  609:  * @buffer = Buffer
  610:  * @buflen = Buffer length
  611:  * return: NULL error or !=NULL new queued task
  612:  */
  613: inline sched_task_t *
  614: schedAIOWrite(sched_root_task_t * __restrict root, sched_task_func_t func, void *arg, int fd, 
  615: 		void *buffer, size_t buflen)
  616: {
  617: #ifndef EVFILT_AIO
  618: 	sched_SetErr(ENOTSUP, "Not supported kevent() filter");
  619: 	return NULL;
  620: #else
  621: 	struct aiocb *acb;
  622: 	off_t off = 0;
  623: 
  624: 	if (!root || !func)
  625: 		return NULL;
  626: 	else
  627: 		memset(buffer, 0, buflen);
  628: 
  629: 	if (!(acb = malloc(sizeof(struct aiocb)))) {
  630: 		LOGERR;
  631: 		return NULL;
  632: 	} else
  633: 		memset(acb, 0, sizeof(struct aiocb));
  634: 
  635: 	acb->aio_fildes = fd;
  636: 	acb->aio_nbytes = buflen;
  637: 	acb->aio_buf = buffer;
  638: 	off = lseek(fd, 0, SEEK_CUR);
  639: 	if (off == -1) {
  640: 		LOGERR;
  641: 		free(acb);
  642: 		return NULL;
  643: 	} else
  644: 		acb->aio_offset = off;
  645: 	acb->aio_sigevent.sigev_notify = SIGEV_KEVENT;
  646: 	acb->aio_sigevent.sigev_notify_kevent_flags = EV_CLEAR | EV_DISPATCH;
  647: 	acb->aio_sigevent.sigev_notify_kqueue = root->root_kq;
  648: 	acb->aio_sigevent.sigev_value.sival_ptr = (void*) O_WRONLY;
  649: 
  650: 	if (aio_write(acb)) {
  651: 		LOGERR;
  652: 		free(acb);
  653: 		return NULL;
  654: 	}
  655: 
  656: 	return schedAIO(root, func, arg, acb, buffer, buflen);
  657: #endif
  658: }
  659: 
  660: /*
  661:  * schedTimer() - Add TIMER task to scheduler queue
  662:  *
  663:  * @root = root task
  664:  * @func = task execution function
  665:  * @arg = 1st func argument
  666:  * @ts = timeout argument structure
  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: schedTimer(sched_root_task_t * __restrict root, sched_task_func_t func, void *arg, struct timespec ts, 
  673: 		void *opt_data, size_t opt_dlen)
  674: {
  675: 	sched_task_t *task, *tmp, *t = NULL;
  676: 	void *ptr;
  677: 	struct timespec now;
  678: 
  679: 	if (!root || !func)
  680: 		return NULL;
  681: 
  682: 	/* get new task */
  683: 	if (!(task = _sched_useTask(root)))
  684: 		return NULL;
  685: 
  686: 	task->task_func = func;
  687: 	TASK_TYPE(task) = taskTIMER;
  688: 	TASK_ROOT(task) = root;
  689: 
  690: 	TASK_ARG(task) = arg;
  691: 
  692: 	TASK_DATA(task) = opt_data;
  693: 	TASK_DATLEN(task) = opt_dlen;
  694: 
  695: 	/* calculate timeval structure */
  696: 	clock_gettime(CLOCK_MONOTONIC, &now);
  697: 	now.tv_sec += ts.tv_sec;
  698: 	now.tv_nsec += ts.tv_nsec;
  699: 	if (now.tv_nsec >= 1000000000L) {
  700: 		now.tv_sec++;
  701: 		now.tv_nsec -= 1000000000L;
  702: 	} else if (now.tv_nsec < 0) {
  703: 		now.tv_sec--;
  704: 		now.tv_nsec += 1000000000L;
  705: 	}
  706: 	TASK_TS(task) = now;
  707: 
  708: 	if (root->root_hooks.hook_add.timer)
  709: 		ptr = root->root_hooks.hook_add.timer(task, NULL);
  710: 	else
  711: 		ptr = NULL;
  712: 
  713: 	if (!ptr) {
  714: #ifdef HAVE_LIBPTHREAD
  715: 		pthread_mutex_lock(&root->root_mtx[taskTIMER]);
  716: #endif
  717: #ifdef TIMER_WITHOUT_SORT
  718: 		TAILQ_INSERT_TAIL(&root->root_timer, TASK_ID(task), task_node);
  719: #else
  720: 		TAILQ_FOREACH_SAFE(t, &root->root_timer, task_node, tmp)
  721: 			if (sched_timespeccmp(&TASK_TS(task), &TASK_TS(t), -) < 1)
  722: 				break;
  723: 		if (!t)
  724: 			TAILQ_INSERT_TAIL(&root->root_timer, TASK_ID(task), task_node);
  725: 		else
  726: 			TAILQ_INSERT_BEFORE(t, TASK_ID(task), task_node);
  727: #endif
  728: #ifdef HAVE_LIBPTHREAD
  729: 		pthread_mutex_unlock(&root->root_mtx[taskTIMER]);
  730: #endif
  731: 	} else
  732: 		task = _sched_unuseTask(task);
  733: 
  734: 	return task;
  735: }
  736: 
  737: /*
  738:  * schedEvent() - Add EVENT task to scheduler queue
  739:  *
  740:  * @root = root task
  741:  * @func = task execution function
  742:  * @arg = 1st func argument
  743:  * @val = additional func argument
  744:  * @opt_data = Optional data
  745:  * @opt_dlen = Optional data length
  746:  * return: NULL error or !=NULL new queued task
  747:  */
  748: sched_task_t *
  749: schedEvent(sched_root_task_t * __restrict root, sched_task_func_t func, void *arg, u_long val, 
  750: 		void *opt_data, size_t opt_dlen)
  751: {
  752: 	sched_task_t *task;
  753: 	void *ptr;
  754: 
  755: 	if (!root || !func)
  756: 		return NULL;
  757: 
  758: 	/* get new task */
  759: 	if (!(task = _sched_useTask(root)))
  760: 		return NULL;
  761: 
  762: 	task->task_func = func;
  763: 	TASK_TYPE(task) = taskEVENT;
  764: 	TASK_ROOT(task) = root;
  765: 
  766: 	TASK_ARG(task) = arg;
  767: 	TASK_VAL(task) = val;
  768: 
  769: 	TASK_DATA(task) = opt_data;
  770: 	TASK_DATLEN(task) = opt_dlen;
  771: 
  772: 	if (root->root_hooks.hook_add.event)
  773: 		ptr = root->root_hooks.hook_add.event(task, NULL);
  774: 	else
  775: 		ptr = NULL;
  776: 
  777: 	if (!ptr) {
  778: #ifdef HAVE_LIBPTHREAD
  779: 		pthread_mutex_lock(&root->root_mtx[taskEVENT]);
  780: #endif
  781: 		TAILQ_INSERT_TAIL(&root->root_event, TASK_ID(task), task_node);
  782: #ifdef HAVE_LIBPTHREAD
  783: 		pthread_mutex_unlock(&root->root_mtx[taskEVENT]);
  784: #endif
  785: 	} else
  786: 		task = _sched_unuseTask(task);
  787: 
  788: 	return task;
  789: }
  790: 
  791: 
  792: /*
  793:  * schedEventLo() - Add EVENT_Lo task to scheduler queue
  794:  *
  795:  * @root = root task
  796:  * @func = task execution function
  797:  * @arg = 1st func argument
  798:  * @val = additional func argument
  799:  * @opt_data = Optional data
  800:  * @opt_dlen = Optional data length
  801:  * return: NULL error or !=NULL new queued task
  802:  */
  803: sched_task_t *
  804: schedEventLo(sched_root_task_t * __restrict root, sched_task_func_t func, void *arg, u_long val, 
  805: 		void *opt_data, size_t opt_dlen)
  806: {
  807: 	sched_task_t *task;
  808: 	void *ptr;
  809: 
  810: 	if (!root || !func)
  811: 		return NULL;
  812: 
  813: 	/* get new task */
  814: 	if (!(task = _sched_useTask(root)))
  815: 		return NULL;
  816: 
  817: 	task->task_func = func;
  818: 	TASK_TYPE(task) = taskEVENT;
  819: 	TASK_ROOT(task) = root;
  820: 
  821: 	TASK_ARG(task) = arg;
  822: 	TASK_VAL(task) = val;
  823: 
  824: 	TASK_DATA(task) = opt_data;
  825: 	TASK_DATLEN(task) = opt_dlen;
  826: 
  827: 	if (root->root_hooks.hook_add.eventlo)
  828: 		ptr = root->root_hooks.hook_add.eventlo(task, NULL);
  829: 	else
  830: 		ptr = NULL;
  831: 
  832: 	if (!ptr) {
  833: #ifdef HAVE_LIBPTHREAD
  834: 		pthread_mutex_lock(&root->root_mtx[taskEVENTLO]);
  835: #endif
  836: 		TAILQ_INSERT_TAIL(&root->root_eventlo, TASK_ID(task), task_node);
  837: #ifdef HAVE_LIBPTHREAD
  838: 		pthread_mutex_unlock(&root->root_mtx[taskEVENTLO]);
  839: #endif
  840: 	} else
  841: 		task = _sched_unuseTask(task);
  842: 
  843: 	return task;
  844: }
  845: 
  846: /*
  847:  * schedSuspend() - Add Suspended task to scheduler queue
  848:  *
  849:  * @root = root task
  850:  * @func = task execution function
  851:  * @arg = 1st func argument
  852:  * @id = Trigger ID
  853:  * @opt_data = Optional data
  854:  * @opt_dlen = Optional data length
  855:  * return: NULL error or !=NULL new queued task
  856:  */
  857: sched_task_t *
  858: schedSuspend(sched_root_task_t * __restrict root, sched_task_func_t func, void *arg, u_long id, 
  859: 		void *opt_data, size_t opt_dlen)
  860: {
  861: 	sched_task_t *task;
  862: 	void *ptr;
  863: 
  864: 	if (!root || !func)
  865: 		return NULL;
  866: 
  867: 	/* get new task */
  868: 	if (!(task = _sched_useTask(root)))
  869: 		return NULL;
  870: 
  871: 	task->task_func = func;
  872: 	TASK_TYPE(task) = taskSUSPEND;
  873: 	TASK_ROOT(task) = root;
  874: 
  875: 	TASK_ARG(task) = arg;
  876: 	TASK_VAL(task) = id;
  877: 
  878: 	TASK_DATA(task) = opt_data;
  879: 	TASK_DATLEN(task) = opt_dlen;
  880: 
  881: 	if (root->root_hooks.hook_add.suspend)
  882: 		ptr = root->root_hooks.hook_add.suspend(task, NULL);
  883: 	else
  884: 		ptr = NULL;
  885: 
  886: 	if (!ptr) {
  887: #ifdef HAVE_LIBPTHREAD
  888: 		pthread_mutex_lock(&root->root_mtx[taskSUSPEND]);
  889: #endif
  890: 		TAILQ_INSERT_TAIL(&root->root_suspend, TASK_ID(task), task_node);
  891: #ifdef HAVE_LIBPTHREAD
  892: 		pthread_mutex_unlock(&root->root_mtx[taskSUSPEND]);
  893: #endif
  894: 	} else
  895: 		task = _sched_unuseTask(task);
  896: 
  897: 	return task;
  898: }
  899: 
  900: /*
  901:  * schedCallOnce() - Call once from scheduler
  902:  *
  903:  * @root = root task
  904:  * @func = task execution function
  905:  * @arg = 1st func argument
  906:  * @val = additional func argument
  907:  * @opt_data = Optional data
  908:  * @opt_dlen = Optional data length
  909:  * return: return value from called func
  910:  */
  911: sched_task_t *
  912: schedCallOnce(sched_root_task_t * __restrict root, sched_task_func_t func, void *arg, u_long val, 
  913: 		void *opt_data, size_t opt_dlen)
  914: {
  915: 	sched_task_t *task;
  916: 	void *ret;
  917: 
  918: 	if (!root || !func)
  919: 		return NULL;
  920: 
  921: 	/* get new task */
  922: 	if (!(task = _sched_useTask(root)))
  923: 		return NULL;
  924: 
  925: 	task->task_func = func;
  926: 	TASK_TYPE(task) = taskEVENT;
  927: 	TASK_ROOT(task) = root;
  928: 
  929: 	TASK_ARG(task) = arg;
  930: 	TASK_VAL(task) = val;
  931: 
  932: 	TASK_DATA(task) = opt_data;
  933: 	TASK_DATLEN(task) = opt_dlen;
  934: 
  935: 	ret = schedCall(task);
  936: 
  937: 	_sched_unuseTask(task);
  938: 	return ret;
  939: }

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