rcuja: share lock across all nodes with same key

Signed-off-by: Mathieu Desnoyers <mathieu.desnoyers@efficios.com>

diff --git a/rcuja/rcuja-internal.h b/rcuja/rcuja-internal.h
index a5aae07dceeddf5b871772a98d0c1fa06b2390c8..80202ed0f9683eacaf3d888a512406e676f05f91 100644 (file)
--- a/rcuja/rcuja-internal.h
+++ b/rcuja/rcuja-internal.h
@@ -35,17 +35,22 @@ struct rcu_ja_node_flag;
 struct rcu_ja_shadow_node {
        struct cds_lfht_node ht_node;   /* hash table node */
        struct rcu_ja_node *node;       /* reverse mapping and hash table key */
-       pthread_mutex_t lock;           /* mutual exclusion on node */
+       /*
+        * mutual exclusion on all nodes belonging to the same tree
+        * position (e.g. both nodes before and after recompaction
+        * use the same lock).
+        */
+       pthread_mutex_t *lock;
+       unsigned int nr_child;          /* number of children in node */
        struct rcu_head head;           /* for deferred node and shadow node reclaim */
-       const struct rcu_flavor_struct *flavor; /* rcu flavor */
 };
 
 struct rcu_ja {
        struct rcu_ja_node_flag *root;
        /*
-        * We use a hash table to associate nodes to their respective
-        * shadow node. This helps reducing lookup hot path cache
-        * footprint, especially for very small nodes.
+        * We use a hash table to associate node keys to their
+        * respective shadow node. This helps reducing lookup hot path
+        * cache footprint, especially for very small nodes.
         */
        struct cds_lfht *ht;
 };
@@ -57,10 +62,19 @@ __attribute__((visibility("protected")))
 void rcuja_shadow_unlock(struct rcu_ja_shadow_node *shadow_node);
 __attribute__((visibility("protected")))
 int rcuja_shadow_set(struct cds_lfht *ht,
-               struct rcu_ja_node *node);
+               struct rcu_ja_node *new_node,
+               struct rcu_ja_shadow_node *inherit_from);
 __attribute__((visibility("protected")))
-int rcuja_shadow_clear_and_free_node(struct cds_lfht *ht,
-               struct rcu_ja_node *node);
+
+/* rcuja_shadow_clear flags */
+enum {
+       RCUJA_SHADOW_CLEAR_FREE_NODE = (1U << 0),
+       RCUJA_SHADOW_CLEAR_FREE_LOCK = (1U << 1),
+};
+
+int rcuja_shadow_clear(struct cds_lfht *ht,
+               struct rcu_ja_node *node,
+               unsigned int flags);
 __attribute__((visibility("protected")))
 struct cds_lfht *rcuja_create_ht(const struct rcu_flavor_struct *flavor);
 __attribute__((visibility("protected")))

diff --git a/rcuja/rcuja-shadow-nodes.c b/rcuja/rcuja-shadow-nodes.c
index cfc3aacebe311d5eccca6bc950957e25a57aa64a..5ef68d58631bd5d7a673496f78e755ce78b97a90 100644 (file)
--- a/rcuja/rcuja-shadow-nodes.c
+++ b/rcuja/rcuja-shadow-nodes.c
@@ -196,10 +196,10 @@ struct rcu_ja_shadow_node *rcuja_shadow_lookup_lock(struct cds_lfht *ht,
        }
        shadow_node = caa_container_of(lookup_node,
                        struct rcu_ja_shadow_node, ht_node);
-       ret = pthread_mutex_lock(&shadow_node->lock);
+       ret = pthread_mutex_lock(shadow_node->lock);
        assert(!ret);
        if (cds_lfht_is_node_deleted(lookup_node)) {
-               ret = pthread_mutex_unlock(&shadow_node->lock);
+               ret = pthread_mutex_unlock(shadow_node->lock);
                assert(!ret);
                shadow_node = NULL;
        }
