技术: 对比 Linux 下并发库

技术

对比 C++11并发库 和 Posix Pthread.

前面已经有两篇 非常详细 的文章说过了 pthread 和 c++并发库, 这里再次对其进行深入探究.
如果对c/c++线程库不太熟悉, 可以参考我以前写的总结:

当然你对进程有非常深入的理解那最好了, 因为本文是基于 linux平台 写的.

引子

到目前为止, 接触的线程部分, 已经说了 pthread , boost线程库 & c++11线程库 , 也就是说 讲了很多 并发编程 的内容. 但是大多是从使用场景上来说, 至于线程本身的原理却讲的比较少, 本文主要就是讲讲原理或者其他细枝末节的东西.

也可能在中间穿插一些我个人的开发经验.

正文

回顾Posix Thread

Posix部分, 权做复习和经验总结, 快速的把以前学过的带一遍.

调度策略

调度策略有三种:

  • SCHED_OTHER：非实时、正常
  • SCHED_RR：实时、轮询法
  • SCHED_FIFO：实时、先入先出

例子:

1
2
3
pthread_attr_t attr;
pthread_attr_init(&attr);
pthread_attr_setschedpolicy(&attr, SCHED_FIFO);//sched_policy

调度继承

线程A创建了线程B，则线程B的调度策略是与线程A的调度策略与线程B的继承策略有关的。如果线程B继承策略为PTHREAD_INHERIT_SCHED，则线程B的调度策略与线程A相同；如果线程B继承策略为PTHREAD_EXPLICIT_SCHE，则线程B的调度策略由attr决定。

1
pthread_attr_setinheritsched(&attr,PTHREAD_EXPLICIT_SCHED);

优先级

线程优先级支持两种设置方式, 一种是创建时设置(通过线程属性), 另一种是创建后动态设置.

1
pthread_attr_setschedparam(&attr, new_priority);

如果是动态设置, 则使用下列函数设置:

1
2
3
4
5
pthread_attr_getschedparam(&attr, &schedparam);
schedparam.sched_priority = new_priority;
pthread_attr_setschedparam(&attr, &schedparam);
//或者
pthread_setschedparam(pthrid, sched_policy, &schedparam);

脱离同步

如果设置了detachstate状态，则pthread_join()会失效，线程会自动释放所占用的资源。
线程的缺省状态为PHREAD_CREATE_JOINABLE状态，线程运行起来后，一旦被设置为PTHREAD_CREATE_DETACH状态，则无法再恢复到joinable状态。

1
pthread_attr_setdetachstate (&attr,PTHREAD_CREATE_DETACHED);

取消线程

有一个线程在使用select监控网口，主控线程此时接到了用户的通知，要放弃监听，则主控线程会向监听线程发送取消请求。 Linux的pthread线程接收到取消请求时，并不会立刻终止线程，而是要等到取消点时才会结束任务。这样我们可以为取消点建立某些特殊的处理(有专门的hook函数)。Select是取消点，所以可以退出。(一般系统调用都是取消点)

发送终止信号给thread线程, 如果成功则返回0, 否则为非0值.发送成功并不意味着thread会终止.

1
int pthread_cancel(pthread_t thread)

1
2
//old_state如果不为NULL则存入原来的Cancel状态以便恢复
int pthread_setcancelstate(int state, int *oldstate)

设置本线程对Cancel信号的反应，state有两种值:

  • PTHREAD_CANCEL_ENABLE（缺省）
  • PTHREAD_CANCEL_DISABLE

分别表示收到信号后设为CANCLED状态和忽略CANCEL信号继续运行.

设置本线程取消动作的执行时机: (仅当Cancel状态为Enable时有效)

1
2
//oldtype如果不为NULL则存入原来的取消动作类型值
int pthread_setcanceltype(int type, int *oldtype)

type由两种取值:

  • PTHREAD_CANCEL_DEFFERED //下一个取消点再推出
  • PTHREAD_CANCEL_ASYCHRONOUS //立即执行(退出)

取消点

Pthread规定了取消点的概念。不论线程何时收到取消请求，都只能在取消点上才能取消线程。这就保证了风险的可控。
Pthread标准指定了以下几个取消点：

  • pthread_testcancel
  • 所有调度点, 如pthread_cond_wait、sigwait、select、sleep等

