Changeset 986ca9cdcea688fff7615506f7a522fed667967b

Timestamp:

05/13/11 03:05:48 (3 years ago)

Author:

Theo Schlossnagle <jesus@omniti.com>

git-committer:

Theo Schlossnagle <jesus@omniti.com> 1305255948 -0400

git-parent:

[d2b2cf6a9fe0bc981e17e3ca09ad4c053f699dff]

git-author:

Theo Schlossnagle <jesus@omniti.com> 1305255948 -0400

Message:

add support for asynchronous jlog'in

Files:

• src/utils/noit_log.c (modified) (10 diffs)

src/utils/noit_log.c

@@ -226 +226 @@
-};
-
+typedef struct jlog_line {
+  char *buf;
+  char buf_static[512];
+  char *buf_dynamic;
+  int len;
+  void *next;
+} jlog_line;
+
+typedef struct {
+  jlog_ctx *log;
+  pthread_t writer;
+  void *head;
+} jlog_asynch_ctx;
+
-static int
-jlog_lspath_to_fspath(noit_log_stream_t ls, char *buff, int len,
@@ -267 +281 @@
-static int
-jlog_logio_cleanse(noit_log_stream_t ls) {
+  jlog_asynch_ctx *actx;
-  jlog_ctx *log;
-  DIR *d;
@@ -274 +289 @@
-  int size = 0;
-
-
+
+
+
-  if(!log) return -1;
-  if(jlog_lspath_to_fspath(ls, path, sizeof(path), NULL) <= 0) return -1;
@@ -315 +332 @@
-jlog_logio_reopen(noit_log_stream_t ls) {
-  char **subs;
+  jlog_asynch_ctx *actx = ls->op_ctx;
-  int i;
-  /* reopening only has the effect of removing temporary subscriptions */
@@ -320 +338 @@
-
-  if(ls->lock) pthread_rwlock_wrlock(ls->lock);
-ls->op_ctx
+actx->log
-    goto bail;
-
-  for(i=0;subs[i];i++)
-    if(subs[i][0] == '~')
-ls->op_ctx
-
-ls->op_ctx
+actx->log
+
+actx->log
-  jlog_logio_cleanse(ls);
- bail:
@@ -333 +351 @@
-  return 0;
-}
+static jlog_line *
+jlog_asynch_pop(jlog_asynch_ctx *actx, jlog_line **iter) {
+  jlog_line *h = NULL, *rev = NULL;
+
+  if(*iter) { /* we have more on the previous list */
+    h = *iter;
+    *iter = h->next;
+    return h;
+  }
+
+  while(1) {
+    h = actx->head;
+    if(noit_atomic_casptr((volatile void **)&actx->head, NULL, h) == h) break;
+  }
+  while(h) {
+    /* which unshifted things into the queue -- it's backwards, reverse it */
+    jlog_line *tmp = h;
+    h = h->next;
+    tmp->next = rev;
+    rev = tmp;
+  }
+  if(rev) *iter = rev->next;
+  else *iter = NULL;
+  return rev;
+}
+void
+jlog_asynch_push(jlog_asynch_ctx *actx, jlog_line *n) {
+  while(1) {
+    n->next = actx->head;
+    if(noit_atomic_casptr((volatile void **)&actx->head, n, n->next) == n->next) return;
+  }
+}
+static void *
+jlog_logio_asynch_writer(void *vls) {
+  noit_log_stream_t ls = vls;
+  jlog_asynch_ctx *actx = ls->op_ctx;
+  jlog_line *iter = NULL;
+  while(1) {
+    int fast = 0, max = 1000;
+    jlog_line *line;
+    pthread_rwlock_rdlock(ls->lock);
+    while(max > 0 && NULL != (line = jlog_asynch_pop(actx, &iter))) {
+      jlog_ctx_write(actx->log, line->buf_dynamic ?
+                                  line->buf_dynamic :
+                                  line->buf_static,
+                     line->len);
+      if(line->buf_dynamic != NULL) free(line->buf_dynamic);
+      free(line);
+      fast = 1;
+      max--;
+    }
+    pthread_rwlock_unlock(ls->lock);
+    if(max > 0) {
+      /* we didn't hit our limit... so we ran the queue dry */
+      /* 200ms if there was nothing, 10ms otherwise */
+      usleep(fast ? 10000 : 200000);
+    }
+  }
+}
-static int
-jlog_logio_open(noit_log_stream_t ls) {
-  char path[PATH_MAX], *sub, **subs, *p;
+  jlog_asynch_ctx *actx;
-  jlog_ctx *log = NULL;
+  pthread_attr_t tattr;
-  int i, listed, found;
-
@@ -417 +496 @@
-  }
-
-
+
+
+
-  /* We do this to clean things up */
-  jlog_logio_reopen(ls);
+
+  pthread_attr_init(&tattr);
+  pthread_attr_setdetachstate(&tattr, PTHREAD_CREATE_DETACHED);
+  if(pthread_create(&actx->writer, &tattr, jlog_logio_asynch_writer, ls) != 0) {
+    return -1;
+  }
-  return 0;
-}
@@ -425 +512 @@
-jlog_logio_write(noit_log_stream_t ls, const void *buf, size_t len) {
-  int rv = -1;
+  jlog_asynch_ctx *actx;
+  jlog_line *line;
-  if(!ls->op_ctx) return -1;
-
-
-
-
+
+
+
+
+
+
+
+
+
+
+
-  return rv;
-}
@@ -435 +531 @@
-jlog_logio_close(noit_log_stream_t ls) {
-  if(ls->op_ctx) {
-
+
+
-    ls->op_ctx = NULL;
-  }
@@ -443 +540 @@
-jlog_logio_size(noit_log_stream_t ls) {
-  size_t size;
+  jlog_asynch_ctx *actx;
-  if(!ls->op_ctx) return -1;
+  actx = ls->op_ctx;
-  pthread_rwlock_rdlock(ls->lock);
-(jlog_ctx *)ls->op_ctx
+actx->log
-  pthread_rwlock_unlock(ls->lock);
-  return size;