d/freebsd-dev
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

1. Always call user callback function in newly created thread, it seems

Browse Source 
POSIX implies that the user callback function must be executed in clean
environment.
2. Use newly introduced pthread stubs in libc.
This commit is contained in:
David Xu 2006-03-07 08:28:07 +00:00
parent 8682859c8c
commit c5a4f4eff2
Notes: svn2git 2020-12-20 02:59:44 +00:00 
svn path=/head/; revision=156383
5 changed files with 176 additions and 418 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
lib/librt/aio.c
View File

@ -34,10 +34,8 @@
#include "namespace.h"
#include <errno.h>
#include <pthread.h>
#include <stddef.h>
#include <signal.h>
#include <unistd.h>
#include "sigev_thread.h"
#include "un-namespace.h"
@ -92,7 +90,6 @@ aio_io(struct aiocb *iocb, int (*sysfunc)(struct aiocb *iocb))
saved_ev = iocb->aio_sigevent;
sn->sn_id = (sigev_id_t)iocb;
sn->sn_flags |= SNF_ONESHOT;
__sigev_get_sigevent(sn, &iocb->aio_sigevent, sn->sn_id);
__sigev_list_lock();

2
lib/librt/mq.c
View File

@ -33,11 +33,9 @@
#include "namespace.h"
#include <errno.h>
#include <pthread.h>
#include <stddef.h>
#include <stdlib.h>
#include <signal.h>
#include <unistd.h>
#include "sigev_thread.h"
#include "un-namespace.h"

549
lib/librt/sigev_thread.c
View File

