前言

在《[apue] 進程控制那些事兒 》一文中，曾提到進程 ID 并不是唯一的，在整個系統(tǒng)運行期間一個進程 ID 可能會出現(xiàn)好多次。

> ./pid
fork and exec child 18687
[18687] child running
wait child 18687 return 0
fork and exec child 18688
[18688] child running
wait child 18688 return 0
fork and exec child 18689
...
wait child 18683 return 0
fork and exec child 18684
[18684] child running
wait child 18684 return 0
fork and exec child 18685
[18685] child running
wait child 18685 return 0
fork and exec child 18687
[18687] child running
wait child 18687 return 0
duplicated pid find: 18687, total 31930, elapse 8

如果一直不停的 fork 子進程，在 Linux 上大約 8 秒就會得到重復的 pid，在 macOS 上大約是一分多鐘。

...
[32765] child running
wait child 32765 return 0
fork and exec child 32766
[32766] child running
wait child 32766 return 0
fork and exec child 32767
[32767] child running
wait child 32767 return 0
fork and exec child 300
[300] child running
wait child 300 return 0
fork and exec child 313
[313] child running
wait child 313 return 0
fork and exec child 314
[314] child running
wait child 314 return 0
...

并且在 Linux 上 pid 的分配范圍是 [300, 32768)，約 3W 個；在 macOS 上是 [100，99999)，約 10W 個。

為何會產(chǎn)生這種差異？Linux 上是如何檢索并分配空閑 pid 的？帶著這個問題，找出系統(tǒng)對應的內(nèi)核源碼看個究竟。

源碼分析

和《[kernel] 帶著問題看源碼 —— setreuid 何時更新 saved-set-uid (SUID)》一樣，這里使用 bootlin 查看內(nèi)核 3.10.0 版本源碼，關(guān)于 bootlin 的簡單介紹也可以參考那篇文章。

進程 ID 是在 fork 時分配的，所以先搜索 sys_fork：

整個搜索過程大概是 sys_fork -> do_fork -> copy_process -> alloc_pid -> alloc_pidmap，下面分別說明。

copy_process

sys_fork & do_fork 都比較簡單，其中 do_fork 主要調(diào)用 copy_process 復制進程內(nèi)容，這個函數(shù)很長，直接搜索關(guān)鍵字 pid ：

 /*
 * This creates a new process as a copy of the old one,
 * but does not actually start it yet.
 *
 * It copies the registers, and all the appropriate
 * parts of the process environment (as per the clone
 * flags). The actual kick-off is left to the caller.
 */
static struct task_struct *copy_process(unsigned long clone_flags,
                    unsigned long stack_start,
                    unsigned long stack_size,
                    int __user *child_tidptr,
                    struct pid *pid,
                    int trace)
{
    int retval;
    struct task_struct *p;

    ...
    /* copy all the process information */
    retval = copy_semundo(clone_flags, p);
    if (retval)
        goto bad_fork_cleanup_audit;
    retval = copy_files(clone_flags, p);
    if (retval)
        goto bad_fork_cleanup_semundo;
    retval = copy_fs(clone_flags, p);
    if (retval)
        goto bad_fork_cleanup_files;
    retval = copy_sighand(clone_flags, p);
    if (retval)
        goto bad_fork_cleanup_fs;
    retval = copy_signal(clone_flags, p);
    if (retval)
        goto bad_fork_cleanup_sighand;
    retval = copy_mm(clone_flags, p);
    if (retval)
        goto bad_fork_cleanup_signal;
    retval = copy_namespaces(clone_flags, p);
    if (retval)
        goto bad_fork_cleanup_mm;
    retval = copy_io(clone_flags, p);
    if (retval)
        goto bad_fork_cleanup_namespaces;
    retval = copy_thread(clone_flags, stack_start, stack_size, p);
    if (retval)
        goto bad_fork_cleanup_io;

    if (pid != &init_struct_pid) {
        retval = -ENOMEM;
        pid = alloc_pid(p->nsproxy->pid_ns);
        if (!pid)
            goto bad_fork_cleanup_io;
    }

    p->pid = pid_nr(pid);
    p->tgid = p->pid;

    ...
    if (likely(p->pid)) {
        ptrace_init_task(p, (clone_flags & CLONE_PTRACE) || trace);

        if (thread_group_leader(p)) {
            if (is_child_reaper(pid)) {
                ns_of_pid(pid)->child_reaper = p;
                p->signal->flags |= SIGNAL_UNKILLABLE;
            }

            p->signal->leader_pid = pid;
            p->signal->tty = tty_kref_get(current->signal->tty);
            attach_pid(p, PIDTYPE_PGID, task_pgrp(current));
            attach_pid(p, PIDTYPE_SID, task_session(current));
            list_add_tail(&p->sibling, &p->real_parent->children);
            list_add_tail_rcu(&p->tasks, &init_task.tasks);
            __this_cpu_inc(process_counts);
        }
        attach_pid(p, PIDTYPE_PID, pid);
        nr_threads++;
    }

    ...
    return p;

bad_fork_free_pid:
    if (pid != &init_struct_pid)
        free_pid(pid);
bad_fork_cleanup_io:
    if (p->io_context)
        exit_io_context(p);
bad_fork_cleanup_namespaces:
    exit_task_namespaces(p);
bad_fork_cleanup_mm:
    if (p->mm)
        mmput(p->mm);
bad_fork_cleanup_signal:
    if (!(clone_flags & CLONE_THREAD))
        free_signal_struct(p->signal);
bad_fork_cleanup_sighand:
    __cleanup_sighand(p->sighand);
bad_fork_cleanup_fs:
    exit_fs(p); /* blocking */
bad_fork_cleanup_files:
    exit_files(p); /* blocking */
bad_fork_cleanup_semundo:
    exit_sem(p);
bad_fork_cleanup_audit:
    audit_free(p);
bad_fork_cleanup_policy:
    perf_event_free_task(p);
    if (clone_flags & CLONE_THREAD)
        threadgroup_change_end(current);
    cgroup_exit(p, 0);
    delayacct_tsk_free(p);
    module_put(task_thread_info(p)->exec_domain->module);
bad_fork_cleanup_count:
    atomic_dec(&p->cred->user->processes);
    exit_creds(p);
bad_fork_free:
    free_task(p);
fork_out:
    return ERR_PTR(retval);
}

