高级Linux程序设计第四章：线程

• 要想使用POSIX标准线程API(pthreads)，需要连接libpthread.so库到程序中。

1、创建线程

• 进程中的每个线程都有一个线程号，类型为pthread_t。

• 用pthread_self函数可以返回当前线程的线程号。

• 线程号之间的比较可以用函数pthread_equal。

if (!pthread_equal (pthread_self (), other_thread))

    pthread_join (other_thread, NULL);

• 每个线程执行一个线程函数：

void * function(void *)

• 用函数pthread_create函数可以创建一个线程：

  • 第一个参数是一个pthread_t类型的变量的指针，用于存储线程号。

  • 第二个参数是一个pthread_attr_t类型的线程属性对象，NULL表示使用默认属性

  • 第三个参数是一个指向线程函数的指针

  • 第四个参数是void指针，指向传给线程的参数

• 编译并链接程序：

% cc -o thread-create thread-create.c –lpthread

• 线程退出有两种方式：

  • 线程函数退出。线程函数的返回值就是线程的返回值。

  • 显式调用pthread_exit，其参数为线程的返回值。

#include <pthread.h>

#include <stdio.h>

/* Prints x’s to stderr. The parameter is unused. Does not return. */

void* print_xs (void* unused)

{

    while (1)

        fputc (‘x’, stderr);

    return NULL;

}

/* The main program. */

int main ()

{

    pthread_t thread_id;

    /* Create a new thread. The new thread will run the print_xs function. */

    pthread_create (&thread_id, NULL, &print_xs, NULL);

    /* Print o’s continuously to stderr. */

    while (1)

        fputc (‘o’, stderr);

    return 0;

}

1.1、向线程传递数据

Listing 4.2 (thread-create2) Create Two Threads

#include <pthread.h>

#include <stdio.h>

/* Parameters to print_function. */

struct char_print_parms

{

    /* The character to print. */

    char character;

    /* The number of times to print it. */

    int count;

};

/* Prints a number of characters to stderr, as given by PARAMETERS,

which is a pointer to a struct char_print_parms. */

void* char_print (void* parameters)

{

    /* Cast the cookie pointer to the right type. */

    struct char_print_parms* p = (struct char_print_parms*) parameters;

    int i;

    for (i = 0; i < p->count; ++i)

        fputc (p->character, stderr);

    return NULL;

}

/* The main program. */

int main ()

{

    pthread_t thread1_id;

    pthread_t thread2_id;

    struct char_print_parms thread1_args;

    struct char_print_parms thread2_args;

    /* Create a new thread to print 30,000 ’x’s. */

    thread1_args.character = ’x’;

    thread1_args.count = 30000;

    pthread_create (&thread1_id, NULL, &char_print, &thread1_args);

    /* Create a new thread to print 20,000 o’s. */

    thread2_args.character = ’o’;

    thread2_args.count = 20000;

    pthread_create (&thread2_id, NULL, &char_print, &thread2_args);

    return 0;

}

此程序有一个Bug，主线程创建线程的时候，穿进去的参数是局部变量，如果主线程在从线程之前退出，则局部变量会被释放，然而从线程却可能仍然要访问这些变量，就会出错。因而应该应用join使得主线程等待从线程结束。

1.2、Join线程

• phread_join有两个参数：

  • 第一个参数是要join的线程号。

  • 第二个参数是一个void指针，指向一个变量来接收线程的返回值。

Revised Main Function for thread-create2.c

int main ()

{

    pthread_t thread1_id;

    pthread_t thread2_id;

    struct char_print_parms thread1_args;

    struct char_print_parms thread2_args;

    /* Create a new thread to print 30,000 x’s. */

    thread1_args.character = ’x’;

    thread1_args.count = 30000;

    pthread_create (&thread1_id, NULL, &char_print, &thread1_args);

    /* Create a new thread to print 20,000 o’s. */

    thread2_args.character = ’o’;

    thread2_args.count = 20000;

    pthread_create (&thread2_id, NULL, &char_print, &thread2_args);

    /* Make sure the first thread has finished. */

    pthread_join (thread1_id, NULL);

    /* Make sure the second thread has finished. */

    pthread_join (thread2_id, NULL);

    /* Now we can safely return. */

    return 0;

}

