POSIX 多线程程序设计

摘要  

​

Pthreads

THREADS WITHIN A UNIX PROCESS 

Pthreads 概述 

Pthreads 概述 

0.67 

6.4 

Pthreads 概述 

Pthreads API 

线程数据键（Thread-specific data keys） 

编译多线程程序 

GNU C++ 

线程管理（Thread Management） 

Q：一个线程被创建后，怎么知道操作系统何时调度该线程使之运行？ 

A：除非使用了Pthreads的调度机制，否则线程何时何地被执行取决于操作系统的实现。强壮的程序应该不依赖于线程执行的顺序。

 

Example Code - Pthread Creation and Termination  

#include <pthread.h>

#include <stdio.h> 

#define NUM_THREADS     5 

void *PrintHello(void *threadid) 

{ 

   int tid; 

   tid = (int)threadid; 

   printf("Hello World! It's me, thread #%d!/n", tid); 

pthread_exit(NULL);

} 

int main (int argc, char *argv[]) 

{ 

pthread_t threads[NUM_THREADS];

   int rc, t; 

   for(t=0; t<NUM_THREADS; t++){ 

      printf("In main: creating thread %d/n", t); 

rc = pthread_create(&threads[t], NULL, PrintHello, (void *)t);

      if (rc){ 

         printf("ERROR; return code from pthread_create() is %d/n", rc); 

         exit(-1); 

      } 

   } 

pthread_exit(NULL);

} 

Example 2 - Thread Argument Passing  

例子展示了用结构体向线程设置/传递参数。每个线程获得一个唯一的结构体实例。 

struct thread_data{ 

   int  thread_id; 

   int  sum; 

   char *message; 

}; 

struct thread_data thread_data_array[NUM_THREADS]; 

void *PrintHello(void *threadarg) 

{ 

   struct thread_data *my_data; 

   ... 

   my_data = (struct thread_data *) threadarg; 

   taskid = my_data->thread_id; 

   sum = my_data->sum; 

   hello_msg = my_data->message; 

   ... 

} 

int main (int argc, char *argv[]) 

{ 

   ... 

   thread_data_array[t].thread_id = t; 

   thread_data_array[t].sum = sum; 

   thread_data_array[t].message = messages[t]; 

rc = pthread_create(&threads[t], NULL, PrintHello,  

        (void *) &thread_data_array[t]);

   ... 

} 

Example Code - Pthread Joining  

这个例子演示了用Pthread join函数去等待线程终止。因为有些实现并不是默认创建线程是可连接状态，例子中显式地将其创建为可连接的。 

#include <pthread.h>

#include <stdio.h> 

#define NUM_THREADS    3 

void *BusyWork(void *null) 

{ 

   int i; 

   double result=0.0; 

   for (i=0; i<1000000; i++) 

   { 

     result = result + (double)random(); 

   } 

   printf("result = %e/n",result); 

pthread_exit((void *) 0);

} 

int main (int argc, char *argv[]) 

