摘要
最近由于开发需要用到多线程，以前看的ARTOOKIT项目中有用到pthread,所以又重新看了一下，重点分析了它的多线程实现。它的实现方式的确很nice,以前没有注意到，现在整理在这里，一方面便于以后查询，另一方面也方便其他人。
Pthread设计思路
多线程编程设计的主要部分无非是线程创建，参数传递，数据同步，结果返回以及线程销毁。这里主要就这几个部分做简单的说明，重点给出代码，方便快速上手。如果有兴趣深入研究, 可以查看POSIX Threads Programming。如果觉得英文阅读有障碍，这里有中文版POSIX 多线程程序设计，其中内容出入不大。
线程创建
POSIX通过pthread_create()函数创建线程，API定义如下：
int pthread_create(pthread_t * thread, //新线程标识符
pthread_attr_t * attr, //新线程的运行属性
void * (*start_routine)(void *), //线程将会执行的函数
void * arg);//执行函数的传入参数，可以为结构体线程的运行属性pthread_attr_t 主要包括：__detachstate，表示新线程是否与进程中其他线程脱离同步，如果置位则新线程不能用pthread_join()来同步，且在退出时自行释放所占用的资源，缺省为PTHREAD_CREATE_JOINABLE状态，这个属性可以在线程创建并运行以后用 pthread_attr_setdetachstate 来设置； __schedpolicy，表示新线程的调度策略，主要包括SCHED_OTHER（正常、非实时）、SCHED_RR（实时、轮转法）和SCHED_FIFO（实时、先入先出）三种，缺省为SCHED_OTHER，后两种调度策略仅对超级用户有效。运行时可以用过pthread_setschedparam()来改变；__schedparam，一个struct sched_param结构，目前仅有一个sched_priority整型变量，表示线程的运行优先级，这个参数仅当调度策略为实时（即SCHED_RR或SCHED_FIFO）时才有效；__inheritsched，有两种值可供选择：PTHREAD_EXPLICIT_SCHED和PTHREAD_INHERIT_SCHED，前者表示新线程使用显式指定调度策略和调度参数（即attr中的值），而后者表示继承调用者线程的值。缺省为PTHREAD_EXPLICIT_SCHED；__scope，表示线程间竞争CPU的范围，也就是说线程优先级的有效范围。POSIX的标准中定义了两个值：PTHREAD_SCOPE_SYSTEM和PTHREAD_SCOPE_PROCESS，前者表示与系统中所有线程一起竞争CPU时间，后者表示仅与同进程中的线程竞争CPU。
参数传递 可以通过void *arg实现，例如创建一个结构体THREAD_HANDLE_T，内部成员变量就是需要传递的参数，先定义THREAD_HANDLE_T *arg, 并进行赋值，然后调用pthread_create(&thread，NULL，fun，arg)进行创建线程，创建的线程执行函数为 void *fun(THREAD_HANDLE_T*arg)，从而实现了通过arg进行数据的传递。
线程同步：互斥锁（量）与条件变量
互斥锁（量）
当多个线程共享一个变量时，一个线程读取这个变量的值，而有另外一个线程会修改这个变量的值，如果不加限制，就难以保证读取的变量值是修改之后还是修改之前的。为了保证变量不会被多个线程同时访问，引入互斥锁（量），互斥锁（量）对共享数据的保护就像一把锁。在Pthreads中，任何时候仅有一个线程可以锁定互斥锁（量），因此，当多个线程尝试去锁定互斥锁（量）时仅有一个会成功，直到锁定互斥锁（量）的线程解锁互斥锁（量）后，其他线程才可以去锁定。如此，线程就会轮流访问受保护数据。这就是互斥锁（量）的概念，在Pthread中定义了一套用于线程同步的mutex函数。
有两种方法创建互斥锁，静态方式和动态方式。
POSIX定义了一个宏PTHREAD_MUTEX_INITIALIZER来静态初始化互斥锁，方法如下：
pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;动态方式是采用pthread_mutex_init()函数来初始化互斥锁，API定义如下：
int pthread_mutex_init(pthread_mutex_t *mutex, const pthread_mutexattr_t *mutexattr);其中mutexattr用于指定互斥锁属性（见下），如果为NULL则使用缺省属性。
pthread_mutex_destroy()用于注销一个互斥锁，API定义如下：
int pthread_mutex_destroy(pthread_mutex_t *mutex);互斥锁（量）的属性在创建锁的时候指定，在试图对一个已经被锁定的互斥锁加锁时，不同类型的锁表现不同。PTHREAD_MUTEX_TIMED_NP，这是缺省值，也就是普通锁。当一个线程加锁以后，其余请求锁的线程将形成一个等待队列，并在解锁后按优先级获得锁。这种锁策略保证了资源分配的公平性； PTHREAD_MUTEX_RECURSIVE_NP，嵌套锁，允许同一个线程对同一个锁成功获得多次，并通过多次unlock解锁。如果是不同线程请求，则在加锁线程解锁时重新竞争；PTHREAD_MUTEX_ERRORCHECK_NP，检错锁，如果同一个线程请求同一个锁，则返回EDEADLK，否则与
PTHREAD_MUTEX_TIMED_NP类型动作相同。这样就保证当不允许多次加锁时不会出现最简单情况下的死锁。
锁操作主要包括加锁pthread_mutex_lock()、解锁pthread_mutex_unlock()和测试加锁pthread_mutex_trylock()三个。
int pthread_mutex_lock(pthread_mutex_t *mutex);
int pthread_mutex_unlock(pthread_mutex_t *mutex);
int pthread_mutex_trylock(pthread_mutex_t *mutex);条件变量
条件变量是利用线程间共享的全局变量进行同步的一种机制，主要包括两个动作：一个线程等待”条件变量的条件成立”而挂起；另一个线程使”条件成立”（给出条件成立信号）。为了防止竞争，条件变量的使用总是和一个互斥锁结合在一起。
条件变量和互斥锁（量）一样，都有静态动态两种创建方式，静态方式使用PTHREAD_COND_INITIALIZER常量，如下：
pthread_cond_t cond=PTHREAD_COND_INITIALIZER;动态方式调用pthread_cond_init()函数，API定义如下：
int pthread_cond_init(pthread_cond_t *cond, pthread_condattr_t *cond_attr);注销一个条件变量需要调用pthread_cond_destroy()，只有在没有线程在该条件变量上等待的时候才能注销这个条件变量，否则返回EBUSY。API定义如下：
int pthread_cond_destroy(pthread_cond_t *cond);等待条件有两种方式：无条件等待pthread_cond_wait()和计时等待pthread_cond_timedwait() , API定义如下：
int pthread_cond_wait(pthread_cond_t *cond, pthread_mutex_t *mutex);
int pthread_cond_timedwait(pthread_cond_t *cond, pthread_mutex_t *mutex,
const struct timespec *abstime);其中计时等待方式如果在给定时刻前条件没有满足，则返回ETIMEOUT，结束等待，其中abstime以与time()系统调用相同意义的绝对时间形式出现，0表示格林尼治时间1970年1月1日0时0分0秒。
无论哪种等待方式，都必须和一个互斥锁配合，以防止多个线程同时请求pthread_cond_wait() 或pthread_cond_timedwait() 的竞争条件。在调用pthread_cond_wait()前必须由本线程加锁（pthread_mutex_lock()），而在更新条件等待队列以前，mutex保持锁定状态，并在线程挂起进入等待前自动解锁。在条件满足从而离开pthread_cond_wait()之前，自动重新加锁，以与进入pthread_cond_wait()前的加锁动作对应。
激发条件有两种形式，pthread_cond_signal()激活一个等待该条件的线程，存在多个等待线程时按入队顺序激活其中一个；而pthread_cond_broadcast()则激活所有等待线程。API定义如下：
int pthread_cond_broadcast(pthread_cond_t *cond);
int pthread_cond_signal(pthread_cond_t *cond);线程取消
一般情况下，线程在其主体函数退出的时候会自动终止，但同时也可以因为接收到另一个线程发来的终止（取消）请求而强制终止。当我们创建的子线程是一个死循环函数，通过取消信号强制终止线程有显得很有必要。
线程取消的方法是向目标线程发Cancel信号，但如何处理Cancel信号则由目标线程自己决定，或者忽略、或者立即终止、或者继续运行至Cancelation-point（取消点），由不同的Cancelation状态决定。线程接收到CANCEL信号的缺省处理（即pthread_create()创建线程的缺省状态）是继续运行至取消点，也就是说设置一个CANCELED状态，线程继续运行，只有运行至Cancelation-point的时候才会退出。
根据POSIX标准，pthread_join()、pthread_testcancel()、pthread_cond_wait()、pthread_cond_timedwait()、sem_wait()、sigwait()等函数以及read()、write()等会引起阻塞的系统调用都是Cancelation-point，而其他pthread函数都不会引起Cancelation动作。
与线程取消相关的pthread函数有：pthread_cancel，pthread_setcancelstate，pthread_setcanceltype ，pthread_testcancel，API定义如下：
int pthread_cancel(pthread_t thread)；//发送终止信号给thread线程,如果成功则返回0
int pthread_setcancelstate(int state, int *oldstate)；//设置本线程对Cancel信号的反应
int pthread_setcanceltype(int type, int *oldtype)；//设置本线程取消动作的执行时机
void pthread_testcancel(void)；//检查本线程是否处于Canceld状态，如果是，则进行取消动作，否则直接返回关于线程终止以及线程终止时的清理、同步问题这里暂不做具体讨论了。
示例
这里给出在ARTOOKIT中使用pthread的方法，其对pthread进行一个封装定义，使用起来更方便。先看函数声明，如下：
/*
thread_sub.c, thread_sub.h
Written by Hirokazu Kato
kato@is.naist.jp Apr.24 2007
*/
#ifndef THREAD_SUB_H
#define THREAD_SUB_H
#ifdef __cplusplus
extern "C" {
#endif
typedef struct _THREAD_HANDLE_T THREAD_HANDLE_T;
//
// Client-side.
//
// Setup.
THREAD_HANDLE_T *threadInit( int ID, void *arg, void *(*start_routine)(THREAD_HANDLE_T*) ); // Create a new thread, and launch start_routine() on it. Returns NULL in case of failure.
int threadFree( THREAD_HANDLE_T **flag ); // Frees structures associated with the thread handle pointed to by the location pointed to by flag. Thread should already have terminated (i.e. threadWaitQuit() has returned). Location pointed to by flag is set to NULL.
// Communication.
int threadStartSignal( THREAD_HANDLE_T *flag ); // Send the worker thread the "start processing" request.
int threadGetStatus( THREAD_HANDLE_T *flag ); // Find out (without waiting) whether a worker has ended. 0 = not started or started but not yet ended, 1 = ended.
int threadGetBusyStatus( THREAD_HANDLE_T *flag ); // Find out if a worker is currently busy. 0 = worker not started or worker ended, 1 = worker started but not yet ended.
int threadEndWait( THREAD_HANDLE_T *flag ); // Wait for thread to end processing.
int threadWaitQuit( THREAD_HANDLE_T *flag ); // Tell a thread waiting for the "start processing" request to quit (exit), and wait until this has happened.
//
// Worker-side.
//
int threadStartWait( THREAD_HANDLE_T *flag ); // Wait for the "start processing" request. Returns 0 if worker should start, or -1 if worker should quit (exit).
int threadEndSignal( THREAD_HANDLE_T *flag ); // Notify anyone waiting that the worker has ended.
int threadGetID( THREAD_HANDLE_T *flag ); // Get the 'ID' value passed to the thread's start routine.
void *threadGetArg( THREAD_HANDLE_T *flag ); // Get the 'arg' value passed to the thread's start routine.
int threadGetCPU(void); // Returns the number of online CPUs in the system.
#ifdef __cplusplus
}
#endif
#endifClient-side表示主线程，Worker-side表示工作子线程。threadInit初始化并创建线程， threadFree释放资源并退出线程，threadStartSignal给子线程发送激活信号，开始执行，threadEndWait等待子线程运行结束信号，等等。。。其他函数可以看注释。实现方法如下：
/*
thread_sub.c, thread_sub.h
Written by Hirokazu Kato
kato@is.naist.jp Apr.24 2007
*/
#include <AR/ar.h>
#ifndef _WIN32
# include <unistd.h>
#endif
#include <stdio.h>
#include <stdlib.h>
#include <thread_sub.h>
//#define ARUTIL_DISABLE_PTHREADS // Uncomment to disable pthreads support.
#if !defined(_WINRT) && !defined(ARUTIL_DISABLE_PTHREADS)
# include <pthread.h>
#else
# define pthread_mutex_t CRITICAL_SECTION
# define pthread_mutex_init(pm, a) InitializeCriticalSectionEx(pm, 4000, CRITICAL_SECTION_NO_DEBUG_INFO)
# define pthread_mutex_lock(pm) EnterCriticalSection(pm)
# define pthread_mutex_unlock(pm) LeaveCriticalSection(pm)
# define pthread_mutex_destroy(pm) DeleteCriticalSection(pm)
# define pthread_cond_t CONDITION_VARIABLE
# define pthread_cond_init(pc, a) InitializeConditionVariable(pc)
# define pthread_cond_wait(pc, pm) SleepConditionVariableCS(pc, pm, INFINITE)
# define pthread_cond_signal(pc) WakeConditionVariable(pc)
# define pthread_cond_broadcast(pc) WakeAllConditionVariable(pc)
# define pthread_cond_destroy(pc)
# ifdef _WINRT
# include "thread_sub_winrt.h"
# else
# include <process.h> // _beginthread
# endif
#endif
struct _THREAD_HANDLE_T {
int ID;
int startF; // 0 = no request pending, 1 = start please, 2 = quit please.
int endF; // 0 = worker not started or worker running, 1 = worker completed, 2 = worker will quit (exit).
int busyF; // 0 = worker not started or worker ended, 1 = worker busy.
//pthread_t thread;
pthread_mutex_t mut;
pthread_cond_t cond1; // Signals from client that startF has changed.
pthread_cond_t cond2; // Signals from worker that endF has changed.
void *arg;
};
//
// Worker-side.
//
int threadStartWait( THREAD_HANDLE_T *flag )
{
pthread_mutex_lock(&(flag->mut));
while(flag->startF == 0) {
pthread_cond_wait(&(flag->cond1), &(flag->mut));
}
if( flag->startF == 1 ) {
flag->startF = 0;
flag->busyF = 1;
pthread_mutex_unlock(&(flag->mut));
return 0;
}
else {
flag->endF = 2;
pthread_cond_signal(&(flag->cond2));
pthread_mutex_unlock(&(flag->mut));
return -1;
}
}
int threadEndSignal( THREAD_HANDLE_T *flag )
{
pthread_mutex_lock(&(flag->mut));
flag->endF = 1;
flag->busyF = 0;
pthread_cond_signal(&(flag->cond2));
pthread_mutex_unlock(&(flag->mut));
return 0;
}
int threadGetID( THREAD_HANDLE_T *flag )
{
return (flag->ID);
}
void *threadGetArg( THREAD_HANDLE_T *flag )
{
return (flag->arg);
}
//
// Client-side.
//
// A construct to manage the difference in start routine signature between pthreads and windows threads.
#if defined(_WIN32) && !defined(_WINRT)
struct start_routine_proxy_arg {
void *(*start_routine)(THREAD_HANDLE_T*);
void *arg;
};
static void __cdecl start_routine_proxy(void *arg)
{
struct start_routine_proxy_arg *arg0 = (struct start_routine_proxy_arg *)arg;
(*(arg0->start_routine))(arg0->arg);
free(arg0);
}
#endif
THREAD_HANDLE_T *threadInit( int ID, void *arg, void *(*start_routine)(THREAD_HANDLE_T*) )
{
THREAD_HANDLE_T *flag;
int err;
#if !defined(_WINRT) && !defined(ARUTIL_DISABLE_PTHREADS)
pthread_t thread;
pthread_attr_t attr;
#endif
if ((flag = malloc(sizeof(THREAD_HANDLE_T))) == NULL) return NULL;
flag->ID = ID;
flag->startF = 0;
flag->endF = 0;
flag->busyF = 0;
flag->arg = arg;
pthread_mutex_init( &(flag->mut), NULL );
pthread_cond_init( &(flag->cond1), NULL );
pthread_cond_init( &(flag->cond2), NULL );
#if !defined(_WINRT) && !defined(ARUTIL_DISABLE_PTHREADS)
pthread_attr_init(&attr);
pthread_attr_setdetachstate(&attr, 1); // Preclude the need to do pthread_join on the thread after it exits.
err = pthread_create(&thread, &attr, (void *(*)(void*))start_routine, flag);
pthread_attr_destroy(&attr);
#elif defined(_WIN32)
# ifdef _WINRT
err = arCreateDetachedThreadWinRT(start_routine, flag);
# else
struct start_routine_proxy_arg *srpa_p = malloc(sizeof(struct start_routine_proxy_arg));
srpa_p->start_routine = start_routine;
srpa_p->arg = flag;
err = (_beginthread(start_routine_proxy, 0, srpa_p) == -1L);
# endif
#else
# error No routine available to create a thread.
#endif
if (err == 0) {
return flag;
} else {
threadFree(&flag);
return NULL;
}
}
int threadFree( THREAD_HANDLE_T **flag )
{
pthread_mutex_destroy(&((*flag)->mut));
pthread_cond_destroy(&((*flag)->cond1));
pthread_cond_destroy(&((*flag)->cond1));
free( *flag );
*flag = NULL;
return 0;
}
int threadStartSignal( THREAD_HANDLE_T *flag )
{
pthread_mutex_lock(&(flag->mut));
flag->startF = 1;
pthread_cond_signal(&(flag->cond1));
pthread_mutex_unlock(&(flag->mut));
return 0;
}
int threadGetStatus( THREAD_HANDLE_T *flag )
{
int endFlag;
pthread_mutex_lock(&(flag->mut));
endFlag = flag->endF;
pthread_mutex_unlock(&(flag->mut));
return endFlag;
}
int threadGetBusyStatus( THREAD_HANDLE_T *flag )
{
int busyFlag;
pthread_mutex_lock(&(flag->mut));
busyFlag = flag->busyF;
pthread_mutex_unlock(&(flag->mut));
return busyFlag;
}
int threadEndWait( THREAD_HANDLE_T *flag )
{
pthread_mutex_lock(&(flag->mut));
while (flag->endF == 0) {
pthread_cond_wait(&(flag->cond2), &(flag->mut));
}
flag->endF = 0;
pthread_mutex_unlock(&(flag->mut));
return 0;
}
int threadWaitQuit( THREAD_HANDLE_T *flag )
{
pthread_mutex_lock(&(flag->mut));
flag->startF = 2;
pthread_cond_signal(&(flag->cond1));
while (flag->endF != 2) {
pthread_cond_wait(&(flag->cond2), &(flag->mut));
}
return 0;
}
int threadGetCPU(void)
{
#ifdef _WIN32
SYSTEM_INFO info;
# ifndef _WINRT
GetSystemInfo(&info);
# else
GetNativeSystemInfo(&info);
# endif
return info.dwNumberOfProcessors;
#else
return (int)sysconf(_SC_NPROCESSORS_ONLN);
#endif
}下面给出我的使用实例。
#define DECODE_THREAD_MAX 4
struct _DecodeParamT{
struct _DecodeHandle *decodeHandle;// Reference to parent _DecodeHandle.
unsigned char *dataPtr; // Input image
int ret;
};
struct _DecodeHandle{
int xsize;
int ysize;
int threadNum;
struct _DecodeParamT arg[DECODE_THREAD_MAX];
THREAD_HANDLE_T *threadHandle[DECODE_THREAD_MAX];
};
//子线程执行函数
void *thread_fun(THREAD_HANDLE_T *threadHandle)
{
_DecodeParamT *arg;
int ID;
arg = (_DecodeParamT*)threadGetArg(threadHandle);
ID = threadGetID(threadHandle);
printf("Start decode_thread #%d
",ID);
for(;;)
{
// 判断startF 0 = no request pending, 1 = start please, 2 = quit please.
if( threadStartWait(threadHandle) < 0 ) break;
//cout<<ID<<" Start..."<<endl;
int width = arg->decodeHandle->xsize;
int height = arg->decodeHandle->ysize;
unsigned char* dataPtr = arg->dataPtr;
Mat image = Mat(height,width,CV_8UC1,dataPtr);
//处理方法
//process(image );
//显示图片
char chId[24];
sprintf(chId,"Thread #%03d",ID);
imshow(string(chId),image);
threadEndSignal(threadHandle); //线程单次运行结束
}
printf("End decode_thread #%d.
",ID);
return NULL;
}
int _tmain(int argc, _TCHAR* argv[])
{
static _DecodeHandle *handle = NULL;
handle = (_DecodeHandle*) malloc(sizeof(_DecodeHandle));
if(handle == NULL)
return -1;
VideoCapture cap(0);
if(!cap.isOpened())
return -1;
int key = -1;
Mat img;
cap >> img;
int w = img.cols;
int h = img.rows;
handle->xsize = w;
handle->ysize = h;
//获取线程数
handle->threadNum = 4;
for (int i=0; i< handle->threadNum; i++)
{
//handle->arg[i].dataPtr = (unsigned char *)malloc(w*h);
handle->arg[i].dataPtr = NULL;
printf("Create thread: %d",i);
handle->threadHandle[i] = threadInit(i,&(handle->arg[i]),thread_fun);
}
printf("Create thread Succeed...");
while( key != 'q')//按键‘q’退出while循环
{
cap >> img;
unsigned char* dataPtr;
dataPtr = gray.data;
for (int j =0;j <handle->threadNum;j++)
{
handle->arg[j].dataPtr = dataPtr ;
handle->arg[j].decodeHandle = handle;
printf("thread #%d Start Signal.
",j);
threadStartSignal(handle->threadHandle[j]);
}
for(int j=0; j< handle->threadNum ; j++)
{
//cout <<j<<" End Wait... "<<endl;
threadEndWait(handle->threadHandle[j]);
// get return data
...
}
imshow("src",img);
key = waitKey(10);
}
//释放子线程资源
for(int i=0; i< handle->threadNum;i++)
{
threadWaitQuit(handle->threadHandle[i]);
threadFree(&(handle->threadHandle[i]));
//free(handle->arg[i].dataPtr);
}
free(handle);
handle = NULL;
return 0;
}thread_sub使用提示：
主线程
// Example client structure:
//
// THREAD_HANDLE_T *threadHandle = threadInit(0, NULL, worker);
// if (!threadHandle) {
// fprintf(stderr, "Error starting thread.
");
// exit(-1);
// }
//
// // Ready to do some work:
// threadStartSignal(threadHandle);
//
// // Wait for the results.
// threadEndWait(threadHandle);
//
// // If all done, quit and cleanup.
// threadWaitQuit(threadHandle);
// threadFree(&threadHandle);
子线程执行函数
// Example worker structure:
//
// void *worker(THREAD_HANDLE_T *threadHandle)
// {
// void *arg = threadHandle->arg; // Get data passed to the worker.
// while (threadStartWait(threadHandle) == 0) {
// // Do work, probably on arg.
// threadEndSignal(threadHandle);
// }
// // Do cleanup.
// return (NULL);
// }由于自身能力有限，如有错误之处，请指正。
转载请注明出处。