第12章：线程控制（Thread Control）

一、导读

1.1 本章学习目标

第12章深入讲解线程的高级特性：线程属性配置、线程特定数据（TSD）、线程取消机制、以及线程与信号的关系。在掌握了基本的线程创建和同步之后，本章将帮助你理解如何更精细地控制线程的行为，编写更健壮、更灵活的多线程程序。

本章的核心学习目标包括：

掌握线程属性配置：分离状态、栈大小、调度策略的设置方法
理解分离线程：分离线程终止时自动释放资源，不可join
学会使用线程特定数据（TSD）：每个线程私有的全局变量
掌握线程取消机制：pthread_cancel请求取消线程，取消点的概念
理解线程与信号的关系：信号可以发送给特定线程或整个进程
了解线程的安全编程实践：如何编写可重入的、线程安全的代码

1.2 为什么本章如此重要

在实际的多线程程序开发中，经常需要更精细的线程控制：

分离线程：后台工作线程不需要其他线程等待它，分离后可以自动释放资源
线程特定数据：某些全局变量需要每个线程有独立的副本（如errno、数据库连接）
线程取消：需要终止一个正在执行的线程，但不想杀死整个进程
线程属性：需要调整线程的栈大小（防止栈溢出）或调度策略（实时应用）

此外，线程取消和信号处理是多线程编程中最复杂的部分，理解它们的语义对于编写正确的并发程序至关重要。

1.3 核心问题引导

在阅读本章时，请带着以下问题思考：

分离线程和可join线程有什么区别？分离线程终止后资源如何释放？

线程特定数据（TSD）和线程局部存储（__thread）有什么区别？

线程取消是立即终止吗？什么是取消点？如何设置线程的取消状态？

为什么pthread_cancel不保证线程一定被取消？线程如何拒绝取消？

信号可以发送给特定线程吗？如何做到？

TSD的析构函数什么时候被调用？如果线程被取消，析构函数还会执行吗？

二、核心概念详解

2.1 线程属性

线程属性（pthread_attr_t）控制线程的创建行为，包括分离状态、栈大小、调度策略等。

#include <pthread.h>

pthread_attr_t attr;
pthread_attr_init(&attr);        // 初始化
pthread_attr_destroy(&attr);     // 销毁

// 创建线程时指定属性
pthread_create(&tid, &attr, thread_func, arg);

2.1.1 分离状态

线程的分离状态（detach state）决定线程终止后如何释放资源：

可join线程（默认）：终止后资源不自动释放，需要其他线程调用pthread_join回收
分离线程：终止后资源自动释放，不能被join

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

void *thread_func(void *arg) {
    printf("Thread running\n");
    sleep(1);
    printf("Thread done\n");
    return NULL;
}

int main() {
    pthread_attr_t attr;
    pthread_attr_init(&attr);
    
    // 设置为分离状态
    pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED);
    
    pthread_t tid;
    pthread_create(&tid, &attr, thread_func, NULL);
    
    pthread_attr_destroy(&attr);
    
    // 不能join分离线程
    // pthread_join(tid, NULL);  // 错误！
    
    // 等待线程完成（实际应用中应该用其他同步机制）
    sleep(2);
    
    printf("Main done\n");
    return 0;
}

也可以在运行时将可join线程设为分离：

pthread_detach(pthread_self());  // 分离自己
// 或
pthread_detach(tid);  // 分离其他线程

分离线程的应用场景：

后台工作线程（如日志记录、心跳检测）
不需要返回结果的并发任务
大量短生命周期的线程（避免join的开销）

2.1.2 栈大小

每个线程有独立的栈空间。默认栈大小通常为几MB。如果线程使用大量局部变量或深层递归，可能需要增大栈大小。

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

void *thread_func(void *arg) {
    // 使用大量栈空间
    char buffer[1024 * 1024];  // 1MB
    printf("Thread using large stack\n");
    return NULL;
}

int main() {
    pthread_attr_t attr;
    pthread_attr_init(&attr);
    
    // 设置栈大小为4MB
    size_t stacksize = 4 * 1024 * 1024;
    pthread_attr_setstacksize(&attr, stacksize);
    
    pthread_t tid;
    pthread_create(&tid, &attr, thread_func, NULL);
    
    pthread_join(tid, NULL);
    pthread_attr_destroy(&attr);
    return 0;
}

注意：栈大小必须在创建线程前设置，创建后不能修改。过大的栈大小会浪费虚拟地址空间。