copy_process 的核心就是各種資源的拷貝，表現(xiàn)為 copy_xxx 函數(shù)的調(diào)用，如果有對應的 copy 函數(shù)失敗了，會 goto 到整個函數(shù)末尾的 bad_fork_cleanup_xxx 標簽進行清理，copy 調(diào)用與清理順序是相反的，保證路徑上的所有資源能得到正確釋放。

在 copy_xxx 調(diào)用的末尾，搜到了一段與 pid 分配相關(guān)的代碼：

    if (pid != &init_struct_pid) {
        retval = -ENOMEM;
        pid = alloc_pid(p->nsproxy->pid_ns);
        if (!pid)
            goto bad_fork_cleanup_io;
    }

    p->pid = pid_nr(pid);
    p->tgid = p->pid;

首先判斷進程不是 init 進程才給分配 pid (參數(shù) pid 在 do_fork 調(diào)用 copy_process 時設(shè)置為 NULL，所以這里 if 條件為 true 可以進入)，然后通過 alloc_pid 為進程分配新的 pid。

在繼續(xù)分析 alloc_pid 之前，先把搜索到的另一段包含 pid 代碼瀏覽下：

    if (likely(p->pid)) {
        ptrace_init_task(p, (clone_flags & CLONE_PTRACE) || trace);

        if (thread_group_leader(p)) {
            if (is_child_reaper(pid)) {
                ns_of_pid(pid)->child_reaper = p;
                p->signal->flags |= SIGNAL_UNKILLABLE;
            }

            p->signal->leader_pid = pid;
            p->signal->tty = tty_kref_get(current->signal->tty);
            attach_pid(p, PIDTYPE_PGID, task_pgrp(current));
            attach_pid(p, PIDTYPE_SID, task_session(current));
            list_add_tail(&p->sibling, &p->real_parent->children);
            list_add_tail_rcu(&p->tasks, &init_task.tasks);
            __this_cpu_inc(process_counts);
        }
        attach_pid(p, PIDTYPE_PID, pid);
        nr_threads++;
    }

如果 pid 分配成功，將它們設(shè)置到進程結(jié)構(gòu)中以便生效，主要工作在 attach_pid，限于篇幅就不深入研究了。

alloc_pid

代碼不長，就不刪減了：

 struct pid *alloc_pid(struct pid_namespace *ns)
{
    struct pid *pid;
    enum pid_type type;
    int i, nr;
    struct pid_namespace *tmp;
    struct upid *upid;

    pid = kmem_cache_alloc(ns->pid_cachep, GFP_KERNEL);
    if (!pid)
        goto out;

    tmp = ns;
    pid->level = ns->level;
    for (i = ns->level; i >= 0; i--) {
        nr = alloc_pidmap(tmp);
        if (nr < 0)
            goto out_free;

        pid->numbers[i].nr = nr;
        pid->numbers[i].ns = tmp;
        tmp = tmp->parent;
    }

    if (unlikely(is_child_reaper(pid))) {
        if (pid_ns_prepare_proc(ns))
            goto out_free;
    }

    get_pid_ns(ns);
    atomic_set(&pid->count, 1);
    for (type = 0; type < PIDTYPE_MAX; ++type)
        INIT_HLIST_HEAD(&pid->tasks[type]);

    upid = pid->numbers + ns->level;
    spin_lock_irq(&pidmap_lock);
    if (!(ns->nr_hashed & PIDNS_HASH_ADDING))
        goto out_unlock;
    for ( ; upid >= pid->numbers; --upid) {
        hlist_add_head_rcu(&upid->pid_chain,
                &pid_hash[pid_hashfn(upid->nr, upid->ns)]);
        upid->ns->nr_hashed++;
    }
    spin_unlock_irq(&pidmap_lock);

out:
    return pid;

out_unlock:
    spin_unlock_irq(&pidmap_lock);
out_free:
    while (++i <= ns->level)
        free_pidmap(pid->numbers + i);

    kmem_cache_free(ns->pid_cachep, pid);
    pid = NULL;
    goto out;
}

代碼不長但是看得云里霧里，查找了一些相關(guān)資料，3.10 內(nèi)核為了支持容器，通過各種 namespace 做資源隔離，與 pid 相關(guān)的就是 pid_namespace 啦。這東西還可以嵌套、還可以對上層可見，所以做的很復雜，可以開一個單獨的文章去講它了。這里為了不偏離主題，暫時擱置，直接看 alloc_pidmap 完事兒，感興趣的可以參考附錄 6。

alloc_pidmap

到這里才涉及到本文核心，每一行都很重要，就不做刪減了：

static int alloc_pidmap(struct pid_namespace *pid_ns)
{
    int i, offset, max_scan, pid, last = pid_ns->last_pid;
    struct pidmap *map;

    pid = last + 1;
    if (pid >= pid_max)
        pid = RESERVED_PIDS;
    offset = pid & BITS_PER_PAGE_MASK;
    map = &pid_ns->pidmap[pid/BITS_PER_PAGE];
    /*
     * If last_pid points into the middle of the map->page we
     * want to scan this bitmap block twice, the second time
     * we start with offset == 0 (or RESERVED_PIDS).
     */
    max_scan = DIV_ROUND_UP(pid_max, BITS_PER_PAGE) - !offset;
    for (i = 0; i <= max_scan; ++i) {
        if (unlikely(!map->page)) {
            void *page = kzalloc(PAGE_SIZE, GFP_KERNEL);
            /*
             * Free the page if someone raced with us
             * installing it:
             */
            spin_lock_irq(&pidmap_lock);
            if (!map->page) {
                map->page = page;
                page = NULL;
            }
            spin_unlock_irq(&pidmap_lock);
            kfree(page);
            if (unlikely(!map->page))
                break;
        }
        if (likely(atomic_read(&map->nr_free))) {
            for ( ; ; ) {
                if (!test_and_set_bit(offset, map->page)) {
                    atomic_dec(&map->nr_free);
                    set_last_pid(pid_ns, last, pid);
                    return pid;
                }
                offset = find_next_offset(map, offset);
                if (offset >= BITS_PER_PAGE)
                    break;
                pid = mk_pid(pid_ns, map, offset);
                if (pid >= pid_max)
                    break;
            }
        }
        if (map < &pid_ns->pidmap[(pid_max-1)/BITS_PER_PAGE]) {
            ++map;
            offset = 0;
        } else {
            map = &pid_ns->pidmap[0];
            offset = RESERVED_PIDS;
            if (unlikely(last == offset))
                break;
        }
        pid = mk_pid(pid_ns, map, offset);
    }
    return -1;
}

