if (!current_thread_info->preempt_count && !irqs_disabled()){
    current_thread_info->preempt_count = PREEMPT_ACTIVE;
    schedule();
    current_thread_info->preempt_count = 0;
}

    其次，要满足什么条件时，其他的内核态任务才可以抢占已运行任务的内核态呢？

    * 没有持有锁（lock）。锁用于保护临界区，不能被抢占。

    * 内核态任务代码可重入（code reentrant）。

    那么，如何判断当前上下文（context）（中断处理例程，系统调用，内核线程等）是没有持有锁的？

    我们在前面已经提及过thread_info中的preempt_count可以通过设置正数来显式地禁用内核抢占。这里，通过控制此变量即可实现持有锁机制，preempt_count初始为0，当加锁时便执行加1操作，当解锁时便执行减1操作，由此可以实现控制内核抢占的目的。

    另外，这里需要补充一些关于可重入函数的知识。

    所谓可重入是指一个可以被多个任务调用的过程，任务在调用时不必担心数据是否会出错。不可重入函数在实时系统设计中被视为不安全函数。

    若一个函数是可重入的，则该函数必须满足以下必要条件： 

    * 不能含有静态（全局）非常量数据。

    * 不能返回静态（全局）非常量数据的地址。 

    * 只能处理由调用者提供的数据。

    作为可重入函数的输入参数，只能由调用者提供，而且所提供的输入数据必须满足下面三点要求。

    * 不能依赖于单实例模式资源的锁。 

    * 不能调用不可重入的函数。 

    * 在函数内部，尽量不能用 malloc 和 free 之类的方法进行内存分配和释放，如果使用，一般情况下会造成该函数的不可重入。 

    可重入函数主要用于多任务环境中。一个可重入的函数简单来说就是可以被中断的函数，也就是说，可以在这个函数执行的任何时刻中断它，转入OS调度下去执行另外一段代码，而返回控制时不会出现什么错误。

    不可重入的函数由于使用了一些系统资源，比如全局变量区，中断向量表等，所以它如果被中断的话，可能会出现问题，这类函数是不能运行在多任务环境下的。 

    可重入函数也可以这样理解，重入即表示重复进入，首先它意味着这个函数可以被中断，其次意味着它除了使用自己栈上的变量以外不依赖于任何环境（包括 static），这样的函数就是purecode（纯代码）可重入，可以允许有该函数的多个副本在运行，由于它们使用的是分离的栈，所以不会互相干扰。

    再则，我们来讨论一下关于内核态需要抢占的触发条件：

    内核提供了一个need_resched标志（这个标志在任务结构thread_info中，其返回的是TIF_NEED_RESCHED）来表明是否需要重新执行调度。当执行调度程序时，内核抢占会根据内核抢占是否禁止来进行内核抢占操作。

    在触发内核抢占及重新调度时，有以下几个重要的函数：

    set_tsk_need_resched()：设置指定进程中的need_resched标志；

    clear_tsk_need_resched()：清除指定进程中的need_resched标志；

    need_resched()：检查need_resched标志的值：如果被设置就返回真，否则返回假。

    那么，何时触发重新调度呢？

    * 时钟中断处理例程检查当前任务的时间片，当任务的时间片消耗完时，scheduler_tick()函数就会设置need_resched标志；

    * 信号量、等待队列等机制唤醒时都是基于等待队列（waitqueue）的，而等待队列的唤醒函数为default_wake_function，其调用try_to_wake_up将被唤醒的任务更改为就绪状态并设置need_resched标志；

    * 设置用户进程的nice值时，可能会使高优先级的任务进入就绪状态；

    * 改变任务的优先级时，可能会使高优先级的任务进入就绪状态；

    * 对CPU（SMP）进行负载均衡时，当前任务可能需要移动至另外一个CPU上运行。

    另外，抢占发生的时机：

    * 当一个中断处理例程退出，在返回到内核态时，此时隐式调用schedule()函数，当前任务没有主动放弃CPU使用权，而是被剥夺了CPU使用权。

    * 当内核代码（程序）从不可抢占状态变为可抢占状态时（preemptible），也就是preempt_count从正数变为0时，此时同样隐式调用schedule()函数。

    * 一个任务在内核态中，显式的调用schedule()函数，任务主动放弃CPU使用权。

    * 一个任务在内核态中被阻塞，导致需要调用schedule()函数，任务主动放弃CPU使用权。

    那些时候不允许内核抢占呢？

    * 内核正在进行中断处理。在Linux内核中不能抢占中断（中断只能被其他中断中止和抢占，进程不能中止和抢占中断，内核抢占是被进程抢占和中止），在中断例程中不允许进行进程调度。进程调度函数schedule()会对此做出判断，如果是在中断中调用，会打印错误。

    * 内核正在进行中断上下文的下半部处理时，硬件中断返回前会执行软中断，此时仍然处于中断上下文中，所以此时无法进行内核抢占。

    * 内核的代码段正持有自旋锁（spinlock）、读写锁（writelock/readlock）时，内核代码段处于锁保护状态。此时，内核不能被抢占，否则由于抢占将导致其他CPU长期不能获得锁而出现死锁状态。

    * 内核正在对每CPU私有的数据结构（Per-CPU data structures）进行操作。在SMP（对称多处理器）中，对于每CPU数据结构并未采用自旋锁进行保护，因为这些数据结构隐含地被保护了（不同的CPU上有不同的每CPU数据，其他CPU上运行的进程不能访问另一个CPU的每CPU数据）。在这种情况下，虽然并未采用锁机制，同样不能进行内核抢占，因为如果允许内核抢占，一个进程被抢占后重新调度，有可能调度到其他的CPU上去，这时定义的每CPU数据变量就会发生错位。因此，对于每CPU数据访问时，同样也无法进行内核抢占。

同步原语

    如何避免由于对共享数据的不安全访问导致的数据崩溃？

    内核使用的各种同步技术：

每CPU变量

a++; 
b=f(a); 
c--;

    由于b=f(a)这条指令依赖于前一条指令a++的执行结果，所以b=f(a)将在“执行”阶段之前被阻塞，直到a++的执行结果被生成出来；而c--跟前面没有依赖，它可能在b=f(a)之前就能执行完。（注意，这里的f(a)并不代表一个以a为参数的函数调用，而是代表以a为操作数的指令。C语言的函数调用是需要若干条指令才能实现的，情况要更复杂些。）

    像这样有依赖关系的指令如果挨得很近，后一条指令必定会因为等待前一条执行的结果，而在流水线中阻塞很久，占用流水线的资源。而编译器的乱序，作为编译优化的一种手段，则试图通过指令重排将这样的两条指令拉开一定的距离，以至于后一条指令进入CPU的时候，前一条指令结果已经得到了，那么也就不再需要阻塞等待了。比如将指令重排为：

