操作系统笔记

并发控制

同步

定义：使用原子操作来确保线程之间的协作。

信号量

定义：信号量具有非负整数值，并支持以下两种操作:

• Down() or P()：等待信号量变为正值，然后将其减 1 的原子操作
  • 将它视为 wait() 操作
• Up() or V()：将信号量加 1 的操作，唤醒等待中的 P（如果有）
  • 这就是 signal() 操作

两种使用方式

互斥（初始值=1）

• 也称为“二元信号量”或“mutex”
• 可以用于互斥，就像lock

例子：

semaP(&mysem);
	//Critical section goes here
semaV(&mysem);
1
2
3

调度限制（初始值=0）

• 让线程1等待线程2的信号
  • 当给定事件发生时线程2调度线程1

例子：

Initial value of semaphore = 0
ThreadJoin {
	semaP(&mysem);
}
ThreadFinish {
	semaV(&mysem);
}
1
2
3
4
5
6
7

高层次的同步实现

硬件原语：只有load和store为原子操作

问题：假设宿舍冰箱里的牛奶没了，无论哪个舍友看到了都会出去买，但只需要一个人买即可，买多了不合适，如何通过合适的同步控制来实现？

方法一

• 使用便条避免购买过多牛奶
  • 在出门买牛奶前留下便条（lock）
  • 买完回来后将便条拿走（unlock）
  • 如果看到有便条，就不要出门买牛奶（wait）

if (noMilk) {
	if (noNote) {
		leave Note;
		buy milk;
		remove note;
	}
}
1
2
3
4
5
6
7

问题：存在某些调度方式使得仍然会购买过多牛奶

方法二

给便条打标签。

# Thread A
leave note A;
if (noNote B) {
	if (noMilk) {
		buy milk;
	}
}
remove note A;

# Thread B
leave note B;
if (noNote A) {
	if (noMilk) {
		buy milk;
	}
}
remove note B;
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

问题：如果先调度A留下便条，再调度B留下便条，那么两个线程都会以为对方去买牛奶了，导致最后没人买牛奶。

方法三

# Thread A
leave note A;
while (note B) {}
if (noMilk) {
	buy milk;
}
remove note A;

# Thread B
leave note B;
if (noNote A) {
	if (noMilk) {
		buy milk;
	}
}
remove note B;
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

可行。这个方法还可以扩展到 $n$ 个线程，详情参考Lamports的论文。
问题：

• 非常复杂。即使是这样一个简单的例子，你也很难说服自己相信这个方法可行。
• 两个线程的代码不一致。当线程数增加后，每个线程的代码都会有所不同。
• 当A在等待时，仍然占用CPU。（busy-waiting）

多原子操作实现

方法一：在指令级别实现lock和unlock操作

问题：被锁的范围如何确定？

方法二：使用硬件中断实现锁

如果在执行一组原子操作前，将处理器的中断关闭，那么就不会受到时钟中断的影响，从而不会中途被切换到其它线程。

LockAcquire { disable Ints; }
LockRelease { enable Ints; }
1
2

问题：

• 不能让用户这么做。否则会使机器崩溃。
• 实时系统的功能无法得到保证。
• 当重要事件发生时，系统无法及时处理。

方法三：通过关闭中断实现锁

• 关键思想：维护一个锁并仅在对该锁的操作期间施加互斥

int value = FREE;

Acquire() {
	disable interrupts;
	if (value == BUSY) {
		put thread on wait queue;
		Go to sleep();
		//Enable interrupts?
	} else {
		value = BUSY;
	}
	enable interrupts;
}

