File:  [ELWIX - Embedded LightWeight unIX -] / libaitsched / src / tasks.c
Revision 1.22.8.1: download - view: text, annotated - select for diffs - revision graph
Tue Jan 28 10:21:43 2014 UTC (10 years, 6 months ago) by misho
Branches: sched4_7
Diff to: branchpoint 1.22: preferred, unified
disable kqueue support

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

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