/*
 * Copyright (c) 2012 Adam Hraska
 * All rights reserved.
 *
 * Redistribution and use in source and binary forms, with or without
 * modification, are permitted provided that the following conditions
 * are met:
 *
 * - Redistributions of source code must retain the above copyright
 *   notice, this list of conditions and the following disclaimer.
 * - Redistributions in binary form must reproduce the above copyright
 *   notice, this list of conditions and the following disclaimer in the
 *   documentation and/or other materials provided with the distribution.
 * - The name of the author may not be used to endorse or promote products
 *   derived from this software without specific prior written permission.
 *
 * THIS SOFTWARE IS PROVIDED BY THE AUTHOR ``AS IS'' AND ANY EXPRESS OR
 * IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES
 * OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED.
 * IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR ANY DIRECT, INDIRECT,
 * INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT
 * NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
 * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
 * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
 * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF
 * THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 */

/** @addtogroup genericadt
 * @{
 */

/**
 * @file
 * @brief Scalable resizable concurrent lock-free hash table.
 *
 * CHT is a concurrent hash table that is scalable resizable and lock-free.
 * resizable = the number of buckets of the table increases or decreases
 *     depending on the average number of elements per bucket (ie load)
 * scalable = accessing the table from more cpus increases performance
 *     almost linearly
 * lock-free = common operations never block; even if any of the operations
 *     is preempted or interrupted at any time, other operations will still
 *     make forward progress
 *
 * CHT is designed for read mostly scenarios. Performance degrades as the
 * fraction of updates (insert/remove) increases. Other data structures
 * significantly outperform CHT if the fraction of updates exceeds ~40%.
 *
 * CHT tolerates hardware exceptions and may be accessed from exception
 * handlers as long as the underlying RCU implementation is exception safe.
 *
 * @par Caveats
 *
 * 0) Never assume an item is still in the table.
 * The table may be accessed concurrently; therefore, other threads may
 * insert or remove an item at any time. Do not assume an item is still
 * in the table if cht_find() just returned it to you. Similarly, an
 * item may have already been inserted by the time cht_find() returns NULL.
 *
 * 1) Always use RCU read locks when searching the table.
 * Holding an RCU lock guarantees that an item found in the table remains
 * valid (eg is not freed) even if the item was removed from the table
 * in the meantime by another thread.
 *
 * 2) Never update values in place.
 * Do not update items in the table in place, ie directly. The changes
 * will not propagate to other readers (on other cpus) immediately or even
 * correctly. Some readers may then encounter items that have only some
 * of their fields changed or are completely inconsistent.
 *
 * Instead consider inserting an updated/changed copy of the item and
 * removing the original item. Or contact the maintainer to provide
 * you with a function that atomically replaces an item with a copy.
 *
 * 3) Use cht_insert_unique() instead of checking for duplicates with cht_find()
 * The following code is prone to race conditions:
 * @code
 * if (NULL == cht_find(&h, key)) {
 *     // If another thread inserts and item here, we'll insert a duplicate.
 *     cht_insert(&h, item);
 * }
 * @endcode
 * See cht_insert_unique() on how to correctly fix this.
 *
 *
 * @par Semantics
 *
 * Lazy readers = cht_find_lazy(), cht_find_next_lazy()
 * Readers = lazy readers, cht_find(), cht_find_next()
 * Updates = cht_insert(), cht_insert_unique(), cht_remove_key(),
 *     cht_remove_item()
 *
 * Readers (but not lazy readers) are guaranteed to see the effects
 * of @e completed updates. In other words, if cht_find() is invoked
 * after a cht_insert() @e returned eg on another cpu, cht_find() is
 * guaranteed to see the inserted item.
 *
 * Similarly, updates see the effects of @e completed updates. For example,
 * issuing cht_remove() after a cht_insert() for that key returned (even
 * on another cpu) is guaranteed to remove the inserted item.
 *
 * Reading or updating the table concurrently with other updates
 * always returns consistent data and never corrupts the table.
 * However the effects of concurrent updates may or may not be
 * visible to all other concurrent readers or updaters. Eg, not
 * all readers may see that an item has already been inserted
 * if cht_insert() has not yet returned.
 *
 * Lazy readers are guaranteed to eventually see updates but it
 * may take some time (possibly milliseconds) after the update
 * completes for the change to propagate to lazy readers on all
 * cpus.
 *
 * @par Implementation
 *
 * Collisions in CHT are resolved with chaining. The number of buckets
 * is always a power of 2. Each bucket is represented with a single linked
 * lock-free list [1]. Items in buckets are sorted by their mixed hashes
 * in ascending order. All buckets are terminated with a single global
 * sentinel node whose mixed hash value is the greatest possible.
 *
 * CHT with 2^k buckets uses the k most significant bits of a hash value
 * to determine the bucket number where an item is to be stored. To
 * avoid storing all items in a single bucket if the user supplied
 * hash function does not produce uniform hashes, hash values are
 * mixed first so that the top bits of a mixed hash change even if hash
 * values differ only in the least significant bits. The mixed hash
 * values are cached in cht_link.hash (which is overwritten once the
 * item is scheduled for removal via rcu_call).
 *
 * A new item is inserted before all other existing items in the bucket
 * with the same hash value as the newly inserted item (a la the original
 * lock-free list [2]). Placing new items at the start of a same-hash
 * sequence of items (eg duplicates) allows us to easily check for duplicates
 * in cht_insert_unique(). The function can first check that there are
 * no duplicates of the newly inserted item amongst the items with the
 * same hash as the new item. If there were no duplicates the new item
 * is linked before the same-hash items. Inserting a duplicate while
 * the function is checking for duplicates is detected as a change of
 * the link to the first checked same-hash item (and the search for
 * duplicates can be restarted).
 *
 * @par Table resize algorithm
 *
 * Table resize is based on [3] and [5]. First, a new bucket head array
 * is allocated and initialized. Second, old bucket heads are moved
 * to the new bucket head array with the protocol mentioned in [5].
 * At this point updaters start using the new bucket heads. Third,
 * buckets are split (or joined) so that the table can make use of
 * the extra bucket head slots in the new array (or stop wasting space
 * with the unnecessary extra slots in the old array). Splitting
 * or joining buckets employs a custom protocol. Last, the new array
 * replaces the original bucket array.
 *
 * A single background work item (of the system work queue) guides
 * resizing of the table. If an updater detects that the bucket it
 * is about to access is undergoing a resize (ie its head is moving
 * or it needs to be split/joined), it helps out and completes the
 * head move or the bucket split/join.
 *
 * The table always grows or shrinks by a factor of 2. Because items
 * are assigned a bucket based on the top k bits of their mixed hash
 * values, when growing the table each bucket is split into two buckets
 * and all items of the two new buckets come from the single bucket in the
 * original table. Ie items from separate buckets in the original table
 * never intermix in the new buckets. Moreover
 * since the buckets are sorted by their mixed hash values the items
 * at the beginning of the old bucket will end up in the first new
 * bucket while all the remaining items of the old bucket will end up
 * in the second new bucket. Therefore, there is a single point where
 * to split the linked list of the old bucket into two correctly sorted
 * linked lists of the new buckets:
 *                            .- bucket split
 *                            |
 *             <-- first -->  v  <-- second -->
 *   [old] --> [00b] -> [01b] -> [10b] -> [11b] -> sentinel
 *              ^                 ^
 *   [new0] -- -+                 |
 *   [new1] -- -- -- -- -- -- -- -+
 *
 * Resize in greater detail:
 *
 * a) First, a resizer (a single background system work queue item
 * in charge of resizing the table) allocates and initializes a new
 * bucket head array. New bucket heads are pointed to the sentinel
 * and marked Invalid (in the lower order bits of the pointer to the
 * next item, ie the sentinel in this case):
 *
 *   [old, N] --> [00b] -> [01b] -> [10b] -> [11b] -> sentinel
 *                                                    ^ ^
 *   [new0, Inv] -------------------------------------+ |
 *   [new1, Inv] ---------------------------------------+
 *
 *
 * b) Second, the resizer starts moving old bucket heads with the following
 * lock-free protocol (from [5]) where cas(variable, expected_val, new_val)
 * is short for compare-and-swap:
 *
 *   old head     new0 head      transition to next state
 *   --------     ---------      ------------------------
 *   addr, N      sentinel, Inv  cas(old, (addr, N), (addr, Const))
 *                               .. mark the old head as immutable, so that
 *                                  updaters do not relink it to other nodes
 *                                  until the head move is done.
 *   addr, Const  sentinel, Inv  cas(new0, (sentinel, Inv), (addr, N))
 *                               .. move the address to the new head and mark
 *                                  the new head normal so updaters can start
 *                                  using it.
 *   addr, Const  addr, N        cas(old, (addr, Const), (addr, Inv))
 *                               .. mark the old head Invalid to signify
 *                                  the head move is done.
 *   addr, Inv    addr, N
 *
 * Notice that concurrent updaters may step in at any point and correctly
 * complete the head move without disrupting the resizer. At worst, the
 * resizer or other concurrent updaters will attempt a number of CAS() that
 * will correctly fail.
 *
 *   [old, Inv] -> [00b] -> [01b] -> [10b] -> [11b] -> sentinel
 *                 ^                                   ^
 *   [new0, N] ----+                                   |
 *   [new1, Inv] --------------------------------------+
 *
 *
 * c) Third, buckets are split if the table is growing; or joined if
 * shrinking (by the resizer or updaters depending on whoever accesses
 * the bucket first). See split_bucket() and join_buckets() for details.
 *
 *  1) Mark the last item of new0 with JOIN_FOLLOWS:
 *   [old, Inv] -> [00b] -> [01b, JF] -> [10b] -> [11b] -> sentinel
 *                 ^                                       ^
 *   [new0, N] ----+                                       |
 *   [new1, Inv] ------------------------------------------+
 *
 *  2) Mark the first item of new1 with JOIN_NODE:
 *   [old, Inv] -> [00b] -> [01b, JF] -> [10b, JN] -> [11b] -> sentinel
 *                 ^                                           ^
 *   [new0, N] ----+                                           |
 *   [new1, Inv] ----------------------------------------------+
 *
 *  3) Point new1 to the join-node and mark new1 NORMAL.
 *   [old, Inv] -> [00b] -> [01b, JF] -> [10b, JN] -> [11b] -> sentinel
 *                 ^                     ^
 *   [new0, N] ----+                     |
 *   [new1, N] --------------------------+
 *
 *
 * d) Fourth, the resizer cleans up extra marks added during bucket
 * splits/joins but only when it is sure all updaters are accessing
 * the table via the new bucket heads only (ie it is certain there
 * are no delayed updaters unaware of the resize and accessing the
 * table via the old bucket head).
 *
 *   [old, Inv] ---+
 *                 v
 *   [new0, N] --> [00b] -> [01b, N] ---+
 *                                      v
 *   [new1, N] --> [10b, N] -> [11b] -> sentinel
 *
 *
 * e) Last, the resizer publishes the new bucket head array for everyone
 * to see and use. This signals the end of the resize and the old bucket
 * array is freed.
 *
 *
 * To understand details of how the table is resized, read [1, 3, 5]
 * and comments in join_buckets(), split_bucket().
 *
 *
 * [1] High performance dynamic lock-free hash tables and list-based sets,
 *     Michael, 2002
 *     http://www.research.ibm.com/people/m/michael/spaa-2002.pdf
 * [2] Lock-free linked lists using compare-and-swap,
 *     Valois, 1995
 *     http://people.csail.mit.edu/bushl2/rpi/portfolio/lockfree-grape/documents/lock-free-linked-lists.pdf
 * [3] Resizable, scalable, concurrent hash tables via relativistic programming,
 *     Triplett, 2011
 *     http://www.usenix.org/event/atc11/tech/final_files/Triplett.pdf
 * [4] Split-ordered Lists: Lock-free Extensible Hash Tables,
 *     Shavit, 2006
 *     http://www.cs.ucf.edu/~dcm/Teaching/COT4810-Spring2011/Literature/SplitOrderedLists.pdf
 * [5] Towards a Scalable Non-blocking Coding Style,
 *     Click, 2008
 *     http://www.azulsystems.com/events/javaone_2008/2008_CodingNonBlock.pdf
 */