2.1.3 获取属性值

int detachstate;
pthread_attr_getdetachstate(&attr, &detachstate);
if (detachstate == PTHREAD_CREATE_DETACHED) {
    printf("Thread is detached\n");
}

size_t stacksize;
pthread_attr_getstacksize(&attr, &stacksize);
printf("Stack size: %zu bytes\n", stacksize);

2.2 线程特定数据（TSD）

线程特定数据（Thread-Specific Data, TSD）是每个线程私有的全局变量。不同的线程可以通过相同的键访问各自的数据。

#include <pthread.h>

int pthread_key_create(pthread_key_t *key, void (*destructor)(void *));
int pthread_key_delete(pthread_key_t key);

int pthread_setspecific(pthread_key_t key, const void *value);
void *pthread_getspecific(pthread_key_t key);

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

pthread_key_t key;

void destructor(void *value) {
    printf("Thread %lu: destroying data %p\n",
           (unsigned long)pthread_self(), value);
    free(value);
}

void *thread_func(void *arg) {
    int id = (int)(long)arg;
    
    // 分配线程私有数据
    int *data = malloc(sizeof(int));
    *data = id * 100;
    
    // 设置TSD
    pthread_setspecific(key, data);
    
    // 获取TSD
    int *val = pthread_getspecific(key);
    printf("Thread %d: my data = %d\n", id, *val);
    
    // 线程退出时，析构函数会被调用
    return NULL;
}

int main() {
    // 创建键，指定析构函数
    pthread_key_create(&key, destructor);
    
    pthread_t t1, t2, t3;
    pthread_create(&t1, NULL, thread_func, (void *)1);
    pthread_create(&t2, NULL, thread_func, (void *)2);
    pthread_create(&t3, NULL, thread_func, (void *)3);
    
    pthread_join(t1, NULL);
    pthread_join(t2, NULL);
    pthread_join(t3, NULL);
    
    pthread_key_delete(key);
    return 0;
}

TSD的应用场景：

每个线程的数据库连接
每个线程的错误信息（如errno）
每个线程的日志上下文
每个线程的内存池

TSD vs 线程局部存储（`__thread`）：

特性	TSD	`__thread`
标准	POSIX	GCC扩展/C11
析构函数	支持	不支持
类型限制	void*	基本类型
可移植性	好	依赖编译器

2.3 线程取消

pthread_cancel请求取消一个线程。目标线程可以选择响应或忽略取消请求。

#include <pthread.h>

int pthread_cancel(pthread_t thread);

2.3.1 取消状态和类型

线程可以设置取消状态（是否响应取消）和取消类型（何时响应）：

// 设置取消状态
int pthread_setcancelstate(int state, int *oldstate);
// state: PTHREAD_CANCEL_ENABLE（默认）或 PTHREAD_CANCEL_DISABLE

// 设置取消类型
int pthread_setcanceltype(int type, int *oldtype);
// type: PTHREAD_CANCEL_DEFERRED（默认）或 PTHREAD_CANCEL_ASYNCHRONOUS

取消类型：

延迟取消（DEFERRED）：线程在取消点（cancellation point）检查取消请求
异步取消（ASYNCHRONOUS）：线程可以在任何时刻被取消（危险！）

取消点是某些可能阻塞的系统调用，如read、write、sleep、pthread_join等。线程在这些点检查取消请求。

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

void *thread_func(void *arg) {
    printf("Thread started\n");
    
    // 设置取消类型
    pthread_setcanceltype(PTHREAD_CANCEL_DEFERRED, NULL);
    
    for (int i = 0; i < 10; i++) {
        printf("Thread: iteration %d\n", i);
        sleep(1);  // 取消点
    }
    
    printf("Thread done\n");
    return NULL;
}

int main() {
    pthread_t tid;
    pthread_create(&tid, NULL, thread_func, NULL);
    
    sleep(3);
    
    printf("Main: cancelling thread\n");
    pthread_cancel(tid);
    
    void *retval;
    pthread_join(tid, &retval);
    
    if (retval == PTHREAD_CANCELED) {
        printf("Thread was canceled\n");
    }
    
    return 0;
}

2.3.2 清理处理程序

线程被取消或调用pthread_exit时，可以执行清理函数：

#include <pthread.h>

void pthread_cleanup_push(void (*routine)(void *), void *arg);
void pthread_cleanup_pop(int execute);

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

void cleanup_handler(void *arg) {
    printf("Cleanup: %s\n", (char *)arg);
}