检查本线程是否处于Canceld状态:

1
2
//如果是, 则进行取消动作, 否则直接返回
void pthread_testcancel(void);

根据POSIX标准，read()、write()等会引起阻塞的系统调用都是Cancelation-point，而其他pthread函数都不会引起Cancelation动作。但是pthread_cancel的手册页声称，由于LinuxThread库与C库结合得不好，因而目前C库函数都不是Cancelation-point，因此可以在需要作为Cancelation-point的系统调用前后调用pthread_testcancel()，从而达到POSIX标准所要求的目标，即如下代码段:

1
2
3
pthread_testcancel();
retcode = read(fd, buffer, length);
pthread_testcancel();

但是这段代码, 比较尴尬的是, read是一个阻塞函数, 如果没有读满字节或者没有数据可读, 那么read阻塞, 没有执行至取消点的必然路径, 则线程无法由外部其他线程的取消请求而终止.

静态初始化

对于mutex, cond等存在静态初始化
例如:

1
2
pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;
pthread_cond_t cond =   PTHREAD_COND_INITIALIZER;

初始化一次

1
2
pthread_once_t  once_control = PTHREAD_ONCE_INIT;
int pthread_once(pthread_once_t *once_control, void (*init_routine) (void))

once_control一般取值 PTHREAD_ONCE_INIT, 该变量保证了init_routine()函数在本进程执行序列中仅执行一次, 但具体是哪个线程执行则不确定. 例如下面代码, 虽然两个线程都有调用 once_run() 函数, 但实际上, 它仅执行一次:

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

#include <stdio.h>
#include <pthread.h>
pthread_once_t  once=PTHREAD_ONCE_INIT;
void once_run(void)
{
	printf("once_run in thread %d\n",pthread_self());
}
void * child1(void *arg)
{
	int tid=pthread_self();
	printf("thread %d enter\n",tid);
	pthread_once(&once,once_run);
	printf("thread %d returns\n",tid);
}
void * child2(void *arg)
{
	int tid=pthread_self();
	printf("thread %d enter\n",tid);
	pthread_once(&once,once_run);
	printf("thread %d returns\n",tid);
}
int main(void)
{
	int tid1,tid2;
	printf("hello\n");
	pthread_create(&tid1,NULL,child1,NULL);
	pthread_create(&tid2,NULL,child2,NULL);
	sleep(10);
	printf("main thread exit\n");
	return 0;
}

具体内部实现, 应该是使用了互斥锁和条件变量保证由pthread_once()指定的函数执行且仅执行一次, once_control变量flag来代表是否执行过了, 执行过了pthread_once()会立即返回0, 而不会去执行指定的函数run_once.
如果once_control的初值不是 PTHREAD_ONCE_INIT（LinuxThreads定义为0），pthread_once()的行为就会不正常, 该方面 Linux Pthread和NPTL实现不统一. 所以最好还是初始化为 PTHREAD_ONCE_INIT .

线程清理

不论是可预见的线程终止还是异常终止，都会存在资源释放的问题，在不考虑因运行出错而退出的前提下，如何保证线程终止时能顺利的释放掉自己所占用的资源，特别是锁资源，就是一个必须考虑解决的问题。最经常出现的情形是资源独占锁的使用：线程为了访问临界资源而为其加上锁，但在访问过程中被外界取消，如果线程处于响应取消状态，且采用异步方式响应，或者在打开独占锁以前的运行路径上存在取消点，则该临界资源将永远处于锁定状态得不到释放。外界取消操作是不可预见的，因此的确需要一个机制来简化用于资源释放的编程。

pthread_cleanup_push() 之后, 如果出现了终止, 例如调用pthread_exit()和取消点终止, 都将执行pthread_cleanup_push()入栈的清理函数; 如果是正常结束呢? 那就手动调用void pthread_cleanup_pop(int execute).

一般流程是这样的: (push, pop总是成对出现的)

1
2
3
4
5
pthread_cleanup_push(pthread_mutex_unlock, (void *) &mut);
pthread_mutex_lock(&mut);
/* do some work */
pthread_mutex_unlock(&mut);
pthread_cleanup_pop(0);

