Index: kernel/Makefile =================================================================== --- kernel/Makefile (revision 947ab77e1138e5b43ed8d4e74ae82a491381bfec) +++ kernel/Makefile (revision aae365bc823f3137811d2c48089c150e81c40606) @@ -163,5 +163,4 @@ generic/src/adt/bitmap.c \ generic/src/adt/btree.c \ - generic/src/adt/cht.c \ generic/src/adt/hash_table.c \ generic/src/adt/list.c \ @@ -229,6 +228,4 @@ generic/src/synch/waitq.c \ generic/src/synch/futex.c \ - generic/src/synch/workqueue.c \ - generic/src/synch/rcu.c \ generic/src/smp/ipi.c \ generic/src/smp/smp.c \ @@ -283,5 +280,4 @@ test/atomic/atomic1.c \ test/btree/btree1.c \ - test/cht/cht1.c \ test/fault/fault1.c \ test/mm/falloc1.c \ @@ -292,7 +288,4 @@ test/synch/semaphore1.c \ test/synch/semaphore2.c \ - test/synch/workqueue2.c \ - test/synch/workqueue3.c \ - test/synch/rcu1.c \ test/print/print1.c \ test/print/print2.c \ Index: kernel/generic/include/adt/cht.h =================================================================== --- kernel/generic/include/adt/cht.h (revision 947ab77e1138e5b43ed8d4e74ae82a491381bfec) +++ (revision ) @@ -1,154 +1,0 @@ -/* - * 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 kernel_generic_adt - * @{ - */ -/** @file - */ - -#ifndef KERN_CONC_HASH_TABLE_H_ -#define KERN_CONC_HASH_TABLE_H_ - -#include -#include -#include -#include -#include -#include - -typedef uintptr_t cht_ptr_t; - -/** Concurrent hash table node link. */ -typedef struct cht_link { - /* - * Must be placed first. - * - * The function pointer (rcu_link.func) is used to store the item's - * mixed memoized hash. If in use by RCU (ie waiting for deferred - * destruction) the hash will contain the value of - * cht_t.op->remove_callback. - */ - union { - rcu_item_t rcu_link; - size_t hash; - }; - /** Link to the next item in the bucket including any marks. */ - cht_ptr_t link; -} cht_link_t; - -/** Set of operations for a concurrent hash table. */ -typedef struct cht_ops { - /** Returns the hash of the item. - * - * Applicable also to items that were logically deleted from the table - * but have yet to be physically removed by means of remove_callback(). - */ - size_t (*hash)(const cht_link_t *item); - /** Returns the hash value of the key used to search for entries. */ - size_t (*key_hash)(void *key); - /** Returns true if the two items store equal search keys. */ - bool (*equal)(const cht_link_t *item1, const cht_link_t *item2); - /** Returns true if the item contains an equal search key. */ - bool (*key_equal)(void *key, const cht_link_t *item); - /** Invoked to free a removed item once all references to it are dropped. */ - void (*remove_callback)(cht_link_t *item); -} cht_ops_t; - -/** Groups hash table buckets with their count. - * - * It allows both the number of buckets as well as the bucket array - * to be swapped atomically when resing the table. - */ -typedef struct cht_buckets { - /** The number of buckets is 2^order. */ - size_t order; - /** Array of single linked list bucket heads along with any marks. */ - cht_ptr_t head[1]; -} cht_buckets_t; - -/** Concurrent hash table structure. */ -typedef struct { - /** Item specific operations. */ - cht_ops_t *op; - - /** Buckets currently in use. */ - cht_buckets_t *b; - /** Resized table buckets that will replace b once resize is complete. */ - cht_buckets_t *new_b; - /** Invalid memoized hash value. - * - * If cht_link.hash contains this value the item had been logically - * removed and is waiting to be freed. Such hashes (and the associated - * items) are disregarded and skipped or the actual hash must be - * determined via op->hash(). - */ - size_t invalid_hash; - - /** Minimum number of buckets is 2^min_order. */ - size_t min_order; - /** Maximum number of items per bucket before the table grows. */ - size_t max_load; - /** Table is resized in the background in a work queue. */ - work_t resize_work; - /** If positive the table should grow or shrink. - * - * If not 0 resize work had already been posted to the system work queue. - */ - atomic_t resize_reqs; - - /** Number of items in the table that have not been logically deleted. */ - atomic_t item_cnt; -} cht_t; - -#define cht_get_inst(item, type, member) \ - member_to_inst((item), type, member) - -#define cht_read_lock() rcu_read_lock() -#define cht_read_unlock() rcu_read_unlock() - -extern bool cht_create_simple(cht_t *h, cht_ops_t *op); -extern 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); -extern void cht_destroy(cht_t *h); -extern void cht_destroy_unsafe(cht_t *h); - -extern cht_link_t *cht_find(cht_t *h, void *key); -extern cht_link_t *cht_find_lazy(cht_t *h, void *key); -extern cht_link_t *cht_find_next(cht_t *h, const cht_link_t *item); -extern cht_link_t *cht_find_next_lazy(cht_t *h, const cht_link_t *item); - -extern void cht_insert(cht_t *h, cht_link_t *item); -extern bool cht_insert_unique(cht_t *h, cht_link_t *item, cht_link_t **dup_item); -extern size_t cht_remove_key(cht_t *h, void *key); -extern bool cht_remove_item(cht_t *h, cht_link_t *item); - -#endif - -/** @} - */ Index: kernel/generic/include/arch.h =================================================================== --- kernel/generic/include/arch.h (revision 947ab77e1138e5b43ed8d4e74ae82a491381bfec) +++ kernel/generic/include/arch.h (revision aae365bc823f3137811d2c48089c150e81c40606) @@ -66,7 +66,4 @@ typedef struct { size_t preemption; /**< Preemption disabled counter and flag. */ -#ifdef RCU_PREEMPT_A - size_t rcu_nesting; /**< RCU nesting count and flag. */ -#endif struct thread *thread; /**< Current thread. */ struct task *task; /**< Current task. */ Index: kernel/generic/include/cpu.h =================================================================== --- kernel/generic/include/cpu.h (revision 947ab77e1138e5b43ed8d4e74ae82a491381bfec) +++ kernel/generic/include/cpu.h (revision aae365bc823f3137811d2c48089c150e81c40606) @@ -38,5 +38,4 @@ #include #include -#include #include #include @@ -104,7 +103,4 @@ list_t smp_pending_calls; - /** RCU per-cpu data. Uses own locking. */ - rcu_cpu_data_t rcu; - /** * Stack used by scheduler when there is no running thread. Index: kernel/generic/include/proc/task.h =================================================================== --- kernel/generic/include/proc/task.h (revision 947ab77e1138e5b43ed8d4e74ae82a491381bfec) +++ kernel/generic/include/proc/task.h (revision aae365bc823f3137811d2c48089c150e81c40606) @@ -43,6 +43,4 @@ #include #include -#include -#include #include #include Index: kernel/generic/include/proc/thread.h =================================================================== --- kernel/generic/include/proc/thread.h (revision 947ab77e1138e5b43ed8d4e74ae82a491381bfec) +++ kernel/generic/include/proc/thread.h (revision aae365bc823f3137811d2c48089c150e81c40606) @@ -41,5 +41,4 @@ #include #include -#include #include #include @@ -194,16 +193,4 @@ thread_id_t tid; - /** Work queue this thread belongs to or NULL. Immutable. */ - struct work_queue *workq; - /** Links work queue threads. Protected by workq->lock. */ - link_t workq_link; - /** True if the worker was blocked and is not running. Use thread->lock. */ - bool workq_blocked; - /** True if the worker will block in order to become idle. Use workq->lock. */ - bool workq_idling; - - /** RCU thread related data. Protected by its own locks. */ - rcu_thread_data_t rcu; - /** Architecture-specific data. */ thread_arch_t arch; Index: kernel/generic/include/synch/rcu.h =================================================================== --- kernel/generic/include/synch/rcu.h (revision 947ab77e1138e5b43ed8d4e74ae82a491381bfec) +++ (revision ) @@ -1,241 +1,0 @@ -/* - * 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 kernel_sync - * @{ - */ -/** @file - */ - -#ifndef KERN_RCU_H_ -#define KERN_RCU_H_ - -#include -#include -#include - -/** Use to assign a pointer to newly initialized data to a rcu reader - * accessible pointer. - * - * Example: - * @code - * typedef struct exam { - * struct exam *next; - * int grade; - * } exam_t; - * - * exam_t *exam_list; - * // .. - * - * // Insert at the beginning of the list. - * exam_t *my_exam = malloc(sizeof(exam_t)); - * my_exam->grade = 5; - * my_exam->next = exam_list; - * rcu_assign(exam_list, my_exam); - * - * // Changes properly propagate. Every reader either sees - * // the old version of exam_list or the new version with - * // the fully initialized my_exam. - * rcu_synchronize(); - * // Now we can be sure every reader sees my_exam. - * - * @endcode - */ -#define rcu_assign(ptr, value) \ - do { \ - memory_barrier(); \ - (ptr) = (value); \ - } while (0) - -/** Use to access RCU protected data in a reader section. - * - * Example: - * @code - * exam_t *exam_list; - * // ... - * - * rcu_read_lock(); - * exam_t *first_exam = rcu_access(exam_list); - * // We can now safely use first_exam, it won't change - * // under us while we're using it. - * - * // .. - * rcu_read_unlock(); - * @endcode - */ -#define rcu_access(ptr) ACCESS_ONCE(ptr) - -#include -#include -#include -#include - -extern bool rcu_read_locked(void); -extern void rcu_synchronize(void); -extern void rcu_synchronize_expedite(void); -extern void rcu_call(rcu_item_t *rcu_item, rcu_func_t func); -extern void rcu_barrier(void); - -extern void rcu_print_stat(void); - -extern void rcu_init(void); -extern void rcu_stop(void); -extern void rcu_cpu_init(void); -extern void rcu_kinit_init(void); -extern void rcu_thread_init(struct thread *); -extern void rcu_thread_exiting(void); -extern void rcu_after_thread_ran(void); -extern void rcu_before_thread_runs(void); - -extern uint64_t rcu_completed_gps(void); -extern void _rcu_call(bool expedite, rcu_item_t *rcu_item, rcu_func_t func); -extern void _rcu_synchronize(bool expedite); - -#ifdef RCU_PREEMPT_A - -#define RCU_CNT_INC (1 << 1) -#define RCU_WAS_PREEMPTED (1 << 0) - -/* Fwd. decl. because of inlining. */ -void _rcu_preempted_unlock(void); - -/** Delimits the start of an RCU reader critical section. - * - * Reader sections may be nested and are preemptible. You must not - * however block/sleep within reader sections. - */ -static inline void rcu_read_lock(void) -{ - CURRENT->rcu_nesting += RCU_CNT_INC; - compiler_barrier(); -} - -/** Delimits the end of an RCU reader critical section. */ -static inline void rcu_read_unlock(void) -{ - compiler_barrier(); - CURRENT->rcu_nesting -= RCU_CNT_INC; - - if (RCU_WAS_PREEMPTED == CURRENT->rcu_nesting) { - _rcu_preempted_unlock(); - } -} - -#elif defined(RCU_PREEMPT_PODZIMEK) - -/* Fwd decl. required by the inlined implementation. Not part of public API. */ -extern rcu_gp_t _rcu_cur_gp; -extern void _rcu_signal_read_unlock(void); - -/** Unconditionally records a quiescent state for the local cpu. */ -static inline void _rcu_record_qs(void) -{ - assert(PREEMPTION_DISABLED || interrupts_disabled()); - - /* - * A new GP was started since the last time we passed a QS. - * Notify the detector we have reached a new QS. - */ - if (CPU->rcu.last_seen_gp != _rcu_cur_gp) { - rcu_gp_t cur_gp = ACCESS_ONCE(_rcu_cur_gp); - /* - * Contain memory accesses within a reader critical section. - * If we are in rcu_lock() it also makes changes prior to the - * start of the GP visible in the reader section. - */ - memory_barrier(); - /* - * Acknowledge we passed a QS since the beginning of rcu.cur_gp. - * Cache coherency will lazily transport the value to the - * detector while it sleeps in gp_sleep(). - * - * Note that there is a theoretical possibility that we - * overwrite a more recent/greater last_seen_gp here with - * an older/smaller value. If this cpu is interrupted here - * while in rcu_lock() reader sections in the interrupt handler - * will update last_seen_gp to the same value as is currently - * in local cur_gp. However, if the cpu continues processing - * interrupts and the detector starts a new GP immediately, - * local interrupt handlers may update last_seen_gp again (ie - * properly ack the new GP) with a value greater than local cur_gp. - * Resetting last_seen_gp to a previous value here is however - * benign and we only have to remember that this reader may end up - * in cur_preempted even after the GP ends. That is why we - * append next_preempted to cur_preempted rather than overwriting - * it as if cur_preempted were empty. - */ - CPU->rcu.last_seen_gp = cur_gp; - } -} - -/** Delimits the start of an RCU reader critical section. - * - * Reader sections may be nested and are preemptable. You must not - * however block/sleep within reader sections. - */ -static inline void rcu_read_lock(void) -{ - assert(CPU); - preemption_disable(); - - /* Record a QS if not in a reader critical section. */ - if (0 == CPU->rcu.nesting_cnt) - _rcu_record_qs(); - - ++CPU->rcu.nesting_cnt; - - preemption_enable(); -} - -/** Delimits the end of an RCU reader critical section. */ -static inline void rcu_read_unlock(void) -{ - assert(CPU); - preemption_disable(); - - if (0 == --CPU->rcu.nesting_cnt) { - _rcu_record_qs(); - - /* - * The thread was preempted while in a critical section or - * the detector is eagerly waiting for this cpu's reader to finish. - */ - if (CPU->rcu.signal_unlock) { - /* Rechecks with disabled interrupts. */ - _rcu_signal_read_unlock(); - } - } - - preemption_enable(); -} -#endif - -#endif - -/** @} - */ Index: kernel/generic/include/synch/rcu_types.h =================================================================== --- kernel/generic/include/synch/rcu_types.h (revision 947ab77e1138e5b43ed8d4e74ae82a491381bfec) +++ (revision ) @@ -1,167 +1,0 @@ -/* - * 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 kernel_sync - * @{ - */ -/** @file - */ - -#ifndef KERN_RCU_TYPES_H_ -#define KERN_RCU_TYPES_H_ - -#include -#include - -#if !defined(RCU_PREEMPT_PODZIMEK) && !defined(RCU_PREEMPT_A) -#error You must select an RCU algorithm. -#endif - -/* Fwd decl. */ -struct thread; -struct rcu_item; - -/** Grace period number typedef. */ -typedef uint64_t rcu_gp_t; - -/** RCU callback type. The passed rcu_item_t maybe freed. */ -typedef void (*rcu_func_t)(struct rcu_item *rcu_item); - -typedef struct rcu_item { - rcu_func_t func; - struct rcu_item *next; -} rcu_item_t; - -/** RCU related per-cpu data. */ -typedef struct rcu_cpu_data { - /** The cpu recorded a quiescent state last time during this grace period.*/ - rcu_gp_t last_seen_gp; - -#ifdef RCU_PREEMPT_PODZIMEK - /** This cpu has not yet passed a quiescent state and it is delaying the - * detector. Once it reaches a QS it must sema_up(rcu.remaining_readers). - */ - bool is_delaying_gp; - - /** True if we should signal the detector that we exited a reader section. - * - * Equal to (THREAD->rcu.was_preempted || CPU->rcu.is_delaying_gp). - */ - bool signal_unlock; - - /** The number of times an RCU reader section is nested on this cpu. - * - * If positive, it is definitely executing reader code. If zero, - * the thread might already be executing reader code thanks to - * cpu instruction reordering. - */ - size_t nesting_cnt; -#endif - - /** Callbacks to invoke once the current grace period ends, ie cur_cbs_gp. - * Accessed by the local reclaimer only. - */ - rcu_item_t *cur_cbs; - /** Number of callbacks in cur_cbs. */ - size_t cur_cbs_cnt; - /** Callbacks to invoke once the next grace period ends, ie next_cbs_gp. - * Accessed by the local reclaimer only. - */ - rcu_item_t *next_cbs; - /** Number of callbacks in next_cbs. */ - size_t next_cbs_cnt; - /** New callbacks are place at the end of this list. */ - rcu_item_t *arriving_cbs; - /** Tail of arriving_cbs list. Disable interrupts to access. */ - rcu_item_t **parriving_cbs_tail; - /** Number of callbacks currently in arriving_cbs. - * Disable interrupts to access. - */ - size_t arriving_cbs_cnt; - - /** At the end of this grace period callbacks in cur_cbs will be invoked.*/ - rcu_gp_t cur_cbs_gp; - /** At the end of this grace period callbacks in next_cbs will be invoked. - * - * Should be the next grace period but it allows the reclaimer to - * notice if it missed a grace period end announcement. In that - * case it can execute next_cbs without waiting for another GP. - * - * Invariant: next_cbs_gp >= cur_cbs_gp - */ - rcu_gp_t next_cbs_gp; - - /** Positive if there are callbacks pending in arriving_cbs. */ - semaphore_t arrived_flag; - - /** The reclaimer should expedite GPs for cbs in arriving_cbs. */ - bool expedite_arriving; - - /** Protected by global rcu.barrier_mtx. */ - rcu_item_t barrier_item; - - /** Interruptable attached reclaimer thread. */ - struct thread *reclaimer_thr; - - /* Some statistics. */ - size_t stat_max_cbs; - size_t stat_avg_cbs; - size_t stat_missed_gps; - size_t stat_missed_gp_in_wait; - size_t stat_max_slice_cbs; - size_t last_arriving_cnt; -} rcu_cpu_data_t; - -/** RCU related per-thread data. */ -typedef struct rcu_thread_data { - /** - * Nesting count of the thread's RCU read sections when the thread - * is not running. - */ - size_t nesting_cnt; - -#ifdef RCU_PREEMPT_PODZIMEK - - /** True if the thread was preempted in a reader section. - * - * The thread is placed into rcu.cur_preempted or rcu.next_preempted - * and must remove itself in rcu_read_unlock(). - * - * Access with interrupts disabled. - */ - bool was_preempted; -#endif - - /** Preempted threads link. Access with rcu.prempt_lock.*/ - link_t preempt_link; -} rcu_thread_data_t; - -#endif - -/** @} - */ Index: kernel/generic/include/synch/workqueue.h =================================================================== --- kernel/generic/include/synch/workqueue.h (revision 947ab77e1138e5b43ed8d4e74ae82a491381bfec) +++ (revision ) @@ -1,80 +1,0 @@ -/* - * 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 kernel_generic - * @{ - */ -/** @file - */ - -#ifndef KERN_WORKQUEUE_H_ -#define KERN_WORKQUEUE_H_ - -#include - -/* Fwd decl. */ -struct thread; -struct work_item; -struct work_queue; -typedef struct work_queue work_queue_t; - -typedef void (*work_func_t)(struct work_item *); - -typedef struct work_item { - link_t queue_link; - work_func_t func; - -#ifdef CONFIG_DEBUG - /* Magic number for integrity checks. */ - uint32_t cookie; -#endif -} work_t; - -extern void workq_global_init(void); -extern void workq_global_worker_init(void); -extern void workq_global_stop(void); -extern bool workq_global_enqueue_noblock(work_t *, work_func_t); -extern bool workq_global_enqueue(work_t *, work_func_t); - -extern struct work_queue *workq_create(const char *); -extern void workq_destroy(struct work_queue *); -extern bool workq_init(struct work_queue *, const char *); -extern void workq_stop(struct work_queue *); -extern bool workq_enqueue_noblock(struct work_queue *, work_t *, work_func_t); -extern bool workq_enqueue(struct work_queue *, work_t *, work_func_t); - -extern void workq_print_info(struct work_queue *); -extern void workq_global_print_info(void); - -extern void workq_after_thread_ran(void); -extern void workq_before_thread_is_ready(struct thread *); - -#endif /* KERN_WORKQUEUE_H_ */ - -/** @} - */ Index: kernel/generic/src/adt/cht.c =================================================================== --- kernel/generic/src/adt/cht.c (revision 947ab77e1138e5b43ed8d4e74ae82a491381bfec) +++ (revision ) @@ -1,2708 +1,0 @@ -/* - * 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 kernel_generic_adt - * @{ - */ - -/** - * @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 -#include -#include -#include -#include -#include -#include -#include - -/* 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, , ), succeeds - * cas-order-barrier - * // Move from old_head to new_head started, now the interesting stuff: - * cas(new_head, <0, Inv>, ), succeeds - * - * cas(new_head, <0, Inv>, ), but fails - * cas-order-barrier - * cas(old_head, , ), succeeds - * - * Sees old_head marked Inv (by cpu2) - * load-MB - * assert(new_head == ) - * - * 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(); -} - -/** @} - */ Index: kernel/generic/src/console/cmd.c =================================================================== --- kernel/generic/src/console/cmd.c (revision 947ab77e1138e5b43ed8d4e74ae82a491381bfec) +++ kernel/generic/src/console/cmd.c (revision aae365bc823f3137811d2c48089c150e81c40606) @@ -70,6 +70,4 @@ #include #include -#include -#include #include @@ -536,22 +534,4 @@ }; -/* Data and methods for the 'workq' command */ -static int cmd_workq(cmd_arg_t *argv); -static cmd_info_t workq_info = { - .name = "workq", - .description = "Show global workq information.", - .func = cmd_workq, - .argc = 0 -}; - -/* Data and methods for the 'workq' command */ -static int cmd_rcu(cmd_arg_t *argv); -static cmd_info_t rcu_info = { - .name = "rcu", - .description = "Show RCU run-time statistics.", - .func = cmd_rcu, - .argc = 0 -}; - /* Data and methods for 'ipc' command */ static int cmd_ipc(cmd_arg_t *argv); @@ -618,5 +598,4 @@ &physmem_info, &reboot_info, - &rcu_info, &sched_info, &set4_info, @@ -628,5 +607,4 @@ &uptime_info, &version_info, - &workq_info, &zones_info, &zone_info, @@ -1302,28 +1280,4 @@ } -/** Prints information about the global work queue. - * - * @param argv Ignores - * - * @return Always 1 - */ -int cmd_workq(cmd_arg_t *argv) -{ - workq_global_print_info(); - return 1; -} - -/** Prints RCU statistics. - * - * @param argv Ignores - * - * @return Always 1 - */ -int cmd_rcu(cmd_arg_t *argv) -{ - rcu_print_stat(); - return 1; -} - /** Command for listing memory zones * Index: kernel/generic/src/cpu/cpu.c =================================================================== --- kernel/generic/src/cpu/cpu.c (revision 947ab77e1138e5b43ed8d4e74ae82a491381bfec) +++ kernel/generic/src/cpu/cpu.c (revision aae365bc823f3137811d2c48089c150e81c40606) @@ -50,5 +50,4 @@ #include #include -#include cpu_t *cpus; @@ -108,5 +107,4 @@ cpu_identify(); cpu_arch_init(); - rcu_cpu_init(); } Index: kernel/generic/src/main/kinit.c =================================================================== --- kernel/generic/src/main/kinit.c (revision 947ab77e1138e5b43ed8d4e74ae82a491381bfec) +++ kernel/generic/src/main/kinit.c (revision aae365bc823f3137811d2c48089c150e81c40606) @@ -79,6 +79,4 @@ #include #include -#include -#include #define ALIVE_CHARS 4 @@ -109,12 +107,4 @@ interrupts_disable(); - - /* Start processing RCU callbacks. RCU is fully functional afterwards. */ - rcu_kinit_init(); - - /* - * Start processing work queue items. Some may have been queued during boot. - */ - workq_global_worker_init(); #ifdef CONFIG_SMP Index: kernel/generic/src/main/main.c =================================================================== --- kernel/generic/src/main/main.c (revision 947ab77e1138e5b43ed8d4e74ae82a491381bfec) +++ kernel/generic/src/main/main.c (revision aae365bc823f3137811d2c48089c150e81c40606) @@ -78,5 +78,4 @@ #include #include -#include #include #include @@ -275,5 +274,4 @@ smp_call_init(); - workq_global_init(); clock_counter_init(); timeout_init(); Index: kernel/generic/src/proc/scheduler.c =================================================================== --- kernel/generic/src/proc/scheduler.c (revision 947ab77e1138e5b43ed8d4e74ae82a491381bfec) +++ kernel/generic/src/proc/scheduler.c (revision aae365bc823f3137811d2c48089c150e81c40606) @@ -54,6 +54,4 @@ #include #include -#include -#include #include #include @@ -90,5 +88,4 @@ { before_thread_runs_arch(); - rcu_before_thread_runs(); #ifdef CONFIG_FPU_LAZY @@ -131,6 +128,4 @@ static void after_thread_ran(void) { - workq_after_thread_ran(); - rcu_after_thread_ran(); after_thread_ran_arch(); } @@ -430,5 +425,4 @@ case Exiting: - rcu_thread_exiting(); repeat: if (THREAD->detached) { Index: kernel/generic/src/proc/the.c =================================================================== --- kernel/generic/src/proc/the.c (revision 947ab77e1138e5b43ed8d4e74ae82a491381bfec) +++ kernel/generic/src/proc/the.c (revision aae365bc823f3137811d2c48089c150e81c40606) @@ -60,7 +60,4 @@ the->as = NULL; the->magic = MAGIC; -#ifdef RCU_PREEMPT_A - the->rcu_nesting = 0; -#endif } Index: kernel/generic/src/proc/thread.c =================================================================== --- kernel/generic/src/proc/thread.c (revision 947ab77e1138e5b43ed8d4e74ae82a491381bfec) +++ kernel/generic/src/proc/thread.c (revision aae365bc823f3137811d2c48089c150e81c40606) @@ -48,6 +48,4 @@ #include #include -#include -#include #include #include @@ -69,4 +67,5 @@ #include #include +#include /** Thread states */ @@ -272,5 +271,4 @@ { assert(irq_spinlock_locked(&thread->lock)); - workq_before_thread_is_ready(thread); } @@ -399,6 +397,4 @@ thread->task = task; - thread->workq = NULL; - thread->fpu_context_exists = false; thread->fpu_context_engaged = false; @@ -414,6 +410,4 @@ /* Might depend on previous initialization */ thread_create_arch(thread); - - rcu_thread_init(thread); if ((flags & THREAD_FLAG_NOATTACH) != THREAD_FLAG_NOATTACH) Index: kernel/generic/src/synch/rcu.c =================================================================== --- kernel/generic/src/synch/rcu.c (revision 947ab77e1138e5b43ed8d4e74ae82a491381bfec) +++ (revision ) @@ -1,1863 +1,0 @@ -/* - * 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 kernel_sync - * @{ - */ - -/** - * @file - * @brief Preemptible read-copy update. Usable from interrupt handlers. - * - * @par Podzimek-preempt-RCU (RCU_PREEMPT_PODZIMEK) - * - * Podzimek-preempt-RCU is a preemptible variant of Podzimek's non-preemptible - * RCU algorithm [1, 2]. Grace period (GP) detection is centralized into a - * single detector thread. The detector requests that each cpu announces - * that it passed a quiescent state (QS), ie a state when the cpu is - * outside of an rcu reader section (CS). Cpus check for QSs during context - * switches and when entering and exiting rcu reader sections. Once all - * cpus announce a QS and if there were no threads preempted in a CS, the - * GP ends. - * - * The detector increments the global GP counter, _rcu_cur_gp, in order - * to start a new GP. Readers notice the new GP by comparing the changed - * _rcu_cur_gp to a locally stored value last_seen_gp which denotes the - * the last GP number for which the cpu noted an explicit QS (and issued - * a memory barrier). Readers check for the change in the outer-most - * (ie not nested) rcu_read_lock()/unlock() as these functions represent - * a QS. The reader first executes a memory barrier (MB) in order to contain - * memory references within a CS (and to make changes made by writers - * visible in the CS following rcu_read_lock()). Next, the reader notes - * that it reached a QS by updating the cpu local last_seen_gp to the - * global GP counter, _rcu_cur_gp. Cache coherency eventually makes - * the updated last_seen_gp visible to the detector cpu, much like it - * delivered the changed _rcu_cur_gp to all cpus. - * - * The detector waits a while after starting a GP and then reads each - * cpu's last_seen_gp to see if it reached a QS. If a cpu did not record - * a QS (might be a long running thread without an RCU reader CS; or cache - * coherency has yet to make the most current last_seen_gp visible to - * the detector; or the cpu is still in a CS) the cpu is interrupted - * via an IPI. If the IPI handler finds the cpu still in a CS, it instructs - * the cpu to notify the detector that it had exited the CS via a semaphore - * (CPU->rcu.is_delaying_gp). - * The detector then waits on the semaphore for any cpus to exit their - * CSs. Lastly, it waits for the last reader preempted in a CS to - * exit its CS if there were any and signals the end of the GP to - * separate reclaimer threads wired to each cpu. Reclaimers then - * execute the callbacks queued on each of the cpus. - * - * - * @par A-RCU algorithm (RCU_PREEMPT_A) - * - * A-RCU is based on the user space rcu algorithm in [3] utilizing signals - * (urcu) and Podzimek's rcu [1]. Like in Podzimek's rcu, callbacks are - * executed by cpu-bound reclaimer threads. There is however no dedicated - * detector thread and the reclaimers take on the responsibilities of the - * detector when they need to start a new GP. A new GP is again announced - * and acknowledged with _rcu_cur_gp and the cpu local last_seen_gp. Unlike - * Podzimek's rcu, cpus check explicitly for QS only during context switches. - * Like in urcu, rcu_read_lock()/unlock() only maintain the nesting count - * and never issue any memory barriers. This makes rcu_read_lock()/unlock() - * simple and fast. - * - * If a new callback is queued for a reclaimer and no GP is in progress, - * the reclaimer takes on the role of a detector. The detector increments - * _rcu_cur_gp in order to start a new GP. It waits a while to give cpus - * a chance to switch a context (a natural QS). Then, it examines each - * non-idle cpu that has yet to pass a QS via an IPI. The IPI handler - * sees the most current _rcu_cur_gp and last_seen_gp and notes a QS - * with a memory barrier and an update to last_seen_gp. If the handler - * finds the cpu in a CS it does nothing and let the detector poll/interrupt - * the cpu again after a short sleep. - * - * @par Caveats - * - * last_seen_gp and _rcu_cur_gp are always 64bit variables and they - * are read non-atomically on 32bit machines. Reading a clobbered - * value of last_seen_gp or _rcu_cur_gp or writing a clobbered value - * of _rcu_cur_gp to last_seen_gp will at worst force the detector - * to unnecessarily interrupt a cpu. Interrupting a cpu makes the - * correct value of _rcu_cur_gp visible to the cpu and correctly - * resets last_seen_gp in both algorithms. - * - * - * - * [1] Read-copy-update for opensolaris, - * 2010, Podzimek - * https://andrej.podzimek.org/thesis.pdf - * - * [2] (podzimek-rcu) implementation file "rcu.patch" - * http://d3s.mff.cuni.cz/projects/operating_systems/rcu/rcu.patch - * - * [3] User-level implementations of read-copy update, - * 2012, appendix - * http://www.rdrop.com/users/paulmck/RCU/urcu-supp-accepted.2011.08.30a.pdf - * - */ - -#include -#include -#include -#include -#include -#include -#include -#include -#include -#include -#include -#include -#include -#include - -/* - * Number of milliseconds to give to preexisting readers to finish - * when non-expedited grace period detection is in progress. - */ -#define DETECT_SLEEP_MS 10 -/* - * Max number of pending callbacks in the local cpu's queue before - * aggressively expediting the current grace period - */ -#define EXPEDITE_THRESHOLD 2000 -/* - * Max number of callbacks to execute in one go with preemption - * enabled. If there are more callbacks to be executed they will - * be run with preemption disabled in order to prolong reclaimer's - * time slice and give it a chance to catch up with callback producers. - */ -#define CRITICAL_THRESHOLD 30000 -/* Half the number of values a uint32 can hold. */ -#define UINT32_MAX_HALF 2147483648U - -/** - * The current grace period number. Increases monotonically. - * Lock rcu.gp_lock or rcu.preempt_lock to get a current value. - */ -rcu_gp_t _rcu_cur_gp; - -/** Global RCU data. */ -typedef struct rcu_data { - /** Detector uses so signal reclaimers that a grace period ended. */ - condvar_t gp_ended; - /** Reclaimers use to notify the detector to accelerate GP detection. */ - condvar_t expedite_now; - /** - * Protects: req_gp_end_cnt, req_expedited_cnt, completed_gp, _rcu_cur_gp; - * or: completed_gp, _rcu_cur_gp - */ - SPINLOCK_DECLARE(gp_lock); - /** - * The number of the most recently completed grace period. At most - * one behind _rcu_cur_gp. If equal to _rcu_cur_gp, a grace period - * detection is not in progress and the detector is idle. - */ - rcu_gp_t completed_gp; - - /** Protects the following 3 fields. */ - IRQ_SPINLOCK_DECLARE(preempt_lock); - /** Preexisting readers that have been preempted. */ - list_t cur_preempted; - /** Reader that have been preempted and might delay the next grace period.*/ - list_t next_preempted; - /** - * The detector is waiting for the last preempted reader - * in cur_preempted to announce that it exited its reader - * section by up()ing remaining_readers. - */ - bool preempt_blocking_det; - -#ifdef RCU_PREEMPT_A - - /** - * The detector waits on this semaphore for any preempted readers - * delaying the grace period once all cpus pass a quiescent state. - */ - semaphore_t remaining_readers; - -#elif defined(RCU_PREEMPT_PODZIMEK) - - /** Reclaimers notify the detector when they request more grace periods.*/ - condvar_t req_gp_changed; - /** Number of grace period ends the detector was requested to announce. */ - size_t req_gp_end_cnt; - /** Number of consecutive grace periods to detect quickly and aggressively.*/ - size_t req_expedited_cnt; - /** - * Number of cpus with readers that are delaying the current GP. - * They will up() remaining_readers. - */ - atomic_t delaying_cpu_cnt; - /** - * The detector waits on this semaphore for any readers delaying the GP. - * - * Each of the cpus with readers that are delaying the current GP - * must up() this sema once they reach a quiescent state. If there - * are any readers in cur_preempted (ie preempted preexisting) and - * they are already delaying GP detection, the last to unlock its - * reader section must up() this sema once. - */ - semaphore_t remaining_readers; -#endif - - /** Excludes simultaneous rcu_barrier() calls. */ - mutex_t barrier_mtx; - /** Number of cpus that we are waiting for to complete rcu_barrier(). */ - atomic_t barrier_wait_cnt; - /** rcu_barrier() waits for the completion of barrier callbacks on this wq.*/ - waitq_t barrier_wq; - - /** Interruptible attached detector thread pointer. */ - thread_t *detector_thr; - - /* Some statistics. */ - size_t stat_expedited_cnt; - size_t stat_delayed_cnt; - size_t stat_preempt_blocking_cnt; - /* Does not contain self/local calls. */ - size_t stat_smp_call_cnt; -} rcu_data_t; - -static rcu_data_t rcu; - -static void start_reclaimers(void); -static void synch_complete(rcu_item_t *rcu_item); -static inline void rcu_call_impl(bool expedite, rcu_item_t *rcu_item, - rcu_func_t func); -static void add_barrier_cb(void *arg); -static void barrier_complete(rcu_item_t *barrier_item); -static bool arriving_cbs_empty(void); -static bool next_cbs_empty(void); -static bool cur_cbs_empty(void); -static bool all_cbs_empty(void); -static void reclaimer(void *arg); -static bool wait_for_pending_cbs(void); -static bool advance_cbs(void); -static void exec_completed_cbs(rcu_gp_t last_completed_gp); -static void exec_cbs(rcu_item_t **phead); -static bool wait_for_cur_cbs_gp_end(bool expedite, rcu_gp_t *last_completed_gp); -static void upd_missed_gp_in_wait(rcu_gp_t completed_gp); - -#ifdef RCU_PREEMPT_PODZIMEK -static void start_detector(void); -static void read_unlock_impl(size_t *pnesting_cnt); -static void req_detection(size_t req_cnt); -static bool cv_wait_for_gp(rcu_gp_t wait_on_gp); -static void detector(void *); -static bool wait_for_detect_req(void); -static void end_cur_gp(void); -static bool wait_for_readers(void); -static bool gp_sleep(void); -static void interrupt_delaying_cpus(cpu_mask_t *cpu_mask); -static bool wait_for_delaying_cpus(void); -#elif defined(RCU_PREEMPT_A) -static bool wait_for_readers(bool expedite); -static bool gp_sleep(bool *expedite); -#endif - -static void start_new_gp(void); -static void rm_quiescent_cpus(cpu_mask_t *cpu_mask); -static void sample_cpus(cpu_mask_t *reader_cpus, void *arg); -static void sample_local_cpu(void *); -static bool wait_for_preempt_reader(void); -static void note_preempted_reader(void); -static void rm_preempted_reader(void); -static void upd_max_cbs_in_slice(size_t arriving_cbs_cnt); - -/** Initializes global RCU structures. */ -void rcu_init(void) -{ - condvar_initialize(&rcu.gp_ended); - condvar_initialize(&rcu.expedite_now); - - spinlock_initialize(&rcu.gp_lock, "rcu.gp_lock"); - _rcu_cur_gp = 0; - rcu.completed_gp = 0; - - irq_spinlock_initialize(&rcu.preempt_lock, "rcu.preempt_lock"); - list_initialize(&rcu.cur_preempted); - list_initialize(&rcu.next_preempted); - rcu.preempt_blocking_det = false; - - mutex_initialize(&rcu.barrier_mtx, MUTEX_PASSIVE); - atomic_store(&rcu.barrier_wait_cnt, 0); - waitq_initialize(&rcu.barrier_wq); - - semaphore_initialize(&rcu.remaining_readers, 0); - -#ifdef RCU_PREEMPT_PODZIMEK - condvar_initialize(&rcu.req_gp_changed); - - rcu.req_gp_end_cnt = 0; - rcu.req_expedited_cnt = 0; - atomic_store(&rcu.delaying_cpu_cnt, 0); -#endif - - rcu.detector_thr = NULL; - - rcu.stat_expedited_cnt = 0; - rcu.stat_delayed_cnt = 0; - rcu.stat_preempt_blocking_cnt = 0; - rcu.stat_smp_call_cnt = 0; -} - -/** Initializes per-CPU RCU data. If on the boot cpu inits global data too.*/ -void rcu_cpu_init(void) -{ - if (config.cpu_active == 1) { - rcu_init(); - } - - CPU->rcu.last_seen_gp = 0; - -#ifdef RCU_PREEMPT_PODZIMEK - CPU->rcu.nesting_cnt = 0; - CPU->rcu.is_delaying_gp = false; - CPU->rcu.signal_unlock = false; -#endif - - CPU->rcu.cur_cbs = NULL; - CPU->rcu.cur_cbs_cnt = 0; - CPU->rcu.next_cbs = NULL; - CPU->rcu.next_cbs_cnt = 0; - CPU->rcu.arriving_cbs = NULL; - CPU->rcu.parriving_cbs_tail = &CPU->rcu.arriving_cbs; - CPU->rcu.arriving_cbs_cnt = 0; - - CPU->rcu.cur_cbs_gp = 0; - CPU->rcu.next_cbs_gp = 0; - - semaphore_initialize(&CPU->rcu.arrived_flag, 0); - - /* BSP creates reclaimer threads before AP's rcu_cpu_init() runs. */ - if (config.cpu_active == 1) - CPU->rcu.reclaimer_thr = NULL; - - CPU->rcu.stat_max_cbs = 0; - CPU->rcu.stat_avg_cbs = 0; - CPU->rcu.stat_missed_gps = 0; - CPU->rcu.stat_missed_gp_in_wait = 0; - CPU->rcu.stat_max_slice_cbs = 0; - CPU->rcu.last_arriving_cnt = 0; -} - -/** Completes RCU init. Creates and runs the detector and reclaimer threads.*/ -void rcu_kinit_init(void) -{ -#ifdef RCU_PREEMPT_PODZIMEK - start_detector(); -#endif - - start_reclaimers(); -} - -/** Initializes any per-thread RCU structures. */ -void rcu_thread_init(thread_t *thread) -{ - thread->rcu.nesting_cnt = 0; - -#ifdef RCU_PREEMPT_PODZIMEK - thread->rcu.was_preempted = false; -#endif - - link_initialize(&thread->rcu.preempt_link); -} - -/** Cleans up global RCU resources and stops dispatching callbacks. - * - * Call when shutting down the kernel. Outstanding callbacks will - * not be processed. Instead they will linger forever. - */ -void rcu_stop(void) -{ - /* Stop and wait for reclaimers. */ - for (unsigned int cpu_id = 0; cpu_id < config.cpu_active; ++cpu_id) { - assert(cpus[cpu_id].rcu.reclaimer_thr != NULL); - - if (cpus[cpu_id].rcu.reclaimer_thr) { - thread_interrupt(cpus[cpu_id].rcu.reclaimer_thr); - thread_join(cpus[cpu_id].rcu.reclaimer_thr); - thread_detach(cpus[cpu_id].rcu.reclaimer_thr); - cpus[cpu_id].rcu.reclaimer_thr = NULL; - } - } - -#ifdef RCU_PREEMPT_PODZIMEK - /* Stop the detector and wait. */ - if (rcu.detector_thr) { - thread_interrupt(rcu.detector_thr); - thread_join(rcu.detector_thr); - thread_detach(rcu.detector_thr); - rcu.detector_thr = NULL; - } -#endif -} - -/** Returns the number of elapsed grace periods since boot. */ -uint64_t rcu_completed_gps(void) -{ - spinlock_lock(&rcu.gp_lock); - uint64_t completed = rcu.completed_gp; - spinlock_unlock(&rcu.gp_lock); - - return completed; -} - -/** Creates and runs cpu-bound reclaimer threads. */ -static void start_reclaimers(void) -{ - for (unsigned int cpu_id = 0; cpu_id < config.cpu_count; ++cpu_id) { - char name[THREAD_NAME_BUFLEN] = { 0 }; - - snprintf(name, THREAD_NAME_BUFLEN - 1, "rcu-rec/%u", cpu_id); - - cpus[cpu_id].rcu.reclaimer_thr = - thread_create(reclaimer, NULL, TASK, THREAD_FLAG_NONE, name); - - if (!cpus[cpu_id].rcu.reclaimer_thr) - panic("Failed to create RCU reclaimer thread on cpu%u.", cpu_id); - - thread_wire(cpus[cpu_id].rcu.reclaimer_thr, &cpus[cpu_id]); - thread_ready(cpus[cpu_id].rcu.reclaimer_thr); - } -} - -#ifdef RCU_PREEMPT_PODZIMEK - -/** Starts the detector thread. */ -static void start_detector(void) -{ - rcu.detector_thr = - thread_create(detector, NULL, TASK, THREAD_FLAG_NONE, "rcu-det"); - - if (!rcu.detector_thr) - panic("Failed to create RCU detector thread."); - - thread_ready(rcu.detector_thr); -} - -/** Returns true if in an rcu reader section. */ -bool rcu_read_locked(void) -{ - preemption_disable(); - bool locked = 0 < CPU->rcu.nesting_cnt; - preemption_enable(); - - return locked; -} - -/** Unlocks the local reader section using the given nesting count. - * - * Preemption or interrupts must be disabled. - * - * @param pnesting_cnt Either &CPU->rcu.tmp_nesting_cnt or - * THREAD->rcu.nesting_cnt. - */ -static void read_unlock_impl(size_t *pnesting_cnt) -{ - assert(PREEMPTION_DISABLED || interrupts_disabled()); - - if (0 == --(*pnesting_cnt)) { - _rcu_record_qs(); - - /* - * The thread was preempted while in a critical section or - * the detector is eagerly waiting for this cpu's reader - * to finish. - * - * Note that THREAD may be NULL in scheduler() and not just during boot. - */ - if ((THREAD && THREAD->rcu.was_preempted) || CPU->rcu.is_delaying_gp) { - /* Rechecks with disabled interrupts. */ - _rcu_signal_read_unlock(); - } - } -} - -/** If necessary, signals the detector that we exited a reader section. */ -void _rcu_signal_read_unlock(void) -{ - assert(PREEMPTION_DISABLED || interrupts_disabled()); - - /* - * If an interrupt occurs here (even a NMI) it may beat us to - * resetting .is_delaying_gp or .was_preempted and up the semaphore - * for us. - */ - - /* - * If the detector is eagerly waiting for this cpu's reader to unlock, - * notify it that the reader did so. - */ - if (local_atomic_exchange(&CPU->rcu.is_delaying_gp, false)) { - semaphore_up(&rcu.remaining_readers); - } - - /* - * This reader was preempted while in a reader section. - * We might be holding up the current GP. Notify the - * detector if so. - */ - if (THREAD && local_atomic_exchange(&THREAD->rcu.was_preempted, false)) { - assert(link_used(&THREAD->rcu.preempt_link)); - - rm_preempted_reader(); - } - - /* If there was something to signal to the detector we have done so. */ - CPU->rcu.signal_unlock = false; -} - -#endif /* RCU_PREEMPT_PODZIMEK */ - -typedef struct synch_item { - waitq_t wq; - rcu_item_t rcu_item; -} synch_item_t; - -/** Blocks until all preexisting readers exit their critical sections. */ -void rcu_synchronize(void) -{ - _rcu_synchronize(false); -} - -/** Blocks until all preexisting readers exit their critical sections. */ -void rcu_synchronize_expedite(void) -{ - _rcu_synchronize(true); -} - -/** Blocks until all preexisting readers exit their critical sections. */ -void _rcu_synchronize(bool expedite) -{ - /* Calling from a reader section will deadlock. */ - assert(!rcu_read_locked()); - - synch_item_t completion; - - waitq_initialize(&completion.wq); - _rcu_call(expedite, &completion.rcu_item, synch_complete); - waitq_sleep(&completion.wq); -} - -/** rcu_synchronize's callback. */ -static void synch_complete(rcu_item_t *rcu_item) -{ - synch_item_t *completion = member_to_inst(rcu_item, synch_item_t, rcu_item); - assert(completion); - waitq_wakeup(&completion->wq, WAKEUP_FIRST); -} - -/** Waits for all outstanding rcu calls to complete. */ -void rcu_barrier(void) -{ - /* - * Serialize rcu_barrier() calls so we don't overwrite cpu.barrier_item - * currently in use by rcu_barrier(). - */ - mutex_lock(&rcu.barrier_mtx); - - /* - * Ensure we queue a barrier callback on all cpus before the already - * enqueued barrier callbacks start signaling completion. - */ - atomic_store(&rcu.barrier_wait_cnt, 1); - - DEFINE_CPU_MASK(cpu_mask); - cpu_mask_active(cpu_mask); - - cpu_mask_for_each(*cpu_mask, cpu_id) { - smp_call(cpu_id, add_barrier_cb, NULL); - } - - if (0 < atomic_predec(&rcu.barrier_wait_cnt)) { - waitq_sleep(&rcu.barrier_wq); - } - - mutex_unlock(&rcu.barrier_mtx); -} - -/** Issues a rcu_barrier() callback on the local cpu. - * - * Executed with interrupts disabled. - */ -static void add_barrier_cb(void *arg) -{ - assert(interrupts_disabled() || PREEMPTION_DISABLED); - atomic_inc(&rcu.barrier_wait_cnt); - rcu_call(&CPU->rcu.barrier_item, barrier_complete); -} - -/** Local cpu's rcu_barrier() completion callback. */ -static void barrier_complete(rcu_item_t *barrier_item) -{ - /* Is this the last barrier callback completed? */ - if (0 == atomic_predec(&rcu.barrier_wait_cnt)) { - /* Notify rcu_barrier() that we're done. */ - waitq_wakeup(&rcu.barrier_wq, WAKEUP_FIRST); - } -} - -/** Adds a callback to invoke after all preexisting readers finish. - * - * May be called from within interrupt handlers or RCU reader sections. - * - * @param rcu_item Used by RCU to track the call. Must remain - * until the user callback function is entered. - * @param func User callback function that will be invoked once a full - * grace period elapsed, ie at a time when all preexisting - * readers have finished. The callback should be short and must - * not block. If you must sleep, enqueue your work in the system - * work queue from the callback (ie workq_global_enqueue()). - */ -void rcu_call(rcu_item_t *rcu_item, rcu_func_t func) -{ - rcu_call_impl(false, rcu_item, func); -} - -/** rcu_call() implementation. See rcu_call() for comments. */ -void _rcu_call(bool expedite, rcu_item_t *rcu_item, rcu_func_t func) -{ - rcu_call_impl(expedite, rcu_item, func); -} - -/** rcu_call() inline-able implementation. See rcu_call() for comments. */ -static inline void rcu_call_impl(bool expedite, rcu_item_t *rcu_item, - rcu_func_t func) -{ - assert(rcu_item); - - rcu_item->func = func; - rcu_item->next = NULL; - - preemption_disable(); - - rcu_cpu_data_t *r = &CPU->rcu; - - rcu_item_t **prev_tail = - local_atomic_exchange(&r->parriving_cbs_tail, &rcu_item->next); - *prev_tail = rcu_item; - - /* Approximate the number of callbacks present. */ - ++r->arriving_cbs_cnt; - - if (expedite) { - r->expedite_arriving = true; - } - - bool first_cb = (prev_tail == &CPU->rcu.arriving_cbs); - - /* Added first callback - notify the reclaimer. */ - if (first_cb && !semaphore_count_get(&r->arrived_flag)) { - semaphore_up(&r->arrived_flag); - } - - preemption_enable(); -} - -static bool cur_cbs_empty(void) -{ - assert(THREAD && THREAD->wired); - return NULL == CPU->rcu.cur_cbs; -} - -static bool next_cbs_empty(void) -{ - assert(THREAD && THREAD->wired); - return NULL == CPU->rcu.next_cbs; -} - -/** Disable interrupts to get an up-to-date result. */ -static bool arriving_cbs_empty(void) -{ - assert(THREAD && THREAD->wired); - /* - * Accessing with interrupts enabled may at worst lead to - * a false negative if we race with a local interrupt handler. - */ - return NULL == CPU->rcu.arriving_cbs; -} - -static bool all_cbs_empty(void) -{ - return cur_cbs_empty() && next_cbs_empty() && arriving_cbs_empty(); -} - -/** Reclaimer thread dispatches locally queued callbacks once a GP ends. */ -static void reclaimer(void *arg) -{ - assert(THREAD && THREAD->wired); - assert(THREAD == CPU->rcu.reclaimer_thr); - - rcu_gp_t last_compl_gp = 0; - bool ok = true; - - while (ok && wait_for_pending_cbs()) { - assert(CPU->rcu.reclaimer_thr == THREAD); - - exec_completed_cbs(last_compl_gp); - - bool expedite = advance_cbs(); - - ok = wait_for_cur_cbs_gp_end(expedite, &last_compl_gp); - } -} - -/** Waits until there are callbacks waiting to be dispatched. */ -static bool wait_for_pending_cbs(void) -{ - if (!all_cbs_empty()) - return true; - - bool ok = true; - - while (arriving_cbs_empty() && ok) { - ok = semaphore_down_interruptable(&CPU->rcu.arrived_flag); - } - - return ok; -} - -static void upd_stat_missed_gp(rcu_gp_t compl) -{ - if (CPU->rcu.cur_cbs_gp < compl) { - CPU->rcu.stat_missed_gps += (size_t)(compl - CPU->rcu.cur_cbs_gp); - } -} - -/** Executes all callbacks for the given completed grace period. */ -static void exec_completed_cbs(rcu_gp_t last_completed_gp) -{ - upd_stat_missed_gp(last_completed_gp); - - /* Both next_cbs and cur_cbs GP elapsed. */ - if (CPU->rcu.next_cbs_gp <= last_completed_gp) { - assert(CPU->rcu.cur_cbs_gp <= CPU->rcu.next_cbs_gp); - - size_t exec_cnt = CPU->rcu.cur_cbs_cnt + CPU->rcu.next_cbs_cnt; - - if (exec_cnt < CRITICAL_THRESHOLD) { - exec_cbs(&CPU->rcu.cur_cbs); - exec_cbs(&CPU->rcu.next_cbs); - } else { - /* - * Getting overwhelmed with too many callbacks to run. - * Disable preemption in order to prolong our time slice - * and catch up with updaters posting new callbacks. - */ - preemption_disable(); - exec_cbs(&CPU->rcu.cur_cbs); - exec_cbs(&CPU->rcu.next_cbs); - preemption_enable(); - } - - CPU->rcu.cur_cbs_cnt = 0; - CPU->rcu.next_cbs_cnt = 0; - } else if (CPU->rcu.cur_cbs_gp <= last_completed_gp) { - - if (CPU->rcu.cur_cbs_cnt < CRITICAL_THRESHOLD) { - exec_cbs(&CPU->rcu.cur_cbs); - } else { - /* - * Getting overwhelmed with too many callbacks to run. - * Disable preemption in order to prolong our time slice - * and catch up with updaters posting new callbacks. - */ - preemption_disable(); - exec_cbs(&CPU->rcu.cur_cbs); - preemption_enable(); - } - - CPU->rcu.cur_cbs_cnt = 0; - } -} - -/** Executes callbacks in the single-linked list. The list is left empty. */ -static void exec_cbs(rcu_item_t **phead) -{ - rcu_item_t *rcu_item = *phead; - - while (rcu_item) { - /* func() may free rcu_item. Get a local copy. */ - rcu_item_t *next = rcu_item->next; - rcu_func_t func = rcu_item->func; - - func(rcu_item); - - rcu_item = next; - } - - *phead = NULL; -} - -static void upd_stat_cb_cnts(size_t arriving_cnt) -{ - CPU->rcu.stat_max_cbs = max(arriving_cnt, CPU->rcu.stat_max_cbs); - if (0 < arriving_cnt) { - CPU->rcu.stat_avg_cbs = - (99 * CPU->rcu.stat_avg_cbs + 1 * arriving_cnt) / 100; - } -} - -/** Prepares another batch of callbacks to dispatch at the nest grace period. - * - * @return True if the next batch of callbacks must be expedited quickly. - */ -static bool advance_cbs(void) -{ - /* Move next_cbs to cur_cbs. */ - CPU->rcu.cur_cbs = CPU->rcu.next_cbs; - CPU->rcu.cur_cbs_cnt = CPU->rcu.next_cbs_cnt; - CPU->rcu.cur_cbs_gp = CPU->rcu.next_cbs_gp; - - /* Move arriving_cbs to next_cbs. */ - - CPU->rcu.next_cbs_cnt = CPU->rcu.arriving_cbs_cnt; - CPU->rcu.arriving_cbs_cnt = 0; - - /* - * Too many callbacks queued. Better speed up the detection - * or risk exhausting all system memory. - */ - bool expedite = (EXPEDITE_THRESHOLD < CPU->rcu.next_cbs_cnt) || - CPU->rcu.expedite_arriving; - CPU->rcu.expedite_arriving = false; - - /* Start moving the arriving_cbs list to next_cbs. */ - CPU->rcu.next_cbs = CPU->rcu.arriving_cbs; - - /* - * At least one callback arrived. The tail therefore does not point - * to the head of arriving_cbs and we can safely reset it to NULL. - */ - if (CPU->rcu.next_cbs) { - assert(CPU->rcu.parriving_cbs_tail != &CPU->rcu.arriving_cbs); - - CPU->rcu.arriving_cbs = NULL; - /* Reset arriving_cbs before updating the tail pointer. */ - compiler_barrier(); - /* Updating the tail pointer completes the move of arriving_cbs. */ - ACCESS_ONCE(CPU->rcu.parriving_cbs_tail) = &CPU->rcu.arriving_cbs; - } else { - /* - * arriving_cbs was null and parriving_cbs_tail pointed to it - * so leave it that way. Note that interrupt handlers may have - * added a callback in the meantime so it is not safe to reset - * arriving_cbs or parriving_cbs. - */ - } - - /* Update statistics of arrived callbacks. */ - upd_stat_cb_cnts(CPU->rcu.next_cbs_cnt); - - /* - * Make changes prior to queuing next_cbs visible to readers. - * See comment in wait_for_readers(). - */ - memory_barrier(); /* MB A, B */ - - /* At the end of next_cbs_gp, exec next_cbs. Determine what GP that is. */ - - if (!next_cbs_empty()) { - spinlock_lock(&rcu.gp_lock); - - /* Exec next_cbs at the end of the next GP. */ - CPU->rcu.next_cbs_gp = _rcu_cur_gp + 1; - - /* - * There are no callbacks to invoke before next_cbs. Instruct - * wait_for_cur_cbs_gp() to notify us of the nearest GP end. - * That could be sooner than next_cbs_gp (if the current GP - * had not yet completed), so we'll create a shorter batch - * of callbacks next time around. - */ - if (cur_cbs_empty()) { - CPU->rcu.cur_cbs_gp = rcu.completed_gp + 1; - } - - spinlock_unlock(&rcu.gp_lock); - } else { - CPU->rcu.next_cbs_gp = CPU->rcu.cur_cbs_gp; - } - - assert(CPU->rcu.cur_cbs_gp <= CPU->rcu.next_cbs_gp); - - return expedite; -} - -#ifdef RCU_PREEMPT_A - -/** Waits for the grace period associated with callbacks cub_cbs to elapse. - * - * @param expedite Instructs the detector to aggressively speed up grace - * period detection without any delay. - * @param completed_gp Returns the most recent completed grace period - * number. - * @return false if the thread was interrupted and should stop. - */ -static bool wait_for_cur_cbs_gp_end(bool expedite, rcu_gp_t *completed_gp) -{ - spinlock_lock(&rcu.gp_lock); - - assert(CPU->rcu.cur_cbs_gp <= CPU->rcu.next_cbs_gp); - assert(CPU->rcu.cur_cbs_gp <= _rcu_cur_gp + 1); - - while (rcu.completed_gp < CPU->rcu.cur_cbs_gp) { - /* GP has not yet started - start a new one. */ - if (rcu.completed_gp == _rcu_cur_gp) { - start_new_gp(); - spinlock_unlock(&rcu.gp_lock); - - if (!wait_for_readers(expedite)) - return false; - - spinlock_lock(&rcu.gp_lock); - /* Notify any reclaimers this GP had ended. */ - rcu.completed_gp = _rcu_cur_gp; - condvar_broadcast(&rcu.gp_ended); - } else { - /* GP detection is in progress.*/ - - if (expedite) - condvar_signal(&rcu.expedite_now); - - /* Wait for the GP to complete. */ - errno_t ret = _condvar_wait_timeout_spinlock(&rcu.gp_ended, &rcu.gp_lock, - SYNCH_NO_TIMEOUT, SYNCH_FLAGS_INTERRUPTIBLE); - - if (ret == EINTR) { - spinlock_unlock(&rcu.gp_lock); - return false; - } - } - } - - upd_missed_gp_in_wait(rcu.completed_gp); - - *completed_gp = rcu.completed_gp; - spinlock_unlock(&rcu.gp_lock); - - return true; -} - -static bool wait_for_readers(bool expedite) -{ - DEFINE_CPU_MASK(reader_cpus); - - cpu_mask_active(reader_cpus); - rm_quiescent_cpus(reader_cpus); - - while (!cpu_mask_is_none(reader_cpus)) { - /* Give cpus a chance to context switch (a QS) and batch callbacks. */ - if (!gp_sleep(&expedite)) - return false; - - rm_quiescent_cpus(reader_cpus); - sample_cpus(reader_cpus, reader_cpus); - } - - /* Update statistic. */ - if (expedite) { - ++rcu.stat_expedited_cnt; - } - - /* - * All cpus have passed through a QS and see the most recent _rcu_cur_gp. - * As a result newly preempted readers will associate with next_preempted - * and the number of old readers in cur_preempted will monotonically - * decrease. Wait for those old/preexisting readers. - */ - return wait_for_preempt_reader(); -} - -static bool gp_sleep(bool *expedite) -{ - if (*expedite) { - scheduler(); - return true; - } else { - spinlock_lock(&rcu.gp_lock); - - errno_t ret = 0; - ret = _condvar_wait_timeout_spinlock(&rcu.expedite_now, &rcu.gp_lock, - DETECT_SLEEP_MS * 1000, SYNCH_FLAGS_INTERRUPTIBLE); - - /* rcu.expedite_now was signaled. */ - if (ret == EOK) { - *expedite = true; - } - - spinlock_unlock(&rcu.gp_lock); - - return (ret != EINTR); - } -} - -static void sample_local_cpu(void *arg) -{ - assert(interrupts_disabled()); - cpu_mask_t *reader_cpus = (cpu_mask_t *)arg; - - bool locked = RCU_CNT_INC <= CURRENT->rcu_nesting; - /* smp_call machinery makes the most current _rcu_cur_gp visible. */ - bool passed_qs = (CPU->rcu.last_seen_gp == _rcu_cur_gp); - - if (locked && !passed_qs) { - /* - * This cpu has not yet passed a quiescent state during this grace - * period and it is currently in a reader section. We'll have to - * try to sample this cpu again later. - */ - } else { - /* Either not in a reader section or already passed a QS. */ - cpu_mask_reset(reader_cpus, CPU->id); - /* Contain new reader sections and make prior changes visible to them.*/ - memory_barrier(); - CPU->rcu.last_seen_gp = _rcu_cur_gp; - } -} - -/** Called by the scheduler() when switching away from the current thread. */ -void rcu_after_thread_ran(void) -{ - assert(interrupts_disabled()); - - /* - * In order not to worry about NMI seeing rcu_nesting change work - * with a local copy. - */ - size_t nesting_cnt = local_atomic_exchange(&CURRENT->rcu_nesting, 0); - - /* - * Ensures NMIs see .rcu_nesting without the WAS_PREEMPTED mark and - * do not accidentally call rm_preempted_reader() from unlock(). - */ - compiler_barrier(); - - /* Preempted a reader critical section for the first time. */ - if (RCU_CNT_INC <= nesting_cnt && !(nesting_cnt & RCU_WAS_PREEMPTED)) { - nesting_cnt |= RCU_WAS_PREEMPTED; - note_preempted_reader(); - } - - /* Save the thread's nesting count when it is not running. */ - THREAD->rcu.nesting_cnt = nesting_cnt; - - if (CPU->rcu.last_seen_gp != _rcu_cur_gp) { - /* - * Contain any memory accesses of old readers before announcing a QS. - * Also make changes from the previous GP visible to this cpu. - * Moreover it separates writing to last_seen_gp from - * note_preempted_reader(). - */ - memory_barrier(); - /* - * The preempted reader has been noted globally. There are therefore - * no readers running on this cpu so this is a quiescent state. - * - * Reading the multiword _rcu_cur_gp non-atomically is benign. - * At worst, the read value will be different from the actual value. - * As a result, both the detector and this cpu will believe - * this cpu has not yet passed a QS although it really did. - * - * Reloading _rcu_cur_gp is benign, because it cannot change - * until this cpu acknowledges it passed a QS by writing to - * last_seen_gp. Since interrupts are disabled, only this - * code may to so (IPIs won't get through). - */ - CPU->rcu.last_seen_gp = _rcu_cur_gp; - } - - /* - * Forcefully associate the reclaimer with the highest priority - * even if preempted due to its time slice running out. - */ - if (THREAD == CPU->rcu.reclaimer_thr) { - THREAD->priority = -1; - } - - upd_max_cbs_in_slice(CPU->rcu.arriving_cbs_cnt); -} - -/** Called by the scheduler() when switching to a newly scheduled thread. */ -void rcu_before_thread_runs(void) -{ - assert(!rcu_read_locked()); - - /* Load the thread's saved nesting count from before it was preempted. */ - CURRENT->rcu_nesting = THREAD->rcu.nesting_cnt; -} - -/** Called from scheduler() when exiting the current thread. - * - * Preemption or interrupts are disabled and the scheduler() already - * switched away from the current thread, calling rcu_after_thread_ran(). - */ -void rcu_thread_exiting(void) -{ - assert(CURRENT->rcu_nesting == 0); - - /* - * The thread forgot to exit its reader critical section. - * It is a bug, but rather than letting the entire system lock up - * forcefully leave the reader section. The thread is not holding - * any references anyway since it is exiting so it is safe. - */ - if (RCU_CNT_INC <= THREAD->rcu.nesting_cnt) { - /* Emulate _rcu_preempted_unlock() with the proper nesting count. */ - if (THREAD->rcu.nesting_cnt & RCU_WAS_PREEMPTED) { - rm_preempted_reader(); - } - - printf("Bug: thread (id %" PRIu64 " \"%s\") exited while in RCU read" - " section.\n", THREAD->tid, THREAD->name); - } -} - -/** Returns true if in an rcu reader section. */ -bool rcu_read_locked(void) -{ - return RCU_CNT_INC <= CURRENT->rcu_nesting; -} - -/** Invoked when a preempted reader finally exits its reader section. */ -void _rcu_preempted_unlock(void) -{ - assert(0 == CURRENT->rcu_nesting || RCU_WAS_PREEMPTED == CURRENT->rcu_nesting); - - size_t prev = local_atomic_exchange(&CURRENT->rcu_nesting, 0); - if (prev == RCU_WAS_PREEMPTED) { - /* - * NMI handlers are never preempted but may call rm_preempted_reader() - * if a NMI occurred in _rcu_preempted_unlock() of a preempted thread. - * The only other rcu code that may have been interrupted by the NMI - * in _rcu_preempted_unlock() is: an IPI/sample_local_cpu() and - * the initial part of rcu_after_thread_ran(). - * - * rm_preempted_reader() will not deadlock because none of the locks - * it uses are locked in this case. Neither _rcu_preempted_unlock() - * nor sample_local_cpu() nor the initial part of rcu_after_thread_ran() - * acquire any locks. - */ - rm_preempted_reader(); - } -} - -#elif defined(RCU_PREEMPT_PODZIMEK) - -/** Waits for the grace period associated with callbacks cub_cbs to elapse. - * - * @param expedite Instructs the detector to aggressively speed up grace - * period detection without any delay. - * @param completed_gp Returns the most recent completed grace period - * number. - * @return false if the thread was interrupted and should stop. - */ -static bool wait_for_cur_cbs_gp_end(bool expedite, rcu_gp_t *completed_gp) -{ - /* - * Use a possibly outdated version of completed_gp to bypass checking - * with the lock. - * - * Note that loading and storing rcu.completed_gp is not atomic - * (it is 64bit wide). Reading a clobbered value that is less than - * rcu.completed_gp is harmless - we'll recheck with a lock. The - * only way to read a clobbered value that is greater than the actual - * value is if the detector increases the higher-order word first and - * then decreases the lower-order word (or we see stores in that order), - * eg when incrementing from 2^32 - 1 to 2^32. The loaded value - * suddenly jumps by 2^32. It would take hours for such an increase - * to occur so it is safe to discard the value. We allow increases - * of up to half the maximum to generously accommodate for loading an - * outdated lower word. - */ - rcu_gp_t compl_gp = ACCESS_ONCE(rcu.completed_gp); - if (CPU->rcu.cur_cbs_gp <= compl_gp && - compl_gp <= CPU->rcu.cur_cbs_gp + UINT32_MAX_HALF) { - *completed_gp = compl_gp; - return true; - } - - spinlock_lock(&rcu.gp_lock); - - if (CPU->rcu.cur_cbs_gp <= rcu.completed_gp) { - *completed_gp = rcu.completed_gp; - spinlock_unlock(&rcu.gp_lock); - return true; - } - - assert(CPU->rcu.cur_cbs_gp <= CPU->rcu.next_cbs_gp); - assert(_rcu_cur_gp <= CPU->rcu.cur_cbs_gp); - - /* - * Notify the detector of how many GP ends we intend to wait for, so - * it can avoid going to sleep unnecessarily. Optimistically assume - * new callbacks will arrive while we're waiting; hence +1. - */ - size_t remaining_gp_ends = (size_t) (CPU->rcu.next_cbs_gp - _rcu_cur_gp); - req_detection(remaining_gp_ends + (arriving_cbs_empty() ? 0 : 1)); - - /* - * Ask the detector to speed up GP detection if there are too many - * pending callbacks and other reclaimers have not already done so. - */ - if (expedite) { - if (0 == rcu.req_expedited_cnt) - condvar_signal(&rcu.expedite_now); - - /* - * Expedite only cub_cbs. If there really is a surge of callbacks - * the arriving batch will expedite the GP for the huge number - * of callbacks currently in next_cbs - */ - rcu.req_expedited_cnt = 1; - } - - /* Wait for cur_cbs_gp to end. */ - bool interrupted = cv_wait_for_gp(CPU->rcu.cur_cbs_gp); - - *completed_gp = rcu.completed_gp; - spinlock_unlock(&rcu.gp_lock); - - if (!interrupted) - upd_missed_gp_in_wait(*completed_gp); - - return !interrupted; -} - -/** Waits for an announcement of the end of the grace period wait_on_gp. */ -static bool cv_wait_for_gp(rcu_gp_t wait_on_gp) -{ - assert(spinlock_locked(&rcu.gp_lock)); - - bool interrupted = false; - - /* Wait until wait_on_gp ends. */ - while (rcu.completed_gp < wait_on_gp && !interrupted) { - int ret = _condvar_wait_timeout_spinlock(&rcu.gp_ended, &rcu.gp_lock, - SYNCH_NO_TIMEOUT, SYNCH_FLAGS_INTERRUPTIBLE); - interrupted = (ret == EINTR); - } - - return interrupted; -} - -/** Requests the detector to detect at least req_cnt consecutive grace periods.*/ -static void req_detection(size_t req_cnt) -{ - if (rcu.req_gp_end_cnt < req_cnt) { - bool detector_idle = (0 == rcu.req_gp_end_cnt); - rcu.req_gp_end_cnt = req_cnt; - - if (detector_idle) { - assert(_rcu_cur_gp == rcu.completed_gp); - condvar_signal(&rcu.req_gp_changed); - } - } -} - -/** The detector thread detects and notifies reclaimers of grace period ends. */ -static void detector(void *arg) -{ - spinlock_lock(&rcu.gp_lock); - - while (wait_for_detect_req()) { - /* - * Announce new GP started. Readers start lazily acknowledging that - * they passed a QS. - */ - start_new_gp(); - - spinlock_unlock(&rcu.gp_lock); - - if (!wait_for_readers()) - goto unlocked_out; - - spinlock_lock(&rcu.gp_lock); - - /* Notify reclaimers that they may now invoke queued callbacks. */ - end_cur_gp(); - } - - spinlock_unlock(&rcu.gp_lock); - -unlocked_out: - return; -} - -/** Waits for a request from a reclaimer thread to detect a grace period. */ -static bool wait_for_detect_req(void) -{ - assert(spinlock_locked(&rcu.gp_lock)); - - bool interrupted = false; - - while (0 == rcu.req_gp_end_cnt && !interrupted) { - int ret = _condvar_wait_timeout_spinlock(&rcu.req_gp_changed, - &rcu.gp_lock, SYNCH_NO_TIMEOUT, SYNCH_FLAGS_INTERRUPTIBLE); - - interrupted = (ret == EINTR); - } - - return !interrupted; -} - -static void end_cur_gp(void) -{ - assert(spinlock_locked(&rcu.gp_lock)); - - rcu.completed_gp = _rcu_cur_gp; - --rcu.req_gp_end_cnt; - - condvar_broadcast(&rcu.gp_ended); -} - -/** Waits for readers that started before the current GP started to finish. */ -static bool wait_for_readers(void) -{ - DEFINE_CPU_MASK(reading_cpus); - - /* All running cpus have potential readers. */ - cpu_mask_active(reading_cpus); - - /* - * Give readers time to pass through a QS. Also, batch arriving - * callbacks in order to amortize detection overhead. - */ - if (!gp_sleep()) - return false; - - /* Non-intrusively determine which cpus have yet to pass a QS. */ - rm_quiescent_cpus(reading_cpus); - - /* Actively interrupt cpus delaying the current GP and demand a QS. */ - interrupt_delaying_cpus(reading_cpus); - - /* Wait for the interrupted cpus to notify us that they reached a QS. */ - if (!wait_for_delaying_cpus()) - return false; - /* - * All cpus recorded a QS or are still idle. Any new readers will be added - * to next_preempt if preempted, ie the number of readers in cur_preempted - * monotonically descreases. - */ - - /* Wait for the last reader in cur_preempted to notify us it is done. */ - if (!wait_for_preempt_reader()) - return false; - - return true; -} - -/** Sleeps a while if the current grace period is not to be expedited. */ -static bool gp_sleep(void) -{ - spinlock_lock(&rcu.gp_lock); - - int ret = 0; - while (0 == rcu.req_expedited_cnt && 0 == ret) { - /* minor bug: sleeps for the same duration if woken up spuriously. */ - ret = _condvar_wait_timeout_spinlock(&rcu.expedite_now, &rcu.gp_lock, - DETECT_SLEEP_MS * 1000, SYNCH_FLAGS_INTERRUPTIBLE); - } - - if (0 < rcu.req_expedited_cnt) { - --rcu.req_expedited_cnt; - /* Update statistic. */ - ++rcu.stat_expedited_cnt; - } - - spinlock_unlock(&rcu.gp_lock); - - return (ret != EINTR); -} - -/** Actively interrupts and checks the offending cpus for quiescent states. */ -static void interrupt_delaying_cpus(cpu_mask_t *cpu_mask) -{ - atomic_store(&rcu.delaying_cpu_cnt, 0); - - sample_cpus(cpu_mask, NULL); -} - -/** Invoked on a cpu delaying grace period detection. - * - * Induces a quiescent state for the cpu or it instructs remaining - * readers to notify the detector once they finish. - */ -static void sample_local_cpu(void *arg) -{ - assert(interrupts_disabled()); - assert(!CPU->rcu.is_delaying_gp); - - /* Cpu did not pass a quiescent state yet. */ - if (CPU->rcu.last_seen_gp != _rcu_cur_gp) { - /* Interrupted a reader in a reader critical section. */ - if (0 < CPU->rcu.nesting_cnt) { - assert(!CPU->idle); - /* - * Note to notify the detector from rcu_read_unlock(). - * - * ACCESS_ONCE ensures the compiler writes to is_delaying_gp - * only after it determines that we are in a reader CS. - */ - ACCESS_ONCE(CPU->rcu.is_delaying_gp) = true; - CPU->rcu.signal_unlock = true; - - atomic_inc(&rcu.delaying_cpu_cnt); - } else { - /* - * The cpu did not enter any rcu reader sections since - * the start of the current GP. Record a quiescent state. - * - * Or, we interrupted rcu_read_unlock_impl() right before - * it recorded a QS. Record a QS for it. The memory barrier - * contains the reader section's mem accesses before - * updating last_seen_gp. - * - * Or, we interrupted rcu_read_lock() right after it recorded - * a QS for the previous GP but before it got a chance to - * increment its nesting count. The memory barrier again - * stops the CS code from spilling out of the CS. - */ - memory_barrier(); - CPU->rcu.last_seen_gp = _rcu_cur_gp; - } - } else { - /* - * This cpu already acknowledged that it had passed through - * a quiescent state since the start of cur_gp. - */ - } - - /* - * smp_call() makes sure any changes propagate back to the caller. - * In particular, it makes the most current last_seen_gp visible - * to the detector. - */ -} - -/** Waits for cpus delaying the current grace period if there are any. */ -static bool wait_for_delaying_cpus(void) -{ - int delaying_cpu_cnt = atomic_load(&rcu.delaying_cpu_cnt); - - for (int i = 0; i < delaying_cpu_cnt; ++i) { - if (!semaphore_down_interruptable(&rcu.remaining_readers)) - return false; - } - - /* Update statistic. */ - rcu.stat_delayed_cnt += delaying_cpu_cnt; - - return true; -} - -/** Called by the scheduler() when switching away from the current thread. */ -void rcu_after_thread_ran(void) -{ - assert(interrupts_disabled()); - - /* - * Prevent NMI handlers from interfering. The detector will be notified - * in this function if CPU->rcu.is_delaying_gp. The current thread is - * no longer running so there is nothing else to signal to the detector. - */ - CPU->rcu.signal_unlock = false; - /* - * Separates clearing of .signal_unlock from accesses to - * THREAD->rcu.was_preempted and CPU->rcu.nesting_cnt. - */ - compiler_barrier(); - - /* Save the thread's nesting count when it is not running. */ - THREAD->rcu.nesting_cnt = CPU->rcu.nesting_cnt; - - /* Preempted a reader critical section for the first time. */ - if (0 < THREAD->rcu.nesting_cnt && !THREAD->rcu.was_preempted) { - THREAD->rcu.was_preempted = true; - note_preempted_reader(); - } - - /* - * The preempted reader has been noted globally. There are therefore - * no readers running on this cpu so this is a quiescent state. - */ - _rcu_record_qs(); - - /* - * Interrupt handlers might use RCU while idle in scheduler(). - * The preempted reader has been noted globally, so the handlers - * may now start announcing quiescent states. - */ - CPU->rcu.nesting_cnt = 0; - - /* - * This cpu is holding up the current GP. Let the detector know - * it has just passed a quiescent state. - * - * The detector waits separately for preempted readers, so we have - * to notify the detector even if we have just preempted a reader. - */ - if (CPU->rcu.is_delaying_gp) { - CPU->rcu.is_delaying_gp = false; - semaphore_up(&rcu.remaining_readers); - } - - /* - * Forcefully associate the detector with the highest priority - * even if preempted due to its time slice running out. - * - * todo: Replace with strict scheduler priority classes. - */ - if (THREAD == rcu.detector_thr) { - THREAD->priority = -1; - } else if (THREAD == CPU->rcu.reclaimer_thr) { - THREAD->priority = -1; - } - - upd_max_cbs_in_slice(CPU->rcu.arriving_cbs_cnt); -} - -/** Called by the scheduler() when switching to a newly scheduled thread. */ -void rcu_before_thread_runs(void) -{ - assert(PREEMPTION_DISABLED || interrupts_disabled()); - assert(0 == CPU->rcu.nesting_cnt); - - /* Load the thread's saved nesting count from before it was preempted. */ - CPU->rcu.nesting_cnt = THREAD->rcu.nesting_cnt; - - /* - * Ensures NMI see the proper nesting count before .signal_unlock. - * Otherwise the NMI may incorrectly signal that a preempted reader - * exited its reader section. - */ - compiler_barrier(); - - /* - * In the unlikely event that a NMI occurs between the loading of the - * variables and setting signal_unlock, the NMI handler may invoke - * rcu_read_unlock() and clear signal_unlock. In that case we will - * incorrectly overwrite signal_unlock from false to true. This event - * is benign and the next rcu_read_unlock() will at worst - * needlessly invoke _rcu_signal_unlock(). - */ - CPU->rcu.signal_unlock = THREAD->rcu.was_preempted || CPU->rcu.is_delaying_gp; -} - -/** Called from scheduler() when exiting the current thread. - * - * Preemption or interrupts are disabled and the scheduler() already - * switched away from the current thread, calling rcu_after_thread_ran(). - */ -void rcu_thread_exiting(void) -{ - assert(THREAD != NULL); - assert(THREAD->state == Exiting); - assert(PREEMPTION_DISABLED || interrupts_disabled()); - - /* - * The thread forgot to exit its reader critical section. - * It is a bug, but rather than letting the entire system lock up - * forcefully leave the reader section. The thread is not holding - * any references anyway since it is exiting so it is safe. - */ - if (0 < THREAD->rcu.nesting_cnt) { - THREAD->rcu.nesting_cnt = 1; - read_unlock_impl(&THREAD->rcu.nesting_cnt); - - printf("Bug: thread (id %" PRIu64 " \"%s\") exited while in RCU read" - " section.\n", THREAD->tid, THREAD->name); - } -} - -#endif /* RCU_PREEMPT_PODZIMEK */ - -/** Announces the start of a new grace period for preexisting readers to ack. */ -static void start_new_gp(void) -{ - assert(spinlock_locked(&rcu.gp_lock)); - - irq_spinlock_lock(&rcu.preempt_lock, true); - - /* Start a new GP. Announce to readers that a quiescent state is needed. */ - ++_rcu_cur_gp; - - /* - * Readers preempted before the start of this GP (next_preempted) - * are preexisting readers now that a GP started and will hold up - * the current GP until they exit their reader sections. - * - * Preempted readers from the previous GP have finished so - * cur_preempted is empty, but see comment in _rcu_record_qs(). - */ - list_concat(&rcu.cur_preempted, &rcu.next_preempted); - - irq_spinlock_unlock(&rcu.preempt_lock, true); -} - -/** Remove those cpus from the mask that have already passed a quiescent - * state since the start of the current grace period. - */ -static void rm_quiescent_cpus(cpu_mask_t *cpu_mask) -{ - /* - * Ensure the announcement of the start of a new GP (ie up-to-date - * cur_gp) propagates to cpus that are just coming out of idle - * mode before we sample their idle state flag. - * - * Cpus guarantee that after they set CPU->idle = true they will not - * execute any RCU reader sections without first setting idle to - * false and issuing a memory barrier. Therefore, if rm_quiescent_cpus() - * later on sees an idle cpu, but the cpu is just exiting its idle mode, - * the cpu must not have yet executed its memory barrier (otherwise - * it would pair up with this mem barrier and we would see idle == false). - * That memory barrier will pair up with the one below and ensure - * that a reader on the now-non-idle cpu will see the most current - * cur_gp. As a result, such a reader will never attempt to semaphore_up( - * pending_readers) during this GP, which allows the detector to - * ignore that cpu (the detector thinks it is idle). Moreover, any - * changes made by RCU updaters will have propagated to readers - * on the previously idle cpu -- again thanks to issuing a memory - * barrier after returning from idle mode. - * - * idle -> non-idle cpu | detector | reclaimer - * ------------------------------------------------------ - * rcu reader 1 | | rcu_call() - * MB X | | - * idle = true | | rcu_call() - * (no rcu readers allowed ) | | MB A in advance_cbs() - * MB Y | (...) | (...) - * (no rcu readers allowed) | | MB B in advance_cbs() - * idle = false | ++cur_gp | - * (no rcu readers allowed) | MB C | - * MB Z | signal gp_end | - * rcu reader 2 | | exec_cur_cbs() - * - * - * MB Y orders visibility of changes to idle for detector's sake. - * - * MB Z pairs up with MB C. The cpu making a transition from idle - * will see the most current value of cur_gp and will not attempt - * to notify the detector even if preempted during this GP. - * - * MB Z pairs up with MB A from the previous batch. Updaters' changes - * are visible to reader 2 even when the detector thinks the cpu is idle - * but it is not anymore. - * - * MB X pairs up with MB B. Late mem accesses of reader 1 are contained - * and visible before idling and before any callbacks are executed - * by reclaimers. - * - * In summary, the detector does not know of or wait for reader 2, but - * it does not have to since it is a new reader that will not access - * data from previous GPs and will see any changes. - */ - memory_barrier(); /* MB C */ - - cpu_mask_for_each(*cpu_mask, cpu_id) { - /* - * The cpu already checked for and passed through a quiescent - * state since the beginning of this GP. - * - * _rcu_cur_gp is modified by local detector thread only. - * Therefore, it is up-to-date even without a lock. - * - * cpu.last_seen_gp may not be up-to-date. At worst, we will - * unnecessarily sample its last_seen_gp with a smp_call. - */ - bool cpu_acked_gp = (cpus[cpu_id].rcu.last_seen_gp == _rcu_cur_gp); - - /* - * Either the cpu is idle or it is exiting away from idle mode - * and already sees the most current _rcu_cur_gp. See comment - * in wait_for_readers(). - */ - bool cpu_idle = cpus[cpu_id].idle; - - if (cpu_acked_gp || cpu_idle) { - cpu_mask_reset(cpu_mask, cpu_id); - } - } -} - -/** Serially invokes sample_local_cpu(arg) on each cpu of reader_cpus. */ -static void sample_cpus(cpu_mask_t *reader_cpus, void *arg) -{ - cpu_mask_for_each(*reader_cpus, cpu_id) { - smp_call(cpu_id, sample_local_cpu, arg); - - /* Update statistic. */ - if (CPU->id != cpu_id) - ++rcu.stat_smp_call_cnt; - } -} - -static void upd_missed_gp_in_wait(rcu_gp_t completed_gp) -{ - assert(CPU->rcu.cur_cbs_gp <= completed_gp); - - size_t delta = (size_t)(completed_gp - CPU->rcu.cur_cbs_gp); - CPU->rcu.stat_missed_gp_in_wait += delta; -} - -/** Globally note that the current thread was preempted in a reader section. */ -static void note_preempted_reader(void) -{ - irq_spinlock_lock(&rcu.preempt_lock, false); - - if (CPU->rcu.last_seen_gp != _rcu_cur_gp) { - /* The reader started before the GP started - we must wait for it.*/ - list_append(&THREAD->rcu.preempt_link, &rcu.cur_preempted); - } else { - /* - * The reader started after the GP started and this cpu - * already noted a quiescent state. We might block the next GP. - */ - list_append(&THREAD->rcu.preempt_link, &rcu.next_preempted); - } - - irq_spinlock_unlock(&rcu.preempt_lock, false); -} - -/** Remove the current thread from the global list of preempted readers. */ -static void rm_preempted_reader(void) -{ - irq_spinlock_lock(&rcu.preempt_lock, true); - - assert(link_used(&THREAD->rcu.preempt_link)); - - bool prev_empty = list_empty(&rcu.cur_preempted); - list_remove(&THREAD->rcu.preempt_link); - bool now_empty = list_empty(&rcu.cur_preempted); - - /* This was the last reader in cur_preempted. */ - bool last_removed = now_empty && !prev_empty; - - /* - * Preempted readers are blocking the detector and - * this was the last reader blocking the current GP. - */ - if (last_removed && rcu.preempt_blocking_det) { - rcu.preempt_blocking_det = false; - semaphore_up(&rcu.remaining_readers); - } - - irq_spinlock_unlock(&rcu.preempt_lock, true); -} - -/** Waits for any preempted readers blocking this grace period to finish.*/ -static bool wait_for_preempt_reader(void) -{ - irq_spinlock_lock(&rcu.preempt_lock, true); - - bool reader_exists = !list_empty(&rcu.cur_preempted); - rcu.preempt_blocking_det = reader_exists; - - irq_spinlock_unlock(&rcu.preempt_lock, true); - - if (reader_exists) { - /* Update statistic. */ - ++rcu.stat_preempt_blocking_cnt; - - return semaphore_down_interruptable(&rcu.remaining_readers); - } - - return true; -} - -static void upd_max_cbs_in_slice(size_t arriving_cbs_cnt) -{ - rcu_cpu_data_t *cr = &CPU->rcu; - - if (arriving_cbs_cnt > cr->last_arriving_cnt) { - size_t arrived_cnt = arriving_cbs_cnt - cr->last_arriving_cnt; - cr->stat_max_slice_cbs = max(arrived_cnt, cr->stat_max_slice_cbs); - } - - cr->last_arriving_cnt = arriving_cbs_cnt; -} - -/** Prints RCU run-time statistics. */ -void rcu_print_stat(void) -{ - /* - * Don't take locks. Worst case is we get out-dated values. - * CPU local values are updated without any locks, so there - * are no locks to lock in order to get up-to-date values. - */ - -#ifdef RCU_PREEMPT_PODZIMEK - const char *algo = "podzimek-preempt-rcu"; -#elif defined(RCU_PREEMPT_A) - const char *algo = "a-preempt-rcu"; -#endif - - printf("Config: expedite_threshold=%d, critical_threshold=%d," - " detect_sleep=%dms, %s\n", - EXPEDITE_THRESHOLD, CRITICAL_THRESHOLD, DETECT_SLEEP_MS, algo); - printf("Completed GPs: %" PRIu64 "\n", rcu.completed_gp); - printf("Expedited GPs: %zu\n", rcu.stat_expedited_cnt); - printf("Delayed GPs: %zu (cpus w/ still running readers after gp sleep)\n", - rcu.stat_delayed_cnt); - printf("Preempt blocked GPs: %zu (waited for preempted readers; " - "running or not)\n", rcu.stat_preempt_blocking_cnt); - printf("Smp calls: %zu\n", rcu.stat_smp_call_cnt); - - printf("Max arrived callbacks per GP and CPU:\n"); - for (unsigned int i = 0; i < config.cpu_count; ++i) { - printf(" %zu", cpus[i].rcu.stat_max_cbs); - } - - printf("\nAvg arrived callbacks per GP and CPU (nonempty batches only):\n"); - for (unsigned int i = 0; i < config.cpu_count; ++i) { - printf(" %zu", cpus[i].rcu.stat_avg_cbs); - } - - printf("\nMax arrived callbacks per time slice and CPU:\n"); - for (unsigned int i = 0; i < config.cpu_count; ++i) { - printf(" %zu", cpus[i].rcu.stat_max_slice_cbs); - } - - printf("\nMissed GP notifications per CPU:\n"); - for (unsigned int i = 0; i < config.cpu_count; ++i) { - printf(" %zu", cpus[i].rcu.stat_missed_gps); - } - - printf("\nMissed GP notifications per CPU while waking up:\n"); - for (unsigned int i = 0; i < config.cpu_count; ++i) { - printf(" %zu", cpus[i].rcu.stat_missed_gp_in_wait); - } - printf("\n"); -} - -/** @} - */ Index: kernel/generic/src/synch/workqueue.c =================================================================== --- kernel/generic/src/synch/workqueue.c (revision 947ab77e1138e5b43ed8d4e74ae82a491381bfec) +++ (revision ) @@ -1,968 +1,0 @@ -/* - * 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 kernel_generic - * @{ - */ - -/** - * @file - * @brief Work queue/thread pool that automatically adjusts its size - * depending on the current load. Queued work functions may sleep.. - */ - -#include -#include -#include -#include -#include -#include -#include -#include -#include -#include -#include - -#define WORKQ_MAGIC 0xf00c1333U -#define WORK_ITEM_MAGIC 0xfeec1777U - -struct work_queue { - /* - * Protects everything except activate_worker. - * Must be acquired after any thread->locks. - */ - IRQ_SPINLOCK_DECLARE(lock); - - /* Activates a worker if new work arrives or if shutting down the queue. */ - condvar_t activate_worker; - - /* Queue of work_items ready to be dispatched. */ - list_t queue; - - /* List of worker threads. */ - list_t workers; - - /* Number of work items queued. */ - size_t item_cnt; - - /* Indicates the work queue is shutting down. */ - bool stopping; - const char *name; - - /* Total number of created worker threads. */ - size_t cur_worker_cnt; - /* Number of workers waiting for work to arrive. */ - size_t idle_worker_cnt; - /* Number of idle workers signaled that have not yet been woken up. */ - size_t activate_pending; - /* Number of blocked workers sleeping in work func() (ie not idle). */ - size_t blocked_worker_cnt; - - /* Number of pending signal_worker_op() operations. */ - size_t pending_op_cnt; - - link_t nb_link; - -#ifdef CONFIG_DEBUG - /* Magic cookie for integrity checks. Immutable. Accessed without lock. */ - uint32_t cookie; -#endif -}; - -/** Min number of idle workers to keep. */ -static size_t min_worker_cnt; -/** Max total number of workers - be it blocked, idle, or active. */ -static size_t max_worker_cnt; -/** Max number of concurrently running active workers, ie not blocked nor idle. */ -static size_t max_concurrent_workers; -/** Max number of work items per active worker before a new worker is activated.*/ -static const size_t max_items_per_worker = 8; - -/** System wide work queue. */ -static struct work_queue g_work_queue; - -static int booting = true; - -typedef struct { - IRQ_SPINLOCK_DECLARE(lock); - condvar_t req_cv; - thread_t *thread; - list_t work_queues; -} nonblock_adder_t; - -static nonblock_adder_t nonblock_adder; - -/** Typedef a worker thread signaling operation prototype. */ -typedef void (*signal_op_t)(struct work_queue *workq); - -/* Fwd decl. */ -static void workq_preinit(struct work_queue *workq, const char *name); -static bool add_worker(struct work_queue *workq); -static void interrupt_workers(struct work_queue *workq); -static void wait_for_workers(struct work_queue *workq); -static int _workq_enqueue(struct work_queue *workq, work_t *work_item, - work_func_t func, bool can_block); -static void init_work_item(work_t *work_item, work_func_t func); -static signal_op_t signal_worker_logic(struct work_queue *workq, bool can_block); -static void worker_thread(void *arg); -static bool dequeue_work(struct work_queue *workq, work_t **pwork_item); -static bool worker_unnecessary(struct work_queue *workq); -static void cv_wait(struct work_queue *workq); -static void nonblock_init(void); - -#ifdef CONFIG_DEBUG -static bool workq_corrupted(struct work_queue *workq); -static bool work_item_corrupted(work_t *work_item); -#endif - -/** Creates worker thread for the system-wide worker queue. */ -void workq_global_worker_init(void) -{ - /* - * No need for additional synchronization. Stores to word-sized - * variables are atomic and the change will eventually propagate. - * Moreover add_worker() includes the necessary memory barriers - * in spinlock lock/unlock(). - */ - booting = false; - - nonblock_init(); - - if (!add_worker(&g_work_queue)) - panic("Could not create a single global work queue worker!\n"); - -} - -/** Initializes the system wide work queue and support for other work queues. */ -void workq_global_init(void) -{ - /* Keep idle workers on 1/4-th of cpus, but at least 2 threads. */ - min_worker_cnt = max(2, config.cpu_count / 4); - /* Allow max 8 sleeping work items per cpu. */ - max_worker_cnt = max(32, 8 * config.cpu_count); - /* Maximum concurrency without slowing down the system. */ - max_concurrent_workers = max(2, config.cpu_count); - - workq_preinit(&g_work_queue, "kworkq"); -} - -/** Stops the system global work queue and waits for all work items to complete.*/ -void workq_global_stop(void) -{ - workq_stop(&g_work_queue); -} - -/** Creates and initializes a work queue. Returns NULL upon failure. */ -struct work_queue *workq_create(const char *name) -{ - struct work_queue *workq = malloc(sizeof(struct work_queue)); - if (!workq) - return NULL; - - if (workq) { - if (workq_init(workq, name)) { - assert(!workq_corrupted(workq)); - return workq; - } - - free(workq); - } - - return NULL; -} - -/** Frees work queue resources and stops it if it had not been done so already.*/ -void workq_destroy(struct work_queue *workq) -{ - assert(!workq_corrupted(workq)); - - irq_spinlock_lock(&workq->lock, true); - bool stopped = workq->stopping; -#ifdef CONFIG_DEBUG - size_t running_workers = workq->cur_worker_cnt; -#endif - irq_spinlock_unlock(&workq->lock, true); - - if (!stopped) { - workq_stop(workq); - } else { - assert(0 == running_workers); - } - -#ifdef CONFIG_DEBUG - workq->cookie = 0; -#endif - - free(workq); -} - -/** Initializes workq structure without creating any workers. */ -static void workq_preinit(struct work_queue *workq, const char *name) -{ -#ifdef CONFIG_DEBUG - workq->cookie = WORKQ_MAGIC; -#endif - - irq_spinlock_initialize(&workq->lock, name); - condvar_initialize(&workq->activate_worker); - - list_initialize(&workq->queue); - list_initialize(&workq->workers); - - workq->item_cnt = 0; - workq->stopping = false; - workq->name = name; - - workq->cur_worker_cnt = 1; - workq->idle_worker_cnt = 0; - workq->activate_pending = 0; - workq->blocked_worker_cnt = 0; - - workq->pending_op_cnt = 0; - link_initialize(&workq->nb_link); -} - -/** Initializes a work queue. Returns true if successful. - * - * Before destroying a work queue it must be stopped via - * workq_stop(). - */ -bool workq_init(struct work_queue *workq, const char *name) -{ - workq_preinit(workq, name); - return add_worker(workq); -} - -/** Add a new worker thread. Returns false if the thread could not be created. */ -static bool add_worker(struct work_queue *workq) -{ - assert(!workq_corrupted(workq)); - - thread_t *thread = thread_create(worker_thread, workq, TASK, - THREAD_FLAG_NONE, workq->name); - - if (!thread) { - irq_spinlock_lock(&workq->lock, true); - - /* cur_worker_cnt proactively increased in signal_worker_logic() .*/ - assert(0 < workq->cur_worker_cnt); - --workq->cur_worker_cnt; - - irq_spinlock_unlock(&workq->lock, true); - return false; - } - - /* Respect lock ordering. */ - irq_spinlock_lock(&thread->lock, true); - irq_spinlock_lock(&workq->lock, false); - - bool success; - - if (!workq->stopping) { - success = true; - - /* Try to distribute workers among cpus right away. */ - unsigned int cpu_id = (workq->cur_worker_cnt) % config.cpu_active; - - if (!cpus[cpu_id].active) - cpu_id = CPU->id; - - thread->workq = workq; - thread->cpu = &cpus[cpu_id]; - thread->workq_blocked = false; - thread->workq_idling = false; - link_initialize(&thread->workq_link); - - list_append(&thread->workq_link, &workq->workers); - } else { - /* - * Work queue is shutting down - we must not add the worker - * and we cannot destroy it without ready-ing it. Mark it - * interrupted so the worker exits right away without even - * touching workq. - */ - success = false; - - /* cur_worker_cnt proactively increased in signal_worker() .*/ - assert(0 < workq->cur_worker_cnt); - --workq->cur_worker_cnt; - } - - irq_spinlock_unlock(&workq->lock, false); - irq_spinlock_unlock(&thread->lock, true); - - if (!success) { - thread_interrupt(thread); - } - - thread_ready(thread); - - return success; -} - -/** Shuts down the work queue. Waits for all pending work items to complete. - * - * workq_stop() may only be run once. - */ -void workq_stop(struct work_queue *workq) -{ - assert(!workq_corrupted(workq)); - - interrupt_workers(workq); - wait_for_workers(workq); -} - -/** Notifies worker threads the work queue is shutting down. */ -static void interrupt_workers(struct work_queue *workq) -{ - irq_spinlock_lock(&workq->lock, true); - - /* workq_stop() may only be called once. */ - assert(!workq->stopping); - workq->stopping = true; - - /* Respect lock ordering - do not hold workq->lock during broadcast. */ - irq_spinlock_unlock(&workq->lock, true); - - condvar_broadcast(&workq->activate_worker); -} - -/** Waits for all worker threads to exit. */ -static void wait_for_workers(struct work_queue *workq) -{ - assert(!PREEMPTION_DISABLED); - - irq_spinlock_lock(&workq->lock, true); - - list_foreach_safe(workq->workers, cur_worker, next_worker) { - thread_t *worker = list_get_instance(cur_worker, thread_t, workq_link); - list_remove(cur_worker); - - /* Wait without the lock. */ - irq_spinlock_unlock(&workq->lock, true); - - thread_join(worker); - thread_detach(worker); - - irq_spinlock_lock(&workq->lock, true); - } - - assert(list_empty(&workq->workers)); - - /* Wait for deferred add_worker_op(), signal_worker_op() to finish. */ - while (0 < workq->cur_worker_cnt || 0 < workq->pending_op_cnt) { - irq_spinlock_unlock(&workq->lock, true); - - scheduler(); - - irq_spinlock_lock(&workq->lock, true); - } - - irq_spinlock_unlock(&workq->lock, true); -} - -/** Queues a function into the global wait queue without blocking. - * - * See workq_enqueue_noblock() for more details. - */ -bool workq_global_enqueue_noblock(work_t *work_item, work_func_t func) -{ - return workq_enqueue_noblock(&g_work_queue, work_item, func); -} - -/** Queues a function into the global wait queue; may block. - * - * See workq_enqueue() for more details. - */ -bool workq_global_enqueue(work_t *work_item, work_func_t func) -{ - return workq_enqueue(&g_work_queue, work_item, func); -} - -/** Adds a function to be invoked in a separate thread without blocking. - * - * workq_enqueue_noblock() is guaranteed not to block. It is safe - * to invoke from interrupt handlers. - * - * Consider using workq_enqueue() instead if at all possible. Otherwise, - * your work item may have to wait for previously enqueued sleeping - * work items to complete if you are unlucky. - * - * @param workq Work queue where to queue the work item. - * @param work_item Work item bookkeeping structure. Must be valid - * until func() is entered. - * @param func User supplied function to invoke in a worker thread. - * - * @return false if work queue is shutting down; function is not - * queued for further processing. - * @return true Otherwise. func() will be invoked in a separate thread. - */ -bool workq_enqueue_noblock(struct work_queue *workq, work_t *work_item, - work_func_t func) -{ - return _workq_enqueue(workq, work_item, func, false); -} - -/** Adds a function to be invoked in a separate thread; may block. - * - * While the workq_enqueue() is unlikely to block, it may do so if too - * many previous work items blocked sleeping. - * - * @param workq Work queue where to queue the work item. - * @param work_item Work item bookkeeping structure. Must be valid - * until func() is entered. - * @param func User supplied function to invoke in a worker thread. - * - * @return false if work queue is shutting down; function is not - * queued for further processing. - * @return true Otherwise. func() will be invoked in a separate thread. - */ -bool workq_enqueue(struct work_queue *workq, work_t *work_item, work_func_t func) -{ - return _workq_enqueue(workq, work_item, func, true); -} - -/** Adds a work item that will be processed by a separate worker thread. - * - * func() will be invoked in another kernel thread and may block. - * - * Prefer to call _workq_enqueue() with can_block set. Otherwise - * your work item may have to wait for sleeping work items to complete. - * If all worker threads are blocked/sleeping a new worker thread cannot - * be create without can_block set because creating a thread might - * block due to low memory conditions. - * - * @param workq Work queue where to queue the work item. - * @param work_item Work item bookkeeping structure. Must be valid - * until func() is entered. - * @param func User supplied function to invoke in a worker thread. - * @param can_block May adding this work item block? - * - * @return false if work queue is shutting down; function is not - * queued for further processing. - * @return true Otherwise. - */ -static int _workq_enqueue(struct work_queue *workq, work_t *work_item, - work_func_t func, bool can_block) -{ - assert(!workq_corrupted(workq)); - - bool success = true; - signal_op_t signal_op = NULL; - - irq_spinlock_lock(&workq->lock, true); - - if (workq->stopping) { - success = false; - } else { - init_work_item(work_item, func); - list_append(&work_item->queue_link, &workq->queue); - ++workq->item_cnt; - success = true; - - if (!booting) { - signal_op = signal_worker_logic(workq, can_block); - } else { - /* - * During boot there are no workers to signal. Just queue - * the work and let future workers take care of it. - */ - } - } - - irq_spinlock_unlock(&workq->lock, true); - - if (signal_op) { - signal_op(workq); - } - - return success; -} - -/** Prepare an item to be added to the work item queue. */ -static void init_work_item(work_t *work_item, work_func_t func) -{ -#ifdef CONFIG_DEBUG - work_item->cookie = WORK_ITEM_MAGIC; -#endif - - link_initialize(&work_item->queue_link); - work_item->func = func; -} - -/** Returns the number of workers running work func() that are not blocked. */ -static size_t active_workers_now(struct work_queue *workq) -{ - assert(irq_spinlock_locked(&workq->lock)); - - /* Workers blocked are sleeping in the work function (ie not idle). */ - assert(workq->blocked_worker_cnt <= workq->cur_worker_cnt); - /* Idle workers are waiting for more work to arrive in condvar_wait. */ - assert(workq->idle_worker_cnt <= workq->cur_worker_cnt); - - /* Idle + blocked workers == sleeping worker threads. */ - size_t sleeping_workers = workq->blocked_worker_cnt + workq->idle_worker_cnt; - - assert(sleeping_workers <= workq->cur_worker_cnt); - /* Workers pending activation are idle workers not yet given a time slice. */ - assert(workq->activate_pending <= workq->idle_worker_cnt); - - /* - * Workers actively running the work func() this very moment and - * are neither blocked nor idle. Exclude ->activate_pending workers - * since they will run their work func() once they get a time slice - * and are not running it right now. - */ - return workq->cur_worker_cnt - sleeping_workers; -} - -/** - * Returns the number of workers that are running or are about to run work - * func() and that are not blocked. - */ -static size_t active_workers(struct work_queue *workq) -{ - assert(irq_spinlock_locked(&workq->lock)); - - /* - * Workers actively running the work func() and are neither blocked nor - * idle. ->activate_pending workers will run their work func() once they - * get a time slice after waking from a condvar wait, so count them - * as well. - */ - return active_workers_now(workq) + workq->activate_pending; -} - -static void add_worker_noblock_op(struct work_queue *workq) -{ - condvar_signal(&nonblock_adder.req_cv); -} - -static void add_worker_op(struct work_queue *workq) -{ - add_worker(workq); -} - -static void signal_worker_op(struct work_queue *workq) -{ - assert(!workq_corrupted(workq)); - - condvar_signal(&workq->activate_worker); - - irq_spinlock_lock(&workq->lock, true); - assert(0 < workq->pending_op_cnt); - --workq->pending_op_cnt; - irq_spinlock_unlock(&workq->lock, true); -} - -/** Determines how to signal workers if at all. - * - * @param workq Work queue where a new work item was queued. - * @param can_block True if we may block while signaling a worker or creating - * a new worker. - * - * @return Function that will notify workers or NULL if no action is needed. - */ -static signal_op_t signal_worker_logic(struct work_queue *workq, bool can_block) -{ - assert(!workq_corrupted(workq)); - assert(irq_spinlock_locked(&workq->lock)); - - /* Only signal workers if really necessary. */ - signal_op_t signal_op = NULL; - - /* - * Workers actively running the work func() and neither blocked nor idle. - * Including ->activate_pending workers that will run their work func() - * once they get a time slice. - */ - size_t active = active_workers(workq); - /* Max total allowed number of work items queued for active workers. */ - size_t max_load = active * max_items_per_worker; - - /* Active workers are getting overwhelmed - activate another. */ - if (max_load < workq->item_cnt) { - - size_t remaining_idle = - workq->idle_worker_cnt - workq->activate_pending; - - /* Idle workers still exist - activate one. */ - if (remaining_idle > 0) { - /* - * Directly changing idle_worker_cnt here would not allow - * workers to recognize spurious wake-ups. Change - * activate_pending instead. - */ - ++workq->activate_pending; - ++workq->pending_op_cnt; - signal_op = signal_worker_op; - } else { - /* No idle workers remain. Request that a new one be created. */ - bool need_worker = (active < max_concurrent_workers) && - (workq->cur_worker_cnt < max_worker_cnt); - - if (need_worker && can_block) { - signal_op = add_worker_op; - /* - * It may take some time to actually create the worker. - * We don't want to swamp the thread pool with superfluous - * worker creation requests so pretend it was already - * created and proactively increase the worker count. - */ - ++workq->cur_worker_cnt; - } - - /* - * We cannot create a new worker but we need one desperately - * because all workers are blocked in their work functions. - */ - if (need_worker && !can_block && 0 == active) { - assert(0 == workq->idle_worker_cnt); - - irq_spinlock_lock(&nonblock_adder.lock, true); - - if (nonblock_adder.thread && !link_used(&workq->nb_link)) { - signal_op = add_worker_noblock_op; - ++workq->cur_worker_cnt; - list_append(&workq->nb_link, &nonblock_adder.work_queues); - } - - irq_spinlock_unlock(&nonblock_adder.lock, true); - } - } - } else { - /* - * There are enough active/running workers to process the queue. - * No need to signal/activate any new workers. - */ - signal_op = NULL; - } - - return signal_op; -} - -/** Executes queued work items. */ -static void worker_thread(void *arg) -{ - /* - * The thread has been created after the work queue was ordered to stop. - * Do not access the work queue and return immediately. - */ - if (thread_interrupted(THREAD)) { - thread_detach(THREAD); - return; - } - - assert(arg != NULL); - - struct work_queue *workq = arg; - work_t *work_item; - - while (dequeue_work(workq, &work_item)) { - /* Copy the func field so func() can safely free work_item. */ - work_func_t func = work_item->func; - - func(work_item); - } -} - -/** Waits and retrieves a work item. Returns false if the worker should exit. */ -static bool dequeue_work(struct work_queue *workq, work_t **pwork_item) -{ - assert(!workq_corrupted(workq)); - - irq_spinlock_lock(&workq->lock, true); - - /* Check if we should exit if load is low. */ - if (!workq->stopping && worker_unnecessary(workq)) { - /* There are too many workers for this load. Exit. */ - assert(0 < workq->cur_worker_cnt); - --workq->cur_worker_cnt; - list_remove(&THREAD->workq_link); - irq_spinlock_unlock(&workq->lock, true); - - thread_detach(THREAD); - return false; - } - - bool stop = false; - - /* Wait for work to arrive. */ - while (list_empty(&workq->queue) && !workq->stopping) { - cv_wait(workq); - - if (0 < workq->activate_pending) - --workq->activate_pending; - } - - /* Process remaining work even if requested to stop. */ - if (!list_empty(&workq->queue)) { - link_t *work_link = list_first(&workq->queue); - *pwork_item = list_get_instance(work_link, work_t, queue_link); - -#ifdef CONFIG_DEBUG - assert(!work_item_corrupted(*pwork_item)); - (*pwork_item)->cookie = 0; -#endif - list_remove(work_link); - --workq->item_cnt; - - stop = false; - } else { - /* Requested to stop and no more work queued. */ - assert(workq->stopping); - --workq->cur_worker_cnt; - stop = true; - } - - irq_spinlock_unlock(&workq->lock, true); - - return !stop; -} - -/** Returns true if for the given load there are too many workers. */ -static bool worker_unnecessary(struct work_queue *workq) -{ - assert(irq_spinlock_locked(&workq->lock)); - - /* No work is pending. We don't need too many idle threads. */ - if (list_empty(&workq->queue)) { - /* There are too many idle workers. Exit. */ - return (min_worker_cnt <= workq->idle_worker_cnt); - } else { - /* - * There is work but we are swamped with too many active workers - * that were woken up from sleep at around the same time. We - * don't need another worker fighting for cpu time. - */ - size_t active = active_workers_now(workq); - return (max_concurrent_workers < active); - } -} - -/** Waits for a signal to activate_worker. Thread marked idle while waiting. */ -static void cv_wait(struct work_queue *workq) -{ - ++workq->idle_worker_cnt; - THREAD->workq_idling = true; - - /* Ignore lock ordering just here. */ - assert(irq_spinlock_locked(&workq->lock)); - - _condvar_wait_timeout_irq_spinlock(&workq->activate_worker, - &workq->lock, SYNCH_NO_TIMEOUT, SYNCH_FLAGS_NONE); - - assert(!workq_corrupted(workq)); - assert(irq_spinlock_locked(&workq->lock)); - - THREAD->workq_idling = false; - --workq->idle_worker_cnt; -} - -/** Invoked from thread_ready() right before the thread is woken up. */ -void workq_before_thread_is_ready(thread_t *thread) -{ - assert(thread); - assert(irq_spinlock_locked(&thread->lock)); - - /* Worker's work func() is about to wake up from sleeping. */ - if (thread->workq && thread->workq_blocked) { - /* Must be blocked in user work func() and not be waiting for work. */ - assert(!thread->workq_idling); - assert(thread->state == Sleeping); - assert(THREAD != thread); - assert(!workq_corrupted(thread->workq)); - - /* Protected by thread->lock */ - thread->workq_blocked = false; - - irq_spinlock_lock(&thread->workq->lock, true); - --thread->workq->blocked_worker_cnt; - irq_spinlock_unlock(&thread->workq->lock, true); - } -} - -/** Invoked from scheduler() before switching away from a thread. */ -void workq_after_thread_ran(void) -{ - assert(THREAD); - assert(irq_spinlock_locked(&THREAD->lock)); - - /* Worker's work func() is about to sleep/block. */ - if (THREAD->workq && THREAD->state == Sleeping && !THREAD->workq_idling) { - assert(!THREAD->workq_blocked); - assert(!workq_corrupted(THREAD->workq)); - - THREAD->workq_blocked = true; - - irq_spinlock_lock(&THREAD->workq->lock, false); - - ++THREAD->workq->blocked_worker_cnt; - - bool can_block = false; - signal_op_t op = signal_worker_logic(THREAD->workq, can_block); - - irq_spinlock_unlock(&THREAD->workq->lock, false); - - if (op) { - assert(add_worker_noblock_op == op || signal_worker_op == op); - op(THREAD->workq); - } - } -} - -/** Prints stats of the work queue to the kernel console. */ -void workq_print_info(struct work_queue *workq) -{ - irq_spinlock_lock(&workq->lock, true); - - size_t total = workq->cur_worker_cnt; - size_t blocked = workq->blocked_worker_cnt; - size_t idle = workq->idle_worker_cnt; - size_t active = active_workers(workq); - size_t items = workq->item_cnt; - bool stopping = workq->stopping; - bool worker_surplus = worker_unnecessary(workq); - const char *load_str = worker_surplus ? "decreasing" : - (0 < workq->activate_pending) ? "increasing" : "stable"; - - irq_spinlock_unlock(&workq->lock, true); - - printf( - "Configuration: max_worker_cnt=%zu, min_worker_cnt=%zu,\n" - " max_concurrent_workers=%zu, max_items_per_worker=%zu\n" - "Workers: %zu\n" - "Active: %zu (workers currently processing work)\n" - "Blocked: %zu (work functions sleeping/blocked)\n" - "Idle: %zu (idle workers waiting for more work)\n" - "Items: %zu (queued not yet dispatched work)\n" - "Stopping: %d\n" - "Load: %s\n", - max_worker_cnt, min_worker_cnt, - max_concurrent_workers, max_items_per_worker, - total, - active, - blocked, - idle, - items, - stopping, - load_str); -} - -/** Prints stats of the global work queue. */ -void workq_global_print_info(void) -{ - workq_print_info(&g_work_queue); -} - -static bool dequeue_add_req(nonblock_adder_t *info, struct work_queue **pworkq) -{ - bool stop = false; - - irq_spinlock_lock(&info->lock, true); - - while (list_empty(&info->work_queues) && !stop) { - errno_t ret = _condvar_wait_timeout_irq_spinlock(&info->req_cv, - &info->lock, SYNCH_NO_TIMEOUT, SYNCH_FLAGS_INTERRUPTIBLE); - - stop = (ret == EINTR); - } - - if (!stop) { - *pworkq = list_get_instance(list_first(&info->work_queues), - struct work_queue, nb_link); - - assert(!workq_corrupted(*pworkq)); - - list_remove(&(*pworkq)->nb_link); - } - - irq_spinlock_unlock(&info->lock, true); - - return !stop; -} - -static void thr_nonblock_add_worker(void *arg) -{ - nonblock_adder_t *info = arg; - struct work_queue *workq = NULL; - - while (dequeue_add_req(info, &workq)) { - add_worker(workq); - } -} - -static void nonblock_init(void) -{ - irq_spinlock_initialize(&nonblock_adder.lock, "kworkq-nb.lock"); - condvar_initialize(&nonblock_adder.req_cv); - list_initialize(&nonblock_adder.work_queues); - - nonblock_adder.thread = thread_create(thr_nonblock_add_worker, - &nonblock_adder, TASK, THREAD_FLAG_NONE, "kworkq-nb"); - - if (nonblock_adder.thread) { - thread_ready(nonblock_adder.thread); - } else { - /* - * We won't be able to add workers without blocking if all workers - * sleep, but at least boot the system. - */ - printf("Failed to create kworkq-nb. Sleeping work may stall the workq.\n"); - } -} - -#ifdef CONFIG_DEBUG -/** Returns true if the workq is definitely corrupted; false if not sure. - * - * Can be used outside of any locks. - */ -static bool workq_corrupted(struct work_queue *workq) -{ - /* - * Needed to make the most current cookie value set by workq_preinit() - * visible even if we access the workq right after it is created but - * on a different cpu. Otherwise, workq_corrupted() would not work - * outside a lock. - */ - memory_barrier(); - return NULL == workq || workq->cookie != WORKQ_MAGIC; -} - -/** Returns true if the work_item is definitely corrupted; false if not sure. - * - * Must be used with the work queue protecting spinlock locked. - */ -static bool work_item_corrupted(work_t *work_item) -{ - return NULL == work_item || work_item->cookie != WORK_ITEM_MAGIC; -} -#endif - -/** @} - */ Index: kernel/test/cht/cht1.c =================================================================== --- kernel/test/cht/cht1.c (revision 947ab77e1138e5b43ed8d4e74ae82a491381bfec) +++ (revision ) @@ -1,568 +1,0 @@ -/* - * 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. - */ - -#include -#include -#include -#include - -typedef struct val { - /* Place at the top to simplify re-casting. */ - cht_link_t link; - size_t hash; - size_t unique_id; - bool deleted; - bool mark; -} val_t; - -static size_t val_hash(const cht_link_t *item) -{ - val_t *v = member_to_inst(item, val_t, link); - assert(v->hash == (v->unique_id % 10)); - return v->hash; -} - -static size_t val_key_hash(void *key) -{ - return (uintptr_t)key % 10; -} - -static bool val_equal(const cht_link_t *item1, const cht_link_t *item2) -{ - val_t *v1 = member_to_inst(item1, val_t, link); - val_t *v2 = member_to_inst(item2, val_t, link); - return v1->unique_id == v2->unique_id; -} - -static bool val_key_equal(void *key, const cht_link_t *item2) -{ - val_t *v2 = member_to_inst(item2, val_t, link); - return (uintptr_t)key == v2->unique_id; -} - -static void val_rm_callback(cht_link_t *item) -{ - val_t *v = member_to_inst(item, val_t, link); - assert(!v->deleted); - v->deleted = true; - free(v); -} - -static cht_ops_t val_ops = { - .hash = val_hash, - .key_hash = val_key_hash, - .equal = val_equal, - .key_equal = val_key_equal, - .remove_callback = val_rm_callback, -}; - -static void set_val(val_t *v, size_t h, size_t uid) -{ - v->hash = h; - v->unique_id = uid; - v->deleted = false; - v->mark = false; -} - -/*-------------------------------------------------------------------*/ - -static const char *do_sanity_test(cht_t *h) -{ - if (cht_find_lazy(h, (void *)0)) - return "Found lazy in empty table."; - - if (cht_find(h, (void *)0)) - return "Found in empty table."; - - if (cht_remove_key(h, (void *)0)) - return "Removed from empty table."; - - const int val_cnt = 6; - val_t *v[6] = { NULL }; - - for (int i = 0; i < val_cnt; ++i) - v[i] = nfmalloc(sizeof(val_t)); - - size_t key[] = { 1, 1, 1, 11, 12, 13 }; - - /* First three are identical */ - for (int i = 0; i < 3; ++i) - set_val(v[i], 1, key[i]); - - /* Same hash, different key.*/ - set_val(v[3], 1, key[3]); - - /* Different hashes and keys. */ - set_val(v[4], 2, key[4]); - set_val(v[5], 3, key[5]); - - cht_link_t *dup; - - if (!cht_insert_unique(h, &v[0]->link, &dup)) - return "Duplicates in empty"; - - if (cht_insert_unique(h, &v[1]->link, &dup)) - return "Inserted a duplicate"; - - if (dup != &v[0]->link) - return "Returned wrong duplicate"; - - if (!cht_insert_unique(h, &v[3]->link, &dup)) - return "Refused non-equal item but with a hash in table."; - - cht_insert(h, &v[1]->link); - cht_insert(h, &v[2]->link); - - bool ok = true; - ok = ok && cht_insert_unique(h, &v[4]->link, &dup); - ok = ok && cht_insert_unique(h, &v[5]->link, &dup); - - if (!ok) - return "Refused unique ins 4, 5."; - - if (cht_find(h, (void *)0)) - return "Phantom find."; - - cht_link_t *item = cht_find(h, (void *)v[5]->unique_id); - if (!item || item != &v[5]->link) - return "Missing 5."; - - item = cht_find_next(h, &v[5]->link); - if (item) - return "Found nonexisting duplicate 5"; - - item = cht_find(h, (void *)v[3]->unique_id); - if (!item || item != &v[3]->link) - return "Missing 3."; - - item = cht_find_next(h, &v[3]->link); - if (item) - return "Found nonexisting duplicate 3, same hash as others."; - - item = cht_find(h, (void *)v[0]->unique_id); - ((val_t *)item)->mark = true; - - for (int k = 1; k < 3; ++k) { - item = cht_find_next(h, item); - if (!item) - return "Did not find an inserted duplicate"; - - val_t *val = ((val_t *)item); - - if (val->unique_id != v[0]->unique_id) - return "Found item with a different key."; - if (val->mark) - return "Found twice the same node."; - val->mark = true; - } - - for (int i = 0; i < 3; ++i) { - if (!v[i]->mark) - return "Did not find all duplicates"; - - v[i]->mark = false; - } - - if (cht_find_next(h, item)) - return "Found non-existing duplicate."; - - item = cht_find_next(h, cht_find(h, (void *)key[0])); - - ((val_t *)item)->mark = true; - if (!cht_remove_item(h, item)) - return "Failed to remove inserted item"; - - item = cht_find(h, (void *)key[0]); - if (!item || ((val_t *)item)->mark) - return "Did not find proper item."; - - item = cht_find_next(h, item); - if (!item || ((val_t *)item)->mark) - return "Did not find proper duplicate."; - - item = cht_find_next(h, item); - if (item) - return "Found removed duplicate"; - - if (2 != cht_remove_key(h, (void *)key[0])) - return "Failed to remove all duplicates"; - - if (cht_find(h, (void *)key[0])) - return "Found removed key"; - - if (!cht_find(h, (void *)key[3])) - return "Removed incorrect key"; - - for (size_t k = 0; k < sizeof(v) / sizeof(v[0]); ++k) { - cht_remove_key(h, (void *)key[k]); - } - - for (size_t k = 0; k < sizeof(v) / sizeof(v[0]); ++k) { - if (cht_find(h, (void *)key[k])) - return "Found a key in a cleared table"; - } - - return NULL; -} - -static const char *sanity_test(void) -{ - cht_t h; - if (!cht_create_simple(&h, &val_ops)) - return "Could not create the table."; - - rcu_read_lock(); - const char *err = do_sanity_test(&h); - rcu_read_unlock(); - - cht_destroy(&h); - - return err; -} - -/*-------------------------------------------------------------------*/ - -static size_t next_rand(size_t seed) -{ - return (seed * 1103515245 + 12345) & ((1U << 31) - 1); -} - -/*-------------------------------------------------------------------*/ -typedef struct { - cht_link_t link; - size_t key; - bool free; - bool inserted; - bool deleted; -} stress_t; - -typedef struct { - cht_t *h; - int *stop; - stress_t *elem; - size_t elem_cnt; - size_t upd_prob; - size_t wave_cnt; - size_t wave_elems; - size_t id; - bool failed; -} stress_work_t; - -static size_t stress_hash(const cht_link_t *item) -{ - return ((stress_t *)item)->key >> 8; -} -static size_t stress_key_hash(void *key) -{ - return ((size_t)key) >> 8; -} -static bool stress_equal(const cht_link_t *item1, const cht_link_t *item2) -{ - return ((stress_t *)item1)->key == ((stress_t *)item2)->key; -} -static bool stress_key_equal(void *key, const cht_link_t *item) -{ - return ((size_t)key) == ((stress_t *)item)->key; -} -static void stress_rm_callback(cht_link_t *item) -{ - if (((stress_t *)item)->free) - free(item); - else - ((stress_t *)item)->deleted = true; -} - -cht_ops_t stress_ops = { - .hash = stress_hash, - .key_hash = stress_key_hash, - .equal = stress_equal, - .key_equal = stress_key_equal, - .remove_callback = stress_rm_callback -}; - -static void resize_stresser(void *arg) -{ - stress_work_t *work = (stress_work_t *)arg; - - for (size_t k = 0; k < work->wave_cnt; ++k) { - TPRINTF("I{"); - for (size_t i = 0; i < work->wave_elems; ++i) { - stress_t *s = malloc(sizeof(stress_t)); - if (!s) { - TPRINTF("[out-of-mem]\n"); - goto out_of_mem; - } - - s->free = true; - s->key = (i << 8) + work->id; - - cht_insert(work->h, &s->link); - } - TPRINTF("}"); - - thread_sleep(2); - - TPRINTF("R<"); - for (size_t i = 0; i < work->wave_elems; ++i) { - size_t key = (i << 8) + work->id; - - if (1 != cht_remove_key(work->h, (void *)key)) { - TPRINTF("Err: Failed to remove inserted item\n"); - goto failed; - } - } - TPRINTF(">"); - } - - /* Request that others stop. */ - *work->stop = 1; - return; - -failed: - work->failed = true; - -out_of_mem: - /* Request that others stop. */ - *work->stop = 1; - - /* Remove anything we may have inserted. */ - for (size_t i = 0; i < work->wave_elems; ++i) { - size_t key = (i << 8) + work->id; - cht_remove_key(work->h, (void *)key); - } -} - -static void op_stresser(void *arg) -{ - stress_work_t *work = (stress_work_t *)arg; - assert(0 == *work->stop); - - size_t loops = 0; - size_t seed = work->id; - - while (0 == *work->stop && !work->failed) { - seed = next_rand(seed); - bool upd = ((seed % 100) <= work->upd_prob); - seed = next_rand(seed); - size_t elem_idx = seed % work->elem_cnt; - - ++loops; - if (0 == loops % (1024 * 1024)) { - /* Make the most current work->stop visible. */ - read_barrier(); - TPRINTF("*"); - } - - if (upd) { - seed = next_rand(seed); - bool item_op = seed & 1; - - if (work->elem[elem_idx].inserted) { - if (item_op) { - rcu_read_lock(); - cht_remove_item(work->h, &work->elem[elem_idx].link); - rcu_read_unlock(); - } else { - void *key = (void *)work->elem[elem_idx].key; - if (1 != cht_remove_key(work->h, key)) { - TPRINTF("Err: did not rm the key\n"); - work->failed = true; - } - } - work->elem[elem_idx].inserted = false; - } else if (work->elem[elem_idx].deleted) { - work->elem[elem_idx].deleted = false; - - if (item_op) { - rcu_read_lock(); - cht_link_t *dup; - if (!cht_insert_unique(work->h, &work->elem[elem_idx].link, - &dup)) { - TPRINTF("Err: already inserted\n"); - work->failed = true; - } - rcu_read_unlock(); - } else { - cht_insert(work->h, &work->elem[elem_idx].link); - } - - work->elem[elem_idx].inserted = true; - } - } else { - rcu_read_lock(); - cht_link_t *item = - cht_find(work->h, (void *)work->elem[elem_idx].key); - rcu_read_unlock(); - - if (item) { - if (!work->elem[elem_idx].inserted) { - TPRINTF("Err: found but not inserted!"); - work->failed = true; - } - if (item != &work->elem[elem_idx].link) { - TPRINTF("Err: found but incorrect item\n"); - work->failed = true; - } - } else { - if (work->elem[elem_idx].inserted) { - TPRINTF("Err: inserted but not found!"); - work->failed = true; - } - } - } - } - - /* Remove anything we may have inserted. */ - for (size_t i = 0; i < work->elem_cnt; ++i) { - void *key = (void *) work->elem[i].key; - cht_remove_key(work->h, key); - } -} - -static bool do_stress(void) -{ - cht_t h; - - if (!cht_create_simple(&h, &stress_ops)) { - TPRINTF("Failed to create the table\n"); - return false; - } - - const size_t wave_cnt = 10; - const size_t max_thread_cnt = 8; - const size_t resize_thread_cnt = 2; - size_t op_thread_cnt = min(max_thread_cnt, 2 * config.cpu_active); - size_t total_thr_cnt = op_thread_cnt + resize_thread_cnt; - size_t items_per_thread = 1024; - - size_t work_cnt = op_thread_cnt + resize_thread_cnt; - size_t item_cnt = op_thread_cnt * items_per_thread; - - /* Alloc hash table items. */ - size_t size = item_cnt * sizeof(stress_t) + work_cnt * sizeof(stress_work_t) + - sizeof(int); - - TPRINTF("Alloc and init table items. \n"); - void *p = malloc(size); - if (!p) { - TPRINTF("Failed to alloc items\n"); - cht_destroy(&h); - return false; - } - - stress_t *pitem = p + work_cnt * sizeof(stress_work_t); - stress_work_t *pwork = p; - int *pstop = (int *)(pitem + item_cnt); - - *pstop = 0; - - /* Init work items. */ - for (size_t i = 0; i < op_thread_cnt; ++i) { - pwork[i].h = &h; - pwork[i].stop = pstop; - pwork[i].elem = &pitem[i * items_per_thread]; - pwork[i].upd_prob = (i + 1) * 100 / op_thread_cnt; - pwork[i].id = i; - pwork[i].elem_cnt = items_per_thread; - pwork[i].failed = false; - } - - for (size_t i = op_thread_cnt; i < op_thread_cnt + resize_thread_cnt; ++i) { - pwork[i].h = &h; - pwork[i].stop = pstop; - pwork[i].wave_cnt = wave_cnt; - pwork[i].wave_elems = item_cnt * 4; - pwork[i].id = i; - pwork[i].failed = false; - } - - /* Init table elements. */ - for (size_t k = 0; k < op_thread_cnt; ++k) { - for (size_t i = 0; i < items_per_thread; ++i) { - pwork[k].elem[i].key = (i << 8) + k; - pwork[k].elem[i].free = false; - pwork[k].elem[i].inserted = false; - pwork[k].elem[i].deleted = true; - } - } - - TPRINTF("Running %zu ins/del/find stress threads + %zu resizers.\n", - op_thread_cnt, resize_thread_cnt); - - /* Create and run threads. */ - thread_t *thr[max_thread_cnt + resize_thread_cnt]; - - for (size_t i = 0; i < total_thr_cnt; ++i) { - if (i < op_thread_cnt) - thr[i] = thread_create(op_stresser, &pwork[i], TASK, 0, "cht-op-stress"); - else - thr[i] = thread_create(resize_stresser, &pwork[i], TASK, 0, "cht-resize"); - - assert(thr[i]); - thread_wire(thr[i], &cpus[i % config.cpu_active]); - thread_ready(thr[i]); - } - - bool failed = false; - - /* Wait for all threads to return. */ - TPRINTF("Joining resize stressers.\n"); - for (size_t i = op_thread_cnt; i < total_thr_cnt; ++i) { - thread_join(thr[i]); - thread_detach(thr[i]); - failed = pwork[i].failed || failed; - } - - TPRINTF("Joining op stressers.\n"); - for (int i = (int)op_thread_cnt - 1; i >= 0; --i) { - TPRINTF("%d threads remain\n", i); - thread_join(thr[i]); - thread_detach(thr[i]); - failed = pwork[i].failed || failed; - } - - cht_destroy(&h); - free(p); - - return !failed; -} - -/*-------------------------------------------------------------------*/ - -const char *test_cht1(void) -{ - const char *err = sanity_test(); - if (err) - return err; - printf("Basic sanity test: ok.\n"); - - if (!do_stress()) - return "CHT stress test failed."; - else - return NULL; -} Index: kernel/test/cht/cht1.def =================================================================== --- kernel/test/cht/cht1.def (revision 947ab77e1138e5b43ed8d4e74ae82a491381bfec) +++ (revision ) @@ -1,6 +1,0 @@ -{ - "cht", - "Concurrent hash table test", - &test_cht1, - true -}, Index: kernel/test/synch/rcu1.c =================================================================== --- kernel/test/synch/rcu1.c (revision 947ab77e1138e5b43ed8d4e74ae82a491381bfec) +++ (revision ) @@ -1,1042 +1,0 @@ -/* - * 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. - */ - -#include -#include -#include -#include -#include -#include -#include -#include -#include