void *thread_func(void *arg) {
    pthread_cleanup_push(cleanup_handler, "resource 1");
    pthread_cleanup_push(cleanup_handler, "resource 2");
    
    printf("Thread doing work...\n");
    sleep(2);
    
    // 弹出并执行清理函数
    pthread_cleanup_pop(1);  // 执行
    pthread_cleanup_pop(1);  // 执行
    
    return NULL;
}

int main() {
    pthread_t tid;
    pthread_create(&tid, NULL, thread_func, NULL);
    
    sleep(1);
    pthread_cancel(tid);
    
    pthread_join(tid, NULL);
    return 0;
}

注意：pthread_cleanup_push和pthread_cleanup_pop必须成对出现，且在同一作用域内。它们通常被实现为宏，包含{和}。

2.3.3 禁用取消

某些代码段不希望被取消（如修改共享数据结构时）：

void *thread_func(void *arg) {
    // 禁用取消
    pthread_setcancelstate(PTHREAD_CANCEL_DISABLE, NULL);
    
    // 临界区：不会被取消
    modify_shared_data();
    
    // 重新启用取消
    pthread_setcancelstate(PTHREAD_CANCEL_ENABLE, NULL);
    
    // 可以继续被取消
    sleep(10);
    
    return NULL;
}

2.4 线程与信号

信号可以发送给整个进程（所有线程共享信号处理函数）或特定线程。

#include <signal.h>
#include <pthread.h>

// 向特定线程发送信号
int pthread_kill(pthread_t thread, int sig);

// 等待信号
int sigwait(const sigset_t *set, int *sig);

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

void *signal_handler_thread(void *arg) {
    sigset_t *set = (sigset_t *)arg;
    int sig;
    
    while (1) {
        // 等待信号
        sigwait(set, &sig);
        
        if (sig == SIGINT) {
            printf("Signal handler thread: caught SIGINT\n");
        } else if (sig == SIGUSR1) {
            printf("Signal handler thread: caught SIGUSR1\n");
        }
    }
    return NULL;
}

int main() {
    // 阻塞信号（所有线程继承）
    sigset_t set;
    sigemptyset(&set);
    sigaddset(&set, SIGINT);
    sigaddset(&set, SIGUSR1);
    pthread_sigmask(SIG_BLOCK, &set, NULL);
    
    // 创建专门的信号处理线程
    pthread_t sig_tid;
    pthread_create(&sig_tid, NULL, signal_handler_thread, &set);
    
    printf("Main thread: PID=%d, TID=%lu\n", getpid(),
           (unsigned long)pthread_self());
    printf("Send signals: kill -INT %d or kill -USR1 %d\n",
           getpid(), getpid());
    
    // 主线程做其他工作
    sleep(30);
    
    pthread_cancel(sig_tid);
    pthread_join(sig_tid, NULL);
    return 0;
}

多线程信号处理的推荐模式：

在主线程中阻塞所有信号

创建一个专门的线程处理信号（使用sigwait）

其他线程不处理信号

2.5 线程安全性

编写线程安全的代码需要注意以下问题：

2.5.1 可重入函数

函数在多线程环境下可以被安全地并发调用，称为可重入的（reentrant）。

不可重入的函数（访问全局/静态数据）：

strtok（使用静态缓冲区）
localtime（返回静态结构）
getpwnam（返回静态结构）

可重入版本（_r后缀）：

strtok_r
localtime_r
getpwnam_r

#include <string.h>
#include <time.h>
#include <pwd.h>

// 不可重入
char *token = strtok(str, " ");  // 使用静态变量
struct tm *tm = localtime(&t);   // 返回静态结构

// 可重入
char *saveptr;
char *token = strtok_r(str, " ", &saveptr);

struct tm tm_buf;
struct tm *tm = localtime_r(&t, &tm_buf);

2.5.2 线程安全的全局变量

某些全局变量（如errno）在每个线程中有独立的副本：

#include <errno.h>
#include <stdio.h>
#include <pthread.h>

void *thread_func(void *arg) {
    // 每个线程有自己的errno
    errno = 0;
    printf("Thread %lu: errno = %d\n",
           (unsigned long)pthread_self(), errno);
    return NULL;
}

2.5.3 线程安全的编程实践

避免全局变量：使用参数传递或TSD

使用可重入函数：优先使用_r后缀的函数

保护共享数据：使用互斥锁

避免在信号处理函数中访问共享数据

