Message ID | 1465412133-3029-11-git-send-email-cota@braap.org (mailing list archive) |
---|---|
State | New, archived |
Headers | show |
On 08/06/16 21:55, Emilio G. Cota wrote: > 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> Reviewed-by: Sergey Fedorov <sergey.fedorov@linaro.org> > --- > include/qemu/qht.h | 183 ++++++++++++ > util/Makefile.objs | 1 + > util/qht.c | 833 +++++++++++++++++++++++++++++++++++++++++++++++++++++ > 3 files changed, 1017 insertions(+) > create mode 100644 include/qemu/qht.h > create mode 100644 util/qht.c > > 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..6f74909 > --- /dev/null > +++ b/util/qht.c > @@ -0,0 +1,833 @@ > +/* > + * 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 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; > + > + do { > + for (i = 0; i < QHT_BUCKET_ENTRIES; i++) { > + if (b->pointers[i]) { > + if (unlikely(b->pointers[i] == p)) { > + return false; > + } > + } else { > + goto found; > + } > + } > + prev = b; > + b = b->next; > + } while (b); > + > + b = qemu_memalign(QHT_BUCKET_ALIGN, sizeof(*b)); > + memset(b, 0, sizeof(*b)); > + new = b; > + i = 0; > + 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); > + 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; > +} > + > +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); > +} > + > +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_remove_entry(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_remove_entry(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_rcu(old, qht_map_destroy, rcu); > +} > + > +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; > +} > + > +/* 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); > +}
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..6f74909 --- /dev/null +++ b/util/qht.c @@ -0,0 +1,833 @@ +/* + * 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 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; + + do { + for (i = 0; i < QHT_BUCKET_ENTRIES; i++) { + if (b->pointers[i]) { + if (unlikely(b->pointers[i] == p)) { + return false; + } + } else { + goto found; + } + } + prev = b; + b = b->next; + } while (b); + + b = qemu_memalign(QHT_BUCKET_ALIGN, sizeof(*b)); + memset(b, 0, sizeof(*b)); + new = b; + i = 0; + 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); + 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; +} + +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); +} + +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_remove_entry(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_remove_entry(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_rcu(old, qht_map_destroy, rcu); +} + +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; +} + +/* 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); +}
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 | 833 +++++++++++++++++++++++++++++++++++++++++++++++++++++ 3 files changed, 1017 insertions(+) create mode 100644 include/qemu/qht.h create mode 100644 util/qht.c