分类： 工程语言基础

原文: 字节跳动在PGO反馈优化技术上的探索与实践

PGO(Profile-guided optimization)通常也叫做 FDO(Feedback-directed optimization)，它是一种编译优化技术，它的原理是编译器使用程序的运行时 profiling 信息，生成更高质量的代码，从而提高程序的性能。

传统的编译器优化通常借助于程序的静态分析结果以及启发式规则实现，而在被提供了运行时的 profiling 信息后，编译器可以对应用进行更好的优化。通常来说编译反馈优化能获得 10%-15% 的性能收益，对于特定特征的应用（例如使用编译反馈优化 Clang本身）收益高达 30%。

编译反馈优化通常包括以下手段：

1. Inlining，例如函数 A 频繁调用函数 B，B 函数相对小，则编译器会根据计算得出的 threshold 和 cost 选择是否将函数 B inline 到函数 A 中。
2. ICP（Indirect call promotion），如果间接调用（Call Register）非常频繁地调用同一个被调用函数，则编译器会插入针对目标地址的比较和跳转指令。使得该被调用函数后续有了 inlining 和更多被优化机会，同时增加了 icache 的命中率，减少了分支预测的失败率。
3. Register allocation，编译器能使用运行时数据做更好的寄存器分配。
4. Basic block optimization，编译器能根据基本块的执行次数进行优化，将频繁执行的基本块放置在接近的位置，从而优化 data locality，减少访存开销。
5. Size/speed optimization，编译器根据函数的运行时信息，对频繁执行的函数选择性能高于代码密度的优化策略。
6. Function layout，类似于 Basic block optimization，编译器根据 Caller/Callee 的信息，将更容易在一条执行路径上的函数放在相同的段中。
7. Condition branch optimization，编译器根据跳转信息，将更容易执行的分支放在比较指令之后，增加icache 命中率。
8. Memory intrinsics，编译器根据 intrinsics 的调用频率选择是否将其展开，也能根据 intrinsics 接收的参数优化 memcpy 等 intrinsics 的实现。

编译器需要 profiling 信息对应用进行优化，profile 的获取通常有两种方式：

• Instrumentation-based（基于插桩）
• Sample-based（基于采样）

Instrumentation

Instrumentation-based PGO 的流程分为三步骤：

• 编译器对程序源码插桩编译，生成插桩后的程序（instrumented program）。
• 运行插桩后的程序，生成 profile 文件。
• 编译器使用 profile 文件，再次对源码进行编译。

Instrumentation-based PGO 对代码插桩包括：

1. 插入计数器（counter）

• 对编译器 IR 计算 MST，计算频繁跳转的边，对不在 MST 上的边插入计数器，用于减少插桩代码对运行时性能的影响。
• 在函数入口插入计数器。

2. 插入探针（probes）

• 收集间接函数调用地址（indirect call addresses）。
• 收集部分函数的参数值。

Sampling

Sample-based PGO 的流程同样分为三步骤：

• 编译器对程序源码进行编译，生成带调试信息的程序（program with debug information）。
• 运行带调试信息的程序，使用 profiler（例如linux perf）采集运行时的性能数据。
• 编译器使用 profile 文件，再次对源码进行编译。

其中步骤2采集的数据为二进制级别采样数据（例如 linux perf 使用 perf record 命令收集得到 perf.data 文件）。二进制采样数据通常包含的是程序的 PC 值，我们需要使用工具，读取被采样程序的调试信息（例如使用 AutoFDO 等工具），将程序的原始二进制采样数据生成程序源码行号对应的采样数据，提供给编译器使用。

比较

对比 sampled-based PGO，Instrumentation-based PGO 的优点采集的性能数据较为准确，但繁琐的流程使其在业务上难以大规模落地，主要原因有以下几点：

• 应用二进制编译时间长，引入的额外编译流程影响了开发、版本发布的效率。
• 产品迭代速度快，代码更新频繁，热点信息与应用瓶颈变化快。而 instrumented-based PGO 无法使用旧版本收集的 profile 数据编译新版本，需要频繁地使用插桩后的最新版本收集性能数据。
• 插桩引入了额外的性能开销，这些性能开销会影响业务应用的性能特征，收集的 profile 不能准确地表示正常版本的性能特征，从而降低优化的效果，使得 instrumented-based PGO 的优点不再明显。

使用 Sample-based PGO 方案可以有效地解决以上问题：

• 无需引入额外的编译流程，为程序添加额外的调试信息不会明显地降低编译效率。
• Sample-based PGO 对过时的 profile 有一定的容忍性，不会对优化效果产生较大影响。
• 采样引入的额外性能开销很小，可以忽略不计，不会对业务应用的性能特征造成影响。

POSIX多线程程序设计. 作者 美 David R.Buten

条件变量是用来通知共享数据状态信息的。可以使用条件变量来通知队列已空、或队列非空、或任何其他需要由线程处理的共享数据状态。

当一个线程互斥地访问共享状态时，它可能发现在其他线程改变状态之前它什么也做不了。状态可能是对的和一致的，即没有破坏不变量，但但是线程就是对当前状态不感兴趣。例如，一个处理队列的线程发现队列为空时，它只能等待，直到有一个节点被填加进队列中。