#include <adt/cht.h>
#include <adt/hash.h>
#include <assert.h>
#include <mm/slab.h>
#include <barrier.h>
#include <barrier.h>
#include <atomic.h>
#include <synch/rcu.h>

/* Logarithm of the min bucket count. Must be at least 3. 2^6 == 64 buckets. */
#define CHT_MIN_ORDER 6
/* Logarithm of the max bucket count. */
#define CHT_MAX_ORDER (8 * sizeof(size_t))
/* Minimum number of hash table buckets. */
#define CHT_MIN_BUCKET_CNT (1 << CHT_MIN_ORDER)
/* Does not have to be a power of 2. */
#define CHT_MAX_LOAD 2

typedef cht_ptr_t marked_ptr_t;
typedef bool (*equal_pred_t)(void *arg, const cht_link_t *item);

/** The following mark items and bucket heads.
 *
 * They are stored in the two low order bits of the next item pointers.
 * Some marks may be combined. Some marks share the same binary value and
 * are distinguished only by context (eg bucket head vs an ordinary item),
 * in particular by walk_mode_t.
 */
typedef enum mark {
	/** Normal non-deleted item or a valid bucket head. */
	N_NORMAL = 0,
	/** Logically deleted item that might have already been unlinked.
	 *
	 * May be combined with N_JOIN and N_JOIN_FOLLOWS. Applicable only
	 * to items; never to bucket heads.
	 *
	 * Once marked deleted an item remains marked deleted.
	 */
	N_DELETED = 1,
	/** Immutable bucket head.
	 *
	 * The bucket is being moved or joined with another and its (old) head
	 * must not be modified.
	 *
	 * May be combined with N_INVALID. Applicable only to old bucket heads,
	 * ie cht_t.b and not cht_t.new_b.
	 */
	N_CONST = 1,
	/** Invalid bucket head. The bucket head must not be modified.
	 *
	 * Old bucket heads (ie cht_t.b) are marked invalid if they have
	 * already been moved to cht_t.new_b or if the bucket had already
	 * been merged with another when shrinking the table. New bucket
	 * heads (ie cht_t.new_b) are marked invalid if the old bucket had
	 * not yet been moved or if an old bucket had not yet been split
	 * when growing the table.
	 */
	N_INVALID = 3,
	/** The item is a join node, ie joining two buckets
	 *
	 * A join node is either the first node of the second part of
	 * a bucket to be split; or it is the first node of the bucket
	 * to be merged into/appended to/joined with another bucket.
	 *
	 * May be combined with N_DELETED. Applicable only to items, never
	 * to bucket heads.
	 *
	 * Join nodes are referred to from two different buckets and may,
	 * therefore, not be safely/atomically unlinked from both buckets.
	 * As a result join nodes are not unlinked but rather just marked
	 * deleted. Once resize completes join nodes marked deleted are
	 * garbage collected.
	 */
	N_JOIN = 2,
	/** The next node is a join node and will soon be marked so.
	 *
	 * A join-follows node is the last node of the first part of bucket
	 * that is to be split, ie it is the last node that will remain
	 * in the same bucket after splitting it.
	 *
	 * May be combined with N_DELETED. Applicable to items as well
	 * as to bucket heads of the bucket to be split (but only in cht_t.new_b).
	 */
	N_JOIN_FOLLOWS = 2,
	/** Bit mask to filter out the address to the next item from the next ptr. */
	N_MARK_MASK = 3
} mark_t;

/** Determines */
typedef enum walk_mode {
	/** The table is not resizing. */
	WM_NORMAL = 4,
	/** The table is undergoing a resize. Join nodes may be encountered. */
	WM_LEAVE_JOIN,
	/** The table is growing. A join-follows node may be encountered. */
	WM_MOVE_JOIN_FOLLOWS
} walk_mode_t;

/** Bucket position window. */
typedef struct wnd {
	/** Pointer to cur's predecessor. */
	marked_ptr_t *ppred;
	/** Current item. */
	cht_link_t *cur;
	/** Last encountered item. Deleted or not. */
	cht_link_t *last;
} wnd_t;

/* Sentinel node used by all buckets. Stores the greatest possible hash value.*/
static const cht_link_t sentinel = {
	/* NULL and N_NORMAL */
	.link = 0 | N_NORMAL,
	.hash = -1
};

static size_t size_to_order(size_t bucket_cnt, size_t min_order);
static cht_buckets_t *alloc_buckets(size_t order, bool set_invalid,
    bool can_block);
static inline cht_link_t *find_lazy(cht_t *h, void *key);
static cht_link_t *search_bucket(cht_t *h, marked_ptr_t head, void *key,
    size_t search_hash);
static cht_link_t *find_resizing(cht_t *h, void *key, size_t hash,
    marked_ptr_t old_head, size_t old_idx);
static bool insert_impl(cht_t *h, cht_link_t *item, cht_link_t **dup_item);
static bool insert_at(cht_link_t *item, const wnd_t *wnd, walk_mode_t walk_mode,
    bool *resizing);
static bool has_duplicate(cht_t *h, const cht_link_t *item, size_t hash,
    cht_link_t *cur, cht_link_t **dup_item);
static cht_link_t *find_duplicate(cht_t *h, const cht_link_t *item, size_t hash,
    cht_link_t *start);
static bool remove_pred(cht_t *h, size_t hash, equal_pred_t pred, void *pred_arg);
static bool delete_at(cht_t *h, wnd_t *wnd, walk_mode_t walk_mode,
    bool *deleted_but_gc, bool *resizing);
static bool mark_deleted(cht_link_t *cur, walk_mode_t walk_mode, bool *resizing);
static bool unlink_from_pred(wnd_t *wnd, walk_mode_t walk_mode, bool *resizing);
static bool find_wnd_and_gc_pred(cht_t *h, size_t hash, walk_mode_t walk_mode,
    equal_pred_t pred, void *pred_arg, wnd_t *wnd, bool *resizing);
static bool find_wnd_and_gc(cht_t *h, size_t hash, walk_mode_t walk_mode,
    wnd_t *wnd, bool *resizing);
static bool gc_deleted_node(cht_t *h, walk_mode_t walk_mode, wnd_t *wnd,
    bool *resizing);
static bool join_completed(cht_t *h, const wnd_t *wnd);
static void upd_resizing_head(cht_t *h, size_t hash, marked_ptr_t **phead,
    bool *join_finishing,  walk_mode_t *walk_mode);
static void item_removed(cht_t *h);
static void item_inserted(cht_t *h);
static void free_later(cht_t *h, cht_link_t *item);
static void help_head_move(marked_ptr_t *psrc_head, marked_ptr_t *pdest_head);
static void start_head_move(marked_ptr_t *psrc_head);
static void mark_const(marked_ptr_t *psrc_head);
static void complete_head_move(marked_ptr_t *psrc_head, marked_ptr_t *pdest_head);
static void split_bucket(cht_t *h, marked_ptr_t *psrc_head,
    marked_ptr_t *pdest_head, size_t split_hash);
static void mark_join_follows(cht_t *h, marked_ptr_t *psrc_head,
    size_t split_hash, wnd_t *wnd);
static void mark_join_node(cht_link_t *join_node);
static void join_buckets(cht_t *h, marked_ptr_t *psrc_head,
    marked_ptr_t *pdest_head, size_t split_hash);
static void link_to_join_node(cht_t *h, marked_ptr_t *pdest_head,
    cht_link_t *join_node, size_t split_hash);
static void resize_table(work_t *arg);
static void grow_table(cht_t *h);
static void shrink_table(cht_t *h);
static void cleanup_join_node(cht_t *h, marked_ptr_t *new_head);
static void clear_join_and_gc(cht_t *h, cht_link_t *join_node,
    marked_ptr_t *new_head);
static void cleanup_join_follows(cht_t *h, marked_ptr_t *new_head);
static marked_ptr_t make_link(const cht_link_t *next, mark_t mark);
static cht_link_t *get_next(marked_ptr_t link);
static mark_t get_mark(marked_ptr_t link);
static void next_wnd(wnd_t *wnd);
static bool same_node_pred(void *node, const cht_link_t *item2);
static size_t calc_key_hash(cht_t *h, void *key);
static size_t node_hash(cht_t *h, const cht_link_t *item);
static size_t calc_node_hash(cht_t *h, const cht_link_t *item);
static void memoize_node_hash(cht_t *h, cht_link_t *item);
static size_t calc_split_hash(size_t split_idx, size_t order);
static size_t calc_bucket_idx(size_t hash, size_t order);
static size_t grow_to_split_idx(size_t old_idx);
static size_t grow_idx(size_t idx);
static size_t shrink_idx(size_t idx);
static marked_ptr_t cas_link(marked_ptr_t *link, const cht_link_t *cur_next,
    mark_t cur_mark, const cht_link_t *new_next, mark_t new_mark);
static marked_ptr_t _cas_link(marked_ptr_t *link, marked_ptr_t cur,
    marked_ptr_t new);
static void cas_order_barrier(void);

static void dummy_remove_callback(cht_link_t *item)
{
	/* empty */
}

/** Creates a concurrent hash table.
 *
 * @param h         Valid pointer to a cht_t instance.
 * @param op        Item specific operations. All operations are compulsory.
 * @return True if successfully created the table. False otherwise.
 */
bool cht_create_simple(cht_t *h, cht_ops_t *op)
{
	return cht_create(h, 0, 0, 0, false, op);
}

/** Creates a concurrent hash table.
 *
 * @param h         Valid pointer to a cht_t instance.
 * @param init_size The initial number of buckets the table should contain.
 *                  The table may be shrunk below this value if deemed necessary.
 *                  Uses the default value if 0.
 * @param min_size  Minimum number of buckets that the table should contain.
 *                  The number of buckets never drops below this value,
 *                  although it may be rounded up internally as appropriate.
 *                  Uses the default value if 0.
 * @param max_load  Maximum average number of items per bucket that allowed
 *                  before the table grows.
 * @param can_block If true creating the table blocks until enough memory
 *                  is available (possibly indefinitely). Otherwise,
 *                  table creation does not block and returns immediately
 *                  even if not enough memory is available.
 * @param op        Item specific operations. All operations are compulsory.
 * @return True if successfully created the table. False otherwise.
 */
bool cht_create(cht_t *h, size_t init_size, size_t min_size, size_t max_load,
    bool can_block, cht_ops_t *op)
{
	assert(h);
	assert(op && op->hash && op->key_hash && op->equal && op->key_equal);
	/* Memoized hashes are stored in the rcu_link.func function pointer. */
	static_assert(sizeof(size_t) == sizeof(rcu_func_t), "");
	assert(sentinel.hash == (uintptr_t)sentinel.rcu_link.func);

	/* All operations are compulsory. */
	if (!op || !op->hash || !op->key_hash || !op->equal || !op->key_equal)
		return false;

	size_t min_order = size_to_order(min_size, CHT_MIN_ORDER);
	size_t order = size_to_order(init_size, min_order);

	h->b = alloc_buckets(order, false, can_block);

	if (!h->b)
		return false;

	h->max_load = (max_load == 0) ? CHT_MAX_LOAD : max_load;
	h->min_order = min_order;
	h->new_b = NULL;
	h->op = op;
	atomic_store(&h->item_cnt, 0);
	atomic_store(&h->resize_reqs, 0);

	if (NULL == op->remove_callback) {
		h->op->remove_callback = dummy_remove_callback;
	}

	/*
	 * Cached item hashes are stored in item->rcu_link.func. Once the item
	 * is deleted rcu_link.func will contain the value of invalid_hash.
	 */
	h->invalid_hash = (uintptr_t)h->op->remove_callback;

	/* Ensure the initialization takes place before we start using the table. */
	write_barrier();

	return true;
}

/** Allocates and initializes 2^order buckets.
 *
 * All bucket heads are initialized to point to the sentinel node.
 *
 * @param order       The number of buckets to allocate is 2^order.
 * @param set_invalid Bucket heads are marked invalid if true; otherwise
 *                    they are marked N_NORMAL.
 * @param can_block   If true memory allocation blocks until enough memory
 *                    is available (possibly indefinitely). Otherwise,
 *                    memory allocation does not block.
 * @return Newly allocated and initialized buckets or NULL if not enough memory.
 */
