Changeset b62cf2be087943dcb29b6e068bd4262862fcb17d

Timestamp:

12/17/07 05:43:26 (5 years ago)

Author:

Theo Schlossnagle <jesus@omniti.com>

git-committer:

Theo Schlossnagle <jesus@omniti.com> 1197870206 +0000

git-parent:

[cd1ab55918856e8682710ede8ef2e44377887173]

git-author:

Theo Schlossnagle <jesus@omniti.com> 1197870206 +0000

Message:

more work... fleshing out the eventer

Files:

• configure.in (modified) (1 diff)
• src/eventer/eventer.c (modified) (1 diff)
• src/eventer/eventer.h (modified) (2 diffs)
• src/eventer/eventer_POSIX_fd_opset.c (modified) (1 diff)
• src/eventer/eventer_kqueue_impl.c (modified) (6 diffs)
• src/noit_defines.h (modified) (1 diff)
• src/noitd.c (modified) (2 diffs)
• src/utils/Makefile.in (modified) (1 diff)
• src/utils/noit_skiplist.c (added)
• src/utils/noit_skiplist.h (added)

configure.in

@@ -12 +12 @@
-AC_PATH_PROGS(PERL, perl)
-AC_SUBST(PERL)
+
+if test "x$CC" = "xgcc" ; then
+  CFLAGS="$CFLAGS -g -Wall"
+fi
-
-# Checks for data types

src/eventer/eventer.c

@@ -8 +8 @@
-  e->opset = eventer_POSIX_fd_opset;
-  return e;
+}
+
+int eventer_timecompare(void *av, void *bv) {
+  /* Herein we avoid equality.  This function is only used as a comparator
+   * for a heap of timed events.  If they are equal, b is considered less
+   * just to maintain an order (despite it not being stable).
+   */
+  eventer_t a = (eventer_t)av;
+  eventer_t b = (eventer_t)bv;
+  if(a->whence.tv_sec < b->whence.tv_sec) return -1;
+  if(a->whence.tv_sec == b->whence.tv_sec &&
+     a->whence.tv_usec < b->whence.tv_usec) return -1;
+  return 1;
-}
-

src/eventer/eventer.h