Linux 實現(xiàn) pid 快速檢索的關(guān)鍵，就是通過位圖這種數(shù)據(jù)結(jié)構(gòu)，在系統(tǒng)頁大小為 4K 的情況下，一個頁就可以表示 4096 * 8 = 32768 個 ID，這個數(shù)據(jù)剛好是《[apue] 進程控制那些事兒 》中實測的最大進程 ID 值，看起來 Linux 只用一個內(nèi)存頁就解決了 pid 的快速檢索、分配、釋放等問題，兼顧了性能與準確性，不得不說確實精妙。

pid 范圍

繼續(xù)進行之前，先確定幾個常量的值：

/* PAGE_SHIFT determines the page size */
#define PAGE_SHIFT	12
#define PAGE_SIZE	(_AC(1,UL) << PAGE_SHIFT)
#define PAGE_MASK	(~(PAGE_SIZE-1))
    
/*
 * This controls the default maximum pid allocated to a process
 */
#define PID_MAX_DEFAULT (CONFIG_BASE_SMALL ? 0x1000 : 0x8000)

/*
 * A maximum of 4 million PIDs should be enough for a while.
 * [NOTE: PID/TIDs are limited to 2^29 ~= 500+ million, see futex.h.]
 */
#define PID_MAX_LIMIT (CONFIG_BASE_SMALL ? PAGE_SIZE * 8 : \
	(sizeof(long) > 4 ? 4 * 1024 * 1024 : PID_MAX_DEFAULT))
    
/*
 * Define a minimum number of pids per cpu.  Heuristically based
 * on original pid max of 32k for 32 cpus.  Also, increase the
 * minimum settable value for pid_max on the running system based
 * on similar defaults.  See kernel/pid.c:pidmap_init() for details.
 */
#define PIDS_PER_CPU_DEFAULT    1024
#define PIDS_PER_CPU_MIN    8


#define BITS_PER_PAGE		(PAGE_SIZE * 8)
#define BITS_PER_PAGE_MASK	(BITS_PER_PAGE-1) 
#define PIDMAP_ENTRIES		((PID_MAX_LIMIT+BITS_PER_PAGE-1)/BITS_PER_PAGE) 
    
int pid_max = PID_MAX_DEFAULT;

#define RESERVED_PIDS		300

int pid_max_min = RESERVED_PIDS + 1;
int pid_max_max = PID_MAX_LIMIT;

它們受頁大小、系統(tǒng)位數(shù)、CONFIG_BASE_SMALL 宏的影響，宏僅用于內(nèi)存受限系統(tǒng)，可以理解為總為 0。列表看下 4K、8K 頁大小與 32 位、64 位系統(tǒng)場景下各個常量的取值：

結(jié)論：

• 32 位系統(tǒng) pid 上限為 32768
• 64 位系統(tǒng) pid 上限為 4194304 (400 W+)
• 32 位系統(tǒng)只需要 1 個頁面就可以存儲所有 pid
• 64 位系統(tǒng)需要 128 個頁面存儲所有 pid，不過具體使用幾個頁面視 PAGE_SIZE 大小而定

搜索 pid_max 全局變量的引用，發(fā)現(xiàn)還有下面的邏輯：

void __init pidmap_init(void)
{
    /* Veryify no one has done anything silly */
    BUILD_BUG_ON(PID_MAX_LIMIT >= PIDNS_HASH_ADDING);

    /* bump default and minimum pid_max based on number of cpus */
    pid_max = min(pid_max_max, max_t(int, pid_max,
                PIDS_PER_CPU_DEFAULT * num_possible_cpus()));
    pid_max_min = max_t(int, pid_max_min,
                PIDS_PER_CPU_MIN * num_possible_cpus());
    pr_info("pid_max: default: %u minimum: %u\n", pid_max, pid_max_min);

    init_pid_ns.pidmap[0].page = kzalloc(PAGE_SIZE, GFP_KERNEL);
    /* Reserve PID 0. We never call free_pidmap(0) */
    set_bit(0, init_pid_ns.pidmap[0].page);
    atomic_dec(&init_pid_ns.pidmap[0].nr_free);
    init_pid_ns.nr_hashed = PIDNS_HASH_ADDING;

    init_pid_ns.pid_cachep = KMEM_CACHE(pid,
            SLAB_HWCACHE_ALIGN | SLAB_PANIC);
}

重點看 pid_max & pid_max_min，它們會受系統(tǒng) CPU 核數(shù)影響，對于我測試機：

> uname -p
x86_64
> getconf PAGE_SIZE
4096
> cat /proc/cpuinfo | grep 'processor' | wc -l
2
> cat /proc/cpuinfo | grep 'cpu cores' | wc -l
2

為 64 位系統(tǒng)，頁大小 4K，共有 2 * 2 = 4 個核，PID_MAX_LIMIT = 4194304、PID_MAX_DEFAULT = 32768、pid_max_cores (按核數(shù)計算的 PID_MAX 上限) 為 1024 * 4 = 4096、pid_min_cores (按核數(shù)計算的 PID_MAX 下限) 為 8 *４= 32；初始化時 pid_max = 32768、pid_max_max = 4194304、pid_max_min = 301；經(jīng)過 pidmap_init 后，pid_max 被設(shè)置為 min (pid_max_max, max (pid_max, pid_max_cores)) = 32768、pid_max_min 被設(shè)置為 max (pid_max_min, pid_min_cores) = 301。

這里有一行 pr_info 打印了最終的 pid_max & pid_max_min 的值，通過 dmesg 查看：

