前言
前面讲了线程的实现跟原理，这篇文章将会把没有完成的内容给梳理一遍，包括通道通信，信号量和 2 个并行算法。这一章的内容需要涉及到上一章节的内容，如果有不清楚的读者可以翻阅上一章
信号量
信号量是底层同步原语，这样的说法比较学院派，读者的理解就是底层用于同步线程的工具。理论上信号量是一个受保护的整数，可以理解为一个计数器，它可以原子地加 1 或减 1(也就是信号量的VP操作)。当计数器为 0 时，获取信号量的线程将会阻塞。
其实现原理也比较简单，每个信号量都使用一个结构体来表达，而结构体中维护一个队列。所以每个信号量都有一个队列。当线程使用信号量进行同步时，当前线程会被挂在这个队列下从而被阻塞运行，直到在适当的实际，这个队列里的线程会被加入 ready 队列再进行执行。
其结构体如下：
typedef struct Sem_T {
int count;
void *queue;
} Sem_T;
创建信号量
信号量的创建代码如下：
void Sem_init(Sem_T *s, int count) {
assert(current);
assert(s);
s->count = count;
s->queue = NULL;
}
Sem_T *Sem_new(int count) {
Sem_T *s;
NEW(s);
Sem_init(s, count);
return s;
}
代码比较简单，分配空间，然后对计数器进行赋值，并将队列置为NULL。
对同一个信号量多次调用 Sem_init 函数是属于不能检查的错误。
信号量VP操作
信号量减 1
void Sem_wait(Sem_T *s) {
assert(current);
assert(s);
testalert();
if (s->count <= 0) {
put(current, (Thread_T *)&s->queue);
run();
testalert();
} else
--s->count;
}
信号量减 1 操作比较简单，如下：
- 做参数检查
- 接着执行
testalert检查当前线程的alerted标志。 - 判断计数器的值
- 如果计数器小于等于 0 ，则说明当前信号量已经用完，将当前线程加入信号量的队列等待，在适当的时机返回执行
- 如果计数器大于 0 ，则将计数器减 1 并返回
通常来说，Sem_wait 不能在 Thread_init 前调用，这是一个可检查的错误。
使用信号量会使线程阻塞，当线程在阻塞期间被设置了 alerted 标志，那么线程会被停止阻塞进入 ready 队列，并且返回时会引发Thread_Alerted异常，而不会将计数器减 1 。
信号量加 1
void Sem_signal(Sem_T *s) {
assert(current);
assert(s);
if (s->count == 0 && !isempty(s->queue)) {
Thread_T t = get((Thread_T *)&s->queue);
assert(!t->alerted);
put(t, &ready);
} else
++s->count;
}
- 参数检查
- 判断信号量是否为 1 ，同时也判断信号量的队列是否为空。
- 如果队列不为空且计数为 0 ，说明有线程在等待信号量，将信号量队列中的线程出入并放入
ready队列等在执行 - 否则将计数器加 1
饥饿
从上述来看，信号量的操作不难理解。但信号量的获取涉及到一个饥饿的问题，也就是如果程序设计得不当，会有线程无法获得信号量，从而一直被阻塞。
我们将信号量的 PV操作 换成下面的代码
void Sem_wait(Sem_T *s)
{
while(s->count <= 0)
{
put(current, &s->queue);
run();
}
--s->count;
}
void Sem_signal(Sem_T *s)
{
if(++s->count > 0 && !isempty(s->queue))
put(get(&s->queue), &ready);
}
假定有 A 和 B 两个线程，且有一个信号量 s 初始化为二值信号量，A 和 B 分别执行下述代码
for（;;）
{
Sem_wait(s);
...//临界区
Sem_signal(s);
}
假定 A 正在临界区执行代码，那么 B 会处于 信号量s 的队列中等在。当 A 调用了Sem-signal 时，线程 B 移动到 ready 队列。此时 B 在Sem_wait 中的 run 返回。在 B 进行下一个 while 判断前，因为调度时间到了线程将转移给 A ，此时 A 比 B更快地再次获取到了信号量，那么 B 又会被放入 信号量s 的队列。这就造成了 饥饿
如果Sem_wait不使用while循环，且Sem_signal不会一直对 信号量s 一直加 1，那么就可以有效解决这个问题。
互斥锁宏
如果信号量被初始化为一个二值信号量，那么每次只有一个线程能够获取信号量，这个方法很常用，所以书中为它导出了一个宏
#define LOCK(mutex) do { Sem_T *_yymutex = &(mutex); \
Sem_wait(_yymutex);
#define END_LOCK Sem_signal(_yymutex); } while (0)
LOCK(mutex)
statements
END_LOCK
如果执行语句 statements 有可能引发异常，那么就不可以使用这个宏了，此时正确的用法如下
TRY
Sem_wait(&mutex)
statements
FINALLY
Sem_signal(&mutex)
END_TRY
通信通道
通信通道可用于线程之间传递数据，调用通信通道的线程会阻塞，直到对方线程接收完数据才会返回。当然，如果调用线程的 alerted 被设置，通道函数会立即引发 Thread_Alerted 异常。
如果是在阻塞期间被设置了 alerted 标志，那么线程将得到执行并引发Thread_Alerted 异常，这种情况下数据可能传输了，有可能为传输。
通信通道使用了信号量到进行临界资源的控制，其结构体如下
typedef struct Chan_T {
const void *ptr;//数据指针
int *size;//数据大小
Sem_T send; //发送信号量
Sem_T recv;//接收信号量
Sem_T sync;//同步信号量
}*Chan_T;
创建通道
Chan_T Chan_new(void) {
Chan_T c;
NEW(c);
Sem_init(&c->send, 1);
Sem_init(&c->recv, 0);
Sem_init(&c->sync, 0);
return c;
}
非常简单，直接常见 3 个信号量，并为它们赋值
发送数据
代码如下：
int Chan_send(Chan_T c, const void *ptr, int size) {
assert(c);
assert(ptr);
assert(size >= 0);
Sem_wait(&c->send);
c->ptr = ptr;
c->size = &size;
Sem_signal(&c->recv);
Sem_wait(&c->sync);
return size;
}
- 参数检查
- 获取通道信号量，确保此时通道不会被占用，如果被占用将进入等待
- 获取数据指针和数据大小
- 将接收信号量增 1 ，以通知对方可以获取接收信号量并开始接收数据
- 等待接收完成
- 返回接收大小
第 5 点需要注意，Sem_wait(&c->sync) 在没有线程接收的时候，他会一直处于阻塞状态，因为sync信号量一开始赋值为 0 。
接收数据
int Chan_receive(Chan_T c, void *ptr, int size) {
int n;
assert(c);
assert(ptr);
assert(size >= 0);
Sem_wait(&c->recv);
n = *c->size;
if (size < n)
n = size;
*c->size = n;
if (n > 0)
memcpy(ptr, c->ptr, n);
Sem_signal(&c->sync);
Sem_signal(&c->send);
return n;
}
- 参数检查
- 获取通道接收信号量，如果没有则一直阻塞等待
- 接收数据
- 释放同步信号量，让发送线程能够从阻塞中返回
- 释放发送信号量，以让通道能够进行下一次的数据传输
并行算法
书中对并行算法举了 3 个例子，这里笔者拿其中 2 个典型例子来讲述
并发排序
并发排序使用的是快速排序算法跟多线程集合起来，这里不对快排进行讲述，有需要的读者自行学习。
了解快排的读者应该知道，快排就是找到锚点后分将数据分为 2 边再继续进行排序，这样是算法很适合使用多线程的，我们为每一边的排序开一个线程去排序，因为两边的的排序互不影响。
江苏将直接写入代码中的注释
struct args {
int *a;
int lb, ub;
};
int cutoff = 10000;
//partition的作用各位了解快排的读者应该知道，就是找出排序后找出中间点
int partition(int a[], int i, int j) {
int v, k, t;
j++;
k = i;
v = a[k];
while (i < j) {
i++; while (a[i] < v && i < j) i++;
j--; while (a[j] > v ) j--;
if (i < j) { t = a[i]; a[i] = a[j]; a[j] = t; }
}
t = a[k]; a[k] = a[j]; a[j] = t;
return j;
}
int quick(void *cl) {
struct args *p = cl;
int lb = p->lb, ub = p->ub;
if (lb < ub) {
//k就是中间点，而lb和ub就是排序数组上下限
int k = partition(p->a, lb, ub);
p->lb = lb;
p->ub = k - 1;
if (k - lb > cutoff) {
Thread_T t;
//跟常规快排不同的是，我们这里并不是在函数里面使用递归
//而是重新开辟一个线程去执行小数值部分的排序，然后继续往下排序数值大的部分
t = Thread_new(quick, p, sizeof *p, NULL);
Fmt_print("thread %p sorted %d..%d\n", t, lb, k - 1);
} else
quick(p);
p->lb = k + 1;
p->ub = ub;
if (ub - k > cutoff) {
Thread_T t;
//同理,这里并没有继续递归,而是重新开辟线程
t = Thread_new(quick, p, sizeof *p, NULL);
Fmt_print("thread %p sorted %d..%d\n", t, k + 1, ub);
} else
quick(p);
}
return EXIT_SUCCESS;
}
void sort(int *x, int n, int argc, char *argv[]) {
struct args args;
if (argc >= 3)
cutoff = atoi(argv[2]);
//初始化要排序的参数
args.a = x;
args.lb = 0;
args.ub = n - 1;
quick(&args);//开始排序
Thread_join(NULL);//等待所有的线程退出，所有线程都退出意味着完成排序
}
main(int argc, char *argv[]) {
int i, n = 100000, *x, preempt;
//初始化线程系统
preempt = Thread_init(1, NULL);
assert(preempt == 1);
//输入数组大小
if (argc >= 2)
n = atoi(argv[1]);
//开辟数组空间并清空
x = CALLOC(n, sizeof (int));
//为数组产生随机数以便进行排序
srand(time(NULL));
for (i = 0; i < n; i++)
x[i] = rand();
//开始排序
sort(x, n, argc, argv);
//检查排序结果
for (i = 1; i < n; i++)
if (x[i] < x[i-1])
break;
assert(i == n);
//退出线程
Thread_exit(EXIT_SUCCESS);
return EXIT_SUCCESS;
}
生成素数
生成素数是指输入数字 n和last，找出小于等于 last 的 n 个素数。
这里使用了艾拉托逊斯筛法，大致可以这样理解：
我们将一个数字 A 通过一个筛子 Z，如果符合筛子的规则，那么 A 就通过。这里是筛子就是一条线程，将数字传入给线程，如果线程判断可以通过，那么数字 A 就进入下一步，为什么说是下一步，因为还有很多条规则要通过
这个算法的规则就是如果一个数不是在它之前所有素数的整数倍，那么这个数就是素数。举个例子，比如在 7 前面有素数 2、3 、5 ，你可以发现 7 并不是他们的整数倍，所以 7 是素数，而 9 是 3 的整数倍，那么 9 就不是素数。那么需要筛选到什么时候才算是素数呢。举个例子，按照我们前面所讲，11 面前的所有素数是 2、3、5、7，如果 11 不是 2、3、5、7的整数倍，那么11就是素数
代码的整体思路如下：
- 创建一个源头线程，这个线程专门向下一个筛子线程发送奇数数字。
- 创建第一个筛子线程，这个筛子线程会不断地根据我们所讲的规则生成一个素数筛子数组比如第一个筛子数组是[2, 3, 5, 7, 11]，后进入过滤函数。
- 创建新的管道和下一个筛子线程，这个筛子线程也会生成自己的素数筛子数组，然后通过新建的管道与下一个筛子线程通信。
- 在过滤函数中会不断地根据筛子数组进行取模，如果取模出来不为0，那么说明这个数符合有可能是素数，将这个数传给下一个筛子线程。
- 继续从 3 步骤 往下，直到找出 n 个素数或者超过 last
sink.jpg
代码如下：
struct args {
Chan_T c;
int n, last;
};
/*
source的作用是创建源头，因为所有的素数一定是奇数(除了2意外)
所以source的作用就是不断地通过通道c给第一个筛子线程发送奇数
*/
int source(void *cl) {
struct args *p = cl;
int i = 2;
if (Chan_send(p->c, &i, sizeof i))//先发送 2
for (i = 3; Chan_send(p->c, &i, sizeof i); )//再发送大于2的所有奇数
i += 2;
return EXIT_SUCCESS;
}
void filter(int primes[], Chan_T input, Chan_T output) {
int j, x;
for (;;) {
Chan_receive(input, &x, sizeof x);
for (j = 0; primes[j] != 0 && x%primes[j] != 0; j++)
;
if (primes[j] == 0)
if (Chan_send(output, &x, sizeof x) == 0)
break;
}
Chan_receive(input, &x, 0);
}
/*
sink线程从上一个筛子线程中接收数据，生成自己的筛子规则，接着就开辟一个线程根据不断接收上一个筛子的数据，并根据筛子规则来过滤数字
*/
int sink(void *cl) {
struct args *p = cl;
Chan_T input = p->c;
int i = 0, j, x, primes[256];
primes[0] = 0;
for (;;) {
Chan_receive(input, &x, sizeof x);
for (j = 0; primes[j] != 0 && x%primes[j] != 0; j++)
;
if (primes[j] == 0) {
if (x > p->last)
break;
//这里就是处理素数的地方，例子中的打印，我们也可以是用一个数组保存下来
Fmt_print(" %d", x);
primes[i++] = x;
primes[i] = 0;
if (i == p->n)
{
p->c = Chan_new();
Thread_new(sink, p, sizeof *p, NULL);
filter(primes, input, p->c);
return EXIT_SUCCESS;
}
}
}
Fmt_print("\n");
Chan_receive(input, &x, 0);
return EXIT_SUCCESS;
}
int main(int argc, char *argv[]) {
struct args args;
Thread_init(1, NULL);
args.c = Chan_new();//初始化通信通道
Thread_new(source, &args, sizeof args, NULL);//开辟source线程
args.n = argc > 2 ? atoi(argv[2]) : 5;//找出素数的个数
//查找的数字上线，也就是小于1000的所有素数
args.last = argc > 1 ? atoi(argv[1]) : 1000;
Thread_new(sink, &args, sizeof args, NULL);//开辟sink线程
Thread_exit(EXIT_SUCCESS);
return EXIT_SUCCESS;
}
后语
学习这一章节，花费了笔者不少时间跟精力。平时工作占据了大部分时间，下班跟假期也只能抽出一点时间来学习跟归纳。虽然有这样的背景，但笔者也是花费的时间确实比较长。侧面说明了笔者的基础达到理想地步。当然，在学习完之后对于线程跟同步的理解也有所加深。文章因为加入了不少笔者的理解，如果有错误的地方还请海涵和指正。
网友评论