@ -31,12 +31,10 @@
#include <machine/atomic.h>
#include "namespace.h"
#include <assert.h>
#include <err.h>
#include <errno.h>
#include <ucontext.h>
#include <sys/thr.h>
#include <sys/time.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
@ -46,96 +44,59 @@
#include "sigev_thread.h"
/* Lowest number of worker threads should be kept. */
#define SIGEV_WORKER_LOW 0
/* Highest number of worker threads can be created. */
#define SIGEV_WORKER_HIGH 20
/* How long an idle worker thread should stay. */
#define SIGEV_WORKER_IDLE 10
struct sigev_worker {
LIST_ENTRY(sigev_worker) sw_link;
pthread_cond_t sw_cv;
struct sigev_node *sw_sn;
int sw_flags;
int *sw_readyptr;
};
#define SWF_READYQ 1
LIST_HEAD(sigev_list_head, sigev_node);
#define HASH_QUEUES 17
#define HASH(t, id) ((((id) << 3) + (t)) % HASH_QUEUES)
static struct sigev_list_head sigev_hash[HASH_QUEUES];
static struct sigev_list_head sigev_all;
static TAILQ_HEAD(, sigev_node) sigev_actq;
static TAILQ_HEAD(, sigev_thread_node) sigev_threads;
static LIST_HEAD(,sigev_thread) sigev_threads;
static int sigev_generation;
static pthread_mutex_t *sigev_list_mtx;
static pthread_once_t sigev_once = PTHREAD_ONCE_INIT;
static pthread_once_t sigev_once_default = PTHREAD_ONCE_INIT;
static pthread_mutex_t *sigev_threads_mtx;
static pthread_cond_t *sigev_threads_cv;
static pthread_cond_t *sigev_actq_cv;
static struct sigev_thread_node *sigev_default_thread;
static struct sigev_thread_attr sigev_default_sna;
static struct sigev_thread *sigev_default_thread;
static pthread_attr_t sigev_default_attr;
static int atfork_registered;
static LIST_HEAD(,sigev_worker) sigev_worker_ready;
static int sigev_worker_count;
static int sigev_worker_start;
static int sigev_worker_high;
static int sigev_worker_low;
static pthread_cond_t *sigev_worker_init_cv;
static void __sigev_fork_prepare(void);
static void __sigev_fork_parent(void);
static void __sigev_fork_child(void);
static struct sigev_thread_node *sigev_thread_create(pthread_attr_t *,
struct sigev_thread_node *, int);
static void *sigev_service_loop(void *);
static void *sigev_worker_routine(void *);
static void sigev_put(struct sigev_node *);
static void worker_cleanup(void *arg);
static void __sigev_fork_prepare(void);
static void __sigev_fork_parent(void);
static void __sigev_fork_child(void);
static struct sigev_thread *sigev_thread_create(int);
static void *sigev_service_loop(void *);
static void *worker_routine(void *);
static void worker_cleanup(void *);
#pragma weak pthread_create
#pragma weak pthread_attr_getschedpolicy
#pragma weak pthread_attr_getinheritsched
#pragma weak pthread_attr_getschedparam
#pragma weak pthread_attr_getscope
#pragma weak pthread_attr_getstacksize
#pragma weak pthread_attr_getstackaddr
#pragma weak pthread_attr_getguardsize
#pragma weak pthread_attr_init
#pragma weak pthread_attr_setscope
#pragma weak pthread_attr_setdetachstate
#pragma weak pthread_atfork
#pragma weak _pthread_once
#pragma weak pthread_cleanup_push
#pragma weak pthread_cleanup_pop
#pragma weak pthread_setcancelstate
static __inline void
attr2sna(pthread_attr_t *attr, struct sigev_thread_attr *sna)
static void
attrcopy(pthread_attr_t *src, pthread_attr_t *dst)
{
struct sched_param sched_param;
struct sched_param sched;
void *a;
size_t u;
int v;
pthread_attr_getschedpolicy(attr, &sna->sna_policy);
pthread_attr_getinheritsched(attr, &sna->sna_inherit);
pthread_attr_getschedparam(attr, &sched_param);
sna->sna_prio = sched_param.sched_priority;
pthread_attr_getstacksize(attr, &sna->sna_stacksize);
pthread_attr_getstackaddr(attr, &sna->sna_stackaddr);
pthread_attr_getguardsize(attr, &sna->sna_guardsize);
}
_pthread_attr_getschedpolicy(src, &v);
_pthread_attr_setschedpolicy(dst, v);
static __inline int
sna_eq(const struct sigev_thread_attr *a, const struct sigev_thread_attr *b)
{
return memcmp(a, b, sizeof(*a)) == 0;
_pthread_attr_getinheritsched(src, &v);
_pthread_attr_setinheritsched(dst, v);
_pthread_attr_getschedparam(src, &sched);
_pthread_attr_setschedparam(dst, &sched);
_pthread_attr_getscope(src, &v);
_pthread_attr_setscope(dst, v);
_pthread_attr_getstacksize(src, &u);
_pthread_attr_setstacksize(dst, u);
_pthread_attr_getstackaddr(src, &a);
_pthread_attr_setstackaddr(src, a);
_pthread_attr_getguardsize(src, &u);
_pthread_attr_setguardsize(dst, u);
}
static __inline int
@ -148,46 +109,36 @@ void
__sigev_thread_init(void)
{
static int inited = 0;
pthread_mutexattr_t mattr;
int i;
_pthread_mutexattr_init(&mattr);
_pthread_mutexattr_settype(&mattr, PTHREAD_MUTEX_NORMAL);
sigev_list_mtx = malloc(sizeof(pthread_mutex_t));
_pthread_mutex_init(sigev_list_mtx, NULL);
sigev_threads_mtx = malloc(sizeof(pthread_mutex_t));
_pthread_mutex_init(sigev_threads_mtx, NULL);
sigev_threads_cv = malloc(sizeof(pthread_cond_t));
_pthread_cond_init(sigev_threads_cv, NULL);
sigev_actq_cv = malloc(sizeof(pthread_cond_t));
_pthread_cond_init(sigev_actq_cv, NULL);
_pthread_mutex_init(sigev_list_mtx, &mattr);
_pthread_mutexattr_destroy(&mattr);
for (i = 0; i < HASH_QUEUES; ++i)
LIST_INIT(&sigev_hash[i]);
LIST_INIT(&sigev_all);
TAILQ_INIT(&sigev_threads);
TAILQ_INIT(&sigev_actq);
LIST_INIT(&sigev_threads);
sigev_default_thread = NULL;
sigev_worker_count = 0;
sigev_worker_start = 0;
LIST_INIT(&sigev_worker_ready);
sigev_worker_high = SIGEV_WORKER_HIGH;
sigev_worker_low = SIGEV_WORKER_LOW;
sigev_worker_init_cv = malloc(sizeof(pthread_cond_t));
_pthread_cond_init(sigev_worker_init_cv, NULL);
if (atfork_registered == 0) {
pthread_atfork(
_pthread_atfork(
__sigev_fork_prepare,
__sigev_fork_parent,
__sigev_fork_child);
atfork_registered = 1;
}
if (!inited) {
pthread_attr_init(&sigev_default_attr);
attr2sna(&sigev_default_attr, &sigev_default_sna);
pthread_attr_setscope(&sigev_default_attr,
_pthread_attr_init(&sigev_default_attr);
_pthread_attr_setscope(&sigev_default_attr,
PTHREAD_SCOPE_SYSTEM);
pthread_attr_setdetachstate(&sigev_default_attr,
_pthread_attr_setdetachstate(&sigev_default_attr,
PTHREAD_CREATE_DETACHED);
inited = 1;
}
sigev_default_thread = sigev_thread_create(NULL, NULL, 0);
sigev_default_thread = sigev_thread_create(0);
}
int
@ -223,49 +174,46 @@ __sigev_fork_child(void)
__sigev_thread_init();
}
int
void
__sigev_list_lock(void)
{
return _pthread_mutex_lock(sigev_list_mtx);
_pthread_mutex_lock(sigev_list_mtx);
}
int
void
__sigev_list_unlock(void)
{
return _pthread_mutex_unlock(sigev_list_mtx);
}
int
__sigev_thread_list_lock(void)
{
return _pthread_mutex_lock(sigev_threads_mtx);
}
int
__sigev_thread_list_unlock(void)
{
return _pthread_mutex_unlock(sigev_threads_mtx);
_pthread_mutex_unlock(sigev_list_mtx);
}
struct sigev_node *
__sigev_alloc(int type, const struct sigevent *evp, struct sigev_node *prev,
int usethreadpool)
int usedefault)
{
struct sigev_node *sn;
sn = calloc(1, sizeof(*sn));
if (sn != NULL) {
sn->sn_value = evp->sigev_value;
sn->sn_func = evp->sigev_notify_function;
sn->sn_gen = atomic_fetchadd_int(&sigev_generation, 1);
sn->sn_type = type;
if (usethreadpool)
sn->sn_flags |= SNF_THREADPOOL;
sn->sn_tn = sigev_thread_create(evp->sigev_notify_attributes,
prev ? prev->sn_tn : NULL, usethreadpool);
if (sn->sn_tn == NULL) {
free(sn);
sn = NULL;
sn->sn_func = evp->sigev_notify_function;
sn->sn_gen = atomic_fetchadd_int(&sigev_generation, 1);
sn->sn_type = type;
_pthread_attr_init(&sn->sn_attr);
_pthread_attr_setdetachstate(&sn->sn_attr, PTHREAD_CREATE_DETACHED);
if (evp->sigev_notify_attributes)
attrcopy(evp->sigev_notify_attributes, &sn->sn_attr);
if (prev) {
__sigev_list_lock();
prev->sn_tn->tn_refcount++;
__sigev_list_unlock();
sn->sn_tn = prev->sn_tn;
} else {
sn->sn_tn = sigev_thread_create(usedefault);
if (sn->sn_tn == NULL) {
_pthread_attr_destroy(&sn->sn_attr);
free(sn);
sn = NULL;
}
}
}
return (sn);
@ -276,11 +224,11 @@ __sigev_get_sigevent(struct sigev_node *sn, struct sigevent *newevp,
sigev_id_t id)
{
/*
* Build a new sigevent, and tell kernel to deliver SIGEV_SIGSERVICE
* Build a new sigevent, and tell kernel to deliver SIGSERVICE
* signal to the new thread.
*/
newevp->sigev_notify = SIGEV_THREAD_ID;
newevp->sigev_signo = SIGEV_SIGSERVICE;
newevp->sigev_signo = SIGSERVICE;
newevp->sigev_notify_thread_id = (lwpid_t)sn->sn_tn->tn_lwpid;
newevp->sigev_value.sival_ptr = (void *)id;
}
@ -288,6 +236,7 @@ __sigev_get_sigevent(struct sigev_node *sn, struct sigevent *newevp,
void
__sigev_free(struct sigev_node *sn)
{
_pthread_attr_destroy(&sn->sn_attr);
free(sn);
}
@ -310,7 +259,6 @@ __sigev_register(struct sigev_node *sn)
int chain = HASH(sn->sn_type, sn->sn_id);
LIST_INSERT_HEAD(&sigev_hash[chain], sn, sn_link);
LIST_INSERT_HEAD(&sigev_all, sn, sn_allist);
return (0);
}
@ -329,26 +277,17 @@ int
__sigev_delete_node(struct sigev_node *sn)
{
LIST_REMOVE(sn, sn_link);
LIST_REMOVE(sn, sn_allist);
__sigev_thread_list_lock();
if (--sn->sn_tn->tn_refcount == 0)
if (!(sn->sn_flags & SNF_THREADPOOL))
pthread_kill(sn->sn_tn->tn_thread, SIGEV_SIGSERVICE);
__sigev_thread_list_unlock();
_pthread_kill(sn->sn_tn->tn_thread, SIGSERVICE);
if (sn->sn_flags & SNF_WORKING)
sn->sn_flags |= SNF_REMOVED;
else {
if (sn->sn_flags & SNF_ACTQ) {
TAILQ_REMOVE(&sigev_actq, sn, sn_actq);
}
else
__sigev_free(sn);
}
return (0);
}
static
sigev_id_t
static sigev_id_t
sigev_get_id(siginfo_t *si)
{
switch(si->si_code) {
@ -362,324 +301,166 @@ sigev_get_id(siginfo_t *si)
return (-1);
}
static struct sigev_thread_node *
sigev_thread_create(pthread_attr_t *pattr, struct sigev_thread_node *prev,
int usepool)
static struct sigev_thread *
sigev_thread_create(int usedefault)
{
struct sigev_thread_node *tn;
struct sigev_thread_attr sna;
struct sigev_thread *tn;
sigset_t set;
int ret;
if (pattr == NULL)
pattr = &sigev_default_attr;
else {
pthread_attr_setscope(pattr, PTHREAD_SCOPE_SYSTEM);
pthread_attr_setdetachstate(pattr, PTHREAD_CREATE_DETACHED);
}
attr2sna(pattr, &sna);
__sigev_thread_list_lock();
if (prev != NULL && sna_eq(&prev->tn_sna, &sna)) {
prev->tn_refcount++;
__sigev_thread_list_unlock();
return (prev);
}
if (sna_eq(&sna, &sigev_default_sna) && usepool &&
sigev_default_thread != NULL) {
if (usedefault && sigev_default_thread) {
__sigev_list_lock();
sigev_default_thread->tn_refcount++;
__sigev_thread_list_unlock();
return (sigev_default_thread);
__sigev_list_unlock();
return (sigev_default_thread);
}
tn = NULL;
/* Search a thread matching the required stack address */
if (sna.sna_stackaddr != NULL) {
TAILQ_FOREACH(tn, &sigev_threads, tn_link) {
if (sna.sna_stackaddr == tn->tn_sna.sna_stackaddr)
break;
}
}
if (tn != NULL) {
tn->tn_refcount++;
__sigev_thread_list_unlock();
return (tn);
}
tn = malloc(sizeof(*tn));
tn->tn_sna = sna;
tn->tn_cur = NULL;
tn->tn_lwpid = -1;
tn->tn_refcount = 1;
TAILQ_INSERT_TAIL(&sigev_threads, tn, tn_link);
_pthread_cond_init(&tn->tn_cv, NULL);
/* for debug */
__sigev_list_lock();
LIST_INSERT_HEAD(&sigev_threads, tn, tn_link);
__sigev_list_unlock();
sigemptyset(&set);
sigaddset(&set, SIGEV_SIGSERVICE);
sigaddset(&set, SIGSERVICE);
_sigprocmask(SIG_BLOCK, &set, NULL);
ret = pthread_create(&tn->tn_thread, pattr, sigev_service_loop, tn);
ret = pthread_create(&tn->tn_thread, &sigev_default_attr,
sigev_service_loop, tn);
_sigprocmask(SIG_UNBLOCK, &set, NULL);
if (ret != 0) {
TAILQ_REMOVE(&sigev_threads, tn, tn_link);
__sigev_thread_list_unlock();
__sigev_list_lock();
LIST_REMOVE(tn, tn_link);
__sigev_list_unlock();
free(tn);
tn = NULL;
} else {
/* wait the thread to get its lwpid */
__sigev_list_lock();
while (tn->tn_lwpid == -1)
_pthread_cond_wait(sigev_threads_cv, sigev_threads_mtx);
__sigev_thread_list_unlock();
_pthread_cond_wait(&tn->tn_cv, sigev_list_mtx);
__sigev_list_unlock();
}
return (tn);
}
static void
after_dispatch(struct sigev_thread_node *tn)
{
struct sigev_node *sn;
if ((sn = tn->tn_cur) != NULL) {
__sigev_list_lock();
sn->sn_flags &= ~SNF_WORKING;
if (sn->sn_flags & SNF_REMOVED)
__sigev_free(sn);
else if (sn->sn_flags & SNF_ONESHOT)
__sigev_delete_node(sn);
tn->tn_cur = NULL;
__sigev_list_unlock();
}
tn->tn_cur = NULL;
}
/*
* This function is called if user callback calls
* pthread_exit() or pthread_cancel() for the thread.
*/
static void
thread_cleanup(void *arg)
{
struct sigev_thread_node *tn = arg;
fprintf(stderr, "Dangerous Robinson, calling pthread_exit() from "
"SIGEV_THREAD is undefined.");
after_dispatch(tn);
/* longjmp(tn->tn_jbuf, 1); */
abort();
}
/*
* Main notification dispatch function, the function either
* run user callback by itself or hand off the notifications
* to worker threads depend on flags.
* The thread receives notification from kernel and creates
* a thread to call user callback function.
*/
static void *
sigev_service_loop(void *arg)
{
static int failure;
siginfo_t si;
sigset_t set;
struct sigev_thread_node *tn;
struct sigev_thread *tn;
struct sigev_node *sn;
sigev_id_t id;
pthread_t td;
int ret;
tn = arg;
thr_self(&tn->tn_lwpid);
__sigev_thread_list_lock();
_pthread_cond_broadcast(sigev_threads_cv);
__sigev_thread_list_unlock();
__sigev_list_lock();
_pthread_cond_broadcast(&tn->tn_cv);
__sigev_list_unlock();
/*
* Service thread should not be killed by callback, if user
* attempts to do so, the thread will be restarted.
*/
setjmp(tn->tn_jbuf);
pthread_setcancelstate(PTHREAD_CANCEL_DISABLE, NULL);
sigemptyset(&set);
sigaddset(&set, SIGEV_SIGSERVICE);
pthread_cleanup_push(thread_cleanup, tn);
sigaddset(&set, SIGSERVICE);
for (;;) {
ret = sigwaitinfo(&set, &si);
__sigev_thread_list_lock();
__sigev_list_lock();
if (tn->tn_refcount == 0) {
TAILQ_REMOVE(&sigev_threads, tn, tn_link);
__sigev_thread_list_unlock();
LIST_REMOVE(tn, tn_link);
__sigev_list_unlock();
free(tn);
break;
}
__sigev_thread_list_unlock();
if (ret == -1)
if (ret == -1) {
__sigev_list_unlock();
continue;
}
id = sigev_get_id(&si);
__sigev_list_lock();
sn = __sigev_find(si.si_code, id);
if (sn != NULL) {
sn->sn_info = si;
if (!(sn->sn_flags & SNF_THREADPOOL)) {
tn->tn_cur = sn;
sn->sn_flags |= SNF_WORKING;
__sigev_list_unlock();
sn->sn_dispatch(sn);
after_dispatch(tn);
} else {
assert(!(sn->sn_flags & SNF_ACTQ));
sigev_put(sn);
}
} else {
if (sn == NULL) {
__sigev_list_unlock();
continue;
}
sn->sn_info = si;
if (sn->sn_flags & SNF_SYNC)
tn->tn_cur = sn;
else
tn->tn_cur = NULL;
__sigev_list_unlock();
}
}
pthread_cleanup_pop(0);
return (0);
}
/*
* Hand off notifications to worker threads.
*
* prerequist: sigev list locked.
*/
static void
sigev_put(struct sigev_node *sn)
{
struct sigev_worker *worker;
pthread_t td;
int ret, ready;
TAILQ_INSERT_TAIL(&sigev_actq, sn, sn_actq);
sn->sn_flags |= SNF_ACTQ;
/*
* check if we should add more worker threads unless quota is hit.
*/
if (LIST_EMPTY(&sigev_worker_ready) &&
sigev_worker_count + sigev_worker_start < sigev_worker_high) {
sigev_worker_start++;
sn->sn_flags |= SNF_WORKING;
__sigev_list_unlock();
worker = malloc(sizeof(*worker));
_pthread_cond_init(&worker->sw_cv, NULL);
worker->sw_flags = 0;
worker->sw_sn = 0;
worker->sw_readyptr = &ready;
ready = 0;
ret = pthread_create(&td, &sigev_default_attr,
sigev_worker_routine, worker);
if (ret) {
warnc(ret, "%s:%s can not create worker thread",
__FILE__, __func__);
__sigev_list_lock();
sigev_worker_start--;
__sigev_list_unlock();
} else {
__sigev_list_lock();
while (ready == 0) {
_pthread_cond_wait(sigev_worker_init_cv,
sigev_list_mtx);
}
__sigev_list_unlock();
}
} else {
worker = LIST_FIRST(&sigev_worker_ready);
if (worker) {
LIST_REMOVE(worker, sw_link);
worker->sw_flags &= ~SWF_READYQ;
_pthread_cond_broadcast(&worker->sw_cv);
__sigev_list_unlock();
}
}
}
/*
* Background thread to dispatch notification to user code.
* These threads are not bound to any realtime objects.
*/
static void *
sigev_worker_routine(void *arg)
{
struct sigev_worker *worker;
struct sigev_node *sn;
struct timespec ts;
int ret;
worker = arg;
__sigev_list_lock();
sigev_worker_count++;
sigev_worker_start--;
(*(worker->sw_readyptr))++;
LIST_INSERT_HEAD(&sigev_worker_ready, worker, sw_link);
worker->sw_flags |= SWF_READYQ;
_pthread_cond_broadcast(sigev_worker_init_cv);
for (;;) {
if (worker->sw_flags & SWF_READYQ) {
LIST_REMOVE(worker, sw_link);
worker->sw_flags &= ~SWF_READYQ;
}
sn = TAILQ_FIRST(&sigev_actq);
if (sn != NULL) {
TAILQ_REMOVE(&sigev_actq, sn, sn_actq);
sn->sn_flags &= ~SNF_ACTQ;
sn->sn_flags |= SNF_WORKING;
__sigev_list_unlock();
worker->sw_sn = sn;
pthread_cleanup_push(worker_cleanup, worker);
sn->sn_dispatch(sn);
pthread_cleanup_pop(0);
worker->sw_sn = NULL;
ret = pthread_create(&td, &sn->sn_attr, worker_routine, sn);
if (ret != 0) {
if (failure++ < 5)
warnc(ret, "%s:%s failed to create thread.\n",
__FILE__, __func__);
__sigev_list_lock();
sn->sn_flags &= ~SNF_WORKING;
if (sn->sn_flags & SNF_REMOVED)
__sigev_free(sn);
else if (sn->sn_flags & SNF_ONESHOT)
__sigev_delete_node(sn);
} else {
LIST_INSERT_HEAD(&sigev_worker_ready, worker, sw_link);
worker->sw_flags |= SWF_READYQ;
clock_gettime(CLOCK_REALTIME, &ts);
ts.tv_sec += SIGEV_WORKER_IDLE;
ret = _pthread_cond_timedwait(&worker->sw_cv,
sigev_list_mtx, &ts);
if (ret == ETIMEDOUT) {
/*
* If we were timeouted and there is nothing
* to do, exit the thread.
*/
if (TAILQ_EMPTY(&sigev_actq) &&
(worker->sw_flags & SWF_READYQ) &&
sigev_worker_count > sigev_worker_low)
goto out;
}
__sigev_list_unlock();
} else if (tn->tn_cur) {
__sigev_list_lock();
while (tn->tn_cur)
_pthread_cond_wait(&tn->tn_cv, sigev_list_mtx);
__sigev_list_unlock();
}
}
out:
if (worker->sw_flags & SWF_READYQ) {
LIST_REMOVE(worker, sw_link);
worker->sw_flags &= ~SWF_READYQ;
}
sigev_worker_count--;
__sigev_list_unlock();
_pthread_cond_destroy(&worker->sw_cv);
free(worker);
return (0);
}
/*
* newly created worker thread to call user callback function.
*/
static void *
worker_routine(void *arg)
{
struct sigev_node *sn = arg;
_pthread_cleanup_push(worker_cleanup, sn);
sn->sn_dispatch(sn);
_pthread_cleanup_pop(1);
return (0);
}
/* clean up a notification after dispatch. */
static void
worker_cleanup(void *arg)
{
struct sigev_worker *worker = arg;
struct sigev_node *sn;
struct sigev_node *sn = arg;
sn = worker->sw_sn;
__sigev_list_lock();
sn->sn_flags &= ~SNF_WORKING;
if (sn->sn_flags & SNF_SYNC) {
sn->sn_tn->tn_cur = NULL;
_pthread_cond_broadcast(&sn->sn_tn->tn_cv);
}
if (sn->sn_flags & SNF_REMOVED)
__sigev_free(sn);
else if (sn->sn_flags & SNF_ONESHOT)
__sigev_delete_node(sn);
sigev_worker_count--;
else
sn->sn_flags &= ~SNF_WORKING;
__sigev_list_unlock();
_pthread_cond_destroy(&worker->sw_cv);
free(worker);
}

38
lib/librt/sigev_thread.h
View File

@ -32,9 +32,8 @@
#include <sys/types.h>
#include <sys/queue.h>
#include <setjmp.h>
struct sigev_thread_node;
struct sigev_thread;
struct sigev_node;
typedef uintptr_t sigev_id_t;
@ -42,8 +41,6 @@ typedef void (*sigev_dispatch_t)(struct sigev_node *);
struct sigev_node {
LIST_ENTRY(sigev_node) sn_link;
LIST_ENTRY(sigev_node) sn_allist;
TAILQ_ENTRY(sigev_node) sn_actq;
int sn_type;
sigev_id_t sn_id;
sigev_dispatch_t sn_dispatch;
@ -51,51 +48,38 @@ struct sigev_node {
void *sn_func;
int sn_flags;
int sn_gen;
int sn_usethreadpool;
siginfo_t sn_info;
struct sigev_thread_node * sn_tn;
pthread_attr_t sn_attr;
struct sigev_thread *sn_tn;
};
struct sigev_thread_attr {
int sna_policy;
int sna_inherit;
int sna_prio;
size_t sna_stacksize;
void *sna_stackaddr;
size_t sna_guardsize;
};
struct sigev_thread_node {
TAILQ_ENTRY(sigev_thread_node) tn_link;
struct sigev_thread {
LIST_ENTRY(sigev_thread) tn_link;
pthread_t tn_thread;
struct sigev_node *tn_cur;
struct sigev_thread_attr tn_sna;
int tn_refcount;
long tn_lwpid;
jmp_buf tn_jbuf;
pthread_cond_t tn_cv;
};
#define SNF_WORKING 0x01
#define SNF_REMOVED 0x02
#define SNF_ONESHOT 0x04
#define SNF_ACTQ 0x08
#define SNF_THREADPOOL 0x10
#define SNF_SYNC 0x04
#define SIGEV_SIGSERVICE (SIGTHR+1)
#define SIGSERVICE (SIGTHR+1)
int __sigev_check_init();
struct sigev_node *__sigev_alloc(int, const struct sigevent *,
struct sigev_node *, int usethreadpool);
struct sigev_node *, int);
struct sigev_node *__sigev_find(int, sigev_id_t);
void __sigev_get_sigevent(struct sigev_node *, struct sigevent *,
sigev_id_t);
int __sigev_register(struct sigev_node *);
int __sigev_delete(int, sigev_id_t);
int __sigev_delete_node(struct sigev_node *);
int __sigev_list_lock(void);
int __sigev_list_unlock(void);
int __sigev_thread_list_lock(void);
int __sigev_thread_list_unlock(void);
void __sigev_list_lock(void);
void __sigev_list_unlock(void);
void __sigev_free(struct sigev_node *);
#endif

2
lib/librt/timer.c
View File

@ -33,12 +33,10 @@
#include "namespace.h"
#include <errno.h>
#include <pthread.h>
#include <stddef.h>
#include <signal.h>
#include <stdlib.h>
#include <time.h>
#include <unistd.h>
#include "sigev_thread.h"
#include "un-namespace.h"