Futex基础

Futex （ futex- “快速用户空间互斥量”的缩写）是Linux开发人员从IBM在2002年提出的一种机制，并于2003年底进入内核。 主要思想是提供一种更有效的方法，以最少的OS内核调用次数来同步用户线程。

在本文中，我们将回顾这些功能模块，尝试了解它们的工作原理，还将它们用作构建更高级别（我们熟悉的）同步对象的基础。

重要一点：futexes是一个相当低级的工具；仅在开发基本库（例如标准C / C ++库）时才值得直接使用它。 您不太可能需要在常规应用程序中使用futex。

动机

在futex出现之前，有必要每次都进行系统调用（例如使用semop ）来控制对多个线程的共享资源的访问，这是资源密集型的，因为每个调用都需要将上下文从用户模式切换到内核模式。 随着现代处理器中内核数量的增加以及应用软件中线程数量的增加，这已经成为一个重要的问题。 鉴于所有这些调用均未携带任何应用功能，未实现任何业务逻辑，而仅保证其余代码的正确操作，因此这甚至更具“冒犯性”。

提议向OS添加新概念“ futex”是基于一个简单的观察结果：在大多数情况下，捕获同步对象的尝试首次成功。 程序员以这样一种方式来编写软件：从锁定锁到解锁，所需的时间尽可能短，这意味着捕获另一个线程的尝试不会遇到障碍的可能性很高。 当流到达这样的“自由”同步对象时，我们可以捕获它而无需使用相对便宜的原子操作进行系统调用。 原子操作成功进行的可能性很大。

在那种罕见的情况下，当我们仍然尝试访问被另一个线程阻塞的资源时，原子操作将返回错误。 在这种情况下，我们有两个选择。 我们可以旋转用户模式的自旋锁，等待资源的释放（这将消耗CPU资源），或者让内核进入睡眠模式，等待资源的释放。 这是futex进入场景的地方。

简单使用futexes-期望和唤醒

futex系统调用结合了多种功能。 我们在这里不考虑复杂的选项（其中一些选项太复杂，甚至在官方文档中也没有描述），而是集中在FUTEX_WAIT和FUTEX_WAKE操作上。 官方文档中的描述将为您提供良好的基础：

futex（）系统调用为程序提供了一种等待特定条件变为真的方法。 通常，此系统调用在共享内存同步的上下文中使用阻塞构造。 使用futex时，主要同步操作在用户空间中执行。 用户空间程序仅在程序需要长时间进入待机模式直到条件变为真时才执行futex（）系统调用。 另外，futex（）可用于唤醒需要特定条件的进程或线程。

简而言之，futex是一种内核构造，可在发生某些情况时帮助用户代码同步线程。 一些进程（或线程）可以等待FUTEX_WAIT调用中的事件，而其他进程（或线程）可以使用FUTEX_WAKE调用这些事件。 等待有效地进行-等待线程被内核挂起，并且不使用处理器资源，直到发生预期事件时将它们唤醒。

花时间阅读整个文档。 好吧，或者至少阅读有关FUTEX_WAIT和FUTEX_WAKE的部分。

让我们看一个简单的示例 ，该示例演示了使用futex来协调两个流程的工作的基本用法。

子进程：

1. 在通用内存插槽中等待0xA
2. 将值0xB写入此插槽

父进程此时：

1. 将0xA值写入共享内存插槽
2. 等待0xB出现在其中

两个过程之间的这种“握手”。 这是代码：