a++; 
c--; 
b=f(a);

    相比于CPU的乱序，编译器的乱序才是真正对指令顺序做了调整。但是编译器的乱序也必须保证程序上下文的因果关系不发生改变。

    2、乱序的后果

*addr=5; 
val=*data;

Thread 1:

obj->data = 123; 
obj->ready = 1;

    当设置了data之后，记下标志，然后在另一个线程中可能执行：

Thread 2:

if (obj->ready) 
    do_something(obj->data);

    虽然这个代码看上去有些别扭，但是似乎没错。不过，考虑到乱序，如果标志被置位先于data被设置，那么结果很可能就悲剧了（本来不会执行do_something函数，但是由于乱序导致执行了该函数）。因为从字面上看，前面的那两条指令其实并不存在显式的因果关系，乱序是有可能发生的。

    总的来说，如果程序具有显式的因果关系的话，乱序一定会尊重这些关系；否则，乱序就可能打破程序原有的逻辑。这时候，就需要使用屏障来抑制乱序，以维持程序所期望的逻辑。

    3、优化和内存屏障的作用

    内存屏障主要有：读屏障、写屏障、通用屏障、优化屏障几种。

    以读屏障为例，它用于保证读操作有序。屏障之前的读操作一定会先于屏障之后的读操作完成，写操作不受影响，同属于屏障的某一侧的读操作也不受影响。类似的，写屏障用于限制写操作。而通用屏障则对读写操作都有作用。而优化屏障则用于限制编译器的指令重排，不区分读写。前三种屏障都隐含了优化屏障的功能。比如：

tmp = 2048; 
*addr = 5; 
mb(); 
val = *data;

    有了内存屏障就了确保先设置地址端口，再读数据端口。而至于设置地址端口与tmp的赋值孰先孰后，屏障则不做干预。有了内存屏障，就可以在隐式因果关系的场景中，保证因果关系逻辑正确。

    4、内存屏障原语

    Linux使用六个内存屏障原语。这些原语也被当做优化屏障，因为我们必须保证编译器程序不在屏障前后移动汇编语言指令。

    内存屏障既用于多处理器系统（MP），也用于单处理器系统（UP）。当内存屏障应该防止仅出现在多处理器系统上的竞争条件时，就使用smp_xxx()原语；在单处理器系统上，它们什么也不做。其他的内存屏障原语防止出现在单处理器和多处理器系统上的竞争条件。

    内存屏障原语的实现依赖于系统地体系结构。

    在80x86微处理器上，如果CPU支持lfence汇编语言指令，就把rmb()宏展开为 asm volatile("lfence")，否则就展开为 asm volatile("lock;addl $0, 0(%%esp)")。asm指令告诉编译器插入一些汇编语言指令并起优化屏障的作用。lock;addl $0, 0(%%esp)汇编指令把0加到栈顶的内存单元；这条指令本身没有什么价值，但是，lock前缀使得这条指令成为CPU的一个内存屏障。

    而对于wmb()宏，其实现即为barrier()宏，这是因为Intel处理器不对写内存访问进行重新排序，因此，没有必要在代码中插入一条串行化汇编指令。不过，此宏禁止编译器重新组合指令。

    5、多处理器系统情况

    前面只是考虑了单处理器指令乱序的问题，而在多处理器下，除了每个处理器要独自面对上面讨论的问题之外，当多个处理器之间存在交互的时候，同样要面对乱序的问题。

    一个处理器（记为a）对内存的写操作并不是直接就在内存上生效的，而是要先经过自身的cache。另一个处理器（记为b）如果要读取相应内存上的新值，先得等a的cache同步到内存，然后b的cache再从内存同步这个新值。而如果需要同步的值不止一个的话，就会存在顺序问题。举一个例子：

<CPU-a>:                 |   <CPU-b>:
                         | 
obj->data = 123;         |   if (obj->ready)
wmb();                   |       do_something(obj->data);
obj->ready = 1;          |   

    前面也说过，必须要使用屏障来保证CPU-a不发生乱序，从而使得ready标记置位的时候，data一定是有效的。但是在多处理器情况下，这还不够。原因在于，data和ready标记的新值可能以相反的顺序更新到CPU-b上。

    其实这种情况在大多数体系结构下并不会发生，不过内核文档memory-barriers.txt举了alpha机器的例子。alpha机器可能使用分列的cache结构，每个cache列可以并行工作，以提升效率。而每个cache列上面缓存的数据是互斥的（如果不互斥就还得解决cache列之间的一致性），于是就可能引发cache更新不同步的问题。

    假设cache被分成两列，而CPU-a和CPU-b上的data和ready都分别被缓存在不同的cache列上。

    首先是CPU-a更新了cache之后，会发送消息让其他CPU的cache来同步新的值，对于data和ready的更新消息是需要按顺序发出的。如果cache只有一列，那么指令执行的顺序就决定了操作cache的顺序，也就决定了cache更新消息发出的顺序。但是现在假设了有两个cache列，可 能由于缓存data的cache列比较繁忙而使得data的更新消息晚于ready发出，那么程序逻辑就没法保证了。不过好在SMP下的内存屏障在解决指令乱序问题之外，也将cache更新消息乱序的问题解决了。只要使用了屏障，就能保证屏障之前的cache更新消息先于屏障之后的消息被发出。

    然后就是CPU-b的问题。在使用了屏障之后，CPU-a已经保证data的更新消息先发出了，那么CPU-b也会先收到data的更新消息。不过同样，CPU-b上缓存data的cache列可能比较繁忙，导致对data的更新晚于对ready的更新。这里同样会出问题。

    所以，在这种情况下，CPU-b也得使用屏障。CPU-a上要使用写屏障，保证两个写操作不乱序，并且相应的两个cache更新消息不乱序。CPU-b上则需要使用读屏障，保证对两个cache单元的同步不乱序。可见，SMP下的内存屏障一定是需要配对使用的。

    所以，上面的例子应该改写成： 

<CPU-a>:                 |   <CPU-b>:
                         | 
obj->data = 123;         |   if (obj->ready){
wmb();                   |       rmb();
obj->ready = 1;          |       do_something(obj->data);
                         |   }

    CPU-b上使用的读屏障还有一种弱化版本，它不保证读操作的有序性，叫做数据依赖屏障。顾名思义，它是在具有数据依赖情况下使用的屏障，因为有数据依赖（也就是之前所说的显式的因果关系），所以CPU和编译器已经能够保证指令的顺序。