但是这样做, 有很大的问题, 如果线程处于PTHREAD_CANCEL_ASYNCHRONOUS状态, 在你的清理函数中有放锁操作时, 试想CANCEL事件有可能在pthread_cleanup_push()和pthread_mutex_lock()之间发生, 或者pthread_mutex_unlock()和pthread_cleanup_pop()之间发生, 从而导致清理函数unlock一个并没有加锁的mutex变量. 所以一般使用前, 要把立即退出, 改换成PTHREAD_CANCEL_DEFERRED 即下一个取消点退出, 这样可以保证资源正确的释放:

1
2
3
4
5
6
7
8
9
10
{
 int oldtype;
 pthread_setcanceltype(PTHREAD_CANCEL_DEFERRED, &oldtype);
 
 pthread_cleanup_push(routine, arg);
 //...
 pthread_cleanup_pop(execute);
 
 pthread_setcanceltype(oldtype, NULL);
}

这里注意一下, 各个编译器对于return的处理可能有一点儿区别, 在线程宿主函数中主动调用return，如果return语句包含在pthread_cleanup_push()/pthread_cleanup_pop()对中，则不会引起清理函数的执行，反而会导致segment fault。

线程回收

进程中各个线程的运行都是相互独立的，线程的终止并不会通知，也不会影响其他线程，终止的线程所占用的资源也并不会随着线程的终止而得到释放。正如进程之间可以用wait()系统调用来同步终止并释放资源一样，线程之间也有类似机制，那就是pthread_join()函数。

1
2
3
4
//线程是Jinable的话
void pthread_exit(void *retval)
int pthread_join(pthread_t th, void **thread_return)
int pthread_detach(pthread_t th)

线程将处于DETACHED状态, 在结束运行时自行释放所占用的内存资源，同时也无法由pthread_join()同步，pthread_detach()执行之后，对th请求pthread_join()将返回错误(-1)。

TSD

在多线程环境下，由于数据空间是共享的，因此全局变量也为所有线程所共有。但有时应用程序设计中有必要提供线程私有的全局变量. 仅在某个线程中有效，但却可以跨多个函数访问，比如程序可能需要每个线程维护一个链表，而使用相同的函数操作，最简单的办法就是使用同名而不同变量地址的线程相关数据结构。这样的数据结构可以由Posix线程库维护，称为线程私有数据（Thread-specific Data，或TSD）.

(其他语言一般称为 ThreadLocal)

实现机制是采用了二级表, 全局变量名一样, 但是各个线程的值不同. TSD池用一个结构数组表示:

1
2
//key-value结构, 所有线程共享的全局数组
static struct pthread_key_struct pthread_keys[PTHREAD_KEYS_MAX] = { { 0, NULL } };

不论哪个线程调用pthread_key_create()，所创建的key都是所有线程可访问的，但各个线程可根据自己的需要往key中填入不同的值，这就相当于提供了一个同名而不同值的全局变量。
具体的创建函数如下:

1
int pthread_key_create(pthread_key_t *key, void (*destr_function) (void *))

该函数从TSD池中分配一项，将其值赋给key供以后访问使用。如果destr_function不为空，在线程退出(pthread_exit())时将以key所关联的数据为参数调用destr_function()，以释放分配的缓冲区。创建一个TSD就相当于将结构数组中的某一项设置为”in_use”，并将其索引返回给*key. key保留了索引, 所以每个线程可以根据自己的情况,在相关索引里存入自己的值, 同时维护一张自己的表. key是全局的.

但是注意, 这个函数, 仅仅是表示某个线程对于某个key是 “正在使用”, 真正设置值, 是在下面的函数里:

1
2
3
4
5
//设置
int  pthread_setspecific(pthread_key_t  key,  const void  *pointer)

//获取
void * pthread_getspecific(pthread_key_t key)

写入 pthread_setspecific() 时, 将pointer指针值(不是所指的内容)与key相关联, 而相应的get函数则将与key相关联的数据读出来. 数据类型都设为 void * 因此可以指向任何类型的数据.

一般实现中, 采用二维 void * 指针数组来存放与key关联的数据, 也就是各个线程自己维护的这张表, 每个线程能有多少个key存储呢? 那要看数组大小.