例如，共享数据由一个互斥量保护。线程必须锁住互斥量来判定队列的当前状态，如判断队列是否为空。线程在等待之前必须释放锁(否则)其他线程就不可能插入数据)，然后等待队列状态的变化。例如，线程可以通过某种方式阻塞自己，以便插入线程能够找到它的ID并唤醒它。但是这里有一个问题，即线程是运行于解锁和阻塞之间。

如果线程仍在运行，而其他线程此时向队列中插入了一个新的元素，则其他线程无法确定是否有线程在等待新元素。等待线程已经查找了队列并发现队列为空，解锁互斥量，然后阻塞自己，所以无法知道队列不再为空。更糟糕的是，它可能没有说明它在等待队列非空，所以其他线程无法找到它的线程ID，它只有永远等下去了。解锁和等待操作必须是原子性的，以防止其他线程在该线程解锁之后、阻塞之前锁住互斥量，这样其他线程才能够唤醒它。

等待条件变量总是返回锁住的互斥量。

这就是为什么要使用条件变量的原因。条件变量是与互斥量相关、也与互斥量保护的共享数据相关的信号机制。在一个条件变量上等待会导致以下原子操作：释放相关互斥量，等待其他线程发给该条件变量的信号(唤醒一个等待者)或广播该条件变量(唤醒所有等待者)。当等待条件变量时，互斥量必须治终锁住：当线程从条件变量等待中醒来时，它重新继续锁住互斥量。

条件变量不提供互斥。需要一个互斥量来同步对共享数据(包括活等待的谓词)的访问，这就是为什么在等待条件变量时必须指定一个互斥量。通过将解锁操作与等待条件变量原子化，Pthreads系统确保了在释放互斥量和等待条件变量之间没有线程可以改变与条件变量相关的“谓词”（如队列满或者队列空）。

为什么不将互斥量作为条件变量的一部分来创建呢？首先，互斥量不仅与条件变量一起使用，而且还要单独使用；其次，通常一个互斥量可以与多个条件变量相关。例如，队列可以为空，也可以为满。虽然可以设置两个条件变量让线程等待不同的条件，但只能有一个互斥量来协调对队列头的访问。

一个条件变量应该与一个谓词相关。如果试图将一个条件变量与多个谓词相关，或者将多个条件变量与一个谓词相关，就有陷入死锁或者竞争问题的危险。只要小心使用，可能不会有什么问题，但是很容易搞混你的程序，并且通常也不值得冒险。原则是:第一，当你在多个谓词之间共享一个条件变量时，必须总是使用广播，而不是发信号；第二，信号要比广播有效。

条件变量和谓词都是程序中的共享数据。它们被多个线程使用，可能是同时使用。由于你认为条件变量和谓词总是一起被锁定的，所以容易让人记住它们总是被相同的互斥量控制。在没有锁住互斥量前就发信号或广播条件变量是可能的(合法的,通常也是合理的)，但是更安全的方式是先锁住互斥量。

下图显示了三个线程与一个条件变量交互的时间图。圆形框代表条件变量，三条线段代表三个线程的活动。

当线段进入圆形框时，既表明线程使用条件变量做了一些事。当线程对应的线段在到达圆形框中线之前停止，表明线程在等待条件变量；当线程线段到达中线之下时，表明它在发信号或广播来唤醒等待线程。

线程1等条件变量发信号，由于此时没有等待线程，所以没有任何效果。线程1然后在条件变量上等待。线程2同样在条件变量上阻塞，随后线程3发信号唤醒在条件变量上等待的线程1。线程3然后在条件变量上等待。线程1广播条件变量，唤醒线程2和线程3。随后，线程3在条件变量上等待。一段时间后，线程3的等待时间超时，线程3被唤醒。

1 创建和释放条件变量

pthread_cond_t cod = PTHREAD_COND_INITIALIZER
int pthread_cond_init (pthread_cond_t *cond,
    pthread_condattr_t *condattr);
int Pthread_cond_destroy (pthread_cond_t *ccond)

程序中由pthread_cond_t类型的变量来表示条件变量。永远不要拷贝条件变量，因为使用条件变量的备份是不可知的，这就像是打一个断线的电话号码并等待回答一样。例如，一个线程可能在等待条件变量的一个拷贝，同时其他线程可能广播或发信号给该条件变量的其他拷贝，则该等待线程就不能被唤醒。不过，可以传递条件变量的指针以使不同函数和线程可以使用它来同步。

大部分时间你可能在整个文件范围内(即不在任何函数内部)声明全局或静态类型条件变量。如果有其他文件需要使用，则使用全局(extern)类型；否则，使用静态(static)类型。如下面实例cond_static.c所示，如果声明了一个使用默认属性值的静态条件变量，则需要使用PTHREAD_COND_INITIALIZER宏初始化。

#include <pthread.h>
#include "errors.h"
/*
 * Declare a structure, with a mutex and conddition variable
 * statically initialized. This is the same asusing
 * pthread_mutex_init and pthread_cond_init, with the ddefault
 * attributes.
 */