> dmesg | grep pid_max
[    0.621979] pid_max: default: 32768 minimum: 301

與預期相符。

CPU 核數(shù)超過多少時會影響 pid_max 上限？簡單計算一下： 32768 / 1024 = 32。當總核數(shù)超過 32 時，pid_max 的上限才會超過 32768；CPU 核數(shù)超過多少時會影響 pid_max 下限？301 / 4 = 75，當總核數(shù)超過 75 時，pid_max 的下限才會超過 301。下表列出了 64 位系統(tǒng) 4K 頁面不同核數(shù)對應的 pid max 的上下限值：

可見雖然 pid_max 能到 400W+，實際根據(jù)核數(shù)計算的話沒有那么多，pidmap 數(shù)組僅占用個位數(shù)的槽位。

另外 pid_max 也可以通過 proc 文件系統(tǒng)調(diào)整：

> su
Password:
$ echo 131072 > /proc/sys/kernel/pid_max
$ cat /proc/sys/kernel/pid_max
131072
$ suspend

[1]+  Stopped                 su
> ./pid
...
[20004] child running
wait child 20004 return 0
duplicated pid find: 20004, total 129344, elapse 74

經(jīng)過測試，未調(diào)整前使用測試程序僅能遍歷 31930 個 pid，調(diào)整到 131072 后可以遍歷 129344 個 pid，看來是實時生效了。

搜索相關(guān)的代碼，發(fā)現(xiàn)在 kernel/sysctl.c 中有如下邏輯：

static struct ctl_table kern_table[] = {
    ...
    {
        .procname= "pid_max",
        .data= &pid_max,
        .maxlen= sizeof (int),
        .mode= 0644,
        .proc_handler= proc_dointvec_minmax,
        .extra1= &pid_max_min,
        .extra2= &pid_max_max,
    },
    ...
    { }
};

看起來 proc 文件系統(tǒng)是搭建在 ctl_table 數(shù)組之上，后者直接包含了要被修改的全局變量地址，實現(xiàn)"實時"修改。而且，ctl_table 還通過 pid_max_min & pid_max_max 的值標識了修改的范圍，如果輸入超出了范圍將返回錯誤：

$ echo 300 > /proc/sys/kernel/pid_max
bash: echo: write error: Invalid argument
$ echo 4194305 > /proc/sys/kernel/pid_max
bash: echo: write error: Invalid argument

可以實時修改 pid_max  的另外一個原因還與 PIDMAP_ENTRIES 有關(guān)，詳情見下節(jié)。

最后補充一點，pidmap_init 是在 start_kernel 中調(diào)用的，后者又被 BIOS setup 程序所調(diào)用，整體調(diào)用鏈是這樣：

boot/head.S -> start_kernel -> pidmap_init

start_kernel 中就是一堆 xxx_init 初始化調(diào)用：

 asmlinkage void __init start_kernel(void)
{
    char * command_line;
    extern const struct kernel_param __start___param[], __stop___param[];

    /*
     * Need to run as early as possible, to initialize the
     * lockdep hash:
     */
    lockdep_init();
    smp_setup_processor_id();
    debug_objects_early_init();

    /*
     * Set up the the initial canary ASAP:
     */
    boot_init_stack_canary();

    cgroup_init_early();

    local_irq_disable();
    early_boot_irqs_disabled = true;

    /*
     * Interrupts are still disabled. Do necessary setups, then
     * enable them
     */
    boot_cpu_init();
    page_address_init();
    pr_notice("%s", linux_banner);
    setup_arch(&command_line);
    mm_init_owner(&init_mm, &init_task);
    mm_init_cpumask(&init_mm);
    setup_command_line(command_line);
    setup_nr_cpu_ids();
    setup_per_cpu_areas();
    smp_prepare_boot_cpu();/* arch-specific boot-cpu hooks */

    build_all_zonelists(NULL, NULL);
    page_alloc_init();

    pr_notice("Kernel command line: %s\n", boot_command_line);
    parse_early_param();
    parse_args("Booting kernel", static_command_line, __start___param,
            __stop___param - __start___param,
            -1, -1, &unknown_bootoption);

    jump_label_init();

    /*
     * These use large bootmem allocations and must precede
     * kmem_cache_init()
     */
    setup_log_buf(0);
    pidhash_init();
    vfs_caches_init_early();
    sort_main_extable();
    trap_init();
    mm_init();

    /*
     * Set up the scheduler prior starting any interrupts (such as the
     * timer interrupt). Full topology setup happens at smp_init()
     * time - but meanwhile we still have a functioning scheduler.
     */
    sched_init();
    /*
     * Disable preemption - early bootup scheduling is extremely
     * fragile until we cpu_idle() for the first time.
     */
    preempt_disable();
    if (WARN(!irqs_disabled(), "Interrupts were enabled *very* early, fixing it\n"))
        local_irq_disable();
    idr_init_cache();
    perf_event_init();
    rcu_init();
    tick_nohz_init();
    radix_tree_init();
    /* init some links before init_ISA_irqs() */
    early_irq_init();
    init_IRQ();
    tick_init();
    init_timers();
    hrtimers_init();
    softirq_init();
    timekeeping_init();
    time_init();
    profile_init();
    call_function_init();
    WARN(!irqs_disabled(), "Interrupts were enabled early\n");
    early_boot_irqs_disabled = false;
    local_irq_enable();

    kmem_cache_init_late();

    /*
     * HACK ALERT! This is early. We're enabling the console before
     * we've done PCI setups etc, and console_init() must be aware of
     * this. But we do want output early, in case something goes wrong.
     */
    console_init();
    if (panic_later)
        panic(panic_later, panic_param);

    lockdep_info();

    /*
     * Need to run this when irqs are enabled, because it wants
     * to self-test [hard/soft]-irqs on/off lock inversion bugs
     * too:
     */
    locking_selftest();

#ifdef CONFIG_BLK_DEV_INITRD
    if (initrd_start && !initrd_below_start_ok &&
            page_to_pfn(virt_to_page((void *)initrd_start)) < min_low_pfn) {
        pr_crit("initrd overwritten (0x%08lx < 0x%08lx) - disabling it.\n",
                page_to_pfn(virt_to_page((void *)initrd_start)),
                min_low_pfn);
        initrd_start = 0;
    }
#endif
    page_cgroup_init();
    debug_objects_mem_init();
    kmemleak_init();
    setup_per_cpu_pageset();
    numa_policy_init();
    if (late_time_init)
        late_time_init();
    sched_clock_init();
    calibrate_delay();
    pidmap_init();
    anon_vma_init();
#ifdef CONFIG_X86
    if (efi_enabled(EFI_RUNTIME_SERVICES))
        efi_enter_virtual_mode();
#endif
    thread_info_cache_init();
    cred_init();
    fork_init(totalram_pages);
    proc_caches_init();
    buffer_init();
    key_init();
    security_init();
    dbg_late_init();
    vfs_caches_init(totalram_pages);
    signals_init();
    /* rootfs populating might need page-writeback */
    page_writeback_init();
#ifdef CONFIG_PROC_FS
    proc_root_init();
#endif
    cgroup_init();
    cpuset_init();
    taskstats_init_early();
    delayacct_init();

    check_bugs();

    acpi_early_init(); /* before LAPIC and SMP init */
    sfi_init_late();

    if (efi_enabled(EFI_RUNTIME_SERVICES)) {
        efi_late_init();
        efi_free_boot_services();
    }

    ftrace_init();

    /* Do the rest non-__init'ed, we're now alive */
    rest_init();
}

