diff mbox

[v6,10/15] qht: QEMU's fast, resizable and scalable Hash Table

Message ID 1464138802-23503-11-git-send-email-cota@braap.org (mailing list archive)
State New, archived
Headers show

Commit Message

Emilio Cota May 25, 2016, 1:13 a.m. UTC
This is a fast, scalable chained hash table with optional auto-resizing, allowing
reads that are concurrent with reads, and reads/writes that are concurrent
with writes to separate buckets.

A hash table with these features will be necessary for the scalability
of the ongoing MTTCG work; before those changes arrive we can already
benefit from the single-threaded speedup that qht also provides.

Signed-off-by: Emilio G. Cota <cota@braap.org>
---
 include/qemu/qht.h | 183 ++++++++++++
 util/Makefile.objs |   1 +
 util/qht.c         | 837 +++++++++++++++++++++++++++++++++++++++++++++++++++++
 3 files changed, 1021 insertions(+)
 create mode 100644 include/qemu/qht.h
 create mode 100644 util/qht.c

Comments

Sergey Fedorov May 29, 2016, 7:52 p.m. UTC | #1
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
Sergey Fedorov May 29, 2016, 7:55 p.m. UTC | #2
On 29/05/16 22:52, Sergey Fedorov wrote:
> On 25/05/16 04:13, Emilio G. Cota wrote:
>> +
>> +/* 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.

Sorry for the chopped email. Hope this one will be better :)

Kind regards,
Sergey
Alex Bennée May 31, 2016, 7:46 a.m. UTC | #3
Sergey Fedorov <serge.fdrv@gmail.com> writes:

> 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.

Wouldn't this be for an expansion of the API when we actually have
something that would use it that way?

>
> (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


--
Alex Bennée
Sergey Fedorov June 1, 2016, 8:53 p.m. UTC | #4
On 31/05/16 10:46, Alex Bennée wrote:
> Sergey Fedorov <serge.fdrv@gmail.com> writes:
>
>> 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.
> Wouldn't this be for an expansion of the API when we actually have
> something that would use it that way?
>

Yes, I think in "tb hash: track translated blocks with qht" we could just pass tb_hash_func() to qht_init().

Kind regards,
Sergey
Emilio Cota June 3, 2016, 9:18 a.m. UTC | #5
On Sun, May 29, 2016 at 22:52:27 +0300, Sergey Fedorov wrote:
> 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.

I considered this when designing the API. I think it's not worth having
in qht; callers could have their own wrapper to do this though.

For the only caller of qht that we have so far I don't see this
as being worth the hassle.

For instance, we couldn't use the same function for lookups and
inserts/removals, since the hash function would look like:

uint32_t hash_func(void *p)
{
	TranslationBlock *tb = p;
	return tb_hash_func(tb->phys_pc, ...);
}

But for lookups we don't yet know *tb (that's what we're looking for!).
All we have is the tb_desc struct that we use for comparisons.

Thanks,

		Emilio
Emilio Cota June 3, 2016, 11:01 a.m. UTC | #6
On Sun, May 29, 2016 at 22:52:27 +0300, Sergey Fedorov wrote:
> > +/**
> > + * 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()?

Probably using 'see also:' for non-qht functions is a little too much.
The pointer to RCU should be enough, me thinks.

> (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?

Good question. Tested what gave the best performance for the ARM
bootup test :-) It matched the performance of the old behavior,
that was, to double the size when n_entries reached
n_buckets * QHT_BUCKET_ENTRIES / 2.

> > +
> > +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.

Done.

(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.

How can prev be NULL here and that not be a bug? NULL here would
mean there was a hole before orig[pos]. Or orig[pos] was
NULL, which is also a bug.

> > +
> > +/* 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.

Changed.

> (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().

True. Originally the RCU struct wasn't at the head, then I changed
it to please valgrind. Changed now.

Thanks,

		Emilio
Sergey Fedorov June 3, 2016, 3:19 p.m. UTC | #7
On 03/06/16 12:18, Emilio G. Cota wrote:
> On Sun, May 29, 2016 at 22:52:27 +0300, Sergey Fedorov wrote:
>> 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.
> I considered this when designing the API. I think it's not worth having
> in qht; callers could have their own wrapper to do this though.
>
> For the only caller of qht that we have so far I don't see this
> as being worth the hassle.
>
> For instance, we couldn't use the same function for lookups and
> inserts/removals, since the hash function would look like:
>
> uint32_t hash_func(void *p)
> {
> 	TranslationBlock *tb = p;
> 	return tb_hash_func(tb->phys_pc, ...);
> }
>
> But for lookups we don't yet know *tb (that's what we're looking for!).
> All we have is the tb_desc struct that we use for comparisons.

Fair enough.

Thanks,
Sergey
Sergey Fedorov June 3, 2016, 3:34 p.m. UTC | #8
On 03/06/16 14:01, Emilio G. Cota wrote:
> On Sun, May 29, 2016 at 22:52:27 +0300, Sergey Fedorov wrote:
>>> +/*
>>> + * 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.
> How can prev be NULL here and that not be a bug? NULL here would
> mean there was a hole before orig[pos]. Or orig[pos] was
> NULL, which is also a bug.

Yeah, you're right.

Kind regards,
Sergey
diff mbox

Patch

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 @@ 
+/*
+ * Copyright (C) 2016, Emilio G. Cota <cota@braap.org>
+ *
+ * License: GNU GPL, version 2 or later.
+ *   See the COPYING file in the top-level directory.
+ */
+#ifndef QEMU_QHT_H
+#define QEMU_QHT_H
+
+#include "qemu/osdep.h"
+#include "qemu/seqlock.h"
+#include "qemu/thread.h"
+#include "qemu/qdist.h"
+
+struct qht {
+    struct qht_map *map;
+    QemuMutex lock; /* serializes setters of ht->map */
+    unsigned int mode;
+};
+
+/**
+ * struct qht_stats - Statistics of a QHT
+ * @head_buckets: number of head buckets
+ * @used_head_buckets: number of non-empty head buckets
+ * @entries: total number of entries
+ * @chain: frequency distribution representing the number of buckets in each
+ *         chain, excluding empty chains.
+ * @occupancy: frequency distribution representing chain occupancy rate.
+ *             Valid range: from 0.0 (empty) to 1.0 (full occupancy).
+ *
+ * An entry is a pointer-hash pair.
+ * Each bucket can host several entries.
+ * Chains are chains of buckets, whose first link is always a head bucket.
+ */
+struct qht_stats {
+    size_t head_buckets;
+    size_t used_head_buckets;
+    size_t entries;
+    struct qdist chain;
+    struct qdist occupancy;
+};
+
+typedef bool (*qht_lookup_func_t)(const void *obj, const void *userp);
+typedef void (*qht_iter_func_t)(struct qht *ht, void *p, uint32_t h, void *up);
+
+#define QHT_MODE_AUTO_RESIZE 0x1 /* auto-resize when heavily loaded */
+
+/**
+ * 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);
+
+/**
+ * qht_destroy - destroy a previously initialized QHT
+ * @ht: QHT to be destroyed
+ *
+ * Call only when there are no readers/writers left.
+ */
+void qht_destroy(struct qht *ht);
+
+/**
+ * qht_insert - Insert a pointer into the hash table
+ * @ht: QHT to insert to
+ * @p: pointer to be inserted
+ * @hash: hash corresponding to @p
+ *
+ * Attempting to insert a NULL @p is a bug.
+ * Inserting the same pointer @p with different @hash values is a bug.
+ *
+ * Returns true on sucess.
+ * Returns false if the @p-@hash pair already exists in the hash table.
+ */
+bool qht_insert(struct qht *ht, void *p, uint32_t hash);
+
+/**
+ * qht_lookup - Look up a pointer in a QHT
+ * @ht: QHT to be looked up
+ * @func: function to compare existing pointers against @userp
+ * @userp: pointer to pass to @func
+ * @hash: hash of the pointer to be looked up
+ *
+ * Needs to be called under an RCU read-critical section.
+ *
+ * The user-provided @func compares pointers in QHT against @userp.
+ * If the function returns true, a match has been found.
+ *
+ * Returns the corresponding pointer when a match is found.
+ * Returns NULL otherwise.
+ */
+void *qht_lookup(struct qht *ht, qht_lookup_func_t func, const void *userp,
+                 uint32_t hash);
+
+/**
+ * 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.
+ *
+ * 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);
+
+/**
+ * qht_reset - reset a QHT
+ * @ht: QHT to be reset
+ *
+ * All entries in the hash table are reset. No resizing is performed.
+ *
+ * If concurrent readers may exist, the objects pointed to by the hash table
+ * must remain valid for the existing RCU grace period -- see qht_remove().
+ * See also: qht_reset_size()
+ */
+void qht_reset(struct qht *ht);
+
+/**
+ * qht_reset_size - reset and resize a QHT
+ * @ht: QHT to be reset and resized
+ * @n_elems: number of entries the resized hash table should be optimized for.
+ *
+ * Returns true if the resize was necessary and therefore performed.
+ * Returns false otherwise.
+ *
+ * If concurrent readers may exist, the objects pointed to by the hash table
+ * must remain valid for the existing RCU grace period -- see qht_remove().
+ * See also: qht_reset(), qht_resize().
+ */
+bool qht_reset_size(struct qht *ht, size_t n_elems);
+
+/**
+ * qht_resize - resize a QHT
+ * @ht: QHT to be resized
+ * @n_elems: number of entries the resized hash table should be optimized for
+ *
+ * Returns true on success.
+ * Returns false if the resize was not necessary and therefore not performed.
+ * See also: qht_reset_size().
+ */
+bool qht_resize(struct qht *ht, size_t n_elems);
+
+/**
+ * qht_iter - Iterate over a QHT
+ * @ht: QHT to be iterated over
+ * @func: function to be called for each entry in QHT
+ * @userp: additional pointer to be passed to @func
+ *
+ * Each time it is called, user-provided @func is passed a pointer-hash pair,
+ * plus @userp.
+ */
+void qht_iter(struct qht *ht, qht_iter_func_t func, void *userp);
+
+/**
+ * qht_statistics_init - Gather statistics from a QHT
+ * @ht: QHT to gather statistics from
+ * @stats: pointer to a struct qht_stats to be filled in
+ *
+ * Does NOT need to be called under an RCU read-critical section,
+ * since it does not dereference any pointers stored in the hash table.
+ *
+ * When done with @stats, pass the struct to qht_statistics_destroy().
+ * Failing to do this will leak memory.
+ */
+void qht_statistics_init(struct qht *ht, struct qht_stats *stats);
+
+/**
+ * qht_statistics_destroy - Destroy a struct qht_stats
+ * @stats: stuct qht_stats to be destroyed
+ *
+ * See also: qht_statistics_init().
+ */
+void qht_statistics_destroy(struct qht_stats *stats);
+
+#endif /* QEMU_QHT_H */
diff --git a/util/Makefile.objs b/util/Makefile.objs
index 702435e..45f8794 100644
--- a/util/Makefile.objs
+++ b/util/Makefile.objs
@@ -33,3 +33,4 @@  util-obj-y += timed-average.o
 util-obj-y += base64.o
 util-obj-y += log.o
 util-obj-y += qdist.o
