Nginx之进程间的通信机制(共享内存、原子操作)

off999 2025-01-05 19:31 30 浏览 0 评论

1. 概述

详细教程资料+课件关注+后台私信；资料；两个字可以免费视频领取+文档+各大厂面试题资料内容包括：C/C++，Linux，golang,Nginx，ZeroMQ，MySQL，Redis，fastdfs，MongoDB，ZK，流媒体，CDN，P2P，K8S，Docker，TCP/IP，协程，DPDK，嵌入式等。

Linux 提供了多种进程间传递消息的方式，如共享内存、套接字、管道、消息队列、信号等，而 Nginx 框架使用了 3 种传递消息的传递方式：共享内存、套接字、信号。

Nginx 主要使用了 3 种同步方式：原子操作、信号量、文件锁。

由于 Nginx 的每个 worker 进程都会同时处理千万个请求，所以处理任何一个请求时都不应该阻塞当前进程处理后续的其他请求。

2. 共享内存

共享内存是 Linux 下提供的最基本的进程间通信方法，它通过 mmap 或者 shmgat 系统调用在内存中创建了一块连续的线性地址空间，而通过 munmap 或者 shmdt 系统调用可以释放这块内存。

虽然 mmap 可以以磁盘文件的方式映射共享内存，但在 Nginx 封装的共享内存操作方法中是没有使用到映射文件功能的。

Nginx 定义了 ngx_shm_t 结构体，用于描述一块共享内存：

typedef struct {
    /* 执行共享内存的起始地址 */
    u_char      *addr;
    /* 共享内存的长度 */
    size_t       size;
    /* 这块共享内存的名称 */
    ngx_str_t    name;
    /* 记录日志的 ngx_log_t 对象 */
    ngx_log_t   *log;
    /* 表示共享内存是否已经分配过的标志位，为 1 时表示已经存在 */
    ngx_uint_t   exists; /* unsigned exists:1 */
}ngx_shm_t;

操作 ngx_shm_t 结构体的方法有以下两个：

ngx_shm_alloc：用于分配新的共享内存；
ngx_shm_free：用于释放已经存在的共享内存。

mmap 系统调用简述

void *mmap(void *start, size_t length, int prot, int flags, 
                int fd, off_t offset);

mmap 可以将磁盘文件映射到内存中，直接操作内存时 Linux 内核将负责同步内存和磁盘文件中的数据：

fd 参数就指向需要同步的磁盘文件
offset 则代表从文件的这个偏移量开始共享。
当 flags 参数中加入 MAP_ANON 或者 MAP_ANONYMOUS 参数时表示不使用文件映射方式，这时 fd 和 offset 参数就没有意义了，也不需要传递，此时的 mmap 方法和 ngx_shm_alloc 的功能几乎完全相同。
length 参数就是将要在内存中开辟的线性地址空间大小。
prot 参数则是操作这段共享内存的方式（如只读或可读可写）。
start 参数说明希望的共享内存起始映射地址，通常设为 NULL，即由内存选择映射的起始地址。

MAP_ANON 是 MAP_ANONYMOUS 的同义词，已过时。表示不使用文件映射方式，并且共享内存被初始化为0，因此忽略 mmap 中的 fd 和 offset 参数，但是为了可移植性，当 MAP_ANONYMOUS（或 MAP_ANON）被指定时，fd 应该设置为 -1。

如下为使用 mmap 实现的 ngx_shm_alloc 方法：

ngx_int_t ngx_shm_alloc(ngx_shm_t *shm) { /* 开辟一块 shm->size 大小且可读/写的共享内存，内存首地址存放在 shm->addr 中 */ shm->addr = (u_char *)mmap(NULL, shm->size, PROT_READ|PROT_WRITE, MAP_ANON|MAP_SHARED, -1, 0); if (shm->addr == MAP_FAILED) { ngx_log_error(NGX_LOG_ALERT, shm->log, ngx_errno, "mmap(MAP_ANON|MAP_SHARED, %uz) failed", shm->size); return NGX_ERROR; } return NGX_OK; }

当不在使用共享内存时，需要调用 munmap 或者 shmdt 来释放共享内存：

start：指向共享内存的首地址
length：表示这段共享内存的长度

Nginx 的 ngx_shm_free 方法封装了该 munmap 方法