typedef struct_my_struct_tag {
    pthread_mutex_t mutex; /* Protects access to value */
    pthread_cond_t cond; /* Signals change to value */
    int value; /* Access protected by mutex */
} my_struct_t;
my_struct_t data = {
    PTHREAD_MUTEX_INITIALIZER, PTHREAD_COND_INITIALIZER0};
int main (int argc, char *argv[]){
    return 0;
}

当声明条件变量时，要记住条件变量与相关的谓词是”链接”在一起的。建议你将一组不变量、谓词和它们的互斥量，以及一个或多个条件变量封装为一个数据结构的元素,并仔细地记录下它们之间的关系。

有时无法静态地初始化一个条件变量，例如，当使用malloc分配一个包含条件变量的结构时。这时，你需要调用pthread_cond_init来动态他初始化条件变量，如以下实例cond_dynamic.c所示。还可以动态初始化静态声明的条件变量，但是必须确保每个条件变量在使用之前初始化且仅初始化次。你可以在建立任何线程前初始化它，或者使用pthread_once。如果需要使用非默认属性初始化条件变量，必须使用动态初始化。

#include <pthread.h>
#include "errors.h"
/*
 * Define a structure, with a mutex and condittion variable.
 */
typedef struct_my_struct_tag {
    pthread_mutex_t mutex; /* Protects access to value */
    pthread_cond_t cond; /* Signals change to value */
    int value; /* Access protected by mutex */
} my_struct_t;
int main (int argc, char *argv[]) {
    my_struct_t *data;
    int status;
    data = malloc (sizeof (my_struct_t));
    if (data == NULL)
        errno_abort ("Allocate structure");
    status = pthread_mutex_init (&data->mutex, NULL);
    if (status != 0)
        err_abort (status, "Init mutex");
    status = pthread_cond_init (&data->cond, NULL);
    if (status != 0)
        err_abort (status, "Init condition");
    status = pthread_cond_destroy (&data->cond);
    if (status != 0)
        err_abort (status, "Destroy condition");
    status = pthread_mutex_destroy (&data->mutex);
    if (status != 0)
        err_abort (status, "Destroy mutex");
    (void)free (data);
    return status;
}

当动态初始化条件变量时，应该在不需要它时调用pthread_coind_destroy来释放它。不必释放一个通过PTHREAD_COND_INITIALIZER宏静不态初始化的条件变量。

当你确信没有其他线程在某条件变量上等待，或者将要等待、发信号或广播时，可以安全地释放该条件变量。判定上述情况的最好方式是在刚刚成功地广播了该条件变量、唤醒了所有等待线程的线程内，且确信不再有线程随后后使用它时安全释放。

2 等待条件变量

int pthread_cond_wait (pthread_cond_t *coned,
    pthread_mutex_t *mutex);
int pthread_cond_timedwait (pthread_cond t *cond, 
    pthread_mutex_t *mutex, struct timespec *expiration);

每个条件变量必须与一个特定的互斥量、一个谓词条件相关联。当线程等待条件变量时，它必须将相关互斥量锁住。记住，在阻塞线程之前，条件变量等待操作将解锁互斥量；而在重新返回线程之前，会再次锁住互斥量。

所有并发地(同时)等待同一个条件变量的线程必须指定同一个相关互斥量。例如，Pthreads不允许线程1使用互斥量A等待条件变量A，而线程2使用互斥量B等待条件变量A。不过，以下情况是十分合理的：线程1使用互斥量A等待条件变量A，而线程2使用互斥量A等待条件变量B。即有任何条件变量在特定时刻只能与一个互斥量相关联，而互斥量则可以同时与多个条件变量关联。

在锁住相关的互斥量之后和在等待条件变量之前，测试谓词是很重要的。如果线程发信号或广播一个条件变量，而没有线程在等待该条件变量时，则什么也没发生。如果在这之后，有线程调用pthread_cond_wait，则它将一直等待下去而无视该条件变量刚刚被广播的事实，这将意味着该线程可能永远不被唤醒。因为在线程等待条件变量之前，互斥量一直被锁住，所以，在测试谓词和等待条件变量之间无法设置谓词——互斥量被锁住，没有其他线程可以修改共享数据，包括谓词。

当线程醒来时，再次测试谓词同样重要。应该总是在循环中等待条件变量，来避免程序错误、多处理器竞争和假唤醒。以下实例cond.c，显示了如何等待条件变量。

wait_thread 线程睡眠一段时间以允许主线程在被唤醒之前条件变量等待操作，设置共享的谓词(data.value)，然后发信号给条件变量。 wait_thread线程等待的时间由hibernation变量控制，默认是1秒。

如果程序带参数运行，则将该参数解析为整数值，保存在hibernation变量中。这将控制wait_thread线程在发送条件变量的信号前等待的时间。

主线程调用pthread_cond_timedwait函数等待至多2秒(从当前时间开始)。如果hibernation变量设置为大于两秒的值，则条件变量等待操作将超时，返回ETIMEDOUT。如果hibernation变量设为2秒，则主线程和wait_thread线程发生竞争，并且每次运行的结果可能不同。如果hibernation变量设置为小于2秒，则条件变量等待操作不会超时。

#include <pthread.h>
#include "errors.h"