<CPU-a>:           |  <CPU-b>:
                   |
init(newval);      |  p = data;
<write barrier>    |  <data dependency barrier>
data = &newval;    |  val = *p;

自旋锁

    ?一种广泛应用的同步技术是加锁（locking）。当内核控制路径必须访问共享数据结构或进入临界区时，就需要为自己获取一把“锁”。由于锁机制保护的资源非常类似与限制于房间内的资源，当某人进入房间时，就把门锁上。如果内核控制路径希望访问资源，就试图获取钥匙“打开门”。当且仅当资源空闲时，它才能成功。然后，只要它还想使用这个资源，门就依然锁着。当内核控制路径释放锁时，门就打开，另一个内核控制路径就可以进入房间使用资源。

    下图展现了锁的使用。

    5个内核控制路径(P1, P2, P3, P4和P0)试图访问两个临界区(C1, C2)。内核控制路径P0正在C1中，而P2和P4正等待进入C1。同时，P1正在C2中，而P3正在等待进入C2。注意P0和P1可以并行运行。临界区C3的锁处于打开状态，因为没有内核控制路径需要进入C3。

    自旋锁（spinlock）是用来在多处理器环境中工作的一种特殊的锁。如果内核控制路径（内核态进程）发现自旋锁“开着”，就获取锁并继续自己的执行。相反，如果内核控制路径发现锁由运行在另一个CPU上的内核控制路径“锁着”，就在周围“旋转”，反复执行一条紧凑的循环指令，直到锁被释放。

    自旋锁的循环指令表示“忙等”。即使等待的内核控制路径无事可做（除了浪费时间），它也在CPU上保持运行。不过，自旋锁通常非常方便，因为很多内核资源只锁1毫秒的时间片段，所以说，释放CPU和随后又获得CPU都不会消耗很多时间。

    一般来说，由自旋锁所保护的每个临界区都是禁止内核抢占的。在单处理器系统上，这种锁本身不起锁的作用，自旋锁原语仅仅是禁止或启用内核抢占。请注意，在自旋锁忙等期间，内核抢占还是有效的，因此，等待自旋锁释放的进程有可能被更高优先级的进程所替代。

    下面，进行几个方面对自旋锁进行相关说明：

    1、为什么使用自旋锁？

    操作系统锁机制的基本原理，就是在某个锁操作过程中不能与其他锁操作交织执行，以免多个执行路径对内核中某些重要的数据及数据结构进行同时操作而造成系统混乱。在不同的系统环境中，根据系统特点和操作需要，锁机制可以用多种方式来实现。在Linux中，其系统内核的锁机制一般通过3种基本方式来实现，即原语、关中断和总线锁。在单CPU系统中，CPU 的读—修改—写原语可以保证是原子的，即执行过程过中不会被中断，所以CPU通过关中断的方式，从芯片级保证该操作所存取的数据不能被多个内核控制路径同时访问，避免交叉执行。然而，在对称多处理器 (SMP) 环境中，单CPU涉及读—修改—写原语不再是原子的，因为在某个CPU执行读—修改—写指令时有多次总线操作，其他CPU竞争总线，可导致对同一存储单元的读—写操作与其他CPU对这一存储单元交叉，这时我们就需要用一个称为自旋锁（spin lock）的原始对象为CPU 提供锁定总线的方法。

    2、关于自旋锁的几个事实

    自旋锁实际上是忙等锁，当锁不可用时，CPU一直循环执行“测试并设置（test-and-set）”，直到该锁可用而取得该锁，CPU在等待自旋锁时不做任何有用的工作，仅仅是等待。这说明只有在占用锁的时间极短的情况下，使用自旋锁是合理的，因为此时某个CPU可能正在等待这个自旋锁。当临界区较为短小时，如只是为了保证对数据修改的原子性，常用自旋锁；当临界区很大，或有共享设备的时候，需要较长时间占用锁，使用自旋锁就不是一个很好的选择，会降低CPU的效率。

    自旋锁也存在死锁（deadlock）问题。引发这个问题最常见的情况是要求递归使用一个自旋锁，即如果一个已经拥有某个自旋锁的CPU希望第二次获得这个自旋锁，则该CPU将死锁。自旋锁没有与其关联的“使用计数器”或“所有者标识”；锁或者被占用或者空闲。如果你在锁被占用时获取它，你将等待到该锁被释放。如果碰巧你的CPU已经拥有了该锁，那么用于释放锁的代码将得不到运行，因为你使CPU永远处于“测试并设置”某个内存变量的自旋状态。另外，如果进程获得自旋锁之后再阻塞，也有可能导致死锁的发生。由于自旋锁造成的死锁，会使整个系统挂起，影响非常大。

    自旋锁一定是由系统内核调用的。不可能在用户程序中由用户请求自旋锁。当一个用户进程拥有自旋锁期间，内核是把代码提升到管态的级别上运行。在内部，内核能获取自旋锁，但任何用户都做不到这一点。

    3、Linux 自旋锁

    在Linux中，每个自旋锁都用spinlock_t结构表示，其中包含两个字段：

    slock：该字段表示自旋锁的状态：值为1时，表示未加锁状态，而任何负数和0都表示加锁状态。

    break_lock：表示进程正在忙等自旋锁。

    自旋锁宏：

    具有内核抢占的spin_lock宏

    针对支持SMP系统的抢占式内核，该宏获取自旋锁的地址slp作为其参数，并执行下面的操作：

    a、调用preempt_disable()以禁用内核抢占；

    b、调用函数_raw_spin_trylock()，它对自旋锁的slock字段执行原子性的测试和设置操作。该函数首先执行等价于下面汇编语言片段的一些指令：

movb $0, %a1
xchgb %a1, slp->slock

while (spin_is_locked(slp) && slp->break_lock)
    cpu_relax();

    非抢占式内核中的spin_lock宏

    如果在内核编译时没有选择内核抢占选项，spin_lock宏就与前面描述的spin_lock宏有着很大的区别。在这种情况下，宏生成一个汇编语言程序片段，本质上等价于下面紧凑的忙等待：

1： lock; decb slp->slock (递减slp->slock, 判断其是否为正数)
    jns 3f  (f表示向前的，它在程序后面出现)
2:  pause
    cmpb $0, slp->slock (判断slp->slock是否为0)
    jle 2b  (b表示向后的，前跳回标签2代码)
    jmp lb  (前跳回标签1代码)
