trylock for rw_semaphores: 2.4.1
PATCH 1/1 trylock for rw_semaphores: 2.4.1
Here is an x86 implementation of down_read_trylock() and down_write_trylock() for read/write semaphores. As with down_trylock() for exclusive semaphores, they don't block if they fail to get the lock. They just return 1, as opposed to 0 in the success case.
The algorithm should be robust. It should be able to handle any race and clean up properly if a task fails to get the lock, but I would appreciate comments if anyone sees a flaw.
Admittedly, the code path for successfully grabbing the lock is longer than it should be. I should have done the nested if tests in down_*_trylock() with js and jc assembly instructions, rather than using sets and setc to copy the flags to C variables. My excuse is that I'm unfamiliar with assembly programming, so I took the path of least resistance.
I've only tested the code to make sure it compiles and works properly when the lock is already held, but there are no waiters. This is the tricky case where it has to weasel out of holding the bias. The success case and non-bias failure case still need to be tested. Stress testing might also be a good idea. I don't have the time to do this right now, but I thought I'd make the patch available in case anyone else is interested the functionality.
Brian Watson
Compaq Computer
[email protected]
arch/i386/kernel/semaphore.c +33 -0
@@ -382,8 +382,41 @@return sem; } +/* We have the bias, but we can't sleep. We have to get rid of it +* as gracefully as we can. +* +* This routine does have the unfortunate side-effect that we could +* spin for awhile if there's a lot of contention for this lock. If +* that's the case, however, then it's less likely that we would hold +* the bias and be running this code. +*/ +void __up_biased (int val,struct rw_semaphore *sem) +{ +int count, newcount; +repeat: +/* Does it look like we're racing with another contender? */ + count =atomic_read (&sem->count); + newcount = count + val; +if (newcount <0 ) +/* Yes: Try again. */ + goto repeat; +else +/* No: Bump the count while no one's looking. Did we race? */ +if (cmpxchg ((volatile int *)&sem->count, count, newcount) + != count) +/* Yes: Try again. */ + goto repeat; +else +/* No: Let the real waiters duke it out for the bias. +* FIXME: This has the same potential stampede problem +* as down_write_failed_biased(). +*/ +if (atomic_read (&sem->count) >=0 ) +wake_up (&sem->wait); +} + asm( " .align 4 .globl __rwsem_wake
include/asm-i386/atomic.h +13 -0
@@ -52,8 +52,21 @@ :"ir" (i), "m" (v->counter) : "memory");return c; } +#define ATOMIC_SUB_SIGN_BIT 0x1 +#define ATOMIC_SUB_CARRY_BIT 0x2 +static __inline__int atomic_sub_sign_and_carry (int i,atomic_t *v) +{ +unsigned char s, c; + + __asm__ __volatile__( +LOCK "subl %3,%0; sets %1; setc %2" + :"=m" (v->counter),"=qm" (s),"=qm" (c) + :"ir" (i),"m" (v->counter) :"memory" ); +return s | c<<1 ; +} +static __inline__void atomic_inc (atomic_t *v) { __asm__ __volatile__( LOCK "incl %0"
include/asm-i386/semaphore.h +70 -0
76 lines not on this page · raw message
Quoted 4 lines · Brian J. Watson
Here is an x86 implementation of
down_read_trylock()anddown_write_trylock()for read/write semaphores. As withdown_trylock()for exclusive semaphores, they don't block if they fail to get the lock. They just return 1, as opposed to 0 in the success case.
How about the following instead? Warning: compiled, not tested.
-ben
include/asm-i386/semaphore.h +27 -0
@@ -382,5 +382,32 @@__up_write (sem); } +/* returns 1 if it successfully obtained the semaphore for write */ +static inline int down_write_trylock (struct rw_semaphore *sem) +{ +int old =RW_LOCK_BIAS , new =0 ; +int res; + + res =cmpxchg (&sem->count.counter, old, new); +return (res ==RW_LOCK_BIAS ); +} + +/* returns 1 if it successfully obtained the semaphore for read */ +static inline int down_read_trylock (struct rw_semaphore *sem) +{ +int ret =1 ; + asmvolatile ( +LOCK "subl $1,%0 + js 2f + 1: + .section .text.lock,\"ax\" + 2:" LOCK "inc %0 + subl %1,%1 + jmp 1b + .previous" + :"=m" (*(volatile int *)sem), "=r" (ret) : "1" (ret) : "memory"); +return ret; +} +#endif #endif
Review
Quoted 13 lines · Ben LaHaise
How about the following instead? Warning: compiled, not tested.
-ben
(quoted diff) +9
-0+/* returns 1 if it successfully obtained the semaphore for write */ +static inline int down_write_trylock(struct rw_semaphore *sem) +{ + int old = RW_LOCK_BIAS, new = 0; + int res; + + res = cmpxchg(&sem->count.counter, old, new); + return (res == RW_LOCK_BIAS); +}
Excellent! This simplifies things greatly. :)
The reason I returned 0 for success and 1 for failure is because that's the semantic of down_trylock(). IMO they should be consistent.
The only other thing this routine needs is the WAITQUEUE_DEBUG code, at least to keep the readers and writers fields accurate.
Quoted 7 lines
(quoted diff) +7
-0+ +/* returns 1 if it successfully obtained the semaphore for read */ +static inline int down_read_trylock(struct rw_semaphore *sem) +{ + int ret = 1; + asm volatile( + LOCK "subl $1,%0
Review
Quoted 10 lines
(quoted diff) +10
-0+ js 2f + 1: + .section .text.lock,\"ax\" + 2:" LOCK "inc %0 + subl %1,%1 + jmp 1b + .previous" + :"=m" (*(volatile int *)sem), "=r" (ret) : "1" (ret) : "memory"); + return ret; +}
There's a couple of races I can see here:
1) Task A holds the write lock and the count is at zero. Simultaneously, task B calls down_read_trylock() and task C calls down_read(). Task B wins and decrements the count to -1. Not long after, task C decrements it to -2, sees the carry bit is clear, and calls down_read_failed(). Meanwhile, task B bumps the count back up to -1 and returns. When task C calls __up_read(), it bumps the count to zero, sees that the zero flag is set, calls __rwsem_wake(), who sets the write_bias_granted field to 1.
Now task D comes along. It calls down_write(), which decrements the count to -BIAS, sees that the zero bit is clear and the carry bit is set, and calls down_write_failed_biased(). Here it sees that the write_bias_granted field is 1, xchg's it for a 0, and continues on, blissfully unaware that both A and D hold the write lock.
2) Task A holds the write lock, and task B is waiting to get the write lock. The count is at -BIAS. Task C calls down_read_trylock(), who decrements the count to -BIAS-1. At this moment, task A releases the write lock. It bumps the count to -1, sees that the carry bit is clear, and continues along. Now task C bumps the count to 0, and returns. Task B continues sleeping, unaware that the write lock is available. The next task who tries to grab the lock will decrement the count below zero. It'll join task B in the biased code where it'll fall into a never-ending sleep, because no one is going to call __rwsem_wake(). Anyone else who tries to grab the lock will fall into a similar deep, deep sleep.
Adapting from your down_write_trylock() code, I implemented a new down_read_trylock() that avoids these races.
Same disclaimer: compiled, not tested.
-Brian
include/asm-i386/semaphore.h +55 -0
@@ -381,6 +381,61 @@#endif __up_write (sem); } +/* returns 0 if it successfully obtained the semaphore for write */ +static inline int down_write_trylock (struct rw_semaphore *sem) +{ +int old =RW_LOCK_BIAS , new =0 ; + +#if WAITQUEUE_DEBUG +if (sem->__magic != (long )&sem->__magic) +BUG (); +#endif +if (cmpxchg (&sem->count.counter, old, new) ==RW_LOCK_BIAS ) { +#if WAITQUEUE_DEBUG +if (atomic_read (&sem->writers)) +BUG (); +if (atomic_read (&sem->readers)) +BUG (); +if (sem->read_bias_granted) +BUG (); +if (sem->write_bias_granted) +BUG (); +atomic_inc (&sem->writers); +#endif +return 0 ; + } +else +return 1 ; +} + +/* returns 0 if it successfully obtained the semaphore for read */ +static inline int down_read_trylock (struct rw_semaphore *sem) +{ +int old, new; + +#if WAITQUEUE_DEBUG +if (sem->__magic != (long )&sem->__magic) +BUG (); +#endif +repeat: + old =atomic_read (&sem->count); +if (old <=0 ) +return 1 ; + new = old -1 ; +if (cmpxchg (&sem->count.counter, old, new) == old) { +#if WAITQUEUE_DEBUG +if (sem->write_bias_granted) +BUG (); +if (atomic_read (&sem->writers)) +BUG (); +atomic_inc (&sem->readers); +#endif +return 0 ; + } +else + goto repeat; +} +#endif #endif