@@ -43 +43 @@
-struct _event {
-  eventer_func_t      callback;
-*
+ 
-  int                 fd;
-  int                 mask;
@@ -52 +52 @@
-API_EXPORT(eventer_t) eventer_alloc();
-API_EXPORT(void)      eventer_free(eventer_t);
+API_EXPORT(int)       eventer_timecompare(void *a, void *b);
-
-typedef struct _eventer_impl {

src/eventer/eventer_POSIX_fd_opset.c

@@ -8 +8 @@
-
-#include <sys/socket.h>
+#include <unistd.h>
-
-static int

src/eventer/eventer_kqueue_impl.c

@@ -7 +7 @@
-#include "eventer/eventer.h"
-#include "utils/noit_atomic.h"
-
+
+
+
+
+
-#include <sys/event.h>
-#include <pthread.h>
-
-
-
+
+
+
+
+
-  eventer_t e;
-  pthread_t executor;
-  noit_spinlock_t lock;
-
-
-
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
-typedef struct kqueue_setup {
-  struct kevent *__ke_vec;
-  unsigned int __ke_vec_a;
-  unsigned int __ke_vec_used;
- 
+
-
-static pthread_mutex_t kqs_lock;
+static pthread_mutex_t te_lock;
-static kqs_t master_kqs = NULL;
-static pthread_key_t kqueue_setup_key;
+static noit_skiplist *timed_events = NULL;
-#define KQUEUE_DECL kqs_t kqs
-#define KQUEUE_SETUP kqs = (kqs_t) pthread_getspecific(kqueue_setup_key)
@@ -34 +66 @@
-#define ke_vec_used kqs->__ke_vec_used
-
+static void kqs_init(kqs_t kqs) {
+  enum { initial_alloc = 64 };
+  ke_vec_a = initial_alloc;
+  ke_vec = (struct kevent *) malloc(ke_vec_a * sizeof (struct kevent));
+}
-static void
-ke_change (register int const ident,
@@ -39 +76 @@
-           register int const flags,
-           register void *const udata) {
-  enum { initial_alloc = 64 };
-  register struct kevent *kep;
-  KQUEUE_DECL;
-
-  KQUEUE_SETUP;
+  if(!kqs) kqs = master_kqs;
+
+  if(kqs == master_kqs) pthread_mutex_lock(&kqs_lock);
-  if (!ke_vec_a) {
-
-
+
-  }
-  else if (ke_vec_used == ke_vec_a) {
@@ -56 +93 @@
-
-  EV_SET(kep, ident, filter, flags, 0, 0, udata);
+  if(kqs == master_kqs) pthread_mutex_unlock(&kqs_lock);
-}
-
-static int eventer_kqueue_impl_init() {
-  struct rlimit rlim;
+  master_thread = pthread_self();
-  kqueue_fd = kqueue();
-  if(kqueue_fd == -1) {
@@ -65 +104 @@
-  }
-  pthread_mutex_init(&kqs_lock, NULL);
+  pthread_mutex_init(&te_lock, NULL);
+  pthread_key_create(&kqueue_setup_key, NULL);
-  master_kqs = calloc(1, sizeof(*master_kqs));
+  kqs_init(master_kqs);
-  getrlimit(RLIMIT_NOFILE, &rlim);
-  maxfds = rlim.rlim_cur;
-  master_fds = calloc(maxfds, sizeof(*master_fds));
+  timed_events = calloc(1, sizeof(*timed_events));
+  noit_skiplist_init(timed_events);
+  noit_skiplist_set_compare(timed_events,
+                            eventer_timecompare, eventer_timecompare);
+  noit_skiplist_add_index(timed_events,
+                          noit_compare_voidptr, noit_compare_voidptr);
-  return 0;
-}
@@ -75 +123 @@
-}
-static void eventer_kqueue_impl_add(eventer_t e) {
+  assert(e->mask);
+  ev_lock_state_t lockstate;
+
+  /* Timed events are simple */
+  if(e->mask == EVENTER_TIMER) {
+    pthread_mutex_lock(&te_lock);
+    noit_skiplist_insert(timed_events, e);
+    pthread_mutex_unlock(&te_lock);
+    return;
+  }
+
+  /* file descriptor event */
+  lockstate = acquire_master_fd(e->fd);
+  master_fds[e->fd].e = e;
+  if(e->mask & (EVENTER_READ | EVENTER_EXCEPTION))
+    ke_change(e->fd, EVFILT_READ, EV_ADD | EV_ENABLE, e);
+  if(e->mask & (EVENTER_WRITE))
+    ke_change(e->fd, EVFILT_WRITE, EV_ADD | EV_ENABLE, e);
+  release_master_fd(e->fd, lockstate);
-}
-static void eventer_kqueue_impl_remove(eventer_t e) {
+  if(e->mask & (EVENTER_READ | EVENTER_WRITE | EVENTER_EXCEPTION)) {
+    ev_lock_state_t lockstate;
+    lockstate = acquire_master_fd(e->fd);
+    if(e == master_fds[e->fd].e) {
+      master_fds[e->fd].e = NULL;
+      if(e->mask & (EVENTER_READ | EVENTER_EXCEPTION))
+        ke_change(e->fd, EVFILT_READ, EV_DELETE | EV_DISABLE, e);
+      if(e->mask & (EVENTER_WRITE))
+        ke_change(e->fd, EVFILT_WRITE, EV_DELETE | EV_DISABLE, e);
+    }
+    release_master_fd(e->fd, lockstate);
+  }
+  else if(e->mask & EVENTER_TIMER) {
+    pthread_mutex_lock(&te_lock);
+    noit_skiplist_remove_compare(timed_events, e, NULL, noit_compare_voidptr);
+    pthread_mutex_unlock(&te_lock);
+  }
+  else {
+    abort();
+  }
-}
-static void eventer_kqueue_impl_update(eventer_t e) {
+  if(e->mask & EVENTER_TIMER) {
+    pthread_mutex_lock(&te_lock);
+    noit_skiplist_remove_compare(timed_events, e, NULL, noit_compare_voidptr);
+    noit_skiplist_insert(timed_events, e);
+    pthread_mutex_unlock(&te_lock);
+    return;
+  }
+  ke_change(e->fd, EVFILT_READ, EV_DELETE | EV_DISABLE, e);
+  ke_change(e->fd, EVFILT_WRITE, EV_DELETE | EV_DISABLE, e);
+  if(e->mask & (EVENTER_READ | EVENTER_EXCEPTION))
+    ke_change(e->fd, EVFILT_READ, EV_DELETE | EV_DISABLE, e);
+  if(e->mask & (EVENTER_WRITE))
+    ke_change(e->fd, EVFILT_WRITE, EV_DELETE | EV_DISABLE, e);
-}
-static eventer_t eventer_kqueue_impl_remove_fd(int fd) {
+  eventer_t eiq = NULL;
+  ev_lock_state_t lockstate;
+  if(master_fds[fd].e) {
+    lockstate = acquire_master_fd(fd);
+    eiq = master_fds[fd].e;
+    master_fds[fd].e = NULL;
+    if(eiq->mask & (EVENTER_READ | EVENTER_EXCEPTION))
+      ke_change(fd, EVFILT_READ, EV_DELETE | EV_DISABLE, eiq);
+    if(eiq->mask & (EVENTER_WRITE))
+      ke_change(fd, EVFILT_WRITE, EV_DELETE | EV_DISABLE, eiq);
+    release_master_fd(fd, lockstate);
+  }
+  return eiq;
-}
-static void eventer_kqueue_impl_loop() {
+  int is_master_thread = 0;
+  pthread_t self;
+  KQUEUE_DECL;
+  KQUEUE_SETUP;
+
+  self = pthread_self();
+  if(pthread_equal(self, master_thread)) is_master_thread = 1;
+
+  if(!kqs) {
+    kqs = calloc(1, sizeof(*kqs));
+    kqs_init(kqs);
+  }
+  pthread_setspecific(kqueue_setup_key, kqs);
+  while(1) {
+    struct timeval __now, __sleeptime;
+    struct timespec __kqueue_sleeptime;
+    int fd_cnt = 0;
+    int max_timed_events_to_process;
+    int newmask;
+
+    __sleeptime = __max_sleeptime;
+
+    /* Handle timed events...
+     * we could be multithreaded, so if we pop forever we could starve
+     * ourselves. */
+    max_timed_events_to_process = timed_events->size;
+    while(max_timed_events_to_process-- > 0) {
+      eventer_t timed_event;
+
+      gettimeofday(&__now, NULL);
+
+      pthread_mutex_lock(&te_lock);
+      /* Peek at our next timed event, if should fire, pop it.
+       * otherwise we noop and NULL it out to break the loop. */
+      timed_event = noit_skiplist_peek(timed_events);
+      if(timed_event) {
+        if(compare_timeval(timed_event->whence, __now) < 0) {
+          timed_event = noit_skiplist_pop(timed_events, NULL);
+        }
+        else {
+          sub_timeval(timed_event->whence, __now, &__sleeptime);
+          timed_event = NULL;
+        }
+      }
+      pthread_mutex_unlock(&te_lock);
+      if(timed_event == NULL) break;
+
+      /* Make our call */
+      newmask = timed_event->callback(timed_event, EVENTER_TIMER,
+                                      timed_event->closure, &__now);
+      if(newmask)
+        eventer_add(timed_event);
+      else
+        eventer_free(timed_event);
+    }
+
+    if(compare_timeval(__max_sleeptime, __sleeptime) < 0) {
+      /* we exceed our configured maximum, set it down */
+      memcpy(&__sleeptime, &__max_sleeptime, sizeof(__sleeptime));
+    }
+
+    /* If we're the master, we need to lock the master_kqs and make mods */
+    if(master_kqs->__ke_vec_used) {
+      struct timespec __zerotime = { 0, 0 };
+      pthread_mutex_lock(&kqs_lock);
+      fd_cnt = kevent(kqueue_fd,
+                      master_kqs->__ke_vec, master_kqs->__ke_vec_used,
+                      NULL, 0,
+                      &__zerotime);
+      if(fd_cnt < 0) {
+        fprintf(stderr, "kevent: %s\n", strerror(errno));
+      }
+      master_kqs->__ke_vec_used = 0;
+      pthread_mutex_unlock(&kqs_lock);
+    }
+
+    /* Now we move on to our fd-based events */
+    __kqueue_sleeptime.tv_sec = __sleeptime.tv_sec;
+    __kqueue_sleeptime.tv_nsec = __sleeptime.tv_usec * 1000;
+    fd_cnt = kevent(kqueue_fd, ke_vec, ke_vec_used,
+                    ke_vec, ke_vec_a,
+                    &__kqueue_sleeptime);
+    ke_vec_used = 0;
+    if(fd_cnt < 0) {
+      fprintf(stderr, "kevent: %s\n", strerror(errno));
+    }
+    else {
+      int idx;
+      for(idx = 0; idx < fd_cnt; idx++) {
+        ev_lock_state_t lockstate;
+        struct kevent *ke;
+        eventer_t e;
+        int fd, evmask, oldmask;
+
+        ke = &ke_vec[idx];
+        e = (eventer_t)ke->udata;
+        fd = ke->ident;
+        assert(e == master_fds[fd].e);
+        lockstate = acquire_master_fd(fd);
+        assert(lockstate == EV_OWNED);
+
+        evmask = 0;
+        if(ke->filter == EVFILT_READ) evmask = EVENTER_READ;
+        if(ke->filter == EVFILT_WRITE) evmask = EVENTER_WRITE;
+        gettimeofday(&__now, NULL);
+        oldmask = e->mask;
+        newmask = e->callback(e, evmask, e->closure, &__now);
+
+        if(newmask) {
+          /* toggle the read bits if needed */
+          if(newmask & (EVENTER_READ | EVENTER_EXCEPTION)) {
+            if(!(oldmask & (EVENTER_READ | EVENTER_EXCEPTION)))
+              ke_change(fd, EVFILT_READ, EV_ADD | EV_ENABLE, e);
+          }
+          else if(oldmask & (EVENTER_READ | EVENTER_EXCEPTION))
+            ke_change(fd, EVFILT_READ, EV_DELETE | EV_DISABLE, e);
+  
+          /* toggle the write bits if needed */
+          if(newmask & EVENTER_WRITE) {
+            if(!(oldmask & EVENTER_WRITE))
+              ke_change(fd, EVFILT_WRITE, EV_ADD | EV_ENABLE, e);
+          }
+          else if(oldmask & EVENTER_WRITE)
+              ke_change(fd, EVFILT_WRITE, EV_DELETE | EV_DISABLE, e);
+  
+          /* Set our mask */
+          e->mask = newmask;
+        }
+        else {
+          eventer_free(e);
+        }
+        release_master_fd(fd, lockstate);
+      }
+    }
+  }
-}
-

src/noit_defines.h

@@ -6 +6 @@
-#define API_EXPORT(type) extern type
-
+static inline int compare_timeval(struct timeval a, struct timeval b) {
+  if (a.tv_sec < b.tv_sec) return -1;
+  if (a.tv_sec > b.tv_sec) return 1;
+  if (a.tv_usec < b.tv_usec) return -1;
+  if (a.tv_usec > b.tv_usec) return 1;
+  return 0;
+}
+
+static inline void sub_timeval(struct timeval a, struct timeval b,
+                               struct timeval *out)
+{
+  out->tv_usec = a.tv_usec - b.tv_usec;
+  if (out->tv_usec < 0L) {
+    a.tv_sec--;
+    out->tv_usec += 1000000L;
+  }
+  out->tv_sec = a.tv_sec - b.tv_sec;
+  if (out->tv_sec < 0L) {
+    out->tv_sec++;
+    out->tv_usec -= 1000000L;
+  }
+}
+
-#endif

src/noitd.c

@@ -5 +5 @@
-#include <stdlib.h>
-
+int stdin_handler(eventer_t e, int mask, void *closure, struct timeval *now) {
+  fprintf(stderr, "in stdin_handler:\n");
+  return EVENTER_READ;
+}
+void stdin_sample() {
+  eventer_t e;
+  e = eventer_alloc();
+  e->fd = 0;
+  e->mask = EVENTER_READ;
+  e->callback = stdin_handler;
+  eventer_add(e);
+}
-int main(int argc, char **argv) {
-  if(eventer_choose("kqueue") == -1) {
@@ -14 +26 @@
-    exit(-1);
-  }
+
+  stdin_sample();
+  eventer_loop();
-  return 0;
-}

src/utils/Makefile.in

@@ -10 +10 @@
-top_srcdir=@top_srcdir@
-
-
+ noit_skiplist.o
-
-all:    libnoit_utils.a