static cht_buckets_t *alloc_buckets(size_t order, bool set_invalid, bool can_block)
{
	size_t bucket_cnt = (1 << order);
	size_t bytes =
	    sizeof(cht_buckets_t) + (bucket_cnt - 1) * sizeof(marked_ptr_t);
	cht_buckets_t *b = can_block ? nfmalloc(bytes) : malloc(bytes);

	if (!b)
		return NULL;

	b->order = order;

	marked_ptr_t head_link = set_invalid ?
	    make_link(&sentinel, N_INVALID) :
	    make_link(&sentinel, N_NORMAL);

	for (size_t i = 0; i < bucket_cnt; ++i) {
		b->head[i] = head_link;
	}

	return b;
}

/** Returns the smallest k such that bucket_cnt <= 2^k and min_order <= k.*/
static size_t size_to_order(size_t bucket_cnt, size_t min_order)
{
	size_t order = min_order;

	/* Find a power of two such that bucket_cnt <= 2^order */
	do {
		if (bucket_cnt <= ((size_t)1 << order))
			return order;

		++order;
	} while (order < CHT_MAX_ORDER);

	return order;
}

/** Destroys a CHT successfully created via cht_create().
 *
 * Waits for all outstanding concurrent operations to complete and
 * frees internal allocated resources. The table is however not cleared
 * and items already present in the table (if any) are leaked.
 */
void cht_destroy(cht_t *h)
{
	cht_destroy_unsafe(h);

	/* You must clear the table of items. Otherwise cht_destroy will leak. */
	assert(atomic_load(&h->item_cnt) == 0);
}

/** Destroys a successfully created CHT but does no error checking. */
void cht_destroy_unsafe(cht_t *h)
{
	/* Wait for resize to complete. */
	while (0 < atomic_load(&h->resize_reqs)) {
		rcu_barrier();
	}

	/* Wait for all remove_callback()s to complete. */
	rcu_barrier();

	free(h->b);
	h->b = NULL;
}

/** Returns the first item equal to the search key or NULL if not found.
 *
 * The call must be enclosed in a rcu_read_lock() unlock() pair. The
 * returned item is guaranteed to be allocated until rcu_read_unlock()
 * although the item may be concurrently removed from the table by another
 * cpu.
 *
 * Further items matching the key may be retrieved via cht_find_next().
 *
 * cht_find() sees the effects of any completed cht_remove(), cht_insert().
 * If a concurrent remove or insert had not yet completed cht_find() may
 * or may not see the effects of it (eg it may find an item being removed).
 *
 * @param h   CHT to operate on.
 * @param key Search key as defined by cht_ops_t.key_equal() and .key_hash().
 * @return First item equal to the key or NULL if such an item does not exist.
 */
cht_link_t *cht_find(cht_t *h, void *key)
{
	/* Make the most recent changes to the table visible. */
	read_barrier();
	return cht_find_lazy(h, key);
}

/** Returns the first item equal to the search key or NULL if not found.
 *
 * Unlike cht_find(), cht_find_lazy() may not see the effects of
 * cht_remove() or cht_insert() even though they have already completed.
 * It may take a couple of milliseconds for those changes to propagate
 * and become visible to cht_find_lazy(). On the other hand, cht_find_lazy()
 * operates a bit faster than cht_find().
 *
 * See cht_find() for more details.
 */
cht_link_t *cht_find_lazy(cht_t *h, void *key)
{
	return find_lazy(h, key);
}

/** Finds the first item equal to the search key. */
static inline cht_link_t *find_lazy(cht_t *h, void *key)
{
	assert(h);
	/* See docs to cht_find() and cht_find_lazy(). */
	assert(rcu_read_locked());

	size_t hash = calc_key_hash(h, key);

	cht_buckets_t *b = rcu_access(h->b);
	size_t idx = calc_bucket_idx(hash, b->order);
	/*
	 * No need for access_once. b->head[idx] will point to an allocated node
	 * even if marked invalid until we exit rcu read section.
	 */
	marked_ptr_t head = b->head[idx];

	/* Undergoing a resize - take the slow path. */
	if (N_INVALID == get_mark(head))
		return find_resizing(h, key, hash, head, idx);

	return search_bucket(h, head, key, hash);
}

/** Returns the next item matching \a item.
 *
 * Must be enclosed in a rcu_read_lock()/unlock() pair. Effects of
 * completed cht_remove(), cht_insert() are guaranteed to be visible
 * to cht_find_next().
 *
 * See cht_find() for more details.
 */
cht_link_t *cht_find_next(cht_t *h, const cht_link_t *item)
{
	/* Make the most recent changes to the table visible. */
	read_barrier();
	return cht_find_next_lazy(h, item);
}

/** Returns the next item matching \a item.
 *
 * Must be enclosed in a rcu_read_lock()/unlock() pair. Effects of
 * completed cht_remove(), cht_insert() may or may not be visible
 * to cht_find_next_lazy().
 *
 * See cht_find_lazy() for more details.
 */
cht_link_t *cht_find_next_lazy(cht_t *h, const cht_link_t *item)
{
	assert(h);
	assert(rcu_read_locked());
	assert(item);

	return find_duplicate(h, item, calc_node_hash(h, item), get_next(item->link));
}

/** Searches the bucket at head for key using search_hash. */
static inline cht_link_t *search_bucket(cht_t *h, marked_ptr_t head, void *key,
    size_t search_hash)
{
	/*
	 * It is safe to access nodes even outside of this bucket (eg when
	 * splitting the bucket). The resizer makes sure that any node we
	 * may find by following the next pointers is allocated.
	 */

	cht_link_t *cur = NULL;
	marked_ptr_t prev = head;

try_again:
	/* Filter out items with different hashes. */
	do {
		cur = get_next(prev);
		assert(cur);
		prev = cur->link;
	} while (node_hash(h, cur) < search_hash);

	/*
	 * Only search for an item with an equal key if cur is not the sentinel
	 * node or a node with a different hash.
	 */
	while (node_hash(h, cur) == search_hash) {
		if (h->op->key_equal(key, cur)) {
			if (!(N_DELETED & get_mark(cur->link)))
				return cur;
		}

		cur = get_next(cur->link);
		assert(cur);
	}

	/*
	 * In the unlikely case that we have encountered a node whose cached
	 * hash has been overwritten due to a pending rcu_call for it, skip
	 * the node and try again.
	 */
	if (node_hash(h, cur) == h->invalid_hash) {
		prev = cur->link;
		goto try_again;
	}

	return NULL;
}

/** Searches for the key while the table is undergoing a resize. */
static cht_link_t *find_resizing(cht_t *h, void *key, size_t hash,
    marked_ptr_t old_head, size_t old_idx)
{
	assert(N_INVALID == get_mark(old_head));
	assert(h->new_b);

	size_t new_idx = calc_bucket_idx(hash, h->new_b->order);
	marked_ptr_t new_head = h->new_b->head[new_idx];
	marked_ptr_t search_head = new_head;

	/* Growing. */
	if (h->b->order < h->new_b->order) {
		/*
		 * Old bucket head is invalid, so it must have been already
		 * moved. Make the new head visible if still not visible, ie
		 * invalid.
		 */
		if (N_INVALID == get_mark(new_head)) {
			/*
			 * We should be searching a newly added bucket but the old
			 * moved bucket has not yet been split (its marked invalid)
			 * or we have not yet seen the split.
			 */
			if (grow_idx(old_idx) != new_idx) {
				/*
				 * Search the moved bucket. It is guaranteed to contain
				 * items of the newly added bucket that were present
				 * before the moved bucket was split.
				 */
				new_head = h->new_b->head[grow_idx(old_idx)];
			}

			/* new_head is now the moved bucket, either valid or invalid. */

			/*
			 * The old bucket was definitely moved to new_head but the
			 * change of new_head had not yet propagated to this cpu.
			 */
			if (N_INVALID == get_mark(new_head)) {
				/*
				 * We could issue a read_barrier() and make the now valid
				 * moved bucket head new_head visible, but instead fall back
				 * on using the old bucket. Although the old bucket head is
				 * invalid, it points to a node that is allocated and in the
				 * right bucket. Before the node can be freed, it must be
				 * unlinked from the head (or another item after that item
				 * modified the new_head) and a grace period must elapse.
				 * As a result had the node been already freed the grace
				 * period preceeding the free() would make the unlink and
				 * any changes to new_head visible. Therefore, it is safe
				 * to use the node pointed to from the old bucket head.
				 */

				search_head = old_head;
			} else {
				search_head = new_head;
			}
		}

		return search_bucket(h, search_head, key, hash);
	} else if (h->b->order > h->new_b->order) {
		/* Shrinking. */

		/* Index of the bucket in the old table that was moved. */
		size_t move_src_idx = grow_idx(new_idx);
		marked_ptr_t moved_old_head = h->b->head[move_src_idx];

		/*
		 * h->b->head[move_src_idx] had already been moved to new_head
		 * but the change to new_head had not yet propagated to us.
		 */
		if (N_INVALID == get_mark(new_head)) {
			/*
			 * new_head is definitely valid and we could make it visible
			 * to this cpu with a read_barrier(). Instead, use the bucket
			 * in the old table that was moved even though it is now marked
			 * as invalid. The node it points to must be allocated because
			 * a grace period would have to elapse before it could be freed;
			 * and the grace period would make the now valid new_head
			 * visible to all cpus.
			 *
			 * Note that move_src_idx may not be the same as old_idx.
			 * If move_src_idx != old_idx then old_idx is the bucket
			 * in the old table that is not moved but instead it is
			 * appended to the moved bucket, ie it is added at the tail
			 * of new_head. In that case an invalid old_head notes that
			 * it had already been merged into (the moved) new_head.
			 * We will try to search that bucket first because it
			 * may contain some newly added nodes after the bucket
			 * join. Moreover, the bucket joining link may already be
			 * visible even if new_head is not. Therefore, if we're
			 * lucky we'll find the item via moved_old_head. In any
			 * case, we'll retry in proper old_head if not found.
			 */
			search_head = moved_old_head;
		}

		cht_link_t *ret = search_bucket(h, search_head, key, hash);

		if (ret)
			return ret;
		/*
		 * Bucket old_head was already joined with moved_old_head
		 * in the new table but we have not yet seen change of the
		 * joining link (or the item is not in the table).
		 */
		if (move_src_idx != old_idx && get_next(old_head) != &sentinel) {
			/*
			 * Note that old_head (the bucket to be merged into new_head)
			 * points to an allocated join node (if non-null) even if marked
			 * invalid. Before the resizer lets join nodes to be unlinked
			 * (and freed) it sets old_head to NULL and waits for a grace period.
			 * So either the invalid old_head points to join node; or old_head
			 * is null and we would have seen a completed bucket join while
			 * traversing search_head.
			 */
			assert(N_JOIN & get_mark(get_next(old_head)->link));
			return search_bucket(h, old_head, key, hash);
		}

		return NULL;
	} else {
		/*
		 * Resize is almost done. The resizer is waiting to make
		 * sure all cpus see that the new table replaced the old one.
		 */
		assert(h->b->order == h->new_b->order);
		/*
		 * The resizer must ensure all new bucket heads are visible before
		 * replacing the old table.
		 */
		assert(N_NORMAL == get_mark(new_head));
		return search_bucket(h, new_head, key, hash);
	}
}

/** Inserts an item. Succeeds even if an equal item is already present. */
void cht_insert(cht_t *h, cht_link_t *item)
{
	insert_impl(h, item, NULL);
}

/** Inserts a unique item. Returns false if an equal item was already present.
 *
 * Use this function to atomically check if an equal/duplicate item had
 * not yet been inserted into the table and to insert this item into the
 * table.
 *
 * The following is @e NOT thread-safe, so do not use:
 * @code
 * if (!cht_find(h, key)) {
 *     // A concurrent insert here may go unnoticed by cht_find() above.
 *     item = malloc(..);
 *     cht_insert(h, item);
 *     // Now we may have two items with equal search keys.
 * }
 * @endcode
 *
 * Replace such code with:
 * @code
 * item = malloc(..);
 * if (!cht_insert_unique(h, item, &dup_item)) {
 *     // Whoops, someone beat us to it - an equal item 'dup_item'
 *     // had already been inserted.
 *     free(item);
 * } else {
 *     // Successfully inserted the item and we are guaranteed that
 *     // there are no other equal items.
 * }
 * @endcode
 *
 */
