POSIX线程之二. 线程同步，属性和多线程

一、线程同步

线程同步的两种基本方法:

A. 信号量

   它的作用如同看守一段代码的看门人。

   控制一组相同对象的访问时，

   比如从5条可用的电话线中分配一条给某个线程的情况，

   就更适合计算信号量。

   

   最简单的信号量--二进制信号量，它只有0和1两种取值;

   更通用的信号量--计数信号量，它有更大的取值范围。

B. 互斥量

   它的作用如同保护代码段的一个互斥设备。

   如果想控制任一时刻只能有一个线程可以访问一些共享内存，

   通常使用互斥量

二、用信号量进行同步

有两组函数用于信号量，

一组取自POSIX的实时扩展，用于线程同步。

一组被称为系统V信号量，  用于进程同步

信号量函数的名字都以sem_开头。

线程中使用的基本信号量函数有四个,它们的定义如下:

#include <semaphore.h>

int sem_init(sem_t *sem, int pshared, unsigned int value);

信号量通过sem_init函数创建，

参数:

sem    : 将被函数初始化信号量对象指针 

pshared: 共享选项

         0,   表示这个信号量是当前进程的局部信号量，

         非0, 表示这个信号量可以在多个进程之间共享，

              目前linux还不支持，将会导致调用失败。

value  : 信号量初始的整数值

控制信号量值的函数:

#include <semaphore.h>

int sem_wait(sem_t * sem);
int sem_post(sem_t * sem);

sem_post: 以原子操作的方式给信号量的值加1 (V操作).

sem_wait: 以原子操作的方式给信号量的值减1 (P操作).

          但它会等待直到信号量有个非零值才会开始减法操作。

  因此，

  如果对值为2的信号量调用sem_wait,

      线程将继续执行，但信号量的值会减1.

  如果对值为0的信号量调用sem_wait,

      这个函数将会等待，

      直到有其他线程增加了该信号量值使其不再为0为止。

  如果两个线程同时在sem_wait调用上等待同一个信号量变为非零值，

      那么当该信号量被第三个线程增加1时，

      只有其中一个等待线程将开始对信号量减1，然后继续执行，

      另外一个线程还将继续等待。

#include <semaphore.h>

int sem_destroy(sem_t * sem);

对用完的信号量进行清理，

并清理该信号量拥有的所有资源。

如果企图清理的信号量正在被一些线程等待，将返回错误。

这些函数在成功进返回0.

示例程序:

#include <stdio.h>
#include <unistd.h>
#include <stdlib.h>
#include <string.h>

#include <pthread.h>
#include <semaphore.h>

void *thread_function(void *arg);
sem_t bin_sem;

#define WORK_SIZE 1024
char work_area[WORK_SIZE];

int main()
{
  int res;
  pthread_t a_thread;
  void *thread_result;

  /* 创建信号量 */
  res = sem_init(&bin_sem, 0, 0);
  if (res != 0)
  {
    perror("Semaphore initialization failed");
    exit(EXIT_FAILURE);
  }

  res = pthread_create(&a_thread, NULL, thread_function, NULL);
  if (res != 0)
  {
    perror("Thread creation failed");
    exit(EXIT_FAILURE);
  }

  printf("Input some text. Enter 'end' to finish\n");
  while(strncmp("end", work_area, 3) != 0)
  {
    fgets(work_area, WORK_SIZE, stdin);
    sem_post(&bin_sem);  //  加1
  }

  printf("\nWaiting for thread to finish...\n");
  res = pthread_join(a_thread, &thread_result);
  if (res != 0)
  {
    perror("Thread join failed");
    exit(EXIT_FAILURE);
  }

  printf("Thread joined\n");
  sem_destroy(&bin_sem);
  exit(EXIT_SUCCESS);
}


void *thread_function(void *arg)
{
  sem_wait(&bin_sem);  //  减1
  while(strncmp("end", work_area, 3) != 0)
  {
    printf("You input %d characters\n", strlen(work_area) -1);
    sem_wait(&bin_sem);
  }
  pthread_exit(NULL);
}

编译与运行:

$ cc -D_REENTRANT thread3.c –o thread3 -lpthread
$ ./thread3
Input some text. Enter ‘end’ to finish
The Wasp Factory
You input 16 characters
Iain Banks
You input 10 characters
end
Waiting for thread to finish...
Thread joined

三、用互斥量进行同步

它允许程序员锁住某个对象，使得每次只能有一个线程访问它。

