[自制操做系统] 原子操做&核间中断&读写锁&PRWLock
本文主要为读论文Scalable Read-mostly Synchronization Using Passive Reader-Writer Locks的记录。
并将其在JOS上实现。其中包括lapic原理，IPI 实现。
本文中支持的新特性：php
- 支持原子操做
- 支持读写锁
- 支持针对单一核心
IPI - 支持
PRWLock
Github : https://github.com/He11oLiu/MOScss
论文阅读记录
研究背景
- 单核性能提高遇到瓶颈 转向多核提高性能
- 单核主要为计算密集型模型，多核主要为并行化模型
- 并行化模型面临的问题：多个处理器共享数据结构，须要同步原语来保证其一致性
- 更底层：内存一致性 然而会致使可扩展性问题
- Strict Consistency Memory Barrier
- Sequential Consistency TSO
可扩展性 Scalable
提到可扩展性，不得不提Amdahl's lawhtml
S(latency)(s) = 1/((1-p)+p/s)
其中1-p极为不能够并行的部分，而对于一个处理器，形成(1-P)部分有如下几种缘由：linux
- 内存屏障时等待以前的指令执行完
- MESI模型中等待获取最新值
- 等待其余处理器释放锁
- 多核之间的通信带宽受限，阻塞
关于读写锁
- 读读不阻塞，读写阻塞
- 适合读数据频率远大于写数据频率的应用
- 单核上的实现思路：读者锁不冲突，当须要加写者锁的时候须要等全部的读者锁释放。利用一个读者计数器来实现。
- 多核上最直观的实现思路：每一个核上保存本身的读者锁，写者须要等到全部的读者锁所有释放了才能开始获取锁。
现有的RWLock所作的尝试
BRLOCK C-SNZI
- 读者申请本身核上的锁
- 当只有读者时，因为只是访问本身的核上的锁，因此有良好的扩展性
- 写者须要获取全部核上的互斥锁，恒定延迟，与处理器数量有关。
- SNZI利用树进行了必定优化
RCU
- 弱化语义，可能会读到脏的数据（逻辑矛盾）
- 读者无锁，写着读者能够同时进行
- 先写值再改指针
- 写者开销大，要处理旧的数据
- 垃圾回收(无抢占调度一圈)
Bounded staleness 短内存可见性
所谓短内存可见性，也就是在很短的时间周期内，因为每一个核上面的单独cache很是的小，很大概率会被替换掉，从而能看到最新的数据。下面是具体的图表git
PRWLock的设计思路
- 在短期内各个处理器都可以看到最新的版本号
- 利用TSO架构的优点，版本控制隐式表示退出CS区域
- 并不彻底依赖于短期可见，对于特殊状况，保证一致性，利用IPI要求进行Report IPI消息传递开销较小，且能够相互掩盖。
- 两种模式 支持调度（睡眠与抢占）
PRWLock 的两种模式
Passive Mode
- 用于处理没有通过调度的读者
- 共享内存陈旧化
- 弱控制，经过版本控制隐式反馈+少数状况IPI
Active Mode (用于支持睡眠与调度相似BRLock)
- 用于处理被调度过的进程（睡眠／抢占）
- 经过维护active检测数量
- 强控制，主动监听
- 主动等待
PRWLock流程视频：
优酷视频github
PRWLock的正确性
- 写者发现读者获取了最新的版本变量时，因为TSO的特性，也必定看到了写者上的写锁，确信其不会再次进入临界区。
- 对于须要较长时间才能看到最新的版本号或没有读者指望获取读者锁提供了IPI来下降等待时间，避免无限等待。
PRWLock 内核中减小IPIs
- 锁域(Lock Domain)用于表示一个锁对应的处理器范围
- 若上下文切换到了其余的进程，就不必管这个核的锁了
- 锁域的上下线能够避免一些没有必要的一致性检测
- 注意利用内存屏障来保证一致性
PRWLock 用户态实现
因为在用户态有如下两个特色sql
- 用户态不能随时关闭抢占(Preemption)
- 用户态不能发送核间中断(IPI)
因此PRWLock在用户态实现的思路以下：shell
- 利用抢断标记位避免特殊窗口时被抢断
- 写者必须陷入内核态发送IPI
PRWLock 性能分析
读者
- 读者之间无内存屏障 （无关联）
- 锁域上下线原本就是极少的操做，用来改善性能的，因此其中的内存屏障影响不大
- 对于长CS区的读者，与传统同样
写者
- IPI只要几百个cycle 自己也要等待
- 多个写者能够直接把锁传递
总结
- 利用了短内存写全局可见时间
- 利用了TSO的特性设计的版本控制来隐式维护语义
- 利用IPI来保证特殊状况
- 利用两种模式支持调度
- 读者之间无关联（内存屏障），提高读者性能
- PWAKE 分布式唤醒，提升了唤醒并行性
移植PRWLock到JOS
JOS的核间中断实现
关于Local APIC
在一个基于
APIC的系统中，每个核心都有一个Local APIC，Local APIC负责处理CPU中特定的中断配置。还有其余事情，它包含了Local Vector Table(LVT)负责配置事件中断。api此外，还有一个CPU外面的
IO APIC(例如Intel82093AA）的芯片组，而且提供基于多处理器的中断管理，在多个处理器之间，实现静态或动态的中断触发路由。数组
Inter-Processor Interrupts（IPIs)是一种由Local APIC触发的中断，通常能够用于多CPU间调度之类的使用想要开启
Local APIC接收中断，则须要设置Spurious Interrupt Vector Register的第8位便可。使用
APIC Timer的最大好处是每一个cpu内核都有一个定时器。相反PIT(Programmable Interval Timer)就不这样，PIT是共用的一个。
周期触发模式
周期触发模式中，程序设置一个”初始计数“寄存器（Initial Count），同时Local APIC会将这个数复制到”当前计数“寄存器（Current Count）。Local APIC会将这个数（当前计数）递减，直到减到0为止，这时候将触发一个IRQ（能够理解为触发一次中断），与此同时将当前计数恢复到初始计数寄存器的值，而后周而复始的执行上述逻辑。可使用这种方法经过Local APIC实现定时按照必定时间间隔触发中断的功能。
一次性触发模式
同以前同样，可是不会恢复到初始计数。
TSC-Deadline Modie
cpu的时间戳到达deadline的时候会触发IRQ每一个本地
APIC都有 32 位的寄存器，一个内部时钟，一个本地定时设备以及为本地中断保留的两条额外的 IRQ 线 LINT0 和 LINT1。全部本地 APIC 都链接到 I/O APIC，造成一个多级 APIC 系统。Intel x86架构提供LINT0和LINT1两个中断引脚，他们一般与Local APIC相连，用于接收Local APIC传递的中断信号，另外，当Local APIC被禁用的时候，LINT0和LINT1即被配置为INTR和NMI管脚，即外部IO中断管脚和非屏蔽中断管脚。
The local APIC registers are memory mapped to an address that can be found in the MP/MADT tables. Make sure you map these to virtual memory if you are using paging. Each register is 32 bits long, and expects to written and read as a 32 bit integer. Although each register is 4 bytes, they are all aligned on a 16 byte boundary.
再一次详细查看JOS中的核间中断的实现方式，
因为这段映射，设置了nocache和直写的特性，便于对于IO的操做。
void lapic_init(void)
{
if (!lapicaddr)
return;
// lapicaddr is the physical address of the LAPIC's 4K MMIO
// region. Map it in to virtual memory so we can access it.
lapic = mmio_map_region(lapicaddr, 4096);
// Enable local APIC; set spurious interrupt vector.
lapicw(SVR, ENABLE | (IRQ_OFFSET + IRQ_SPURIOUS));
// The timer repeatedly counts down at bus frequency
// from lapic[TICR] and then issues an interrupt.
// If we cared more about precise timekeeping,
// TICR would be calibrated using an external time source.
lapicw(TDCR, X1);
lapicw(TIMER, PERIODIC | (IRQ_OFFSET + IRQ_TIMER));
lapicw(TICR, 10000000);
// Leave LINT0 of the BSP enabled so that it can get
// interrupts from the 8259A chip.
//
// According to Intel MP Specification, the BIOS should initialize
// BSP's local APIC in Virtual Wire Mode, in which 8259A's
// INTR is virtually connected to BSP's LINTIN0. In this mode,
// we do not need to program the IOAPIC.
if (thiscpu != bootcpu)
lapicw(LINT0, MASKED);
// Disable NMI (LINT1) on all CPUs
lapicw(LINT1, MASKED);
// Disable performance counter overflow interrupts
// on machines that provide that interrupt entry.
if (((lapic[VER] >> 16) & 0xFF) >= 4)
lapicw(PCINT, MASKED);
// Map error interrupt to IRQ_ERROR.
lapicw(ERROR, IRQ_OFFSET + IRQ_ERROR);
// Clear error status register (requires back-to-back writes).
lapicw(ESR, 0);
lapicw(ESR, 0);
// Ack any outstanding interrupts.
lapicw(EOI, 0);
// Send an Init Level De-Assert to synchronize arbitration ID's.
lapicw(ICRHI, 0);
lapicw(ICRLO, BCAST | INIT | LEVEL);
while (lapic[ICRLO] & DELIVS)
;
// Enable interrupts on the APIC (but not on the processor).
lapicw(TPR, 0);
}
lapic_init将LAPIC映射到lapicaddr地址上，而且初始化LAPIC各类中断参数。
// Local APIC registers, divided by 4 for use as uint32_t[] indices.
#define ID (0x0020 / 4) // ID
这里的宏定义为/4是由于MMIO映射到MMIOaddr，保存在volatile uint32_t *lapic;中。这个单位是uint32_t，故全部的地址均/4
下面来看一下主要的APIC Registers
-
EOI Register
Write to the register with offset 0xB0 using the value 0 to signal an end of interrupt. A non-zero values causes a general protection fault.
#define EOI (0x00B0 / 4) // EOI // Acknowledge interrupt. void lapic_eoi(void) { if (lapic) lapicw(EOI, 0); } -
Local Vector Table Registers
There are some special interrupts that the processor and LAPIC can generate themselves. While external interrupts are configured in the I/O APIC, these interrupts must be configured using registers in the LAPIC. The most interesting registers are:
0x320 = lapic timer0x350 = lint00x360 = lint1JOS在这里只保留了BSP的LINT0用于接受8259A的中断，其余的LINT0与LINT1非屏蔽中断，均设置为MASKED// Leave LINT0 of the BSP enabled so that it can get // interrupts from the 8259A chip. // // According to Intel MP Specification, the BIOS should initialize // BSP's local APIC in Virtual Wire Mode, in which 8259A's // INTR is virtually connected to BSP's LINTIN0. In this mode, // we do not need to program the IOAPIC. if (thiscpu != bootcpu) lapicw(LINT0, MASKED); // Disable NMI (LINT1) on all CPUs lapicw(LINT1, MASKED); -
Spurious Interrupt Vector Register
The offset is 0xF0. The low byte contains the number of the spurious interrupt. As noted above, you should probably set this to 0xFF. To enable the APIC, set bit 8 (or 0x100) of this register. If bit 12 is set then EOI messages will not be broadcast. All the other bits are currently reserved.
// Enable local APIC; set spurious interrupt vector. lapicw(SVR, ENABLE | (IRQ_OFFSET + IRQ_SPURIOUS)); -
Interrupt Command Register
The interrupt command register is made of two 32-bit registers; one at 0x300 and the other at 0x310.
#define ICRHI (0x0310 / 4) // Interrupt Command [63:32] #define ICRLO (0x0300 / 4) // Interrupt Command [31:0]It is used for sending interrupts to different processors.
The interrupt is issued when 0x300 is written to, but not when 0x310 is written to. Thus, to send an interrupt command one should first write to 0x310, then to 0x300.
须要先写
ICRHI，而后在写ICRLO的时候就会产生中断。At 0x310 there is one field at bits 24-27, which is local APIC ID of the target processor (for a physical destination mode).
lapicw(ICRHI, apicid << 24);给
ICRHI中断目标核心的local APIC ID。这里的apicid是在MP Floating Pointer Structure读的时候顺序给的cpu_id。ICRLO的分布比较重要
- 其中目标模式有(
8-10)
#define INIT 0x00000500 // INIT/RESET #define STARTUP 0x00000600 // Startup IPI- 其中发送模式有(
18~19)
#define SELF 0x00040000 // Send to self #define BCAST 0x00080000 // Send to all APICs, including self. #define OTHERS 0x000C0000 // Send to all APICs, excluding self.不设置的话则为发送给
0x310 ICRHI制定的核心。 - 其中目标模式有(
综上，打包了一个IPI发送的接口，
void lapic_ipi(int vector)
{
lapicw(ICRLO, OTHERS | FIXED | vector);
while (lapic[ICRLO] & DELIVS)
;
}
用于发送IPI与IPI ACK均是利用MMIO直接对相应地址书写，比较简单。
这里测试一下，先设置trap中的IPI中断
#define T_PRWIPI 20 // IPI report for PRWLock
void prw_ipi_report(struct Trapframe *tf)
{
cprintf("%d in ipi report\n",cpunum());
}
在trap_dispatch中加入对这个中断的分发
case T_PRWIPI:
prw_ipi_report(tf);
break;
最后在init的时候用bsp发送IPI给全部其余核心
lapic_ipi(T_PRWIPI);
设置QEMU模拟4个核心来测试IPI是否正确
1 in ipi report
3 in ipi report
2 in ipi report
非BSP能够正确的接受IPI并进入中断处理历程。
JOS实现传统内核态读写锁
typedef struct dumbrwlock {
struct spinlock lock;
atomic_t readers;
}dumbrwlock;
void rw_initlock(dumbrwlock *rwlk)
{
spin_initlock(&rwlk->lock);
rwlk->readers.counter = 0;
}
void dumb_wrlock(dumbrwlock *rwlk)
{
spin_lock(&rwlk->lock);
while (rwlk->readers.counter > 0)
asm volatile("pause");
}
void dumb_wrunlock(dumbrwlock *rwlk)
{
spin_unlock(&rwlk->lock);
}
void dumb_rdlock(dumbrwlock *rwlk)
{
while (1)
{
atomic_inc(&rwlk->readers);
if (!rwlk->lock.locked)
return;
atomic_dec(&rwlk->readers);
while (rwlk->lock.locked)
asm volatile("pause");
}
}
void dumb_rdunlock(dumbrwlock *rwlk)
{
atomic_dec(&rwlk->readers);
}
而后发现一个比较大的问题，JOS没有实现原子操做，先实现原子操做再进行下面的尝试。
JOS 实现原子操做
仿造linux 2.6内核，实现原子操做
#ifndef JOS_INC_ATOMIC_H_
#define JOS_INC_ATOMIC_H_
/* * Atomic operations that C can't guarantee us. Useful for * resource counting etc.. */
#include <inc/types.h>
#define LOCK "lock ; "
/* * Make sure gcc doesn't try to be clever and move things around * on us. We need to use _exactly_ the address the user gave us, * not some alias that contains the same information. */
typedef struct
{
volatile int counter;
} atomic_t;
#define ATOMIC_INIT(i) \
{ \
(i) \
}
/** * atomic_read - read atomic variable * @v: pointer of type atomic_t * * Atomically reads the value of @v. */
#define atomic_read(v) ((v)->counter)
/** * atomic_set - set atomic variable * @v: pointer of type atomic_t * @i: required value * * Atomically sets the value of @v to @i. */
#define atomic_set(v, i) (((v)->counter) = (i))
/** * atomic_add - add integer to atomic variable * @i: integer value to add * @v: pointer of type atomic_t * * Atomically adds @i to @v. */
static __inline__ void atomic_add(int i, atomic_t *v)
{
__asm__ __volatile__(
LOCK "addl %1,%0"
: "=m"(v->counter)
: "ir"(i), "m"(v->counter));
}
/** * atomic_sub - subtract the atomic variable * @i: integer value to subtract * @v: pointer of type atomic_t * * Atomically subtracts @i from @v. */
static __inline__ void atomic_sub(int i, atomic_t *v)
{
__asm__ __volatile__(
LOCK "subl %1,%0"
: "=m"(v->counter)
: "ir"(i), "m"(v->counter));
}
/** * atomic_sub_and_test - subtract value from variable and test result * @i: integer value to subtract * @v: pointer of type atomic_t * * Atomically subtracts @i from @v and returns * true if the result is zero, or false for all * other cases. */
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;
}
/** * atomic_inc - increment atomic variable * @v: pointer of type atomic_t * * Atomically increments @v by 1. */
static __inline__ void atomic_inc(atomic_t *v)
{
__asm__ __volatile__(
LOCK "incl %0"
: "=m"(v->counter)
: "m"(v->counter));
}
/** * atomic_dec - decrement atomic variable * @v: pointer of type atomic_t * * Atomically decrements @v by 1. */
static __inline__ void atomic_dec(atomic_t *v)
{
__asm__ __volatile__(
LOCK "decl %0"
: "=m"(v->counter)
: "m"(v->counter));
}
/** * atomic_dec_and_test - decrement and test * @v: pointer of type atomic_t * * Atomically decrements @v by 1 and * returns true if the result is 0, or false for all other * cases. */
static __inline__ int atomic_dec_and_test(atomic_t *v)
{
unsigned char c;
__asm__ __volatile__(
LOCK "decl %0; sete %1"
: "=m"(v->counter), "=qm"(c)
: "m"(v->counter)
: "memory");
return c != 0;
}
/** * atomic_inc_and_test - increment and test * @v: pointer of type atomic_t * * Atomically increments @v by 1 * and returns true if the result is zero, or false for all * other cases. */
static __inline__ int atomic_inc_and_test(atomic_t *v)
{
unsigned char c;
__asm__ __volatile__(
LOCK "incl %0; sete %1"
: "=m"(v->counter), "=qm"(c)
: "m"(v->counter)
: "memory");
return c != 0;
}
/** * atomic_add_negative - add and test if negative * @v: pointer of type atomic_t * @i: integer value to add * * Atomically adds @i to @v and returns true * if the result is negative, or false when * result is greater than or equal to zero. */
static __inline__ int atomic_add_negative(int i, atomic_t *v)
{
unsigned char c;
__asm__ __volatile__(
LOCK "addl %2,%0; sets %1"
: "=m"(v->counter), "=qm"(c)
: "ir"(i), "m"(v->counter)
: "memory");
return c;
}
/** * atomic_add_return - add and return * @v: pointer of type atomic_t * @i: integer value to add * * Atomically adds @i to @v and returns @i + @v */
static __inline__ int atomic_add_return(int i, atomic_t *v)
{
int __i;
/* Modern 486+ processor */
__i = i;
__asm__ __volatile__(
LOCK "xaddl %0, %1;"
: "=r"(i)
: "m"(v->counter), "0"(i));
return i + __i;
}
static __inline__ int atomic_sub_return(int i, atomic_t *v)
{
return atomic_add_return(-i, v);
}
#define atomic_inc_return(v) (atomic_add_return(1, v))
#define atomic_dec_return(v) (atomic_sub_return(1, v))
/* These are x86-specific, used by some header files */
#define atomic_clear_mask(mask, addr) \
__asm__ __volatile__(LOCK "andl %0,%1" \
: \
: "r"(~(mask)), "m"(*addr) \
: "memory")
#define atomic_set_mask(mask, addr) \
__asm__ __volatile__(LOCK "orl %0,%1" \
: \
: "r"(mask), "m"(*(addr)) \
: "memory")
#endif
而后在内核中对读写锁的功能进行测试。
遇到两个问题
-
一个是
asm volatile("pause");容易死在那个循环里面，不会从新换到这个CPU中，在DEBUG的时候发如今先后加上cprintf其就会顺利换回来。while (rwlk->lock.locked) { cprintf(""); asm volatile("pause"); } -
另外一个是设计内核中的测试
- 多核上的输出可能会并行化，要减短输出内容。
- 在用户空间的锁分享目前很差作，linux是基于文件的。
- 故设计了两个锁来进行测试
一个是
CPU 0的writer锁，一个是reader锁。// test reader-writer lock rw_initlock(&lock1); rw_initlock(&lock2); dumb_wrlock(&lock1); cprintf("[rw] CPU %d gain writer lock1\n", cpunum()); dumb_rdlock(&lock2); cprintf("[rw] CPU %d gain reader lock2\n", cpunum()); // Starting non-boot CPUs boot_aps(); cprintf("[rw] CPU %d going to release writer lock1\n", cpunum()); dumb_wrunlock(&lock1); cprintf("[rw] CPU %d going to release reader lock2\n", cpunum()); dumb_rdunlock(&lock2);对于每一个核上，分别获取
lock1的读着锁与lock2的写者锁。添加asm volatile("pause");是想让其余核模拟上线来检测各类状况。dumb_rdlock(&lock1); cprintf("[rw] %d l1\n", cpunum()); asm volatile("pause"); dumb_rdunlock(&lock1); cprintf("[rw] %d unl1\n", cpunum()); dumb_wrlock(&lock2); cprintf("[rw] %d l2\n", cpunum()); asm volatile("pause"); cprintf("[rw] %d unl2\n", cpunum()); dumb_wrunlock(&lock2);在给
QEMU四核参数CPUS=4的时候下的运行状况以下：
[rw] CPU 0 gain writer lock1
[rw] CPU 0 gain reader lock2
[MP] CPU 1 starting
[MP] CPU 2 starting
[MP] CPU 3 starting
[rw] CPU 0 going to release writer lock1
[rw] CPU 0 going to release reader lock2
[rw] 1 l1
[rw] 2 l1
[rw] 3 l1
[rw] 2 unl1
[rw] 2 l2
[rw] 3 unl1
[rw] 1 unl1
[rw] 2 unl2
[MP] CPU 2 sched
[rw] 3 l2
[rw] 3 unl2
[rw] 1 l2
[MP] CPU 3 sched
[rw] 1 unl2
[MP] CPU 1 sched
能够观察到一旦CPU0释放了lock1的写者锁，全部的核都可以得到lock1的读者锁。然后CPU2得到了lock2的写者锁后，其余核上线，CPU3与CPU1只是释放了lock1，没法得到lock2，只有等CPU2释放了lock2才能获取。
这与指望的读写锁的功能是一致的。至此普通读写锁的实现完成。
JOS实现PRWLock
首先有几个重点：
PRWLock数据结构设计- 锁的具体实现
- 调度时调用内容
PRWLock的测试
PRWLock的数据结构
enum lock_status
{
FREE = 0,
LOCKED,
PASS,
PASSIVE
};
struct percpu_prwlock
{
enum lock_status reader;
atomic_t version;
};
typedef struct prwlock
{
enum lock_status writer;
struct percpu_prwlock lockinfo[NCPU];
atomic_t active;
atomic_t version;
} prwlock;
对于一个prwlock，除了其主要的版本以及ACTIVE的读者数量，还须要保存每一个核心持有该锁的版本号，以及每一个核上该锁的读者状态。这里直接经过lockinfo数组索引每一个核对应的该锁信息。
而全局内核所拥有的读写锁经过locklist进行索引，在init的时候加入到这个list中去。
extern unsigned int prwlocknum;
extern prwlock *locklist[MAXPRWLock];
锁的具体操做
初始化操做的时候须要设置各类初值，并将其添加到list中
void prw_initlock(prwlock *rwlk)
{
int i = 0;
rwlk->writer = FREE;
for (i = 0; i < NCPU; i++)
{
rwlk->lockinfo[i].reader = FREE;
atomic_set(&rwlk->lockinfo[i].version, 0);
}
atomic_set(&rwlk->active, 0);
atomic_set(&rwlk->version, 0);
locklist[prwlocknum++] = rwlk;
}
剩下的与论文中伪代码的实现思路相同，只是具体调用的函数有一些差异。
读者锁中包括向核心发送ipi。这里只是示意，就没有写PASS的具体部分，能够经过添加一个等待标志变量来实现。
void prw_wrlock(prwlock *rwlk)
{
int newVersion;
int id = 0;
unsigned int corewait = 0;
if (rwlk->writer == PASS)
return;
rwlk->writer = LOCKED;
newVersion = atomic_inc_return(&rwlk->version);
for (id = 0; id < ncpu; id++)
{
#ifdef TESTPRW
cprintf("CPU %d Ver %d\n", id, atomic_read(&rwlk->lockinfo[id].version));
#endif
if (id != cpunum() && atomic_read(&rwlk->lockinfo[id].version) != newVersion)
{
lapic_ipi_dest(id, PRWIPI);
corewait |= binlist[id];
#ifdef TESTPRW
cprintf("send ipi %d\n", id);
#endif
}
}
for (id = 0; id < ncpu; id++)
{
if (corewait & binlist[id])
{
while (atomic_read(&rwlk->lockinfo[id].version) != newVersion)
asm volatile("pause");
}
}
while (atomic_read(&rwlk->active) != 0)
{
lock_kernel();
sched_yield();
}
}
void prw_wrunlock(prwlock *rwlk)
{
// if someone waiting to gain write lock rwlk->writer should be PASS
rwlk->writer = FREE;
}
void prw_rdlock(prwlock *rwlk)
{
struct percpu_prwlock *st;
int lockversion;
st = &rwlk->lockinfo[cpunum()];
st->reader = PASSIVE;
while (rwlk->writer != FREE)
{
st->reader = FREE;
lockversion = atomic_read(&rwlk->version);
atomic_set(&st->version, lockversion);
while (rwlk->writer != FREE)
asm volatile("pause");
st = &rwlk->lockinfo[cpunum()];
st->reader = PASSIVE;
}
}
void prw_rdunlock(prwlock *rwlk)
{
struct percpu_prwlock *st;
int lockversion;
st = &rwlk->lockinfo[cpunum()];
if (st->reader == PASSIVE)
st->reader = FREE;
else
atomic_dec(&rwlk->active);
lockversion = atomic_read(&rwlk->version);
atomic_set(&st->version, lockversion);
}
每一个核心接到PRWIPI的处理函数
void prw_ipi_report(struct Trapframe *tf)
{
int lockversion, i;
struct percpu_prwlock *st;
cprintf("In IPI_report CPU %d\n", cpunum());
for (i = 0; i < prwlocknum; i++)
{
st = &locklist[i]->lockinfo[cpunum()];
if (st->reader != PASSIVE)
{
lockversion = atomic_read(&locklist[i]->version);
atomic_set(&st->version, lockversion);
}
}
}
调度时调用内容
调度时须要将全部的锁均进行处理，因此要遍历locklist
// Implement PRWLock
if (prwlocknum != 0)
for (j = 0; j < prwlocknum; j++)
prw_sched(locklist[j]);
具体的prw_sched以下：
void prw_sched(prwlock *rwlk)
{
struct percpu_prwlock *st;
int lockversion;
st = &rwlk->lockinfo[cpunum()];
if (st->reader == PASSIVE)
{
atomic_inc(&rwlk->active);
st->reader = FREE;
}
lockversion = atomic_read(&rwlk->version);
atomic_set(&st->version, lockversion);
}
PRWLock的测试
测试PRWLock也比较复杂，因为咱们使用的是big kernel lock，因此内核态里面很差测试，直接在初始化开始RR以前测试。这里引入一个新的IPI进行测试。
void prw_debug(struct Trapframe *tf)
{
int needlock = 0;
cprintf("====CPU %d in prw debug====\n",cpunum());
if(kernel_lock.cpu == thiscpu && kernel_lock.locked == 1)
{
unlock_kernel();
needlock = 1;
}
prw_wrlock(&lock1);
cprintf("====%d gain lock1====\n",cpunum());
prw_wrunlock(&lock1);
cprintf("====%d release lock1====\n",cpunum());
if(needlock)
lock_kernel();
}
给一个核心发送DEBUGPRW中断，即让其获取lock1的写者锁。
#ifdef TESTPRW
unlock_kernel();
prw_initlock(&lock1);
prw_wrlock(&lock1);
prw_wrunlock(&lock1);
prw_rdlock(&lock1);
cprintf("====%d Gain Reader Lock====\n", cpunum());
lapic_ipi_dest(3, DEBUGPRW);
for (int i = 0; i < 10000; i++)
asm volatile("pause");
prw_rdunlock(&lock1);
cprintf("====%d release Reader Lock====\n", cpunum());
lock_kernel();
#endif
这里先用unlock_kernel，避免其余核心没法接收中断，最后再lock_kernel，才能开始sched。
测试选择6个核心
SMP: CPU 0 found 6 CPU(s)
enabled interrupts: 1 2 4
[MP] CPU 1 starting
[MP] CPU 2 starting
[MP] CPU 3 starting
[MP] CPU 4 starting
[MP] CPU 5 starting
[MP] CPU 1 sched
[MP] CPU 2 sched
[MP] CPU 3 sched
[MP] CPU 4 sched
[MP] CPU 5 sched
CPU 0 Ver 0
CPU 1 Ver 0
send ipi 1
CPU 2 Ver 0
send ipi 2
CPU 3 Ver 0
send ipi 3
CPU 4 Ver 0
send ipi 4
CPU 5 Ver 0
====0 Gain Reader Lock==== In IPI_report CPU 1 In IPI_report CPU 2 In IPI_report CPU 4 FS is running ====CPU 3 in prw debug==== FS can do I/O CPU 0 Ver 0 Dsend ipi 0 evice 1 presence: 1 CPU 1 Ver 1 send ipi 1 CPU 2 Ver 2 CPU 3 Ver 1 CPU 4 Ver 1 send ipi 4 CPU 5 Ver 1 send ipi 5 In IPI_report CPU 5 $ block cache is good superblock is good bitmap is good alloc_block is good file_open is good file_get_block is good file_flush is good file_truncate is good file rewrite is good ====0 release Reader Lock==== Init finish! Sched start... ====3 gain lock1==== ====3 release lock1====
当CPU0释放了读着锁以后，CPU3才可以获取lock1，测试正确
- 1. 原子操做&普通锁&读写锁
- 2. 原子操做、互斥锁、读写锁
- 3. 【自制操做系统08】中断
- 4. 【自制操做系统13】锁
- 5. 操做系统概念-内核同步-原子操做
- 6. 操做系统核心原理-1.操做系统导论
- 7. 原子操做和锁
- 8. 锁----和 原子操做
- 9. 无锁-CAS原子操做
- 10. 操做系统-中断（1）中断
- 更多相关文章...
- • 操作系统（OS）平台 统计 - 浏览器信息
- • Java 中操作 R - R 语言教程
- • 漫谈MySQL的锁机制
- • Docker容器实战(七) - 容器眼光下的文件系统
-
每一个你不满意的现在，都有一个你没有努力的曾经。
- 1. 原子操做&普通锁&读写锁
- 2. 原子操做、互斥锁、读写锁
- 3. 【自制操做系统08】中断
- 4. 【自制操做系统13】锁
- 5. 操做系统概念-内核同步-原子操做
- 6. 操做系统核心原理-1.操做系统导论
- 7. 原子操做和锁
- 8. 锁----和 原子操做
- 9. 无锁-CAS原子操做
- 10. 操做系统-中断（1）中断