Robj/deadlock fix

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);
    }