1 // SPDX-License-Identifier: GPL-2.0-or-later
5  * (C) Copyright 2013-2015 Hewlett-Packard Development Company, L.P.
6  * (C) Copyright 2013-2014,2018 Red Hat, Inc.
8  * (C) Copyright 2015 Hewlett-Packard Enterprise Development LP
9  * (C) Copyright 2024-2025 Meta Platforms, Inc. and affiliates.
44  * The basic principle of a queue-based spinlock can best be understood
45  * by studying a classic queue-based spinlock implementation called the
47  * Synchronization on Shared-Memory Multiprocessors by Mellor-Crummey and
58  * unlock the next pending (next->locked), we compress both these: {tail,
59  * next->locked} into a single u32 value.
63  * are at most 4 nesting levels, it can be encoded by a 2-bit number. Now
64  * we can encode the tail by combining the 2-bit nesting level with the cpu
66  * 32-bit word is now needed. Even though we only need 1 bit for the lock,
76  *      atomic operations on smaller 8-bit and 16-bit data types.
94 	if (!(atomic_read_acquire(&lock->val) & (mask)))  in is_lock_released()
103 	int cnt = min(RES_NR_HELD, rqh->cnt);  in check_deadlock_AA()
109 	for (int i = 0; i < cnt - 1; i++) {  in check_deadlock_AA()
110 		if (rqh->locks[i] == lock)  in check_deadlock_AA()
111 			return -EDEADLK;  in check_deadlock_AA()
125 	int rqh_cnt = min(RES_NR_HELD, rqh->cnt);  in check_deadlock_ABBA()
138 		int real_cnt = READ_ONCE(rqh_cpu->cnt);  in check_deadlock_ABBA()
165 		remote_lock = READ_ONCE(rqh_cpu->locks[cnt - 1]);  in check_deadlock_ABBA()
176 		 * attempt to acquire a lock we hold, so for this search only 'cnt - 1'  in check_deadlock_ABBA()
179 		for (int i = 0; i < cnt - 1; i++) {  in check_deadlock_ABBA()
180 			if (READ_ONCE(rqh_cpu->locks[i]) != lock)  in check_deadlock_ABBA()
188 			for (int i = 0; i < rqh_cnt - 1; i++) {  in check_deadlock_ABBA()
189 				if (rqh->locks[i] == remote_lock)  in check_deadlock_ABBA()
190 					return -EDEADLK;  in check_deadlock_ABBA()
220 	u64 prev = ts->cur;  in check_timeout()
222 	if (!ts->timeout_end) {  in check_timeout()
223 		ts->cur = time;  in check_timeout()
224 		ts->timeout_end = time + ts->duration;  in check_timeout()
228 	if (time > ts->timeout_end)  in check_timeout()
229 		return -ETIMEDOUT;  in check_timeout()
236 		ts->cur = time;  in check_timeout()
272  * Provide a test-and-set fallback for cases when queued spin lock support is
290 	val = atomic_read(&lock->val);  in resilient_tas_spin_lock()
292 	if (val || !atomic_try_cmpxchg(&lock->val, &val, 1)) {  in resilient_tas_spin_lock()
309  * Per-CPU queue node structures; we can never have more than 4 nested
312  * Exactly fits one 64-byte cacheline on a 64-bit architecture.
320 #define res_atomic_cond_read_acquire(v, c) res_smp_cond_load_acquire(&(v)->counter, (c))
323  * resilient_queued_spin_lock_slowpath - acquire the queued spinlock
325  * @val: Current value of the queued spinlock 32-bit word
328  * * 0		- Lock was acquired successfully.
329  * * -EDEADLK	- Lock acquisition failed because of AA/ABBA deadlock.
330  * * -ETIMEDOUT - Lock acquisition failed because of timeout.
332  * (queue tail, pending bit, lock value)
336  * uncontended  (0,0,0) -:--> (0,0,1) ------------------------------:--> (*,*,0)
337  *                       :       | ^--------.------.             /  :
339  * pending               :    (0,1,1) +--> (0,1,0)   \           |  :
340  *                       :       | ^--'              |           |  :
342  * uncontended           :    (n,x,y) +--> (n,0,0) --'           |  :
343  *   queue               :       | ^--'                          |  :
345  * contended             :    (*,x,y) +--> (*,0,0) ---> (*,0,1) -'  :
346  *   queue               :         ^--'                             :
363 	 * Wait for in-progress pending->locked hand-overs with a bounded  in resilient_queued_spin_lock_slowpath()
366 	 * 0,1,0 -> 0,0,1  in resilient_queued_spin_lock_slowpath()
370 		val = atomic_cond_read_relaxed(&lock->val,  in resilient_queued_spin_lock_slowpath()
371 					       (VAL != _Q_PENDING_VAL) || !cnt--);  in resilient_queued_spin_lock_slowpath()
375 	 * If we observe any contention; queue.  in resilient_queued_spin_lock_slowpath()
378 		goto queue;  in resilient_queued_spin_lock_slowpath()
383 	 * 0,0,* -> 0,1,* -> 0,0,1 pending, trylock  in resilient_queued_spin_lock_slowpath()
390 	 * Undo and queue; our setting of PENDING might have made the  in resilient_queued_spin_lock_slowpath()
391 	 * n,0,0 -> 0,0,0 transition fail and it will now be waiting  in resilient_queued_spin_lock_slowpath()
400 		goto queue;  in resilient_queued_spin_lock_slowpath()
409 	 * We're pending, wait for the owner to go away.  in resilient_queued_spin_lock_slowpath()
411 	 * 0,1,1 -> *,1,0  in resilient_queued_spin_lock_slowpath()
413 	 * this wait loop must be a load-acquire such that we match the  in resilient_queued_spin_lock_slowpath()
414 	 * store-release that clears the locked bit and create lock  in resilient_queued_spin_lock_slowpath()
421 		res_smp_cond_load_acquire(&lock->locked, !VAL || RES_CHECK_TIMEOUT(ts, ret, _Q_LOCKED_MASK));  in resilient_queued_spin_lock_slowpath()
432 		 * *,1,* -> *,0,*  in resilient_queued_spin_lock_slowpath()
442 	 * 0,1,0 -> 0,0,1  in resilient_queued_spin_lock_slowpath()
452 queue:  in resilient_queued_spin_lock_slowpath()
455 	 * Grab deadlock detection entry for the queue path.  in resilient_queued_spin_lock_slowpath()
460 	idx = node->count++;  in resilient_queued_spin_lock_slowpath()
490 	 * Keep counts of non-zero index values:  in resilient_queued_spin_lock_slowpath()
492 	lockevent_cond_inc(lock_use_node2 + idx - 1, idx);  in resilient_queued_spin_lock_slowpath()
495 	 * Ensure that we increment the head node->count before initialising  in resilient_queued_spin_lock_slowpath()
501 	node->locked = 0;  in resilient_queued_spin_lock_slowpath()
502 	node->next = NULL;  in resilient_queued_spin_lock_slowpath()
505 	 * We touched a (possibly) cold cacheline in the per-cpu queue node;  in resilient_queued_spin_lock_slowpath()
515 	 * @node into the waitqueue via WRITE_ONCE(prev->next, node) below.  in resilient_queued_spin_lock_slowpath()
524 	 * p,*,* -> n,*,*  in resilient_queued_spin_lock_slowpath()
530 	 * if there was a previous node; link it and wait until reaching the  in resilient_queued_spin_lock_slowpath()
539 		WRITE_ONCE(prev->next, node);  in resilient_queued_spin_lock_slowpath()
541 		val = arch_mcs_spin_lock_contended(&node->locked);  in resilient_queued_spin_lock_slowpath()
543 			ret = -EDEADLK;  in resilient_queued_spin_lock_slowpath()
553 		next = READ_ONCE(node->next);  in resilient_queued_spin_lock_slowpath()
559 	 * we're at the head of the waitqueue, wait for the owner & pending to  in resilient_queued_spin_lock_slowpath()
562 	 * *,x,y -> *,0,0  in resilient_queued_spin_lock_slowpath()
564 	 * this wait loop must use a load-acquire such that we match the  in resilient_queued_spin_lock_slowpath()
565 	 * store-release that clears the locked bit and create lock  in resilient_queued_spin_lock_slowpath()
575 	val = res_atomic_cond_read_acquire(&lock->val, !(VAL & _Q_LOCKED_PENDING_MASK) ||  in resilient_queued_spin_lock_slowpath()
586 		 * n,*,* -> 0,*,*  in resilient_queued_spin_lock_slowpath()
590 		 * when we remain the head of wait queue. However, eventually,  in resilient_queued_spin_lock_slowpath()
592 		 * will queue behind us. This will leave the lock owner in  in resilient_queued_spin_lock_slowpath()
596 		 * We terminate the whole wait queue for two reasons. Firstly,  in resilient_queued_spin_lock_slowpath()
597 		 * we eschew per-waiter timeouts with one applied at the head of  in resilient_queued_spin_lock_slowpath()
598 		 * the wait queue.  This allows everyone to break out faster  in resilient_queued_spin_lock_slowpath()
605 			next = smp_cond_load_relaxed(&node->next, VAL);  in resilient_queued_spin_lock_slowpath()
606 			WRITE_ONCE(next->locked, RES_TIMEOUT_VAL);  in resilient_queued_spin_lock_slowpath()
615 	 * n,0,0 -> 0,0,1 : lock, uncontended  in resilient_queued_spin_lock_slowpath()
616 	 * *,*,0 -> *,*,1 : lock, contended  in resilient_queued_spin_lock_slowpath()
618 	 * If the queue head is the only one in the queue (lock value == tail)  in resilient_queued_spin_lock_slowpath()
625 	 *       above wait condition, therefore any concurrent setting of  in resilient_queued_spin_lock_slowpath()
629 		if (atomic_try_cmpxchg_relaxed(&lock->val, &val, _Q_LOCKED_VAL))  in resilient_queued_spin_lock_slowpath()
635 	 * which will then detect the remaining tail and queue behind us  in resilient_queued_spin_lock_slowpath()
641 	 * contended path; wait for next if not observed yet, release.  in resilient_queued_spin_lock_slowpath()
644 		next = smp_cond_load_relaxed(&node->next, (VAL));  in resilient_queued_spin_lock_slowpath()
646 	arch_mcs_spin_unlock_contended(&next->locked);  in resilient_queued_spin_lock_slowpath()