3:  

    JNS（jump if not sign），汇编语言中的条件转移指令 。结果为正则转移。
    JLE（或JNG）（jump if less or equal or not greater），汇编语言中的条件转移指令。小于或等于，或者不大于则转移。

    汇编语言指令decb递减自旋锁的值，该指令是原子的，因为它带有lock字节前缀。随后检测符号标志，如果它被清零，说明自旋锁被设置为1（未锁），因此，从标记3处继续正常执行。否则，在标签2处执行紧凑循环直到自旋锁出现正值。然后，从标签1处开始重新执行，因为不检查其他的处理器是否抢占了锁就继续执行是不安全的。

    spin_unlock宏

    spin_unlock宏释放以前获得的自旋锁，它本质上执行了下面的汇编语言指令：

movb $1, slp->slock  (slock赋值为1，标为未锁)

读/写自旋锁

    我们从如下几个点进行讨论：

    1、什么是读写自旋锁？

    自旋锁（Spinlock）是一种常用的互斥（Mutual Exclusion）同步原语（Synchronization Primitive），试图进入临界区（Critical Section）的线程使用忙等待（Busy Waiting）的方式检测锁的状态，若锁未被持有则尝试获取。这种忙等待的做法无谓地消耗了处理器资源，因此只适用于临界区非常短小的代码片段，例如Linux内核的中断处理函数。

    由于互斥的特点，使用自旋锁的代码毫无线程并发性可言，多处理器系统的性能受到限制。通过观察线程（内核控制路径）在临界区的访问行为，我们发现有些线程只是简单地读取信息，并不修改任何东西，那么允许它们同时进入临界区不会有任何危险，反而能大大提高系统的并发性。这种将线程区分为读者和写者、多个读者允许同时访问共享资源、申请线程在等待期内依然使用忙等待方式的锁，我们称之为读写自旋锁（Reader-Writer Spinlock）。

    读写自旋锁同样是在保护SMP体系下的共享数据结构而引入的，它的引入是为了增加内核的并发能力。只要内核控制路径没有对数据结构进行修改，读/写自旋锁就允许多个内核控制路径同时读同一数据结构。如果一个内核控制路径想对这个结构进行写操作，那么它必须首先获取读/写锁的写锁，写锁授权独占访问这个资源。这样设计的目的，即允许对数据结构并发读可以提高系统性能。

    下图显示有两个受读写自旋锁保护的临界区。内核控制路径R0和R1正在同时读取C1中的数据结构，而W0正在等待获取写锁。内核控制路径W1正对C2中的数据进行写操作，而R2和W1分别等待获取读锁和写锁。

    每个读/写自旋锁都是一个rwlock_t结构：

typedef struct {
    raw_rwlock_t raw_lock;
#if defined(CONFIG_PREEMPT) && defined(CONFIG_SMP)
    unsigned int break_lock;
#endif
#ifdef CONFIG_DEBUG_SPINLOCK
    unsigned int magic, owner_cpu;
    void *owner;
#endif
#ifdef CONFIG_DEBUG_LOCK_ALLOC
    struct lockdep_map dep_map;
#endif
} rwlock_t;