1.3、线程返回值

用线程计算素数

#include <pthread.h>

#include <stdio.h>

/* Compute successive prime numbers (very inefficiently). Return the

Nth prime number, where N is the value pointed to by *ARG. */

void* compute_prime (void* arg)

{

    int candidate = 2;

    int n = *((int*) arg);

    while (1) {

        int factor;

        int is_prime = 1;

        /* Test primality by successive division. */

        for (factor = 2; factor < candidate; ++factor)

            if (candidate % factor == 0) {

                is_prime = 0;

                break;

            }

        /* Is this the prime number we’re looking for? */

        if (is_prime) {

            if (--n == 0)

            /* Return the desired prime number as the thread return value. */

                return (void*) candidate;

        }

        ++candidate;

    }

    return NULL;

}

int main ()

{

    pthread_t thread;

    int which_prime = 5000;

    int prime;

    /* Start the computing thread, up to the 5,000th prime number. */

    pthread_create (&thread, NULL, &compute_prime, &which_prime);

    /* Do some other work here... */

    /* Wait for the prime number thread to complete, and get the result. */

    pthread_join (thread, (void*) &prime);

    /* Print the largest prime it computed. */

    printf(“The %dth prime number is %d.\n”, which_prime, prime);

    return 0;

} 

1.4、线程属性

• 使用线程属性：

  • 创建pthread_attr_t对象。

  • 调用pthread_attr_init来初始化为默认属性。

  • 可以修改线程属性对象，设置需要的属性值。

  • 调用pthread_create的时候将属性对象指针传入。

  • 调用pthread_attr_destroy来释放属性对象。

  • pthread_attr_t本身不能自动回收，但是可以用pthread_attr_init函数重新初始化线程对象。

• 使用detach state属性

  • 一个线程可以默认状态下被创建称为可join的线程，也可以创建为detached线程。

  • 一个可join的线程在结束后不会自动被回收，而是要其他线程调用pthread_join来获得此线程的返回值。

  • 一个detached线程在结束后会自动被回收。

  • 用函数pthread_attr_setdetachstate来设定线程属性对象的detach state.

#include <pthread.h>

void* thread_function (void* thread_arg)

{

/* Do work here... */

}

int main ()

{

    pthread_attr_t attr;

    pthread_t thread;

    pthread_attr_init (&attr);

    pthread_attr_setdetachstate (&attr, PTHREAD_CREATE_DETACHED);

    pthread_create (&thread, &attr, &thread_function, NULL);

    pthread_attr_destroy (&attr);

    /* Do work here... */

    /* No need to join the second thread. */

    return 0;

}

2、取消线程

• 一个线程可以调用pthread_cancel来取消另一个线程。

• 被取消的线程需要被join来释放资源。

• 被取消的线程的返回值为PTHREAD_CANCELED

• 有关线程的取消，一个线程可以为如下三个状态：

  • 可异步取消：一个线程可以在任何时刻被取消。

  • 可同步取消：取消的请求被放在队列中，直到线程到达某个点，才被取消。

  • 不可取消：取消的请求被忽略。

  • 默认状态下，线程是可同步取消的。

• 调用pthread_setcanceltype来设定线程取消的方式：

  • pthread_setcanceltype (PTHREAD_CANCEL_ASYNCHRONOUS, NULL);

  • pthread_setcanceltype (PTHREAD_CANCEL_DEFERRED, NULL);

  • pthread_setcanceltype (PTHREAD_CANCEL_DISABLE, NULL);

3、线程私有数据

• 每个线程私有数据都有一个key