数组大小由以下几个宏来说明:

1
2
3
#define PTHREAD_KEY_2NDLEVEL_SIZE       32
#define PTHREAD_KEY_1STLEVEL_SIZE   \
  ((PTHREAD_KEYS_MAX + PTHREAD_KEY_2NDLEVEL_SIZE - 1)/ PTHREAD_KEY_2NDLEVEL_SIZE)

其中在 /usr/include/bits/local_lim.h 中定义了 PTHREAD_KEYS_MAX为1024, 因此一维数组大小为32.
而具体存放的位置由key值经过以下计算得到: idx1st = key / PTHREAD_KEY_2NDLEVEL_SIZEidx2nd = key % PTHREAD_KEY_2NDLEVEL_SIZE, 说明第二维度(value长度, 也为32), 也就是说, 数据存放在一个32×32的稀疏矩阵中. 同样, 访问的时候也由key值经过类似计算得到数据所在位置索引, 再取出其中内容返回.

上面说了, 真正调用destr_function去释放key所对应的全局空间的是 pthread_exit() 线程退出时, 那么下面的delete函数呢?

1
int pthread_key_delete(pthread_key_t key)

这个函数并不检查当前是否有线程正使用该TSD，也不会调用清理函数（destr_function），而只是将TSD释放以供下一次调用pthread_key_create()使用。 也就是说, 把所有线程共享的全局数组的某个key所对应的索引位置置空NULL, 好在重用.

放一个简单的案例吧:

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
#include <stdio.h>
#include <pthread.h>

pthread_key_t key;//key是全局的

void echomsg(int t)
{
	printf("destructor excuted in thread %d,param=%d\n",
	       pthread_self(),t);
}
void * child1(void *arg)
{
	int tid=pthread_self();
	printf("thread %d enter\n",tid);
	pthread_setspecific(key,(void *)tid);
	sleep(2);
	printf("thread %d returns %d\n",tid,pthread_getspecific(key));
	sleep(5);
}
void * child2(void *arg)
{
	int tid=pthread_self();
	printf("thread %d enter\n",tid);
	pthread_setspecific(key,(void *)tid);
	sleep(1);
	printf("thread %d returns %d\n",tid,pthread_getspecific(key));
	sleep(5);
}
int main(void)
{
	int tid1,tid2;
	printf("hello\n");
	pthread_key_create(&key,echomsg);
	pthread_create(&tid1,NULL,child1,NULL);
	pthread_create(&tid2,NULL,child2,NULL);
	sleep(10);
	pthread_key_delete(key);
	printf("main thread exit\n");
	return 0;
}

还是要强调一下, 只有线程退出的时候, 采取主动调用那个pthread_key_create() 指定的清理函数.

同步，pthread_detach

同步里面 mutex 和 condition 用的最多, 其次是稍稍强大的 sem 信号量, 读写锁则是特别情况才用. 同步的内容比较多, 还是参考我的博客 posix-thread 吧.

(主要弄熟 posix thread 中的 mutex, detach)

回顾C++11并发库

技术细节探讨

尾巴

本文标题:技术: 对比 Linux 下并发库

文章作者:Merlin

发布时间:2016-07-04, 10:16:26

最后更新:2018-04-12, 14:55:17

原始链接:http://www.merlinblog.site/posts/b84fe139/

许可协议: "署名-非商用-相同方式共享 4.0" 转载请保留原文链接及作者。

文章目录
  1. 1. 引子
  2. 2. 正文
    1. 2.1. 回顾Posix Thread
      1. 2.1.1. 调度策略
    2. 2.2. 调度继承
      1. 2.2.1. 优先级
      2. 2.2.2. 脱离同步
      3. 2.2.3. 取消线程
      4. 2.2.4. 取消点
      5. 2.2.5. 静态初始化
      6. 2.2.6. 初始化一次
      7. 2.2.7. 线程清理
      8. 2.2.8. 线程回收
      9. 2.2.9. TSD
      10. 2.2.10. 同步，pthread_detach
    3. 2.3. 回顾C++11并发库
    4. 2.4. 技术细节探讨
  3. 3. 尾巴
|