typedef struct_my_struct_tag {
    pthread_mutex_t mutex; /* Protects access to value */
    pthread_cond_t cond; /* Signals change to value */
    int value; /* Access protected by mutex */
} my_struct_t;
my_struct_t data = {
    PTHREAD_MUTEX_INITIALIZER, PTHREAD_COND_INITIALIZER0};

int hibernation = 1; /* Default to 1 second */
/*
 * Thread start routine. It will set the mainthread's predicate
 * and signal the condition variable.
 */
void * wait_thread (void *arg) {
    int status;
    sleep (hibernation);
    status = pthread_mutex_lock (&data.mutex);
    if (status != 0)
        err_abort (status, "Lock mutex");
    data.value = 1; /* Set predicate */
    status = pthread_cond_signal (&data.cond);
    if (status != 0)
        err_abort (status, "Signal condition");
    status = pthread_mutex_unlock (&data.mutex);
    if (status != 0)
        err_abort (status, "Unlock mutex");
    return NULL;
}

int main (int argc, char *argv[]) {
    int status;
    pthread_t wait_thread_id;
    struct timespec timeout;
    /* 
     * If an argument is specified, interpret it asthe number
     * of seconds for wait thread to sleep before signaling the
     * condition variable. You can play with this to see the
     * condition wait below time out or wake normally.
     */
    if (argc > 1)
        hibernation = atoi (argv[1]);
    /*
     * Create wait_thread.
     */
    status = pthread_create (
        &wait_thread_id, NULL, wait_thread, NULL);
    if (status != 0)
        err_abort (status, "Create wait thread");

    /*
     * Wait on the condition variable for 2 seconnds, or until
     * signaled by the wait_thread. Normally, wait_tthread
     * should signal. If you raise "hibernation" above 2
     * seconds, it will time out.
     */
    timeout.tv sec = time (NULL) + 2;
    timeout.tv nsec = 0;
    status = pthread_mutex_lock (&data.mutex);
    if (status != 0)
        err_abort (status, "Lock mutex");
    while (data.value == 0) {
        status = pthread_cond_timedwait (
            &data.cond, &data.mutex, &timeout);
        if (status == ETIMEDOUT) {
            printf ("Condition wait timed out.\n");
            break;
        }
        else if (status != 0)
            err_abort (status, "Wait on condition");
    }
    if (data.value 1= 0)
        printf ("Condition was signaled.\n");
    status = pthread_mutex_unlock (&data.mutex)1
    if (status != 0)
        err_abort (status, "Unlock mutex");
    return 0;
}

pthread_cond_wait函数是POSIX线程库中用于等待条件变量的函数之一。它的原理涉及到线程同步和互斥锁的概念。

在调用pthread_cond_wait之前，通常需要先获取一个互斥锁（pthread_mutex_lock），以确保在等待条件变量期间的线程安全性。然后，线程会检查一个条件，如果条件不满足，线程就会阻塞在pthread_cond_wait调用处，等待其他线程发出条件变量的信号。

当其他线程调用pthread_cond_signal或pthread_cond_broadcast函数时，条件变量会被发出信号。这些函数用于通知等待在条件变量上的线程，条件已经满足，或者在广播情况下，通知所有等待线程。此时，被阻塞的线程会被唤醒，并开始重新尝试获取互斥锁。

pthread_cond_wait的原理可以简述如下：

1. 线程调用pthread_mutex_lock获取互斥锁，确保线程安全。
2. 线程检查条件是否满足。如果条件满足，线程不会调用pthread_cond_wait，而是继续执行后续操作。
3. 如果条件不满足，线程调用pthread_cond_wait，释放互斥锁并进入阻塞状态，等待条件变量的信号。
4. 其他线程调用pthread_cond_signal或pthread_cond_broadcast发出条件变量的信号。
5. 被阻塞的线程被唤醒，重新尝试获取互斥锁。
6. 线程成功获取互斥锁后，继续执行后续操作。

需要注意的是，pthread_cond_wait函数的阻塞和唤醒是由操作系统内核实现的，因此具体的实现细节可能因操作系统而异。但是，POSIX线程库提供了一种标准接口，确保了跨平台的可移植性。

POSIX多线程程序设计. 作者 美 David R.Buten

大部分多线程程序需要在线程间共享数据。如果两个线程同时访问共享数据就可能会有问题，因为一个线程可能在另一个线程修改共享数据的过程中使用该数据，并认为共享数据保持未变。

使线程同步最通用和常用的方法就是确保对相同(或相关)数据的内存访问”互斥地”进行，即一次只能允许一个线程写数据，其他线程必须等待。Pthreads使用了一种特殊形式的信号灯——互斥量。互斥量(mutex)是由单词互相(mutual)的首部”mut”和排斥(eexclusion)的首部”ex”组合而成的。

同步不仅仅在修改数据时重要，当线程需要读取其他线程写入的数据时，而且数据写入的顺序也有影响时，同样需要同步。

考虑以下实例：一个线程向数组中某个元素填加新数据，并更新max_index变量以表明该数组元素有效。在另一个处理器上同时运行的处理线程，负责遍历数组并处理每个有效元素。如果处理线程在读取数组元素的新数据：之前先读取了更新后的max_index值，计算就会出错。这可能显得有些不太合理，但是以这种方式工作的内存系统比按照确定顺序访问内存的系统要快得多。互斤量是解决此类问题的通用方法：在访问共享数据的代码段周围加锁互斥量，则一次只能有一个线程进入该代码段。