void ngx_shm_free(ngx_shm_t *shm) { if (munmap((void*) shm->addr, shm->size) == -1) { ngx_log_error(NGX_LOG_ALERT, shm->log, ngx_errno, "munmap(%p, %uz) failed", shm->addr, shm->size); } }

Nginx 各进程间共享数据的主要方式就是使用共享内存（在使用共享内存时，Nginx 一般是由 master 进程创建，在 master 进程 fork 出 worker 子进程后，所有的进程开始使用这块内存中的数据）。

Nginx 的共享内存有三种实现：

不映射文件使用 mmap 分配共享内存（即上面的代码）
以 /dev/zero 文件使用 mmap 映射共享内存
用 shmget 调用来分配共享内存

3. 原子操作

原子操作（atomic operation）指的是由多步操作组成的一个操作。如果该操作不能原子地执行，则要么执行完所有步骤，要么一步也不执行，不可能只执行所有步骤的一个子集。

typedef volatile ngx_atomic_uint_t  ngx_atomic_t;

Nginx 提供了两个方法来修改原子变量的值.

ngx_atomic_cmp_set

static ngx_inline ngx_atomic_uint_t
ngx_atomic_cmp_set(ngx_atomic_t *lock, ngx_atomic_uint_t old,
    ngx_atomic_uint_t set)

该方法会将 old 参数与原子变量 lock 的值进行比较，若相等，则将 lock 设为参数 set，同时返回 1；若不等，则直接返回 0。

ngx_atomic_fetch_add

static ngx_inline ngx_atomic_int_t
ngx_atomic_fetch_add(ngx_atomic_t *value, ngx_atomic_int_t add)

该方法会把原子变量 value 的值加上参数 add，同时返回之前 value 的值。

由于各种硬件体系架构，原子操作的实现不尽相同，如下为 Nginx 基于几个硬件体系关于原子操作的实现。

当无法实现原子操作时，就只能用 volatile 关键字在 C 语言级别上模拟原子操作了。事实上，绝大多数体系架构都支持原子操作。

ngx_atomic_cmp_set 的实现如下：

static ngx_inline ngx_atomic_uint_t
ngx_atomic_cmp_set(ngx_atomic_t *lock, ngx_atomic_uint_t old,
    ngx_atomic_uint_t set)
{
    /* 当原子变量 lock 与 old 相等时，才能把 set 设置到 lock 中 */
    if (*lock == old) {
        *lock = set;
        return 1;
    }

    /* 若 lock 与 set 不等，返回 0 */
    return 0;
}

ngx_atomic_fetch_add 的实现如下：