typedef struct {
    volatile unsigned int lock;
} raw_rwlock_t;

    2、读写自旋锁的属性

    上面提及的共享资源可以是简单的单一变量或多个变量，也可以是像文件这样的复杂数据结构。为了防止错误地使用读写自旋锁而引发的bug，我们假定每个共享资源关联一把唯一的读写自旋锁，线程只允许按照类似大象装冰箱的方式访问共享资源：

    申请锁 ==>> 获得锁后，读写共享资源 ==>> 释放锁。

    对于线程（内核控制路径）的执行，我们假设：

    a、系统存在一个全局时钟，我们讨论的时间是离散的，不是连续的、数学意义上的时间。

    b、任意时刻，系统中活跃线程的总数目是有限的。

    c、线程的执行不会因为调度、缺页异常等原因无限期地被延迟。理论上，线程的执行可以被系统无限期地延迟，因此任何互斥算法都有死锁的危险。我们希望排除系统的干扰，集中关注算法及具体实现本身。

    d、线程对共享资源的访问在有限步骤内结束。

    e、当线程释放锁时，我们希望：线程在有限步骤内释放锁。

    因为每个程序步骤花费有限时间，所以如果满足上述 5 个条件，那么：获得锁的线程必然在有限时间内将锁释放掉。

    我们说某个读写自旋锁算法是正确的，是指该锁满足如下三个属性：

    a、互斥。任意时刻读者和写者不能同时访问共享资源（即获得锁）；任意时刻只能有至多一个写者访问共享资源。

    b、读者并发。在满足“互斥”的前提下，多个读者可以同时访问共享资源。

    c、无死锁（Freedom from Deadlock）。如果线程A试图获取锁，那么某个线程必将获得锁，这个线程可能是A自己；如果线程A试图但是却永远没有获得锁，那么某个或某些线程必定无限次地获得锁。

    读写自旋锁主要用于比较短小的代码片段，线程等待期间不应该进入睡眠状态，因为睡眠 / 唤醒操作相当耗时，大大延长了获得锁的等待时间，所以我们要求：

    d. 忙等待。申请锁的线程必须不断地查询是否发生退出等待的事件，不能进入睡眠状态。这个要求只是描述线程执行锁申请操作未成功时的行为，并不涉及锁自身的正确性。

    “无死锁”属性告诉我们，从全局来看一定会有申请线程获得锁，但对于某个或某些申请线程而言，它们可能永远无法获得锁，这种现象称为饥饿（Starvation）。一种原因源于计算机体系结构的特点：例如在使用基于单一共享变量的读写自旋锁的多核系统中，如果锁的持有者A所处的处理器和等待者B所处的处理器相邻（也许还能共享二级缓存），B更容易获知锁被释放，增大获得锁的几率，而距离较远的处理器上的线程则难与之PK，导致饥饿的发生。还有一种原因源于设计策略，即读写自旋锁刻意偏好某类角色的线程。

    为了提高并发性，读写自旋锁可以选择偏好读者，即读者能够优先获得锁：

    a、读者优先（Reader Preference）。如果锁被读者持有，那么新来的读者可以立即获得锁，无需忙等待。至于当锁被“写者持有”或“未被持有”时，新来的读者是否可以“阻塞”到正在等待的写者之前，依赖于具体实现。

    如果读者持续不断地到来，等待的写者很可能永远无法获得锁，导致饥饿。在现实中，写者的数目一般较读者少许多，而且到来的频率很低，因此读写自旋锁可以选择偏好写者来有效地缓解饥饿现象：

    b、写者优先（Writer Preference）。写者必须在后到的读者 / 写者之前获得锁。因为在写者之前到来的等待线程数目是有限的，所以可以保证写者的等待时间有个合理的上界。但是多个读者之间获得锁的顺序不确定，且先到的读者不一定能在后到的写者之前获得锁。可见，如果写者持续到来，读者仍然可能产生饥饿。

    为了彻底消除饥饿现象，完美的读写自旋锁还需满足下面任一属性：

    c、无饥饿（Freedom from Starvation）。如果线程A试图获取锁，那么A必定能在有限时间内获得锁。当然，这个“有限时间”也许相当漫长。

    d、公平（Fairness）。我们把“锁申请”操作的执行分为两个阶段：准备阶段（Doorway Section），能在有限程序步骤结束；等待阶段（Waiting Section），也许永远无法结束等待阶段一旦结束，线程即获得读写自旋锁。如果线程A和B同时申请锁，但是A的等待阶段完成于B之前，那么公平读写自旋锁保证A在B之前获得锁。如果A和B的等待阶段在时间上有重叠，那么它们获得锁的顺序是不确定的。

    “公平”意味着申请锁的线程必定在有限时间内获得锁。若不然，假设A申请一个公平读写自旋锁但是永远不能获得，那么在A之后完成准备阶段的线程显然也永远不能获得锁。而在A之前或“重叠”地完成等待阶段的申请线程数目是 有限的，可见必然发生了“死锁”，矛盾。同时这也说明释放锁的时间也是有限的。使用公平读写自旋锁杜绝了饥饿现象的发生，如果假定线程访问共享资源和释放锁的时间有一个合理的上界，那么锁申请线程的等待时间只与前面等待的线程数目有关，不依赖其它因素。

    P.S. 我们也可以自己去进行相关算法的设计与实现，比如说从博弈论和统计学的方向来思考（如利用概率进行读写者优先权分配等）。

    3、以自动机的观点看读写自旋锁

    前面关于读写自旋锁的定义和描述虽然通俗易懂，但是并不精确，很多细节比较含糊。例如，读者和写者这种角色到底是什么含义？“先来”，“后到”，“新来”以及“同时到来”如何界定？申请和释放锁的过程到底是怎样的？

    现在，我们集中精力思考一下读写自旋锁到底是什么东西？读写自旋锁其实就是一个有限状态自动机（Finite State Machine）。自动机模型是一种强大的武器，可以帮助我们精确描述和理解各种算法。在给出严格定义之前，我们先规范一下上节中出现的各种概念：

    a、首先，我们把读写自旋锁看成一个独立的串行系统，线程对锁函数的调用本质上是向其独立地提交操作（Operation）。操作必须是基本的，语义清晰的。所谓“基本”，是指任一种类操作的执行效果都不能由其它一种或多种操作的执行累积而成。

    b、读写自旋锁的函数调用的全过程现在可以建模为：

    线程提交了一个操作，然后等待读写自旋锁在某个时刻选择并执行该操作。我们举个读者申请锁的例子来具体说明。前面提到申请锁分成两个阶段，其中准备阶段我们认为线程向读写自旋锁提交了一个“读者申请”的操作。读者在等待阶段不停地测试锁的最新状态，其实就是在等待读写自旋锁的选择。最终读者在被许可的情况下“原子地”更新锁的状态，从而获得锁，说明读写自旋锁在某个合适的时刻选择并执行了该“读者申请”的操作。一旦某个操作被选中，它将不受干扰地在有限时间内成功完成并且在执行过程中读写自旋锁不能选择其它的操作。读者可能会有些奇怪，直观上锁的释放操作似乎是立即执行，难道也需要“等待”么？为了保证锁状态的一致性（Consistency），某些实现的释放函数使用了忙等待方式（参见本文的第一个实现），亦或由于调度、处理器相对速度等原因，总之锁的释放操作同样有一个不确定的等待执行的延时，因此可以和其它操作统一到相同的执行模型中。在操作成功提交至执行完毕这段时间内，线程不能睡眠。

    c、某个线程对锁的一次使用既可以用读者身份申请，也可以用写者身份申请，但是不能以两种身份同时申请。可见“角色”实质上是线程分别提交了“读者申请”或“写者申请”的操作，而不能提交类似“读者写者同时申请”的操作。

    d、读者 / 写者可以不停地到来 / 离去，这意味着线程能够持续地向读写自旋锁提交各种操作，但是每次只能提交一个。只有当上次提交的操作被执行后，线程才被允许提交新操作。读写自旋锁有能力知道某个操作是哪个线程提交的。

    e、线程对锁的使用必须采用前面提及的规范化流程，这是指线程必须提交配对的“申请”/“释放”操作，即“申请”操作成功执行后，线程应当在有限时间内提交相应的“释放”操作，且在此之前不准提交其它操作。

    f、关于读者 / 写者先来后到的顺序问题，我们转换成确定操作的提交顺序。我们认为操作的提交效果是“瞬间”产生的，即使多个线程在所谓的“同一时刻”提交操作，这些操作彼此之间也有严格的先后顺序，不存在两个或多个操作是“同时”提交成功的。在现实中，提交显然是需要一定时间的，不同线程的提交过程可能在时间上重叠，但是我们认为总可以按照一种策略规定它们的提交顺序，虽然这可能影响锁的实际执行过程，但并不影响正确性；对于同一线程提交的各个操作，它们彼此之间显然有着严格的时序关系，当然能够确定提交顺序。在此，我们彻底取消同时性的概念。

    令 A(t) 为在时间段 (0, t] 内所有提交的操作构成的集合，A(t) 中的任两个操作 o1和 o2，要么 o1在 o2之前提交，要么 o1在 o2之后提交，这种提交顺序是一种全序关系（Total Order）。

    读写自旋锁的形式化定义是一个 6 元组（Q，O，T，S，q0，qf），其中：

    Q = {q0，q1，…，qn}，是一个有限集合，称为状态集。状态 qi描述了读写自旋锁在某时刻t0所处于的一种真实状况。

    O = {o0，o1，…，om}，是一个有限集合，称为操作种类集。

    T：Q x O -> Q 是转移函数。T 是一个偏函数（Partial Function），即 T 的定义域是 Q x O 的子集。如果 T 在 (q, o) 有定义，即存在 q’ = T(q, o)，我们称状态 q 允许操作 o，在状态 q 可以执行操作 o，成功完成后读写自旋锁转换到状态 q ’；反之，如果 T 在 (q, o) 没有定义，我们称状态 q 不允许操作 o，说明在状态 q 不能执行操作 o，例如在锁被写者持有时，不能选择 “读者申请获取锁”的操作。

    S 是选择函数，从已提交但未执行的操作实例集合中选择一个让读写自旋锁执行，后文详细讨论。由于任意时刻活跃线程的总数目是有限的，这个集合必然是有限集，因此我们认为 S 的每一次选择能在有限步骤内结束。

    q0是初始状态。

    qf是结束状态，对于任一种操作 o，T 在 (qf, o) 无定义。也就是说到达该状态后，读写自旋锁不再执行任何操作。

    我们先画出与定义等价的状态图，然后描述 6 元组具体是什么。

    a、状态图中的每个圆圈代表一个状态。状态集合Q至少应该有3个状态：“未被持有”，“读者持有”和“写者持有”。因为可能执行“析构”操作，所以还需要增加一个结束状态“停止”。除此之外不需要新的状态。

    b、有向边上的文字代表了一种操作。读写自旋锁需要 6 种操作： “初始化”、“析构”、“读者申请”、 “读者释放”、 “写者申请”和“写者释放”。操作后面括号内的文字，例如“最后持有者”，只是辅助理解，并不表示一种新的操作。

    c、有向边及其上的操作定义了转移函数。如果一条有向边从状态q指向q’，且标注的操作是 o，那么表明状态q允许操作o，且 q’ = T(q, o)。

    d、初始状态是“未被持有”。

    e、结束状态是“停止”，双圆圈表示，该状态不射出任何有向边，表明此后锁停止执行任何操作。

    结合状态图，我们描述读写自旋锁的工作原理：

    a、我们规定在时刻 0 执行全局唯一一次的“初始化”操作，将锁置为初始状态“未被持有”，图中即为那条没有起点、标注“初始化“操作的有向边。如果决定停止使用读写自旋锁，则执行全局唯一一次的“析构”操作，将锁置为结束状态“停止”。

    b、读写自旋锁可以被看成一个从初始状态“未被持有”开始依次“吃”操作、不断转换状态的串行机器。令 W(t) 为时间段 (0, t] 内已提交但未执行的操作构成的集合，W(t) 是所有提交的操作集合 A(t) 的子集。在时刻t，如果锁准备执行新的操作，假设当前处于状态q，W(t)不是空集且存在状态q允许的操作，那么读写自旋锁使用选择函数S在W(t)集合中选出一个来执行，执行完成后将自身状态置为 q’ = T(q, o)。

    c、我们称序列 < qI1，oI1，qI2，oI2，…，oIn，qI(n+1)> 是读写自旋锁在 t 时刻的执行序列，如果：

        ①. oIk是操作，1 <= k <= (n + 1) 且 oI1，oI2，…，oIn属于集合 A(t)。

        ②. qIk是状态，1 <= k <= (n + 1)。

        ③. 读写自旋锁在 t 时刻的状态是 qI(n+1)。

        ④. qI1= q0。

        ⑤. T 在 (qIk, oIk) 有定义，且 qI(k+1)= T(qIk, oIk)（1 <= k <= n）。

    c、假如执行序列的最后一个状态 qI(n+1)不是结束状态 qf，且在时刻 t0，W(t0) 为空或者 qI(n+1)不允许 W(t0) 中的任一个操作 o，我们称读写自旋锁在时刻t0处于潜在死锁状态。这并不表明读写自旋锁真的死锁了，因为随后线程可以提交新的操作，使其继续工作下去。例如 qI(n+1)是“写者持有”状态，而 W(t0) 中全是“读者申请”的操作。但是我们知道锁的持有者一会定在 t0之后的有限时间内提交“写者释放”操作，届时读写自旋锁可以选择执行它，将状态置为“未被持有”，而现存的“读者申请”的操作随后也可被执行了。

    d、如果存在t0 > 0，且对于任意 t >= t0，读写自旋锁在时刻t都处于潜在死锁状态，我们称读写自旋锁从时刻t0开始“死锁”。

    以下是状态图正确性的证明概要：

    a、互斥。从图可知，状态“读者持有”只能转换到自身和“未被持有”，不能转换到“写者持有”，同时状态“写者持有”只能转换到“未被持有”，不能转换到“读者持有”，所以锁一旦被持有，另一种角色的线程只有等到“未被持有”的状态才有机会获得锁，因此读者和写者不可能同时获得锁。状态“写者持有”不允许“写者申请”操作，故而任何时刻只有至多一个写者获得锁。

    b、读者并发。状态“读者持有”允许“读者申请”操作，因此可以有多个读者同时持有锁。

    c、无死锁。证明关于线程执行的 3 个假设。反证法，假设对任意t >= t0，锁在时刻t都处于潜在死锁状态。令q为t0时刻锁的状态，分 3 种情况讨论：

    “未被持有”。如果线程A在 t1 > t0 的时刻提交“读者申请”或“写者申请”的操作，那么锁在t1时刻并不处于潜在死锁状态。

    “读者持有”。持有者必须在某个 t1 > t0 的时刻提交“读者释放”的操作，那么锁在t1时刻并不处于潜在死锁状态。

    “写者持有”。持有者必须在某个 t1 > t0 的时刻提交“写者释放”的操作，那么锁在t1时刻并不处于潜在死锁状态。

    从线程 A 申请锁的角度来看，由状态图知对于任意时刻 t0，不论锁在 t0的状态如何，总存在 t1> t0，锁在时刻 t1必定处于“未被持有”的状态，那么在时刻 t1允许锁申请操作，不是 A 就是别的线程获得锁。如果 A 永远不能获得锁，说明锁一旦处于“未被持有”的状态，就选择了别的线程提交的锁申请操作，那么某个或某些线程必然无限次地获得锁。

    上面提到读写自旋锁有一种选择未执行的操作的能力，即选择函数 S，正是这个函数的差异，导致锁展现不同属性：

    a、读者优先。在任意时刻 t，如果锁处于状态“读者持有”，S 以大于 0 的概率选择一个尚未执行的“读者申请”操作。这意味着：首先，即使有先提交但尚未执行的“写者申请”操作，“读者申请”操作可以被优先执行；其次，没有刻意规定如何选“读者申请”操作，因此多个“读者申请”操作间的执行顺序是不确定的；最后，不排除连续选择“读者释放”操作，使得锁状态迅速变为“未被持有”，只不过这种几率很小。

    b、写者优先。在任意时刻 t，如果 o1是尚未执行的“写者申请”操作，o2是尚未执行的“读者申请”或“写者申请”操作，且 o1在 o2之前提交，那么 S 保证一定在 o2之前选择 o1。

    c、无饥饿。如果线程提交了操作 o，那么 S 必定在有限时间内选择 o。即存在时刻 t，读写自旋锁在 t 的执行序列 < qI1，oI1，qI2，oI2，…，oIn，qI(n+1)> 满足 o = oIn。狭义上， o 限定为“读者申请”或“写者申请”操作。

    d、公平。如果操作 o1在 o2之前提交，那么 S 保证一定在在 o2之前选择执行 o1。狭义上，o1和 o2限定为“读者申请”或“写者申请”操作。

    4、读写自旋锁的实现细节

    上文阐述的自动机模型是个抽象的机器，用于帮助我们理解读写自旋锁的工作原理，但是忽略了很多实现的关键细节：

    a、操作的执行者。如果按照前面的描述，为读写自旋锁创建专门的操作执行线程，那么锁的实际性能将会比较低下，因此我们要求申请线程自己执行提交的操作。

    b、操作类别的区分。可以提供多个调用接口来区分不同种类的操作，避免使用额外变量存放类别信息。

    c、确定操作的提交顺序，即线程的到来的先后关系。写者优先和公平读写自旋锁需要这个信息。可以有 3 种方法：

        ①、假定系统有一个非常精确的实时时钟，线程到来的时刻用于确定顺序。但是寻找直接后继者比较困难，因为事先无法预知线程到来的精确时间。

        ②、参考银行的做法，即每个到来的线程领取一张号码牌，号码的大小决定先后关系。

        ③、将线程组织成一个先进先出（FIFO）的队列，具体实现可以使用单向链表，双向链表等。

    d、在状态 q，确定操作（线程）是否被允许执行。这有 2 个条件：首先 q 必须允许该操作；其次对于写者优先和公平读写自旋锁，不存在先提交但尚未执行的写者（读者 / 写者）申请操作。可以有 3 种方法：

        ①、不停地主动查询这 2 个条件。

        ②、被动等待前一个执行线程通知。

        ③、主动/被动相结合。

    e、选择执行的线程。在状态 q，如果存在多个被允许执行的线程，那么它们必须达成一致（Consensus），保证只有一个线程执行成功，否则会破坏锁状态的一致性。有 2 种简单方法：

        ①、互斥执行。原子指令（总线级别的互斥），或使用锁（高级互斥原语）。

        ②、投机执行。线程不管三七二十一先执行再说，然后检查是否成功。如果不成功，可能需要执行回滚操作。

    f、因为多个读者可以同时持有锁，那么读者释放锁时，有可能需要知道自己是不是最后一个持有者（例如通知后面的写者）。一个简单的方法是用共享计数器保存当前持有锁的读者数目。如果我们对具体数目并不关心，只是想知道计数器是大于 0 还是等于 0，那么用一种称为“非零指示器”（Non-Zero Indicator）的数据结构效果更好。还可以使用双向链表等特殊数据结构。

    5、为读获取和释放一个锁

    read_lock宏，作用于读/写自旋锁的地址*lock，与前面所描述的spin_lock宏非常相似。如果编译内核时选择了内核抢占选项，read_lock宏执行与spin_lock()非常相似的操作，只有一点不同：该宏执行_raw_read_trylock( )函数以在第2步有效地获取读/写自旋锁。

