自己动手实现自旋锁(spinlock)

大多数的并行程序都需要在底层使用锁机制进行同步，简单来讲，锁无非是一套简单的原语，它们保证程序（或进程）对某一资源的互斥访问来维持数据的一致性，如果没有锁机制作为保证，多个线程可能同时访问某一资源，假设没有精心设计的（很复杂）无锁算法保证程序正确执行，那么后果往往非常严重的。无锁算法难于使用，所以一般而言都使用锁来保证程序的一致性。

如果更新某一数据结构的操作比较缓慢，那么互斥的锁是一个比较好的选择，此时如果某一进程或线程被阻塞，操作系统会重新接管控制权，并调度其他进程（或线程）继续执行，原先被阻塞的进程处于睡眠状态。控制权的转换伴随着进程上下文的切换，而这往往是一个昂贵而耗时的操作，所以对于等待锁的时间比较短，那么应该使用其他更高效的方法。

自旋锁（spinlock）

自旋锁（Spinlock）是一种常用的互斥（Mutual Exclusion）同步原语（Synchronization Primitive），试图进入临界区（Critical Section）的线程使用忙等待（Busy Waiting）的方式检测锁的状态，若锁未被持有则尝试获取。与其他锁不同，自旋锁仅仅只是“自旋”，即不停地检查某一锁是否已经被解开，自旋锁是非常快的，所以加锁-解锁操作耗时很短，然而，自旋锁也不是万精油，当因互斥导致进程睡眠的时间很长时，使用自旋锁是不明智的选择。

下面我们考虑实现自己的自旋锁，首先我们需要一些原语，幸好GCC已经为我们提供了一些内置函数，

#define atomic_xadd(P, V) __sync_fetch_and_add((P), (V))
#define cmpxchg(P, O, N) __sync_val_compare_and_swap((P), (O), (N))
#define atomic_inc(P) __sync_add_and_fetch((P), 1)
#define atomic_dec(P) __sync_add_and_fetch((P), -1)
#define atomic_add(P, V) __sync_add_and_fetch((P), (V))
#define atomic_set_bit(P, V) __sync_or_and_fetch((P), 1<<(V))
#define atomic_clear_bit(P, V) __sync_and_and_fetch((P), ~(1<<(V)))

然而，我们也需要自己实现其他的几个原子操作，如下：

/* Compile read-write barrier */
#define barrier() asm volatile("": : :"memory")
/* Pause instruction to prevent excess processor bus usage */
#define cpu_relax() asm volatile("pause\n": : :"memory")
/* Atomic exchange (of various sizes) */
static inline void *xchg_64(void *ptr, void *x)
{
__asm__ __volatile__("xchgq %0,%1"
:"=r" ((unsigned long long) x)
:"m" (*(volatile long long *)ptr), "0" ((unsigned long long) x)
:"memory");
return x;
}
static inline unsigned xchg_32(void *ptr, unsigned x)
{
__asm__ __volatile__("xchgl %0,%1"
:"=r" ((unsigned) x)
:"m" (*(volatile unsigned *)ptr), "0" (x)
:"memory");
return x;
}
static inline unsigned short xchg_16(void *ptr, unsigned short x)
{
__asm__ __volatile__("xchgw %0,%1"
:"=r" ((unsigned short) x)
:"m" (*(volatile unsigned short *)ptr), "0" (x)
:"memory");
return x;
}
/* Test and set a bit */
static inline char atomic_bitsetandtest(void *ptr, int x)
{
char out;
__asm__ __volatile__("lock; bts %2,%1\n"
"sbb %0,%0\n"
:"=r" (out), "=m" (*(volatile long long *)ptr)
:"Ir" (x)
:"memory");
return out;
}

自旋锁可以使用交换原语实现，如下：

#define EBUSY 1
typedef unsigned spinlock;
static void spin_lock(spinlock *lock)
{
while (1)
{
if (!xchg_32(lock, EBUSY)) return;
while (*lock) cpu_relax();
}
}
static void spin_unlock(spinlock *lock)
{
barrier();
*lock = 0;
}
static int spin_trylock(spinlock *lock)
{
return xchg_32(lock, EBUSY);
}

