木慕的博客

Stay naive, stay simple.

0%

POSIX thread

发表于 分类于 Valine：

POSIX thread

Portable operating system interface (POSIX) 是IEEE Computer Society为维持操作系统兼容性定义的一套标准。

POSIX 线程库规定了C/C++的线程API，相比于fork这种创造新进程的方式，使用线程开销更小。

  • 多处理器用多线程：效果最好
  • 单处理器使用多线程： 也有增益（一个线程等IO，另外的执行）

适用场景

  • 多任务同时进行
  • 阻止潜在的长时间IO等待
  • 一些地方很多CPU周期
  • 必须响应异步
  • 优先级中断

线程基础

线程共享模型

所有线程都可以访问相同的全局共享内存,但是也有自己的私有数据。

线程安全

由多个线程共享资源引起的，如果例程没有使用某种类型的同步结构来防止数据损坏，那么它就不是线程安全的。

线程限制

依赖于具体实现（我的WSLunlimit -Hu）：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
tigerroarm@Mudifan:~$ ulimit -a
core file size          (blocks, -c) 0
data seg size           (kbytes, -d) unlimited
scheduling priority             (-e) 0
file size               (blocks, -f) unlimited
pending signals                 (-i) 50633
max locked memory       (kbytes, -l) 64
max memory size         (kbytes, -m) unlimited
open files                      (-n) 1024
pipe size            (512 bytes, -p) 8
POSIX message queues     (bytes, -q) 819200
real-time priority              (-r) 0
stack size              (kbytes, -s) 8192
cpu time               (seconds, -t) unlimited
max user processes              (-u) 50633
virtual memory          (kbytes, -v) unlimited
file locks                      (-x) unlimited

线程共享如下资源

  • 进程指令（ Process instructions）
  • 大部分数据
  • 文件描述符
  • signals 和 signal handlers
  • 当前目录
  • User 和 group id

每个线程独立有：

  • Tread ID
  • registers, stack pointer集合
  • local variables, return addresses 栈
  • signal mask
  • priority
  • Return value: errno

pthread 如果OK会return 0

编译程序需要用：

gcc -pthread thread_program.c

创建线程

1
2
3
4
int pthread_create(pthread_t * thread, 
                      const pthread_attr_t * attr,
                      void * (*start_routine)(void *), 
                      void *arg);
  • thread : 返回的thread ID
  • attr: 使用默认thread 就NULL
  • start_routine: 需要线程运行的函数
  • arg: 函数参数

进程同步

thread 库提供了三种同步机制

  • 互斥（mutexes）:互斥锁（mutual exclusion lock）
  • joins: 等待其他进程完成
  • 条件变量: pthread_cond_t

mutex:

用pthread_mutex_lock和pthread_mutex_unlock加锁解锁

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

void *functionC();
pthread_mutex_t mutex1 = PTHREAD_MUTEX_INITIALIZER;
int  counter = 0;

main()
{
   int rc1, rc2;
   pthread_t thread1, thread2;

   /* Create independent threads each of which will execute functionC */

   if( (rc1=pthread_create( &thread1, NULL, &functionC, NULL)) )
   {
      printf("Thread creation failed: %d\n", rc1);
   }

   if( (rc2=pthread_create( &thread2, NULL, &functionC, NULL)) )
   {
      printf("Thread creation failed: %d\n", rc2);
   }

   /* Wait till threads are complete before main continues. Unless we  */
   /* wait we run the risk of executing an exit which will terminate   */
   /* the process and all threads before the threads have completed.   */

   pthread_join( thread1, NULL);
   pthread_join( thread2, NULL); 

   exit(0);
}

void *functionC()
{
   pthread_mutex_lock( &mutex1 );
   counter++;
   printf("Counter value: %d\n",counter);
   pthread_mutex_unlock( &mutex1 );
}

Joins

等待线程完成。

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

#define NTHREADS 10
void *thread_function(void *);
pthread_mutex_t mutex1 = PTHREAD_MUTEX_INITIALIZER;
int  counter = 0;

main()
{
   pthread_t thread_id[NTHREADS];
   int i, j;

   for(i=0; i < NTHREADS; i++)
   {
      pthread_create( &thread_id[i], NULL, thread_function, NULL );
   }

   for(j=0; j < NTHREADS; j++)
   {
      pthread_join( thread_id[j], NULL); 
   }

   /* Now that all threads are complete I can print the final result.     */
   /* Without the join I could be printing a value before all the threads */
   /* have been completed.                                                */

   printf("Final counter value: %d\n", counter);
}

void *thread_function(void *dummyPtr)
{
   printf("Thread number %ld\n", pthread_self());
   pthread_mutex_lock( &mutex1 );
   counter++;
   pthread_mutex_unlock( &mutex1 );
}

Condition Variables

为什么需要条件变量

使用互斥锁的时候，如果有一个下面的情况

  • 线程1: 如果有变化，就把本地buff save 到cloud上
  • 线程2：改本地buff，标记变化
1
2
3
4
5
6
7
8
9
//thread 1
while(1) {
    pthread_mutex_lock(lock); 
    if(is_change) {
        save_to_cloud(buff);
        is_change = false;
    }
    pthread_mutex_unlock(lock);
}
1
2
3
4
5
6
7
8
9
10
//thread 2
while(1) {
    char = tmp[50];
    read(tmp, 50);

    pthread_mutex_lock(lock);
    strcpy(buff, tmp);
    is_change = true;
    pthead_mutex_unlock(lock);
}

如果这时候is_change 一直是false， 那么thread1一直在执行循环。

想达到的目的是：直到某个状态改变，才开始执行

需要的是lock+condition

就需要有条件变量

互斥锁通过控制线程对数据的访问来实现同步，而条件变量允许线程根据数据的实际值进行同步。如果没有条件变量，程序员将需要线程持续轮询(可能在关键部分) ，以检查条件是否满足。这可能会非常消耗资源，因为线程在此活动中将持续忙碌。条件变量是一种不用轮询就可以实现相同目标的方法。

使用

  • 主线程：
    • 声明和初始化需要同步的全局数据/变量(例如“ count”)
    • 声明和初始化条件变量对象
    • 声明并初始化一个的互斥锁
    • 创建线程 a 和 b 来做工作
  • 线程A
    • 一直工作到某个条件必须发生的点(例如“ count”必须达到指定的值)
    • 锁住互斥锁，然后检查值
    • pthread_cond_wait() ，阻塞等待来自B线程信号（自动解锁相关的互斥锁）
    • 当收到信号时就唤醒进程
  • 线程B
    • 锁住互斥锁
    • 更改线程A正在等待的全局变量的值
    • 检查全局变量的值，如果满足条件，给A发信号
    • 解锁互斥锁

具体方法

  • pthread_cond_wait() 阻塞线程直到条件达到

  • pthread_cond_init

    • pthread_cond_t cond = PTHREAD_COND_INITIALIZER;

  • pthread_cond_signal()：唤醒另一个等待条件变量的信号

参考