并行计算学习之Pthread

1.线程与进程

 

   相对进程而言，线程是一个更加接近于执行体的概念，它可以与同进程中的其他线程共享数据，但拥有自己的栈空间，拥有独立的执行序列。在串行程序基础上引入线程和进程是为了提高程序的并发度，从而提高程序运行效率和响应时间。

 

    线程和进程在使用上各有优缺点：线程执行开销小，但不利于资源的管理和保护；而进程正相反。同时，线程适合于在SMP机器上运行，而进程则可以跨机器迁移。

 

 

2.Pthread基本概念

       Pthread是一套通用的线程库, 它广泛的被各种Unix所支持, 是由POSIX提出的. 因此, 它具有很好的客移植性. 在Linux上, 由于它是通过内核级线程来实现的, 就没有完全的实现它. 但从功能上来看, 它丝毫不逊色.

 

3.Pthread函数

   (1).创建线程

      int pthread_create (pthread_t *THREAD,pthread_attr_t * ATTR,void * (*START_ROUTINE)(void *),void * ARG);
   第一个参数是一个pthread_t型的指针用于保存线程id. 以后对该线程的操作都要用id来标示.
   第二个参数是一个pthread_attr_t的指针用于说明要创建的线程的属性, 使用NULL, 表示要使用缺省的属性.
   第三个参数指明了线程的如口, 是一个只有一个(void *)参数的函数.
   第四个参数指明了要传到线程如口函数的参数.

  pthread_create 返回值
  pthread_create() 在调用成功完成之后返回零。其他任何返回值都表示出现了错误。如果检测到以下任一情况,pthread_create() 将失败并返回相应的值。

  (2).等待线程终止

   pthread_join() 函数会一直阻塞调用线程,直到指定的线程终止。

  (3)分离线程
    pthread_detach(3C) 是 pthread_join(3C) 的替代函数,可回收创建时 detachstate 属性设置为 PTHREAD_CREATE_JOINABLE 的线程的存储空间。

  (4)终止线程
   使用 pthread_exit(3C) 终止线程。

 

    在POSIX线程API中提供了一个pthread_cleanup_push()/pthread_cleanup_pop()函数对用于自动释放资源--从pthread_cleanup_push()的调用点到pthread_cleanup_pop()之间的程序段中的终止动作（包括调用pthread_exit()和取消点终止）都将执行pthread_cleanup_push()所指定的清理函数。API定义如下：

void pthread_cleanup_push(void (*routine) (void *), void *arg)

void pthread_cleanup_pop(int execute)

 

4.线程同步

   (1).创建和销毁

   有两种方法创建互斥锁，静态方式和动态方式。

  POSIX定义了一个宏PTHREAD_MUTEX_INITIALIZER来静态初始化互斥锁，

  方法如下： pthread_mutex_t mutex=PTHREAD_MUTEX_INITIALIZER;

     在Linux Threads实现中，pthread_mutex_t是一个结构，而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)

   销毁一个互斥锁即意味着释放它所占用的资源，且要求锁当前处于开放状态。

  由于在Linux中，互斥锁并不占用任何资源，因此Linux Threads中的pthread_mutex_destroy()除了检查锁状态以外（锁定状态则返回EBUSY）没有其他动作。

  (2).锁操作

 

     锁操作主要包括加锁pthread_mutex_lock()、解锁pthread_mutex_unlock()和测试加锁pthread_mutex_trylock()三个。

    不论哪种类型的锁，都不可能被两个不同的线程同时得到，而必须等待解锁。对于普通锁和适应锁类型，解锁者可以是同进程内任何线程；而检错锁则必须由加锁者解锁才有效，否则返回EPERM；对于嵌套锁，文档和实现要求必须由加锁者解锁，但实验结果表明并没有这种限制，这个不同目前还没有得到解释。在同一进程中的线程，如果加锁后没有解锁，则任何其他线程都无法再获得锁。

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_mutex_trylock()语义与pthread_mutex_lock()类似，不同的是在锁已经被占据时返回EBUSY而不是挂起等待。

  (3).条件变量

条件变量是利用线程间共享的全局变量进行同步的一种机制，主要包括两个动作：一个线程等待"条件变量的条件成立"而挂起；另一个线程使"条件成立"（给出条件成立信号）。为了防止竞争，条件变量的使用总是和一个互斥锁结合在一起。

 

5.一个例子

 

#include <pthread.h>
#include <sys/time.h>
#include <stdlib.h>
#include <stdio.h>

#define MaxThreadNum 32  
#define kSamplePoints 100000000  
#define kSpace 1  

void *compute_pi(void *);
inline double WallTime();

int total_hits, hits[MaxThreadNum][kSpace];
int sample_points_per_thread, num_threads;
pthread_mutex_t MUTEX;
int main(void)
{
    int i;  
    double time_start, time_end;

    pthread_t p_threads[MaxThreadNum];

    pthread_mutex_init(&MUTEX,NULL);
    printf("Enter num_threads\n");
    scanf("%d", &num_threads);
    pthread_attr_t attr;
    pthread_attr_init(&attr);
    pthread_attr_setscope(&attr,PTHREAD_SCOPE_SYSTEM);
    time_start = WallTime();

    total_hits = 0;  
    sample_points_per_thread = kSamplePoints / num_threads;

    for(i=0; i<num_threads; i++)
    {
        pthread_create(&p_threads[i], &attr, compute_pi, (void *)i);
    }

    for(i=0; i<num_threads; i++)
    {
        pthread_join(p_threads[i], NULL);
        total_hits += hits[i][0];
    }

    double pi = 4.0 * (double)total_hits / kSamplePoints;
    time_end = WallTime();  
    printf("Elasped time: %lf, Pi: %lf\n", time_end - time_start, pi);
    return 0;
 
}
void *compute_pi(void * s)
{
    unsigned int seed;
    int i;

    double rand_no_x, rand_no_y;

    seed =(int) s;
    /*local_hits = 0;*/

    for(i=0; i<sample_points_per_thread; i++)
    {
        rand_no_x = (double)(rand_r(&seed))/(double)(RAND_MAX);
        rand_no_y = (double)(rand_r(&seed))/(double)(RAND_MAX);
        if((rand_no_x - 0.5)*(rand_no_x - 0.5) + (rand_no_y - 0.5) * (rand_no_y - 0.5) < 0.25)
        {
         pthread_mutex_lock(&MUTEX);
           total_hits++;
         pthread_mutex_unlock(&MUTEX);
        }
        seed *= i;
    }
    pthread_exit(0);
}

inline double WallTime()
{
    struct timeval tv;
    struct timezone tz;

    gettimeofday(&tv, &tz);

    double currTime = (double)tv.tv_sec + (double)tv.tv_usec/1000000.0;

    return currTime;
}