bool cht_insert_unique(cht_t *h, cht_link_t *item, cht_link_t **dup_item)
{
	assert(rcu_read_locked());
	assert(dup_item);
	return insert_impl(h, item, dup_item);
}

/** Inserts the item into the table and checks for duplicates if dup_item. */
static bool insert_impl(cht_t *h, cht_link_t *item, cht_link_t **dup_item)
{
	rcu_read_lock();

	cht_buckets_t *b = rcu_access(h->b);
	memoize_node_hash(h, item);
	size_t hash = node_hash(h, item);
	size_t idx = calc_bucket_idx(hash, b->order);
	marked_ptr_t *phead = &b->head[idx];

	bool resizing = false;
	bool inserted = false;

	do {
		walk_mode_t walk_mode = WM_NORMAL;
		bool join_finishing;

		resizing = resizing || (N_NORMAL != get_mark(*phead));

		/* The table is resizing. Get the correct bucket head. */
		if (resizing) {
			upd_resizing_head(h, hash, &phead, &join_finishing, &walk_mode);
		}

		wnd_t wnd = {
			.ppred = phead,
			.cur = get_next(*phead),
			.last = NULL
		};

		if (!find_wnd_and_gc(h, hash, walk_mode, &wnd, &resizing)) {
			/* Could not GC a node; or detected an unexpected resize. */
			continue;
		}

		if (dup_item && has_duplicate(h, item, hash, wnd.cur, dup_item)) {
			rcu_read_unlock();
			return false;
		}

		inserted = insert_at(item, &wnd, walk_mode, &resizing);
	} while (!inserted);

	rcu_read_unlock();

	item_inserted(h);
	return true;
}

/** Inserts item between wnd.ppred and wnd.cur.
 *
 * @param item      Item to link to wnd.ppred and wnd.cur.
 * @param wnd       The item will be inserted before wnd.cur. Wnd.ppred
 *                  must be N_NORMAL.
 * @param walk_mode
 * @param resizing  Set to true only if the table is undergoing resize
 *         and it was not expected (ie walk_mode == WM_NORMAL).
 * @return True if the item was successfully linked to wnd.ppred. False
 *         if whole insert operation must be retried because the predecessor
 *         of wnd.cur has changed.
 */
inline static bool insert_at(cht_link_t *item, const wnd_t *wnd,
    walk_mode_t walk_mode, bool *resizing)
{
	marked_ptr_t ret;

	if (walk_mode == WM_NORMAL) {
		item->link = make_link(wnd->cur, N_NORMAL);
		/* Initialize the item before adding it to a bucket. */
		memory_barrier();

		/* Link a clean/normal predecessor to the item. */
		ret = cas_link(wnd->ppred, wnd->cur, N_NORMAL, item, N_NORMAL);

		if (ret == make_link(wnd->cur, N_NORMAL)) {
			return true;
		} else {
			/* This includes an invalid head but not a const head. */
			*resizing = ((N_JOIN_FOLLOWS | N_JOIN) & get_mark(ret));
			return false;
		}
	} else if (walk_mode == WM_MOVE_JOIN_FOLLOWS) {
		/* Move JOIN_FOLLOWS mark but filter out the DELETED mark. */
		mark_t jf_mark = get_mark(*wnd->ppred) & N_JOIN_FOLLOWS;
		item->link = make_link(wnd->cur, jf_mark);
		/* Initialize the item before adding it to a bucket. */
		memory_barrier();

		/* Link the not-deleted predecessor to the item. Move its JF mark. */
		ret = cas_link(wnd->ppred, wnd->cur, jf_mark, item, N_NORMAL);

		return ret == make_link(wnd->cur, jf_mark);
	} else {
		assert(walk_mode == WM_LEAVE_JOIN);

		item->link = make_link(wnd->cur, N_NORMAL);
		/* Initialize the item before adding it to a bucket. */
		memory_barrier();

		mark_t pred_mark = get_mark(*wnd->ppred);
		/* If the predecessor is a join node it may be marked deleted.*/
		mark_t exp_pred_mark = (N_JOIN & pred_mark) ? pred_mark : N_NORMAL;

		ret = cas_link(wnd->ppred, wnd->cur, exp_pred_mark, item, exp_pred_mark);
		return ret == make_link(wnd->cur, exp_pred_mark);
	}
}

/** Returns true if the chain starting at cur has an item equal to \a item.
 *
 * @param h    CHT to operate on.
 * @param item Item whose duplicates the function looks for.
 * @param hash Hash of \a item.
 * @param[in] cur  The first node with a hash greater to or equal to item's hash.
 * @param[out] dup_item The first duplicate item encountered.
 * @return True if a non-deleted item equal to \a item exists in the table.
 */
static inline bool has_duplicate(cht_t *h, const cht_link_t *item, size_t hash,
    cht_link_t *cur, cht_link_t **dup_item)
{
	assert(cur);
	assert(cur == &sentinel || hash <= node_hash(h, cur) ||
	    node_hash(h, cur) == h->invalid_hash);

	/* hash < node_hash(h, cur) */
	if (hash != node_hash(h, cur) && h->invalid_hash != node_hash(h, cur))
		return false;

	/*
	 * Load the most recent node marks. Otherwise we might pronounce a
	 * logically deleted node for a duplicate of the item just because
	 * the deleted node's DEL mark had not yet propagated to this cpu.
	 */
	read_barrier();

	*dup_item = find_duplicate(h, item, hash, cur);
	return NULL != *dup_item;
}

/** Returns an item that is equal to \a item starting in a chain at \a start. */
static cht_link_t *find_duplicate(cht_t *h, const cht_link_t *item, size_t hash,
    cht_link_t *start)
{
	assert(hash <= node_hash(h, start) || h->invalid_hash == node_hash(h, start));

	cht_link_t *cur = start;

try_again:
	assert(cur);

	while (node_hash(h, cur) == hash) {
		assert(cur != &sentinel);

		bool deleted = (N_DELETED & get_mark(cur->link));

		/* Skip logically deleted nodes. */
		if (!deleted && h->op->equal(item, cur))
			return cur;

		cur = get_next(cur->link);
		assert(cur);
	}

	/* Skip logically deleted nodes with rcu_call() in progress. */
	if (h->invalid_hash == node_hash(h, cur)) {
		cur = get_next(cur->link);
		goto try_again;
	}

	return NULL;
}

/** Removes all items matching the search key. Returns the number of items removed.*/
size_t cht_remove_key(cht_t *h, void *key)
{
	assert(h);

	size_t hash = calc_key_hash(h, key);
	size_t removed = 0;

	while (remove_pred(h, hash, h->op->key_equal, key))
		++removed;

	return removed;
}

/** Removes a specific item from the table.
 *
 * The called must hold rcu read lock.
 *
 * @param item Item presumably present in the table and to be removed.
 * @return True if the item was removed successfully; or false if it had
 *     already been deleted.
 */
bool cht_remove_item(cht_t *h, cht_link_t *item)
{
	assert(h);
	assert(item);
	/* Otherwise a concurrent cht_remove_key might free the item. */
	assert(rcu_read_locked());

	/*
	 * Even though we know the node we want to delete we must unlink it
	 * from the correct bucket and from a clean/normal predecessor. Therefore,
	 * we search for it again from the beginning of the correct bucket.
	 */
	size_t hash = calc_node_hash(h, item);
	return remove_pred(h, hash, same_node_pred, item);
}

/** Removes an item equal to pred_arg according to the predicate pred. */
static bool remove_pred(cht_t *h, size_t hash, equal_pred_t pred, void *pred_arg)
{
	rcu_read_lock();

	bool resizing = false;
	bool deleted = false;
	bool deleted_but_gc = false;

	cht_buckets_t *b = rcu_access(h->b);
	size_t idx = calc_bucket_idx(hash, b->order);
	marked_ptr_t *phead = &b->head[idx];

	do {
		walk_mode_t walk_mode = WM_NORMAL;
		bool join_finishing = false;

		resizing = resizing || (N_NORMAL != get_mark(*phead));

		/* The table is resizing. Get the correct bucket head. */
		if (resizing) {
			upd_resizing_head(h, hash, &phead, &join_finishing, &walk_mode);
		}

		wnd_t wnd = {
			.ppred = phead,
			.cur = get_next(*phead),
			.last = NULL
		};

		if (!find_wnd_and_gc_pred(
		    h, hash, walk_mode, pred, pred_arg, &wnd, &resizing)) {
			/* Could not GC a node; or detected an unexpected resize. */
			continue;
		}

		/*
		 * The item lookup is affected by a bucket join but effects of
		 * the bucket join have not been seen while searching for the item.
		 */
		if (join_finishing && !join_completed(h, &wnd)) {
			/*
			 * Bucket was appended at the end of another but the next
			 * ptr linking them together was not visible on this cpu.
			 * join_completed() makes this appended bucket visible.
			 */
			continue;
		}

		/* Already deleted, but delete_at() requested one GC pass. */
		if (deleted_but_gc)
			break;

		bool found = (wnd.cur != &sentinel && pred(pred_arg, wnd.cur));

		if (!found) {
			rcu_read_unlock();
			return false;
		}

		deleted = delete_at(h, &wnd, walk_mode, &deleted_but_gc, &resizing);
	} while (!deleted || deleted_but_gc);

	rcu_read_unlock();
	return true;
}

/** Unlinks wnd.cur from wnd.ppred and schedules a deferred free for the item.
 *
 * Ignores nodes marked N_JOIN if walk mode is WM_LEAVE_JOIN.
 *
 * @param h   CHT to operate on.
 * @param wnd Points to the item to delete and its N_NORMAL predecessor.
 * @param walk_mode Bucket chaing walk mode.
 * @param deleted_but_gc Set to true if the item had been logically deleted,
 *         but a garbage collecting walk of the bucket is in order for
 *         it to be fully unlinked.
 * @param resizing Set to true if the table is undergoing an unexpected
 *         resize (ie walk_mode == WM_NORMAL).
 * @return False if the wnd.ppred changed in the meantime and the whole
 *         delete operation must be retried.
 */
static inline bool delete_at(cht_t *h, wnd_t *wnd, walk_mode_t walk_mode,
    bool *deleted_but_gc, bool *resizing)
{
	assert(wnd->cur && wnd->cur != &sentinel);

	*deleted_but_gc = false;

	if (!mark_deleted(wnd->cur, walk_mode, resizing)) {
		/* Already deleted, or unexpectedly marked as JOIN/JOIN_FOLLOWS. */
		return false;
	}

	/* Marked deleted. Unlink from the bucket. */

	/* Never unlink join nodes. */
	if (walk_mode == WM_LEAVE_JOIN && (N_JOIN & get_mark(wnd->cur->link)))
		return true;

	cas_order_barrier();

	if (unlink_from_pred(wnd, walk_mode, resizing)) {
		free_later(h, wnd->cur);
	} else {
		*deleted_but_gc = true;
	}

	return true;
}

/** Marks cur logically deleted. Returns false to request a retry. */
static inline bool mark_deleted(cht_link_t *cur, walk_mode_t walk_mode,
    bool *resizing)
{
	assert(cur && cur != &sentinel);

	/*
	 * Btw, we could loop here if the cas fails but let's not complicate
	 * things and let's retry from the head of the bucket.
	 */

	cht_link_t *next = get_next(cur->link);

	if (walk_mode == WM_NORMAL) {
		/* Only mark clean/normal nodes - JF/JN is used only during resize. */
		marked_ptr_t ret = cas_link(&cur->link, next, N_NORMAL, next, N_DELETED);

		if (ret != make_link(next, N_NORMAL)) {
			*resizing = (N_JOIN | N_JOIN_FOLLOWS) & get_mark(ret);
			return false;
		}
	} else {
		static_assert(N_JOIN == N_JOIN_FOLLOWS, "");

		/* Keep the N_JOIN/N_JOIN_FOLLOWS mark but strip N_DELETED. */
		mark_t cur_mark = get_mark(cur->link) & N_JOIN_FOLLOWS;

		marked_ptr_t ret =
		    cas_link(&cur->link, next, cur_mark, next, cur_mark | N_DELETED);

		if (ret != make_link(next, cur_mark))
			return false;
	}

	return true;
}