• 每个线程都使用此key来访问属于它的那份数据。

• 调用pthread_key_create来为此线程创建一个新的key和数据。

  • 第一个参数是pthread_key_t

  • 第二个参数是一个回收函数，会在线程退出的时候被调用。

• 每个线程可以调用pthread_setspecific，根据key来设定线程私有数据的值。

• 调用pthread_getspecific来得到一个线程私有数据。

#include <malloc.h>

#include <pthread.h>

#include <stdio.h>

/* The key used to associate a log file pointer with each thread. */

static pthread_key_t thread_log_key;

/* Write MESSAGE to the log file for the current thread. */

void write_to_thread_log (const char* message)

{

    FILE* thread_log = (FILE*) pthread_getspecific (thread_log_key);

    fprintf (thread_log, “%s\n”, message);

}

/* Close the log file pointer THREAD_LOG. */

void close_thread_log (void* thread_log)

{

    fclose ((FILE*) thread_log);

}

void* thread_function (void* args)

{

    char thread_log_filename[20];

    FILE* thread_log;

    /* Generate the filename for this thread’s log file. */

    sprintf (thread_log_filename, “thread%d.log”, (int) pthread_self ());

    /* Open the log file. */

    thread_log = fopen (thread_log_filename, “w”);

    /* Store the file pointer in thread-specific data under thread_log_key. */

    pthread_setspecific (thread_log_key, thread_log);

    write_to_thread_log (“Thread starting.”);

    /* Do work here... */

    return NULL;

}

int main ()

{

    int i;

    pthread_t threads[5];

    /* Create a key to associate thread log file pointers in thread-specific data. Use close_thread_log to clean up the file

pointers. */

    pthread_key_create (&thread_log_key, close_thread_log);

    /* Create threads to do the work. */

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

        pthread_create (&(threads[i]), NULL, thread_function, NULL);

        /* Wait for all threads to finish. */

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

            pthread_join (threads[i], NULL);

    return 0;

}

3.1、线程回收

Linux可以提供回收器(cleanup handler)，它是一个函数，在线程退出的时候被调用。

调用pthread_cleanup_push可以注册一个回收器。

调用pthread_cleanup_pop可以注销一个回收器。

pthread_cleanup_pop(0)仅仅注销一个回收器。

pthread_cleanup_pop(1)不仅仅注销这个回收器，而且调用它。

#include <malloc.h>

#include <pthread.h>

/* Allocate a temporary buffer. */

void* allocate_buffer (size_t size)

{

    return malloc (size);

}

/* Deallocate a temporary buffer. */

void deallocate_buffer (void* buffer)

{

    free (buffer);

}

void do_some_work ()

{

    /* Allocate a temporary buffer. */

    void* temp_buffer = allocate_buffer (1024);

    /* Register a cleanup handler for this buffer, to deallocate it in case the thread exits or is cancelled. */

    pthread_cleanup_push (deallocate_buffer, temp_buffer);

    /* Do some work here that might call pthread_exit or might be cancelled... */

    /* Unregister the cleanup handler. Because we pass a nonzero value, this actually performs the cleanup by calling

deallocate_buffer. */

    pthread_cleanup_pop (1);

}

4、线程同步及临界区

4.1、互斥锁(Mutexes)

• Mutex全称MUTual EXclusion lock，也即互斥锁。

• 同一时刻，只有一个线程可以访问互斥锁。

• 调用pthread_mutex_init可以创建互斥锁：

  • 第一个参数是pthread_mutex_t

  • 第二个参数是互斥锁属性对象，设为NULL表示使用默认属性。

pthread_mutex_t mutex;

pthread_mutex_init (&mutex, NULL);

• 创建互斥锁的第二种方式：pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;

• 调用pthread_mutex_lock来锁定互斥锁。

• 调用pthread_mutex_unlock来解锁互斥锁。

#include <stdio.h>

#include <stdlib.h>