下图显示了共享互斥量的三个线程的时序图。处于标有”互斥量”的圆形框之上的线段表示相关的线程没有拥有互斥量。处于圆形框中心线之下的线段表示相关的线程拥有互斥量。处于中心线之上的线段表明相关的线程等待拥有互斥量。

最初，互斥量没有被加锁。当线程1试图加锁该互斥量时，由于于没有竞争，线程1立即加锁成功，对应线段也移到中心线之下。然后线程2试图加锁互斥量，由于互斥量已经被锁住，所以线程2被阻塞，对应线段在中心线之上。线程1解锁互斥量，解除线程2的阻塞，使其对互斥量加锁成功。稍后，线程3试图加锁互斥量被阻塞。线程1调用函数pthread_mutex_trylock试着锁住互斥量，而立刻返回EBUSY。线程2解锁互斥量，解除线程3的阻塞，线程3加锁成功。最后，线程3完成工作,解锁互斥量。

1 创建和销毁互斥量

pthread_mutex_t_mutex = PTHREAD_MUTEX_INITTALIZER;
int pthread nutex init(
    pthread_mutex_t *mutex, pthread_mutexattr_t*attr);
int Pthread_mutex_destroy (Pthread_mutex_t*mutex);

程序中的互斥量是用pthread_mutex_t类型的变量来表示的。不能拷贝互斥量变量，因为使用拷贝的互斥量是不确定的。可以拷贝指向互斥量的指针，这样就可以使多个函数或线程共享互斥量来实现同步。

大部分时间你可能在”文件范围”内，即函数体外，声明互斥量为外部或静态存储类型。如果有其他文件使用互斥量，则将其声明为外部类型；如果仅在本文件内使用,则将其声明为静态类型。你应该使用宏 PTHREAD_MUTEEX_INITIALIZER 来声明具有默认属性的静态互斥量，如下面例程mutex_staltic.c所示(你可以编译并运行该程序,不过因为main函数为空,所以不会有任何结果)。

#include <pthread.h>
#include "errors,h"
/*
 * Declare a structure, with a mutex, statically initialized. This
 * is the same as using pthread_mutex_init, withthe default,
 * attributes.
 */
typedef struct my_struct_tag {
    pthread_mutex_t mutex; /* Protects access to value */
    int value; /* Access protected by mutex */
} my_struct_t;

my_struct_t data = {PTHREAD_MUTEX_INITIALIZER, 0};

int main (int argc, char *argv[])
{
    return 0;
}

通常不能静态地初始化一个互斥量，例如当使用malloc动态分配一个包含互斥量的数据结构时。这时，应该使用pthread_mutex_init调用来动态地初始化互斥量，如下面程序mutex_dynamic.c所示。也可以动态地初始化静态声明的互斥量，但必须保证每个互斥量在使用前被初始化，而且只被始化一次。可以在创建任何线程之前初始化它，如通过调用pthread_once。如果需要初始化一个非缺省属性的互斥量，必须使用动态初始化。

#include <pthread.h>
#include "errors.h"
/*
 * Define a structure, with a mutex.
 */
typedef struct my_struct_tag {
    pthread_mutex_t mutex; /* Protects access to value */
    int value; /* Access protected by mutex */
} my_struct_t;

int main (int argc, char *argv[]) {
    my_struct_t *data;
    int status;
    data = malloc (sizeof (my_struct_t));
    if (data == NULL)
        errno_abort ("Allocate structure");
    status = pthread_mutex_init (&data->mutex, NULL);
    if (status != 0)
        err abort (status, "Init mutex");
    status = pthread_mutex_destroy (&data->mutex);
    if (status != 0)
        err_abort (status, "Destroy mutex");
    (void)free (data);
    return status;
}

将互斥量与它要保护的数据明显地联系起来是个不错的注意。如果可能的话，将互斥量和数据定义在一起。例如，在mutex_static.c和mutex_dynamic.c中，互斥量和它要保护的数据就被定义在同一个数据结构中，并通过注释语句记录了这种关系。

当不再需要一个通过pthread_mutex_init调用动态初始化的互斥量时，应该调用pthread_mutex_destroy来释放它。不需要释放一个使用
PTHREAD_MUTEX_INITIALIZER 宏静态初始化的互斥量。当确信没有线程在互斥量上阻塞时,可以立刻释放它。

当知道没有线程在互斥量上阻塞，且互斥量也没有被锁住时，可以安全地释放它。获知此信息的最好方式是在刚刚解锁互斥量的线程内，程序逻辑确保随后不再有线程会加锁该互斥量。当线程在某个堆栈数据结构中锁住互斥量，以从列表中删除该结构并释放内存时，在释放互斥量占有的空间之前先将互斥量解锁和释放是个安全且不错的主意。

2 加锁和解锁互斥量

int pthread_mutex_lock (pthread_mutex_t *mutex) 
int pthread_mutex_trylock (pthread_mutex_t *mutex)
int pthread_mutex_unlock (Pthread_mutex_t *mutex)