/** Unlinks wnd.cur from wnd.ppred. Returns false if it should be retried. */
static inline bool unlink_from_pred(wnd_t *wnd, walk_mode_t walk_mode,
    bool *resizing)
{
	assert(wnd->cur != &sentinel);
	assert(wnd->cur && (N_DELETED & get_mark(wnd->cur->link)));

	cht_link_t *next = get_next(wnd->cur->link);

	if (walk_mode == WM_LEAVE_JOIN) {
		/* Never try to unlink join nodes. */
		assert(!(N_JOIN & get_mark(wnd->cur->link)));

		mark_t pred_mark = get_mark(*wnd->ppred);
		/* Succeed only if the predecessor is clean/normal or a join node. */
		mark_t exp_pred_mark = (N_JOIN & pred_mark) ? pred_mark : N_NORMAL;

		marked_ptr_t pred_link = make_link(wnd->cur, exp_pred_mark);
		marked_ptr_t next_link = make_link(next, exp_pred_mark);

		if (pred_link != _cas_link(wnd->ppred, pred_link, next_link))
			return false;
	} else {
		assert(walk_mode == WM_MOVE_JOIN_FOLLOWS || walk_mode == WM_NORMAL);
		/* Move the JF mark if set. Clear DEL mark. */
		mark_t cur_mark = N_JOIN_FOLLOWS & get_mark(wnd->cur->link);

		/* The predecessor must be clean/normal. */
		marked_ptr_t pred_link = make_link(wnd->cur, N_NORMAL);
		/* Link to cur's successor keeping/copying cur's JF mark. */
		marked_ptr_t next_link = make_link(next, cur_mark);

		marked_ptr_t ret = _cas_link(wnd->ppred, pred_link, next_link);

		if (pred_link != ret) {
			/* If we're not resizing the table there are no JF/JN nodes. */
			*resizing = (walk_mode == WM_NORMAL) &&
			    (N_JOIN_FOLLOWS & get_mark(ret));
			return false;
		}
	}

	return true;
}

/** Finds the first non-deleted item equal to \a pred_arg according to \a pred.
 *
 * The function returns the candidate item in \a wnd. Logically deleted
 * nodes are garbage collected so the predecessor will most likely not
 * be marked as deleted.
 *
 * Unlike find_wnd_and_gc(), this function never returns a node that
 * is known to have already been marked N_DELETED.
 *
 * Any logically deleted nodes (ie those marked N_DELETED) are garbage
 * collected, ie free in the background via rcu_call (except for join-nodes
 * if walk_mode == WM_LEAVE_JOIN).
 *
 * @param h         CHT to operate on.
 * @param hash      Hash the search for.
 * @param walk_mode Bucket chain walk mode.
 * @param pred      Predicate used to find an item equal to pred_arg.
 * @param pred_arg  Argument to pass to the equality predicate \a pred.
 * @param[in,out] wnd The search starts with wnd.cur. If the desired
 *                  item is found wnd.cur will point to it.
 * @param resizing  Set to true if the table is resizing but it was not
 *                  expected (ie walk_mode == WM_NORMAL).
 * @return False if the operation has to be retried. True otherwise
 *        (even if an equal item had not been found).
 */
static bool find_wnd_and_gc_pred(cht_t *h, size_t hash, walk_mode_t walk_mode,
    equal_pred_t pred, void *pred_arg, wnd_t *wnd, bool *resizing)
{
	assert(wnd->cur);

	if (wnd->cur == &sentinel)
		return true;

	/*
	 * A read barrier is not needed here to bring up the most recent
	 * node marks (esp the N_DELETED). At worst we'll try to delete
	 * an already deleted node; fail in delete_at(); and retry.
	 */

	size_t cur_hash;

try_again:
	cur_hash = node_hash(h, wnd->cur);

	while (cur_hash <= hash) {
		assert(wnd->cur && wnd->cur != &sentinel);

		/* GC any deleted nodes on the way. */
		if (N_DELETED & get_mark(wnd->cur->link)) {
			if (!gc_deleted_node(h, walk_mode, wnd, resizing)) {
				/* Retry from the head of a bucket. */
				return false;
			}
		} else {
			/* Is this the node we were looking for? */
			if (cur_hash == hash && pred(pred_arg, wnd->cur))
				return true;

			next_wnd(wnd);
		}

		cur_hash = node_hash(h, wnd->cur);
	}

	if (cur_hash == h->invalid_hash) {
		next_wnd(wnd);
		assert(wnd->cur);
		goto try_again;
	}

	/* The searched for node is not in the current bucket. */
	return true;
}

/** Find the first item (deleted or not) with a hash greater or equal to \a hash.
 *
 * The function returns the first item with a hash that is greater or
 * equal to \a hash in \a wnd. Moreover it garbage collects logically
 * deleted node that have not yet been unlinked and freed. Therefore,
 * the returned node's predecessor will most likely be N_NORMAL.
 *
 * Unlike find_wnd_and_gc_pred(), this function may return a node
 * that is known to had been marked N_DELETED.
 *
 * @param h         CHT to operate on.
 * @param hash      Hash of the item to find.
 * @param walk_mode Bucket chain walk mode.
 * @param[in,out] wnd wnd.cur denotes the first node of the chain. If the
 *                  the operation is successful, \a wnd points to the desired
 *                  item.
 * @param resizing  Set to true if a table resize was detected but walk_mode
 *                  suggested the table was not undergoing a resize.
 * @return False indicates the operation must be retried. True otherwise
 *       (even if an item with exactly the same has was not found).
 */
static bool find_wnd_and_gc(cht_t *h, size_t hash, walk_mode_t walk_mode,
    wnd_t *wnd, bool *resizing)
{
try_again:
	assert(wnd->cur);

	while (node_hash(h, wnd->cur) < hash) {
		/* GC any deleted nodes along the way to our desired node. */
		if (N_DELETED & get_mark(wnd->cur->link)) {
			if (!gc_deleted_node(h, walk_mode, wnd, resizing)) {
				/* Failed to remove the garbage node. Retry. */
				return false;
			}
		} else {
			next_wnd(wnd);
		}

		assert(wnd->cur);
	}

	if (node_hash(h, wnd->cur) == h->invalid_hash) {
		next_wnd(wnd);
		goto try_again;
	}

	/* wnd->cur may be NULL or even marked N_DELETED. */
	return true;
}

/** Garbage collects the N_DELETED node at \a wnd skipping join nodes. */
static bool gc_deleted_node(cht_t *h, walk_mode_t walk_mode, wnd_t *wnd,
    bool *resizing)
{
	assert(N_DELETED & get_mark(wnd->cur->link));

	/* Skip deleted JOIN nodes. */
	if (walk_mode == WM_LEAVE_JOIN && (N_JOIN & get_mark(wnd->cur->link))) {
		next_wnd(wnd);
	} else {
		/* Ordinary deleted node or a deleted JOIN_FOLLOWS. */
		assert(walk_mode != WM_LEAVE_JOIN ||
		    !((N_JOIN | N_JOIN_FOLLOWS) & get_mark(wnd->cur->link)));

		/* Unlink an ordinary deleted node, move JOIN_FOLLOWS mark. */
		if (!unlink_from_pred(wnd, walk_mode, resizing)) {
			/* Retry. The predecessor was deleted, invalid, const, join_follows. */
			return false;
		}

		free_later(h, wnd->cur);

		/* Leave ppred as is. */
		wnd->last = wnd->cur;
		wnd->cur = get_next(wnd->cur->link);
	}

	return true;
}

/** Returns true if a bucket join had already completed.
 *
 * May only be called if upd_resizing_head() indicates a bucket join
 * may be in progress.
 *
 * If it returns false, the search must be retried in order to guarantee
 * all item that should have been encountered have been seen.
 */
static bool join_completed(cht_t *h, const wnd_t *wnd)
{
	/*
	 * The table is shrinking and the searched for item is in a bucket
	 * appended to another. Check that the link joining these two buckets
	 * is visible and if not, make it visible to this cpu.
	 */

	/*
	 * Resizer ensures h->b->order stays the same for the duration of this
	 * func. We got here because there was an alternative head to search.
	 * The resizer waits for all preexisting readers to finish after
	 * it
	 */
	assert(h->b->order > h->new_b->order);
	assert(wnd->cur);

	/* Either we did not need the joining link or we have already followed it.*/
	if (wnd->cur != &sentinel)
		return true;

	/* We have reached the end of a bucket. */

	if (wnd->last != &sentinel) {
		size_t last_seen_hash = node_hash(h, wnd->last);

		if (last_seen_hash == h->invalid_hash) {
			last_seen_hash = calc_node_hash(h, wnd->last);
		}

		size_t last_old_idx = calc_bucket_idx(last_seen_hash, h->b->order);
		size_t move_src_idx = grow_idx(shrink_idx(last_old_idx));

		/*
		 * Last node seen was in the joining bucket - if the searched
		 * for node is there we will find it.
		 */
		if (move_src_idx != last_old_idx)
			return true;
	}

	/*
	 * Reached the end of the bucket but no nodes from the joining bucket
	 * were seen. There should have at least been a JOIN node so we have
	 * definitely not seen (and followed) the joining link. Make the link
	 * visible and retry.
	 */
	read_barrier();
	return false;
}

/** When resizing returns the bucket head to start the search with in \a phead.
 *
 * If a resize had been detected (eg cht_t.b.head[idx] is marked immutable).
 * upd_resizing_head() moves the bucket for \a hash from the old head
 * to the new head. Moreover, it splits or joins buckets as necessary.
 *
 * @param h     CHT to operate on.
 * @param hash  Hash of an item whose chain we would like to traverse.
 * @param[out] phead Head of the bucket to search for \a hash.
 * @param[out] join_finishing Set to true if a bucket join might be
 *              in progress and the bucket may have to traversed again
 *              as indicated by join_completed().
 * @param[out] walk_mode Specifies how to interpret node marks.
 */
static void upd_resizing_head(cht_t *h, size_t hash, marked_ptr_t **phead,
    bool *join_finishing,  walk_mode_t *walk_mode)
{
	cht_buckets_t *b = rcu_access(h->b);
	size_t old_idx = calc_bucket_idx(hash, b->order);
	size_t new_idx = calc_bucket_idx(hash, h->new_b->order);

	marked_ptr_t *pold_head = &b->head[old_idx];
	marked_ptr_t *pnew_head = &h->new_b->head[new_idx];

	/* In any case, use the bucket in the new table. */
	*phead = pnew_head;

	/* Growing the table. */
	if (b->order < h->new_b->order) {
		size_t move_dest_idx = grow_idx(old_idx);
		marked_ptr_t *pmoved_head = &h->new_b->head[move_dest_idx];

		/* Complete moving the bucket from the old to the new table. */
		help_head_move(pold_head, pmoved_head);

		/* The hash belongs to the moved bucket. */
		if (move_dest_idx == new_idx) {
			assert(pmoved_head == pnew_head);
			/*
			 * move_head() makes the new head of the moved bucket visible.
			 * The new head may be marked with a JOIN_FOLLOWS
			 */
			assert(!(N_CONST & get_mark(*pmoved_head)));
			*walk_mode = WM_MOVE_JOIN_FOLLOWS;
		} else {
			assert(pmoved_head != pnew_head);
			/*
			 * The hash belongs to the bucket that is the result of splitting
			 * the old/moved bucket, ie the bucket that contains the second
			 * half of the split/old/moved bucket.
			 */

			/* The moved bucket has not yet been split. */
			if (N_NORMAL != get_mark(*pnew_head)) {
				size_t split_hash = calc_split_hash(new_idx, h->new_b->order);
				split_bucket(h, pmoved_head, pnew_head, split_hash);
				/*
				 * split_bucket() makes the new head visible. No
				 * JOIN_FOLLOWS in this part of split bucket.
				 */
				assert(N_NORMAL == get_mark(*pnew_head));
			}

			*walk_mode = WM_LEAVE_JOIN;
		}
	} else if (h->new_b->order < b->order) {
		/* Shrinking the table. */

		size_t move_src_idx = grow_idx(new_idx);

		/*
		 * Complete moving the bucket from the old to the new table.
		 * Makes a valid pnew_head visible if already moved.
		 */
		help_head_move(&b->head[move_src_idx], pnew_head);

		/* Hash belongs to the bucket to be joined with the moved bucket. */
		if (move_src_idx != old_idx) {
			/* Bucket join not yet completed. */
			if (N_INVALID != get_mark(*pold_head)) {
				size_t split_hash = calc_split_hash(old_idx, b->order);
				join_buckets(h, pold_head, pnew_head, split_hash);
			}

			/*
			 * The resizer sets pold_head to &sentinel when all cpus are
			 * guaranteed to see the bucket join.
			 */
			*join_finishing = (&sentinel != get_next(*pold_head));
		}

		/* move_head() or join_buckets() makes it so or makes the mark visible.*/
		assert(N_INVALID == get_mark(*pold_head));
		/* move_head() makes it visible. No JOIN_FOLLOWS used when shrinking. */
		assert(N_NORMAL == get_mark(*pnew_head));

		*walk_mode = WM_LEAVE_JOIN;
	} else {
		/*
		 * Final stage of resize. The resizer is waiting for all
		 * readers to notice that the old table had been replaced.
		 */
		assert(b == h->new_b);
		*walk_mode = WM_NORMAL;
	}
}

