libaitsched/src/tasks.c - diff

Return to tasks.c CVS log

Up to [ELWIX - Embedded LightWeight unIX -] / libaitsched / src

Diff for /libaitsched/src/tasks.c between versions 1.18 and 1.29.4.1

version 1.18, 2013/08/22 15:28:04	version 1.29.4.1, 2022/11/28 23:29:59
Line 12 terms:	Line 12 terms:
All of the documentation and software included in the ELWIX and AITNET	All of the documentation and software included in the ELWIX and AITNET
Releases is copyrighted by ELWIX - Sofia/Bulgaria <info@elwix.org>	Releases is copyrighted by ELWIX - Sofia/Bulgaria <info@elwix.org>

Copyright 2004, 2005, 2006, 2007, 2008, 2009, 2010, 2011, 2012, 2013	Copyright 2004 - 2022
by Michael Pounov <misho@elwix.org>. All rights reserved.	by Michael Pounov <misho@elwix.org>. All rights reserved.

Redistribution and use in source and binary forms, with or without	Redistribution and use in source and binary forms, with or without
Line 57 sched_useTask(sched_root_task_t * __restrict root)	Line 57 sched_useTask(sched_root_task_t * __restrict root)
{	{
sched_task_t task, tmp;	sched_task_t task, tmp;

#ifdef HAVE_LIBPTHREAD	SCHED_QLOCK(root, taskUNUSE);
pthread_mutex_lock(&root->root_mtx[taskUNUSE]);
#endif
TAILQ_FOREACH_SAFE(task, &root->root_unuse, task_node, tmp) {	TAILQ_FOREACH_SAFE(task, &root->root_unuse, task_node, tmp) {
if (!TASK_ISLOCKED(task)) {	if (!TASK_ISLOCKED(task)) {
TAILQ_REMOVE(&root->root_unuse, task, task_node);	TAILQ_REMOVE(&root->root_unuse, task, task_node);
break;	break;
}	}
}	}
#ifdef HAVE_LIBPTHREAD	SCHED_QUNLOCK(root, taskUNUSE);
pthread_mutex_unlock(&root->root_mtx[taskUNUSE]);
#endif

if (!task) {	if (!task) {
task = malloc(sizeof(sched_task_t));	task = e_malloc(sizeof(sched_task_t));
if (!task) {	if (!task) {
LOGERR;	LOGERR;
return NULL;	return NULL;
Line 93 sched_task_t *	Line 89 sched_task_t *
sched_unuseTask(sched_task_t * __restrict task)	sched_unuseTask(sched_task_t * __restrict task)
{	{
TASK_UNLOCK(task);	TASK_UNLOCK(task);

TASK_TYPE(task) = taskUNUSE;	TASK_TYPE(task) = taskUNUSE;
#ifdef HAVE_LIBPTHREAD	insert_task_to(task, &(TASK_ROOT(task))->root_unuse);
pthread_mutex_lock(&TASK_ROOT(task)->root_mtx[taskUNUSE]);
#endif
TAILQ_INSERT_TAIL(&TASK_ROOT(task)->root_unuse, TASK_ID(task), task_node);
#ifdef HAVE_LIBPTHREAD
pthread_mutex_unlock(&TASK_ROOT(task)->root_mtx[taskUNUSE]);
#endif
task = NULL;

	task = NULL;
return task;	return task;
}	}

#pragma GCC visibility push(hidden)

#ifdef HAVE_LIBPTHREAD
static void
_sched_threadCleanup(sched_task_t *t)
{
if (!t \|\| !TASK_ROOT(t))
return;

if (TASK_FLAG(t) == PTHREAD_CREATE_JOINABLE)
pthread_detach(pthread_self());

pthread_mutex_lock(&TASK_ROOT(t)->root_mtx[taskTHREAD]);
TAILQ_REMOVE(&TASK_ROOT(t)->root_thread, t, task_node);
pthread_mutex_unlock(&TASK_ROOT(t)->root_mtx[taskTHREAD]);

sched_unuseTask(t);
}
void *
_sched_threadWrapper(sched_task_t *t)
{
void *ret = NULL;
sem_t *s = NULL;

if (!t \|\| !TASK_ROOT(t) \|\| !TASK_RET(t))
pthread_exit(ret);
else
s = (sem_t*) TASK_RET(t);

pthread_cleanup_push((void ()(void)) _sched_threadCleanup, t);

pthread_setcancelstate(PTHREAD_CANCEL_ENABLE, NULL);
pthread_setcanceltype(PTHREAD_CANCEL_ASYNCHRONOUS, NULL);

/* notify parent, thread is ready for execution */
sem_post(s);
pthread_testcancel();

ret = TASK_FUNC(t)(t);

pthread_cleanup_pop(42);
TASK_ROOT(t)->root_ret = ret;
pthread_exit(ret);
}
#endif

#pragma GCC visibility pop

/*	/*
* sched_taskExit() - Exit routine for scheduler task, explicit required for thread tasks	* sched_taskExit() - Exit routine for scheduler task, explicit required for thread tasks
*	*
Line 190 sched_task_t *	Line 133 sched_task_t *
schedRead(sched_root_task_t * __restrict root, sched_task_func_t func, void *arg, int fd,	schedRead(sched_root_task_t * __restrict root, sched_task_func_t func, void *arg, int fd,
void *opt_data, size_t opt_dlen)	void *opt_data, size_t opt_dlen)
{	{
	return schedReadExt(root, func, arg, fd, opt_data, opt_dlen, 0);
	}

	/*
	* schedReadExt() - Add READ I/O task to scheduler queue with custom event mask
	*
	* @root = root task
	* @func = task execution function
	* @arg = 1st func argument
	* @fd = fd handle
	* @opt_data = Optional data
	* @opt_dlen = Optional data length
	* @mask = Event mask
	* return: NULL error or !=NULL new queued task
	*/
	sched_task_t *
	schedReadExt(sched_root_task_t * __restrict root, sched_task_func_t func, void *arg, int fd,
	void *opt_data, size_t opt_dlen, u_long mask)
	{
sched_task_t *task;	sched_task_t *task;
void *ptr;	void *ptr;

Line 200 schedRead(sched_root_task_t * __restrict root, sched_t	Line 162 schedRead(sched_root_task_t * __restrict root, sched_t
if (!(task = sched_useTask(root)))	if (!(task = sched_useTask(root)))
return NULL;	return NULL;

task->task_func = func;	TASK_FUNC(task) = func;
TASK_TYPE(task) = taskREAD;	TASK_TYPE(task) = taskREAD;
TASK_ROOT(task) = root;	TASK_ROOT(task) = root;

Line 210 schedRead(sched_root_task_t * __restrict root, sched_t	Line 172 schedRead(sched_root_task_t * __restrict root, sched_t
TASK_DATA(task) = opt_data;	TASK_DATA(task) = opt_data;
TASK_DATLEN(task) = opt_dlen;	TASK_DATLEN(task) = opt_dlen;

	TASK_HARG(task) = (void*) mask;

if (root->root_hooks.hook_add.read)	if (root->root_hooks.hook_add.read)
ptr = root->root_hooks.hook_add.read(task, NULL);	ptr = root->root_hooks.hook_add.read(task, task->task_harg);
else	else
ptr = NULL;	ptr = NULL;

if (!ptr) {	if (!ptr)
#ifdef HAVE_LIBPTHREAD	insert_task_to(task, &root->root_read);
pthread_mutex_lock(&root->root_mtx[taskREAD]);	else
#endif
TAILQ_INSERT_TAIL(&root->root_read, TASK_ID(task), task_node);
#ifdef HAVE_LIBPTHREAD
pthread_mutex_unlock(&root->root_mtx[taskREAD]);
#endif
} else
task = sched_unuseTask(task);	task = sched_unuseTask(task);

return task;	return task;
Line 254 schedWrite(sched_root_task_t * __restrict root, sched_	Line 212 schedWrite(sched_root_task_t * __restrict root, sched_
if (!(task = sched_useTask(root)))	if (!(task = sched_useTask(root)))
return NULL;	return NULL;

task->task_func = func;	TASK_FUNC(task) = func;
TASK_TYPE(task) = taskWRITE;	TASK_TYPE(task) = taskWRITE;
TASK_ROOT(task) = root;	TASK_ROOT(task) = root;

Line 269 schedWrite(sched_root_task_t * __restrict root, sched_	Line 227 schedWrite(sched_root_task_t * __restrict root, sched_
else	else
ptr = NULL;	ptr = NULL;

if (!ptr) {	if (!ptr)
#ifdef HAVE_LIBPTHREAD	insert_task_to(task, &root->root_write);
pthread_mutex_lock(&root->root_mtx[taskWRITE]);	else
#endif
TAILQ_INSERT_TAIL(&root->root_write, TASK_ID(task), task_node);
#ifdef HAVE_LIBPTHREAD
pthread_mutex_unlock(&root->root_mtx[taskWRITE]);
#endif
} else
task = sched_unuseTask(task);	task = sched_unuseTask(task);

return task;	return task;
Line 298 sched_task_t *	Line 250 sched_task_t *
schedNode(sched_root_task_t * __restrict root, sched_task_func_t func, void *arg, int fd,	schedNode(sched_root_task_t * __restrict root, sched_task_func_t func, void *arg, int fd,
void *opt_data, size_t opt_dlen)	void *opt_data, size_t opt_dlen)
{	{
	#if SUP_ENABLE != KQ_SUPPORT
	sched_SetErr(ENOTSUP, "disabled kqueue support");
	return NULL;
	#else
sched_task_t *task;	sched_task_t *task;
void *ptr;	void *ptr;

Line 308 schedNode(sched_root_task_t * __restrict root, sched_t	Line 264 schedNode(sched_root_task_t * __restrict root, sched_t
if (!(task = sched_useTask(root)))	if (!(task = sched_useTask(root)))
return NULL;	return NULL;

task->task_func = func;	TASK_FUNC(task) = func;
TASK_TYPE(task) = taskNODE;	TASK_TYPE(task) = taskNODE;
TASK_ROOT(task) = root;	TASK_ROOT(task) = root;

Line 323 schedNode(sched_root_task_t * __restrict root, sched_t	Line 279 schedNode(sched_root_task_t * __restrict root, sched_t
else	else
ptr = NULL;	ptr = NULL;

if (!ptr) {	if (!ptr)
#ifdef HAVE_LIBPTHREAD	insert_task_to(task, &root->root_node);
pthread_mutex_lock(&root->root_mtx[taskNODE]);	else
	task = sched_unuseTask(task);

	return task;
	#endif /* KQ_SUPPORT */
	}

	/*
	* schedNode2() - Add NODE task with all events to scheduler queue
	*
	* @root = root task
	* @func = task execution function
	* @arg = 1st func argument
	* @fd = fd handle
	* @opt_data = Optional data
	* @opt_dlen = Optional data length
	* return: NULL error or !=NULL new queued task
	*/
	sched_task_t *
	schedNode2(sched_root_task_t * __restrict root, sched_task_func_t func, void *arg, int fd,
	void *opt_data, size_t opt_dlen)
	{
	#if SUP_ENABLE != KQ_SUPPORT
	sched_SetErr(ENOTSUP, "disabled kqueue support");
	return NULL;
	#else
	sched_task_t *task;
	void *ptr;

	if (!root \|\| !func)
	return NULL;

	/* get new task */
	if (!(task = sched_useTask(root)))
	return NULL;

	TASK_FUNC(task) = func;
	TASK_TYPE(task) = taskNODE;
	TASK_ROOT(task) = root;

	TASK_ARG(task) = arg;
	TASK_FD(task) = fd;

	TASK_DATA(task) = opt_data;
	TASK_DATLEN(task) = opt_dlen;

	if (root->root_hooks.hook_add.node)
	#ifdef __FreeBSD__
	ptr = root->root_hooks.hook_add.node(task,
	(void*) (NOTE_READ \| NOTE_CLOSE_WRITE \| NOTE_CLOSE \| NOTE_OPEN));
	#else
	ptr = root->root_hooks.hook_add.node(task, NULL);
#endif	#endif
TAILQ_INSERT_TAIL(&root->root_node, TASK_ID(task), task_node);	else
#ifdef HAVE_LIBPTHREAD	ptr = NULL;
pthread_mutex_unlock(&root->root_mtx[taskNODE]);
#endif	if (!ptr)
} else	insert_task_to(task, &root->root_node);
	else
task = sched_unuseTask(task);	task = sched_unuseTask(task);

return task;	return task;
	#endif /* KQ_SUPPORT */
}	}

/*	/*
Line 352 sched_task_t *	Line 361 sched_task_t *
schedProc(sched_root_task_t * __restrict root, sched_task_func_t func, void *arg, u_long pid,	schedProc(sched_root_task_t * __restrict root, sched_task_func_t func, void *arg, u_long pid,
void *opt_data, size_t opt_dlen)	void *opt_data, size_t opt_dlen)
{	{
	#if SUP_ENABLE != KQ_SUPPORT
	sched_SetErr(ENOTSUP, "disabled kqueue support");
	return NULL;
	#else
sched_task_t *task;	sched_task_t *task;
void *ptr;	void *ptr;

Line 362 schedProc(sched_root_task_t * __restrict root, sched_t	Line 375 schedProc(sched_root_task_t * __restrict root, sched_t
if (!(task = sched_useTask(root)))	if (!(task = sched_useTask(root)))
return NULL;	return NULL;

task->task_func = func;	TASK_FUNC(task) = func;
TASK_TYPE(task) = taskPROC;	TASK_TYPE(task) = taskPROC;
TASK_ROOT(task) = root;	TASK_ROOT(task) = root;

Line 377 schedProc(sched_root_task_t * __restrict root, sched_t	Line 390 schedProc(sched_root_task_t * __restrict root, sched_t
else	else
ptr = NULL;	ptr = NULL;

if (!ptr) {	if (!ptr)
#ifdef HAVE_LIBPTHREAD	insert_task_to(task, &root->root_proc);
pthread_mutex_lock(&root->root_mtx[taskPROC]);	else
#endif
TAILQ_INSERT_TAIL(&root->root_proc, TASK_ID(task), task_node);
#ifdef HAVE_LIBPTHREAD
pthread_mutex_unlock(&root->root_mtx[taskPROC]);
#endif
} else
task = sched_unuseTask(task);	task = sched_unuseTask(task);

return task;	return task;
	#endif /* KQ_SUPPORT */
}	}

/*	/*
Line 406 sched_task_t *	Line 414 sched_task_t *
schedUser(sched_root_task_t * __restrict root, sched_task_func_t func, void *arg, u_long id,	schedUser(sched_root_task_t * __restrict root, sched_task_func_t func, void *arg, u_long id,
void *opt_data, size_t opt_dlen)	void *opt_data, size_t opt_dlen)
{	{
	#if SUP_ENABLE != KQ_SUPPORT
	sched_SetErr(ENOTSUP, "disabled kqueue support");
	return NULL;
	#else
#ifndef EVFILT_USER	#ifndef EVFILT_USER
sched_SetErr(ENOTSUP, "Not supported kevent() filter");	sched_SetErr(ENOTSUP, "Not supported kevent() filter");
return NULL;	return NULL;
Line 420 schedUser(sched_root_task_t * __restrict root, sched_t	Line 432 schedUser(sched_root_task_t * __restrict root, sched_t
if (!(task = sched_useTask(root)))	if (!(task = sched_useTask(root)))
return NULL;	return NULL;

task->task_func = func;	TASK_FUNC(task) = func;
TASK_TYPE(task) = taskUSER;	TASK_TYPE(task) = taskUSER;
TASK_ROOT(task) = root;	TASK_ROOT(task) = root;

Line 435 schedUser(sched_root_task_t * __restrict root, sched_t	Line 447 schedUser(sched_root_task_t * __restrict root, sched_t
else	else
ptr = NULL;	ptr = NULL;

if (!ptr) {	if (!ptr)
#ifdef HAVE_LIBPTHREAD	insert_task_to(task, &root->root_user);
pthread_mutex_lock(&root->root_mtx[taskUSER]);	else
#endif
TAILQ_INSERT_TAIL(&root->root_user, TASK_ID(task), task_node);
#ifdef HAVE_LIBPTHREAD
pthread_mutex_unlock(&root->root_mtx[taskUSER]);
#endif
} else
task = sched_unuseTask(task);	task = sched_unuseTask(task);

return task;	return task;
#endif	#endif /* EVFILT_USER */
	#endif /* KQ_SUPPORT */
}	}

/*	/*
Line 465 sched_task_t *	Line 472 sched_task_t *
schedSignal(sched_root_task_t * __restrict root, sched_task_func_t func, void *arg, u_long sig,	schedSignal(sched_root_task_t * __restrict root, sched_task_func_t func, void *arg, u_long sig,
void *opt_data, size_t opt_dlen)	void *opt_data, size_t opt_dlen)
{	{
	#if SUP_ENABLE != KQ_SUPPORT
	sched_SetErr(ENOTSUP, "disabled kqueue support");
	return NULL;
	#else
sched_task_t *task;	sched_task_t *task;
void *ptr;	void *ptr;

Line 475 schedSignal(sched_root_task_t * __restrict root, sched	Line 486 schedSignal(sched_root_task_t * __restrict root, sched
if (!(task = sched_useTask(root)))	if (!(task = sched_useTask(root)))
return NULL;	return NULL;

task->task_func = func;	TASK_FUNC(task) = func;
TASK_TYPE(task) = taskSIGNAL;	TASK_TYPE(task) = taskSIGNAL;
TASK_ROOT(task) = root;	TASK_ROOT(task) = root;

Line 490 schedSignal(sched_root_task_t * __restrict root, sched	Line 501 schedSignal(sched_root_task_t * __restrict root, sched
else	else
ptr = NULL;	ptr = NULL;

if (!ptr) {	if (!ptr)
#ifdef HAVE_LIBPTHREAD	insert_task_to(task, &root->root_signal);
pthread_mutex_lock(&root->root_mtx[taskSIGNAL]);	else
#endif
TAILQ_INSERT_TAIL(&root->root_signal, TASK_ID(task), task_node);
#ifdef HAVE_LIBPTHREAD
pthread_mutex_unlock(&root->root_mtx[taskSIGNAL]);
#endif
} else
task = sched_unuseTask(task);	task = sched_unuseTask(task);

return task;	return task;
	#endif /* KQ_SUPPORT */
}	}

/*	/*
Line 519 sched_task_t *	Line 525 sched_task_t *
schedAlarm(sched_root_task_t * __restrict root, sched_task_func_t func, void *arg, struct timespec ts,	schedAlarm(sched_root_task_t * __restrict root, sched_task_func_t func, void *arg, struct timespec ts,
void *opt_data, size_t opt_dlen)	void *opt_data, size_t opt_dlen)
{	{
	#if SUP_ENABLE != KQ_SUPPORT
	sched_SetErr(ENOTSUP, "disabled kqueue support");
	return NULL;
	#else
sched_task_t *task;	sched_task_t *task;
void *ptr;	void *ptr;

Line 529 schedAlarm(sched_root_task_t * __restrict root, sched_	Line 539 schedAlarm(sched_root_task_t * __restrict root, sched_
if (!(task = sched_useTask(root)))	if (!(task = sched_useTask(root)))
return NULL;	return NULL;

task->task_func = func;	TASK_FUNC(task) = func;
TASK_TYPE(task) = taskALARM;	TASK_TYPE(task) = taskALARM;
TASK_ROOT(task) = root;	TASK_ROOT(task) = root;

Line 544 schedAlarm(sched_root_task_t * __restrict root, sched_	Line 554 schedAlarm(sched_root_task_t * __restrict root, sched_
else	else
ptr = NULL;	ptr = NULL;

if (!ptr) {	if (!ptr)
#ifdef HAVE_LIBPTHREAD	insert_task_to(task, &root->root_alarm);
pthread_mutex_lock(&root->root_mtx[taskALARM]);	else
#endif
TAILQ_INSERT_TAIL(&root->root_alarm, TASK_ID(task), task_node);
#ifdef HAVE_LIBPTHREAD
pthread_mutex_unlock(&root->root_mtx[taskALARM]);
#endif
} else
task = sched_unuseTask(task);	task = sched_unuseTask(task);

return task;	return task;
	#endif /* KQ_SUPPORT */
}	}

#ifdef AIO_SUPPORT	#ifdef AIO_SUPPORT
Line 574 sched_task_t *	Line 579 sched_task_t *
schedAIO(sched_root_task_t * __restrict root, sched_task_func_t func, void *arg,	schedAIO(sched_root_task_t * __restrict root, sched_task_func_t func, void *arg,
struct aiocb * __restrict acb, void *opt_data, size_t opt_dlen)	struct aiocb * __restrict acb, void *opt_data, size_t opt_dlen)
{	{
	#if SUP_ENABLE != KQ_SUPPORT
	sched_SetErr(ENOTSUP, "disabled kqueue support");
	return NULL;
	#else
sched_task_t *task;	sched_task_t *task;
void *ptr;	void *ptr;

Line 584 schedAIO(sched_root_task_t * __restrict root, sched_ta	Line 593 schedAIO(sched_root_task_t * __restrict root, sched_ta
if (!(task = sched_useTask(root)))	if (!(task = sched_useTask(root)))
return NULL;	return NULL;

task->task_func = func;	TASK_FUNC(task) = func;
TASK_TYPE(task) = taskAIO;	TASK_TYPE(task) = taskAIO;
TASK_ROOT(task) = root;	TASK_ROOT(task) = root;

Line 599 schedAIO(sched_root_task_t * __restrict root, sched_ta	Line 608 schedAIO(sched_root_task_t * __restrict root, sched_ta
else	else
ptr = NULL;	ptr = NULL;

if (!ptr) {	if (!ptr)
#ifdef HAVE_LIBPTHREAD	insert_task_to(task, &root->root_aio);
pthread_mutex_lock(&root->root_mtx[taskAIO]);	else
#endif
TAILQ_INSERT_TAIL(&root->root_aio, TASK_ID(task), task_node);
#ifdef HAVE_LIBPTHREAD
pthread_mutex_unlock(&root->root_mtx[taskAIO]);
#endif
} else
task = sched_unuseTask(task);	task = sched_unuseTask(task);

return task;	return task;
	#endif /* KQ_SUPPORT */
}	}

/*	/*
Line 629 sched_task_t *	Line 633 sched_task_t *
schedAIORead(sched_root_task_t * __restrict root, sched_task_func_t func, void *arg, int fd,	schedAIORead(sched_root_task_t * __restrict root, sched_task_func_t func, void *arg, int fd,
void *buffer, size_t buflen, off_t offset)	void *buffer, size_t buflen, off_t offset)
{	{
	#if SUP_ENABLE != KQ_SUPPORT
	sched_SetErr(ENOTSUP, "disabled kqueue support");
	return NULL;
	#else
struct aiocb *acb;	struct aiocb *acb;
off_t off;	off_t off;

Line 644 schedAIORead(sched_root_task_t * __restrict root, sche	Line 652 schedAIORead(sched_root_task_t * __restrict root, sche
} else	} else
off = offset;	off = offset;

if (!(acb = malloc(sizeof(struct aiocb)))) {	if (!(acb = e_malloc(sizeof(struct aiocb)))) {
LOGERR;	LOGERR;
return NULL;	return NULL;
} else	} else
Line 660 schedAIORead(sched_root_task_t * __restrict root, sche	Line 668 schedAIORead(sched_root_task_t * __restrict root, sche

if (aio_read(acb)) {	if (aio_read(acb)) {
LOGERR;	LOGERR;
free(acb);	e_free(acb);
return NULL;	return NULL;
}	}

return schedAIO(root, func, arg, acb, buffer, buflen);	return schedAIO(root, func, arg, acb, buffer, buflen);
	#endif /* KQ_SUPPORT */
}	}

/*	/*
Line 683 sched_task_t *	Line 692 sched_task_t *
schedAIOWrite(sched_root_task_t * __restrict root, sched_task_func_t func, void *arg, int fd,	schedAIOWrite(sched_root_task_t * __restrict root, sched_task_func_t func, void *arg, int fd,
void *buffer, size_t buflen, off_t offset)	void *buffer, size_t buflen, off_t offset)
{	{
	#if SUP_ENABLE != KQ_SUPPORT
	sched_SetErr(ENOTSUP, "disabled kqueue support");
	return NULL;
	#else
struct aiocb *acb;	struct aiocb *acb;
off_t off;	off_t off;

Line 698 schedAIOWrite(sched_root_task_t * __restrict root, sch	Line 711 schedAIOWrite(sched_root_task_t * __restrict root, sch
} else	} else
off = offset;	off = offset;

if (!(acb = malloc(sizeof(struct aiocb)))) {	if (!(acb = e_malloc(sizeof(struct aiocb)))) {
LOGERR;	LOGERR;
return NULL;	return NULL;
} else	} else
Line 714 schedAIOWrite(sched_root_task_t * __restrict root, sch	Line 727 schedAIOWrite(sched_root_task_t * __restrict root, sch

if (aio_write(acb)) {	if (aio_write(acb)) {
LOGERR;	LOGERR;
free(acb);	e_free(acb);
return NULL;	return NULL;
}	}

return schedAIO(root, func, arg, acb, buffer, buflen);	return schedAIO(root, func, arg, acb, buffer, buflen);
	#endif /* KQ_SUPPORT */
}	}

#ifdef EVFILT_LIO	#ifdef EVFILT_LIO
Line 737 sched_task_t *	Line 751 sched_task_t *
schedLIO(sched_root_task_t * __restrict root, sched_task_func_t func, void *arg,	schedLIO(sched_root_task_t * __restrict root, sched_task_func_t func, void *arg,
struct aiocb ** __restrict acbs, void *opt_data, size_t opt_dlen)	struct aiocb ** __restrict acbs, void *opt_data, size_t opt_dlen)
{	{
	#if SUP_ENABLE != KQ_SUPPORT
	sched_SetErr(ENOTSUP, "disabled kqueue support");
	return NULL;
	#else
sched_task_t *task;	sched_task_t *task;
void *ptr;	void *ptr;

Line 747 schedLIO(sched_root_task_t * __restrict root, sched_ta	Line 765 schedLIO(sched_root_task_t * __restrict root, sched_ta
if (!(task = sched_useTask(root)))	if (!(task = sched_useTask(root)))
return NULL;	return NULL;

task->task_func = func;	TASK_FUNC(task) = func;
TASK_TYPE(task) = taskLIO;	TASK_TYPE(task) = taskLIO;
TASK_ROOT(task) = root;	TASK_ROOT(task) = root;

Line 762 schedLIO(sched_root_task_t * __restrict root, sched_ta	Line 780 schedLIO(sched_root_task_t * __restrict root, sched_ta
else	else
ptr = NULL;	ptr = NULL;

if (!ptr) {	if (!ptr)
#ifdef HAVE_LIBPTHREAD	insert_task_to(task, &root->root_lio);
pthread_mutex_lock(&root->root_mtx[taskLIO]);	else
#endif
TAILQ_INSERT_TAIL(&root->root_lio, TASK_ID(task), task_node);
#ifdef HAVE_LIBPTHREAD
pthread_mutex_unlock(&root->root_mtx[taskLIO]);
#endif
} else
task = sched_unuseTask(task);	task = sched_unuseTask(task);

return task;	return task;
	#endif /* KQ_SUPPORT */
}	}

/*	/*
Line 792 sched_task_t *	Line 805 sched_task_t *
schedLIORead(sched_root_task_t * __restrict root, sched_task_func_t func, void *arg, int fd,	schedLIORead(sched_root_task_t * __restrict root, sched_task_func_t func, void *arg, int fd,
struct iovec *bufs, size_t nbufs, off_t offset)	struct iovec *bufs, size_t nbufs, off_t offset)
{	{
	#if SUP_ENABLE != KQ_SUPPORT
	sched_SetErr(ENOTSUP, "disabled kqueue support");
	return NULL;
	#else
struct sigevent sig;	struct sigevent sig;
struct aiocb **acb;	struct aiocb **acb;
off_t off;	off_t off;
Line 809 schedLIORead(sched_root_task_t * __restrict root, sche	Line 826 schedLIORead(sched_root_task_t * __restrict root, sche
} else	} else
off = offset;	off = offset;

if (!(acb = calloc(sizeof(void*), nbufs))) {	if (!(acb = e_calloc(sizeof(void*), nbufs))) {
LOGERR;	LOGERR;
return NULL;	return NULL;
} else	} else
memset(acb, 0, sizeof(void) nbufs);	memset(acb, 0, sizeof(void) nbufs);
for (i = 0; i < nbufs; off += bufs[i++].iov_len) {	for (i = 0; i < nbufs; off += bufs[i++].iov_len) {
acb[i] = malloc(sizeof(struct aiocb));	acb[i] = e_malloc(sizeof(struct aiocb));
if (!acb[i]) {	if (!acb[i]) {
LOGERR;	LOGERR;
for (i = 0; i < nbufs; i++)	for (i = 0; i < nbufs; i++)
if (acb[i])	if (acb[i])
free(acb[i]);	e_free(acb[i]);
free(acb);	e_free(acb);
return NULL;	return NULL;
} else	} else
memset(acb[i], 0, sizeof(struct aiocb));	memset(acb[i], 0, sizeof(struct aiocb));
Line 840 schedLIORead(sched_root_task_t * __restrict root, sche	Line 857 schedLIORead(sched_root_task_t * __restrict root, sche
LOGERR;	LOGERR;
for (i = 0; i < nbufs; i++)	for (i = 0; i < nbufs; i++)
if (acb[i])	if (acb[i])
free(acb[i]);	e_free(acb[i]);
free(acb);	e_free(acb);
return NULL;	return NULL;
}	}

return schedLIO(root, func, arg, (void*) acb, bufs, nbufs);	return schedLIO(root, func, arg, (void*) acb, bufs, nbufs);
	#endif /* KQ_SUPPORT */
}	}

/*	/*
Line 864 sched_task_t *	Line 882 sched_task_t *
schedLIOWrite(sched_root_task_t * __restrict root, sched_task_func_t func, void *arg, int fd,	schedLIOWrite(sched_root_task_t * __restrict root, sched_task_func_t func, void *arg, int fd,
struct iovec *bufs, size_t nbufs, off_t offset)	struct iovec *bufs, size_t nbufs, off_t offset)
{	{
	#if SUP_ENABLE != KQ_SUPPORT
	sched_SetErr(ENOTSUP, "disabled kqueue support");
	return NULL;
	#else
struct sigevent sig;	struct sigevent sig;
struct aiocb **acb;	struct aiocb **acb;
off_t off;	off_t off;
Line 881 schedLIOWrite(sched_root_task_t * __restrict root, sch	Line 903 schedLIOWrite(sched_root_task_t * __restrict root, sch
} else	} else
off = offset;	off = offset;

if (!(acb = calloc(sizeof(void*), nbufs))) {	if (!(acb = e_calloc(sizeof(void*), nbufs))) {
LOGERR;	LOGERR;
return NULL;	return NULL;
} else	} else
memset(acb, 0, sizeof(void) nbufs);	memset(acb, 0, sizeof(void) nbufs);
for (i = 0; i < nbufs; off += bufs[i++].iov_len) {	for (i = 0; i < nbufs; off += bufs[i++].iov_len) {
acb[i] = malloc(sizeof(struct aiocb));	acb[i] = e_malloc(sizeof(struct aiocb));
if (!acb[i]) {	if (!acb[i]) {
LOGERR;	LOGERR;
for (i = 0; i < nbufs; i++)	for (i = 0; i < nbufs; i++)
if (acb[i])	if (acb[i])
free(acb[i]);	e_free(acb[i]);
free(acb);	e_free(acb);
return NULL;	return NULL;
} else	} else
memset(acb[i], 0, sizeof(struct aiocb));	memset(acb[i], 0, sizeof(struct aiocb));
Line 912 schedLIOWrite(sched_root_task_t * __restrict root, sch	Line 934 schedLIOWrite(sched_root_task_t * __restrict root, sch
LOGERR;	LOGERR;
for (i = 0; i < nbufs; i++)	for (i = 0; i < nbufs; i++)
if (acb[i])	if (acb[i])
free(acb[i]);	e_free(acb[i]);
free(acb);	e_free(acb);
return NULL;	return NULL;
}	}

return schedLIO(root, func, arg, (void*) acb, bufs, nbufs);	return schedLIO(root, func, arg, (void*) acb, bufs, nbufs);
	#endif /* KQ_SUPPORT */
}	}
#endif /* EVFILT_LIO */	#endif /* EVFILT_LIO */
#endif /* AIO_SUPPORT */	#endif /* AIO_SUPPORT */
Line 948 schedTimer(sched_root_task_t * __restrict root, sched_	Line 971 schedTimer(sched_root_task_t * __restrict root, sched_
if (!(task = sched_useTask(root)))	if (!(task = sched_useTask(root)))
return NULL;	return NULL;

task->task_func = func;	TASK_FUNC(task) = func;
TASK_TYPE(task) = taskTIMER;	TASK_TYPE(task) = taskTIMER;
TASK_ROOT(task) = root;	TASK_ROOT(task) = root;

Line 976 schedTimer(sched_root_task_t * __restrict root, sched_	Line 999 schedTimer(sched_root_task_t * __restrict root, sched_
ptr = NULL;	ptr = NULL;

if (!ptr) {	if (!ptr) {
#ifdef HAVE_LIBPTHREAD	SCHED_QLOCK(root, taskTIMER);
pthread_mutex_lock(&root->root_mtx[taskTIMER]);
#endif
#ifdef TIMER_WITHOUT_SORT	#ifdef TIMER_WITHOUT_SORT
TAILQ_INSERT_TAIL(&root->root_timer, TASK_ID(task), task_node);	TAILQ_INSERT_TAIL(&root->root_timer, task, task_node);
#else	#else
TAILQ_FOREACH_SAFE(t, &root->root_timer, task_node, tmp)	TAILQ_FOREACH_SAFE(t, &root->root_timer, task_node, tmp)
if (sched_timespeccmp(&TASK_TS(task), &TASK_TS(t), -) < 1)	if (sched_timespeccmp(&TASK_TS(task), &TASK_TS(t), -) < 1)
break;	break;
if (!t)	if (!t)
TAILQ_INSERT_TAIL(&root->root_timer, TASK_ID(task), task_node);	TAILQ_INSERT_TAIL(&root->root_timer, task, task_node);
else	else
TAILQ_INSERT_BEFORE(t, TASK_ID(task), task_node);	TAILQ_INSERT_BEFORE(t, task, task_node);
#endif	#endif
#ifdef HAVE_LIBPTHREAD	SCHED_QUNLOCK(root, taskTIMER);
pthread_mutex_unlock(&root->root_mtx[taskTIMER]);
#endif
} else	} else
task = sched_unuseTask(task);	task = sched_unuseTask(task);

Line 1024 schedEvent(sched_root_task_t * __restrict root, sched_	Line 1043 schedEvent(sched_root_task_t * __restrict root, sched_
if (!(task = sched_useTask(root)))	if (!(task = sched_useTask(root)))
return NULL;	return NULL;

task->task_func = func;	TASK_FUNC(task) = func;
TASK_TYPE(task) = taskEVENT;	TASK_TYPE(task) = taskEVENT;
TASK_ROOT(task) = root;	TASK_ROOT(task) = root;

Line 1039 schedEvent(sched_root_task_t * __restrict root, sched_	Line 1058 schedEvent(sched_root_task_t * __restrict root, sched_
else	else
ptr = NULL;	ptr = NULL;

if (!ptr) {	if (!ptr)
#ifdef HAVE_LIBPTHREAD	insert_task_to(task, &root->root_event);
pthread_mutex_lock(&root->root_mtx[taskEVENT]);	else
#endif
TAILQ_INSERT_TAIL(&root->root_event, TASK_ID(task), task_node);
#ifdef HAVE_LIBPTHREAD
pthread_mutex_unlock(&root->root_mtx[taskEVENT]);
#endif
} else
task = sched_unuseTask(task);	task = sched_unuseTask(task);

return task;	return task;
Line 1079 schedTask(sched_root_task_t * __restrict root, sched_t	Line 1092 schedTask(sched_root_task_t * __restrict root, sched_t
if (!(task = sched_useTask(root)))	if (!(task = sched_useTask(root)))
return NULL;	return NULL;

task->task_func = func;	TASK_FUNC(task) = func;
TASK_TYPE(task) = taskTASK;	TASK_TYPE(task) = taskTASK;
TASK_ROOT(task) = root;	TASK_ROOT(task) = root;

Line 1095 schedTask(sched_root_task_t * __restrict root, sched_t	Line 1108 schedTask(sched_root_task_t * __restrict root, sched_t
ptr = NULL;	ptr = NULL;

if (!ptr) {	if (!ptr) {
#ifdef HAVE_LIBPTHREAD	SCHED_QLOCK(root, taskTASK);
pthread_mutex_lock(&root->root_mtx[taskTASK]);
#endif
TAILQ_FOREACH_SAFE(t, &root->root_task, task_node, tmp)	TAILQ_FOREACH_SAFE(t, &root->root_task, task_node, tmp)
if (TASK_VAL(task) < TASK_VAL(t))	if (TASK_VAL(task) < TASK_VAL(t))
break;	break;
if (!t)	if (!t)
TAILQ_INSERT_TAIL(&root->root_task, TASK_ID(task), task_node);	TAILQ_INSERT_TAIL(&root->root_task, task, task_node);
else	else
TAILQ_INSERT_BEFORE(t, TASK_ID(task), task_node);	TAILQ_INSERT_BEFORE(t, task, task_node);
#ifdef HAVE_LIBPTHREAD	SCHED_QUNLOCK(root, taskTASK);
pthread_mutex_unlock(&root->root_mtx[taskTASK]);
#endif
} else	} else
task = sched_unuseTask(task);	task = sched_unuseTask(task);

Line 1139 schedSuspend(sched_root_task_t * __restrict root, sche	Line 1148 schedSuspend(sched_root_task_t * __restrict root, sche
if (!(task = sched_useTask(root)))	if (!(task = sched_useTask(root)))
return NULL;	return NULL;

task->task_func = func;	TASK_FUNC(task) = func;
TASK_TYPE(task) = taskSUSPEND;	TASK_TYPE(task) = taskSUSPEND;
TASK_ROOT(task) = root;	TASK_ROOT(task) = root;

Line 1154 schedSuspend(sched_root_task_t * __restrict root, sche	Line 1163 schedSuspend(sched_root_task_t * __restrict root, sche
else	else
ptr = NULL;	ptr = NULL;

if (!ptr) {	if (!ptr)
#ifdef HAVE_LIBPTHREAD	insert_task_to(task, &root->root_suspend);
pthread_mutex_lock(&root->root_mtx[taskSUSPEND]);	else
#endif
TAILQ_INSERT_TAIL(&root->root_suspend, TASK_ID(task), task_node);
#ifdef HAVE_LIBPTHREAD
pthread_mutex_unlock(&root->root_mtx[taskSUSPEND]);
#endif
} else
task = sched_unuseTask(task);	task = sched_unuseTask(task);

return task;	return task;
Line 1193 schedCallOnce(sched_root_task_t * __restrict root, sch	Line 1196 schedCallOnce(sched_root_task_t * __restrict root, sch
if (!(task = sched_useTask(root)))	if (!(task = sched_useTask(root)))
return NULL;	return NULL;

task->task_func = func;	TASK_FUNC(task) = func;
TASK_TYPE(task) = taskEVENT;	TASK_TYPE(task) = taskEVENT;
TASK_ROOT(task) = root;	TASK_ROOT(task) = root;

Line 1215 schedCallOnce(sched_root_task_t * __restrict root, sch	Line 1218 schedCallOnce(sched_root_task_t * __restrict root, sch
* @root = root task	* @root = root task
* @func = task execution function	* @func = task execution function
* @arg = 1st func argument	* @arg = 1st func argument
* @detach = Detach thread from scheduler, if !=0
* @ss = stack size	* @ss = stack size
* @opt_data = Optional data	* @opt_data = Optional data
* @opt_dlen = Optional data length	* @opt_dlen = Optional data length
* return: NULL error or !=NULL new queued task	* return: NULL error or !=NULL new queued task
*/	*/
sched_task_t *	sched_task_t *
schedThread(sched_root_task_t * __restrict root, sched_task_func_t func, void *arg, int detach,	schedThread(sched_root_task_t * __restrict root, sched_task_func_t func, void *arg,
size_t ss, void *opt_data, size_t opt_dlen)	size_t ss, void *opt_data, size_t opt_dlen)
{	{
#ifndef HAVE_LIBPTHREAD	#ifndef HAVE_LIBPTHREAD
Line 1230 schedThread(sched_root_task_t * __restrict root, sched	Line 1232 schedThread(sched_root_task_t * __restrict root, sched
return NULL;	return NULL;
#endif	#endif
sched_task_t *task;	sched_task_t *task;
void *ptr;
pthread_attr_t attr;	pthread_attr_t attr;
sem_t *s = NULL;	void *ptr;

if (!root \|\| !func)	if (!root \|\| !func)
return NULL;	return NULL;
else {
/* normalizing stack size & detach state */
if (ss)
ss &= 0x7FFFFFFF;
detach = detach ? PTHREAD_CREATE_DETACHED : PTHREAD_CREATE_JOINABLE;
}

if (!(s = (sem_t*) malloc(sizeof(sem_t)))) {
LOGERR;
return NULL;
}
if (sem_init(s, 0, 1)) {
LOGERR;
free(s);
return NULL;
}

/* get new task */	/* get new task */
if (!(task = sched_useTask(root))) {	if (!(task = sched_useTask(root)))
sem_destroy(s);
free(s);

return NULL;	return NULL;
}

task->task_func = func;	TASK_FUNC(task) = func;
TASK_TYPE(task) = taskTHREAD;	TASK_TYPE(task) = taskTHREAD;
TASK_ROOT(task) = root;	TASK_ROOT(task) = root;

TASK_ARG(task) = arg;	TASK_ARG(task) = arg;
TASK_FLAG(task) = detach;
TASK_RET(task) = (intptr_t) s;

TASK_DATA(task) = opt_data;	TASK_DATA(task) = opt_data;
TASK_DATLEN(task) = opt_dlen;	TASK_DATLEN(task) = opt_dlen;

pthread_attr_init(&attr);	pthread_attr_init(&attr);
pthread_attr_setdetachstate(&attr, detach);	pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED);
if (ss && (errno = pthread_attr_setstacksize(&attr, ss))) {	if (ss && (errno = pthread_attr_setstacksize(&attr, ss))) {
LOGERR;	LOGERR;
pthread_attr_destroy(&attr);	pthread_attr_destroy(&attr);
sem_destroy(s);
free(s);
return sched_unuseTask(task);	return sched_unuseTask(task);
}	}
if ((errno = pthread_attr_getstacksize(&attr, &ss))) {	if ((errno = pthread_attr_getstacksize(&attr, &ss))) {
LOGERR;	LOGERR;
pthread_attr_destroy(&attr);	pthread_attr_destroy(&attr);
sem_destroy(s);
free(s);
return sched_unuseTask(task);	return sched_unuseTask(task);
} else	} else
TASK_FLAG(task) \|= (ss << 1);	TASK_FLAG(task) = ss;
if ((errno = pthread_attr_setguardsize(&attr, ss))) {
LOGERR;
pthread_attr_destroy(&attr);
sem_destroy(s);
free(s);
return sched_unuseTask(task);
}
#ifdef SCHED_RR	#ifdef SCHED_RR
pthread_attr_setschedpolicy(&attr, SCHED_RR);	pthread_attr_setschedpolicy(&attr, SCHED_RR);
#else	#else
pthread_attr_setschedpolicy(&attr, SCHED_OTHER);	pthread_attr_setschedpolicy(&attr, SCHED_OTHER);
#endif	#endif

if (root->root_hooks.hook_add.thread)	if (root->root_hooks.hook_add.thread)
ptr = root->root_hooks.hook_add.thread(task, &attr);	ptr = root->root_hooks.hook_add.thread(task, &attr);
else	else
ptr = NULL;	ptr = NULL;
pthread_attr_destroy(&attr);

if (!ptr) {	if (!ptr)
pthread_mutex_lock(&root->root_mtx[taskTHREAD]);	insert_task_to(task, &root->root_thread);
TAILQ_INSERT_TAIL(&root->root_thread, TASK_ID(task), task_node);	else
pthread_mutex_unlock(&root->root_mtx[taskTHREAD]);

/* wait for init thread actions */
sem_wait(s);
} else
task = sched_unuseTask(task);	task = sched_unuseTask(task);

sem_destroy(s);	pthread_attr_destroy(&attr);
free(s);
return task;	return task;
}	}

Line 1337 sched_task_t *	Line 1300 sched_task_t *
schedRTC(sched_root_task_t * __restrict root, sched_task_func_t func, void *arg, struct timespec ts,	schedRTC(sched_root_task_t * __restrict root, sched_task_func_t func, void *arg, struct timespec ts,
void *opt_data, size_t opt_dlen)	void *opt_data, size_t opt_dlen)
{	{
#if defined(HAVE_TIMER_CREATE) && defined(HAVE_TIMER_SETTIME)	#if defined(HAVE_LIBRT) && defined(HAVE_TIMER_CREATE) && \
	defined(HAVE_TIMER_SETTIME) && defined(HAVE_TIMER_DELETE)
sched_task_t *task;	sched_task_t *task;
void *ptr;	void *ptr;

Line 1348 schedRTC(sched_root_task_t * __restrict root, sched_ta	Line 1312 schedRTC(sched_root_task_t * __restrict root, sched_ta
if (!(task = sched_useTask(root)))	if (!(task = sched_useTask(root)))
return NULL;	return NULL;

task->task_func = func;	TASK_FUNC(task) = func;
TASK_TYPE(task) = taskRTC;	TASK_TYPE(task) = taskRTC;
TASK_ROOT(task) = root;	TASK_ROOT(task) = root;

Line 1363 schedRTC(sched_root_task_t * __restrict root, sched_ta	Line 1327 schedRTC(sched_root_task_t * __restrict root, sched_ta
else	else
ptr = NULL;	ptr = NULL;

if (!ptr) {	if (!ptr)
#ifdef HAVE_LIBPTHREAD	insert_task_to(task, &root->root_rtc);
pthread_mutex_lock(&root->root_mtx[taskRTC]);	else
#endif
TAILQ_INSERT_TAIL(&root->root_rtc, TASK_ID(task), task_node);
#ifdef HAVE_LIBPTHREAD
pthread_mutex_unlock(&root->root_mtx[taskRTC]);
#endif
} else
task = sched_unuseTask(task);	task = sched_unuseTask(task);

return task;	return task;

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

Removed from v.1.18
changed lines
	Added in v.1.29.4.1