在最简单的情况下，使用互斥量是容易的事情。通过调用pthread_mutex_lock或pthread_mutex_trylock锁住互斥量，处理共享数据，然后调用pthread_mutex_unlock解锁互斥量。为确保线程能够读取一组变量的一致的值，需要在任何读写这些变量的代码段周围锁住互斥量。

当调用线程已经锁住互斥量之后，就不能再加锁该互斥量。试图这样做的结果可能是返回错误(EDEADLK)，或者可能陷入”自死锁”，使不幸的线程永远等待下去。不能解锁一个已经解锁的互斥量，也不能解锁由其他线程锁住的互斥量。锁住的互斥量是属于加锁线程的。

下面程序alarm_mutex.c是alarm_thread.c的改进版本。它在一个alarmserver线程中处理多个闹铃的请求。

结构体alarm_t现在包含了一个标准UNIX time_t类型的绝对时间，表示从UNIX纪元(1970年1月1日00:00)开始到闹铃时的秒数。这样alarm_t结构体就可以按照闹铃时间排序,而不是按照请求的秒数排序。另外，还有一个link元素将所有请求链接起来。

互斥量alarm_mutex负责协调对闹铃请求列表alarm_list的头节点的访问。互斥量是使用默认属性调用宏 PTHREAD_MUTEX_INITIALIZER 静态初始化的。列首指针初始化为空。

#include <pthread.h>
#include <time.h>
#include "errors.h"
/*
 * The "alarm" structure now contains the tinme t (time since the
 * Epoch, in seconds) for each alarm, so thatthey can be
 * sorted. Storing the requested number of sedconds would not be
 * enough, since the "alarm thread" cannot tel11 how long it has
 * been on the list.
 */
typedef struct alarm_tag {
    struct alarm_tag *link;
    int seconds;
    time_t time; /* seconds from EPOCH */
    char message[64];
} alarm_t;

pthread_mutex_t alarm_mutex = PTHREAD_MUTEX_INITIALIZER;
alarm_t *alarm_list = NULL;

下面讲述函数alarm_thread的代码。该函数作为线程运行，并依次处理列表alarm_list中的每个闹铃请求。线程永不停止，当main函数返回时，线程”蒸发”。这种做法的惟一后果是任何剩余请求都不会被传送，线程没有保留任何能够在进程之外可见的状态。

如果希望程序在退出之前处理所有未完结的闹铃请求,可以简简单地修改程序以达到该目标。当主线程发现列表 alarm_list为空时，需要通过某种方式通知alarm_thread线程终止。例如，可以在主线程中设置一个全局变量alarm_done的值，然后调用pthread_exit而不是调用exit终止。当alarm_thread线程发现列表为空且alarm_done被置位时,它会立即调用pthread_exit，而不是等待下一个请求。

如果列表中没有新的请求，alarm_thread线程需要阻塞自己一小段时间，解锁互斥量，以便主线程能够添加新的闹铃请求。通过将sleep_time置为1秒来作到这点。

如果列表中发现请求，则将它从列表中删除。调用time函数放获得当前时间，并将其与请求时间比较。如果闹铃时间已经过期，则alarm_thread线程将sleep_time置为0；否则，alarm_thread线程计算闹铃时间与当前时间的差，并将sleep_time置为该差值(以秒为单位)。

在线程睡眠或阻塞之前，总是要解锁互斥量。如果互斥量仍被锁住，则主线程就无法向列表中增加请求。这将使程序变成同步工作方式，因为用户必须等到闹铃之后才能做其他事(用户可能输入一个命令,但是必须等到下一闹钟到期时才能获得系统提示)。调用sleep将阻塞alarm_thread线程指定的时间，直到经过该时间后线程才能运行。

调用sched_yield的效果果是将处理器交给另一个等待运行的线程，但是如果没有就绪的线程，则立即返回。在程序中,调用sched_yield意味着:如果有等待处理的用户输入，则主线程运行，处理用户请求；如果用户没有输入请求,则该函数立即返回。

如果alam指针非空，即如果已经从alarm_list列表中处理里了一个闹铃请求，则函数打印消息显示闹铃已到期。然后，线程释放alarm结构，准备处理下一个闹铃请求。

/*
 * The alarm thread's start routine.
 */