int main(int argc, char** argv) { int shm_id = shmget(IPC_PRIVATE, 4096, IPC_CREAT | 0666); if (shm_id < 0) { perror("shmget"); exit(1); } int* shared_data = shmat(shm_id, NULL, 0); *shared_data = 0; int forkstatus = fork(); if (forkstatus < 0) { perror("fork"); exit(1); } if (forkstatus == 0) { //   printf("child waiting for A\n"); wait_on_futex_value(shared_data, 0xA); printf("child writing B\n"); //  0xB         *shared_data = 0xB; wake_futex_blocking(shared_data); } else { //   printf("parent writing A\n"); //  0xA         *shared_data = 0xA; wake_futex_blocking(shared_data); printf("parent waiting for B\n"); wait_on_futex_value(shared_data, 0xB); // Wait for the child to terminate. wait(NULL); shmdt(shared_data); } return 0; } 

注意POSIX调用以在进程之间分配共享内存。 我们在这里不能使用通常的内存分配，因为即使不同进程中的指针的相同地址实际上也会指向不同的内存块（每个进程唯一）。

应当注意的是，该示例与规范有所不同，因为futex最初是为了等待特定含义的更改而创建的，即“从某物变为某物”，而不是“从某物变为某物”。 为了说明这种可能性，我给出了这个示例，下面我们将考虑基本版本（在该版本上我们实现互斥体）。

这是wait_on_futex_value函数代码：

 void wait_on_futex_value(int* futex_addr, int val) { while (1) { int futex_rc = futex(futex_addr, FUTEX_WAIT, val, NULL, NULL, 0); if (futex_rc == -1) { if (errno != EAGAIN) { perror("futex"); exit(1); } } else if (futex_rc == 0) { if (*futex_addr == val) { //    return; } } else { abort(); } } } 

该函数的主要任务（实际上是futex系统调用之外）是一个循环，当我们唤醒false（对我们不感兴趣）时，我们将在该循环中运行。 如果在共享内存插槽中安装了新值，但我们没有期望，则会发生这种情况。 好吧，或者在另一个进程比我们的进程更早唤醒的情况下（这在我们的特定情况下不会发生，但是以更一般的方式是可能的）。

Futex语义非常棘手！ 如果futex地址处的值不等于传递的参数val，则FUTEX_WAIT调用将立即返回。 在我们的情况下，如果子进程去等待，而父进程在插槽中写入值0xA，则可能会发生这种情况。 在这种情况下，futex返回值EAGAIN。

这是wake_futex_blocking函数代码：

 void wake_futex_blocking(int* futex_addr) { while (1) { int futex_rc = futex(futex_addr, FUTEX_WAKE, 1, NULL, NULL, 0); if (futex_rc == -1) { perror("futex wake"); exit(1); } else if (futex_rc > 0) { return; } } } 

这是FUTEX_WAKE的阻塞包装器，无论有多少侦听器期望它，它都将快速计算并返回一个值。 在我们的示例中，这被用作“握手”的一部分，但其他用途也是可能的。

Futex是自定义代码的内核队列。

简而言之，futex是内核驱动的队列，用于解决自定义代码任务。 它允许用户代码请求内核挂起其线程的执行直到事件发生，并同时通知另一个线程该事件并唤醒所有等待该事件的线程。 前面我们提到了在用户模式下组织自旋锁的功能，可以等待某些条件得到满足。 但是，内核中的队列是更好的选择，因为它使我们免于在等待循环中执行的数十亿浪费的处理器指令。

这是有关LWN的文章“ A Futex概述和更新”中的图表：



在Linux内核代码中，futex是在kernel / futex.c文件中实现的。 内核存储一个哈希表，其中的键是地址，以便快速找到所需的队列并将调用过程添加到其中。 当然，一切都不是那么简单-毕竟，内核本身需要同步对内部数据的访问，并支持futeksov的各种其他选项。

限时等待FUTEX_WAIT

futex系统调用具有超时参数，该参数允许用户指定准备等待多长时间。 这是一个实现的完整示例 ，但这是关键部分：

 printf("child waiting for A\n"); struct timespec timeout = {.tv_sec = 0, .tv_nsec = 500000000}; while (1) { unsigned long long t1 = time_ns(); int futex_rc = futex(shared_data, FUTEX_WAIT, 0xA, &timeout, NULL, 0); printf("child woken up rc=%d errno=%s, elapsed=%llu\n", futex_rc, futex_rc ? strerror(errno) : "", time_ns() - t1); if (futex_rc == 0 && *shared_data == 0xA) { break; } } 

如果等待被延迟了500毫秒，那么futex函数将结束，在循环的下一个迭代中，我们可以对此做出反应（在屏幕上显示某些内容，写入日志，继续等待或停止）。

使用futex实现互斥

我们从以下事实开始本文：futex在实现更高级别的同步对象中具有实际用途。 让我们尝试使用它们（以及原子）来实现经典的互斥体。 下面的实现基于Ulrich Drepper撰写的文章“ Futexes are Tricky”中的代码。

对于此示例，我使用C ++，主要是为了使用C ++ 11标准中的原子。 您可以在此处找到完整的代码，但最重要的部分是：

 class Mutex { public: Mutex() : atom_(0) {} void lock() { int c = cmpxchg(&atom_, 0, 1); // If the lock was previously unlocked, there's nothing else for us to do. // Otherwise, we'll probably have to wait. if (c != 0) { do { // If the mutex is locked, we signal that we're waiting by setting the // atom to 2. A shortcut checks is it's 2 already and avoids the atomic // operation in this case. if (c == 2 || cmpxchg(&atom_, 1, 2) != 0) { // Here we have to actually sleep, because the mutex is actually // locked. Note that it's not necessary to loop around this syscall; // a spurious wakeup will do no harm since we only exit the do...while // loop when atom_ is indeed 0. syscall(SYS_futex, (int*)&atom_, FUTEX_WAIT, 2, 0, 0, 0); } // We're here when either: // (a) the mutex was in fact unlocked (by an intervening thread). // (b) we slept waiting for the atom and were awoken. // // So we try to lock the atom again. We set teh state to 2 because we // can't be certain there's no other thread at this exact point. So we // prefer to err on the safe side. } while ((c = cmpxchg(&atom_, 0, 2)) != 0); } } void unlock() { if (atom_.fetch_sub(1) != 1) { atom_.store(0); syscall(SYS_futex, (int*)&atom_, FUTEX_WAKE, 1, 0, 0, 0); } } private: // 0 means unlocked // 1 means locked, no waiters // 2 means locked, there are waiters in lock() std::atomic<int> atom_; }; 

在此代码中，cmpxhg函数是一个简单的包装器，可更方便地使用原子：

 // An atomic_compare_exchange wrapper with semantics expected by the paper's // mutex - return the old value stored in the atom. int cmpxchg(std::atomic<int>* atom, int expected, int desired) { int* ep = &expected; std::atomic_compare_exchange_strong(atom, ep, desired); return *ep; } 

此代码示例包含许多解释其操作逻辑的注释。 这不会造成伤害，因为存在很大的风险，您可能会想要编写一个稍微简单但完全不正确的版本。 至于此代码-并不是所有代码都完美。 例如，他尝试对std :: atomic类型的内部设备进行假设，将其内容转换为int *，以传递给futex调用。 通常情况并非如此。 该代码可以在Linux x64上编译并运行，但是我们不能保证与其他平台的兼容性。 为了获得它，我们需要为原子添加一个平台依赖层。 由于这不是本文的主题（并且因为您不太可能在同一个C ++模块中混合使用futex），因此我们将省略此实现。 这只是一个示范！

Glibc互斥锁和低级锁

因此，我们到了glibc实现POSIX线程的地步，其中一部分是pthread_mutex_t类型。 正如我在本文开头所说的那样，futex并不是普通开发人员所需要的。 它们由运行时库或非常专门用于实现更高级别的同步原语的东西使用。 在这种情况下，查看NPTL互斥锁的实现很有趣。 在glibc代码中，这是nptl / pthread_mutex_lock.c文件。

由于需要支持各种类型的互斥锁，因此代码非常复杂，但是如果需要，我们可以找到非常熟悉的块。 您还可以查看sysdeps / unix / sysv / linux / x86_64 / lowlevellock.h和nptl / lowlevellock.c文件。 该代码有些混乱，但是比较交换和futex调用的组合仍然很容易。

您应该已经很好地理解了systeds /nptl/lowlevellock.h文件的初始注释：

 /* Low-level locks use a combination of atomic operations (to acquire and release lock ownership) and futex operations (to block until the state of a lock changes). A lock can be in one of three states: 0: not acquired, 1: acquired with no waiters; no other threads are blocked or about to block for changes to the lock state, >1: acquired, possibly with waiters; there may be other threads blocked or about to block for changes to the lock state. We expect that the common case is an uncontended lock, so we just need to transition the lock between states 0 and 1; releasing the lock does not need to wake any other blocked threads. If the lock is contended and a thread decides to block using a futex operation, then this thread needs to first change the state to >1; if this state is observed during lock release, the releasing thread will wake one of the potentially blocked threads. .. */ 

去运行时futexes

Rantime Go不使用libc（在大多数情况下）。 因此，它不能依赖POSIX线程的实现。 而是直接调用较低级别的系统调用。 这使其成为使用futex的一个很好的例子。 由于无法调用pthread_mutex_t，因此必须编写自己的替代文件。 让我们看看如何做到这一点，让我们从对用户可见的sync.Mutex类型开始（在src / sync / mutex.go中）。

这种类型的Lock方法尝试使用原子交换操作来快速捕获锁。 如果事实证明您需要等待，它将调用runtime_SemacquireMutex，后者将调用runtime.lock。 该函数在src / runtime / lock_futex.go中定义，它声明了您可能熟悉的几个常量：

 const ( mutex_unlocked = 0 mutex_locked = 1 mutex_sleeping = 2 ... ) // Possible lock states are mutex_unlocked, mutex_locked and mutex_sleeping. // mutex_sleeping means that there is presumably at least one sleeping thread. 

runtime.lock还试图使用原子函数来捕获锁。 这是有道理的，因为在Go运行时的许多地方都调用了runtime.lock，但是在我看来，通过从Mutex.lock调用runtime.lock时删除原子函数的两个连续调用，可以通过某种方式优化代码。

如果事实证明您需要等待，则会调用与平台相关的函数futexsleep，该函数在src / runtime / os_linux.go文件中为Linux定义。 该函数使用代码FUTEX_WAIT_PRIVATE进行futex系统调用（在这种情况下，这是合适的，因为Go运行时位于一个进程中）。