{ 

pthread_t thread[NUM_THREADS];

pthread_attr_t attr;

   int rc, t; 

   void *status; 

   /* Initialize and set thread detached attribute */ 

pthread_attr_init(&attr);

pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_JOINABLE);

   for(t=0; t<NUM_THREADS; t++) 

   { 

      printf("Creating thread %d/n", t); 

rc = pthread_create(&thread[t], &attr, BusyWork, NULL); 

      if (rc) 

      { 

         printf("ERROR; return code from pthread_create()  

                is %d/n", rc); 

         exit(-1); 

      } 

   } 

   /* Free attribute and wait for the other threads */ 

pthread_attr_destroy(&attr);

   for(t=0; t<NUM_THREADS; t++) 

   { 

rc = pthread_join(thread[t], &status);

      if (rc) 

      { 

         printf("ERROR; return code from pthread_join()  

                is %d/n", rc); 

         exit(-1); 

      } 

      printf("Completed join with thread %d status= %ld/n",t, (long)status); 

   } 

pthread_exit(NULL);

} 

98,304 

Example Code - Stack Management  

这个例子演示了如何去查询和设定线程栈大小。  

#include <pthread.h>

#include <stdio.h> 

#define NTHREADS 4 

#define N 1000 

#define MEGEXTRA 1000000 

pthread_attr_t attr; 

void *dowork(void *threadid) 

{ 

   double A[N][N]; 

   int i,j,tid; 

size_t mystacksize;

   tid = (int)threadid; 

pthread_attr_getstacksize (&attr, &mystacksize);

   printf("Thread %d: stack size = %li bytes /n", tid, mystacksize); 

   for (i=0; i<N; i++) 

     for (j=0; j<N; j++) 

      A[i][j] = ((i*j)/3.452) + (N-i); 

pthread_exit(NULL);

} 

int main(int argc, char *argv[]) 

{ 

pthread_t threads[NTHREADS];

size_t stacksize;

   int rc, t; 

pthread_attr_init(&attr);

pthread_attr_getstacksize (&attr, &stacksize);

   printf("Default stack size = %li/n", stacksize); 

   stacksize = sizeof(double)*N*N+MEGEXTRA; 

   printf("Amount of stack needed per thread = %li/n",stacksize); 

pthread_attr_setstacksize (&attr, stacksize);

   printf("Creating threads with stack size = %li bytes/n",stacksize); 

   for(t=0; t<NTHREADS; t++){ 

pthread_create(&threads[t], &attr, dowork, (void *)t);

      if (rc){ 

         printf("ERROR; return code from pthread_create() is %d/n", rc); 

         exit(-1); 

      } 

   } 

   printf("Created %d threads./n", t); 

pthread_exit(NULL); 

} 

互斥量（Mutex Variables） 

$1200 

互斥量（Mutex Variables） 

​​pthread_mutex_init​​​ (mutex,attr)  

​​pthread_mutex_destroy​​​ (mutex)  

​​pthread_mutexattr_init​​​ (attr)  

​​pthread_mutexattr_destroy​​​ (attr)  

互斥量（Mutex Variables） 

​​pthread_mutex_lock​​​ (mutex)  

​​pthread_mutex_trylock​​​ (mutex)  

​​pthread_mutex_unlock​​​ (mutex)  

Q：有多个线程等待同一个锁定的互斥量，当互斥量被解锁后，那个线程会第一个锁定互斥量？ 

A：除非线程使用了优先级调度机制，否则，线程会被系统调度器去分配，那个线程会第一个锁定互斥量是随机的。 

Example Code - Using Mutexes  

例程演示了线程使用互斥量处理一个点积（dot product）计算。主数据通过一个可全局访问的数据结构被所有线程使用，每个线程处理数据的不同部分，主线程等待其他线程完成计算并输出结果。 

#include <pthread.h>

#include <stdio.h> 

#include <malloc.h> 

/*    

The following structure contains the necessary information   

to allow the function "dotprod" to access its input data and  

place its output into the structure.   

*/ 

typedef struct  

 { 

   double      *a; 

   double      *b; 

   double     sum;  

   int     veclen;  

 } DOTDATA; 

/* Define globally accessible variables and a mutex */ 

#define NUMTHRDS 4 

#define VECLEN 100 

   DOTDATA dotstr;  

pthread_t callThd[NUMTHRDS];

pthread_mutex_t mutexsum;

/* 

The function dotprod is activated when the thread is created. 

All input to this routine is obtained from a structure  

of type DOTDATA and all output from this function is written into 

this structure. The benefit of this approach is apparent for the  

multi-threaded program: when a thread is created we pass a single 

argument to the activated function - typically this argument 

is a thread number. All  the other information required by the  

function is accessed from the globally accessible structure.  

*/ 

void *dotprod(void *arg) 

{ 

   /* Define and use local variables for convenience */ 

   int i, start, end, offset, len ; 

   double mysum, *x, *y; 

   offset = (int)arg; 

   len = dotstr.veclen; 

   start = offset*len; 

   end   = start + len; 

   x = dotstr.a; 

   y = dotstr.b; 

   /* 

   Perform the dot product and assign result 

   to the appropriate variable in the structure.  

   */ 

   mysum = 0; 

   for (i=start; i<end ; i++)  

    { 

      mysum += (x[i] * y[i]); 

    } 

   /* 

   Lock a mutex prior to updating the value in the shared 

   structure, and unlock it upon updating. 

   */ 

pthread_mutex_lock (&mutexsum);

   dotstr.sum += mysum; 

pthread_mutex_unlock (&mutexsum);

pthread_exit((void*) 0);

} 

/*  

The main program creates threads which do all the work and then  

print out result upon completion. Before creating the threads, 

the input data is created. Since all threads update a shared structure,  

we need a mutex for mutual exclusion. The main thread needs to wait for 

all threads to complete, it waits for each one of the threads. We specify 

a thread attribute value that allow the main thread to join with the 

threads it creates. Note also that we free up handles when they are 

no longer needed. 

*/ 

int main (int argc, char *argv[]) 

{ 

   int i; 

   double *a, *b; 

   void *status; 

pthread_attr_t attr;

   /* Assign storage and initialize values */ 

   a = (double*) malloc (NUMTHRDS*VECLEN*sizeof(double)); 

   b = (double*) malloc (NUMTHRDS*VECLEN*sizeof(double)); 

   for (i=0; i<VECLEN*NUMTHRDS; i++) 

    { 

     a[i]=1.0; 

     b[i]=a[i]; 

    } 

   dotstr.veclen = VECLEN;  

   dotstr.a = a;  

   dotstr.b = b;  

   dotstr.sum=0; 

pthread_mutex_init(&mutexsum, NULL);

   /* Create threads to perform the dotproduct  */ 

pthread_attr_init(&attr);

pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_JOINABLE);

        for(i=0; i<NUMTHRDS; i++) 

        { 

        /*  

        Each thread works on a different set of data. 

        The offset is specified by 'i'. The size of 

        the data for each thread is indicated by VECLEN. 

        */ 

pthread_create( &callThd[i], &attr, dotprod, (void *)i);

        } 

pthread_attr_destroy(&attr);

        /* Wait on the other threads */ 

        for(i=0; i<NUMTHRDS; i++) 

        { 

pthread_join( callThd[i], &status);

        } 

   /* After joining, print out the results and cleanup */ 

   printf ("Sum =  %f /n", dotstr.sum); 

   free (a); 

   free (b); 

pthread_mutex_destroy(&mutexsum);

pthread_exit(NULL);

}    