#if 0
static void move_head(marked_ptr_t *psrc_head, marked_ptr_t *pdest_head)
{
	start_head_move(psrc_head);
	cas_order_barrier();
	complete_head_move(psrc_head, pdest_head);
}
#endif

/** Moves an immutable head \a psrc_head of cht_t.b to \a pdest_head of cht_t.new_b.
 *
 * The function guarantees the move will be visible on this cpu once
 * it completes. In particular, *pdest_head will not be N_INVALID.
 *
 * Unlike complete_head_move(), help_head_move() checks if the head had already
 * been moved and tries to avoid moving the bucket heads if possible.
 */
static inline void help_head_move(marked_ptr_t *psrc_head,
    marked_ptr_t *pdest_head)
{
	/* Head move has to in progress already when calling this func. */
	assert(N_CONST & get_mark(*psrc_head));

	/* Head already moved. */
	if (N_INVALID == get_mark(*psrc_head)) {
		/* Effects of the head move have not yet propagated to this cpu. */
		if (N_INVALID == get_mark(*pdest_head)) {
			/* Make the move visible on this cpu. */
			read_barrier();
		}
	} else {
		complete_head_move(psrc_head, pdest_head);
	}

	assert(!(N_CONST & get_mark(*pdest_head)));
}

/** Initiates the move of the old head \a psrc_head.
 *
 * The move may be completed with help_head_move().
 */
static void start_head_move(marked_ptr_t *psrc_head)
{
	/* Mark src head immutable. */
	mark_const(psrc_head);
}

/** Marks the head immutable. */
static void mark_const(marked_ptr_t *psrc_head)
{
	marked_ptr_t ret, src_link;

	/* Mark src head immutable. */
	do {
		cht_link_t *next = get_next(*psrc_head);
		src_link = make_link(next, N_NORMAL);

		/* Mark the normal/clean src link immutable/const. */
		ret = cas_link(psrc_head, next, N_NORMAL, next, N_CONST);
	} while (ret != src_link && !(N_CONST & get_mark(ret)));
}

/** Completes moving head psrc_head to pdest_head (started by start_head_move()).*/
static void complete_head_move(marked_ptr_t *psrc_head, marked_ptr_t *pdest_head)
{
	assert(N_JOIN_FOLLOWS != get_mark(*psrc_head));
	assert(N_CONST & get_mark(*psrc_head));

	cht_link_t *next = get_next(*psrc_head);

#ifdef CONFIG_DEBUG
	marked_ptr_t ret =
#endif
	    cas_link(pdest_head, &sentinel, N_INVALID, next, N_NORMAL);
	assert(ret == make_link(&sentinel, N_INVALID) || (N_NORMAL == get_mark(ret)));
	cas_order_barrier();

#ifdef CONFIG_DEBUG
	ret =
#endif
	    cas_link(psrc_head, next, N_CONST, next, N_INVALID);
	assert(ret == make_link(next, N_CONST) || (N_INVALID == get_mark(ret)));
	cas_order_barrier();
}

/** Splits the bucket at psrc_head and links to the remainder from pdest_head.
 *
 * Items with hashes greater or equal to \a split_hash are moved to bucket
 * with head at \a pdest_head.
 *
 * @param h           CHT to operate on.
 * @param psrc_head   Head of the bucket to split (in cht_t.new_b).
 * @param pdest_head  Head of the bucket that points to the second part
 *                    of the split bucket in psrc_head. (in cht_t.new_b)
 * @param split_hash  Hash of the first possible item in the remainder of
 *                    psrc_head, ie the smallest hash pdest_head is allowed
 *                    to point to..
 */
static void split_bucket(cht_t *h, marked_ptr_t *psrc_head,
    marked_ptr_t *pdest_head, size_t split_hash)
{
	/* Already split. */
	if (N_NORMAL == get_mark(*pdest_head))
		return;

	/*
	 * L == Last node of the first part of the split bucket. That part
	 *      remains in the original/src bucket.
	 * F == First node of the second part of the split bucket. That part
	 *      will be referenced from the dest bucket head.
	 *
	 * We want to first mark a clean L as JF so that updaters unaware of
	 * the split (or table resize):
	 * - do not insert a new node between L and F
	 * - do not unlink L (that is why it has to be clean/normal)
	 * - do not unlink F
	 *
	 * Then we can safely mark F as JN even if it has been marked deleted.
	 * Once F is marked as JN updaters aware of table resize will not
	 * attempt to unlink it (JN will have two predecessors - we cannot
	 * safely unlink from both at the same time). Updaters unaware of
	 * ongoing resize can reach F only via L and that node is already
	 * marked JF, so they won't unlink F.
	 *
	 * Last, link the new/dest head to F.
	 *
	 *
	 * 0)                           ,-- split_hash, first hash of the dest bucket
	 *                              v
	 *  [src_head | N] -> .. -> [L] -> [F]
	 *  [dest_head | Inv]
	 *
	 * 1)                             ,-- split_hash
	 *                                v
	 *  [src_head | N] -> .. -> [JF] -> [F]
	 *  [dest_head | Inv]
	 *
	 * 2)                             ,-- split_hash
	 *                                v
	 *  [src_head | N] -> .. -> [JF] -> [JN]
	 *  [dest_head | Inv]
	 *
	 * 3)                             ,-- split_hash
	 *                                v
	 *  [src_head | N] -> .. -> [JF] -> [JN]
	 *                                   ^
	 *  [dest_head | N] -----------------'
	 */
	wnd_t wnd;

	rcu_read_lock();

	/* Mark the last node of the first part of the split bucket as JF. */
	mark_join_follows(h, psrc_head, split_hash, &wnd);
	cas_order_barrier();

	/* There are nodes in the dest bucket, ie the second part of the split. */
	if (wnd.cur != &sentinel) {
		/*
		 * Mark the first node of the dest bucket as a join node so
		 * updaters do not attempt to unlink it if it is deleted.
		 */
		mark_join_node(wnd.cur);
		cas_order_barrier();
	} else {
		/*
		 * Second part of the split bucket is empty. There are no nodes
		 * to mark as JOIN nodes and there never will be.
		 */
	}

	/* Link the dest head to the second part of the split. */
#ifdef CONFIG_DEBUG
	marked_ptr_t ret =
#endif
	    cas_link(pdest_head, &sentinel, N_INVALID, wnd.cur, N_NORMAL);
	assert(ret == make_link(&sentinel, N_INVALID) || (N_NORMAL == get_mark(ret)));
	cas_order_barrier();

	rcu_read_unlock();
}

/** Finds and marks the last node of psrc_head w/ hash less than split_hash.
 *
 * Finds a node in psrc_head with the greatest hash that is strictly less
 * than split_hash and marks it with N_JOIN_FOLLOWS.
 *
 * Returns a window pointing to that node.
 *
 * Any logically deleted nodes along the way are
 * garbage collected; therefore, the predecessor node (if any) will most
 * likely not be marked N_DELETED.
 *
 * @param h          CHT to operate on.
 * @param psrc_head  Bucket head.
 * @param split_hash The smallest hash a join node (ie the node following
 *                   the desired join-follows node) may have.
 * @param[out] wnd   Points to the node marked with N_JOIN_FOLLOWS.
 */
static void mark_join_follows(cht_t *h, marked_ptr_t *psrc_head,
    size_t split_hash, wnd_t *wnd)
{
	/* See comment in split_bucket(). */

	bool done = false;

	do {
		bool resizing = false;
		wnd->ppred = psrc_head;
		wnd->cur = get_next(*psrc_head);

		/*
		 * Find the split window, ie the last node of the first part of
		 * the split bucket and the its successor - the first node of
		 * the second part of the split bucket. Retry if GC failed.
		 */
		if (!find_wnd_and_gc(h, split_hash, WM_MOVE_JOIN_FOLLOWS, wnd, &resizing))
			continue;

		/* Must not report that the table is resizing if WM_MOVE_JOIN_FOLLOWS.*/
		assert(!resizing);
		/*
		 * Mark the last node of the first half of the split bucket
		 * that a join node follows. It must be clean/normal.
		 */
		marked_ptr_t ret =
		    cas_link(wnd->ppred, wnd->cur, N_NORMAL, wnd->cur, N_JOIN_FOLLOWS);

		/*
		 * Successfully marked as a JF node or already marked that way (even
		 * if also marked deleted - unlinking the node will move the JF mark).
		 */
		done = (ret == make_link(wnd->cur, N_NORMAL)) ||
		    (N_JOIN_FOLLOWS & get_mark(ret));
	} while (!done);
}

/** Marks join_node with N_JOIN. */
static void mark_join_node(cht_link_t *join_node)
{
	/* See comment in split_bucket(). */

	bool done;
	do {
		cht_link_t *next = get_next(join_node->link);
		mark_t mark = get_mark(join_node->link);

		/*
		 * May already be marked as deleted, but it won't be unlinked
		 * because its predecessor is marked with JOIN_FOLLOWS or CONST.
		 */
		marked_ptr_t ret =
		    cas_link(&join_node->link, next, mark, next, mark | N_JOIN);

		/* Successfully marked or already marked as a join node. */
		done = (ret == make_link(next, mark)) ||
		    (N_JOIN & get_mark(ret));
	} while (!done);
}

/** Appends the bucket at psrc_head to the bucket at pdest_head.
 *
 * @param h          CHT to operate on.
 * @param psrc_head  Bucket to merge with pdest_head.
 * @param pdest_head Bucket to be joined by psrc_head.
 * @param split_hash The smallest hash psrc_head may contain.
 */