上面的自旋锁已经能够工作，但是也会产生问题，因为多个线程可能产生竞争，因为在锁释放的时候其他的每个线程都想获得锁。这会导致处理器总线的负载增大，从而使性能降低，所以接下来我们将实现另外一种自旋锁，该自旋锁能够感知下一个获得锁的进程或线程，因此能够大大减轻处理器总线负载。

下面我们介绍另外一种自旋锁，MCS自旋锁，该锁使用链表维护申请者的请求序列，

typedef struct mcs_lock_t mcs_lock_t;
struct mcs_lock_t
{
mcs_lock_t *next;
int spin;
};
typedef struct mcs_lock_t *mcs_lock;
static void lock_mcs(mcs_lock *m, mcs_lock_t *me)
{
mcs_lock_t *tail;
me->next = NULL;
me->spin = 0;
tail = xchg_64(m, me);
/* No one there? */
if (!tail) return;
/* Someone there, need to link in */
tail->next = me;
/* Make sure we do the above setting of next. */
barrier();
/* Spin on my spin variable */
while (!me->spin) cpu_relax();
return;
}
static void unlock_mcs(mcs_lock *m, mcs_lock_t *me)
{
/* No successor yet? */
if (!me->next)
{
/* Try to atomically unlock */
if (cmpxchg(m, me, NULL) == me) return;
/* Wait for successor to appear */
while (!me->next) cpu_relax();
}
/* Unlock next one */
me->next->spin = 1;
}
static int trylock_mcs(mcs_lock *m, mcs_lock_t *me)
{
mcs_lock_t *tail;
me->next = NULL;
me->spin = 0;
/* Try to lock */
tail = cmpxchg(m, NULL, &me);
/* No one was there - can quickly return */
if (!tail) return 0;
return EBUSY;
}

当然，MCS锁也是有问题的，因为它的API除了需要传递锁的地址外，还需要传递另外一个结构，下面介绍另外一种自旋锁算法，K42锁算法，

typedef struct k42lock k42lock;
struct k42lock
{
k42lock *next;
k42lock *tail;
};
static void k42_lock(k42lock *l)
{
k42lock me;
k42lock *pred, *succ;
me.next = NULL;
barrier();
pred = xchg_64(&l->tail, &me);
if (pred)
{
me.tail = (void *) 1;
barrier();
pred->next = &me;
barrier();
while (me.tail) cpu_relax();
}
succ = me.next;
if (!succ)
{
barrier();
l->next = NULL;
if (cmpxchg(&l->tail, &me, &l->next) != &me)
{
while (!me.next) cpu_relax();
l->next = me.next;
}
}
else
{
l->next = succ;
}
}
static void k42_unlock(k42lock *l)
{
k42lock *succ = l->next;
barrier();
if (!succ)
{
if (cmpxchg(&l->tail, &l->next, NULL) == (void *) &l->next) return;
while (!l->next) cpu_relax();
succ = l->next;
}
succ->tail = NULL;
}
static int k42_trylock(k42lock *l)
{
if (!cmpxchg(&l->tail, NULL, &l->next)) return 0;
return EBUSY;
}

K42和MCS锁都需要遍历链表才能找到下一个最可能获得锁的进程（或线程），有时查找可能比较费时，所以我们再次改进后：