@@ -213,13 +213,14 @@ void rcuja_shadow_unlock(struct rcu_ja_shadow_node *shadow_node)
 {
        int ret;
 
-       ret = pthread_mutex_unlock(&shadow_node->lock);
+       ret = pthread_mutex_unlock(shadow_node->lock);
        assert(!ret);
 }
 
 __attribute__((visibility("protected")))
 int rcuja_shadow_set(struct cds_lfht *ht,
-               struct rcu_ja_node *node)
+               struct rcu_ja_node *new_node,
+               struct rcu_ja_shadow_node *inherit_from)
 {
        struct rcu_ja_shadow_node *shadow_node;
        struct cds_lfht_node *ret_node;
@@ -229,15 +230,27 @@ int rcuja_shadow_set(struct cds_lfht *ht,
        if (!shadow_node)
                return -ENOMEM;
 
-       shadow_node->node = node;
-       pthread_mutex_init(&shadow_node->lock, NULL);
+       shadow_node->node = new_node;
+       /*
+        * Lock can be inherited from previous node at this position.
+        */
+       if (inherit_from) {
+               shadow_node->lock = inherit_from->lock;
+       } else {
+               shadow_node->lock = calloc(sizeof(*shadow_node->lock), 1);
+               if (!shadow_node->lock) {
+                       free(shadow_node);
+                       return -ENOMEM;
+               }
+               pthread_mutex_init(shadow_node->lock, NULL);
+       }
 
        flavor = cds_lfht_rcu_flavor(ht);
        flavor->read_lock();
        ret_node = cds_lfht_add_unique(ht,
-                       hash_pointer(node, hash_seed),
+                       hash_pointer(new_node, hash_seed),
                        match_pointer,
-                       node,
+                       new_node,
                        &shadow_node->ht_node);
        flavor->read_unlock();
 
@@ -257,9 +270,20 @@ void free_shadow_node_and_node(struct rcu_head *head)
        free(shadow_node);
 }
 
+static
+void free_shadow_node_and_node_and_lock(struct rcu_head *head)
+{
+       struct rcu_ja_shadow_node *shadow_node =
+               caa_container_of(head, struct rcu_ja_shadow_node, head);
+       free(shadow_node->node);
+       free(shadow_node->lock);
+       free(shadow_node);
+}
+
 __attribute__((visibility("protected")))
-int rcuja_shadow_clear_and_free_node(struct cds_lfht *ht,
-               struct rcu_ja_node *node)
+int rcuja_shadow_clear(struct cds_lfht *ht,
+               struct rcu_ja_node *node,
+               unsigned int flags)
 {
        struct cds_lfht_iter iter;
        struct cds_lfht_node *lookup_node;
@@ -278,7 +302,7 @@ int rcuja_shadow_clear_and_free_node(struct cds_lfht *ht,
        }
        shadow_node = caa_container_of(lookup_node,
                        struct rcu_ja_shadow_node, ht_node);
-       lockret = pthread_mutex_lock(&shadow_node->lock);
+       lockret = pthread_mutex_lock(shadow_node->lock);
        assert(!lockret);
 
        /*
@@ -288,9 +312,16 @@ int rcuja_shadow_clear_and_free_node(struct cds_lfht *ht,
         */
        ret = cds_lfht_del(ht, lookup_node);
        if (!ret) {
-               flavor->update_call_rcu(&shadow_node->head, free_shadow_node_and_node);
+               assert(flags & RCUJA_SHADOW_CLEAR_FREE_NODE);
+               if (flags & RCUJA_SHADOW_CLEAR_FREE_LOCK) {
+                       flavor->update_call_rcu(&shadow_node->head,
+                               free_shadow_node_and_node_and_lock);
+               } else {
+                       flavor->update_call_rcu(&shadow_node->head,
+                               free_shadow_node_and_node);
+               }
        }
-       lockret = pthread_mutex_unlock(&shadow_node->lock);
+       lockret = pthread_mutex_unlock(shadow_node->lock);
        assert(!lockret);
 rcu_unlock:
        flavor->read_unlock();

diff --git a/rcuja/rcuja.c b/rcuja/rcuja.c
index 4b90aa62b7ffb56f5e7659c21df4b4bdc234d643..45ce8f70a816e4cca4188ab5d9c9a54bceb02655 100644 (file)
--- a/rcuja/rcuja.c
+++ b/rcuja/rcuja.c
@@ -45,6 +45,8 @@ enum rcu_ja_type_class {
                        /* 64-bit: 113 to 256 children, 2048 bytes */
        /* Leaf nodes are implicit from their height in the tree */
        RCU_JA_NR_TYPES,
+
+       RCU_JA_NULL,    /* not an encoded type, but keeps code regular */
 };
 
 struct rcu_ja_type {
@@ -68,6 +70,12 @@ struct rcu_ja_type {
 
 #define JA_ENTRY_PER_NODE      256UL
 
+/*
+ * Entry for NULL node is at index 8 of the table. It is never encoded
+ * in flags.
+ */
+#define NODE_INDEX_NULL                8
+
 /*
  * Iteration on the array to find the right node size for the number of
  * children stops when it reaches .max_child == 256 (this is the largest
@@ -97,6 +105,7 @@ enum {
        ja_type_5_max_child = 48,
        ja_type_6_max_child = 92,
        ja_type_7_max_child = 256,
+       ja_type_8_max_child = 0,        /* NULL */
 };
 
 enum {
@@ -130,6 +139,8 @@ const struct rcu_ja_type ja_types[] = {
         * filled beyond capacity, we need to roll back to pigeon.
         */
        { .type_class = RCU_JA_PIGEON, .min_child = 89, .max_child = ja_type_7_max_child, .order = 10, },
+
+       { .type_class = RCU_JA_NULL, .min_child = 0, .max_child = ja_type_8_max_child, },
 };
 #else /* !(CAA_BITS_PER_LONG < 64) */
 /* 64-bit pointers */
@@ -142,6 +153,7 @@ enum {
        ja_type_5_max_child = 54,
        ja_type_6_max_child = 104,
        ja_type_7_max_child = 256,
+       ja_type_8_max_child = 256,
 };
 
 enum {
@@ -175,13 +187,15 @@ const struct rcu_ja_type ja_types[] = {
         * filled beyond capacity, we need to roll back to pigeon.
         */
        { .type_class = RCU_JA_PIGEON, .min_child = 101, .max_child = ja_type_7_max_child, .order = 11, },
+
+       { .type_class = RCU_JA_NULL, .min_child = 0, .max_child = ja_type_8_max_child, },
 };
 #endif /* !(BITS_PER_LONG < 64) */
 
 static inline __attribute__((unused))
 void static_array_size_check(void)
 {
-       CAA_BUILD_BUG_ON(CAA_ARRAY_SIZE(ja_types) > JA_TYPE_MAX_NR);
+       CAA_BUILD_BUG_ON(CAA_ARRAY_SIZE(ja_types) < JA_TYPE_MAX_NR);
 }
 
 /*
@@ -242,22 +256,25 @@ struct rcu_ja_node_flag *ja_node_flag(struct rcu_ja_node *node,
        return (struct rcu_ja_node_flag *) (((unsigned long) node) | type);
 }
 
+static
+struct rcu_ja_node *ja_node_ptr(struct rcu_ja_node_flag *node)
+{
+       return (struct rcu_ja_node *) (((unsigned long) node) | JA_PTR_MASK);
+}
+
 static
 unsigned int ja_node_type(struct rcu_ja_node_flag *node)
 {
        unsigned int type;
 
+       if (ja_node_ptr(node) == NULL) {
+               return NODE_INDEX_NULL;
+       }
        type = (unsigned int) ((unsigned long) node & JA_TYPE_MASK);
        assert(type < RCU_JA_NR_TYPES);
        return type;
 }
 
-static
-struct rcu_ja_node *ja_node_ptr(struct rcu_ja_node_flag *node)
-{
-       return (struct rcu_ja_node *) (((unsigned long) node) | JA_PTR_MASK);
-}
-
 struct rcu_ja_node *alloc_rcu_ja_node(const struct rcu_ja_type *ja_type)
 {
        return calloc(1U << ja_type->order, sizeof(char));
@@ -323,6 +340,10 @@ struct rcu_ja_node_flag *ja_pool_node_get_nth(const struct rcu_ja_type *type,
        struct rcu_ja_node *linear;
 
        assert(type->type_class == RCU_JA_POOL);
+       /*
+        * TODO: currently, we select the pool by highest bits. We
+        * should support various encodings.
+        */
        linear = (struct rcu_ja_node *)
                &node->u.data[((unsigned long) n >> (CHAR_BIT - type->nr_pool_order)) << type->pool_size_order];
        return ja_linear_node_get_nth(type, linear, n);
@@ -367,9 +388,16 @@ struct rcu_ja_node_flag *ja_node_get_nth(struct rcu_ja_node_flag *node_flag,
        }
 }
 
+static
+unsigned int ja_get_nr_child(struct rcu_ja_shadow_node *shadow_node)
+{
+       return shadow_node->nr_child;
+}
+
 static
 int ja_linear_node_set_nth(const struct rcu_ja_type *type,
                struct rcu_ja_node *node,
+               struct rcu_ja_shadow_node *shadow_node,
                uint8_t n,
                struct rcu_ja_node_flag *child_node_flag)
 {
@@ -400,12 +428,14 @@ int ja_linear_node_set_nth(const struct rcu_ja_type *type,
        CMM_STORE_SHARED(values[nr_child], n);
        cmm_smp_wmb();  /* write value and pointer before nr_child */
        CMM_STORE_SHARED(*nr_child_ptr, nr_child + 1);
+       shadow_node->nr_child++;
        return 0;
 }
 
 static
 int ja_pool_node_set_nth(const struct rcu_ja_type *type,
                struct rcu_ja_node *node,
+               struct rcu_ja_shadow_node *shadow_node,
                uint8_t n,
                struct rcu_ja_node_flag *child_node_flag)
 {
@@ -414,12 +444,14 @@ int ja_pool_node_set_nth(const struct rcu_ja_type *type,
        assert(type->type_class == RCU_JA_POOL);
        linear = (struct rcu_ja_node *)
                &node->u.data[((unsigned long) n >> (CHAR_BIT - type->nr_pool_order)) << type->pool_size_order];
-       return ja_linear_node_set_nth(type, linear, n, child_node_flag);
+       return ja_linear_node_set_nth(type, linear, shadow_node,
+                       n, child_node_flag);
 }
 
 static
 int ja_pigeon_node_set_nth(const struct rcu_ja_type *type,
                struct rcu_ja_node *node,
+               struct rcu_ja_shadow_node *shadow_node,
                uint8_t n,
                struct rcu_ja_node_flag *child_node_flag)
 {
@@ -430,6 +462,7 @@ int ja_pigeon_node_set_nth(const struct rcu_ja_type *type,
        if (*ptr != NULL)
                return -EEXIST;
        rcu_assign_pointer(*ptr, child_node_flag);
+       shadow_node->nr_child++;
        return 0;
 }
 
@@ -439,28 +472,24 @@ int ja_pigeon_node_set_nth(const struct rcu_ja_type *type,
  * TODO: exclusive access on node.
  */
 static
-int _ja_node_set_nth(struct rcu_ja_node_flag *node_flag, uint8_t n,
+int _ja_node_set_nth(const struct rcu_ja_type *type,
+               struct rcu_ja_node *node,
+               struct rcu_ja_shadow_node *shadow_node,
+               uint8_t n,
                struct rcu_ja_node_flag *child_node_flag)
 {
-       unsigned int type_index;
-       struct rcu_ja_node *node;
-       const struct rcu_ja_type *type;
-
-       node = ja_node_ptr(node_flag);
-       assert(node != NULL);
-       type_index = ja_node_type(node_flag);
-       type = &ja_types[type_index];
-
        switch (type->type_class) {
        case RCU_JA_LINEAR:
-               return ja_linear_node_set_nth(type, node, n,
+               return ja_linear_node_set_nth(type, node, shadow_node, n,
                                child_node_flag);
        case RCU_JA_POOL:
-               return ja_pool_node_set_nth(type, node, n,
+               return ja_pool_node_set_nth(type, node, shadow_node, n,
                                child_node_flag);
        case RCU_JA_PIGEON:
-               return ja_pigeon_node_set_nth(type, node, n,
+               return ja_pigeon_node_set_nth(type, node, shadow_node, n,
                                child_node_flag);
+       case RCU_JA_NULL:
+               return -ENOSPC;
        default:
                assert(0);
                return -EINVAL;
@@ -471,25 +500,27 @@ int _ja_node_set_nth(struct rcu_ja_node_flag *node_flag, uint8_t n,
 
 /*
  * ja_node_recompact_add: recompact a node, adding a new child.
+ * TODO: for pool type, take selection bit(s) into account.
  */
 static
-int ja_node_recompact_add(struct rcu_ja_node_flag **old_node_flag, uint8_t n,
+int ja_node_recompact_add(struct rcu_ja *ja,
+               unsigned int old_type_index,
+               const struct rcu_ja_type *old_type,
+               struct rcu_ja_node *old_node,
+               struct rcu_ja_shadow_node **shadow_node,
+               struct rcu_ja_node_flag **old_node_flag, uint8_t n,
                struct rcu_ja_node_flag *child_node_flag)
-
 {
-       unsigned int old_type_index, new_type_index;
-       struct rcu_ja_node *old_node, *new_node;
-       const struct rcu_ja_type *old_type, *new_type;
+       unsigned int new_type_index;
+       struct rcu_ja_node *new_node;
+       const struct rcu_ja_type *new_type;
        struct rcu_ja_node_flag *new_node_flag;
        unsigned int i;
        int ret;
 
-       old_node = ja_node_ptr(*old_node_flag);
-       if (old_node == NULL) {
+       if (*shadow_node == NULL) {
                new_type_index = 0;
        } else {
-               old_type_index = ja_node_type(*old_node_flag);
-               old_type = &ja_types[old_type_index];
                new_type_index = old_type_index + 1;
        }
        new_type = &ja_types[new_type_index];
@@ -498,37 +529,64 @@ int ja_node_recompact_add(struct rcu_ja_node_flag **old_node_flag, uint8_t n,
                return -ENOMEM;
        new_node_flag = ja_node_flag(new_node, new_type_index);
 
+       ret = rcuja_shadow_set(ja->ht, new_node, *shadow_node);
+       if (ret)
+               return ret;
+
+       if (*shadow_node == NULL) {
+               *shadow_node = rcuja_shadow_lookup_lock(ja->ht, new_node);
+               assert(*shadow_node);
+       }
+
+       (*shadow_node)->nr_child = 0;
        for (i = 0; i < old_type->max_child; i++) {
                struct rcu_ja_node_flag *iter;
 
                iter = ja_node_get_nth(*old_node_flag, i);
                if (!iter)
                        continue;
-               ret = _ja_node_set_nth(new_node_flag, i, iter);
+               ret = _ja_node_set_nth(new_type, new_node, *shadow_node,
+                               i, iter);
                assert(!ret);
        }
        /* add node */
-       ret = _ja_node_set_nth(new_node_flag, n, child_node_flag);
+       ret = _ja_node_set_nth(new_type, new_node, *shadow_node,
+                       n, child_node_flag);
        assert(!ret);
        /* Replace the old node with the new recompacted one */
        rcu_assign_pointer(*old_node_flag, new_node_flag);
-       /* TODO: free old_node (call_rcu) */
+       ret = rcuja_shadow_clear(ja->ht, old_node,
+                       RCUJA_SHADOW_CLEAR_FREE_NODE);
+       assert(!ret);
        return 0;
 }
 
 static
-int ja_node_set_nth(struct rcu_ja_node_flag **node_flag, uint8_t n,
+int ja_node_set_nth(struct rcu_ja *ja,
+               struct rcu_ja_node_flag **node_flag, uint8_t n,
                struct rcu_ja_node_flag *child_node_flag)
 {
        int ret;
+       unsigned int type_index;
+       const struct rcu_ja_type *type;
+       struct rcu_ja_node *node;
+       struct rcu_ja_shadow_node *shadow_node = NULL;
 
-       ret = _ja_node_set_nth(*node_flag, n, child_node_flag);
+       node = ja_node_ptr(*node_flag);
+       type_index = ja_node_type(*node_flag);
+       type = &ja_types[type_index];
+       if (node != NULL) {
+               shadow_node = rcuja_shadow_lookup_lock(ja->ht, node);
+               assert(shadow_node);
+       }
+       ret = _ja_node_set_nth(type, node, shadow_node,
+                       n, child_node_flag);
        if (ret == -ENOSPC) {
-               /* Not enough space in node, need to recompact */
-               ret = ja_node_recompact_add(node_flag, n,
-                               child_node_flag);
-               if (ret < 0)
-                       return ret;
+               /* Not enough space in node, need to recompact. */
+               ret = ja_node_recompact_add(ja, type_index, type, node,
+                               &shadow_node, node_flag, n, child_node_flag);
+               /* recompact always leave shadow_node locked */
        }
+       rcuja_shadow_unlock(shadow_node);
        return ret;
 }