Release() {
	disable interrupts;
	if (anyone on wait queue) {
		take thread off wait queue;
		Place on ready queue;
	} else {
		value = FREE;
	}
	enable interrupts;
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

这个方法还有一个问题，那就是被放入等待队列的线程，何时恢复中断？
如果在放入等待队列前或进入睡眠前恢复中断，很有可能切换到另一个线程，那个线程将value改为了FREE，然后我们的线程还傻乎乎地睡眠了。
所以必须在睡眠后将中断打开。

要解决这个问题，我们回想一下线程调度的过程。我们知道，线程执行 sleep() 系统调用后，变为睡眠状态，然后CPU执行调度程序，此时由于处于系统调用中，系统也会关闭所有中断，而当新线程准备执行前，系统会自动恢复中断。

因此，这里的情况也是一样的，我们的线程关闭中断，进入睡眠后，由被调度执行的线程恢复中断。

我们下面看一个例子来更好的理解这种锁是如何工作的。
在下图中，有两个线程A和B,其中A正在执行，B处于就绪队列中，随时可以被CPU调度执行。这里的锁变量为value，虽然图上记录了它的当前持有者（owner），但这里只是为了方便演示，实际上没有这种显式记录。

在下图中，A执行Acquire()尝试获得锁，它首先将中断关闭。在后面的图中，中断关闭由红点标识，当中断重新恢复后，由绿点标识。

此时value为0，因此A“获得了锁”，将value改为1。值得注意的是，在这个过程中，由于中断被关闭，因此时钟中断无效，CPU不会执行其它线程，这就保证了锁操作的原子性。

最后A重新打开中断，结束Acquire()操作，执行临界区的代码。




中途发生时钟中断，于是操作系统调度就绪队列中的线程B。这里要注意的是，执行系统调用时硬件会自动关闭中断。

完成调度，线程B恢复执行后，硬件又自动恢复中断。

B也尝试获取锁。

由于此时value为1，因此B进入等待队列，并睡眠。




B睡眠后，系统执行调度程序，恢复线程A的执行。值得注意的是，B是在关闭中断后进入睡眠的，之后由于执行了线程切换，在恢复A的上下文后，硬件自动恢复中断。

A继续执行临界区代码，最后执行Release()来“释放锁”。

释放锁时，检查等待队列，选择其中一个线程放入就绪队列。

B进入就绪队列。

A恢复中断，结束Release()。

A继续执行后面的代码，直到发生时钟中断，系统恢复B的执行。

B从睡眠中返回并恢复中断。

由于B是被选择唤醒的线程，因此B实际上就已经持有锁了。B结束Acquire()后开始执行它的临界区代码。再次说明，这里的owner仅仅是为了演示，线程持有者是隐式记录的，因为只有线程B从等待队列中恢复执行。

问题：

• 不能在用户级实现该锁
• 不适合多核/多处理器系统

方法四：使用 Read-Modify-Write 原子指令

示例：

test&set(&address) {
	result = M[address];
	M[address] = 1;
	return result;
}

swap(&address, register) {
	temp = M[address];
	M[address] = register;
	register = temp;
}

// 我们常说的CAS
compare&swap(&address, reg1, reg2) {
	if (reg1 == M[address]) {
		M[address] = reg2;
		return success;
	} else {
		return failure;
	}
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

方法一：test&set

int mylock = 0;

acquire(int* thelock) {
	while (test&set(thelock));
}

release(int* thelock) {
	*thelock = 0;
}
1
2
3
4
5
6
7
8
9

• 如果 lock 空闲，test&set 读取 0 并设置 lock=1，因此 lock 现在被持有。它在退出时返回 0。
• 如果 lock 已被持有，test&set 读取1并设置 lock=1（没有变化）返回1，所以while循环继续。
• 当我们设置 thelock = 0 时，其他人可以获得锁。

问题：Busy-Waiting，并且在多核缓存中发生乒乓现象（不停更新缓存）。

方法二：test&test&set

int mylock = 0;
acquire(int* thelock) {
	do {
		while(*thelock);         
	} while(test&set(thelock)); 
}
release(int* thelock) {
	*thelock = 0;		 
}
1
2
3
4
5
6
7
8
9

大部分情况下只会读而没有写，因此解决了缓存乒乓的问题。并且读取的是缓存，效率比读内存高。

问题：Busy-Waiting

方法三：只在检查锁的值时 Busy-Waiting

这种方法的实现和前面使用中断实现的锁很像，唯一不同仅在于这种方法不涉及系统调用。

int guard = 0;  //全局变量
int mylock = FREE;

acquire(int* thelock) {
	while (test&set(&guard));  //Busy-Wait的时间很短，因为检查锁变量的值耗费很少时间
	if (*thelock == BUSY) {
		put thread on wait queue;
		go to sleep() & guard = 0;
	} else {
		*thelock = BUSY;
		guard = 0;
	}
}

release(int* thelock) {
	while (test&set(guard));
	if anyone on wait queue {
		take thread off wait queue
		Place on ready queue;
	} else {
		*thelock = FREE;
	}
	guard = 0;
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

尝试1

上面的代码仍然比较抽象，因为我们在用户程序中无法将线程放入队列以及睡眠。下面我们看看如何借助linux的futex来实现上面的方法：

#include  
#include  
int futex(int *uaddr, int futex_op, int val, const struct timespec *timeout);

// uaddr: points to a 32-bit value in user space
// futex_op:
// FUTEX_WAIT – if val == *uaddr sleep till FUTEX_WAIT
// Atomic check that condition still holds after we disable interrupts (in kernel!)
// FUTEX_WAKE – wake up at most val waiting threads
// FUTEX_FD, FUTEX_WAKE_OP, FUTEX_CMP_REQUEUE: More interesting operations!
// timeout: ptr to a timespec structure that specifies a timeout for the op
1
2
3
4
5
6
7
8
9
10
11

int mylock = 0;

acquire(int* thelock) {
	while (test&set(thelock)) {
		futex(thelock, FUTEX_WAIT, 1);
	}
}

release(int* thelock) {
	*thelock = 0;
	futex(thelock, FUTEX_WAKE, 1);
}
1
2
3
4
5
6
7
8
9
10
11
12

为什么在acquire中不用if而是while？因为被唤醒的线程需要重新获取锁，而在此之前有可能出现其它线程抢先一步获取锁，使得该线程又得进入睡眠。

优点：无Busy-Waiting
问题：释放锁时，即使等待队列为空，也会执行系统调用，在无竞争的情况下并不能完全不陷入内核

尝试2

bool maybe = false;
int mylock = 0;

acquire(int* thelock, bool* maybe) {
	while (test&set(thelock)) {
		// Sleep, since lock busy!
		*maybe = true;
		futex(thelock, FUTEX_WAIT, 1);

		// Make sure other sleepers not stuck
		*maybe = true;
	}
}

release(int* thelock, bool *maybe) {
	*thelock = 0;
	if (*maybe) {
		*maybe = false;
		// Try to wake up someone
		futex(&value, FUTEX_WAKE, 1);
	}
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

这种方法能够在无竞争的情况下完全不涉及系统调用。不过它还能进一步优化，这里不做扩展，详情请参考Ulrich Drepper的“Futexes are Tricky”。

尝试3

使用其它原子操作实现：

typedef enum { 
	UNLOCKED,
	LOCKED,
	CONTESTED 
} Lock;
Lock mylock = UNLOCKED; 

acquire(Lock* thelock) {
	// If unlocked, grab lock!
	if (compare&swap(thelock, UNLOCKED, LOCKED))
		return;

	// Keep trying to grab lock, sleep in futex
	while (swap(mylock, CONTESTED) != UNLOCKED))
		// Sleep unless someone releases hear!
		futex(thelock, FUTEX_WAIT, CONTESTED);
}

release(Lock *thelock) {
	// If someone sleeping
	if (swap(thelock,UNLOCKED) == CONTESTED)
		futex(thelock,FUTEX_WAKE,1);
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

• 接口简洁
• 在无竞争的情况下无开销
• 可以通过类似的方法实现信号量

管程 (Monitor)

信号量的问题在于它有两个用途：

• 它们用于互斥锁和调度约束
• 在有界缓冲区中翻转 P 操作并不是很明显

更清晰的想法：用锁进行互斥控制，用条件变量调度约束

Monitor: 一个锁以及零个或多个条件变量管理对共享数据的并发访问

• 一些语言（如 Java）本身就提供了
• 大多数其他的使用实际的锁和条件变量

Monitor 是并发编程的范式。

条件变量

条件变量：等待临界区内某事的线程队列

• 关键思想：通过在我们进入睡眠时自动释放锁来允许在临界区内睡眠
• 信号量与此相反：不能在临界区中睡眠

操作：

• Wait(&lock)：自动释放锁并进入睡眠状态。稍后在运行之前重新获取锁
• Signal()：唤醒一位等待者，如果有的话
• Broadcast()：唤醒所有等待者

规则：执行条件变量操作时必须持有锁

示例

lock buf_lock = <initially unlocked>
condition producer_CV = <initially empty>
condition consumer_CV = <initially empty>

Producer(item) {
  acquire(&buf_lock);
  while (buffer.full()) { cond_wait(&producer_CV, &buf_lock); }
  enqueue(item);
  cond_signal(&consumer_CV);
  release(&buf_lock);
}

Consumer() {
  acquire(buf_lock);
  while (buffer.empty()) { cond_wait(&consumer_CV, &buf_lock); }
  item = dequeue();
  cond_signal(&producer_CV);
  release(buf_lock);
  return item
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

为什么要有 while 呢？因为等待线程被唤醒后只是进入了就绪队列，并不意味着立刻被调度，有可能前面的人先把蛋糕吃了，后面的人再从 cond_wait 中返回，因此需要再对缓冲区检查。

Basic Readers/Writers Solution

• Readers 只能在没有 Writers 时访问数据库
• Writers 只能在没有人时访问数据库

状态变量 (由名为 “lock” 的锁进行保护):

• int AR: Number of active readers; initially = 0
• int WR: Number of waiting readers; initially = 0
• int AW: Number of active writers; initially = 0
• int WW: Number of waiting writers; initially = 0
• Condition okToRead = NIL
• Condition okToWrite = NIL

Reader() {
	// First check self into system
	acquire(&lock);
	while ((AW + WW) > 0) {	// Is it safe to read?
		WR++;	            // No. Writers exist
		cond_wait(&okToRead, &lock);// Sleep on cond var
		WR--;	            // No longer waiting
	}
	AR++;		            // Now we are active!
	release(&lock);
	// Perform actual read-only access
	AccessDatabase(ReadOnly);
	// Now, check out of system
	acquire(&lock);
	AR--;		            // No longer active
	if (AR == 0 && WW > 0)	// No other active readers
		cond_signal(&okToWrite);// Wake up one writer
	release(&lock);
}

Writer() {
	// First check self into system
	acquire(&lock);
	while ((AW + AR) > 0) {	// Is it safe to write?
		WW++;	            // No. Active users exist
		cond_wait(&okToWrite, &lock);	// Sleep on cond var
		WW--;	            // No longer waiting
	}
	AW++;		            // Now we are active!
	release(&lock);
	// Perform actual read/write access
	AccessDatabase(ReadWrite);
	// Now, check out of system
	acquire(&lock);
	AW--;		            // No longer active
	if (WW > 0){	        // Give priority to writers
		cond_signal(&okToWrite);// Wake up one writer
	} else if (WR > 0) {	// Otherwise, wake reader
		cond_broadcast(&okToRead); // Wake all readers
	}	
	release(&lock);
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42

饥饿与死锁

饥饿：线程间的不确定性等待，例如低优先级线程等待高优先级线程的资源
死锁：持有资源的线程互相等待其他线程的资源
饥饿随着时间推移总能解决，但死锁不是。

发生死锁的四个必要条件

• 互斥性：一个资源在某一时刻只能被一个线程使用
• 持有并等待：持有至少一个资源的线程等待获取其他线程持有的资源
• 非争夺：资源只能由持有它的线程主动释放
• 环形等待：$T_1\to T_2\to ...\to T_n\to T_1$

死锁解决的四个办法：

• 死锁预防：编写不容易发生死锁的代码
• 死锁恢复：允许死锁发生，并思考出如何恢复
• 死锁避免：动态延迟资源请求，让死锁不会发生（银行家算法）
• 死锁无视：忽略死锁的可能性

死锁检测算法

[Avail] = [FreeResources]
Add all threads to UNFINISHED
do {
	done = true
	for thread in UNFINISHED {
		if [thread.request] <= [Avail] {
			UNFINISHED.remove(thread)
			[Avail] = [Avail] + [thread.alloc]
			done = false
		}
	}
} while (!done)
1
2
3
4
5
6
7
8
9
10
11
12

当算法结束时，若UNFINISHED不为空，则会发生死锁。