Re: [Qemu-devel] [PATCH v6 10/15] qht: QEMU's fast, resizable and scalable Hash Table

[Sergey Fedorov <serge.fdrv@gmail.com>]

On 25/05/16 04:13, Emilio G. Cota wrote:
> diff --git a/include/qemu/qht.h b/include/qemu/qht.h
> new file mode 100644
> index 0000000..aec60aa
> --- /dev/null
> +++ b/include/qemu/qht.h
> @@ -0,0 +1,183 @@
(snip)
> +/**
> + * qht_init - Initialize a QHT
> + * @ht: QHT to be initialized
> + * @n_elems: number of entries the hash table should be optimized for.
> + * @mode: bitmask with OR'ed QHT_MODE_*
> + */
> +void qht_init(struct qht *ht, size_t n_elems, unsigned int mode);

First of all, thank you for spending your time on the documentation of
the API!

I was just wondering if it could be worthwhile to pass a hash function
when initializing a QHT. Then we could have variants of qht_insert(),
qht_remove() and qht_lookup() which does not require a computed hash
value but call the function by themselves. This could make sense since a
hash value passed the the functions should always be exactly the same
for the same object.

(snip)
> +/**
> + * qht_remove - remove a pointer from the hash table
> + * @ht: QHT to remove from
> + * @p: pointer to be removed
> + * @hash: hash corresponding to @p
> + *
> + * Attempting to remove a NULL @p is a bug.
> + *
> + * Just-removed @p pointers cannot be immediately freed; they need to remain
> + * valid until the end of the RCU grace period in which qht_remove() is called.
> + * This guarantees that concurrent lookups will always compare against valid
> + * data.

Mention rcu_call1()/call_rcu()/g_free_rcu()?

> + *
> + * Returns true on success.
> + * Returns false if the @p-@hash pair was not found.
> + */
> +bool qht_remove(struct qht *ht, const void *p, uint32_t hash);
> +
(snip)
> diff --git a/util/qht.c b/util/qht.c
> new file mode 100644
> index 0000000..ca5a620
> --- /dev/null
> +++ b/util/qht.c
> @@ -0,0 +1,837 @@
(snip)
> +/* trigger a resize when n_added_buckets > n_buckets / div */
> +#define QHT_NR_ADDED_BUCKETS_THRESHOLD_DIV 8

Just out of curiosity, how did you get this number?

> +
> +static void qht_do_resize(struct qht *ht, struct qht_map *new);
> +static void qht_grow_maybe(struct qht *ht);

qht_grow_maybe() is used just once. Please consider reordering of
definitions and removing this forward declaration.

(snip)
> +
> +/* call with head->lock held */
> +static bool qht_insert__locked(struct qht *ht, struct qht_map *map,
> +                               struct qht_bucket *head, void *p, uint32_t hash,
> +                               bool *needs_resize)
> +{
> +    struct qht_bucket *b = head;
> +    struct qht_bucket *prev = NULL;
> +    struct qht_bucket *new = NULL;
> +    int i;
> +
> +    for (;;) {
> +        if (b == NULL) {
> +            b = qemu_memalign(QHT_BUCKET_ALIGN, sizeof(*b));
> +            memset(b, 0, sizeof(*b));
> +            new = b;
> +            atomic_inc(&map->n_added_buckets);
> +            if (unlikely(qht_map_needs_resize(map)) && needs_resize) {
> +                *needs_resize = true;
> +            }
> +        }
> +        for (i = 0; i < QHT_BUCKET_ENTRIES; i++) {
> +            if (b->pointers[i]) {
> +                if (unlikely(b->pointers[i] == p)) {
> +                    return false;
> +                }
> +                continue;
> +            }
> +            /* found an empty key: acquire the seqlock and write */
> +            seqlock_write_begin(&head->sequence);
> +            if (new) {
> +                atomic_rcu_set(&prev->next, b);
> +            }
> +            b->hashes[i] = hash;
> +            atomic_set(&b->pointers[i], p);
> +            seqlock_write_end(&head->sequence);
> +            return true;
> +        }
> +        prev = b;
> +        b = b->next;
> +    }
> +}

Here is my attempt:

static bool qht_insert__locked(struct qht *ht, struct qht_map
*map,           
                               struct qht_bucket *head, void *p,
uint32_t hash,
                               bool
*needs_resize)                            
{                                                                             

    struct qht_bucket **bpp = &head,
*new;                                    
    int i =
0;                                                                
                                                                              

    do
{                                                                      
        while (i < QHT_BUCKET_ENTRIES)
{                                      
            if ((*bpp)->pointers[i])
{                                        
                if (unlikely((*bpp)->pointers[i] == p))
{                     
                    return
false;                                             
               
}                                                             
               
i++;                                                          
               
continue;                                                     
           
}                                                                 
            goto
found;                                                       
       
}                                                                     
        bpp =
&(*bpp)->next;                                                  
        i =
0;                                                                
    } while
(*bpp);                                                           
                                                                              

    new = qemu_memalign(QHT_BUCKET_ALIGN,
sizeof(*new));                      
    memset(new, 0,
sizeof(*new));                                             
    atomic_rcu_set(bpp,
new);                                                 
   
atomic_inc(&map->n_added_buckets);                                        
    if (unlikely(qht_map_needs_resize(map)) && needs_resize)
{                
        *needs_resize =
true;                                                 
   
}                                                                         
found:                                                                        

    /* found an empty key: acquire the seqlock and write
*/                   
   
seqlock_write_begin(&head->sequence);                                     
    (*bpp)->hashes[i] =
hash;                                                 
    atomic_set(&(*bpp)->pointers[i],
p);                                      
   
seqlock_write_end(&head->sequence);                                       
    return
true;                                                              
}                                                                             


Feel free to use it as you wish.

>
(snip)
> +/*
> + * Find the last valid entry in @head, and swap it with @orig[pos], which has
> + * just been invalidated.
> + */
> +static inline void qht_bucket_fill_hole(struct qht_bucket *orig, int pos)
> +{
> +    struct qht_bucket *b = orig;
> +    struct qht_bucket *prev = NULL;
> +    int i;
> +
> +    if (qht_entry_is_last(orig, pos)) {
> +        orig->hashes[pos] = 0;
> +        atomic_set(&orig->pointers[pos], NULL);
> +        return;
> +    }
> +    do {
> +        for (i = 0; i < QHT_BUCKET_ENTRIES; i++) {
> +            if (b->pointers[i]) {
> +                continue;
> +            }
> +            if (i > 0) {
> +                return qht_entry_move(orig, pos, b, i - 1);
> +            }
> +            qht_debug_assert(prev);

'prev' can be NULL if this is the first iteration.

> +            return qht_entry_move(orig, pos, prev, QHT_BUCKET_ENTRIES - 1);
> +        }
> +        prev = b;
> +        b = b->next;
> +    } while (b);
> +    /* no free entries other than orig[pos], so swap it with the last one */
> +    qht_entry_move(orig, pos, prev, QHT_BUCKET_ENTRIES - 1);
> +}
> +
> +/* call with b->lock held */
> +static inline
> +bool qht_remove__locked(struct qht_map *map, struct qht_bucket *head,
> +                        const void *p, uint32_t hash)
> +{
> +    struct qht_bucket *b = head;
> +    int i;
> +
> +    do {
> +        for (i = 0; i < QHT_BUCKET_ENTRIES; i++) {
> +            void *q = b->pointers[i];
> +
> +            if (unlikely(q == NULL)) {
> +                return false;
> +            }
> +            if (q == p) {
> +                qht_debug_assert(b->hashes[i] == hash);
> +                seqlock_write_begin(&head->sequence);
> +                qht_bucket_fill_hole(b, i);

"Fill hole" doesn't correspond to the function's new job since there's
no hole. "Remove entry" would make more sense, I think.

> +                seqlock_write_end(&head->sequence);
> +                return true;
> +            }
> +        }
> +        b = b->next;
> +    } while (b);
> +    return false;
> +}
(snip)
> +/*
> + * Call with ht->lock and all bucket locks held.
> + *
> + * Creating the @new map here would add unnecessary delay while all the locks
> + * are held--holding up the bucket locks is particularly bad, since no writes
> + * can occur while these are held. Thus, we let callers create the new map,
> + * hopefully without the bucket locks held.
> + */
> +static void qht_do_resize(struct qht *ht, struct qht_map *new)
> +{
> +    struct qht_map *old;
> +
> +    old = ht->map;
> +    g_assert_cmpuint(new->n_buckets, !=, old->n_buckets);
> +
> +    qht_map_iter__all_locked(ht, old, qht_map_copy, new);
> +    qht_map_debug__all_locked(new);
> +
> +    atomic_rcu_set(&ht->map, new);
> +    call_rcu1(&old->rcu, qht_map_reclaim);

call_rcu() macro is a more convenient way to do this and you wouldn't
need qht_map_reclaim().

> +}

Kind regards,
Sergey