static void join_buckets(cht_t *h, marked_ptr_t *psrc_head,
    marked_ptr_t *pdest_head, size_t split_hash)
{
	/* Buckets already joined. */
	if (N_INVALID == get_mark(*psrc_head))
		return;
	/*
	 * F == First node of psrc_head, ie the bucket we want to append
	 *      to (ie join with) the bucket starting at pdest_head.
	 * L == Last node of pdest_head, ie the bucket that psrc_head will
	 *      be appended to.
	 *
	 * (1) We first mark psrc_head immutable to signal that a join is
	 * in progress and so that updaters unaware of the join (or table
	 * resize):
	 * - do not insert new nodes between the head psrc_head and F
	 * - do not unlink F (it may already be marked deleted)
	 *
	 * (2) Next, F is marked as a join node. Updaters aware of table resize
	 * will not attempt to unlink it. We cannot safely/atomically unlink
	 * the join node because it will be pointed to from two different
	 * buckets. Updaters unaware of resize will fail to unlink the join
	 * node due to the head being marked immutable.
	 *
	 * (3) Then the tail of the bucket at pdest_head is linked to the join
	 * node. From now on, nodes in both buckets can be found via pdest_head.
	 *
	 * (4) Last, mark immutable psrc_head as invalid. It signals updaters
	 * that the join is complete and they can insert new nodes (originally
	 * destined for psrc_head) into pdest_head.
	 *
	 * Note that pdest_head keeps pointing at the join node. This allows
	 * lookups and updaters to determine if they should see a link between
	 * the tail L and F when searching for nodes originally in psrc_head
	 * via pdest_head. If they reach the tail of pdest_head without
	 * encountering any nodes of psrc_head, either there were no nodes
	 * in psrc_head to begin with or the link between L and F did not
	 * yet propagate to their cpus. If psrc_head was empty, it remains
	 * NULL. Otherwise psrc_head points to a join node (it will not be
	 * unlinked until table resize completes) and updaters/lookups
	 * should issue a read_barrier() to make the link [L]->[JN] visible.
	 *
	 * 0)                           ,-- split_hash, first hash of the src bucket
	 *                              v
	 *  [dest_head | N]-> .. -> [L]
	 *  [src_head | N]--> [F] -> ..
	 *  ^
	 *  ` split_hash, first hash of the src bucket
	 *
	 * 1)                            ,-- split_hash
	 *                               v
	 *  [dest_head | N]-> .. -> [L]
	 *  [src_head | C]--> [F] -> ..
	 *
	 * 2)                            ,-- split_hash
	 *                               v
	 *  [dest_head | N]-> .. -> [L]
	 *  [src_head | C]--> [JN] -> ..
	 *
	 * 3)                            ,-- split_hash
	 *                               v
	 *  [dest_head | N]-> .. -> [L] --+
	 *                                v
	 *  [src_head | C]-------------> [JN] -> ..
	 *
	 * 4)                            ,-- split_hash
	 *                               v
	 *  [dest_head | N]-> .. -> [L] --+
	 *                                v
	 *  [src_head | Inv]-----------> [JN] -> ..
	 */

	rcu_read_lock();

	/* Mark src_head immutable - signals updaters that bucket join started. */
	mark_const(psrc_head);
	cas_order_barrier();

	cht_link_t *join_node = get_next(*psrc_head);

	if (join_node != &sentinel) {
		mark_join_node(join_node);
		cas_order_barrier();

		link_to_join_node(h, pdest_head, join_node, split_hash);
		cas_order_barrier();
	}

#ifdef CONFIG_DEBUG
	marked_ptr_t ret =
#endif
	    cas_link(psrc_head, join_node, N_CONST, join_node, N_INVALID);
	assert(ret == make_link(join_node, N_CONST) || (N_INVALID == get_mark(ret)));
	cas_order_barrier();

	rcu_read_unlock();
}

/** Links the tail of pdest_head to join_node.
 *
 * @param h          CHT to operate on.
 * @param pdest_head Head of the bucket whose tail is to be linked to join_node.
 * @param join_node  A node marked N_JOIN with a hash greater or equal to
 *                   split_hash.
 * @param split_hash The least hash that is greater than the hash of any items
 *                   (originally) in pdest_head.
 */
static void link_to_join_node(cht_t *h, marked_ptr_t *pdest_head,
    cht_link_t *join_node, size_t split_hash)
{
	bool done = false;

	do {
		wnd_t wnd = {
			.ppred = pdest_head,
			.cur = get_next(*pdest_head)
		};

		bool resizing = false;

		if (!find_wnd_and_gc(h, split_hash, WM_LEAVE_JOIN, &wnd, &resizing))
			continue;

		assert(!resizing);

		if (wnd.cur != &sentinel) {
			/* Must be from the new appended bucket. */
			assert(split_hash <= node_hash(h, wnd.cur) ||
			    h->invalid_hash == node_hash(h, wnd.cur));
			return;
		}

		/* Reached the tail of pdest_head - link it to the join node. */
		marked_ptr_t ret =
		    cas_link(wnd.ppred, &sentinel, N_NORMAL, join_node, N_NORMAL);

		done = (ret == make_link(&sentinel, N_NORMAL));
	} while (!done);
}

/** Instructs RCU to free the item once all preexisting references are dropped.
 *
 * The item is freed via op->remove_callback().
 */
static void free_later(cht_t *h, cht_link_t *item)
{
	assert(item != &sentinel);

	/*
	 * remove_callback only works as rcu_func_t because rcu_link is the first
	 * field in cht_link_t.
	 */
	rcu_call(&item->rcu_link, (rcu_func_t)h->op->remove_callback);

	item_removed(h);
}

/** Notes that an item had been unlinked from the table and shrinks it if needed.
 *
 * If the number of items in the table drops below 1/4 of the maximum
 * allowed load the table is shrunk in the background.
 */
static inline void item_removed(cht_t *h)
{
	size_t items = (size_t) atomic_predec(&h->item_cnt);
	size_t bucket_cnt = (1 << h->b->order);

	bool need_shrink = (items == h->max_load * bucket_cnt / 4);
	bool missed_shrink = (items == h->max_load * bucket_cnt / 8);

	if ((need_shrink || missed_shrink) && h->b->order > h->min_order) {
		size_t resize_reqs = atomic_preinc(&h->resize_reqs);
		/* The first resize request. Start the resizer. */
		if (1 == resize_reqs) {
			workq_global_enqueue_noblock(&h->resize_work, resize_table);
		}
	}
}

/** Notes an item had been inserted and grows the table if needed.
 *
 * The table is resized in the background.
 */
static inline void item_inserted(cht_t *h)
{
	size_t items = (size_t) atomic_preinc(&h->item_cnt);
	size_t bucket_cnt = (1 << h->b->order);

	bool need_grow = (items == h->max_load * bucket_cnt);
	bool missed_grow = (items == 2 * h->max_load * bucket_cnt);

	if ((need_grow || missed_grow) && h->b->order < CHT_MAX_ORDER) {
		size_t resize_reqs = atomic_preinc(&h->resize_reqs);
		/* The first resize request. Start the resizer. */
		if (1 == resize_reqs) {
			workq_global_enqueue_noblock(&h->resize_work, resize_table);
		}
	}
}

/** Resize request handler. Invoked on the system work queue. */
static void resize_table(work_t *arg)
{
	cht_t *h = member_to_inst(arg, cht_t, resize_work);

#ifdef CONFIG_DEBUG
	assert(h->b);
	/* Make resize_reqs visible. */
	read_barrier();
	assert(0 < atomic_load(&h->resize_reqs));
#endif

	bool done = false;

	do {
		/* Load the most recent h->item_cnt. */
		read_barrier();
		size_t cur_items = (size_t) atomic_load(&h->item_cnt);
		size_t bucket_cnt = (1 << h->b->order);
		size_t max_items = h->max_load * bucket_cnt;

		if (cur_items >= max_items && h->b->order < CHT_MAX_ORDER) {
			grow_table(h);
		} else if (cur_items <= max_items / 4 && h->b->order > h->min_order) {
			shrink_table(h);
		} else {
			/* Table is just the right size. */
			size_t reqs = atomic_predec(&h->resize_reqs);
			done = (reqs == 0);
		}
	} while (!done);
}

/** Increases the number of buckets two-fold. Blocks until done. */
static void grow_table(cht_t *h)
{
	if (h->b->order >= CHT_MAX_ORDER)
		return;

	h->new_b = alloc_buckets(h->b->order + 1, true, false);

	/* Failed to alloc a new table - try next time the resizer is run. */
	if (!h->new_b)
		return;

	/* Wait for all readers and updaters to see the initialized new table. */
	rcu_synchronize();
	size_t old_bucket_cnt = (1 << h->b->order);

	/*
	 * Give updaters a chance to help out with the resize. Do the minimum
	 * work needed to announce a resize is in progress, ie start moving heads.
	 */
	for (size_t idx = 0; idx < old_bucket_cnt; ++idx) {
		start_head_move(&h->b->head[idx]);
	}

	/* Order start_head_move() wrt complete_head_move(). */
	cas_order_barrier();

	/* Complete moving heads and split any buckets not yet split by updaters. */
	for (size_t old_idx = 0; old_idx < old_bucket_cnt; ++old_idx) {
		marked_ptr_t *move_dest_head = &h->new_b->head[grow_idx(old_idx)];
		marked_ptr_t *move_src_head = &h->b->head[old_idx];

		/* Head move not yet completed. */
		if (N_INVALID != get_mark(*move_src_head)) {
			complete_head_move(move_src_head, move_dest_head);
		}

		size_t split_idx = grow_to_split_idx(old_idx);
		size_t split_hash = calc_split_hash(split_idx, h->new_b->order);
		marked_ptr_t *split_dest_head = &h->new_b->head[split_idx];

		split_bucket(h, move_dest_head, split_dest_head, split_hash);
	}

	/*
	 * Wait for all updaters to notice the new heads. Once everyone sees
	 * the invalid old bucket heads they will know a resize is in progress
	 * and updaters will modify the correct new buckets.
	 */
	rcu_synchronize();

	/* Clear the JOIN_FOLLOWS mark and remove the link between the split buckets.*/
	for (size_t old_idx = 0; old_idx < old_bucket_cnt; ++old_idx) {
		size_t new_idx = grow_idx(old_idx);

		cleanup_join_follows(h, &h->new_b->head[new_idx]);
	}

	/*
	 * Wait for everyone to notice that buckets were split, ie link connecting
	 * the join follows and join node has been cut.
	 */
	rcu_synchronize();

	/* Clear the JOIN mark and GC any deleted join nodes. */
	for (size_t old_idx = 0; old_idx < old_bucket_cnt; ++old_idx) {
		size_t new_idx = grow_to_split_idx(old_idx);

		cleanup_join_node(h, &h->new_b->head[new_idx]);
	}

	/* Wait for everyone to see that the table is clear of any resize marks. */
	rcu_synchronize();

	cht_buckets_t *old_b = h->b;
	rcu_assign(h->b, h->new_b);

	/* Wait for everyone to start using the new table. */
	rcu_synchronize();

	free(old_b);

	/* Not needed; just for increased readability. */
	h->new_b = NULL;
}

/** Halfs the number of buckets. Blocks until done. */
static void shrink_table(cht_t *h)
{
	if (h->b->order <= h->min_order)
		return;

	h->new_b = alloc_buckets(h->b->order - 1, true, false);

	/* Failed to alloc a new table - try next time the resizer is run. */
	if (!h->new_b)
		return;

	/* Wait for all readers and updaters to see the initialized new table. */
	rcu_synchronize();

	size_t old_bucket_cnt = (1 << h->b->order);

	/*
	 * Give updaters a chance to help out with the resize. Do the minimum
	 * work needed to announce a resize is in progress, ie start moving heads.
	 */
	for (size_t old_idx = 0; old_idx < old_bucket_cnt; ++old_idx) {
		size_t new_idx = shrink_idx(old_idx);

		/* This bucket should be moved. */
		if (grow_idx(new_idx) == old_idx) {
			start_head_move(&h->b->head[old_idx]);
		} else {
			/* This bucket should join the moved bucket once the move is done.*/
		}
	}

	/* Order start_head_move() wrt to complete_head_move(). */
	cas_order_barrier();

	/* Complete moving heads and join buckets with the moved buckets. */
	for (size_t old_idx = 0; old_idx < old_bucket_cnt; ++old_idx) {
		size_t new_idx = shrink_idx(old_idx);
		size_t move_src_idx = grow_idx(new_idx);

		/* This bucket should be moved. */
		if (move_src_idx == old_idx) {
			/* Head move not yet completed. */
			if (N_INVALID != get_mark(h->b->head[old_idx])) {
				complete_head_move(&h->b->head[old_idx], &h->new_b->head[new_idx]);
			}
		} else {
			/* This bucket should join the moved bucket. */
			size_t split_hash = calc_split_hash(old_idx, h->b->order);
			join_buckets(h, &h->b->head[old_idx], &h->new_b->head[new_idx],
			    split_hash);
		}
	}

	/*
	 * Wait for all updaters to notice the new heads. Once everyone sees
	 * the invalid old bucket heads they will know a resize is in progress
	 * and updaters will modify the correct new buckets.
	 */
	rcu_synchronize();

	/* Let everyone know joins are complete and fully visible. */
	for (size_t old_idx = 0; old_idx < old_bucket_cnt; ++old_idx) {
		size_t move_src_idx = grow_idx(shrink_idx(old_idx));

		/* Set the invalid joinee head to NULL. */
		if (old_idx != move_src_idx) {
			assert(N_INVALID == get_mark(h->b->head[old_idx]));

			if (&sentinel != get_next(h->b->head[old_idx]))
				h->b->head[old_idx] = make_link(&sentinel, N_INVALID);
		}
	}

	/* todo comment join node vs reset joinee head*/
	rcu_synchronize();

	size_t new_bucket_cnt = (1 << h->new_b->order);

	/* Clear the JOIN mark and GC any deleted join nodes. */
	for (size_t new_idx = 0; new_idx < new_bucket_cnt; ++new_idx) {
		cleanup_join_node(h, &h->new_b->head[new_idx]);
	}

	/* Wait for everyone to see that the table is clear of any resize marks. */
	rcu_synchronize();

	cht_buckets_t *old_b = h->b;
	rcu_assign(h->b, h->new_b);

	/* Wait for everyone to start using the new table. */
	rcu_synchronize();

	free(old_b);

	/* Not needed; just for increased readability. */
	h->new_b = NULL;
}