void __lockfunc _read_lock(rwlock_t *lock)
{
    preempt_disable();
    rwlock_acquire_read(&lock->dep_map, 0, 0, _RET_IP_);
    LOCK_CONTENDED(lock, _raw_read_trylock, _raw_read_lock);
}

# define _raw_read_trylock(rwlock)    __raw_read_trylock(&(rwlock)->raw_lock)
static inline int __raw_read_trylock(raw_rwlock_t *lock)
{
    atomic_t *count = (atomic_t *)lock;
    atomic_dec(count);
    if (atomic_read(count) >= 0)
        return 1;
    atomic_inc(count);
    return 0;
}

        movl $rwlp->lock,%eax
        lock; subl $1,(%eax)
        jns 1f
        call _ _read_lock_failed
    1:

    _ _read_lock_failed:
        lock; incl (%eax)
    1:  pause
        cmpl $1,(%eax)
        js 1b
        lock; decl (%eax)
        js _ _read_lock_failed
        ret

    lock; incl rwlp->lock

#define write_lock(lock)        _write_lock(lock)
void __lockfunc _write_lock(rwlock_t *lock)
{
    preempt_disable();
    rwlock_acquire(&lock->dep_map, 0, 0, _RET_IP_);
    LOCK_CONTENDED(lock, _raw_write_trylock, _raw_write_lock);
}