void *alarm_thread (void *arg) {
    alarm t *alarm;
    int sleep time;
    time t_now;
    int status;
    /*
    * Loop forever, processing commands. The alarmthread will
    * be disintegrated when the process exits.
    */
    while (1) {
        status = pthread_mutex_lock (&alarm_mutex);
        if (status != 0)
            err_abort (status, "Lock mutex");
        alarm = alarm_list;
        /*
         * If the alarm list is empty, wait for one second. This
         * allows the main thread to run, and readanother
         * command. If the list is not empty, remove tthe first
         * item. Compute the number of seconds to wait-- if the
         * result is less than 0 (the time has passed), t!hen set
         * the sleep time to 0.
         */
        if (alarm == NULL)
            sleep_time = 1;
        else {
            alarm_lişt = alarm->link;
            now = time (NULL);
            if (alarm->time <= now)
                sleep_time = 0;
            else
                sleep_time = alarm->time - now;
#ifdef DEBUG
            printf ("[waiting: %d(%d)\"Bs\"]\n", alarm->timne,
                sleep time, alarm->message);
#endif
        }

        /*
         * Unlock the mutex before waiting, so that thhe main
         * thread can lock it to insert a new alarm request.If
         * the sleep_time is 0, then call sched yield, giving9
         * /the main thread a chance to run if it has been n
         * readied by user input, without delaying the message
         * if there's no input.
         */
        status = pthread_mutex_unlock (salarm_mutex);
        if (status != 0)
            err_abort (status, "Unlock mutex");
        if (sleep_time > 0)
            sleep (sleep_time);
        else
            sched_yield ();
        /*
         * If a timer expired, print the message and free the
         * structure.
         */
        if (alarm != NULL) {
            printf("(td) ts\n", alarm->seconds, alarm->me(ssage)
            free (alarm);
        }
    }
}

最后,我们来讨论alarm_mutex.c的主程序代码。基本结构我们已经开发过的版本相同，包括一个循环、从stdin中读取用户输入的请求并依次处理它们。这一次,没有像alarm.c中那样同步地等待，也没有像alarm_fork.c 或alarm_thread.c 那样为每个请求创建一个异步处理实体进程或线程)，而是将所有请求排队，等待服务线程alarm_thread处理。一旦主线程将所有请求排队,它就可以读取下一个请求了。

建立一个服务线程来处理所有请求。返回的线程ID保存在局部变量thread中(尽管我们不使用它)。

与其他的闹铃版本一样读取并处理用户请求。就像在alarm_thread.c中那样，数据保存在malloc分配的堆结构中。

程序将闹铃请求添加到alarm_list列表中，该列表由主线程和alarm_thread线程共享。所以在访问共享数据之前，需要将互斥量alarm_mutex加锁。

由于线程alarm_thread串行地处理列表中的请求,所以没有办法知道从读取用户请求到处理请求的时间间隔。因此,alarm结构中包含了闹铃的绝对时间。绝对时间是通过将用户输入的闹铃时间间隔加上由time调用返回的当前时间获得。

闹铃请求在列表alarm_list中按照闹铃时间先后顺序排序。插入代码遍历列表，直到找到第一个闹铃时间大于或等于新闹铃请求时间的节点，然后将新请求插入到找到的节点前。因为alarm_list是个简单的链表，遍历维护了两个指针：一个next指针指向当前节点，一个last指针指向前一个节点的link指针或者指向列表头指针。

如果没有找到大于或等于当前闹铃时间的节点，则将新请求节点插入列表尾部。当退出遍历时，如果当前节点指针为NULL，则前一节点(或链表头)指向新请求节点。

int main (int argc, char *argv[]) {
    int status;
    char line[128];
    alarm_t *alarm, **last, *next;
    pthread_t thread;

    status = pthread_create {
        &thread, NULL, alarm_thread, NULL);
    if (status != 0)
        err_abort (status, "Create alarm thread");
    while {1) {
        printf ("alarm> ");
        if(fgets(line, sizeof(line), stdin) == NULL) exit(0)
        if (strlen (line)<= 1) continue;
        alarm = (alarm_t*)malloc (sizeof (alarm_t));
        if (alarm == NULL)
            errno_abort ("Allocate alarm");
        /*
         * Parse input line into seconds (%d) and a message
         * (864[^\n]), consisting of up to 64 charadters
         * separated from the seconds by whitespace.
         */
        if (sscanf (line, "%d 864[^\n]",
            &alarm->seconds, alarm->message) < 2) {
            fprintf (stderr, "Bad command\n");
            free (alarm);
        } else {
            status = pthread_mutex_lock (&alarm_mutex);
            if (status != 0)
                err_abort (status, "Lock mutex");
            alarm->time = time (NULL) + alarm->seconds;
            /*
             * Insert the new alarm into the list of alarmis,
             * sorted by expiration time.
             */
            last = &alarm_list;
            next = *last;
            while (next!= NULL) {
                if (next->time >= alarm->time) {
                    alarm->link = next;
                    *last = alarm;
                    break;
                }
                last = &next->link;
                next = next->link;
            }
            /*
             * If we reached the end of the list, insert tthe new
             * alarm there. ("next" is NULL, and "last" pointts
             * to the link field of the last item, or tothe
             * list header).
             */
            if (next == NULL){
                *last = alarm;
                alarm->link = NULL;
            }
#ifdef DEBUG
            printf ("[list: ");
            for (next = alarm_list; next != NULL; next = 1next->link)
                printf ("#d{%d)[\'\'\'] ", next->time,
                    next->time - time (NULL), next->message);
            printf ("]\n");
#endif
            status = pthread_mutex_unlock (&alarm_mutex);
            if (status 1= 0)
                err_abort (status, "Unlock mutex");
        }
    }
}

这个简单的例子存在几个严重的缺点。尽管与alarm_fork.c 和 alarm_thread.c相比，该实例具有占用更少资源的优势,但它的响应性能不够。一旦alarm_thread线程从列表中接收了一个闹铃请求，它就进入睡眠直到闹铃到期。当它发现列表中没有闹铃请求时，也会睡眠1秒，以允许主线程接收新的用户请求。当alarm_thread线程睡眠时，它就不能注意到由主线程添加到请求列表中的任何闹铃请求，直到它从睡眠中返回。