類似 Linux 0.11 中的 main。

pid 分配

先看看 pid 在 Linux 中是如何存放的：

struct pidmap {
    atomic_t nr_free;
    void *page;
};

struct pid_namespace {
    ...
    struct pidmap pidmap[PIDMAP_ENTRIES];
    int last_pid;
    ...
};

做個簡單說明：

• pidmap.page 指向分配的內(nèi)存頁
• pidmap.nr_free 表示空閑的 pid 數(shù)量，如果為零就表示分配滿了，不必浪費時間檢索
• pid_namespace.pidmap 數(shù)組用于存儲多個 pidmap，數(shù)組大小是固定的，以 64 位 4K 頁面計算是 128；實際并不分配這么多，與上一節(jié)中的 pid_max 有關(guān)，并且是在分配 pid 時才分配相關(guān)的頁面，屬于懶加載策略，這也是上一節(jié)可以實時修改 pid_max 值的原因之一
• pid_namespace.last_pid 用于記錄上次分配位置，方便下次繼續(xù)檢索空閑 pid

下面進入代碼。

初始化

    pid = last + 1;
    if (pid >= pid_max)
        pid = RESERVED_PIDS;
    offset = pid & BITS_PER_PAGE_MASK;
    map = &pid_ns->pidmap[pid/BITS_PER_PAGE];

函數(shù)開頭，已經(jīng)完成了下面的工作：

• 將起始檢索位置設(shè)置為 last 的下個位置、達到最大位置時回卷 (pid)
• 確定起始 pid 所在頁面 (map)
• 確定起始 pid 所在頁中的位偏移 (offset)

這里簡單補充一點位圖的相關(guān)操作：

• pid / BITS_PER_PAGE：獲取 bit 所在位圖的索引，對于測試機這里總為 0 (只分配一個內(nèi)存頁)；
• pid & BITS_PER_PAGE_MAX：獲取 bit 在位圖內(nèi)部偏移，與操作相當于取余，而性能更好

經(jīng)過處理，可使用 pid_ns->pidmap[map].page[offset] 定位這個 pid (注：page[offset] 是種形象的寫法，表示頁面中第 N 位，實際需要使用位操作宏)。

遍歷頁面

    max_scan = DIV_ROUND_UP(pid_max, BITS_PER_PAGE) - !offset;
    for (i = 0; i <= max_scan; ++i) {
        if (unlikely(!map->page)) {
            ...
        }
        if (likely(atomic_read(&map->nr_free))) {
            ...
        }
        if (map < &pid_ns->pidmap[(pid_max-1)/BITS_PER_PAGE]) {
            ++map;
            offset = 0;
        } else {
            map = &pid_ns->pidmap[0];
            offset = RESERVED_PIDS;
            if (unlikely(last == offset))
                break;
        }
        pid = mk_pid(pid_ns, map, offset);
    }

做個簡單說明：

• 外層 for 循環(huán)用來遍歷 pidmap 數(shù)組，對于測試機遍歷次數(shù) max_scan == 1，會遍歷兩遍
  • 第一遍是 (last_pid, max_pid)
  • 第二遍是 (RESERVED_PIDS, last_pid]
  • 保證即使 last_pid 位于頁面中間，也能完整的遍歷整個 bitmap
• 第一個 if 用于首次訪問時分配內(nèi)存頁
• 第二個 if 用于當前 pidmap 內(nèi)搜索空閑 pid
• 第三個 if 用于判斷是否遍歷到 pidmap 數(shù)組末尾。注意 map 是個 pidmap 指針，所以需要對比地址；(pid_max-1)/BITS_PER_PAGE 就是最后一個有效 pidmap 的索引
  • 若未超過末尾，遞增 map 指向下一個 pidmap，重置 offset 為 0
  • 若超過末尾，回卷 map 指向第一個 pidmap，offset 設(shè)置為 RESERVED_PIDS
    • 若回卷后到了之前遍歷的位置 (last)，說明所有 pid 均已耗盡，退出外層 for 循環(huán)
• 根據(jù)新的位置生成 pid 繼續(xù)上面的嘗試

對于回卷后 offset = RESERVED_PIDS 有個疑問——是否設(shè)置為 pid_max_min 更為合理？否則打破了之前設(shè)置 pid_max_min 的努力，特別是當 CPU 核數(shù)大于 75 時，pid_max_min 是有可能超過 300 的。

列表考察下“不同的頁面數(shù)” & “pid 是否位于頁面第一個位置” (offset == 0) 對于多次遍歷的影響：

表中根據(jù)頁面數(shù)和 offset 推算出了 max_scan 的值，從而得到遍歷次數(shù)，example 列每一子列都是一個獨立的用例，其中：N-rear 表示第 N 頁的后半部分，N-front 表示前半部分，不帶后綴的就是整頁遍歷。逗號分隔的數(shù)字表示一個可能的頁面遍歷順序。