使用原子操作：对于简单的计数器，可以使用__atomic或__sync内置函数

#include <stdatomic.h>

atomic_int counter = 0;

void *thread_func(void *arg) {
    for (int i = 0; i < 1000; i++) {
        atomic_fetch_add(&counter, 1);  // 原子操作
    }
    return NULL;
}

三、实战案例

3.1 线程池实现

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

#define POOL_SIZE 4
#define QUEUE_SIZE 100

typedef void (*task_func_t)(void *arg);

typedef struct {
    task_func_t func;
    void *arg;
} Task;

typedef struct {
    Task tasks[QUEUE_SIZE];
    int head, tail, count;
    pthread_mutex_t mutex;
    pthread_cond_t not_full;
    pthread_cond_t not_empty;
    int shutdown;
} TaskQueue;

TaskQueue queue;
pthread_t workers[POOL_SIZE];

void queue_init() {
    queue.head = queue.tail = queue.count = 0;
    queue.shutdown = 0;
    pthread_mutex_init(&queue.mutex, NULL);
    pthread_cond_init(&queue.not_full, NULL);
    pthread_cond_init(&queue.not_empty, NULL);
}

void queue_push(task_func_t func, void *arg) {
    pthread_mutex_lock(&queue.mutex);
    while (queue.count == QUEUE_SIZE && !queue.shutdown) {
        pthread_cond_wait(&queue.not_full, &queue.mutex);
    }
    if (queue.shutdown) {
        pthread_mutex_unlock(&queue.mutex);
        return;
    }
    queue.tasks[queue.tail].func = func;
    queue.tasks[queue.tail].arg = arg;
    queue.tail = (queue.tail + 1) % QUEUE_SIZE;
    queue.count++;
    pthread_cond_signal(&queue.not_empty);
    pthread_mutex_unlock(&queue.mutex);
}

int queue_pop(Task *task) {
    pthread_mutex_lock(&queue.mutex);
    while (queue.count == 0 && !queue.shutdown) {
        pthread_cond_wait(&queue.not_empty, &queue.mutex);
    }
    if (queue.shutdown && queue.count == 0) {
        pthread_mutex_unlock(&queue.mutex);
        return -1;
    }
    *task = queue.tasks[queue.head];
    queue.head = (queue.head + 1) % QUEUE_SIZE;
    queue.count--;
    pthread_cond_signal(&queue.not_full);
    pthread_mutex_unlock(&queue.mutex);
    return 0;
}

void *worker_func(void *arg) {
    Task task;
    while (queue_pop(&task) == 0) {
        task.func(task.arg);
    }
    return NULL;
}

void pool_init() {
    queue_init();
    for (int i = 0; i < POOL_SIZE; i++) {
        pthread_create(&workers[i], NULL, worker_func, NULL);
    }
}

void pool_shutdown() {
    pthread_mutex_lock(&queue.mutex);
    queue.shutdown = 1;
    pthread_cond_broadcast(&queue.not_empty);
    pthread_cond_broadcast(&queue.not_full);
    pthread_mutex_unlock(&queue.mutex);
    
    for (int i = 0; i < POOL_SIZE; i++) {
        pthread_join(workers[i], NULL);
    }
    
    pthread_mutex_destroy(&queue.mutex);
    pthread_cond_destroy(&queue.not_full);
    pthread_cond_destroy(&queue.not_empty);
}

// 示例任务
void print_task(void *arg) {
    int id = (int)(long)arg;
    printf("Task %d executed by thread %lu\n", id,
           (unsigned long)pthread_self());
    usleep(100000);
}

int main() {
    pool_init();
    
    for (int i = 0; i < 20; i++) {
        queue_push(print_task, (void *)(long)i);
    }
    
    sleep(3);
    pool_shutdown();
    
    printf("All tasks completed\n");
    return 0;
}

3.2 使用TSD实现线程局部日志

#include <stdio.h>
#include <pthread.h>
#include <stdlib.h>
#include <string.h>
#include <time.h>

pthread_key_t log_key;

typedef struct {
    FILE *fp;
    char prefix[64];
} ThreadLog;

void log_destructor(void *value) {
    ThreadLog *log = (ThreadLog *)value;
    if (log) {
        if (log->fp) fclose(log->fp);
        free(log);
    }
}

void log_init(const char *prefix) {
    ThreadLog *log = malloc(sizeof(ThreadLog));
    snprintf(log->prefix, sizeof(log->prefix), "[%s] ", prefix);
    
    char filename[256];
    snprintf(filename, sizeof(filename), "thread_%lu.log",
             (unsigned long)pthread_self());
    log->fp = fopen(filename, "a");
    
    pthread_setspecific(log_key, log);
}

