Retry three times, then bail out and return not found.

Trace line numbers within macros.
WIP: race condition on close causes double-free
2013-09-04 13:11:02 -04:00 · 2013-09-04 13:10:16 -04:00 · 2013-09-04 13:09:17 -04:00 · 2013-08-26 13:07:46 -04:00 · 2013-08-21 15:56:31 -04:00 · 2013-08-13 08:54:22 -04:00
13 changed files with 475 additions and 345 deletions
--- a/.gitignore
+++ b/.gitignore
@ -7,8 +7,6 @@ c_src/*.o
 c_src/bzip2-1.0.6
 c_src/snappy-1.0.4
 deps/
-priv/wt
-priv/*.so*
-priv/*.dylib*
+priv/
 log/
 *~
--- a/4
+++ b/4
@ -52,17 +52,19 @@ endif
 .PHONY: all compile doc clean test dialyzer typer shell distclean pdf \
  update-deps clean-common-test-data rebuild

-all: deps compile
+all: deps compile test

 # =============================================================================
 # Rules to build the system
 # =============================================================================

 deps:
+	c_src/build_deps.sh get-deps
 	$(REBAR) get-deps
 	$(REBAR) compile

 update-deps:
+	c_src/build_deps.sh update-deps
 	$(REBAR) update-deps
 	$(REBAR) compile

--- a/c_src/async_nif.h
+++ b/c_src/async_nif.h
@ -24,9 +24,11 @@
 extern "C" {
 #endif

-#include <assert.h>
-
-#include "queue.h"
+#include <stdlib.h>
+#include <urcu.h>		/* RCU flavor */
+#include <urcu/arch.h>          /* RCU Architecture */
+#include <urcu/tls-compat.h>    /* RCU Thread-local storage */
+#include <urcu/rculfqueue.h>	/* RCU Lock-free queue */

 #ifndef UNUSED
 #define UNUSED(v) ((void)(v))