int _raw_write_trylock(rwlock_t *lock)
{
    atomic_t *count = (atomic_t *)lock->lock;
    if (atomic_sub_and_test(0x01000000, count))
        return 1;
    atomic_add(0x01000000, count);
    return 0;
}
 
static __inline__ int atomic_sub_and_test(int i, atomic_t *v)
{
    unsigned char c;
 
    __asm__ __volatile__(
        LOCK "subl %2,%0; sete %1"
        :"=m" (v->counter), "=qm" (c)
        :"ir" (i), "m" (v->counter) : "memory");
    return c;
}

lock; addl $0x01000000,rwlp

typedef struct{
    unsigned int sequence;
    spinlock_t lock;
} seqlock_t;

void write_seqlock(seqlock_t *sl);

void write_sequnlock(seqlock_t *sl);

write_seqlock(&seqlock_a);
//写操作代码块
…
write_sequnlock(&seqlock_a);

int write_tryseqlock(seqlock_t *sl);

unsigned read_seqbegin(const seqlock_t *sl);

int read_seqretry(const seqlock_t *sl, unsigned iv);

do {
   seqnum = read_seqbegin(&seqlock_a);
//读操作代码块
...
} while (read_seqretry(&seqlock_a, seqnum));

write_seqlock_irqsave(lock, flags)

write_seqlock_irq(lock)

write_seqlock_bh(lock)

write_sequnlock_irqrestore(lock, flags)

write_sequnlock_irq(lock)

write_sequnlock_bh(lock)

read_seqbegin_irqsave(lock, flags)

read_seqretry_irqrestore(lock, iv, flags)

do {
   seqnum = read_seqbegin_irqsave(&seqlock_a, flags);
//读操作代码块
...
} while (read_seqretry_irqrestore(&seqlock_a, seqnum, flags));

unsigned read_seqcount_begin(const seqcount_t *s);

int read_seqcount_retry(const seqcount_t *s, unsigned iv);

do {
   seqnum = read_seqbegin_count(&seqcount_a);
//读操作代码块
...
} while (read_seqretry(&seqcount_a, seqnum));

