BadVPN – Rev 1
?pathlinks?
/**
* @file BReactor_glib.c
* @author Ambroz Bizjak <ambrop7@gmail.com>
*
* @section LICENSE
*
* Redistribution and use in source and binary forms, with or without
* modification, are permitted provided that the following conditions are met:
* 1. Redistributions of source code must retain the above copyright
* notice, this list of conditions and the following disclaimer.
* 2. Redistributions in binary form must reproduce the above copyright
* notice, this list of conditions and the following disclaimer in the
* documentation and/or other materials provided with the distribution.
* 3. Neither the name of the author nor the
* names of its contributors may be used to endorse or promote products
* derived from this software without specific prior written permission.
*
* THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND
* ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
* WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
* DISCLAIMED. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR ANY
* DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES
* (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
* LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND
* ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
* (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
* SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
*/
#include <stdlib.h>
#include <string.h>
#include <misc/offset.h>
#include <base/BLog.h>
#include "BReactor_glib.h"
#include <generated/blog_channel_BReactor.h>
struct fd_source {
GSource source;
BFileDescriptor *bfd;
};
static void assert_timer (BSmallTimer *bt)
{
ASSERT(bt->is_small == 0 || bt->is_small == 1)
ASSERT(bt->active == 0 || bt->active == 1)
ASSERT(!bt->active || bt->reactor)
ASSERT(!bt->active || bt->source)
}
static void dispatch_pending (BReactor *o)
{
while (!o->exiting && BPendingGroup_HasJobs(&o->pending_jobs)) {
BPendingGroup_ExecuteJob(&o->pending_jobs);
}
}
static void reset_limits (BReactor *o)
{
LinkedList1Node *list_node;
while (list_node = LinkedList1_GetFirst(&o->active_limits_list)) {
BReactorLimit *limit = UPPER_OBJECT(list_node, BReactorLimit, active_limits_list_node);
ASSERT(limit->count > 0)
limit->count = 0;
LinkedList1_Remove(&o->active_limits_list, &limit->active_limits_list_node);
}
}
static gushort get_glib_wait_events (int ev)
{
gushort gev = G_IO_ERR | G_IO_HUP;
if (ev & BREACTOR_READ) {
gev |= G_IO_IN;
}
if (ev & BREACTOR_WRITE) {
gev |= G_IO_OUT;
}
return gev;
}
static int get_fd_dispatchable_events (BFileDescriptor *bfd)
{
ASSERT(bfd->active)
int ev = 0;
if ((bfd->waitEvents & BREACTOR_READ) && (bfd->pollfd.revents & G_IO_IN)) {
ev |= BREACTOR_READ;
}
if ((bfd->waitEvents & BREACTOR_WRITE) && (bfd->pollfd.revents & G_IO_OUT)) {
ev |= BREACTOR_WRITE;
}
if ((bfd->pollfd.revents & G_IO_ERR)) {
ev |= BREACTOR_ERROR;
}
if ((bfd->pollfd.revents & G_IO_HUP)) {
ev |= BREACTOR_HUP;
}
return ev;
}
static gboolean timer_source_handler (gpointer data)
{
BSmallTimer *bt = (void *)data;
assert_timer(bt);
ASSERT(bt->active)
BReactor *reactor = bt->reactor;
if (reactor->exiting) {
return FALSE;
}
g_source_destroy(bt->source);
g_source_unref(bt->source);
bt->active = 0;
DebugCounter_Decrement(&reactor->d_timers_ctr);
if (bt->is_small) {
bt->handler.smalll(bt);
} else {
BTimer *btimer = UPPER_OBJECT(bt, BTimer, base);
bt->handler.heavy(btimer->user);
}
dispatch_pending(reactor);
reset_limits(reactor);
return FALSE;
}
static gboolean fd_source_func_prepare (GSource *source, gint *timeout)
{
BFileDescriptor *bfd = ((struct fd_source *)source)->bfd;
ASSERT(bfd->active)
ASSERT(bfd->source == source)
*timeout = -1;
return FALSE;
}
static gboolean fd_source_func_check (GSource *source)
{
BFileDescriptor *bfd = ((struct fd_source *)source)->bfd;
ASSERT(bfd->active)
ASSERT(bfd->source == source)
return (get_fd_dispatchable_events(bfd) ? TRUE : FALSE);
}
static gboolean fd_source_func_dispatch (GSource *source, GSourceFunc callback, gpointer user_data)
{
BFileDescriptor *bfd = ((struct fd_source *)source)->bfd;
BReactor *reactor = bfd->reactor;
ASSERT(bfd->active)
ASSERT(bfd->source == source)
if (reactor->exiting) {
return TRUE;
}
int events = get_fd_dispatchable_events(bfd);
if (!events) {
return TRUE;
}
bfd->handler(bfd->user, events);
dispatch_pending(reactor);
reset_limits(reactor);
return TRUE;
}
void BSmallTimer_Init (BSmallTimer *bt, BSmallTimer_handler handler)
{
bt->handler.smalll = handler;
bt->active = 0;
bt->is_small = 1;
}
int BSmallTimer_IsRunning (BSmallTimer *bt)
{
assert_timer(bt);
return bt->active;
}
void BTimer_Init (BTimer *bt, btime_t msTime, BTimer_handler handler, void *user)
{
bt->base.handler.heavy = handler;
bt->base.active = 0;
bt->base.is_small = 0;
bt->user = user;
bt->msTime = msTime;
}
int BTimer_IsRunning (BTimer *bt)
{
return BSmallTimer_IsRunning(&bt->base);
}
void BFileDescriptor_Init (BFileDescriptor *bs, int fd, BFileDescriptor_handler handler, void *user)
{
bs->fd = fd;
bs->handler = handler;
bs->user = user;
bs->active = 0;
}
int BReactor_Init (BReactor *bsys)
{
return BReactor_InitFromExistingGMainLoop(bsys, g_main_loop_new(NULL, FALSE), 1);
}
void BReactor_Free (BReactor *bsys)
{
DebugObject_Free(&bsys->d_obj);
DebugCounter_Free(&bsys->d_timers_ctr);
DebugCounter_Free(&bsys->d_limits_ctr);
DebugCounter_Free(&bsys->d_fds_counter);
ASSERT(!BPendingGroup_HasJobs(&bsys->pending_jobs))
ASSERT(LinkedList1_IsEmpty(&bsys->active_limits_list))
// free job queue
BPendingGroup_Free(&bsys->pending_jobs);
// unref main loop if needed
if (bsys->unref_gloop_on_free) {
g_main_loop_unref(bsys->gloop);
}
}
int BReactor_Exec (BReactor *bsys)
{
DebugObject_Access(&bsys->d_obj);
// dispatch pending jobs (until exiting) and reset limits
dispatch_pending(bsys);
reset_limits(bsys);
// if exiting, do not enter glib loop
if (bsys->exiting) {
return bsys->exit_code;
}
// enter glib loop
g_main_loop_run(bsys->gloop);
ASSERT(bsys->exiting)
return bsys->exit_code;
}
void BReactor_Quit (BReactor *bsys, int code)
{
DebugObject_Access(&bsys->d_obj);
// remember exiting
bsys->exiting = 1;
bsys->exit_code = code;
// request termination of glib loop
g_main_loop_quit(bsys->gloop);
}
void BReactor_SetSmallTimer (BReactor *bsys, BSmallTimer *bt, int mode, btime_t time)
{
DebugObject_Access(&bsys->d_obj);
assert_timer(bt);
// remove timer if it's already set
BReactor_RemoveSmallTimer(bsys, bt);
// if mode is absolute, subtract current time
if (mode == BTIMER_SET_ABSOLUTE) {
btime_t now = btime_gettime();
time = (time < now ? 0 : time - now);
}
// set active and reactor
bt->active = 1;
bt->reactor = bsys;
// init source
bt->source = g_timeout_source_new(time);
g_source_set_callback(bt->source, timer_source_handler, bt, NULL);
g_source_attach(bt->source, g_main_loop_get_context(bsys->gloop));
DebugCounter_Increment(&bsys->d_timers_ctr);
}
void BReactor_RemoveSmallTimer (BReactor *bsys, BSmallTimer *bt)
{
DebugObject_Access(&bsys->d_obj);
assert_timer(bt);
// do nothing if timer is not active
if (!bt->active) {
return;
}
// free source
g_source_destroy(bt->source);
g_source_unref(bt->source);
// set not active
bt->active = 0;
DebugCounter_Decrement(&bsys->d_timers_ctr);
}
void BReactor_SetTimer (BReactor *bsys, BTimer *bt)
{
BReactor_SetSmallTimer(bsys, &bt->base, BTIMER_SET_RELATIVE, bt->msTime);
}
void BReactor_SetTimerAfter (BReactor *bsys, BTimer *bt, btime_t after)
{
BReactor_SetSmallTimer(bsys, &bt->base, BTIMER_SET_RELATIVE, after);
}
void BReactor_SetTimerAbsolute (BReactor *bsys, BTimer *bt, btime_t time)
{
BReactor_SetSmallTimer(bsys, &bt->base, BTIMER_SET_ABSOLUTE, time);
}
void BReactor_RemoveTimer (BReactor *bsys, BTimer *bt)
{
return BReactor_RemoveSmallTimer(bsys, &bt->base);
}
BPendingGroup * BReactor_PendingGroup (BReactor *bsys)
{
DebugObject_Access(&bsys->d_obj);
return &bsys->pending_jobs;
}
int BReactor_Synchronize (BReactor *bsys, BSmallPending *ref)
{
DebugObject_Access(&bsys->d_obj);
ASSERT(ref)
while (!bsys->exiting) {
ASSERT(BPendingGroup_HasJobs(&bsys->pending_jobs))
if (BPendingGroup_PeekJob(&bsys->pending_jobs) == ref) {
return 1;
}
BPendingGroup_ExecuteJob(&bsys->pending_jobs);
}
return 0;
}
int BReactor_AddFileDescriptor (BReactor *bsys, BFileDescriptor *bs)
{
DebugObject_Access(&bsys->d_obj);
ASSERT(!bs->active)
// set active, no wait events, and set reactor
bs->active = 1;
bs->waitEvents = 0;
bs->reactor = bsys;
// create source
bs->source = g_source_new(&bsys->fd_source_funcs, sizeof(struct fd_source));
((struct fd_source *)bs->source)->bfd = bs;
// init pollfd
bs->pollfd.fd = bs->fd;
bs->pollfd.events = get_glib_wait_events(bs->waitEvents);
bs->pollfd.revents = 0;
// start source
g_source_add_poll(bs->source, &bs->pollfd);
g_source_attach(bs->source, g_main_loop_get_context(bsys->gloop));
DebugCounter_Increment(&bsys->d_fds_counter);
return 1;
}
void BReactor_RemoveFileDescriptor (BReactor *bsys, BFileDescriptor *bs)
{
DebugObject_Access(&bsys->d_obj);
DebugCounter_Decrement(&bsys->d_fds_counter);
ASSERT(bs->active)
// free source
g_source_destroy(bs->source);
g_source_unref(bs->source);
// set not active
bs->active = 0;
}
void BReactor_SetFileDescriptorEvents (BReactor *bsys, BFileDescriptor *bs, int events)
{
DebugObject_Access(&bsys->d_obj);
ASSERT(bs->active)
ASSERT(!(events&~(BREACTOR_READ|BREACTOR_WRITE)))
// set new wait events
bs->waitEvents = events;
// update pollfd wait events
bs->pollfd.events = get_glib_wait_events(bs->waitEvents);
}
int BReactor_InitFromExistingGMainLoop (BReactor *bsys, GMainLoop *gloop, int unref_gloop_on_free)
{
ASSERT(gloop)
ASSERT(unref_gloop_on_free == !!unref_gloop_on_free)
// set not exiting
bsys->exiting = 0;
// set gloop and unref on free flag
bsys->gloop = gloop;
bsys->unref_gloop_on_free = unref_gloop_on_free;
// init fd source functions table
memset(&bsys->fd_source_funcs, 0, sizeof(bsys->fd_source_funcs));
bsys->fd_source_funcs.prepare = fd_source_func_prepare;
bsys->fd_source_funcs.check = fd_source_func_check;
bsys->fd_source_funcs.dispatch = fd_source_func_dispatch;
bsys->fd_source_funcs.finalize = NULL;
// init job queue
BPendingGroup_Init(&bsys->pending_jobs);
// init active limits list
LinkedList1_Init(&bsys->active_limits_list);
DebugCounter_Init(&bsys->d_fds_counter);
DebugCounter_Init(&bsys->d_limits_ctr);
DebugCounter_Init(&bsys->d_timers_ctr);
DebugObject_Init(&bsys->d_obj);
return 1;
}
GMainLoop * BReactor_GetGMainLoop (BReactor *bsys)
{
DebugObject_Access(&bsys->d_obj);
return bsys->gloop;
}
int BReactor_SynchronizeAll (BReactor *bsys)
{
DebugObject_Access(&bsys->d_obj);
dispatch_pending(bsys);
return !bsys->exiting;
}
void BReactorLimit_Init (BReactorLimit *o, BReactor *reactor, int limit)
{
DebugObject_Access(&reactor->d_obj);
ASSERT(limit > 0)
// init arguments
o->reactor = reactor;
o->limit = limit;
// set count zero
o->count = 0;
DebugCounter_Increment(&reactor->d_limits_ctr);
DebugObject_Init(&o->d_obj);
}
void BReactorLimit_Free (BReactorLimit *o)
{
BReactor *reactor = o->reactor;
DebugObject_Free(&o->d_obj);
DebugCounter_Decrement(&reactor->d_limits_ctr);
// remove from active limits list
if (o->count > 0) {
LinkedList1_Remove(&reactor->active_limits_list, &o->active_limits_list_node);
}
}
int BReactorLimit_Increment (BReactorLimit *o)
{
BReactor *reactor = o->reactor;
DebugObject_Access(&o->d_obj);
// check count against limit
if (o->count >= o->limit) {
return 0;
}
// increment count
o->count++;
// if limit was zero, add to active limits list
if (o->count == 1) {
LinkedList1_Append(&reactor->active_limits_list, &o->active_limits_list_node);
}
return 1;
}
void BReactorLimit_SetLimit (BReactorLimit *o, int limit)
{
DebugObject_Access(&o->d_obj);
ASSERT(limit > 0)
// set limit
o->limit = limit;
}