Tue, 10 Nov 2015 21:18:51 +0100
merge
/* * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS HEADER. * * Copyright 2013 Olaf Wintermann. All rights reserved. * * 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. * * 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 COPYRIGHT HOLDER OR CONTRIBUTORS 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 <stdio.h> #include <stdlib.h> #include "../util/atomic.h" #include "event_bsd.h" event_handler_t* evhandler_create(int numthreads) { event_handler_t *ev = malloc(sizeof(event_handler_t)); if(ev == NULL) { return NULL; } ev->ports = calloc(numthreads, sizeof(int)); if(ev->ports == NULL) { free(ev); return NULL; } ev->nports = numthreads; ev->lp = 0; /* create ports event threads */ for(int i=0;i<numthreads;i++) { /* create port */ //ev->ports[i] = port_create(); ev->ports[i] = kqueue(); if(ev->ports[i] == 0) { free(ev->ports); free(ev); return NULL; } /* * start a new handler thread * the thread needs the event port and a pointer to the event handler */ ev_thr_conf_t *conf = malloc(sizeof(ev_thr_conf_t)); if(conf == NULL) { free(ev->ports); free(ev); return NULL; } conf->handler = ev; conf->port = ev->ports[i]; systhread_start(0, 0, (thrstartfunc)ev_handle_events, conf); /* TODO: error handling */ } return ev; } void ev_handle_events(ev_thr_conf_t *conf) { event_handler_t *ev = conf->handler; int kq = conf->port; free(conf); struct timespec timeout; timeout.tv_nsec = 0; timeout.tv_sec = 600; struct kevent events[16]; for(;;) { // wait for events int nev; nev = kevent(kq, NULL, 0, events, 16, &timeout); if(nev == -1) { // TODO: check for error perror("kevent"); continue; } for(int i=0;i<nev;i++) { event_t *event = (event_t*)events[i].udata; if(event->fn) { if(event->fn(ev, event)) { // TODO: reassociate? } else if(event->finish) { event->finish(ev, event); } } } } } /* returns a event handler port */ int ev_get_port(event_handler_t *h) { int nps = h->nports; if(nps == 1) { return h->ports[0]; } int cp = h->lp % nps; ws_atomic_inc32(&h->lp); return h->ports[cp]; } int ev_pollin(event_handler_t *h, int fd, event_t *event) { struct kevent kev; EV_SET(&kev, fd, EVFILT_READ, EV_ADD, 0, 0, event); return kevent(ev_get_port(h), &kev, 1, NULL, 0, NULL); } int ev_pollout(event_handler_t *h, int fd, event_t *event) { struct kevent kev; EV_SET(&kev, fd, EVFILT_WRITE, EV_ADD, 0, 0, event); return kevent(ev_get_port(h), &kev, 1, NULL, 0, NULL); } int evt_send(event_handler_t *h, event_t *event) { return 0; }