@ -37,6 +39,9 @@ extern "C" {
 #define ASYNC_NIF_WORKER_QUEUE_SIZE 8192
 #define ASYNC_NIF_MAX_QUEUED_REQS ASYNC_NIF_WORKER_QUEUE_SIZE * ASYNC_NIF_MAX_WORKERS

+static DEFINE_URCU_TLS(unsigned long long, nr_enqueues);
+static DEFINE_URCU_TLS(unsigned long long, nr_dequeues);
+
 /* Atoms (initialized in on_load) */
 static ERL_NIF_TERM ATOM_EAGAIN;
 static ERL_NIF_TERM ATOM_ENOMEM;
@ -45,7 +50,6 @@ static ERL_NIF_TERM ATOM_ERROR;
 static ERL_NIF_TERM ATOM_OK;
 static ERL_NIF_TERM ATOM_SHUTDOWN;

-
 struct async_nif_req_entry {
  ERL_NIF_TERM ref;
  ErlNifEnv *env;
@ -53,17 +57,15 @@ struct async_nif_req_entry {
  void *args;
  void (*fn_work)(ErlNifEnv*, ERL_NIF_TERM, ErlNifPid*, unsigned int, void *);
  void (*fn_post)(void *);
-  STAILQ_ENTRY(async_nif_req_entry) entries;
+  struct cds_lfq_node_rcu queue_entry;
+  struct rcu_head rcu_head;
 };

-
 struct async_nif_work_queue {
  unsigned int num_workers;
  unsigned int depth;
-  ErlNifMutex *reqs_mutex;
-  ErlNifCond *reqs_cnd;
+  struct cds_lfq_queue_rcu req_queue;
  struct async_nif_work_queue *next;
-  STAILQ_HEAD(reqs, async_nif_req_entry) reqs;
 };

 struct async_nif_worker_entry {
@ -71,40 +73,39 @@ struct async_nif_worker_entry {
  unsigned int worker_id;
  struct async_nif_state *async_nif;
  struct async_nif_work_queue *q;
-  SLIST_ENTRY(async_nif_worker_entry) entries;
+  struct cds_lfq_node_rcu queue_entry;
+  struct rcu_head rcu_head;
 };

 struct async_nif_state {
  unsigned int shutdown;
-  ErlNifMutex *we_mutex;
-  unsigned int we_active;
-  SLIST_HEAD(joining, async_nif_worker_entry) we_joining;
+  unsigned int num_active_workers;
+  struct cds_lfq_queue_rcu worker_join_queue;
  unsigned int num_queues;
  unsigned int next_q;
-  STAILQ_HEAD(recycled_reqs, async_nif_req_entry) recycled_reqs;
+  struct cds_lfq_queue_rcu recycled_req_queue;
  unsigned int num_reqs;
-  ErlNifMutex *recycled_req_mutex;
  struct async_nif_work_queue queues[];
 };

 #define ASYNC_NIF_DECL(decl, frame, pre_block, work_block, post_block)  \
-  struct decl ## _args frame;                                           \
-  static void fn_work_ ## decl (ErlNifEnv *env, ERL_NIF_TERM ref, ErlNifPid *pid, unsigned int worker_id, struct decl ## _args *args) { \
+  struct decl##_args frame;                                             \
+  static void fn_work_##decl (ErlNifEnv *env, ERL_NIF_TERM ref, ErlNifPid *pid, unsigned int worker_id, struct decl##_args *args) { \
  UNUSED(worker_id);                                                    \
  DPRINTF("async_nif: calling \"%s\"", __func__);                       \
  do work_block while(0);                                               \
  DPRINTF("async_nif: returned from \"%s\"", __func__);                 \
  }                                                                     \
-  static void fn_post_ ## decl (struct decl ## _args *args) {           \
+  static void fn_post_##decl (struct decl##_args *args) {               \
    UNUSED(args);                                                       \
    DPRINTF("async_nif: calling \"fn_post_%s\"", #decl);                \
    do post_block while(0);                                             \
    DPRINTF("async_nif: returned from \"fn_post_%s\"", #decl);          \
  }                                                                     \
  static ERL_NIF_TERM decl(ErlNifEnv* env, int argc, const ERL_NIF_TERM argv_in[]) { \
-    struct decl ## _args on_stack_args;                                 \
-    struct decl ## _args *args = &on_stack_args;                        \
-    struct decl ## _args *copy_of_args;                                 \
+    struct decl##_args on_stack_args;                                   \
+    struct decl##_args *args = &on_stack_args;                          \
+    struct decl##_args *copy_of_args;                                   \
    struct async_nif_req_entry *req = NULL;                             \
    unsigned int affinity = 0;                                          \
    ErlNifEnv *new_env = NULL;                                          \
@ -124,22 +125,22 @@ struct async_nif_state {
    DPRINTF("async_nif: returned from \"%s\"", __func__);               \
    copy_of_args = (struct decl ## _args *)malloc(sizeof(struct decl ## _args)); \
    if (!copy_of_args) {                                                \
-      fn_post_ ## decl (args);                                          \
+      fn_post_##decl (args);                                            \
      async_nif_recycle_req(req, async_nif);                            \
      return enif_make_tuple2(env, ATOM_ERROR, ATOM_ENOMEM);		\
    }                                                                   \
-    memcpy(copy_of_args, args, sizeof(struct decl ## _args));           \
+    memcpy(copy_of_args, args, sizeof(struct decl##_args));             \
    req->ref = enif_make_copy(new_env, argv_in[0]);                     \
    enif_self(env, &req->pid);                                          \
    req->args = (void*)copy_of_args;                                    \
-    req->fn_work = (void (*)(ErlNifEnv *, ERL_NIF_TERM, ErlNifPid*, unsigned int, void *))fn_work_ ## decl ; \
-    req->fn_post = (void (*)(void *))fn_post_ ## decl;                 \
+    req->fn_work = (void (*)(ErlNifEnv *, ERL_NIF_TERM, ErlNifPid*, unsigned int, void *))fn_work_##decl ; \
+    req->fn_post = (void (*)(void *))fn_post_##decl;                   \
    int h = -1;                                                        \
    if (affinity)                                                      \
        h = ((unsigned int)affinity) % async_nif->num_queues;          \
    ERL_NIF_TERM reply = async_nif_enqueue_req(async_nif, req, h);     \
-    if (!reply) {                                                      \
-      fn_post_ ## decl (args);                                         \
+    if (!reply) {						       \
+      fn_post_##decl (args);                                           \
      async_nif_recycle_req(req, async_nif);                           \
      free(copy_of_args);					       \
      return enif_make_tuple2(env, ATOM_ERROR, ATOM_EAGAIN);	       \
@ -189,30 +190,38 @@ struct async_nif_state {
 struct async_nif_req_entry *
 async_nif_reuse_req(struct async_nif_state *async_nif)
 {
+    struct cds_lfq_node_rcu *node;
    struct async_nif_req_entry *req = NULL;
    ErlNifEnv *env = NULL;

-    enif_mutex_lock(async_nif->recycled_req_mutex);
-    if (STAILQ_EMPTY(&async_nif->recycled_reqs)) {
-        if (async_nif->num_reqs < ASYNC_NIF_MAX_QUEUED_REQS) {
+    /* Look for a request on our Lock-Free/RCU Queue first. */
+    rcu_read_lock();
+    node = cds_lfq_dequeue_rcu(&async_nif->recycled_req_queue);
+    rcu_read_unlock();
+
+    if (node) {
+        /* The goal is to reuse these req structs, not malloc/free them
+           repeatedly so we don't `call_rcu(&async_nif->rcu, free_req_cb);`.
+           We reuse this req, then when exiting we'll free all of them at
+           once. */
+        req = caa_container_of(node, struct async_nif_req_entry, queue_entry);
+    } else {
+        /* The reuse queue is empty, create a new request. */
+        if (uatomic_read(&async_nif->num_reqs) < ASYNC_NIF_MAX_QUEUED_REQS) {
            req = malloc(sizeof(struct async_nif_req_entry));
            if (req) {
                memset(req, 0, sizeof(struct async_nif_req_entry));
                env = enif_alloc_env();
                if (env) {
                    req->env = env;
-                    __sync_fetch_and_add(&async_nif->num_reqs, 1);
+                    uatomic_inc(&async_nif->num_reqs);
                } else {
                    free(req);
                    req = NULL;
                }
            }
        }
-    } else {
-        req = STAILQ_FIRST(&async_nif->recycled_reqs);
-        STAILQ_REMOVE(&async_nif->recycled_reqs, req, async_nif_req_entry, entries);
    }
-    enif_mutex_unlock(async_nif->recycled_req_mutex);
    return req;
 }

@ -226,14 +235,21 @@ async_nif_reuse_req(struct async_nif_state *async_nif)
 void
 async_nif_recycle_req(struct async_nif_req_entry *req, struct async_nif_state *async_nif)
 {
-    ErlNifEnv *env = NULL;
-    enif_mutex_lock(async_nif->recycled_req_mutex);
+    /* Three things to do here to prepare this request struct for reuse.
+       1) clear the NIF Environment
+       2) zero out the req struct except...
+       3) keep a pointer to the env so we can reset it in the req */
+    ErlNifEnv *env = req->env;
    enif_clear_env(req->env);
-    env = req->env;
+    if (req->args) free(req->args);
    memset(req, 0, sizeof(struct async_nif_req_entry));
    req->env = env;
-    STAILQ_INSERT_TAIL(&async_nif->recycled_reqs, req, entries);
-    enif_mutex_unlock(async_nif->recycled_req_mutex);
+
+    /* Now enqueue this request on our Lock-Free/RCU Queue to be reused later. */
+    cds_lfq_node_init_rcu(&req->queue_entry);
+    rcu_read_lock();
+    cds_lfq_enqueue_rcu(&async_nif->recycled_req_queue, &req->queue_entry);
+    rcu_read_unlock();
 }

 static void *async_nif_worker_fn(void *);
@ -244,41 +260,38 @@ static void *async_nif_worker_fn(void *);
 static int
 async_nif_start_worker(struct async_nif_state *async_nif, struct async_nif_work_queue *q)
 {
-  struct async_nif_worker_entry *we;
+  struct async_nif_worker_entry *worker;

  if (0 == q)
      return EINVAL;

-  enif_mutex_lock(async_nif->we_mutex);
+  /* Before creating a new worker thread join threads which have exited. */
+  for(;;) {
+      struct cds_lfq_node_rcu *node;

-  we = SLIST_FIRST(&async_nif->we_joining);
-  while(we != NULL) {
-    struct async_nif_worker_entry *n = SLIST_NEXT(we, entries);
-    SLIST_REMOVE(&async_nif->we_joining, we, async_nif_worker_entry, entries);
-    void *exit_value = 0; /* We ignore the thread_join's exit value. */
-    enif_thread_join(we->tid, &exit_value);
-    free(we);
-    async_nif->we_active--;
-    we = n;
+      rcu_read_lock();
+      node = cds_lfq_dequeue_rcu(&async_nif->worker_join_queue);
+      rcu_read_unlock();
+
+      if (!node) break; /* Queue is empty. */
+
+      worker = caa_container_of(node, struct async_nif_worker_entry, queue_entry);
+      void *exit_value = 0; /* We ignore the thread_join's exit value. */
+      enif_thread_join(worker->tid, &exit_value);
+      free(worker);
+      uatomic_dec(&async_nif->num_active_workers);
  }

-  if (async_nif->we_active == ASYNC_NIF_MAX_WORKERS) {
-      enif_mutex_unlock(async_nif->we_mutex);
+  if (uatomic_read(&async_nif->num_active_workers) >= ASYNC_NIF_MAX_WORKERS)
      return EAGAIN;
-  }

-  we = malloc(sizeof(struct async_nif_worker_entry));
-  if (!we) {
-      enif_mutex_unlock(async_nif->we_mutex);
-      return ENOMEM;
-  }
-  memset(we, 0, sizeof(struct async_nif_worker_entry));
-  we->worker_id = async_nif->we_active++;
-  we->async_nif = async_nif;
-  we->q = q;
-
-  enif_mutex_unlock(async_nif->we_mutex);
-  return enif_thread_create(NULL,&we->tid, &async_nif_worker_fn, (void*)we, 0);
+  worker = malloc(sizeof(struct async_nif_worker_entry));
+  if (!worker) return ENOMEM;
+  memset(worker, 0, sizeof(struct async_nif_worker_entry));
+  worker->worker_id = uatomic_add_return(&async_nif->num_active_workers, 1);
+  worker->async_nif = async_nif;
+  worker->q = q;
+  return enif_thread_create(NULL,&worker->tid, &async_nif_worker_fn, (void*)worker, 0);
 }

 /**
@ -295,6 +308,8 @@ async_nif_enqueue_req(struct async_nif_state* async_nif, struct async_nif_req_en
  struct async_nif_work_queue *q = NULL;
  double avg_depth = 0.0;

+  //rcu_register_thread();
+
  /* Either we're choosing a queue based on some affinity/hinted value or we
     need to select the next queue in the rotation and atomically update that
     global value (next_q is shared across worker threads) . */
@ -312,59 +327,58 @@ async_nif_enqueue_req(struct async_nif_state* async_nif, struct async_nif_req_en
  for (i = 0; i < async_nif->num_queues; i++) {
      /* Compute the average queue depth not counting queues which are empty or
         the queue we're considering right now. */
-      unsigned int j, n = 0;
+      unsigned int j, d, n = 0;
      for (j = 0; j < async_nif->num_queues; j++) {
-          if (j != qid && async_nif->queues[j].depth != 0) {
+          d = uatomic_read(&async_nif->queues[j].depth);
+          if (j != qid && d != 0) {
              n++;
-              avg_depth += async_nif->queues[j].depth;
+              avg_depth += d;
          }
      }
      if (avg_depth) avg_depth /= n;
-
-      /* Lock this queue under consideration, then check for shutdown.  While
-         we hold this lock either a) we're shutting down so exit now or b) this
-         queue will be valid until we release the lock. */
      q = &async_nif->queues[qid];
-      enif_mutex_lock(q->reqs_mutex);

      /* Try not to enqueue a request into a queue that isn't keeping up with
         the request volume. */
-      if (q->depth <= avg_depth) break;
-      else {
-          enif_mutex_unlock(q->reqs_mutex);
-          qid = (qid + 1) % async_nif->num_queues;
-      }
+      if (uatomic_read(&q->depth) <= avg_depth) break;
+      else qid = (qid + 1) % async_nif->num_queues;
  }

  /* If the for loop finished then we didn't find a suitable queue for this
-     request, meaning we're backed up so trigger eagain.  Note that if we left
-     the loop in this way we hold no lock. */
-  if (i == async_nif->num_queues) return 0;
+     request (e.g. we're backed up servicing requests) or the shutdown flag was
+     set. Returning '0' will toss this request and free its resources.*/
+  if (i == async_nif->num_queues || uatomic_read(&async_nif->shutdown))
+      return 0;

  /* Add the request to the queue. */
-  STAILQ_INSERT_TAIL(&q->reqs, req, entries);
-  __sync_fetch_and_add(&q->depth, 1);
+  cds_lfq_node_init_rcu(&req->queue_entry);
+  rcu_read_lock();
+  cds_lfq_enqueue_rcu(&q->req_queue, &req->queue_entry);
+  rcu_read_unlock();
+  URCU_TLS(nr_enqueues)++;
+  uatomic_inc(&q->depth);
+  uatomic_inc(&q->num_workers);

  /* We've selected a queue for this new request now check to make sure there are
     enough workers actively processing requests on this queue. */
-  while (q->depth > q->num_workers) {
+  while (uatomic_read(&q->depth) > uatomic_read(&q->num_workers)) {
      switch(async_nif_start_worker(async_nif, q)) {
-      case EINVAL: case ENOMEM: default: return 0;
-      case EAGAIN: continue;
-      case 0:      __sync_fetch_and_add(&q->num_workers, 1); goto done;
+        case EINVAL:
+        case ENOMEM:
+	    return 0;
+        default:
+        case EAGAIN:
+	    continue;
+        case 0:
+	    uatomic_inc(&q->num_workers);
+	    goto done;
      }
  }done:;

-  /* Build the term before releasing the lock so as not to race on the use of
-     the req pointer (which will soon become invalid in another thread
-     performing the request). */
  double pct_full = (double)avg_depth / (double)ASYNC_NIF_WORKER_QUEUE_SIZE;
-  ERL_NIF_TERM reply = enif_make_tuple2(req->env, ATOM_OK,
-					enif_make_tuple2(req->env, ATOM_ENQUEUED,
-							 enif_make_double(req->env, pct_full)));
-  enif_cond_signal(q->reqs_cnd);
-  enif_mutex_unlock(q->reqs_mutex);
-  return reply;
+  return enif_make_tuple2(req->env, ATOM_OK,
+			  enif_make_tuple2(req->env, ATOM_ENQUEUED,
+					   enif_make_double(req->env, pct_full)));
 }

 /**
@ -375,73 +389,46 @@ async_nif_enqueue_req(struct async_nif_state* async_nif, struct async_nif_req_en
 static void *
 async_nif_worker_fn(void *arg)
 {
-  struct async_nif_worker_entry *we = (struct async_nif_worker_entry *)arg;
-  unsigned int worker_id = we->worker_id;
-  struct async_nif_state *async_nif = we->async_nif;
-  struct async_nif_work_queue *q = we->q;
-  struct async_nif_req_entry *req = NULL;
-  unsigned int tries = async_nif->num_queues;
+  struct async_nif_worker_entry *worker = (struct async_nif_worker_entry *)arg;
+  unsigned int worker_id = worker->worker_id;
+  struct async_nif_state *async_nif = worker->async_nif;
+  struct async_nif_work_queue *l = NULL, *q = worker->q;

-  for(;;) {
-    /* Examine the request queue, are there things to be done? */
-    enif_mutex_lock(q->reqs_mutex);
-    check_again_for_work:
-    if (async_nif->shutdown) {
-        enif_mutex_unlock(q->reqs_mutex);
+  // TODO(gburd): set_affinity(); to the CPU_ID for this queue
+  rcu_register_thread();
+
+  while(q != l) {
+    struct cds_lfq_node_rcu *node;
+    struct async_nif_req_entry *req = NULL;
+
+    if (uatomic_read(&async_nif->shutdown))
        break;
-    }
-    if (STAILQ_EMPTY(&q->reqs)) {
-      /* Queue is empty so we wait for more work to arrive. */
-	enif_mutex_unlock(q->reqs_mutex);
-	if (tries == 0 && q == we->q) {
-	    if (q->num_workers > ASYNC_NIF_MIN_WORKERS) {
-		/* At this point we've tried to find/execute work on all queues
-		 * and there are at least MIN_WORKERS on this queue so we
-		 * leaving this loop (break) which leads to a thread exit/join. */
-		break;
-	    } else {
-		enif_mutex_lock(q->reqs_mutex);
-		enif_cond_wait(q->reqs_cnd, q->reqs_mutex);
-		goto check_again_for_work;
-	    }
-	} else {
-	    tries--;
-	    __sync_fetch_and_add(&q->num_workers, -1);
-	    q = q->next;
-	    __sync_fetch_and_add(&q->num_workers, 1);
-	    continue; // try next queue
-	}
+
+    rcu_read_lock();
+    node = cds_lfq_dequeue_rcu(&q->req_queue);
+    rcu_read_unlock();
+
+    if (node) {
+	req = caa_container_of(node, struct async_nif_req_entry, queue_entry);
+        uatomic_dec(&q->depth);
+        URCU_TLS(nr_dequeues)++;
+        req->fn_work(req->env, req->ref, &req->pid, worker_id, req->args);
+        req->fn_post(req->args);
+        async_nif_recycle_req(req, async_nif);
+        l = q;
    } else {
-      /* At this point the next req is ours to process and we hold the
-         reqs_mutex lock.  Take the request off the queue. */
-      req = STAILQ_FIRST(&q->reqs);
-      STAILQ_REMOVE(&q->reqs, req, async_nif_req_entry, entries);
-      __sync_fetch_and_add(&q->depth, -1);
-
-      /* Wake up other worker thread watching this queue to help process work. */
-      enif_cond_signal(q->reqs_cnd);
-      enif_mutex_unlock(q->reqs_mutex);
-
-      /* Perform the work. */
-      req->fn_work(req->env, req->ref, &req->pid, worker_id, req->args);
-
-      /* Now call the post-work cleanup function. */
-      req->fn_post(req->args);
-
-      /* Clean up req for reuse. */
-      req->ref = 0;
-      req->fn_work = 0;
-      req->fn_post = 0;
-      free(req->args);
-      req->args = NULL;
-      async_nif_recycle_req(req, async_nif);
-      req = NULL;
+        /* This queue is empty, cycle through other queues looking for work. */
+        uatomic_dec(&q->num_workers);
+        q = q->next;
+        uatomic_inc(&q->num_workers);
    }
  }
-  enif_mutex_lock(async_nif->we_mutex);
-  SLIST_INSERT_HEAD(&async_nif->we_joining, we, entries);
-  enif_mutex_unlock(async_nif->we_mutex);
-  __sync_fetch_and_add(&q->num_workers, -1);
+  uatomic_dec(&q->num_workers);
+  cds_lfq_node_init_rcu(&worker->queue_entry);
+  rcu_read_lock();
+  cds_lfq_enqueue_rcu(&async_nif->worker_join_queue, &worker->queue_entry);
+  rcu_read_unlock();
+  rcu_unregister_thread();
  enif_thread_exit(0);
  return 0;
 }
@ -451,83 +438,74 @@ async_nif_unload(ErlNifEnv *env, struct async_nif_state *async_nif)
 {
  unsigned int i;
  unsigned int num_queues = async_nif->num_queues;
+  struct cds_lfq_node_rcu *node;
  struct async_nif_work_queue *q = NULL;
-  struct async_nif_req_entry *req = NULL;
-  struct async_nif_worker_entry *we = NULL;
  UNUSED(env);

-  /* Signal the worker threads, stop what you're doing and exit.  To ensure
-     that we don't race with the enqueue() process we first lock all the worker
-     queues, then set shutdown to true, then unlock.  The enqueue function will
-     take the queue mutex, then test for shutdown condition, then enqueue only
-     if not shutting down. */
-  for (i = 0; i < num_queues; i++) {
-      q = &async_nif->queues[i];
-      enif_mutex_lock(q->reqs_mutex);
-  }
-  /* Set the shutdown flag so that worker threads will no continue
-     executing requests. */
-  async_nif->shutdown = 1;
-  for (i = 0; i < num_queues; i++) {
-      q = &async_nif->queues[i];
-      enif_mutex_unlock(q->reqs_mutex);
-  }
+  /* Signal the worker threads, stop what you're doing and exit. */
+  uatomic_set(&async_nif->shutdown, 1);

  /* Join for the now exiting worker threads. */
-  while(async_nif->we_active > 0) {
-      for (i = 0; i < num_queues; i++)
-          enif_cond_broadcast(async_nif->queues[i].reqs_cnd);
-      enif_mutex_lock(async_nif->we_mutex);
-      we = SLIST_FIRST(&async_nif->we_joining);
-      while(we != NULL) {
-          struct async_nif_worker_entry *n = SLIST_NEXT(we, entries);
-          SLIST_REMOVE(&async_nif->we_joining, we, async_nif_worker_entry, entries);
+  while(uatomic_read(&async_nif->num_active_workers) > 0) {
+      struct async_nif_worker_entry *worker;
+      struct cds_lfq_node_rcu *node;
+
+      rcu_read_lock();
+      node = cds_lfq_dequeue_rcu(&async_nif->worker_join_queue);
+      rcu_read_unlock();
+
+      if (node) {
+	  worker = caa_container_of(node, struct async_nif_worker_entry, queue_entry);
          void *exit_value = 0; /* We ignore the thread_join's exit value. */
-          enif_thread_join(we->tid, &exit_value);
-          free(we);
-          async_nif->we_active--;
-          we = n;
+          enif_thread_join(worker->tid, &exit_value);
+          free(worker);
+          uatomic_dec(&async_nif->num_active_workers);
      }
-      enif_mutex_unlock(async_nif->we_mutex);
  }
-  enif_mutex_destroy(async_nif->we_mutex);
+  cds_lfq_destroy_rcu(&async_nif->worker_join_queue); // TODO(gburd): check return val

  /* Cleanup in-flight requests, mutexes and conditions in each work queue. */
  for (i = 0; i < num_queues; i++) {
      q = &async_nif->queues[i];

      /* Worker threads are stopped, now toss anything left in the queue. */
-      req = NULL;
-      req = STAILQ_FIRST(&q->reqs);
-      while(req != NULL) {
-          struct async_nif_req_entry *n = STAILQ_NEXT(req, entries);
-          enif_clear_env(req->env);
-          enif_send(NULL, &req->pid, req->env,
-		    enif_make_tuple2(req->env, ATOM_ERROR, ATOM_SHUTDOWN));
-          req->fn_post(req->args);
-          enif_free_env(req->env);
-          free(req->args);
-          free(req);
-          req = n;
+      for (;;) {
+	  rcu_read_lock();
+          node = cds_lfq_dequeue_rcu(&q->req_queue);
+	  rcu_read_unlock();
+
+          if (!node) break; /* Queue is empty. */
+
+	  struct async_nif_req_entry *req;
+	  req = caa_container_of(node, struct async_nif_req_entry, queue_entry);
+	  enif_clear_env(req->env);
+	  enif_send(NULL, &req->pid, req->env, enif_make_tuple2(req->env, ATOM_ERROR, ATOM_SHUTDOWN));
+	  req->fn_post(req->args);
+	  free(req->args);
+	  enif_free_env(req->env);
+	  free(req);
      }
-      enif_mutex_destroy(q->reqs_mutex);
-      enif_cond_destroy(q->reqs_cnd);
+      cds_lfq_destroy_rcu(&q->req_queue); // TODO(gburd): check return val
  }

  /* Free any req structures sitting unused on the recycle queue. */
-  enif_mutex_lock(async_nif->recycled_req_mutex);
-  req = NULL;
-  req = STAILQ_FIRST(&async_nif->recycled_reqs);
-  while(req != NULL) {
-      struct async_nif_req_entry *n = STAILQ_NEXT(req, entries);
-      enif_free_env(req->env);
-      free(req);
-      req = n;
-  }
+  for (;;) {
+      rcu_read_lock();
+      node = cds_lfq_dequeue_rcu(&async_nif->recycled_req_queue);
+      rcu_read_unlock();
+
+      if (!node) break; /* Queue is empty. */
+
+      struct async_nif_req_entry *req;
+      req = caa_container_of(node, struct async_nif_req_entry, queue_entry);
+      enif_free_env(req->env);
+      free(req->args);
+      free(req);
+  }
+  cds_lfq_destroy_rcu(&async_nif->recycled_req_queue);  // TODO(gburd): check return val

-  enif_mutex_unlock(async_nif->recycled_req_mutex);
-  enif_mutex_destroy(async_nif->recycled_req_mutex);
  memset(async_nif, 0, sizeof(struct async_nif_state) + (sizeof(struct async_nif_work_queue) * async_nif->num_queues));
+  free_all_cpu_call_rcu_data();
  free(async_nif);
 }

@ -551,6 +529,10 @@ async_nif_load(ErlNifEnv *env)
  ATOM_OK = enif_make_atom(env, "ok");
  ATOM_SHUTDOWN = enif_make_atom(env, "shutdown");

+  /* Init the RCU library. */
+  rcu_init();
+  (void)create_all_cpu_call_rcu_data(0);
+
  /* Find out how many schedulers there are. */
  enif_system_info(&info, sizeof(ErlNifSysInfo));

@ -566,6 +548,7 @@ async_nif_load(ErlNifEnv *env)
      if (num_queues < 2)
          num_queues = 2;
  }
+  num_queues = 1; //TODO remove me.

  /* Init our portion of priv_data's module-specific state. */
  async_nif = malloc(sizeof(struct async_nif_state) +
@ -576,19 +559,15 @@ async_nif_load(ErlNifEnv *env)
                       sizeof(struct async_nif_work_queue) * num_queues);

  async_nif->num_queues = num_queues;
-  async_nif->we_active = 0;
+  async_nif->num_active_workers = 0;
  async_nif->next_q = 0;
  async_nif->shutdown = 0;
-  STAILQ_INIT(&async_nif->recycled_reqs);
-  async_nif->recycled_req_mutex = enif_mutex_create("recycled_req");
-  async_nif->we_mutex = enif_mutex_create("we");
-  SLIST_INIT(&async_nif->we_joining);
+  cds_lfq_init_rcu(&async_nif->recycled_req_queue, call_rcu);
+  cds_lfq_init_rcu(&async_nif->worker_join_queue, call_rcu);

  for (i = 0; i < async_nif->num_queues; i++) {
      struct async_nif_work_queue *q = &async_nif->queues[i];
-      STAILQ_INIT(&q->reqs);
-      q->reqs_mutex = enif_mutex_create("reqs");
-      q->reqs_cnd = enif_cond_create("reqs");
+      cds_lfq_init_rcu(&q->req_queue, call_rcu);
      q->next = &async_nif->queues[(i + 1) % num_queues];
  }
  return async_nif;
--- a/c_src/build_deps.sh
+++ b/c_src/build_deps.sh
@ -1,5 +1,8 @@
 #!/bin/bash

+#t=__.$$
+#trap 'rm -f $t; exit 0' 0 1 2 3 13 15
+
 # /bin/sh on Solaris is not a POSIX compatible shell, but /usr/bin/ksh is.
 if [ `uname -s` = 'SunOS' -a "${POSIX_SHELL}" != "true" ]; then
    POSIX_SHELL="true"
@ -10,15 +13,22 @@ unset POSIX_SHELL # clear it so if we invoke other scripts, they run as ksh as w

 set -e

+# WiredTiger
 WT_REPO=http://github.com/wiredtiger/wiredtiger.git
-WT_BRANCH=develop
-WT_DIR=wiredtiger-`basename $WT_BRANCH`
-#WT_REF="tags/1.6.6"
-#WT_DIR=wiredtiger-`basename $WT_REF`
+#WT_BRANCH=develop
+#WT_DIR=wiredtiger-`basename $WT_BRANCH`
+WT_REF="tags/1.6.4"
+WT_DIR=wiredtiger-`basename $WT_REF`

+# Google's Snappy Compression
 SNAPPY_VSN="1.0.4"
 SNAPPY_DIR=snappy-$SNAPPY_VSN

+# User-space Read-Copy-Update (RCU)
+URCU_REPO=git://git.lttng.org/userspace-rcu.git
+URCU_REF="tags/v0.7.7"
+URCU_DIR=urcu-`basename $URCU_REF`
+
 [ `basename $PWD` != "c_src" ] && cd c_src

 export BASEDIR="$PWD"
@ -26,10 +36,41 @@ export BASEDIR="$PWD"
 which gmake 1>/dev/null 2>/dev/null && MAKE=gmake
 MAKE=${MAKE:-make}

-export CPPFLAGS="$CPPLAGS -I $BASEDIR/system/include -O3 -mtune=native -march=native"
+export CFLAGS="$CFLAGS -I $BASEDIR/system/include"
+export CXXFLAGS="$CXXFLAGS -I $BASEDIR/system/include"
 export LDFLAGS="$LDFLAGS -L$BASEDIR/system/lib"
 export LD_LIBRARY_PATH="$LD_LIBRARY_PATH:$BASEDIR/system/lib:$LD_LIBRARY_PATH"

+get_urcu ()
+{
+    if [ -d $BASEDIR/$URCU_DIR/.git ]; then
+        (cd $BASEDIR/$URCU_DIR && git pull -u) || exit 1
+    else
+        if [ "X$URCU_REF" != "X" ]; then
+            git clone ${URCU_REPO} ${URCU_DIR} && \
+                (cd $BASEDIR/$URCU_DIR && git checkout refs/$URCU_REF || exit 1)
+        else
+            git clone ${URCU_REPO} ${URCU_DIR} && \
+                (cd $BASEDIR/$URCU_DIR && git checkout -b $URCU_BRANCH origin/$URCU_BRANCH || exit 1)
+        fi
+    fi
+    [ -d $BASEDIR/$URCU_DIR ] || (echo "Missing WiredTiger source directory" && exit 1)
+    (cd $BASEDIR/$URCU_DIR
+        [ -e $BASEDIR/urcu-build.patch ] && \
+            (patch -p1 --forward < $BASEDIR/urcu-build.patch || exit 1 )
+        autoreconf -fis || exit 1
+        urcu_configure;
+    )
+}
+
+urcu_configure ()
+{
+    (cd $BASEDIR/$URCU_DIR
+	LDFLAGS+="-Wl,-rpath,lib/urcu-v0.7.7/priv:lib/urcu/priv:priv" \
+	CFLAGS+="-m64 -Os -g -march=native -mtune=native -fPIC" \
+            ./configure --prefix=${BASEDIR}/system || exit 1)
+}
+
 get_wt ()
 {
    if [ -d $BASEDIR/$WT_DIR/.git ]; then
@ -57,8 +98,9 @@ get_wt ()
 wt_configure ()
 {
    (cd $BASEDIR/$WT_DIR/build_posix
-        CFLAGS+=-g $BASEDIR/$WT_DIR/configure --with-pic \
+        CFLAGS+=-g ../configure --with-pic \
                     --enable-snappy \
+                     --disable-python --disable-java \
                     --prefix=${BASEDIR}/system || exit 1)
 }

@ -75,8 +117,9 @@ get_snappy ()

 get_deps ()
 {
-    get_snappy;
    get_wt;
+    get_snappy;
+    get_urcu;
 }

 update_deps ()
@ -90,6 +133,22 @@ update_deps ()
            fi
        )
    fi
+
+    if [ -d $BASEDIR/$URCU_DIR/.git ]; then
+        (cd $BASEDIR/$URCU_DIR
+            if [ "X$URCU_VSN" == "X" ]; then
+                git pull -u || exit 1
+            else
+                git checkout $URCU_VSN || exit 1
+            fi
+        )
+    fi
+}
+
+build_urcu ()
+{
+    urcu_configure;
+    (cd $BASEDIR/$URCU_DIR && $MAKE -j && $MAKE install)
 }

 build_wt ()
@ -110,6 +169,7 @@ case "$1" in
    clean)
        [ -e $BASEDIR/$WT_DIR/build_posix/Makefile ] && \
            (cd $BASEDIR/$WT_DIR/build_posix && $MAKE clean)
+        [ -e $BASEDIR/$URCU_DIR/Makefile ] && (cd $BASEDIR/$URCU_DIR && $MAKE clean)
        rm -rf system $SNAPPY_DIR
        rm -f ${BASEDIR}/../priv/wt
        rm -f ${BASEDIR}/../priv/libwiredtiger-*.so
@ -130,23 +190,30 @@ case "$1" in
        ;;

    *)
-	shopt -s extglob
-	SUFFIXES='@(so|dylib)'
-
        # Build Snappy
        [ -d $SNAPPY_DIR ] || get_snappy;
        [ -d $BASEDIR/$SNAPPY_DIR ] || (echo "Missing Snappy source directory" && exit 1)
-        test -f $BASEDIR/system/lib/libsnappy.so.[0-9].[0-9].[0-9].* || build_snappy;
+        test -f $BASEDIR/system/lib/libsnappy.so.[0-9].[0-9].[0-9] || build_snappy;
+
+        # Build URCU
+        [ -d $URCU_DIR ] || get_urcu;
+        [ -d $BASEDIR/$URCU_DIR ] || (echo "Missing URCU source directory" && exit 1)
+        test -f $BASEDIR/system/lib/liburcu.a || build_urcu;

        # Build WiredTiger
        [ -d $WT_DIR ] || get_wt;
        [ -d $BASEDIR/$WT_DIR ] || (echo "Missing WiredTiger source directory" && exit 1)
-        test -f  $BASEDIR/system/lib/libwiredtiger-[0-9].[0-9].[0-9].${SUFFIXES} -a \
-	    -f $BASEDIR/system/lib/libwiredtiger_snappy.${SUFFIXES} || build_wt;
+        test -f $BASEDIR/system/lib/libwiredtiger-[0-9].[0-9].[0-9].so \
+             -a -f $BASEDIR/system/lib/libwiredtiger_snappy.so || build_wt;
+
        [ -d $BASEDIR/../priv ] || mkdir ${BASEDIR}/../priv
-	cp -p -P $BASEDIR/system/bin/wt ${BASEDIR}/../priv
-	cp -p -P ${BASEDIR}/system/lib/libwiredtiger-[0-9].[0-9].[0-9].${SUFFIXES} ${BASEDIR}/../priv
-	cp -p -P ${BASEDIR}/system/lib/libwiredtiger_snappy.${SUFFIXES} ${BASEDIR}/../priv
-	cp -p -P ${BASEDIR}/system/lib/libsnappy.${SUFFIXES}* ${BASEDIR}/../priv
+        cp -p -P $BASEDIR/system/bin/wt ${BASEDIR}/../priv
+        cp -p -P $BASEDIR/system/lib/libwiredtiger-[0-9].[0-9].[0-9].so ${BASEDIR}/../priv
+        cp -p -P $BASEDIR/system/lib/libwiredtiger_snappy.so* ${BASEDIR}/../priv
+        cp -p -P $BASEDIR/system/lib/libsnappy.so* ${BASEDIR}/../priv
+        cp -p -P $BASEDIR/system/lib/liburcu.so* ${BASEDIR}/../priv
+        cp -p -P $BASEDIR/system/lib/liburcu-*.so* ${BASEDIR}/../priv
        ;;
 esac
+
+exit 0
--- a/c_src/common.h
+++ b/c_src/common.h
@ -42,8 +42,8 @@ extern "C" {
 #define DPUTS(arg)		((void) 0)
 #endif

-#ifndef __UNUSED
-#define __UNUSED(v) ((void)(v))
+#ifndef UNUSED
+#define UNUSED(v) ((void)(v))
 #endif

 #ifndef COMPQUIET
--- a/c_src/urcu-build.patch
+++ b/c_src/urcu-build.patch
@ -0,0 +1,23 @@
+diff --git a/urcu/rculfhash.h b/urcu/rculfhash.h
+index 23bd1ce..98e00b4 100644
+--- a/urcu/rculfhash.h
+++ b/urcu/rculfhash.h
+@@ -31,6 +31,10 @@
+ #include <urcu-call-rcu.h>
+ #include <urcu-flavor.h>
+
+#ifndef UNUSED
+#define UNUSED(v) ((void)(v))
+#endif
+
+ #ifdef __cplusplus
+ extern "C" {
+ #endif
+@@ -92,6 +96,7 @@ typedef int (*cds_lfht_match_fct)(struct cds_lfht_node *node, const void *key);
+ static inline
+ void cds_lfht_node_init(struct cds_lfht_node *node)
+ {
+    UNUSED(node);
+ }
+
+ /*
--- a/c_src/wiredtiger-build.patch
+++ b/c_src/wiredtiger-build.patch
@ -1,12 +1,106 @@
 diff --git a/ext/compressors/snappy/Makefile.am b/ext/compressors/snappy/Makefile.am
-index 6d78823..c423590 100644
+index 6d78823..2122cf8 100644
 --- a/ext/compressors/snappy/Makefile.am
 +++ b/ext/compressors/snappy/Makefile.am
@@ -2,5 +2,6 @@ AM_CPPFLAGS = -I$(top_builddir) -I$(top_srcdir)/src/include
- 
+
 lib_LTLIBRARIES = libwiredtiger_snappy.la
 libwiredtiger_snappy_la_SOURCES = snappy_compress.c
 -libwiredtiger_snappy_la_LDFLAGS = -avoid-version -module
 +libwiredtiger_snappy_la_CFLAGS = -I$(abs_top_builddir)/../../system/include
 +libwiredtiger_snappy_la_LDFLAGS = -avoid-version -module -L$(abs_top_builddir)/../../system/lib -Wl,-rpath,lib/wterl-0.9.0/priv:lib/wterl/priv:priv
 libwiredtiger_snappy_la_LIBADD = -lsnappy
+diff --git a/src/support/cksum.c b/src/support/cksum.c
+index 7e9befe..b924db7 100644
+--- a/src/support/cksum.c
+++ b/src/support/cksum.c
+@@ -27,6 +27,13 @@
+
+ #include "wt_internal.h"
+
+#if defined(__amd64) || defined(__x86_64)
+#define USE_HARDWARE_CRC32 1
+#else
+#undef USE_HARDWARE_CRC32
+#endif
+
+#ifdef USE_HARDWARE_CRC32
+ static const uint32_t g_crc_slicing[8][256] = {
+ #ifdef WORDS_BIGENDIAN
+ 	/*
+@@ -1078,6 +1085,7 @@ static const uint32_t g_crc_slicing[8][256] = {
+ 	}
+ #endif
+ };
+#endif /* USE_HARDWARE_CRC32 */
+
+ /*
+  * __wt_cksum --
+@@ -1106,15 +1114,29 @@ __wt_cksum(const void *chunk, size_t len)
+ 	/* Checksum one byte at a time to the first 4B boundary. */
+ 	for (p = chunk;
+ 	    ((uintptr_t)p & (sizeof(uint32_t) - 1)) != 0 &&
+-	    len > 0; ++p, --len)
+	    len > 0; ++p, --len) {
+#ifdef USE_HARDWARE_CRC32
+	    __asm__ __volatile__(
+		".byte 0xF2, 0x0F, 0x38, 0xF0, 0xF1"
+		: "=S" (crc)
+		: "0" (crc), "c" (*p));
+#else
+ #ifdef WORDS_BIGENDIAN
+ 		crc = g_crc_slicing[0][((crc >> 24) ^ *p) & 0xFF] ^ (crc << 8);
+ #else
+ 		crc = g_crc_slicing[0][(crc ^ *p) & 0xFF] ^ (crc >> 8);
+ #endif
+#endif
+	}
+
+ 	/* Checksum in 8B chunks. */
+ 	for (nqwords = len / sizeof(uint64_t); nqwords; nqwords--) {
+#ifdef USE_HARDWARE_CRC32
+	    __asm__ __volatile__ (
+		".byte 0xf2, 0x48, 0x0f, 0x38, 0xf0, 0xf1;"
+		: "=S"(crc)
+		: "S"(crc), "c"(*p));
+#else
+ 		crc ^= *(uint32_t *)p;
+ 		p += sizeof(uint32_t);
+ 		next = *(uint32_t *)p;
+@@ -1139,22 +1161,32 @@ __wt_cksum(const void *chunk, size_t len)
+ 			g_crc_slicing[1][(next >> 16) & 0xFF] ^
+ 			g_crc_slicing[0][(next >> 24)];
+ #endif
+#endif
+ 	}
+
+ 	/* Checksum trailing bytes one byte at a time. */
+	for (len &= 0x7; len > 0; ++p, len--) {
+#ifdef USE_HARDWARE_CRC32
+	    __asm__ __volatile__(
+		".byte 0xF2, 0x0F, 0x38, 0xF0, 0xF1"
+		: "=S" (crc)
+		: "0" (crc), "c" (*p));
+#else
+ #ifdef WORDS_BIGENDIAN
+-	for (len &= 0x7; len > 0; ++p, len--)
+ 		crc = g_crc_slicing[0][((crc >> 24) ^ *p) & 0xFF] ^ (crc << 8);
+#else
+		crc = g_crc_slicing[0][(crc ^ *p) & 0xFF] ^ (crc >> 8);
+#endif
+#endif
+	}
+
+#ifdef WORDS_BIGENDIAN
+ 	/* Do final byte swap to produce a result identical to little endian */
+ 	crc =
+ 		((crc << 24) & 0xFF000000) |
+ 		((crc <<  8) & 0x00FF0000) |
+ 		((crc >>  8) & 0x0000FF00) |
+ 		((crc >> 24) & 0x000000FF);
+-#else
+-	for (len &= 0x7; len > 0; ++p, len--)
+-		crc = g_crc_slicing[0][(crc ^ *p) & 0xFF] ^ (crc >> 8);
+ #endif
+ 	return (~crc);
+ }
--- a/c_src/wterl.c
+++ b/c_src/wterl.c
@ -440,7 +440,6 @@ __close_cursors_on(WterlConnHandle *conn_handle, const char *uri)
    return;
 }

-
 /**
 * Callback to handle error messages.
 *
@ -455,15 +454,13 @@ __close_cursors_on(WterlConnHandle *conn_handle, const char *uri)
 *          operation or library failure.
 */
 int
-__wterl_error_handler(WT_EVENT_HANDLER *handler, WT_SESSION *session,
-    int error, const char *message)
+__wterl_error_handler(WT_EVENT_HANDLER *handler, int error, const char *message)
 {
    struct wterl_event_handlers *eh = (struct wterl_event_handlers *)handler;
    ErlNifEnv *msg_env;
    ErlNifPid *to_pid;
    int rc = 0;

-    UNUSED(session);
    enif_mutex_lock(eh->error_mutex);
    msg_env = eh->msg_env_error;
    to_pid = &eh->to_pid;
@ -495,14 +492,13 @@ __wterl_error_handler(WT_EVENT_HANDLER *handler, WT_SESSION *session,
 *          operation or library failure.
 */
 int
-__wterl_message_handler(WT_EVENT_HANDLER *handler, WT_SESSION *session, const char *message)
+__wterl_message_handler(WT_EVENT_HANDLER *handler, const char *message)
 {
    struct wterl_event_handlers *eh = (struct wterl_event_handlers *)handler;
    ErlNifEnv *msg_env;
    ErlNifPid *to_pid;
    int rc = 0;

-    UNUSED(session);
    enif_mutex_lock(eh->message_mutex);
    msg_env = eh->msg_env_message;
    to_pid = &eh->to_pid;
@ -533,14 +529,13 @@ __wterl_message_handler(WT_EVENT_HANDLER *handler, WT_SESSION *session, const ch
 *          operation or library failure.
 */
 int
-__wterl_progress_handler(WT_EVENT_HANDLER *handler, WT_SESSION *session, const char *operation, uint64_t counter)
+__wterl_progress_handler(WT_EVENT_HANDLER *handler, const char *operation, uint64_t counter)
 {
    struct wterl_event_handlers *eh = (struct wterl_event_handlers *)handler;
    ErlNifEnv *msg_env;
    ErlNifPid *to_pid;
    int rc = 0;

-    UNUSED(session);
    enif_mutex_lock(eh->progress_mutex);
    msg_env = eh->msg_env_progress;
    to_pid = &eh->to_pid;
@ -2308,7 +2303,7 @@ on_load(ErlNifEnv *env, void **priv_data, ERL_NIF_TERM load_info)

    char msg[1024];
    snprintf(msg, 1024, "NIF on_load complete (wterl version: %s, wiredtiger version: %s)", priv->wterl_vsn, priv->wiredtiger_vsn);
-    __wterl_message_handler((WT_EVENT_HANDLER *)&priv->eh, NULL, msg);
+    __wterl_message_handler((WT_EVENT_HANDLER *)&priv->eh, msg);
    return 0;
 }

--- a/priv/wterl.schema
+++ b/priv/wterl.schema
@ -1,6 +0,0 @@
-%%%% This is the WiredTiger section
-
-%% @doc wiredtiger data_root
-{mapping, "wiredtiger.data_root", "wterl.data_root", [
-  {default, "{{platform_data_dir}}/wiredtiger"}
-]}.
--- a/rebar.config
+++ b/rebar.config
@ -38,8 +38,8 @@
 {port_specs, [{"priv/wterl.so", ["c_src/*.c"]}]}.

 {port_env, [
-            {"DRV_CFLAGS",  "$DRV_CFLAGS  -O3 -mtune=native -march=native -fPIC -Wall -Wextra -Werror -I c_src/system/include"},
-            {"DRV_LDFLAGS", "$DRV_LDFLAGS -Wl,-rpath,lib/wterl/priv:lib/wterl-0.9.0/priv:priv -Lc_src/system/lib -lwiredtiger"}
+            {"DRV_CFLAGS",  "$DRV_CFLAGS -fPIC -Wall -Wextra -Werror -fdebug-cpp -ftrack-macro-expansion=2 -I c_src/system/include"},
+            {"DRV_LDFLAGS", "$DRV_LDFLAGS -Wl,-rpath,lib/wterl/priv:priv -Lc_src/system/lib -lurcu -lurcu-cds -lwiredtiger"}
           ]}.

 {pre_hooks, [{compile, "c_src/build_deps.sh compile"}]}.
--- a/src/riak_kv_wterl_backend.erl
+++ b/src/riak_kv_wterl_backend.erl
@ -22,7 +22,6 @@

 -module(riak_kv_wterl_backend).
 -behavior(temp_riak_kv_backend).
-compile([{parse_transform, lager_transform}]).

 %% KV Backend API
 -export([api_version/0,
@ -43,7 +42,7 @@

 -ifdef(TEST).
 -include_lib("eunit/include/eunit.hrl").
-compile(export_all).
+-compiel(export_all).
 -endif.

 -define(API_VERSION, 1).
@ -120,14 +119,14 @@ start(Partition, Config) ->
                    "lsm" ->
                        [{internal_page_max, "128K"},
                         {leaf_page_max, "16K"},
-                         {lsm, [
-                             {bloom_config, [{leaf_page_max, "8MB"}]},
-                             {bloom_bit_count, 28},
-                             {bloom_hash_count, 19},
-                             {bloom_oldest, true},
-                             {chunk_size, "100MB"},
-                             {merge_threads, 2}
-                         ]}
+                         {lsm_chunk_size, "100MB"},
+                         {lsm_merge_threads, 2},
+                         {prefix_compression, true},
+                         {lsm_bloom_newest, true},
+                         {lsm_bloom_oldest, true} ,
+                         {lsm_bloom_bit_count, 28},
+                         {lsm_bloom_hash_count, 19},
+                         {lsm_bloom_config, [{leaf_page_max, "8MB"}]}
                        ] ++ Compressor;
                    "table" ->
                        Compressor
@ -342,23 +341,22 @@ is_empty(#state{connection=Connection, table=Table}) ->
 %% @doc Get the status information for this wterl backend
 -spec status(state()) -> [{atom(), term()}].
 status(#state{connection=Connection, table=Table}) ->
-    [].
-%%     case wterl:cursor_open(Connection, "statistics:" ++ Table, [{statistics_fast, true}]) of
-%%         {ok, Cursor} ->
-%% 	    TheStats =
-%% 		case fetch_status(Cursor) of
-%% 		    {ok, Stats} ->
-%% 			Stats;
-%% 		    {error, {eperm, _}} -> % TODO: review/fix this logic
-%% 			{ok, []};
-%% 		    _ ->
-%% 			{ok, []}
-%% 		end,
-%% 	    wterl:cursor_close(Cursor),
-%% 	    TheStats;
-%%         {error, Reason2} ->
-%%             {error, Reason2}
-%%    end.
+    case wterl:cursor_open(Connection, Table) of
+        {ok, Cursor} ->
+	    TheStats =
+		case fetch_status(Cursor) of
+		    {ok, Stats} ->
+			Stats;
+		    {error, {eperm, _}} -> % TODO: review/fix this logic
+			{ok, []};
+		    _ ->
+			{ok, []}
+		end,
+	    wterl:cursor_close(Cursor),
+	    TheStats;
+        {error, Reason2} ->
+            {error, Reason2}
+    end.

 %% @doc Register an asynchronous callback
 -spec callback(reference(), any(), state()) -> {ok, state()}.
@ -401,41 +399,30 @@ establish_connection(Config, Type) ->
            ok = filelib:ensure_dir(filename:join(DataRoot, "x")),

            %% WT Connection Options:
-	    LogSetting = app_helper:get_prop_or_env(log, Config, wterl, false),
+            %% NOTE: LSM auto-checkpoints, so we don't have too.
            CheckpointSetting =
                case Type =:= "lsm" of
                    true ->
-			case LogSetting of
-			    true ->
-				%% Turn checkpoints on if logging is on, checkpoints enable log archival.
-				app_helper:get_prop_or_env(checkpoint, Config, wterl, [{wait, 30}]);  % in seconds
-			    _ ->
-				[]
-			end;
+                        [];
                    false ->
-                        app_helper:get_prop_or_env(checkpoint, Config, wterl, [{wait, 30}])
+                        app_helper:get_prop_or_env(checkpoint, Config, wterl, [{wait, 10}])
                end,
            RequestedCacheSize = app_helper:get_prop_or_env(cache_size, Config, wterl),
            ConnectionOpts =
                orddict:from_list(
                  [ wterl:config_value(create, Config, true),
-                    wterl:config_value(checkpoint_sync, Config, false),
-                    wterl:config_value(transaction_sync, Config, "none"),
-                    wterl:config_value(log, Config, [{enabled, LogSetting}]),
-                    wterl:config_value(mmap, Config, false),
-                    wterl:config_value(checkpoint, Config, CheckpointSetting),
+                    wterl:config_value(sync, Config, false),
                    wterl:config_value(session_max, Config, max_sessions(Config)),
                    wterl:config_value(cache_size, Config, size_cache(RequestedCacheSize)),
-                    wterl:config_value(statistics, Config, [ "fast", "clear"]),
                    wterl:config_value(statistics_log, Config, [{wait, 600}]), % in seconds
                    wterl:config_value(verbose, Config, [ "salvage", "verify"
                         % Note: for some unknown reason, if you add these additional
                         % verbose flags Erlang SEGV's "size_object: bad tag for 0x80"
-                         % no idea why... you've been warned.
+                         % no idea why... yet... you've been warned.
                         %"block", "shared_cache", "reconcile", "evict", "lsm",
                         %"fileops", "read", "write", "readserver", "evictserver",
                         %"hazard", "mutex", "ckpt"
-                         ]) ] ++ proplists:get_value(wterl, Config, [])), % sec
+                         ]) ] ++ CheckpointSetting ++ proplists:get_value(wterl, Config, [])), % sec

            %% WT Session Options:
            SessionOpts = [{isolation, "snapshot"}],
@ -556,15 +543,15 @@ from_index_key(LKey) ->

 %% @private
 %% Return all status from wterl statistics cursor
-%% fetch_status(Cursor) ->
-%%    {ok, fetch_status(Cursor, wterl:cursor_next_value(Cursor), [])}.
-%% fetch_status(_Cursor, {error, _}, Acc) ->
-%%     lists:reverse(Acc);
-%% fetch_status(_Cursor, not_found, Acc) ->
-%%     lists:reverse(Acc);
-%% fetch_status(Cursor, {ok, Stat}, Acc) ->
-%%     [What,Val|_] = [binary_to_list(B) || B <- binary:split(Stat, [<<0>>], [global])],
-%%     fetch_status(Cursor, wterl:cursor_next_value(Cursor), [{What,Val}|Acc]).
+fetch_status(Cursor) ->
+   {ok, fetch_status(Cursor, wterl:cursor_next_value(Cursor), [])}.
+fetch_status(_Cursor, {error, _}, Acc) ->
+    lists:reverse(Acc);
+fetch_status(_Cursor, not_found, Acc) ->
+    lists:reverse(Acc);
+fetch_status(Cursor, {ok, Stat}, Acc) ->
+    [What,Val|_] = [binary_to_list(B) || B <- binary:split(Stat, [<<0>>], [global])],
+    fetch_status(Cursor, wterl:cursor_next_value(Cursor), [{What,Val}|Acc]).

 size_cache(RequestedSize) ->
    Size =
--- a/src/wterl.erl
+++ b/src/wterl.erl
@ -96,8 +96,8 @@ nif_stub_error(Line) ->
 -spec init() -> ok | {error, any()}.
 init() ->
    erlang:load_nif(filename:join([priv_dir(), atom_to_list(?MODULE)]),
-           [{wterl_vsn, "942e51b"},
-	    {wiredtiger_vsn, "1.6.4-275-g9c44420"}]). %% TODO automate these
+           [{wterl_vsn, "53307e8"},
+	    {wiredtiger_vsn, "1.6.2-0-g07cb0a5"}]).

 -spec connection_open(string(), config_list()) -> {ok, connection()} | {error, term()}.
 -spec connection_open(string(), config_list(), config_list()) -> {ok, connection()} | {error, term()}.
@ -454,26 +454,17 @@ config_to_bin([], Acc) ->
 config_to_bin([{Key, Value} | Rest], Acc) ->
    ConfigTypes =
    [{block_compressor, {string, quoted}},
-     {bloom_bit_count, integer},
-     {bloom_config, config},
-     {bloom_hash_count, integer},
-     {bloom_newest, bool},
-     {bloom_oldest, bool},
     {cache_size, string},
     {checkpoint, config},
-     {checkpoint_sync, bool},
     {checksum, string},
-     {chunk_size, string},
     {create, bool},
     {direct_io, list},
     {drop, list},
-     {enabled, bool},
     {error_prefix, string},
     {eviction_target, integer},
     {eviction_trigger, integer},
     {extensions, {list, quoted}},
     {statistics_fast, bool},
-     {file_max, string},
     {force, bool},
     {from, string},
     {hazard_max, integer},
@ -483,21 +474,24 @@ config_to_bin([{Key, Value} | Rest], Acc) ->
     {isolation, string},
     {key_type, string},
     {leaf_page_max, string},
-     {log, config},
-     {lsm, config},
-     {mmap, bool},
-     {merge_threads, integer},
+     {logging, bool},
+     {lsm_bloom_bit_count, integer},
+     {lsm_bloom_config, config},
+     {lsm_bloom_hash_count, integer},
+     {lsm_bloom_newest, bool},
+     {lsm_bloom_oldest, bool},
+     {lsm_chunk_size, string},
+     {prefix_compression, bool},
+     {lsm_merge_threads, integer},
     {multiprocess, bool},
     {name, string},
     {overwrite, bool},
-     {prefix_compression, bool},
     {raw, bool},
     {session_max, integer},
-     {statistics, list},
     {statistics_log, config},
+     {sync, bool},
     {target, {list, quoted}},
     {to, string},
-     {transaction_sync, string},
     {transactional, bool},
     {verbose, list},
     {wait, integer}],
--- a/tools/wterl-b_b.config
+++ b/tools/wterl-b_b.config
@ -25,7 +25,7 @@

 {mode, max}.
 {duration, 10}.
-{concurrent, 16}.
+{concurrent, 4}.
 {report_interval, 1}.
 {pb_timeout_general, 1000}. % ms
 %{pb_timeout_read, ?}.
@ -43,9 +43,7 @@
 {wterl, [
        {connection, [
                      {create, true},
-                      {session_sync, false},
-		      {transaction_sync, "none"},
-		      {log, [{enabled, false}]},
+                      {sync, false},
                      {session_max, 1024},
                      {cache_size, 4294967296},
                      {verbose, []},
@ -60,11 +58,11 @@
                     ]},
        {session, [ {isolation, "snapshot"} ]},
        {table_uri, "lsm:test"},
-	{lsm_merge_threads, 2},
        {table, [
                 {internal_page_max, "128K"},
                 {leaf_page_max, "128K"},
                 {lsm_chunk_size, "25MB"},
+                 {prefix_compression, false},
                 {lsm_bloom_newest, true},
                 {lsm_bloom_oldest, true} ,
                 {lsm_bloom_bit_count, 128},
@ -78,9 +76,7 @@
 {wterl_, [
        {connection, [
                      {create, true},
-                      {session_sync, false},
-		      {transaction_sync, "none"},
-		      {log, [{enabled, false}]},
+                      {sync, false},
                      {session_max, 1024},
                      {cache_size, 4294967296},
                      {verbose, []},
@ -97,6 +93,7 @@
        {session, [ {isolation, "snapshot"} ]},
        {table_uri, "table:test"},
        {table, [
+                 {prefix_compression, false},
                 {block_compressor, "snappy"} % bzip2
                ]}
        ]}.
Author	SHA1	Message	Date
Gregory Burd	08acdbefc8	Retry three times, then bail out and return not found.	2013-09-04 13:11:02 -04:00
Gregory Burd	033661875e	Trace line numbers within macros.	2013-09-04 13:10:16 -04:00
Gregory Burd	9496f99019	WIP: race condition on close causes double-free	2013-09-04 13:09:17 -04:00
Gregory Burd	f4a1126fde	Fix a few mistakes from the merge.	2013-08-26 13:07:46 -04:00
Gregory Burd	da418b4abf	Merge remote-tracking branch 'origin/master' into gsb-rcu Conflicts: c_src/async_nif.h c_src/build_deps.sh src/async_nif.hrl	2013-08-21 15:56:31 -04:00
Gregory Burd	a984a6dae0	Adding a small patch to URCU build process.	2013-08-13 08:54:22 -04:00
Gregory Burd	45037cbcca	Selective receive requires that the ref is created in the scope of the receive.	2013-07-30 12:53:33 -04:00
Gregory Burd	e9f9d13e47	WIP: replaced all locked queues in async_nif with lock-free RCU queues, compiles but SEGVs.	2013-07-28 20:16:52 -04:00
Gregory Burd	866b2a2ed1	Add the User-Space RCU library.	2013-07-27 08:21:40 -04:00