/** Finds and clears the N_JOIN mark from a node in new_head (if present). */
static void cleanup_join_node(cht_t *h, marked_ptr_t *new_head)
{
	rcu_read_lock();

	cht_link_t *cur = get_next(*new_head);

	while (cur != &sentinel) {
		/* Clear the join node's JN mark - even if it is marked as deleted. */
		if (N_JOIN & get_mark(cur->link)) {
			clear_join_and_gc(h, cur, new_head);
			break;
		}

		cur = get_next(cur->link);
	}

	rcu_read_unlock();
}

/** Clears the join_node's N_JOIN mark frees it if marked N_DELETED as well. */
static void clear_join_and_gc(cht_t *h, cht_link_t *join_node,
    marked_ptr_t *new_head)
{
	assert(join_node != &sentinel);
	assert(join_node && (N_JOIN & get_mark(join_node->link)));

	bool done;

	/* Clear the JN mark. */
	do {
		marked_ptr_t jn_link = join_node->link;
		cht_link_t *next = get_next(jn_link);
		/* Clear the JOIN mark but keep the DEL mark if present. */
		mark_t cleared_mark = get_mark(jn_link) & N_DELETED;

		marked_ptr_t ret =
		    _cas_link(&join_node->link, jn_link, make_link(next, cleared_mark));

		/* Done if the mark was cleared. Retry if a new node was inserted. */
		done = (ret == jn_link);
		assert(ret == jn_link || (get_mark(ret) & N_JOIN));
	} while (!done);

	if (!(N_DELETED & get_mark(join_node->link)))
		return;

	/* The join node had been marked as deleted - GC it. */

	/* Clear the JOIN mark before trying to unlink the deleted join node.*/
	cas_order_barrier();

	size_t jn_hash = node_hash(h, join_node);
	do {
		bool resizing = false;

		wnd_t wnd = {
			.ppred = new_head,
			.cur = get_next(*new_head)
		};

		done = find_wnd_and_gc_pred(h, jn_hash, WM_NORMAL, same_node_pred,
		    join_node, &wnd, &resizing);

		assert(!resizing);
	} while (!done);
}

/** Finds a non-deleted node with N_JOIN_FOLLOWS and clears the mark. */
static void cleanup_join_follows(cht_t *h, marked_ptr_t *new_head)
{
	assert(new_head);

	rcu_read_lock();

	wnd_t wnd = {
		.ppred = NULL,
		.cur = NULL
	};
	marked_ptr_t *cur_link = new_head;

	/*
	 * Find the non-deleted node with a JF mark and clear the JF mark.
	 * The JF node may be deleted and/or the mark moved to its neighbors
	 * at any time. Therefore, we GC deleted nodes until we find the JF
	 * node in order to remove stale/deleted JF nodes left behind eg by
	 * delayed threads that did not yet get a chance to unlink the deleted
	 * JF node and move its mark.
	 *
	 * Note that the head may be marked JF (but never DELETED).
	 */
	while (true) {
		bool is_jf_node = N_JOIN_FOLLOWS & get_mark(*cur_link);

		/* GC any deleted nodes on the way - even deleted JOIN_FOLLOWS. */
		if (N_DELETED & get_mark(*cur_link)) {
			assert(cur_link != new_head);
			assert(wnd.ppred && wnd.cur && wnd.cur != &sentinel);
			assert(cur_link == &wnd.cur->link);

			bool dummy;
			bool deleted = gc_deleted_node(h, WM_MOVE_JOIN_FOLLOWS, &wnd, &dummy);

			/* Failed to GC or collected a deleted JOIN_FOLLOWS. */
			if (!deleted || is_jf_node) {
				/* Retry from the head of the bucket. */
				cur_link = new_head;
				continue;
			}
		} else {
			/* Found a non-deleted JF. Clear its JF mark. */
			if (is_jf_node) {
				cht_link_t *next = get_next(*cur_link);
				marked_ptr_t ret =
				    cas_link(cur_link, next, N_JOIN_FOLLOWS, &sentinel, N_NORMAL);

				assert(next == &sentinel ||
				    ((N_JOIN | N_JOIN_FOLLOWS) & get_mark(ret)));

				/* Successfully cleared the JF mark of a non-deleted node. */
				if (ret == make_link(next, N_JOIN_FOLLOWS)) {
					break;
				} else {
					/*
					 * The JF node had been deleted or a new node inserted
					 * right after it. Retry from the head.
					 */
					cur_link = new_head;
					continue;
				}
			} else {
				wnd.ppred = cur_link;
				wnd.cur = get_next(*cur_link);
			}
		}

		/* We must encounter a JF node before we reach the end of the bucket. */
		assert(wnd.cur && wnd.cur != &sentinel);
		cur_link = &wnd.cur->link;
	}

	rcu_read_unlock();
}

/** Returns the first possible hash following a bucket split point.
 *
 * In other words the returned hash is the smallest possible hash
 * the remainder of the split bucket may contain.
 */
static inline size_t calc_split_hash(size_t split_idx, size_t order)
{
	assert(1 <= order && order <= 8 * sizeof(size_t));
	return split_idx << (8 * sizeof(size_t) - order);
}

/** Returns the bucket head index given the table size order and item hash. */
static inline size_t calc_bucket_idx(size_t hash, size_t order)
{
	assert(1 <= order && order <= 8 * sizeof(size_t));
	return hash >> (8 * sizeof(size_t) - order);
}

/** Returns the bucket index of destination*/
static inline size_t grow_to_split_idx(size_t old_idx)
{
	return grow_idx(old_idx) | 1;
}

/** Returns the destination index of a bucket head when the table is growing. */
static inline size_t grow_idx(size_t idx)
{
	return idx << 1;
}

/** Returns the destination index of a bucket head when the table is shrinking.*/
static inline size_t shrink_idx(size_t idx)
{
	return idx >> 1;
}

/** Returns a mixed hash of the search key.*/
static inline size_t calc_key_hash(cht_t *h, void *key)
{
	/* Mimic calc_node_hash. */
	return hash_mix(h->op->key_hash(key)) & ~(size_t)1;
}

/** Returns a memoized mixed hash of the item. */
static inline size_t node_hash(cht_t *h, const cht_link_t *item)
{
	assert(item->hash == h->invalid_hash ||
	    item->hash == sentinel.hash ||
	    item->hash == calc_node_hash(h, item));

	return item->hash;
}

/** Calculates and mixed the hash of the item. */
static inline size_t calc_node_hash(cht_t *h, const cht_link_t *item)
{
	assert(item != &sentinel);
	/*
	 * Clear the lowest order bit in order for sentinel's node hash
	 * to be the greatest possible.
	 */
	return hash_mix(h->op->hash(item)) & ~(size_t)1;
}

/** Computes and memoizes the hash of the item. */
static inline void memoize_node_hash(cht_t *h, cht_link_t *item)
{
	item->hash = calc_node_hash(h, item);
}

/** Packs the next pointer address and the mark into a single pointer. */
static inline marked_ptr_t make_link(const cht_link_t *next, mark_t mark)
{
	marked_ptr_t ptr = (marked_ptr_t) next;

	assert(!(ptr & N_MARK_MASK));
	assert(!((unsigned)mark & ~N_MARK_MASK));

	return ptr | mark;
}

/** Strips any marks from the next item link and returns the next item's address.*/
static inline cht_link_t *get_next(marked_ptr_t link)
{
	return (cht_link_t *)(link & ~N_MARK_MASK);
}

/** Returns the current node's mark stored in the next item link. */
static inline mark_t get_mark(marked_ptr_t link)
{
	return (mark_t)(link & N_MARK_MASK);
}

/** Moves the window by one item so that is points to the next item. */
static inline void next_wnd(wnd_t *wnd)
{
	assert(wnd);
	assert(wnd->cur);

	wnd->last = wnd->cur;
	wnd->ppred = &wnd->cur->link;
	wnd->cur = get_next(wnd->cur->link);
}

/** Predicate that matches only exactly the same node. */
static bool same_node_pred(void *node, const cht_link_t *item2)
{
	const cht_link_t *item1 = (const cht_link_t *) node;
	return item1 == item2;
}

/** Compare-and-swaps a next item link. */
static inline marked_ptr_t cas_link(marked_ptr_t *link, const cht_link_t *cur_next,
    mark_t cur_mark, const cht_link_t *new_next, mark_t new_mark)
{
	return _cas_link(link, make_link(cur_next, cur_mark),
	    make_link(new_next, new_mark));
}

/** Compare-and-swaps a next item link. */
static inline marked_ptr_t _cas_link(marked_ptr_t *link, marked_ptr_t cur,
    marked_ptr_t new)
{
	assert(link != &sentinel.link);
	/*
	 * cas(x) on the same location x on one cpu must be ordered, but do not
	 * have to be ordered wrt to other cas(y) to a different location y
	 * on the same cpu.
	 *
	 * cas(x) must act as a write barrier on x, ie if cas(x) succeeds
	 * and is observed by another cpu, then all cpus must be able to
	 * make the effects of cas(x) visible just by issuing a load barrier.
	 * For example:
	 * cpu1         cpu2            cpu3
	 *                              cas(x, 0 -> 1), succeeds
	 *              cas(x, 0 -> 1), fails
	 *              MB, to order load of x in cas and store to y
	 *              y = 7
	 * sees y == 7
	 * loadMB must be enough to make cas(x) on cpu3 visible to cpu1, ie x == 1.
	 *
	 * If cas() did not work this way:
	 * a) our head move protocol would not be correct.
	 * b) freeing an item linked to a moved head after another item was
	 *   inserted in front of it, would require more than one grace period.
	 *
	 * Ad (a): In the following example, cpu1 starts moving old_head
	 * to new_head, cpu2 completes the move and cpu3 notices cpu2
	 * completed the move before cpu1 gets a chance to notice cpu2
	 * had already completed the move. Our requirements for cas()
	 * assume cpu3 will see a valid and mutable value in new_head
	 * after issuing a load memory barrier once it has determined
	 * the old_head's value had been successfully moved to new_head
	 * (because it sees old_head marked invalid).
	 *
	 *  cpu1             cpu2             cpu3
	 *   cas(old_head, <addr, N>, <addr, Const>), succeeds
	 *   cas-order-barrier
	 *   // Move from old_head to new_head started, now the interesting stuff:
	 *   cas(new_head, <0, Inv>, <addr, N>), succeeds
	 *
	 *                    cas(new_head, <0, Inv>, <addr, N>), but fails
	 *                    cas-order-barrier
	 *                    cas(old_head, <addr, Const>, <addr, Inv>), succeeds
	 *
	 *                                     Sees old_head marked Inv (by cpu2)
	 *                                     load-MB
	 *                                     assert(new_head == <addr, N>)
	 *
	 *   cas-order-barrier
	 *
	 * Even though cpu1 did not yet issue a cas-order-barrier, cpu1's store
	 * to new_head (successful cas()) must be made visible to cpu3 with
	 * a load memory barrier if cpu1's store to new_head is visible
	 * on another cpu (cpu2) and that cpu's (cpu2's) store to old_head
	 * is already visible to cpu3.	 *
	 */
	void *expected = (void *)cur;

	/*
	 * Use the acquire-release model, although we could probably
	 * get away even with the relaxed memory model due to our use
	 * of explicit memory barriers.
	 */
	__atomic_compare_exchange_n((void **)link, &expected, (void *)new, false,
	    __ATOMIC_ACQ_REL, __ATOMIC_ACQUIRE);

	return (marked_ptr_t) expected;
}

/** Orders compare-and-swaps to different memory locations. */
static inline void cas_order_barrier(void)
{
	/* Make sure CAS to different memory locations are ordered. */
	write_barrier();
}

/** @}
 */