void write_seqcount_begin(seqcount_t *s);

void write_seqcount_end(seqcount_t *s);

write_seqcount_begin(&seqcount_a);
//写操作代码块
…
write_seqcount_end(&seqcount_a);

struct semaphore {
	atomic_t count;
	int sleepers;
	wait_queue_head_t wait;
};

struct semaphore{
    spinlock_t lock;
    unsigned int count;
    struct list_head wait_list;
};

struct __wait_queue_head {
    wq_lock_t lock;
    struct list_head task_list;
#if WAITQUEUE_DEBUG
    long __magic;
    long __creator;
#endif
};
 
typedef struct __wait_queue_head wait_queue_head_t;

static inline void sema_init (struct semaphore *sem, int val)
{
/*
 *	*sem = (struct semaphore)__SEMAPHORE_INITIALIZER((*sem),val);
 *
 * i‘d rather use the more flexible initialization above, but sadly
 * GCC 2.7.2.3 emits a bogus warning. EGCS doesn‘t. Oh well.
 */
	atomic_set(&sem->count, val);
	sem->sleepers = 0;
	init_waitqueue_head(&sem->wait);
}
 
static inline void init_MUTEX (struct semaphore *sem)
{
	sema_init(sem, 1);
}
 
static inline void init_MUTEX_LOCKED (struct semaphore *sem)
{
	sema_init(sem, 0);
}

#define __SEMAPHORE_INITIALIZER(name, n)				\
{									\
	.count		= ATOMIC_INIT(n),				\
	.sleepers	= 0,						\
	.wait		= __WAIT_QUEUE_HEAD_INITIALIZER((name).wait)	\
}
 
#define __MUTEX_INITIALIZER(name) \
	__SEMAPHORE_INITIALIZER(name,1)
 
#define __DECLARE_SEMAPHORE_GENERIC(name,count) \
	struct semaphore name = __SEMAPHORE_INITIALIZER(name,count)
 
#define DECLARE_MUTEX(name) __DECLARE_SEMAPHORE_GENERIC(name,1)

#define DECLARE_MUTEX_LOCKED(name) __DECLARE_SEMAPHORE_GENERIC(name,0)

  movl $sem->count, %ecx   
  lock; incl (%ecx)
  jg lf                  # 大于0跳转至1标记处
  lea %ecx, %eax         # lea(Load effect address): 取有效地址，也就是取偏移地址---- lea 目的 源：即将源中的地址传给目的.
  pushl %edx             # 保存现场
  pushl %ecx
  call __up              # 队列释放（唤醒队列中的睡眠进程）
  popl %ecx
  popl %edx

1：

fastcall void __up(struct semaphore *sem)
{
    wake_up(&sem->wait);
}

down:
    movl $sem->count, %ecx
    lock; decl (%ecx)
    jns 1f            #JNS（jump if not sign）,汇编语言中的条件转移指令.结果为正则转移.
    lea %ecx, %eax
    pushl %edx
    pushl %ecx
    call __down
    popl %ecx
    popl %edx
1:

fastcall void __sched __down(struct semaphore * sem)
{
	struct task_struct *tsk = current;
	DECLARE_WAITQUEUE(wait, tsk);
	unsigned long flags;
 
	tsk->state = TASK_UNINTERRUPTIBLE;
	spin_lock_irqsave(&sem->wait.lock, flags);
	add_wait_queue_exclusive_locked(&sem->wait, &wait);
 
	sem->sleepers++;
	for (;;) {
		int sleepers = sem->sleepers;
 
		/*
		 * Add "everybody else" into it. They aren‘t
		 * playing, because we own the spinlock in
		 * the wait_queue_head.
		 */
		if (!atomic_add_negative(sleepers - 1, &sem->count))  
       // atomic_add_negative(i,v):把i加到*v,如果结果为负，返回1,如果结果为0或正数,返回0.
       // 同时注意该函数会将sleepers-1加到sem->count并保存该值至sem->count,这就会有这样一个细节：
       // 如果有睡眠进程,sleepers=1, sleepers++;=>>2, 这样sleepers-1+sem->count就相当于恢复了前面的sem->count
       // (注意刚开始时由汇编语言导致sem->count--,现在通过+1即保证了count的取值范围即为1/0/-1).
        {
			sem->sleepers = 0;
			break;
		}
		sem->sleepers = 1;	/* us - see -1 above */
		spin_unlock_irqrestore(&sem->wait.lock, flags);
 
		schedule();
 
		spin_lock_irqsave(&sem->wait.lock, flags);
		tsk->state = TASK_UNINTERRUPTIBLE;
	}

	remove_wait_queue_locked(&sem->wait, &wait);
	wake_up_locked(&sem->wait);
	spin_unlock_irqrestore(&sem->wait.lock, flags);
	tsk->state = TASK_RUNNING;
}

/*
 * the semaphore definition
 */
struct rw_semaphore {
	signed long		count;
#define RWSEM_UNLOCKED_VALUE		0x00000000
#define RWSEM_ACTIVE_BIAS		0x00000001
#define RWSEM_ACTIVE_MASK		0x0000ffff
#define RWSEM_WAITING_BIAS		(-0x00010000)
#define RWSEM_ACTIVE_READ_BIAS		RWSEM_ACTIVE_BIAS
#define RWSEM_ACTIVE_WRITE_BIAS		(RWSEM_WAITING_BIAS + RWSEM_ACTIVE_BIAS)
	spinlock_t		wait_lock;
	struct list_head	wait_list;
#if RWSEM_DEBUG
	int			debug;
#endif
};

/* interrupt control.. */
#define local_save_flags(x)	do { typecheck(unsigned long,x); __asm__ __volatile__("pushfl ; popl %0":"=g" (x): /* no input */); } while (0)
#define local_irq_restore(x) 	do { typecheck(unsigned long,x); __asm__ __volatile__("pushl %0 ; popfl": /* no output */ :"g" (x):"memory", "cc"); } while (0)
#define local_irq_disable() 	__asm__ __volatile__("cli": : :"memory")
#define local_irq_enable()	__asm__ __volatile__("sti": : :"memory")
/* For spinlocks etc */
#define local_irq_save(x)	__asm__ __volatile__("pushfl ; popl %0 ; cli":"=g" (x): /* no input */ :"memory")

new->next = list_element->next;
list_element->next = new;

new->next = list_element->next;
wmb();
list_element->next = new;

depth = current->lock_depth + 1;
if(depth == 0)
    down(&kernel_sem);
current->lock_depth = depth;

if(--current->lock_depth < 0)
    up(&kernel_sem);?