typedef struct listlock_t listlock_t;
struct listlock_t
{
listlock_t *next;
int spin;
};
typedef struct listlock_t *listlock;
#define LLOCK_FLAG (void *)1
static void listlock_lock(listlock *l)
{
listlock_t me;
listlock_t *tail;
/* Fast path - no users */
if (!cmpxchg(l, NULL, LLOCK_FLAG)) return;
me.next = LLOCK_FLAG;
me.spin = 0;
/* Convert into a wait list */
tail = xchg_64(l, &me);
if (tail)
{
/* Add myself to the list of waiters */
if (tail == LLOCK_FLAG) tail = NULL;
me.next = tail;
/* Wait for being able to go */
while (!me.spin) cpu_relax();
return;
}
/* Try to convert to an exclusive lock */
if (cmpxchg(l, &me, LLOCK_FLAG) == &me) return;
/* Failed - there is now a wait list */
tail = *l;
/* Scan to find who is after me */
while (1)
{
/* Wait for them to enter their next link */
while (tail->next == LLOCK_FLAG) cpu_relax();
if (tail->next == &me)
{
/* Fix their next pointer */
tail->next = NULL;
return;
}
tail = tail->next;
}
}
static void listlock_unlock(listlock *l)
{
listlock_t *tail;
listlock_t *tp;
while (1)
{
tail = *l;
barrier();
/* Fast path */
if (tail == LLOCK_FLAG)
{
if (cmpxchg(l, LLOCK_FLAG, NULL) == LLOCK_FLAG) return;
continue;
}
tp = NULL;
/* Wait for partially added waiter */
while (tail->next == LLOCK_FLAG) cpu_relax();
/* There is a wait list */
if (tail->next) break;
/* Try to convert to a single-waiter lock */
if (cmpxchg(l, tail, LLOCK_FLAG) == tail)
{
/* Unlock */
tail->spin = 1;
return;
}
cpu_relax();
}
/* A long list */
tp = tail;
tail = tail->next;
/* Scan wait list */
while (1)
{
/* Wait for partially added waiter */
while (tail->next == LLOCK_FLAG) cpu_relax();
if (!tail->next) break;
tp = tail;
tail = tail->next;
}
tp->next = NULL;
barrier();
/* Unlock */
tail->spin = 1;
}
static int listlock_trylock(listlock *l)
{
/* Simple part of a spin-lock */
if (!cmpxchg(l, NULL, LLOCK_FLAG)) return 0;
/* Failure! */
return EBUSY;

等等，还可以改进，可以在自旋锁里面嵌套一层自旋锁，

typedef struct bitlistlock_t bitlistlock_t;
struct bitlistlock_t
{
bitlistlock_t *next;
int spin;
};
typedef bitlistlock_t *bitlistlock;
#define BLL_USED ((bitlistlock_t *) -2LL)
static void bitlistlock_lock(bitlistlock *l)
{
bitlistlock_t me;
bitlistlock_t *tail;
/* Grab control of list */
while (atomic_bitsetandtest(l, 0)) cpu_relax();
/* Remove locked bit */
tail = (bitlistlock_t *) ((uintptr_t) *l & ~1LL);
/* Fast path, no waiters */
if (!tail)
{
/* Set to be a flag value */
*l = BLL_USED;
return;
}
if (tail == BLL_USED) tail = NULL;
me.next = tail;
me.spin = 0;
barrier();
/* Unlock, and add myself to the wait list */
*l = &me;
/* Wait for the go-ahead */
while (!me.spin) cpu_relax();
}
static void bitlistlock_unlock(bitlistlock *l)
{
bitlistlock_t *tail;
bitlistlock_t *tp;
/* Fast path - no wait list */
if (cmpxchg(l, BLL_USED, NULL) == BLL_USED) return;
/* Grab control of list */
while (atomic_bitsetandtest(l, 0)) cpu_relax();
tp = *l;
barrier();
/* Get end of list */
tail = (bitlistlock_t *) ((uintptr_t) tp & ~1LL);
/* Actually no users? */
if (tail == BLL_USED)
{
barrier();
*l = NULL;
return;
}
/* Only one entry on wait list? */
if (!tail->next)
{
barrier();
/* Unlock bitlock */
*l = BLL_USED;
barrier();
/* Unlock lock */
tail->spin = 1;
return;
}
barrier();
/* Unlock bitlock */
*l = tail;
barrier();
/* Scan wait list for start */
do
{
tp = tail;
tail = tail->next;
}
while (tail->next);
tp->next = NULL;
barrier();
/* Unlock */
tail->spin = 1;
}
static int bitlistlock_trylock(bitlistlock *l)
{
if (!*l && (cmpxchg(l, NULL, BLL_USED) == NULL)) return 0;
return EBUSY;
}

还可以再次改进，如下

/* Bit-lock for editing the wait block */
#define SLOCK_LOCK 1
#define SLOCK_LOCK_BIT 0
/* Has an active user */
#define SLOCK_USED 2
#define SLOCK_BITS 3
typedef struct slock slock;
struct slock
{
uintptr_t p;
};
typedef struct slock_wb slock_wb;
struct slock_wb
{
/*
* last points to the last wait block in the chain.
* The value is only valid when read from the first wait block.
*/
slock_wb *last;
/* next points to the next wait block in the chain. */
slock_wb *next;
/* Wake up? */
int wake;
};
/* Wait for control of wait block */
static slock_wb *slockwb(slock *s)
{
uintptr_t p;
/* Spin on the wait block bit lock */
while (atomic_bitsetandtest(&s->p, SLOCK_LOCK_BIT))
{
cpu_relax();
}
p = s->p;
if (p <= SLOCK_BITS)
{
/* Oops, looks like the wait block was removed. */
atomic_dec(&s->p);
return NULL;
}
return (slock_wb *)(p - SLOCK_LOCK);
}
static void slock_lock(slock *s)
{
slock_wb swblock;
/* Fastpath - no other readers or writers */
if (!s->p && (cmpxchg(&s->p, 0, SLOCK_USED) == 0)) return;
/* Initialize wait block */
swblock.next = NULL;
swblock.last = &swblock;
swblock.wake = 0;
while (1)
{
uintptr_t p = s->p;
cpu_relax();
/* Fastpath - no other readers or writers */
if (!p)
{
if (cmpxchg(&s->p, 0, SLOCK_USED) == 0) return;
continue;
}
if (p > SLOCK_BITS)
{
slock_wb *first_wb, *last;
first_wb = slockwb(s);
if (!first_wb) continue;
last = first_wb->last;
last->next = &swblock;
first_wb->last = &swblock;
/* Unlock */
barrier();
s->p &= ~SLOCK_LOCK;
break;
}
/* Try to add the first wait block */
if (cmpxchg(&s->p, p, (uintptr_t)&swblock) == p) break;
}
/* Wait to acquire exclusive lock */
while (!swblock.wake) cpu_relax();
}
static void slock_unlock(slock *s)
{
slock_wb *next;
slock_wb *wb;
uintptr_t np;
while (1)
{
uintptr_t p = s->p;
/* This is the fast path, we can simply clear the SRWLOCK_USED bit. */
if (p == SLOCK_USED)
{
if (cmpxchg(&s->p, SLOCK_USED, 0) == SLOCK_USED) return;
continue;
}
/* There's a wait block, we need to wake the next pending user */
wb = slockwb(s);
if (wb) break;
cpu_relax();
}
next = wb->next;
if (next)
{
/*
* There's more blocks chained, we need to update the pointers
* in the next wait block and update the wait block pointer.
*/
np = (uintptr_t) next;
next->last = wb->last;
}
else
{
/* Convert the lock to a simple lock. */
np = SLOCK_USED;
}
barrier();
/* Also unlocks lock bit */
s->p = np;
barrier();
/* Notify the next waiter */
wb->wake = 1;
/* We released the lock */
}
static int slock_trylock(slock *s)
{
/* No other readers or writers? */
if (!s->p && (cmpxchg(&s->p, 0, SLOCK_USED) == 0)) return 0;
return EBUSY;
}

下面是另外一种实现方式，称为stack-lock算法，

typedef struct stlock_t stlock_t;
struct stlock_t
{
stlock_t *next;
};
typedef struct stlock_t *stlock;
static __attribute__((noinline)) void stlock_lock(stlock *l)
{
stlock_t *me = NULL;
barrier();
me = xchg_64(l, &me);
/* Wait until we get the lock */
while (me) cpu_relax();
}
#define MAX_STACK_SIZE (1<<12)
static __attribute__((noinline)) int on_stack(void *p)
{
int x;
uintptr_t u = (uintptr_t) &x;
return ((u - (uintptr_t)p + MAX_STACK_SIZE) 2);
}
static __attribute__((noinline)) void stlock_unlock(stlock *l)
{
stlock_t *tail = *l;
barrier();
/* Fast case */
if (on_stack(tail))
{
/* Try to remove the wait list */
if (cmpxchg(l, tail, NULL) == tail) return;
tail = *l;
}
/* Scan wait list */
while (1)
{
/* Wait for partially added waiter */
while (!tail->next) cpu_relax();
if (on_stack(tail->next)) break;
tail = tail->next;
}
barrier();
/* Unlock */
tail->next = NULL;
}
static int stlock_trylock(stlock *l)
{
stlock_t me;
if (!cmpxchg(l, NULL, &me)) return 0;
return EBUSY;
}

改进后变成，

typedef struct plock_t plock_t;
struct plock_t
{
plock_t *next;
};
typedef struct plock plock;
struct plock
{
plock_t *next;
plock_t *prev;
plock_t *last;
};
static void plock_lock(plock *l)
{
plock_t *me = NULL;
plock_t *prev;
barrier();
me = xchg_64(l, &me);
prev = NULL;
/* Wait until we get the lock */
while (me)
{
/* Scan wait list for my previous */
if (l->next != (plock_t *) &me)
{
plock_t *t = l->next;
while (me)
{
if (t->next == (plock_t *) &me)
{
prev = t;
while (me) cpu_relax();
goto done;
}
if (t->next) t = t->next;
cpu_relax();
}
}
cpu_relax();
}
done:
l->prev = prev;
l->last = (plock_t *) &me;
}
static void plock_unlock(plock *l)
{
plock_t *tail;
/* Do I know my previous? */
if (l->prev)
{
/* Unlock */
l->prev->next = NULL;
return;
}
tail = l->next;
barrier();
/* Fast case */
if (tail == l->last)
{
/* Try to remove the wait list */
if (cmpxchg(&l->next, tail, NULL) == tail) return;
tail = l->next;
}
/* Scan wait list */
while (1)
{
/* Wait for partially added waiter */
while (!tail->next) cpu_relax();
if (tail->next == l->last) break;
tail = tail->next;
}
barrier();
/* Unlock */
tail->next = NULL;
}
static int plock_trylock(plock *l)
{
plock_t me;
if (!cmpxchg(&l->next, NULL, &me))
{
l->last = &me;
return 0;
}
return EBUSY;
}

下面介绍另外一种算法，ticket lock算法，实际上，Linux内核正是采用了该算法，不过考虑到执行效率，人家是以汇编形式写的，

typedef union ticketlock ticketlock;
union ticketlock
{
unsigned u;
struct
{
unsigned short ticket;
unsigned short users;
} s;
};
static void ticket_lock(ticketlock *t)
{
unsigned short me = atomic_xadd(&t->s.users, 1);
while (t->s.ticket != me) cpu_relax();
}
static void ticket_unlock(ticketlock *t)
{
barrier();
t->s.ticket++;
}
static int ticket_trylock(ticketlock *t)
{
unsigned short me = t->s.users;
unsigned short menew = me + 1;
unsigned cmp = ((unsigned) me <<16) + me;
unsigned cmpnew = ((unsigned) menew <<16) + me;
if (cmpxchg(&t->u, cmp, cmpnew) == cmp) return 0;
return EBUSY;
}
static int ticket_lockable(ticketlock *t)
{
ticketlock u = *t;
barrier();
return (u.s.ticket == u.s.users);
}

至此，自旋锁各种不同的实现介绍完毕，亲，你明白了吗？:)

(全文完)