從表中可以觀察到，當 offset == 0 時，整個頁面是從頭到尾遍歷的，不需要多一次遍歷；而當 offset > 0 時，頁面是從中間開始遍歷的，需要多一次遍歷。這就是代碼 - !offset 蘊藏的奧妙：當 offset == 0 時會減去一次多余的遍歷！

下面考察下第一次進入的場景 (以測試機為例)：

/*
 * PID-map pages start out as NULL, they get allocated upon
 * first use and are never deallocated. This way a low pid_max
 * value does not cause lots of bitmaps to be allocated, but
 * the scheme scales to up to 4 million PIDs, runtime.
 */
struct pid_namespace init_pid_ns = {
    .kref = {
        .refcount       = ATOMIC_INIT(2),
    },
    .pidmap = {
        [ 0 ... PIDMAP_ENTRIES-1] = { ATOMIC_INIT(BITS_PER_PAGE), NULL }
    },
    .last_pid = 0,
    .level = 0,
    .child_reaper = &init_task,
    .user_ns = &init_user_ns,
    .proc_inum = PROC_PID_INIT_INO,
};
EXPORT_SYMBOL_GPL(init_pid_ns);

last_pid 初始化為 0，所以初始 pid = 1，offset != 0，遍歷次數(shù)為 2。不過因為是首次分配，找到第一個空閑的 pid 就會返回，不會真正遍歷 2 次。這里我有個疑惑：空閑的 pid 會返回 < RESERVED_PIDS 的值嗎？這與觀察到的現(xiàn)象不符，看起來有什么地方設(shè)置了 last_pid，使其從 RESERVED_PIDS 開始，不過搜索整個庫也沒有找到與 RESERVED_PIDS、pid_max_min、last_pid 相關(guān)的代碼，暫時存疑。

再考察運行中的情況，offset > 0，遍歷次數(shù)仍然為 2，會先遍歷后半部分，如沒有找到空閑 pid，設(shè)置 offset = RESERVED_PIDS、同頁面再進行第 2 次遍歷，此時遍歷前半部分，符合預期。

多頁面的情況與此類似，就不再推理了。

頁面分配

        if (unlikely(!map->page)) {
            void *page = kzalloc(PAGE_SIZE, GFP_KERNEL);
            /*
             * Free the page if someone raced with us
             * installing it:
             */
            spin_lock_irq(&pidmap_lock);
            if (!map->page) {
                map->page = page;
                page = NULL;
            }
            spin_unlock_irq(&pidmap_lock);
            kfree(page);
            if (unlikely(!map->page))
                break;
        }

之前講過，頁面采用懶加載策略，所以每次進來得先判斷下內(nèi)存頁是否分配，如果未分配，調(diào)用 kzalloc 進行分配，注意在設(shè)置 map->page 時使用了自旋鎖保證多線程安全性。若分配頁面成功但設(shè)置失敗，釋放內(nèi)存頁面，直接使用別人分配好的頁面；若頁面分配失敗，則直接中斷外層 for 循環(huán)、失敗退出。

頁內(nèi)遍歷

        if (likely(atomic_read(&map->nr_free))) {
            for ( ; ; ) {
                if (!test_and_set_bit(offset, map->page)) {
                    atomic_dec(&map->nr_free);
                    set_last_pid(pid_ns, last, pid);
                    return pid;
                }
                offset = find_next_offset(map, offset);
                if (offset >= BITS_PER_PAGE)
                    break;
                pid = mk_pid(pid_ns, map, offset);
                if (pid >= pid_max)
                    break;
            }
        }

檢查 map->nr_free 字段，若大于 0 表示還有空閑 pid，進入頁面查找，否則跳過。第一次分配頁面時會將內(nèi)容全部設(shè)置為 0，但 nr_free 是在另外的地方初始化的：

    .pidmap = {
        [ 0 ... PIDMAP_ENTRIES-1] = { ATOMIC_INIT(BITS_PER_PAGE), NULL }
    },

它將被設(shè)置為 BITS_PER_PAGE，對于 4K 頁面就是 32768。接下來通過兩個宏進行空閑位查找：test_and_set_bit & find_next_offset，前者是一個位操作宏，后者也差不多：

#define find_next_offset(map, off)					\
		find_next_zero_bit((map)->page, BITS_PER_PAGE, off)

委托給 find_next_zero_bit，這個位操作函數(shù)。定義位于匯編語言中，太過底層沒有貼上來，不過看名稱應該能猜個七七八八。因為是整數(shù)位操作，可以使用一些類似 atomic 的手段保證多線程安全，所以這里沒有施加額外的鎖，例如對于 test_and_set_bit 來說，返回 0 就是設(shè)置成功，那就能保證同一時間沒有其它線程在設(shè)置同一個比特位，是線程安全的；反之，返回 1 表示已有其它線程占了這個坑，咱們就只能繼續(xù)“負重前行”了~

對于占坑成功的線程，atomic_dec 減少空閑 nr_free 數(shù)，注意在占坑和減少計數(shù)之間還是有其它線程插進來的可能，這會導致插入線程以為有坑位實際上沒有，從而白遍歷一遍。不過這樣做不會產(chǎn)生錯誤結(jié)果，且這個間隔也比較短，插進來的機率并不高，可以容忍。

在返回新 pid 之前記得更新 pid_namespace.last_pid：

/*
 * We might be racing with someone else trying to set pid_ns->last_pid
 * at the pid allocation time (there's also a sysctl for this, but racing
 * with this one is OK, see comment in kernel/pid_namespace.c about it).
 * We want the winner to have the "later" value, because if the
 * "earlier" value prevails, then a pid may get reused immediately.
 *
 * Since pids rollover, it is not sufficient to just pick the bigger
 * value.  We have to consider where we started counting from.
 *
 * 'base' is the value of pid_ns->last_pid that we observed when
 * we started looking for a pid.
 *
 * 'pid' is the pid that we eventually found.
 */
static void set_last_pid(struct pid_namespace *pid_ns, int base, int pid)
{
    int prev;
    int last_write = base;
    do {
        prev = last_write;
        last_write = cmpxchg(&pid_ns->last_pid, prev, pid);
    } while ((prev != last_write) && (pid_before(base, last_write, pid)));
}