必须在进入一段代码之前锁住一个互斥量，然后在完成操作之后解锁它。

操作函数如下:

#include <pthread.h>

int pthread_mutex_init(pthread_mutex_t *mutex,
                       const pthread_mutexattr_t *mutexattr);
int pthread_mutex_lock(pthread_mutex_t *mutex));
int pthread_mutex_unlock(pthread_mutex_t *mutex);
int pthread_mutex_destroy(pthread_mutex_t *mutex);

参数:

mutex : 声明过的对象指针

mutexattr: 互斥量的属性，默认为fast.

但它有一个缺点:

如果程序试图对一个已经加了锁的互斥量调用pthread_mutex_lock,

    程序将会被阻塞。

    而又因为拥有互斥量的这个线程正是现在被阻塞的线程，

    所以互斥量就永远也不会解锁了，

    程序就进入死锁状态。

这个问题可以通过改变互斥量的属性来解决:

    可以让它检查这种情况并返回一个错误，

    或者让它递归地操作，给同一个线程加上多个锁，

    但必须注意在后面执行同等数量的解锁操作。

返回值:

成功时，返回0；

失败时，将返回错误代码，但并不设置errno,

        所以必须对函数的返回代码进行检查。

示例代码:

#include <stdio.h>
#include <unistd.h>
#include <stdlib.h>
#include <string.h>
#include <pthread.h>
#include <semaphore.h>

void *thread_function(void *arg);
pthread_mutex_t work_mutex; /* protects both work_area and time_to_exit */

#define WORK_SIZE 1024
char work_area[WORK_SIZE];
int time_to_exit = 0;

int main()
{
  int res;
  pthread_t a_thread;
  void *thread_result;

  res = pthread_mutex_init(&work_mutex, NULL);
  if (res != 0)
  {
    perror("Mutex initialization failed");
    exit(EXIT_FAILURE);
  }

  res = pthread_create(&a_thread, NULL, thread_function, NULL);
  if (res != 0)
  {
    perror("Thread creation failed");
    exit(EXIT_FAILURE);
  }

  pthread_mutex_lock(&work_mutex);
  printf("Input some text. Enter 'end' to finish\n");
  while(!time_to_exit)
  {
    fgets(work_area, WORK_SIZE, stdin);
    pthread_mutex_unlock(&work_mutex);
    while(1)
    {
      pthread_mutex_lock(&work_mutex);
      if (work_area[0] != '\0')
      {
        pthread_mutex_unlock(&work_mutex);
        sleep(1);
      }
      else
      {
        break;
      }

    }
  }

  pthread_mutex_unlock(&work_mutex);
  printf("\nWaiting for thread to finish...\n");
  res = pthread_join(a_thread, &thread_result);
  if (res != 0)
  {
    perror("Thread join failed");
    exit(EXIT_FAILURE);
  }

  printf("Thread joined\n");
  pthread_mutex_destroy(&work_mutex);
  exit(EXIT_SUCCESS);
}

void *thread_function(void *arg)
{
  sleep(1);
  pthread_mutex_lock(&work_mutex);
  while(strncmp("end", work_area, 3) != 0)
  {
    printf("You input %d characters\n", strlen(work_area) -1);
    work_area[0] = '\0';
    pthread_mutex_unlock(&work_mutex);
    sleep(1);
    pthread_mutex_lock(&work_mutex);
    while (work_area[0] == '\0' )
    {
      pthread_mutex_unlock(&work_mutex);
      sleep(1);
      pthread_mutex_lock(&work_mutex);
    }
  }
  time_to_exit = 1;
  work_area[0] = '\0';
  pthread_mutex_unlock(&work_mutex);
  pthread_exit(0);
}

编译与运行:

$ cc -D_REENTRANT thread4.c –o thread4 -lpthread
$ ./thread4
Input some text. Enter 'end' to finish
Whit
You input 4 characters
The Crow Road
You input 13 characters
end
Waiting for thread to finish...
Thread joined

分析:

这里通过轮询来获得结果的方法并不是好的编程方式。

在实际编程中，

应该尽可能用信号量来避免出现这种情况。

四、线程的属性

#include <pthread.h>

int pthread_attr_init(pthread_attr_t *attr);

初始化一个线程对象的属性

成功时，返回0，

失败时，返回错误代码。

#include <pthread.h>