#include <malloc.h>

#include <pthread.h>

struct job

{

    struct job* next;

    int value;

};

struct job* job_queue;

void process_job(struct job*);

pthread_mutex_t job_queue_mutex = PTHREAD_MUTEX_INITIALIZER;

void * process_queue_function (void * arg)

{

    while(1)

    {

        struct job* next_job;

        pthread_mutex_lock(&job_queue_mutex);

        if(job_queue == NULL)

            next_job = NULL;

        else

        {

            printf("begin removing a job...\n");

            next_job = job_queue;

            job_queue = job_queue->next;

            printf("after removing a job...\n");

         }

         pthread_mutex_unlock(&job_queue_mutex);

         if(next_job == NULL)

        {

             sleep(5);

             continue;

        }

        process_job(next_job);

        free(next_job);

    }

    return NULL;

}

void process_job(struct job* p)

{

    printf("The value is : %d.\n", p->value);

}

void enqueue_job(struct job* new_job)

{

    pthread_mutex_lock(&job_queue_mutex);

    printf("begin inserting a job...\n");

    new_job->next = job_queue;

    job_queue = new_job;

    printf("after inserting a job...\n");

    pthread_mutex_unlock(&job_queue_mutex);

}

void * insert_queue_function(void * arg)

{

    int i = 0;

    while(i < 20)

    {

        sleep(1);

        printf("put the value: %d.\n", i);

        struct job* new_job = (struct job*)malloc(sizeof(struct job));

        new_job->next = NULL;

        new_job->value = i;

        enqueue_job(new_job);

        i++;

    }

}

int main()

{

    pthread_t insert_thread, process_thread;

    pthread_create(&insert_thread, NULL, &insert_queue_function, NULL);

    pthread_create(&process_thread, NULL, &process_queue_function, NULL);

    pthread_join(insert_thread, NULL);

    pthread_join(process_thread, NULL);

}

• 互斥锁有三种：

  • 默认是快锁(fast mutex)：两次锁定互斥锁会造成死锁。

  • 递归锁(recursive mutex)：允许多次锁定互斥锁，同样次数的解锁可以解除互斥锁。

  • 检错锁(error checking mutex)：第二次锁定互斥锁将返回错误码EDEADLK

• 使用互斥锁属性对象可以创建不同类型的锁：

  • pthread_mutexattr_t attr;

  • pthread_mutex_t mutex;

  • pthread_mutexattr_init (&attr);

  • pthread_mutexattr_setkind_np (&attr, PTHREAD_MUTEX_ERRORCHECK_NP);

  • pthread_mutexattr_setkind_np (&attr, PTHREAD_MUTEX_RECURSIVE_NP);

  • pthread_mutex_init (&mutex, &attr);

  • pthread_mutexattr_destroy (&attr);

 

#include <stdio.h>

#include <stdlib.h>

#include <pthread.h>

pthread_mutex_t mutex;

void * thread_function(void * arg)

{

    printf("try the mutex...\n");

    pthread_mutex_lock(&mutex);

    printf("get the mutex...\n");

    pthread_mutex_unlock(&mutex);

    printf("release the mutex...\n");

}

int main()

{

    pthread_mutexattr_t attr;

    pthread_mutex_t mutex;

    pthread_t thread;

    pthread_mutexattr_init(&attr);

    pthread_mutexattr_setkind_np(&attr, PTHREAD_MUTEX_RECURSIVE_NP);

    pthread_mutex_init(&mutex, &attr);

    pthread_mutexattr_destroy(&attr);

    pthread_mutex_lock(&mutex);

    printf("lock the first time...\n");

    pthread_mutex_lock(&mutex);

    printf("lock the second time...\n");

    pthread_create(&thread, NULL, &thread_function, NULL);

    pthread_mutex_unlock(&mutex);

    printf("unlock the first time...\n");

    sleep(5);

    pthread_mutex_unlock(&mutex);

    printf("unlock the second time...\n");

}