+util-obj-y += qht.o
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 @@ 
+/*
+ * qht.c - QEMU Hash Table, designed to scale for read-mostly workloads.
+ *
+ * Copyright (C) 2016, Emilio G. Cota <cota@braap.org>
+ *
+ * License: GNU GPL, version 2 or later.
+ *   See the COPYING file in the top-level directory.
+ *
+ * Assumptions:
+ * - NULL cannot be inserted/removed as a pointer value.
+ * - Trying to insert an already-existing hash-pointer pair is OK. However,
+ *   it is not OK to insert into the same hash table different hash-pointer
+ *   pairs that have the same pointer value, but not the hashes.
+ * - Lookups are performed under an RCU read-critical section; removals
+ *   must wait for a grace period to elapse before freeing removed objects.
+ *
+ * Features:
+ * - Reads (i.e. lookups and iterators) can be concurrent with other reads.
+ *   Lookups that are concurrent with writes to the same bucket will retry
+ *   via a seqlock; iterators acquire all bucket locks and therefore can be
+ *   concurrent with lookups and are serialized wrt writers.
+ * - Writes (i.e. insertions/removals) can be concurrent with writes to
+ *   different buckets; writes to the same bucket are serialized through a lock.
+ * - Optional auto-resizing: the hash table resizes up if the load surpasses
+ *   a certain threshold. Resizing is done concurrently with readers; writes
+ *   are serialized with the resize operation.
+ *
+ * The key structure is the bucket, which is cacheline-sized. Buckets
+ * contain a few hash values and pointers; the u32 hash values are stored in
+ * full so that resizing is fast. Having this structure instead of directly
+ * chaining items has two advantages:
+ * - Failed lookups fail fast, and touch a minimum number of cache lines.
+ * - Resizing the hash table with concurrent lookups is easy.
+ *
+ * There are two types of buckets:
+ * 1. "head" buckets are the ones allocated in the array of buckets in qht_map.
+ * 2. all "non-head" buckets (i.e. all others) are members of a chain that
+ *    starts from a head bucket.
+ * Note that the seqlock and spinlock of a head bucket applies to all buckets
+ * chained to it; these two fields are unused in non-head buckets.
+ *
+ * On removals, we move the last valid item in the chain to the position of the
+ * just-removed entry. This makes lookups slightly faster, since the moment an
+ * invalid entry is found, the (failed) lookup is over.
+ *
+ * Resizing is done by taking all bucket spinlocks (so that no other writers can
+ * race with us) and then copying all entries into a new hash map. Then, the
+ * ht->map pointer is set, and the old map is freed once no RCU readers can see
+ * it anymore.
+ *
+ * Writers check for concurrent resizes by comparing ht->map before and after
+ * acquiring their bucket lock. If they don't match, a resize has occured
+ * while the bucket spinlock was being acquired.
+ *
+ * Related Work:
+ * - Idea of cacheline-sized buckets with full hashes taken from:
+ *   David, Guerraoui & Trigonakis, "Asynchronized Concurrency:
+ *   The Secret to Scaling Concurrent Search Data Structures", ASPLOS'15.
+ * - Why not RCU-based hash tables? They would allow us to get rid of the
+ *   seqlock, but resizing would take forever since RCU read critical
+ *   sections in QEMU take quite a long time.
+ *   More info on relativistic hash tables:
+ *   + Triplett, McKenney & Walpole, "Resizable, Scalable, Concurrent Hash
+ *     Tables via Relativistic Programming", USENIX ATC'11.
+ *   + Corbet, "Relativistic hash tables, part 1: Algorithms", @ lwn.net, 2014.
+ *     https://lwn.net/Articles/612021/
+ */
+#include "qemu/qht.h"
+#include "qemu/atomic.h"
+#include "qemu/rcu.h"
+
+//#define QHT_DEBUG
+
+/*
+ * We want to avoid false sharing of cache lines. Most systems have 64-byte
+ * cache lines so we go with it for simplicity.
+ *
+ * Note that systems with smaller cache lines will be fine (the struct is
+ * almost 64-bytes); systems with larger cache lines might suffer from
+ * some false sharing.
+ */
+#define QHT_BUCKET_ALIGN 64
+
+/* define these to keep sizeof(qht_bucket) within QHT_BUCKET_ALIGN */
+#if HOST_LONG_BITS == 32
+#define QHT_BUCKET_ENTRIES 6
+#else /* 64-bit */
+#define QHT_BUCKET_ENTRIES 4
+#endif
+
+/*
+ * Note: reading partially-updated pointers in @pointers could lead to
+ * segfaults. We thus access them with atomic_read/set; this guarantees
+ * that the compiler makes all those accesses atomic. We also need the
+ * volatile-like behavior in atomic_read, since otherwise the compiler
+ * might refetch the pointer.
+ * atomic_read's are of course not necessary when the bucket lock is held.
+ *
+ * If both ht->lock and b->lock are grabbed, ht->lock should always
+ * be grabbed first.
+ */
+struct qht_bucket {
+    QemuSpin lock;
+    QemuSeqLock sequence;
+    uint32_t hashes[QHT_BUCKET_ENTRIES];
+    void *pointers[QHT_BUCKET_ENTRIES];
+    struct qht_bucket *next;
+} QEMU_ALIGNED(QHT_BUCKET_ALIGN);
+
+QEMU_BUILD_BUG_ON(sizeof(struct qht_bucket) > QHT_BUCKET_ALIGN);
+
+/**
+ * struct qht_map - structure to track an array of buckets
+ * @rcu: used by RCU. Keep it as the top field in the struct to help valgrind
+ *       find the whole struct.
+ * @buckets: array of head buckets. It is constant once the map is created.
+ * @n_buckets: number of head buckets. It is constant once the map is created.
+ * @n_added_buckets: number of added (i.e. "non-head") buckets
+ * @n_added_buckets_threshold: threshold to trigger an upward resize once the
+ *                             number of added buckets surpasses it.
+ *
+ * Buckets are tracked in what we call a "map", i.e. this structure.
+ */
+struct qht_map {
+    struct rcu_head rcu;
+    struct qht_bucket *buckets;
+    size_t n_buckets;
+    size_t n_added_buckets;
+    size_t n_added_buckets_threshold;
+};
+
+/* trigger a resize when n_added_buckets > n_buckets / div */
+#define QHT_NR_ADDED_BUCKETS_THRESHOLD_DIV 8
+
+static void qht_do_resize(struct qht *ht, struct qht_map *new);
+static void qht_grow_maybe(struct qht *ht);
+
+#ifdef QHT_DEBUG
+
+#define qht_debug_assert(X) do { assert(X); } while (0)
+
+static void qht_bucket_debug__locked(struct qht_bucket *b)
+{
+    bool seen_empty = false;
+    bool corrupt = false;
+    int i;
+
+    do {
+        for (i = 0; i < QHT_BUCKET_ENTRIES; i++) {
+            if (b->pointers[i] == NULL) {
+                seen_empty = true;
+                continue;
+            }
+            if (seen_empty) {
+                fprintf(stderr, "%s: b: %p, pos: %i, hash: 0x%x, p: %p\n",
+                       __func__, b, i, b->hashes[i], b->pointers[i]);
+                corrupt = true;
+            }
+        }
+        b = b->next;
+    } while (b);
+    qht_debug_assert(!corrupt);
+}
+
+static void qht_map_debug__all_locked(struct qht_map *map)
+{
+    int i;
+
+    for (i = 0; i < map->n_buckets; i++) {
+        qht_bucket_debug__locked(&map->buckets[i]);
+    }
+}
+#else
+
+#define qht_debug_assert(X) do { (void)(X); } while (0)
+
+static inline void qht_bucket_debug__locked(struct qht_bucket *b)
+{ }
+
+static inline void qht_map_debug__all_locked(struct qht_map *map)
+{ }
+#endif /* QHT_DEBUG */
+
+static inline size_t qht_elems_to_buckets(size_t n_elems)
+{
+    return pow2ceil(n_elems / QHT_BUCKET_ENTRIES);
+}
+
+static inline void qht_head_init(struct qht_bucket *b)
+{
+    memset(b, 0, sizeof(*b));
+    qemu_spin_init(&b->lock);
+    seqlock_init(&b->sequence);
+}
+
+static inline
+struct qht_bucket *qht_map_to_bucket(struct qht_map *map, uint32_t hash)
+{
+    return &map->buckets[hash & (map->n_buckets - 1)];
+}
+
+/* acquire all bucket locks from a map */
+static void qht_map_lock_buckets(struct qht_map *map)
+{
+    size_t i;
+
+    for (i = 0; i < map->n_buckets; i++) {
+        struct qht_bucket *b = &map->buckets[i];
+
+        qemu_spin_lock(&b->lock);
+    }
+}
+
+static void qht_map_unlock_buckets(struct qht_map *map)
+{
+    size_t i;
+
+    for (i = 0; i < map->n_buckets; i++) {
+        struct qht_bucket *b = &map->buckets[i];
+
+        qemu_spin_unlock(&b->lock);
+    }
+}
+
+/*
+ * Call with at least a bucket lock held.
+ * @map should be the value read before acquiring the lock (or locks).
+ */
+static inline bool qht_map_is_stale__locked(struct qht *ht, struct qht_map *map)
+{
+    return map != ht->map;
+}
+
+/*
+ * Grab all bucket locks, and set @pmap after making sure the map isn't stale.
+ *
+ * Pairs with qht_map_unlock_buckets(), hence the pass-by-reference.
+ *
+ * Note: callers cannot have ht->lock held.
+ */
+static inline
+void qht_map_lock_buckets__no_stale(struct qht *ht, struct qht_map **pmap)
+{
+    struct qht_map *map;
+
+    map = atomic_rcu_read(&ht->map);
+    qht_map_lock_buckets(map);
+    if (likely(!qht_map_is_stale__locked(ht, map))) {
+        *pmap = map;
+        return;
+    }
+    qht_map_unlock_buckets(map);
+
+    /* we raced with a resize; acquire ht->lock to see the updated ht->map */
+    qemu_mutex_lock(&ht->lock);
+    map = ht->map;
+    qht_map_lock_buckets(map);
+    qemu_mutex_unlock(&ht->lock);
+    *pmap = map;
+    return;
+}
+
+/*
+ * Get a head bucket and lock it, making sure its parent map is not stale.
+ * @pmap is filled with a pointer to the bucket's parent map.
+ *
+ * Unlock with qemu_spin_unlock(&b->lock).
+ *
+ * Note: callers cannot have ht->lock held.
+ */
+static inline
+struct qht_bucket *qht_bucket_lock__no_stale(struct qht *ht, uint32_t hash,
+                                             struct qht_map **pmap)
+{
+    struct qht_bucket *b;
+    struct qht_map *map;
+
+    map = atomic_rcu_read(&ht->map);
+    b = qht_map_to_bucket(map, hash);
+
+    qemu_spin_lock(&b->lock);
+    if (likely(!qht_map_is_stale__locked(ht, map))) {
+        *pmap = map;
+        return b;
+    }
+    qemu_spin_unlock(&b->lock);
+
+    /* we raced with a resize; acquire ht->lock to see the updated ht->map */
+    qemu_mutex_lock(&ht->lock);
+    map = ht->map;
+    b = qht_map_to_bucket(map, hash);
+    qemu_spin_lock(&b->lock);
+    qemu_mutex_unlock(&ht->lock);
+    *pmap = map;
+    return b;
+}
+
+static inline bool qht_map_needs_resize(struct qht_map *map)
+{
+    return atomic_read(&map->n_added_buckets) > map->n_added_buckets_threshold;
+}
+
+static inline void qht_chain_destroy(struct qht_bucket *head)
+{
+    struct qht_bucket *curr = head->next;
+    struct qht_bucket *prev;
+
+    while (curr) {
+        prev = curr;
+        curr = curr->next;
+        qemu_vfree(prev);
+    }
+}
+
+/* pass only an orphan map */
+static void qht_map_destroy(struct qht_map *map)
+{
+    size_t i;
+
+    for (i = 0; i < map->n_buckets; i++) {
+        qht_chain_destroy(&map->buckets[i]);
+    }
+    qemu_vfree(map->buckets);
+    g_free(map);
+}
+
+static void qht_map_reclaim(struct rcu_head *rcu)
+{
+    struct qht_map *map = container_of(rcu, struct qht_map, rcu);
+
+    qht_map_destroy(map);
+}
+
+static struct qht_map *qht_map_create(size_t n_buckets)
+{
+    struct qht_map *map;
+    size_t i;
+
+    map = g_malloc(sizeof(*map));
+    map->n_buckets = n_buckets;
+
+    map->n_added_buckets = 0;
+    map->n_added_buckets_threshold = n_buckets /
+        QHT_NR_ADDED_BUCKETS_THRESHOLD_DIV;
+
+    /* let tiny hash tables to at least add one non-head bucket */
+    if (unlikely(map->n_added_buckets_threshold == 0)) {
+        map->n_added_buckets_threshold = 1;
+    }
+
+    map->buckets = qemu_memalign(QHT_BUCKET_ALIGN,
+                                 sizeof(*map->buckets) * n_buckets);
+    for (i = 0; i < n_buckets; i++) {
+        qht_head_init(&map->buckets[i]);
+    }
+    return map;
+}
+
+void qht_init(struct qht *ht, size_t n_elems, unsigned int mode)
+{
+    struct qht_map *map;
+    size_t n_buckets = qht_elems_to_buckets(n_elems);
+
+    ht->mode = mode;
+    qemu_mutex_init(&ht->lock);
+    map = qht_map_create(n_buckets);
+    atomic_rcu_set(&ht->map, map);
+}
+
+/* call only when there are no readers/writers left */
+void qht_destroy(struct qht *ht)
+{
+    qht_map_destroy(ht->map);
+    memset(ht, 0, sizeof(*ht));
+}
+
+static void qht_bucket_reset__locked(struct qht_bucket *head)
+{
+    struct qht_bucket *b = head;
+    int i;
+
+    seqlock_write_begin(&head->sequence);
+    do {
+        for (i = 0; i < QHT_BUCKET_ENTRIES; i++) {
+            if (b->pointers[i] == NULL) {
+                goto done;
+            }
+            b->hashes[i] = 0;
+            atomic_set(&b->pointers[i], NULL);
+        }
+        b = b->next;
+    } while (b);
+ done:
+    seqlock_write_end(&head->sequence);
+}
+
+/* call with all bucket locks held */
+static void qht_map_reset__all_locked(struct qht_map *map)
+{
+    size_t i;
+
+    for (i = 0; i < map->n_buckets; i++) {
+        qht_bucket_reset__locked(&map->buckets[i]);
+    }
+    qht_map_debug__all_locked(map);
+}
+
+void qht_reset(struct qht *ht)
+{
+    struct qht_map *map;
+
+    qht_map_lock_buckets__no_stale(ht, &map);
+    qht_map_reset__all_locked(map);
+    qht_map_unlock_buckets(map);
+}
+
+bool qht_reset_size(struct qht *ht, size_t n_elems)
+{
+    struct qht_map *new;
+    struct qht_map *map;
+    size_t n_buckets;
+    bool resize = false;
+
+    n_buckets = qht_elems_to_buckets(n_elems);
+
+    qemu_mutex_lock(&ht->lock);
+    map = ht->map;
+    if (n_buckets != map->n_buckets) {
+        new = qht_map_create(n_buckets);
+        resize = true;
+    }
+
+    qht_map_lock_buckets(map);
+    qht_map_reset__all_locked(map);
+    if (resize) {
+        qht_do_resize(ht, new);
+    }
+    qht_map_unlock_buckets(map);
+    qemu_mutex_unlock(&ht->lock);
+
+    return resize;
+}
+
+static inline
+void *qht_do_lookup(struct qht_bucket *head, qht_lookup_func_t func,
+                    const void *userp, uint32_t hash)
+{
+    struct qht_bucket *b = head;
+    int i;
+
+    do {
+        for (i = 0; i < QHT_BUCKET_ENTRIES; i++) {
+            if (b->hashes[i] == hash) {
+                void *p = atomic_read(&b->pointers[i]);
+
+                if (likely(p) && likely(func(p, userp))) {
+                    return p;
+                }
+            }
+        }
+        b = atomic_rcu_read(&b->next);
+    } while (b);
+
+    return NULL;
+}
+
+static __attribute__((noinline))
+void *qht_lookup__slowpath(struct qht_bucket *b, qht_lookup_func_t func,
+                           const void *userp, uint32_t hash)
+{
+    unsigned int version;
+    void *ret;
+
+    do {
+        version = seqlock_read_begin(&b->sequence);
+        ret = qht_do_lookup(b, func, userp, hash);
+    } while (seqlock_read_retry(&b->sequence, version));
+    return ret;
+}
+
+void *qht_lookup(struct qht *ht, qht_lookup_func_t func, const void *userp,
+                 uint32_t hash)
+{
+    struct qht_bucket *b;
+    struct qht_map *map;
+    unsigned int version;
+    void *ret;
+
+    map = atomic_rcu_read(&ht->map);
+    b = qht_map_to_bucket(map, hash);
+
+    version = seqlock_read_begin(&b->sequence);
+    ret = qht_do_lookup(b, func, userp, hash);
+    if (likely(!seqlock_read_retry(&b->sequence, version))) {
+        return ret;
+    }
+    /*
+     * Removing the do/while from the fastpath gives a 4% perf. increase when
+     * running a 100%-lookup microbenchmark.
+     */
+    return qht_lookup__slowpath(b, func, userp, hash);
+}
+
+/* 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;
+    }
+}
+
+bool qht_insert(struct qht *ht, void *p, uint32_t hash)
+{
+    struct qht_bucket *b;
+    struct qht_map *map;
+    bool needs_resize = false;
+    bool ret;
+
+    /* NULL pointers are not supported */
+    qht_debug_assert(p);
+
+    b = qht_bucket_lock__no_stale(ht, hash, &map);
+    ret = qht_insert__locked(ht, map, b, p, hash, &needs_resize);
+    qht_bucket_debug__locked(b);
+    qemu_spin_unlock(&b->lock);
+
+    if (unlikely(needs_resize) && ht->mode & QHT_MODE_AUTO_RESIZE) {
+        qht_grow_maybe(ht);
+    }
+    return ret;
+}
+
+static inline bool qht_entry_is_last(struct qht_bucket *b, int pos)
+{
+    if (pos == QHT_BUCKET_ENTRIES - 1) {
+        if (b->next == NULL) {
+            return true;
+        }
+        return b->next->pointers[0] == NULL;
+    }
+    return b->pointers[pos + 1] == NULL;
+}
+
+static void
+qht_entry_move(struct qht_bucket *to, int i, struct qht_bucket *from, int j)
+{
+    qht_debug_assert(!(to == from && i == j));
+    qht_debug_assert(to->pointers[i]);
+    qht_debug_assert(from->pointers[j]);
+
+    to->hashes[i] = from->hashes[j];
+    atomic_set(&to->pointers[i], from->pointers[j]);
+
+    from->hashes[j] = 0;
+    atomic_set(&from->pointers[j], NULL);
+}
+
+/*
+ * 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);
+            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);
+                seqlock_write_end(&head->sequence);
+                return true;
+            }
+        }
+        b = b->next;
+    } while (b);
+    return false;
+}
+
+bool qht_remove(struct qht *ht, const void *p, uint32_t hash)
+{
+    struct qht_bucket *b;
+    struct qht_map *map;
+    bool ret;
+
+    /* NULL pointers are not supported */
+    qht_debug_assert(p);
+
+    b = qht_bucket_lock__no_stale(ht, hash, &map);
+    ret = qht_remove__locked(map, b, p, hash);
+    qht_bucket_debug__locked(b);
+    qemu_spin_unlock(&b->lock);
+    return ret;
+}
+
+static inline void qht_bucket_iter(struct qht *ht, struct qht_bucket *b,
+                                   qht_iter_func_t func, void *userp)
+{
+    int i;
+
+    do {
+        for (i = 0; i < QHT_BUCKET_ENTRIES; i++) {
+            if (b->pointers[i] == NULL) {
+                return;
+            }
+            func(ht, b->pointers[i], b->hashes[i], userp);
+        }
+        b = b->next;
+    } while (b);
+}
+
+/* call with all of the map's locks held */
+static inline void qht_map_iter__all_locked(struct qht *ht, struct qht_map *map,
+                                            qht_iter_func_t func, void *userp)
+{
+    size_t i;
+
+    for (i = 0; i < map->n_buckets; i++) {
+        qht_bucket_iter(ht, &map->buckets[i], func, userp);
+    }
+}
+
+void qht_iter(struct qht *ht, qht_iter_func_t func, void *userp)
+{
+    struct qht_map *map;
+
+    map = atomic_rcu_read(&ht->map);
+    qht_map_lock_buckets(map);
+    /* Note: ht here is merely for carrying ht->mode; ht->map won't be read */
+    qht_map_iter__all_locked(ht, map, func, userp);
+    qht_map_unlock_buckets(map);
+}
+
+static void qht_map_copy(struct qht *ht, void *p, uint32_t hash, void *userp)
+{
+    struct qht_map *new = userp;
+    struct qht_bucket *b = qht_map_to_bucket(new, hash);
+
+    /* no need to acquire b->lock because no thread has seen this map yet */
+    qht_insert__locked(ht, new, b, p, hash, NULL);
+}
+
+/*
+ * 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);
+}
+
+bool qht_resize(struct qht *ht, size_t n_elems)
+{
+    size_t n_buckets = qht_elems_to_buckets(n_elems);
+    size_t ret = false;
+
+    qemu_mutex_lock(&ht->lock);
+    if (n_buckets != ht->map->n_buckets) {
+        struct qht_map *new;
+        struct qht_map *old = ht->map;
+
+        new = qht_map_create(n_buckets);
+        qht_map_lock_buckets(old);
+        qht_do_resize(ht, new);
+        qht_map_unlock_buckets(old);
+        ret = true;
+    }
+    qemu_mutex_unlock(&ht->lock);
+
+    return ret;
+}
+
+static __attribute__((noinline)) void qht_grow_maybe(struct qht *ht)
+{
+    struct qht_map *map;
+
+    /*
+     * If the lock is taken it probably means there's an ongoing resize,
+     * so bail out.
+     */
+    if (qemu_mutex_trylock(&ht->lock)) {
+        return;
+    }
+    map = ht->map;
+    /* another thread might have just performed the resize we were after */
+    if (qht_map_needs_resize(map)) {
+        struct qht_map *new = qht_map_create(map->n_buckets * 2);
+
+        qht_map_lock_buckets(map);
+        qht_do_resize(ht, new);
+        qht_map_unlock_buckets(map);
+    }
+    qemu_mutex_unlock(&ht->lock);
+}
+
+/* pass @stats to qht_statistics_destroy() when done */
+void qht_statistics_init(struct qht *ht, struct qht_stats *stats)
+{
+    struct qht_map *map;
+    int i;
+
+    map = atomic_rcu_read(&ht->map);
+
+    stats->head_buckets = map->n_buckets;
+    stats->used_head_buckets = 0;
+    stats->entries = 0;
+    qdist_init(&stats->chain);
+    qdist_init(&stats->occupancy);
+
+    for (i = 0; i < map->n_buckets; i++) {
+        struct qht_bucket *head = &map->buckets[i];
+        struct qht_bucket *b;
+        unsigned int version;
+        size_t buckets;
+        size_t entries;
+        int j;
+
+        do {
+            version = seqlock_read_begin(&head->sequence);
+            buckets = 0;
+            entries = 0;
+            b = head;
+            do {
+                for (j = 0; j < QHT_BUCKET_ENTRIES; j++) {
+                    if (atomic_read(&b->pointers[j]) == NULL) {
+                        break;
+                    }
+                    entries++;
+                }
+                buckets++;
+                b = atomic_rcu_read(&b->next);
+            } while (b);
+        } while (seqlock_read_retry(&head->sequence, version));
+
+        if (entries) {
+            qdist_inc(&stats->chain, buckets);
+            qdist_inc(&stats->occupancy,
+                      (double)entries / QHT_BUCKET_ENTRIES / buckets);
+            stats->used_head_buckets++;
+            stats->entries += entries;
+        } else {
+            qdist_inc(&stats->occupancy, 0);
+        }
+    }
+}
+
+void qht_statistics_destroy(struct qht_stats *stats)
+{
+    qdist_destroy(&stats->occupancy);
+    qdist_destroy(&stats->chain);
+}