int pthread_attr_setdetachstate(pthread_attr_t *attr, int detachstate);
int pthread_attr_getdetachstate(const pthread_attr_t *attr, int *detachstate);
int pthread_attr_setschedpolicy(pthread_attr_t *attr, int policy);
int pthread_attr_getschedpolicy(const pthread_attr_t *attr, int *policy);
int pthread_attr_setschedparam(pthread_attr_t *attr, const struct sched_param
*param);
int pthread_attr_getschedparam(const pthread_attr_t *attr, struct sched_param
*param);
int pthread_attr_setinheritsched(pthread_attr_t *attr, int inherit);
int pthread_attr_getinheritsched(const pthread_attr_t *attr, int *inherit);
int pthread_attr_setscope(pthread_attr_t *attr, int scope);
int pthread_attr_getscope(const pthread_attr_t *attr, int *scope);
int pthread_attr_setstacksize(pthread_attr_t *attr, int scope);
int pthread_attr_getstacksize(const pthread_attr_t *attr, int *scope);

detachedstate: 这个属性允许我们无需对线程进行重新合并。

   pthread_attr_setdetachstate函数可用到的两个标志分别是:

     PTHREAD_CREATE_JOINABLE, 默认值，允许两个线程重新合并。

     PTHREAD_CREATE_DETACHED, 不能调用pthread_join来获得另一个线程的退出状态。

schedpolicy:  控制线程的调度方式。

   取值可以是:

     SCHED_OTHER, 默认值, 

     SCHED_RP,    超级用户权限运行的进程，使用循环调度机制;

     SCHED_FIFO,  超级用户权限运行的进程，使用先进先出策略。

schedparam:  和schedpolicy结合使用

inheritsched: 这个属性可取两个值

     PTHREAD_EXPLICIT_SCHED, 

     PTHREAD_INHERIT_SCHED

scope: 控制一个线程调度的计算方式。

stacksize: 控制线程创建的栈大小，单位为字节

示例程序:

创建一个线程属性，

并其设置为脱离状态，

然后用这个属性创建一个线程。

子线程结束时，它照常调用pthread_exit,

但原先的线程不再等待与它创建的子线程重新合并。

主线程通过一个简单的thread_finished标志来检测子线程是否已经结束，

并显示线程之间仍然共享着变量。

#include <stdio.h>
#include <unistd.h>
#include <stdlib.h>
#include <pthread.h>

void *thread_function(void *arg);

char message[] = "Hello World";
int thread_finished = 0;

int main()
{
  int res;
  pthread_t a_thread;
  pthread_attr_t thread_attr;

  res = pthread_attr_init(&thread_attr);
  if (res != 0)
  {
    perror("Attribute creation failed");
    exit(EXIT_FAILURE);
  }

  res = pthread_attr_setdetachstate(&thread_attr, PTHREAD_CREATE_DETACHED);
  if (res != 0)
  {
    perror("Setting detached attribute failed");
    exit(EXIT_FAILURE);
  }

  res = pthread_create(&a_thread, &thread_attr,
                       thread_function, (void *)message);
  if (res != 0)
  {
    perror("Thread creation failed");
    exit(EXIT_FAILURE);
  }

  (void)pthread_attr_destroy(&thread_attr);
  while(!thread_finished)
  {
    printf("Waiting for thread to say it's finished...\n");
    sleep(1);
  }

  printf("Other thread finished, bye!\n");
  exit(EXIT_SUCCESS);
}

void *thread_function(void *arg)
{
  printf("thread_function is running. Argument was %s\n", (char *)arg);
  sleep(4);
  printf("Second thread setting finished flag, and exiting now\n");
  thread_finished = 1;
  pthread_exit(NULL);
}

五、取消一个线程

#include <pthread.h>

int pthread_cancel(pthread_t thread);

提供一个线程标识，发送请求来取消它。

线程可以用pthread_setcancelstate来设置自己的取消状态。

#include <pthread.h>

int pthread_setcancelstate(int state, int *oldstate);

参数:

state: 取值可以是

  PTHREAD_CANCEL_ENABLE, 允许线程接收取消请求;

  PTHREAD_CANCEL_DISABLE, 忽略取消请求。

oldstate: 用于获取先前的取消状态，

  如果没有兴趣，可以传NULL。

如果取消请求接受了，线程可以进入第二个控制层次，

用pthread_setcanceltype来设置取消类型:

#include <pthread.h>

int pthread_setcanceltype(int type, int *oldtype);

参数:

type : 取值为

  PTHREAD_CANCEL_ASYNCHRONOUS, 在接收到取消请求后立即执行

  PTHREAD_CANCEL_DEFERRED, 它将使得在接收到请求后，

     一直等待直到线程执行了下述函数之一后才采取行动。

     具体函数是:

       pthread_join, 

       pthread_cond_wait,

       pthread_cond_timedwait, 

       pthread_testcancel, 

       sem_wait,

       sigwait.

oldtype:保存先前的状态，可以传NULL

默认情况下，

线程在启动时

取消状态为: PTHREAD_CANCEL_ENABLE 

取消类型为: PTHREAD_CANCEL_DEFERRED.

示例程序:

主线程向它创建的线程发送一个取消请求

#include <stdio.h>
#include <unistd.h>
#include <stdlib.h>
#include <pthread.h>

void *thread_function(void *arg);

int main()
{
  int res;
  pthread_t a_thread;
  void *thread_result;

  res = pthread_create(&a_thread, NULL, thread_function, NULL);
  if (res != 0)
  {
    perror(“Thread creation failed”);
    exit(EXIT_FAILURE);
  }

  sleep(3);
  printf(“Canceling thread...\n”);
  res = pthread_cancel(a_thread);
  if (res != 0)
  {
    perror(“Thread cancelation failed”);
    exit(EXIT_FAILURE);
  }

  printf(“Waiting for thread to finish...\n”);
  res = pthread_join(a_thread, &thread_result);
  if (res != 0)
  {
    perror(“Thread join failed”);
    exit(EXIT_FAILURE);
  }

  exit(EXIT_SUCCESS);
}


void *thread_function(void *arg)
{
  int i, res;

  res = pthread_setcancelstate(PTHREAD_CANCEL_ENABLE, NULL);
  if (res != 0)
  {
    perror(“Thread pthread_setcancelstate failed”);
    exit(EXIT_FAILURE);
  }

  res = pthread_setcanceltype(PTHREAD_CANCEL_DEFERRED, NULL);
  if (res != 0)
  {
    perror(“Thread pthread_setcanceltype failed”);
    exit(EXIT_FAILURE);
  }

  printf(“thread_function is running\n”);
  for(i = 0; i < 10; i++)
  {
    printf(“Thread is still running (%d)...\n”, i);
    sleep(1);
  }
  pthread_exit(0);
}

编译运行:

$ ./thread7
thread_function is running
Thread is still running (0)...
Thread is still running (1)...
Thread is still running (2)...
Canceling thread...
Waiting for thread to finish...
$

六、多线程

在同一程序中创建多个线程，

然后以不同于其启动的顺序将它们合并到一起。

#include <stdio.h>
#include <unistd.h>
#include <stdlib.h>
#include <pthread.h>

#define NUM_THREADS 6

void *thread_function(void *arg);

int main()
{
  int res;
  pthread_t a_thread[NUM_THREADS];
  void *thread_result;
  int lots_of_threads;

  for(lots_of_threads = 0; lots_of_threads < NUM_THREADS; lots_of_threads++)
  {
    res = pthread_create(&(a_thread[lots_of_threads]),
                         NULL, thread_function, (void *)&lots_of_threads);
    if (res != 0)
    {
      perror(“Thread creation failed”);
      exit(EXIT_FAILURE);
    }
    sleep(1);
  }

  printf(“Waiting for threads to finish...\n”);
  for(lots_of_threads = NUM_THREADS - 1; lots_of_threads >= 0; lots_of_threads--)
  {
    res = pthread_join(a_thread[lots_of_threads], &thread_result);
    if (res == 0)
    {
      printf(“Picked up a thread\n”);
    }
    else
    {
      perror(“pthread_join failed”);
    }
  }

  printf(“All done\n”);
  exit(EXIT_SUCCESS);
}

void *thread_function(void *arg)
{
  int my_number = *(int *)arg;
  int rand_num;

  printf(“thread_function is running. Argument was %d\n”, my_number);

  rand_num=1+(int)(9.0*rand()/(RAND_MAX+1.0));
  sleep(rand_num);

  printf(“Bye from %d\n”, my_number);
  pthread_exit(NULL);
}

编译与运行:

$ ./thread8
thread_function is running. Argument was 0
thread_function is running. Argument was 1
thread_function is running. Argument was 2
thread_function is running. Argument was 3
thread_function is running. Argument was 4
Bye from 1
thread_function is running. Argument was 5
Waiting for threads to finish...
Bye from 5
Picked up a thread
Bye from 0
Bye from 2
Bye from 3
Bye from 4
Picked up a thread
Picked up a thread
Picked up a thread
Picked up a thread
Picked up a thread
All done