• pthread_mutex_trylock函数检测是否互斥锁已经被锁定，并不被阻塞。

• 如果pthread_mutex_trylock测试一个没有锁定的互斥锁，则锁定此锁，返回0。

• 如果pthread_mutex_trylock测试一个已经锁定的互斥锁，则不阻塞，立即返回错误码EBUSY。

4.2、信号量(Semaphores)

• 信号量是一种计数器，被用来同步多线程。

• Linux提供两种信号量实现：

  • 此处是POSIX标准信号量，用于线程间的通信。

  • 另外一种实现是用于进程间通信。

• 每个信号量包含一个计数器。

• 信号量包含两种操作：

  • wait操作对信号量减一。如果值已经为0，则wait操作阻塞，直到信号量变为正。

  • post操作对信号量加一。如果值上次为0，其他线程在上次的wait操作中阻塞在此信号量上，则阻塞的线程之一结束其wait操作，继续运行。

• 信号量类型sem_t

• sem_init函数用来初始化信号量

  • 第一个参数是指向sem_t变量的指针

  • 第二个参数为0

  • 第三个参数是信号量的初始值

• sem_destroy函数用来释放信号量

• sem_wait函数用来执行wait操作

• sem_post函数用来执行post操作

• 非阻塞的wait函数是sem_trywait

• sem_getvalue函数可以得到当前信号量的值。

#include <stdio.h>

#include <stdlib.h>

#include <malloc.h>

#include <pthread.h>

#include <semaphore.h>

struct job

{

    struct job* next;

    int value;

};

struct job* job_queue;

void process_job(struct job*);

pthread_mutex_t job_queue_mutex = PTHREAD_MUTEX_INITIALIZER;

sem_t job_queue_count;

void initialize_job_queue()

{

     job_queue = NULL;

     sem_init(&job_queue_count, 0, 0);

}

void * process_queue_function (void * arg)

{

    while(1)

    {

        struct job* next_job;

        sem_wait(&job_queue_count);

        pthread_mutex_lock(&job_queue_mutex);

        printf("begin removing a job...\n");

        next_job = job_queue;

        job_queue = job_queue->next;

        printf("after removing a job...\n");

        pthread_mutex_unlock(&job_queue_mutex);

        process_job(next_job);

        free(next_job);

     }

    return NULL;

}

void process_job(struct job* p)

{

    printf("The value is : %d.\n", p->value);

}

void enqueue_job(struct job* new_job)

{

    pthread_mutex_lock(&job_queue_mutex);

    printf("begin inserting a job...\n");

    new_job->next = job_queue;

    job_queue = new_job;

    sem_post(&job_queue_count);

    printf("after inserting a job...\n");

    pthread_mutex_unlock(&job_queue_mutex);

}

void * insert_queue_function(void * arg)

{

    int i = 0;

    while(i < 20)

    {

        sleep(1);

        printf("put the value: %d.\n", i);

        struct job* new_job = (struct job*)malloc(sizeof(struct job));

        new_job->next = NULL;

        new_job->value = i;

        enqueue_job(new_job);

        i++;

    }

}

int main()

{

    pthread_t insert_thread, process_thread;

    pthread_create(&insert_thread, NULL, &insert_queue_function, NULL);

    pthread_create(&process_thread, NULL, &process_queue_function, NULL);

    pthread_join(insert_thread, NULL);

    pthread_join(process_thread, NULL);

}

4.3、条件变量(Condition Variables)

• 如果一个线程A等待一个条件变量，则阻塞，直到另一个线程B触发此条件变量。

• 线程A必须在线程B触发前等待此条件变量，如果线程B在线程A等待条件变量之前触发它，则触发丢失。

• 条件变量必须和互斥锁共同使用。

• 条件变量类型pthread_cond_t

• pthread_cond_init初始化一个条件变量。