void log_message(const char *msg) {
    ThreadLog *log = pthread_getspecific(log_key);
    if (log && log->fp) {
        time_t now = time(NULL);
        char timebuf[64];
        strftime(timebuf, sizeof(timebuf), "%H:%M:%S", localtime(&now));
        fprintf(log->fp, "%s%s %s\n", log->prefix, timebuf, msg);
        fflush(log->fp);
    }
}

void *thread_func(void *arg) {
    int id = (int)(long)arg;
    char prefix[32];
    snprintf(prefix, sizeof(prefix), "Thread-%d", id);
    log_init(prefix);
    
    for (int i = 0; i < 5; i++) {
        char msg[64];
        snprintf(msg, sizeof(msg), "Message %d", i);
        log_message(msg);
        usleep(100000);
    }
    return NULL;
}

int main() {
    pthread_key_create(&log_key, log_destructor);
    
    pthread_t threads[3];
    for (int i = 0; i < 3; i++) {
        pthread_create(&threads[i], NULL, thread_func, (void *)(long)(i + 1));
    }
    for (int i = 0; i < 3; i++) {
        pthread_join(threads[i], NULL);
    }
    
    pthread_key_delete(log_key);
    printf("Logs written to thread_*.log files\n");
    return 0;
}

3.3 安全的线程取消

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

typedef struct {
    int *data;
    pthread_mutex_t *mutex;
} Resource;

void cleanup_resource(void *arg) {
    Resource *res = (Resource *)arg;
    printf("Cleanup: releasing resources\n");
    if (res->data) {
        free(res->data);
        res->data = NULL;
    }
    if (res->mutex) {
        pthread_mutex_unlock(res->mutex);
    }
}

void *thread_func(void *arg) {
    Resource *res = (Resource *)arg;
    
    // 注册清理函数
    pthread_cleanup_push(cleanup_resource, res);
    
    // 分配资源
    res->data = malloc(100 * sizeof(int));
    printf("Thread: allocated data\n");
    
    pthread_mutex_lock(res->mutex);
    printf("Thread: locked mutex\n");
    
    // 工作循环
    for (int i = 0; i < 10; i++) {
        printf("Thread: working... %d\n", i);
        sleep(1);  // 取消点
    }
    
    pthread_mutex_unlock(res->mutex);
    
    pthread_cleanup_pop(1);  // 执行清理
    return NULL;
}

int main() {
    pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;
    Resource res = {NULL, &mutex};
    
    pthread_t tid;
    pthread_create(&tid, NULL, thread_func, &res);
    
    sleep(3);
    printf("Main: cancelling thread\n");
    pthread_cancel(tid);
    
    pthread_join(tid, NULL);
    pthread_mutex_destroy(&mutex);
    
    printf("Main: done\n");
    return 0;
}

四、本章小结

第12章深入讲解了线程的高级控制特性。核心知识点包括：

线程属性：分离状态（自动释放资源）、栈大小（防止溢出）、调度策略。使用pthread_attr_t配置。

线程特定数据（TSD）：每个线程私有的全局变量。pthread_key_create创建键，pthread_setspecific/pthread_getspecific设置/获取值。析构函数在线程退出时调用。

线程取消：pthread_cancel请求取消线程。延迟取消在取消点检查请求。pthread_cleanup_push/pop注册清理函数。可以禁用取消保护临界区。

线程与信号：推荐模式是创建一个专门的信号处理线程，使用sigwait同步等待信号。

线程安全：使用可重入函数（_r后缀）、避免全局变量、保护共享数据、使用原子操作。

五、思考与练习

思考题

分离线程和可join线程在资源管理上有什么区别？什么场景下应该使用分离线程？

TSD的析构函数在什么情况下不会被调用？如何确保资源被正确释放？

为什么异步取消（ASYNCHRONOUS）是危险的？在什么情况下可以使用？

为什么多线程程序推荐使用专门的信号处理线程，而不是在每个线程中注册信号处理函数？

编程练习

实现一个线程池，支持动态调整线程数量。

使用TSD实现一个线程局部的随机数生成器（每个线程有独立的种子）。

编写一个程序，演示线程取消的清理处理程序如何释放资源。

实现一个多线程的日志系统：每个线程有独立的日志文件，使用TSD管理日志上下文。