这个问题可以通过不同的方式解决。当然最简单的为方式就是像alarm_thread.c那样为每个闹铃请求建立一个线程。当然这也不坏，因为线程还是比较廉价的。不过还是没有alarm_t数据结构廉价，并且我们更喜欢构造高效的程序，而不仅仅是响应性好的程序。最好的办法是使用条件变量来通知共享数据的状态变化。

3 非阻塞式互斥量锁

当调用pthread_mutex_lock加锁互斥量时，如果此时互斥量已经被锁住，则调用线程将被阻塞。通常这是你希望的结果，但有时你可能希望如果互斥量已被锁住，则执行另外的代码路线，你的程序可能做其他一些有益的工作而不仅仅是等待。为此，Pthreads提供了pthread_mutex_trylock函数，当调用互斥量已被锁住时调用该函数将返回错误代码EBUSY。

使用非阻塞式互斥量加锁函数时，需要确保只有当pthread_mutex_trylock函数调用成功时，才能解锁互斥量。只有拥有互斥量的线程才能解锁它。一个错误的pthread_mutex_trylock函数调用可能返回错误代码，或者可能解锁其他线程依赖的互斥量,这将导致程序产生难以调试的错误。

4 调整互斥量满足工作

互斥量有多大？我可不是指一个pthread_mutex_t类型的结构构占了多少内存。我是使用了一种通俗的、不完全准确的、但是可以被大多数人接受的说法。这种有趣的用法是在有关如何将非线程代码改为线程安全代码的过程中流行起来的。实现线程安全的函数库的一个相对简单的做法是创建一个互斥量，在每次进入函数库时锁住它，在退出库的时候解锁它，这样函数库就变成了一个串行区域，从而防止了线程问的任何冲突。我们就把保护如此大的串行区域的互斥量称为”大”互斥量，并形而上学地认为比那些只保护几行代码的互斥量要明显地大。

进一步扩展，保护两个变量的互斥量比保护一个变量的互斥量要”大”。那么到底互斥量该多大呢?答案只能是:足够大,但不要太大。

当需要保护两个共享变量时，你有两种基本策略：可以为每个个变量指派一个”小”的互斥量，或者为两个变量指派一个”大”的互斥量。哪一种方法更好取决于很多因素。并且，在开发过程中影响因素可能发生改变，这依赖于有多少线程需要共享数据和如何使用共享变量。

以下是主要的设计因素:

• 互斥量不是免费的，需要时间来加锁和解锁。锁住较少互斥量的程序通常运行得更快。所以，互斥量应该尽量少，够用即可，每个互斥量保护的区域应则尽量大。
• 互斥量的本质是串行执行。如果很多线程需要频繁地加锁同一个互斥量，则线程的大部分时间就会在等待，这对性能是有害的。如果互斥量保护的数据(或代码)包含彼此无关的片段，则可以将大的互斥量分解解为几个小的互斥量来提高性能。这样，任意时刻需要小互斥量的线程减少，线程等待时间就会减少。所以，互斥量应该足够多(到有意义的地步)，每个互斥量保护的区域则应尽量的少。
• 上述两方面看似互相矛盾,但是这不是我们头一次遇到的情清况。一旦当你解了互斥量的性能后,就能够正确地处理它们。

在复杂的程序中，通常需要一些经验来获得正确的平衡。在大多数情况下，如果你开始使用较大的互斥量，然后当经验或性能数据告诉你哪个地方存在频繁的竞争时，你应改用较小的互斥量，则你的代码通常会更为简单。简单单是好的。除非发现问题，否则不要轻易花费太多时间优化你的代码。

另一方面，如果从一开始就能明晰你的算法必然导致频繁的竞争,则不要过于简单化。开始使用必须的互斥量和数据结构比后来增加它们容易得多。

5 使用多个互斥量

有时，一个互斥量是不够的，特别是当你的代码需要跨越软件体系内部的界限时。例如，当多个线程同时访问一个队列结构时，你需要两个互斥量，一个用来保护队列头，一个用来保护队列元素内的数据。当为多线程建立一个树型结构时，你可能需要为每个节点设置一个互斥量。

同时使用多个互斥量会导致复杂度的增加。最坏的情况就是死锁的发生，即两个线程分别锁住一个互斥量而等待对方的互斥量。

如果可以在独立的数据上使用两个分离的互斥量，那么就应该这样做。这样,通过减少线程必须等待其他线程完成数据操作(甚至是该线程不需要的数据)的时间，你的程序会最终取得成功。如果数据独立，则某个特定函数就不太可能经常需要同时加锁两个互斥量。

当数据不是完全独立的时候，情况就复杂了。如果你的程序中有一个不变量，影响着由两个互斥量保护的数据，即使该不变量很少被改变或引用，你迟早需要编写同时锁住两个互斥量的代码，来确保不变量的完整。如果一个线程锁住互斥量A后，加锁互斥量B；同时另一个线程锁住互斥量B而等待互斥量A，则你的代码就产生了一个经典的死锁现象。