• pthread_cond_signal触发一个条件变量。一个阻塞在一个条件变量上的线程会因为此条件变量被触发而继续执行。

• pthread_cond_broadcast唤醒所有的阻塞在一个条件变量上的线程。

• pthread_cond_wait阻塞一个线程，直到被触发。

• 当pthread_cond_wait被调用的时候，和它配合使用的互斥锁必须已经锁定，此函数自动解锁互斥锁，并阻塞在条件变量上。当条件变量被触发的时候，阻塞的线程被唤醒，pthread_cond_wait自动获取互斥锁。

#include <pthread.h>

int thread_flag;

pthread_cond_t thread_flag_cv;

pthread_mutex_t thread_flag_mutex;

void initialize_flag ()

{

    /* Initialize the mutex and condition variable. */

    pthread_mutex_init (&thread_flag_mutex, NULL);

    pthread_cond_init (&thread_flag_cv, NULL);

    /* Initialize the flag value. */

    thread_flag = 0;

}

/* Calls do_work repeatedly while the thread flag is set; blocks if the flag is clear. */

void* thread_function (void* thread_arg)

{

    /* Loop infinitely. */

    while (1) {

    /* Lock the mutex before accessing the flag value. */

        pthread_mutex_lock (&thread_flag_mutex);

        while (!thread_flag)

        /* The flag is clear. Wait for a signal on the condition variable, indicating that the flag value has changed. When the signal arrives and this thread unblocks, loop and check the flag again. */

            pthread_cond_wait (&thread_flag_cv, &thread_flag_mutex);

        /* When we’ve gotten here, we know the flag must be set. Unlock the mutex. */

        pthread_mutex_unlock (&thread_flag_mutex);

        /* Do some work. */

       do_work ();

    }

    return NULL;

}

/* Sets the value of the thread flag to FLAG_VALUE. */

void set_thread_flag (int flag_value)

{

    /* Lock the mutex before accessing the flag value. */

    pthread_mutex_lock (&thread_flag_mutex);

    /* Set the flag value, and then signal in case thread_function is blocked, waiting for the flag to become set. However, thread_function can’t actually check the flag until the mutex is unlocked. */

    thread_flag = flag_value;

    pthread_cond_signal (&thread_flag_cv);

    /* Unlock the mutex. */

    pthread_mutex_unlock (&thread_flag_mutex);

}

5、Linux线程实现

• Linux中线程是用进程来实现的。

(thread-pid) Print Process IDs for Threads (The test result of fedora is not like this.)

#include <pthread.h>

#include <stdio.h>

#include <unistd.h>

void* thread_function (void* arg)

{

    fprintf (stderr, “child thread pid is %d\n”, (int) getpid ());

    /* Spin forever. */ 

    while (1);

    return NULL;

}

int main ()

{

    pthread_t thread;

    fprintf (stderr, “main thread pid is %d\n”, (int) getpid ());

    pthread_create (&thread, NULL, &thread_function, NULL);

    /* Spin forever. */

    while (1);

    return 0;

}

编译运行

% cc thread-pid.c -o thread-pid -lpthread

% ./thread-pid &

[1] 14608

main thread pid is 14608

child thread pid is 14610

% ps x

PID TTY STAT TIME COMMAND

14042 pts/9 S 0:00 bash

14608 pts/9 R 0:01 ./thread-pid

14609 pts/9 S 0:00 ./thread-pid

14610 pts/9 R 0:01 ./thread-pid

14611 pts/9 R 0:00 ps x

% kill 14608

[1]+ Terminated ./thread-pid

总共有三个进程运行thread-pid程序

第一个进程号为14608，是主函数

第二个进程号为14610，是执行thread_function的线程

还有一个进程14609是管理线程，是由Linux线程库创建的

管理线程在pthread_create第一次调用的时候被创建。

在进程外发送给此进程的信号将发送给进程的主线程

一个线程可以通过函数pthread_kill向另一个线程发送信号