thread.c
Go to the documentation of this file.
#define _GNU_SOURCE
#include <stdio.h>
#include <stdlib.h>
#include <stdint.h>
#include <limits.h>
#ifndef WIN32
#include <sys/queue.h>
#endif
#include <sys/ioctl.h>
#include <unistd.h>
#include <pthread.h>
#include <event2/event.h>
#include <event2/thread.h>
#include "internal.h"
#include "evhtp/thread.h"
typedef struct evthr_cmd evthr_cmd_t;
typedef struct evthr_pool_slist evthr_pool_slist_t;
struct evthr_cmd {
uint8_t stop;
void * args;
evthr_cb cb;
} __attribute__((packed));
TAILQ_HEAD(evthr_pool_slist, evthr);
struct evthr_pool {
#ifdef EVTHR_SHARED_PIPE
int rdr;
int wdr;
#endif
int nthreads;
evthr_pool_slist_t threads;
};
struct evthr {
int rdr;
int wdr;
char err;
ev_t * event;
evbase_t * evbase;
pthread_mutex_t lock;
pthread_t * thr;
evthr_init_cb init_cb;
evthr_exit_cb exit_cb;
void * arg;
void * aux;
#ifdef EVTHR_SHARED_PIPE
int pool_rdr;
struct event * shared_pool_ev;
#endif
TAILQ_ENTRY(evthr) next;
};
#define _evthr_read(thr, cmd, sock) \
(recv(sock, cmd, sizeof(evthr_cmd_t), 0) == sizeof(evthr_cmd_t)) ? 1 : 0
static void
_evthr_read_cmd(evutil_socket_t sock, short which, void * args)
{
evthr_t * thread;
evthr_cmd_t cmd;
int stopped;
if (!(thread = (evthr_t *)args)) {
return;
}
stopped = 0;
if (evhtp_likely(_evthr_read(thread, &cmd, sock) == 1)) {
stopped = cmd.stop;
if (evhtp_likely(cmd.cb != NULL)) {
(cmd.cb)(thread, cmd.args, thread->arg);
}
}
if (evhtp_unlikely(stopped == 1)) {
event_base_loopbreak(thread->evbase);
}
return;
} /* _evthr_read_cmd */
static void *
_evthr_loop(void * args)
{
evthr_t * thread;
if (!(thread = (evthr_t *)args)) {
return NULL;
}
if (thread == NULL || thread->thr == NULL) {
pthread_exit(NULL);
}
thread->evbase = event_base_new();
thread->event = event_new(thread->evbase, thread->rdr,
EV_READ | EV_PERSIST, _evthr_read_cmd, args);
event_add(thread->event, NULL);
#ifdef EVTHR_SHARED_PIPE
if (thread->pool_rdr > 0) {
thread->shared_pool_ev = event_new(thread->evbase, thread->pool_rdr,
EV_READ | EV_PERSIST, _evthr_read_cmd, args);
event_add(thread->shared_pool_ev, NULL);
}
#endif
pthread_mutex_lock(&thread->lock);
if (thread->init_cb != NULL) {
(thread->init_cb)(thread, thread->arg);
}
pthread_mutex_unlock(&thread->lock);
event_base_loop(thread->evbase, 0);
pthread_mutex_lock(&thread->lock);
if (thread->exit_cb != NULL) {
(thread->exit_cb)(thread, thread->arg);
}
pthread_mutex_unlock(&thread->lock);
if (thread->err == 1) {
fprintf(stderr, "FATAL ERROR!\n");
}
pthread_exit(NULL);
} /* _evthr_loop */
evthr_res
evthr_defer(evthr_t * thread, evthr_cb cb, void * arg)
{
evthr_cmd_t cmd = {
.cb = cb,
.args = arg,
.stop = 0
};
if (send(thread->wdr, &cmd, sizeof(cmd), 0) <= 0) {
return EVTHR_RES_RETRY;
}
return EVTHR_RES_OK;
}
evthr_res
evthr_stop(evthr_t * thread)
{
evthr_cmd_t cmd = {
.cb = NULL,
.args = NULL,
.stop = 1
};
if (send(thread->wdr, &cmd, sizeof(evthr_cmd_t), 0) < 0) {
return EVTHR_RES_RETRY;
}
pthread_join(*thread->thr, NULL);
return EVTHR_RES_OK;
}
evbase_t *
evthr_get_base(evthr_t * thr)
{
return thr ? thr->evbase : NULL;
}
void
evthr_set_aux(evthr_t * thr, void * aux)
{
if (thr) {
thr->aux = aux;
}
}
void *
evthr_get_aux(evthr_t * thr)
{
return thr ? thr->aux : NULL;
}
int
evthr_set_initcb(evthr_t * thr, evthr_init_cb cb)
{
if (thr == NULL) {
return -1;
}
thr->init_cb = cb;
return 0;
}
int
evthr_set_exitcb(evthr_t * thr, evthr_exit_cb cb)
{
if (thr == NULL) {
return -1;
}
thr->exit_cb = cb;
return 0;
}
static evthr_t *
_evthr_new(evthr_init_cb init_cb, evthr_exit_cb exit_cb, void * args)
{
evthr_t * thread;
int fds[2];
if (evutil_socketpair(AF_UNIX, SOCK_STREAM, 0, fds) == -1) {
return NULL;
}
evutil_make_socket_nonblocking(fds[0]);
evutil_make_socket_nonblocking(fds[1]);
if (!(thread = calloc(sizeof(evthr_t), 1))) {
return NULL;
}
thread->thr = malloc(sizeof(pthread_t));
thread->arg = args;
thread->rdr = fds[0];
thread->wdr = fds[1];
thread->init_cb = init_cb;
thread->exit_cb = exit_cb;
if (pthread_mutex_init(&thread->lock, NULL)) {
evthr_free(thread);
return NULL;
}
return thread;
} /* evthr_new */
evthr_t *
evthr_new(evthr_init_cb init_cb, void * args)
{
return _evthr_new(init_cb, NULL, args);
}
evthr_t *
evthr_wexit_new(evthr_init_cb init_cb, evthr_exit_cb exit_cb, void * args)
{
return _evthr_new(init_cb, exit_cb, args);
}
int
evthr_start(evthr_t * thread)
{
if (thread == NULL || thread->thr == NULL) {
return -1;
}
if (pthread_create(thread->thr, NULL, _evthr_loop, (void *)thread)) {
return -1;
}
return 0;
}
void
evthr_free(evthr_t * thread)
{
if (thread == NULL) {
return;
}
if (thread->rdr > 0) {
close(thread->rdr);
}
if (thread->wdr > 0) {
close(thread->wdr);
}
if (thread->thr) {
free(thread->thr);
}
if (thread->event) {
event_free(thread->event);
}
#ifdef EVTHR_SHARED_PIPE
if (thread->shared_pool_ev) {
event_free(thread->shared_pool_ev);
}
#endif
if (thread->evbase) {
event_base_free(thread->evbase);
}
free(thread);
} /* evthr_free */
void
evthr_pool_free(evthr_pool_t * pool)
{
evthr_t * thread;
evthr_t * save;
if (pool == NULL) {
return;
}
TAILQ_FOREACH_SAFE(thread, &pool->threads, next, save) {
TAILQ_REMOVE(&pool->threads, thread, next);
evthr_free(thread);
}
free(pool);
}
evthr_res
evthr_pool_stop(evthr_pool_t * pool)
{
evthr_t * thr;
evthr_t * save;
if (pool == NULL) {
return EVTHR_RES_FATAL;
}
TAILQ_FOREACH_SAFE(thr, &pool->threads, next, save) {
evthr_stop(thr);
}
return EVTHR_RES_OK;
}
static inline int
get_backlog_(evthr_t * thread)
{
int backlog = 0;
ioctl(thread->rdr, FIONREAD, &backlog);
return (int)(backlog / sizeof(evthr_cmd_t));
}
evthr_res
evthr_pool_defer(evthr_pool_t * pool, evthr_cb cb, void * arg)
{
#ifdef EVTHR_SHARED_PIPE
evthr_cmd_t cmd = {
.cb = cb,
.args = arg,
.stop = 0
};
if (evhtp_unlikely(send(pool->wdr, &cmd, sizeof(cmd), 0) == -1)) {
return EVTHR_RES_RETRY;
}
return EVTHR_RES_OK;
#endif
evthr_t * thread = NULL;
evthr_t * min_thread = NULL;
int min_backlog = 0;
if (pool == NULL) {
return EVTHR_RES_FATAL;
}
if (cb == NULL) {
return EVTHR_RES_NOCB;
}
TAILQ_FOREACH(thread, &pool->threads, next) {
int backlog = get_backlog_(thread);
if (backlog == 0) {
min_thread = thread;
break;
}
if (min_thread == NULL || backlog < min_backlog) {
min_thread = thread;
min_backlog = backlog;
}
}
return evthr_defer(min_thread, cb, arg);
} /* evthr_pool_defer */
static evthr_pool_t *
_evthr_pool_new(int nthreads,
evthr_init_cb init_cb,
evthr_exit_cb exit_cb,
void * shared)
{
evthr_pool_t * pool;
int i;
#ifdef EVTHR_SHARED_PIPE
int fds[2];
#endif
if (nthreads == 0) {
return NULL;
}
if (!(pool = calloc(sizeof(evthr_pool_t), 1))) {
return NULL;
}
pool->nthreads = nthreads;
TAILQ_INIT(&pool->threads);
#ifdef EVTHR_SHARED_PIPE
if (evutil_socketpair(AF_UNIX, SOCK_DGRAM, 0, fds) == -1) {
return NULL;
}
evutil_make_socket_nonblocking(fds[0]);
evutil_make_socket_nonblocking(fds[1]);
pool->rdr = fds[0];
pool->wdr = fds[1];
#endif
for (i = 0; i < nthreads; i++) {
evthr_t * thread;
if (!(thread = evthr_wexit_new(init_cb, exit_cb, shared))) {
evthr_pool_free(pool);
return NULL;
}
#ifdef EVTHR_SHARED_PIPE
thread->pool_rdr = fds[0];
#endif
TAILQ_INSERT_TAIL(&pool->threads, thread, next);
}
return pool;
} /* _evthr_pool_new */
evthr_pool_t *
evthr_pool_new(int nthreads, evthr_init_cb init_cb, void * shared)
{
return _evthr_pool_new(nthreads, init_cb, NULL, shared);
}
evthr_pool_t *
evthr_pool_wexit_new(int nthreads,
evthr_init_cb init_cb,
evthr_exit_cb exit_cb, void * shared)
{
return _evthr_pool_new(nthreads, init_cb, exit_cb, shared);
}
int
evthr_pool_start(evthr_pool_t * pool)
{
evthr_t * evthr = NULL;
if (pool == NULL) {
return -1;
}
TAILQ_FOREACH(evthr, &pool->threads, next) {
if (evthr_start(evthr) < 0) {
return -1;
}
usleep(5000);
}
return 0;
}
internal.h
TAILQ_FOREACH_SAFE
#define TAILQ_FOREACH_SAFE(var, head, field, tvar)
Definition: internal.h:22
evhtp_unlikely
#define evhtp_unlikely(x)
Definition: internal.h:18
evhtp_likely
#define evhtp_likely(x)
Definition: internal.h:17
evthr_cmd
Definition: thread.c:23
evthr_cmd::args
void * args
Definition: thread.c:25
evthr_cmd::stop
uint8_t stop
Definition: thread.c:24
evthr_cmd::cb
evthr_cb cb
Definition: thread.c:26
evthr_pool
Definition: thread.c:31
evthr_pool::nthreads
int nthreads
Definition: thread.c:36
evthr_pool::threads
evthr_pool_slist_t threads
Definition: thread.c:37
evthr
Definition: thread.c:40
evthr::evbase
evbase_t * evbase
Definition: thread.c:45
evthr::aux
void * aux
Definition: thread.c:51
evthr::err
char err
Definition: thread.c:43
evthr::exit_cb
evthr_exit_cb exit_cb
Definition: thread.c:49
evthr::rdr
int rdr
Definition: thread.c:41
evthr::lock
pthread_mutex_t lock
Definition: thread.c:46
evthr::arg
void * arg
Definition: thread.c:50
evthr::wdr
int wdr
Definition: thread.c:42
evthr::thr
pthread_t * thr
Definition: thread.c:47
evthr::event
ev_t * event
Definition: thread.c:44
evthr::init_cb
evthr_init_cb init_cb
Definition: thread.c:48
evthr_start
int evthr_start(evthr_t *thread)
Definition: thread.c:265
_evthr_loop
static void * _evthr_loop(void *args)
Definition: thread.c:92
evthr_set_aux
void evthr_set_aux(evthr_t *thr, void *aux)
Definition: thread.c:182
evthr_get_aux
void * evthr_get_aux(evthr_t *thr)
Definition: thread.c:190
evthr_new
evthr_t * evthr_new(evthr_init_cb init_cb, void *args)
Definition: thread.c:253
TAILQ_HEAD
TAILQ_HEAD(evthr_pool_slist, evthr)
get_backlog_
static int get_backlog_(evthr_t *thread)
Definition: thread.c:352
_evthr_pool_new
static evthr_pool_t * _evthr_pool_new(int nthreads, evthr_init_cb init_cb, evthr_exit_cb exit_cb, void *shared)
Definition: thread.c:408
evthr_stop
evthr_res evthr_stop(evthr_t *thread)
Definition: thread.c:159
evthr_defer
evthr_res evthr_defer(evthr_t *thread, evthr_cb cb, void *arg)
Definition: thread.c:143
evthr_wexit_new
evthr_t * evthr_wexit_new(evthr_init_cb init_cb, evthr_exit_cb exit_cb, void *args)
Definition: thread.c:259
evthr_get_base
evbase_t * evthr_get_base(evthr_t *thr)
Definition: thread.c:176
evthr_pool_free
void evthr_pool_free(evthr_pool_t *pool)
Definition: thread.c:316
_evthr_read_cmd
static void _evthr_read_cmd(evutil_socket_t sock, short which, void *args)
Definition: thread.c:64
evthr_set_initcb
int evthr_set_initcb(evthr_t *thr, evthr_init_cb cb)
Definition: thread.c:196
evthr_pool_slist_t
struct evthr_pool_slist evthr_pool_slist_t
Definition: thread.c:21
evthr_free
void evthr_free(evthr_t *thread)
Definition: thread.c:279
__attribute__
struct evthr_pool __attribute__
cb
evthr_cb cb
Definition: thread.c:2
evthr_pool_new
evthr_pool_t * evthr_pool_new(int nthreads, evthr_init_cb init_cb, void *shared)
Definition: thread.c:462
evthr_pool_defer
evthr_res evthr_pool_defer(evthr_pool_t *pool, evthr_cb cb, void *arg)
Definition: thread.c:362
_evthr_read
#define _evthr_read(thr, cmd, sock)
Definition: thread.c:60
args
void * args
Definition: thread.c:1
_evthr_new
static evthr_t * _evthr_new(evthr_init_cb init_cb, evthr_exit_cb exit_cb, void *args)
Definition: thread.c:220
evthr_pool_wexit_new
evthr_pool_t * evthr_pool_wexit_new(int nthreads, evthr_init_cb init_cb, evthr_exit_cb exit_cb, void *shared)
Definition: thread.c:468
evthr_set_exitcb
int evthr_set_exitcb(evthr_t *thr, evthr_exit_cb cb)
Definition: thread.c:208
evthr_pool_start
int evthr_pool_start(evthr_pool_t *pool)
Definition: thread.c:476
evthr_pool_stop
evthr_res evthr_pool_stop(evthr_pool_t *pool)
Definition: thread.c:335