static ngx_inline ngx_atomic_int_t
ngx_atomic_fetch_add(ngx_atomic_t *value, ngx_atomic_int_t add)
{
    ngx_atomic_int_t  old;

    /* 将原子变量 value 加上 add 后，返回原先 value 的值 */
    old = *value;
    *value += add;

    return old;

x86 架构下的原子操作

gnu lib提供原子操作的实现
#include <stdio.h>
#include <pthread.h>
#include <stdlib.h>
 
static int count = 0;
 
 
void *test_func(void *arg)
{
        int i=0;
        for(i=0;i<20000;++i){
//                __sync_fetch_and_add(&count,1);
                count ++;
        }
        return NULL;
}
 
int main(int argc, const char *argv[])
{
        pthread_t id[20];
        int i = 0;
 
        for(i=0;i<20;++i){
                pthread_create(&id[i],NULL,test_func,NULL);
        }
 
        for(i=0;i<20;++i){
                pthread_join(id[i],NULL);
        }
 
        printf("%d\n",count);
        return 0;
}

2.利用汇编自己实现

#include <stdio.h>
#include <pthread.h>
#include <stdlib.h>
#define LOCK "lock ; "
typedef struct { volatile int counter; } atomic_t;
 
static __inline__ void atomic_inc(atomic_t *v)
 
{
 
    __asm__ __volatile__(
 
       LOCK "incl %0"
 
       :"=m" (v->counter)
 
       :"m" (v->counter));
 
}
 
static atomic_t count = {0};
void *test_func(void *arg)
{
        int i=0;
        for(i=0;i<20000;++i){
                atomic_inc(&count);
 
        }
        return NULL;
}
 
int main(int argc, const char *argv[])
{
        pthread_t id[20];
        int i = 0;
 
        for(i=0;i<20;++i){
                pthread_create(&id[i],NULL,test_func,NULL);
        }
 
        for(i=0;i<20;++i){
                pthread_join(id[i],NULL);
        }
 
        printf("%d\n",count.counter);
        return 0;
}

3.3 自旋锁

基于原子的操作，Nginx 实现了一个自旋锁。当发现锁已经被其他进程获得时，那么不会使得当前进程进入睡眠状态，始终保持进程的可执行状态，每当内核调度到这个进程执行时就持续检查是否可以获取到锁。

自旋锁主要是为多处理器操作系统而设置的，它要解决的共享资源保护场景就是进程使用锁的时间非常短（如果锁的使用时间很久，自旋锁就不合适，会占用大量的 CPU 资源）。如果使用锁的进程不太希望自己进入睡眠状态，特别它处理的是非常核心的事件时，这时就应该使用自旋锁，其实大部分情况下 Nginx 的 worker 进程最好不要进入睡眠状态，因为它非常繁忙，在这个进程的 epoll 上可能会有十万甚至百万的 TCP 连接等待着处理，进程一旦睡眠后必须等待其他事件的唤醒，这中间及其频繁的进程间切换带来的负载消耗可能无法让用户接受。

自旋锁对于单处理器操作系统来说一样是有效的，不进入睡眠状态并不意味着其他可执行状态的进程得不到执行。Linux 内核中对于每个处理器都有一个运行队列，自旋锁可以仅仅调整当前进程在运行队列中的顺序，或者调整进程的时间片，这都会为当前处理器上的其他进程提供被调度的机会，以使得锁被其他进程释放。

如下为 Nginx 实现的基于原子操作的自旋锁方法 ngx_spinlock：

void
ngx_spinlock(ngx_atomic_t *lock, ngx_atomic_int_t value, ngx_uint_t spin)
{
    ngx_uint_t i, n;
    
    // 无法获取锁时进程的代码将一直在这个循环中执行
    for ( ;; ) {
        // lock 为 0 表示锁是没有被其他进程持有的，这时将 lock 值设为 value 
        // 参数表示当前进程持有了锁
        if (*lock == 0 && ngx_atomic_cmp_set(lock, 0, value)) {
            // 获取到锁后 ngx_spinlock 方法才会返回
            return;
        }
        
        // 该变量是处理器的个数，当它大于 1 时表示处理多处理器系统中
        if (ngx_ncpu > 1) {
            // 在多处理器下，更好的做法是当前进程不要立刻"让出"正在使用的 CPU 
            // 处理器，而是等待一段时间，看看其他处理器上的进程是否会释放锁，
            // 这会减少进程间切换的次数
            for (n = 1; n < spin; n <<= 1) {
            
                // 注意，随着等待的次数越来越多，实际去检查 lock 是否被释放
                // 的频繁会越来越小。为什么？因为检查 lock 值更消耗 CPU，
                // 而执行 ngx_cpu_pause 对于 CPU 的能耗来说更为省电
                for (i = 0; i < n; i++) {
                
                    // ngx_cpu_pause 是在许多架构体系中专门为了自旋锁而提供的
                    // 指令，它会告诉CPU现在处于自旋锁等待状态，通常一些CPU
                    // 会将自己置于节能状态，降低功耗。注意，在执行
                    // ngx_cpu_pause 后，当前进程没有 "让出" 正使用的处理器
                    ngx_cpu_pasue();
                }
                
                // 检查锁是否被释放了，如果 lock 值为0且释放了锁后，就把它的值设为
                // value，当前进程持有锁成功并返回
                if (*lock == 0 && ngx_atomic_cmp_set(lock, 0, value)) {
                    return;
                }
            }
        }
        
        // 当前进程仍然处理可执行状态，但暂时"让出"处理器，使得处理器优先调度其他
        // 可执行状态的进程，这样，在进程被内核再次调度时，在 for 循环代码中可以期望
        // 其他进程释放锁。注意，不同的内核版本对于 sched_yield 系统调用的实现可能
        // 不同，但它们的目的都是暂时 "让出" 处理器
        ngx_sched_yield();
    }
}