Serial version 
Pthreads version  

Main Thread  

Join / Continue  

条件变量（Condition Variables） 

​​pthread_cond_init​​​ (condition,attr)  

​​pthread_cond_destroy​​​ (condition)  

​​pthread_condattr_init​​​ (attr)  

​​pthread_condattr_destroy​​​ (attr)  

条件变量（Condition Variables） 

​​pthread_cond_wait​​​ (condition,mutex)  

​​pthread_cond_signal​​​ (condition)  

​​pthread_cond_broadcast​​​ (condition)  

使用这些函数时适当的锁定和解锁相关的互斥量是非常重要的。如： 

• 调用pthread_cond_wait()前锁定互斥量失败可能导致线程不会阻塞。 
• 调用pthread_cond_signal()后解锁互斥量失败可能会不允许相应的pthread_cond_wait()函数结束（保存阻塞）。 

Example Code - Using Condition Variables  

例子演示了使用Pthreads条件变量的几个函数。主程序创建了三个线程，两个线程工作，根系“count”变量。第三个线程等待count变量值达到指定的值。 

#include <pthread.h>

#include <stdio.h> 

#define NUM_THREADS  3 

#define TCOUNT 10 

#define COUNT_LIMIT 12 

int     count = 0; 

int     thread_ids[3] = {0,1,2}; 

pthread_mutex_t count_mutex;

pthread_cond_t count_threshold_cv;

void *inc_count(void *idp)  

{ 

  int j,i; 

  double result=0.0; 

  int *my_id = idp; 

  for (i=0; i<TCOUNT; i++) { 

pthread_mutex_lock(&count_mutex);

    count++; 

    /*  

    Check the value of count and signal waiting thread when condition is 

    reached.  Note that this occurs while mutex is locked.  

    */ 

    if (count == COUNT_LIMIT) { 

pthread_cond_signal(&count_threshold_cv);

      printf("inc_count(): thread %d, count = %d  Threshold reached./n",  

             *my_id, count); 

      } 

    printf("inc_count(): thread %d, count = %d, unlocking mutex/n",  

           *my_id, count); 

pthread_mutex_unlock(&count_mutex);

    /* Do some work so threads can alternate on mutex lock */ 

    for (j=0; j<1000; j++) 

      result = result + (double)random(); 

    } 

pthread_exit(NULL);

} 

void *watch_count(void *idp)  

{ 

  int *my_id = idp; 

  printf("Starting watch_count(): thread %d/n", *my_id); 

  /* 

  Lock mutex and wait for signal.  Note that the pthread_cond_wait  

  routine will automatically and atomically unlock mutex while it waits.  

  Also, note that if COUNT_LIMIT is reached before this routine is run by 

  the waiting thread, the loop will be skipped to prevent pthread_cond_wait 

  from never returning.  

  */ 

pthread_mutex_lock(&count_mutex);

  if (count<COUNT_LIMIT) { 

pthread_cond_wait(&count_threshold_cv, &count_mutex);

    printf("watch_count(): thread %d Condition signal  

           received./n", *my_id); 

    } 

pthread_mutex_unlock(&count_mutex);

pthread_exit(NULL);

} 

int main (int argc, char *argv[]) 

{ 

  int i, rc; 

pthread_t threads[3];

pthread_attr_t attr;

  /* Initialize mutex and condition variable objects */ 

pthread_mutex_init(&count_mutex, NULL);

pthread_cond_init (&count_threshold_cv, NULL);

  /* For portability, explicitly create threads in a joinable state */ 

pthread_attr_init(&attr);

pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_JOINABLE);

pthread_create(&threads[0], &attr, inc_count, (void *)&thread_ids[0]);

pthread_create(&threads[1], &attr, inc_count, (void *)&thread_ids[1]);

pthread_create(&threads[2], &attr, watch_count, (void *)&thread_ids[2]);

  /* Wait for all threads to complete */ 

  for (i=0; i<NUM_THREADS; i++) { 

pthread_join(threads[i], NULL);

  } 

  printf ("Main(): Waited on %d  threads. Done./n", NUM_THREADS); 

  /* Clean up and exit */ 

pthread_attr_destroy(&attr);

pthread_mutex_destroy(&count_mutex);

pthread_cond_destroy(&count_threshold_cv);

pthread_exit(NULL);

} 

没有覆盖的主题 

Pthread 库API参考 

​​pthread_condattr_setpshared​​