Robj/deadlock fix (#644)

This patch fixes two deadlocks, also enabling threads to perform updates to splinterdb while also performing asynchronous queries.

The first deadlock was caused by trunk_merge_lookup acquiring two locks on the root. This deadlock prone because, if another thread begins to GC the root, it will set the writer bit on the root lock and then wait for the readers to clear out. But, if the query code has already acquired its first read lock on the root and attempts to acquire its second read lock after the writer bit is set, then it will deadlock with the GC thread.

The fix is to acquire a lock on the root only once.

The second deadlock could occur when a thread performing asynchronous queries performs an update which results in a GC. In this case, one of the thread's async queries may hold a read lock on a node-to-be-GCed, but the GC code will wait forever for that read lock to be released, causing the thread to deadlock with itself.

The fix in this case is to defer GC until there are no more readers. So there's now a queue of GC tasks in the trunk context. Any time GC reaches a node with a read lock, it stops and puts that node on the queue. Every trunk modification checks the queue and processes any GCs that are now available to perform.

Author: rtjohnso

src/cache.h

@@ -129,7 +129,8 @@ typedef void (*extent_sync_fn)(cache  *cc,
                                uint64 *pages_outstanding);
 typedef void (*page_prefetch_fn)(cache *cc, uint64 addr, page_type type);
 typedef int (*evict_fn)(cache *cc, bool32 ignore_pinned);
-typedef void (*assert_ungot_fn)(cache *cc, uint64 addr);
+typedef bool32 (*page_addr_pred_fn)(cache *cc, uint64 addr);
+typedef void (*page_addr_fn)(cache *cc, uint64 addr);
 typedef void (*validate_page_fn)(cache *cc, page_handle *page, uint64 addr);
 typedef void (*io_stats_fn)(cache *cc, uint64 *read_bytes, uint64 *write_bytes);
 typedef uint32 (*count_dirty_fn)(cache *cc);
@@ -168,7 +169,8 @@ typedef struct cache_ops {
    cache_generic_fn     flush;
    evict_fn             evict;
    cache_generic_fn     cleanup;
-   assert_ungot_fn      assert_ungot;
+   page_addr_pred_fn    in_use;
+   page_addr_fn         assert_ungot;
    cache_generic_fn     assert_free;
    validate_page_fn     validate_page;
    cache_present_fn     cache_present;
@@ -552,6 +554,20 @@ cache_cleanup(cache *cc)
    return cc->ops->cleanup(cc);
 }
 
+/*
+ *-----------------------------------------------------------------------------
+ * cache_in_use
+ *
+ * Returns TRUE if there is a reader or writer holding a reference
+ * to the page at addr.
+ *-----------------------------------------------------------------------------
+ */
+static inline bool32
+cache_in_use(cache *cc, uint64 addr)
+{
+   return cc->ops->in_use(cc, addr);
+}
+
 /*
  *-----------------------------------------------------------------------------
  * cache_assert_ungot

src/clockcache.c

@@ -1480,7 +1480,7 @@ void
 clockcache_extent_discard(clockcache *cc, uint64 addr, page_type type)
 {
    debug_assert(addr % clockcache_extent_size(cc) == 0);
-   debug_assert(allocator_get_refcount(cc->al, addr) == 1);
+   debug_assert(allocator_get_refcount(cc->al, addr) == AL_NO_REFS);
 
    clockcache_log(addr, 0, "hard evict extent: addr %lu\n", addr);
    for (uint64 i = 0; i < cc->cfg->pages_per_extent; i++) {
@@ -1697,10 +1697,10 @@ clockcache_get_internal(clockcache   *cc,       // IN
    refcount extent_ref_count = allocator_get_refcount(cc->al, base_addr);
 
    // Dump allocated extents info for deeper debugging.
-   if (extent_ref_count <= 1) {
+   if (extent_ref_count == AL_FREE) {
       allocator_print_allocated(cc->al);
    }
-   debug_assert((extent_ref_count > 1),
+   debug_assert((extent_ref_count != AL_FREE),
                 "Attempt to get a buffer for page addr=%lu"
                 ", page type=%d ('%s'),"
                 " from extent addr=%lu, (extent number=%lu)"
@@ -1917,10 +1917,10 @@ clockcache_get_internal_async(clockcache_get_async_state *state, uint64 depth)
       allocator_get_refcount(state->cc->al, state->base_addr);
 
    // Dump allocated extents info for deeper debugging.
-   if (state->extent_ref_count <= 1) {
+   if (state->extent_ref_count == AL_FREE) {
       allocator_print_allocated(state->cc->al);
    }
-   debug_assert((state->extent_ref_count > 1),
+   debug_assert((state->extent_ref_count != AL_FREE),
                 "Attempt to get a buffer for page addr=%lu"
                 ", page type=%d ('%s'),"
                 " from extent addr=%lu, (extent number=%lu)"
@@ -2680,6 +2680,21 @@ clockcache_count_dirty(clockcache *cc)
    return dirty_count;
 }
 
+bool32
+clockcache_in_use(clockcache *cc, uint64 addr)
+{
+   uint32 entry_no = clockcache_lookup(cc, addr);
+   if (entry_no == CC_UNMAPPED_ENTRY) {
+      return FALSE;
+   }
+   for (threadid thr_i = 0; thr_i < CC_RC_WIDTH; thr_i++) {
+      if (clockcache_get_ref(cc, entry_no, thr_i) > 0) {
+         return TRUE;
+      }
+   }
+   return clockcache_test_flag(cc, entry_no, CC_WRITELOCKED);
+}
+
 uint16
 clockcache_get_read_ref(clockcache *cc, page_handle *page)
 {
@@ -2896,6 +2911,13 @@ clockcache_wait_virtual(cache *c)
    return clockcache_wait(cc);
 }
 
+bool32
+clockcache_in_use_virtual(cache *c, uint64 addr)
+{
+   clockcache *cc = (clockcache *)c;
+   return clockcache_in_use(cc, addr);
+}
+
 void
 clockcache_assert_ungot_virtual(cache *c, uint64 addr)
 {
@@ -3010,6 +3032,7 @@ static cache_ops clockcache_ops = {
    .flush             = clockcache_flush_virtual,
    .evict             = clockcache_evict_all_virtual,
    .cleanup           = clockcache_wait_virtual,
+   .in_use            = clockcache_in_use_virtual,
    .assert_ungot      = clockcache_assert_ungot_virtual,
    .assert_free       = clockcache_assert_no_locks_held_virtual,
    .print             = clockcache_print_virtual,

src/core.c

@@ -156,8 +156,11 @@ core_set_super_block(core_handle *spl,
    super                = (core_super_block *)super_page->data;
    uint64 old_root_addr = super->root_addr;
 
-   if (spl->trunk_context.root != NULL) {
-      super->root_addr = spl->trunk_context.root->addr;
+   trunk_ondisk_node_handle root_handle;
+   trunk_init_root_handle(&spl->trunk_context, &root_handle);
+   uint64 root_addr = trunk_ondisk_node_handle_addr(&root_handle);
+   if (root_addr != 0) {
+      super->root_addr = root_addr;
       rc               = trunk_inc_ref(spl->cfg.trunk_node_cfg,
                          spl->heap_id,
                          spl->cc,
@@ -169,6 +172,8 @@ core_set_super_block(core_handle *spl,
    } else {
       super->root_addr = 0;
    }
+   trunk_ondisk_node_handle_deinit(&root_handle);
+
    if (spl->cfg.use_log) {
       if (spl->log) {
          super->log_addr      = log_addr(spl->log);
@@ -1259,7 +1264,6 @@ core_lookup(core_handle *spl, key target, merge_accumulator *result)
 
    rc = trunk_merge_lookup(
       &spl->trunk_context, &root_handle, target, result, NULL);
-   // Release the node handle before handling any errors
    trunk_ondisk_node_handle_deinit(&root_handle);
    if (!SUCCESS(rc)) {
       return rc;
@@ -1339,10 +1343,8 @@ core_lookup_async(core_lookup_async_state *state)
                     NULL,
                     state->callback,
                     state->callback_arg);
-   rc = async_result(&state->trunk_node_state);
-
-   // Release the node handle before handling any errors
    trunk_ondisk_node_handle_deinit(&state->root_handle);
+   rc = async_result(&state->trunk_node_state);
    if (!SUCCESS(rc)) {
       async_return(state, rc);
    }