/*
 * If we started walking pids at 'base', is 'a' seen before 'b'?
 */
static int pid_before(int base, int a, int b)
{
    /*
     * This is the same as saying
     *
     * (a - base + MAXUINT) % MAXUINT < (b - base + MAXUINT) % MAXUINT
     * and that mapping orders 'a' and 'b' with respect to 'base'.
     */
    return (unsigned)(a - base) < (unsigned)(b - base);
}

更新也得考慮線程競爭的問題：這里在判斷 compare_exchange 的返回值之外，還判斷了新的 last_pid (last_write) 和給定的 pid 參數(shù)哪個距離原 last_pid (base) 更遠，只設(shè)置更遠的那個，從而保證在競爭后，last_pid 能反應更真實的情況。

內(nèi)層 for 是無窮循環(huán)且 offset 單調(diào)增長，需要一個結(jié)束條件，這就是 offset > BITS_PER_PAGE；另外一個條件是pid >= pid_max，這個主要用于 max_pid 不是整數(shù)頁面的情況，例如 43 個 CPU 核對應的 pid_max = 44032，占用 2 個內(nèi)存頁且第二頁并不完整 (44032 - 32768 = 11264，< 32768)，此時就需要通過 pid 來終止內(nèi)層遍歷了。為此需要根據(jù)最新 offset 更新當前遍歷的 pid：

static inline int mk_pid(struct pid_namespace *pid_ns,
        struct pidmap *map, int off)
{
    return (map - pid_ns->pidmap)*BITS_PER_PAGE + off;
}

細心的讀者可能發(fā)現(xiàn)了，對于 pid 位于頁面中間的場景，回卷后第二次遍歷該頁面時，仍然是從頭遍歷到尾，沒有在中間提前結(jié)束 (last_pid)，多遍歷了 N-rear 這部分。

對于這一點，我是這樣理解的：這一點點浪費其實微不足道，多寫幾個 if 判斷節(jié)約的 CPU 時間可能還補償不了指令流水被打斷造成的性能損失。

pid 釋放

進程結(jié)束時釋放 pid，由于之前說過的原因，Linux 支持容器需要對 pid 進行 namespace 隔離，導致這一塊前期的邏輯有點偏離主題 (且沒太看懂)，就看看具體的 pid 釋放過程得了：

static void free_pidmap(struct upid *upid)
{
    int nr = upid->nr;
    struct pidmap *map = upid->ns->pidmap + nr / BITS_PER_PAGE;
    int offset = nr & BITS_PER_PAGE_MASK;

    clear_bit(offset, map->page);
    atomic_inc(&map->nr_free);
}

還是經(jīng)典的 nr / BITS_PER_PAGE 確認頁面索引、nr & BITS_PER_PAGE_MASK 確認 pid 所在比特位偏移；一個 clear_bit 優(yōu)雅的將比特位清零；一個 atomic_inc 優(yōu)雅的增加頁面剩余空閑 pid 數(shù)。簡潔明了，毋庸多言。

內(nèi)核小知識

第一次看內(nèi)核源碼，發(fā)現(xiàn)有很多有趣的東西，下面一一說明。

likely & unlikely

很多 if 條件中都有這個，不清楚是干什么的，翻來定義看一看：

# ifndef likely
#  define likely(x)	(__builtin_expect(!!(x), 1))
# endif
# ifndef unlikely
#  define unlikely(x)	(__builtin_expect(!!(x), 0))
# endif

條件 x 使用 !! 處理后將由整數(shù)變?yōu)?0 或 1，然后傳遞給 __builtin_expect，likely 第二個參數(shù)為 1，unlikely 為 0。經(jīng)過一翻 google，這個是編譯器 (gcc) 提供的分支預測優(yōu)化函數(shù)：

long __builtin_expect(long exp, long c);

第一個參數(shù)是條件；第二個是期望值，必需是編譯期常量；函數(shù)返回值為 exp 參數(shù)。GCC v2.96 引入，用來幫助編譯器生成匯編代碼，如果期望值為 1，編譯器將條件失敗放在 jmp 語句；如果期望值為 0，編譯器將條件成功放在 jmp 語句。實現(xiàn)更小概率的指令跳轉(zhuǎn)，這樣做的目的是提升 CPU 指令流水成功率，從而提升性能。

        if (unlikely(!map->page)) {
            void *page = kzalloc(PAGE_SIZE, GFP_KERNEL);
            /*
             * Free the page if someone raced with us
             * installing it:
             */
            spin_lock_irq(&pidmap_lock);
            if (!map->page) {
                map->page = page;
                page = NULL;
            }
            spin_unlock_irq(&pidmap_lock);
            kfree(page);
            if (unlikely(!map->page))
                break;
        }
        if (likely(atomic_read(&map->nr_free))) {
            for ( ; ; ) {
                if (!test_and_set_bit(offset, map->page)) {
                    atomic_dec(&map->nr_free);
                    set_last_pid(pid_ns, last, pid);
                    return pid;
                }
                offset = find_next_offset(map, offset);
                if (offset >= BITS_PER_PAGE)
                    break;
                pid = mk_pid(pid_ns, map, offset);
                if (pid >= pid_max)
                    break;
            }
        }

以頁面分配和頁內(nèi)遍歷為例，這里有 1 個 likely 和 2 個 unlikely，分別說明：

• 第一個 unlikely 用來判斷頁面是否為空，除第一次進入外，其它情況下此頁面都是已分配狀態(tài)，所以 !map->page 傾向于 0，這里使用 unlikely；
• 第二個 unlikely 用來判斷頁面是否分配失敗，正常情況下 !map->page 傾向于 0，這里使用 unlikely；
• 第三個 likely 用來判斷頁面是否已分配完畢，正常情況下 atomic_read(&map->nr_free) 結(jié)果傾向于 > 0，這里使用 likely。

總結(jié)一下，likely & unlikely 并不改變條件結(jié)果本身，在判斷是否進入條件時完全可以忽略它們！如果大部分場景進入條件，使用 likely；如果大多數(shù)場景不進入條件，使用 unlikely。

為何編譯器不能自己做這個工作？深入想想，代碼只有在執(zhí)行時才能知道哪些條件經(jīng)常返回 true，而這已經(jīng)離開編譯型語言生成機器代碼太遠了，所以需要程序員提前告知編譯器怎么生成代碼。對于解釋執(zhí)行的語言，這方面可能稍好一些。

最后，如果程序員也不清楚哪種場景占優(yōu)，最好就留空什么也不添加，千萬不要畫蛇添足。

pr_info 輸出

這個是在 pidmap_init 中遇到的，看看定義：

#ifndef pr_fmt
#define pr_fmt(fmt) fmt
#endif

#define pr_emerg(fmt, ...) \
	printk(KERN_EMERG pr_fmt(fmt), ##__VA_ARGS__)
#define pr_alert(fmt, ...) \
	printk(KERN_ALERT pr_fmt(fmt), ##__VA_ARGS__)
#define pr_crit(fmt, ...) \
	printk(KERN_CRIT pr_fmt(fmt), ##__VA_ARGS__)
#define pr_err(fmt, ...) \
	printk(KERN_ERR pr_fmt(fmt), ##__VA_ARGS__)
#define pr_warning(fmt, ...) \
	printk(KERN_WARNING pr_fmt(fmt), ##__VA_ARGS__)
#define pr_warn pr_warning
#define pr_notice(fmt, ...) \
	printk(KERN_NOTICE pr_fmt(fmt), ##__VA_ARGS__)
#define pr_info(fmt, ...) \
	printk(KERN_INFO pr_fmt(fmt), ##__VA_ARGS__)
#define pr_cont(fmt, ...) \
	printk(KERN_CONT fmt, ##__VA_ARGS__)

原來就是 printk 的包裝，pr_info 使用的級別是 KERN_INFO。下面是網(wǎng)上搜到的 printk 分派圖：

打到 console 的是系統(tǒng)初始化時在屏幕輸出的，一閃而過不太容易看，所以這里是使用基于 /dev/kmsg 的方式，具體點就是直接使用 dmesg：

$ dmesg | grep -C 10 pid_max
[    0.000000] Hierarchical RCU implementation.
[    0.000000]  RCU restricting CPUs from NR_CPUS=5120 to nr_cpu_ids=2.
[    0.000000] NR_IRQS:327936 nr_irqs:440 0
[    0.000000] Console: colour VGA+ 80x25
[    0.000000] console [tty0] enabled
[    0.000000] console [ttyS0] enabled
[    0.000000] allocated 436207616 bytes of page_cgroup
[    0.000000] please try 'cgroup_disable=memory' option if you don't want memory cgroups
[    0.000000] tsc: Detected 2394.374 MHz processor
[    0.620597] Calibrating delay loop (skipped) preset value.. 4788.74 BogoMIPS (lpj=2394374)
[    0.621979] pid_max: default: 32768 minimum: 301
[    0.622732] Security Framework initialized
[    0.623423] SELinux:  Initializing.
[    0.624063] SELinux:  Starting in permissive mode
[    0.624064] Yama: becoming mindful.
[    0.625585] Dentry cache hash table entries: 2097152 (order: 12, 16777216 bytes)
[    0.629691] Inode-cache hash table entries: 1048576 (order: 11, 8388608 bytes)
[    0.632167] Mount-cache hash table entries: 32768 (order: 6, 262144 bytes)
[    0.633123] Mountpoint-cache hash table entries: 32768 (order: 6, 262144 bytes)
[    0.634607] Initializing cgroup subsys memory
[    0.635326] Initializing cgroup subsys devices

也可以直接 cat /dev/kmsg：

$ cat /dev/kmsg | grep -C 10 pid_max
6,144,0,-;Hierarchical RCU implementation.
6,145,0,-;\x09RCU restricting CPUs from NR_CPUS=5120 to nr_cpu_ids=2.
6,146,0,-;NR_IRQS:327936 nr_irqs:440 0
6,147,0,-;Console: colour VGA+ 80x25
6,148,0,-;console [tty0] enabled
6,149,0,-;console [ttyS0] enabled
6,150,0,-;allocated 436207616 bytes of page_cgroup
6,151,0,-;please try 'cgroup_disable=memory' option if you don't want memory cgroups
6,152,0,-;tsc: Detected 2394.374 MHz processor
6,153,620597,-;Calibrating delay loop (skipped) preset value.. 4788.74 BogoMIPS (lpj=2394374)
6,154,621979,-;pid_max: default: 32768 minimum: 301
6,155,622732,-;Security Framework initialized
6,156,623423,-;SELinux:  Initializing.
7,157,624063,-;SELinux:  Starting in permissive mode
6,158,624064,-;Yama: becoming mindful.
6,159,625585,-;Dentry cache hash table entries: 2097152 (order: 12, 16777216 bytes)
6,160,629691,-;Inode-cache hash table entries: 1048576 (order: 11, 8388608 bytes)
6,161,632167,-;Mount-cache hash table entries: 32768 (order: 6, 262144 bytes)
6,162,633123,-;Mountpoint-cache hash table entries: 32768 (order: 6, 262144 bytes)
6,163,634607,-;Initializing cgroup subsys memory
6,164,635326,-;Initializing cgroup subsys devices

這種會 hang 在結(jié)尾，需要 Ctrl+C 才能退出。甚至也可以自己寫程序撈取：

/* The glibc interface */
#include <sys/klog.h>

int klogctl(int type, char *bufp, int len);

不過與前兩個不同，它是基于 /proc/kmsg 的，cat 查看這個文件內(nèi)容通常為空，與 /dev/kmesg 還有一些區(qū)別。限于篇幅就不一一介紹了，感興趣的讀者自己 man 查看下吧。

參考

[1]. Linux內(nèi)核入門-- likely和unlikely

[2]. Linux內(nèi)核輸出的日志去哪里了

[3]. Pid Namespace 詳解

[4]. namaspace之pid namespace

[5]. struct pid & pid_namespace

[6]. 一文看懂Linux進程ID的內(nèi)核管理

[9]. linux系統(tǒng)pid的最大值研究

[10]. What is CONFIG_BASE_SMALL=0