Linux 内核之旅（二十九）：内核并发与同步机制全解

static inline void __list_add(struct list_head *new,
                              struct list_head *prev,
                              struct list_head *next)
{
    next->prev = new;                 /* STEP1 */
    new->next  = next;                /* STEP2 */
    new->prev  = prev;                /* STEP3 到这里:new 已"半接入",但 prev 还没指向它 */
    WRITE_ONCE(prev->next, new);      /* STEP4 直到这一步,从 prev 前向遍历才看得到 new */
}

void add_wait_queue(wait_queue_head_t *q, wait_queue_t *wait)
{
    unsigned long flags;
    wait->flags &= ~WQ_FLAG_EXCLUSIVE;
    spin_lock_irqsave(&q->lock, flags);   /* 进入临界区 */
    __add_wait_queue(q, wait);            /* 内部就是 list_add → __list_add */
    spin_unlock_irqrestore(&q->lock, flags);
}

static inline void __raw_spin_lock(raw_spinlock_t *lock)
{
    preempt_disable();                    /* ★ 关键:持锁期间本 CPU 不可被抢占 */
    spin_acquire(&lock->dep_map, 0, 0, _RET_IP_);
    LOCK_CONTENDED(lock, do_raw_spin_trylock, do_raw_spin_lock);
}

/* irqsave 版还会先关中断,把三个来源一次性堵死 */
static inline unsigned long __raw_spin_lock_irqsave(raw_spinlock_t *lock)
{
    unsigned long flags;
    local_irq_save(flags);                /* 关中断:挡住硬中断里的并发访问 */
    preempt_disable();                    /* 关抢占:挡住被高优先级进程抢占 */
    /* ... 拿锁 ... */
    return flags;
}

static inline void __list_add_rcu(struct list_head *new,
                                  struct list_head *prev, struct list_head *next)
{
    new->next = next;                              /* 先把 new 内部指针备好 */
    new->prev = prev;
    rcu_assign_pointer(list_next_rcu(prev), new);  /* 带写屏障地"发布"前向指针 */
    next->prev = new;                              /* 反向指针后补(读者只走前向) */
}

static inline void list_add_rcu(struct list_head *new, struct list_head *head)
{
	__list_add_rcu(new, head, head->next);
}

static inline void list_add_tail_rcu(struct list_head *new,
					struct list_head *head)
{
	__list_add_rcu(new, head->prev, head);
}

static inline void list_add_tail(struct list_head *new, struct list_head *head)
{
	__list_add(new, head->prev, head);
}

//https://elixir.bootlin.com/linux/v4.11.6/source/include/linux/list.h#L55
static inline void __list_add(struct list_head *new,
			      struct list_head *prev,
			      struct list_head *next)
{
	if (!__list_add_valid(new, prev, next))
		return;

	next->prev = new;
	new->next = next;
	new->prev = prev;
	WRITE_ONCE(prev->next, new);
}

#define list_next_rcu(list)	(*((struct list_head __rcu **)(&(list)->next)))

static inline void __list_add_rcu(struct list_head *new,
		struct list_head *prev, struct list_head *next)
{
	if (!__list_add_valid(new, prev, next))
		return;

	new->next = next;
	new->prev = prev;
	rcu_assign_pointer(list_next_rcu(prev), new);   //？
	next->prev = new;
}

//https://elixir.bootlin.com/linux/v4.11.6/source/kernel/fork.c#L1491
static __latent_entropy 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,
					unsigned long tls,
					int node)
{
    ......
    write_lock_irq(&tasklist_lock);

	/* CLONE_PARENT re-uses the old parent */
	if (clone_flags & (CLONE_PARENT|CLONE_THREAD)) {
		p->real_parent = current->real_parent;
		p->parent_exec_id = current->parent_exec_id;
	} else {
		p->real_parent = current;
		p->parent_exec_id = current->self_exec_id;
	}

	spin_lock(&current->sighand->siglock);

	/*
	 * Copy seccomp details explicitly here, in case they were changed
	 * before holding sighand lock.
	 */
	copy_seccomp(p);

	/*
	 * Process group and session signals need to be delivered to just the
	 * parent before the fork or both the parent and the child after the
	 * fork. Restart if a signal comes in before we add the new process to
	 * it's process group.
	 * A fatal signal pending means that current will exit, so the new
	 * thread can't slip out of an OOM kill (or normal SIGKILL).
	*/
	recalc_sigpending();
	if (signal_pending(current)) {
		retval = -ERESTARTNOINTR;
		goto bad_fork_cancel_cgroup;
	}
	if (unlikely(!(ns_of_pid(pid)->nr_hashed & PIDNS_HASH_ADDING))) {
		retval = -ENOMEM;
		goto bad_fork_cancel_cgroup;
	}

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

		init_task_pid(p, PIDTYPE_PID, pid);
		if (thread_group_leader(p)) {
			init_task_pid(p, PIDTYPE_PGID, task_pgrp(current));
			init_task_pid(p, PIDTYPE_SID, task_session(current));

			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);
			/*
			 * Inherit has_child_subreaper flag under the same
			 * tasklist_lock with adding child to the process tree
			 * for propagate_has_child_subreaper optimization.
			 */
			p->signal->has_child_subreaper = p->real_parent->signal->has_child_subreaper ||
							 p->real_parent->signal->is_child_subreaper;
			list_add_tail(&p->sibling, &p->real_parent->children);
			list_add_tail_rcu(&p->tasks, &init_task.tasks);
			attach_pid(p, PIDTYPE_PGID);
			attach_pid(p, PIDTYPE_SID);
			__this_cpu_inc(process_counts);
		} else {
			current->signal->nr_threads++;
			atomic_inc(&current->signal->live);
			atomic_inc(&current->signal->sigcnt);
			list_add_tail_rcu(&p->thread_group,
					  &p->group_leader->thread_group);
			list_add_tail_rcu(&p->thread_node,
					  &p->signal->thread_head);
		}
		attach_pid(p, PIDTYPE_PID);
		nr_threads++;
	}

	total_forks++;
	spin_unlock(&current->sighand->siglock);
	syscall_tracepoint_update(p);
	write_unlock_irq(&tasklist_lock);
    ......
}

#define READ_ONCE(x)  __READ_ONCE(x, 1)

//https://elixir.bootlin.com/linux/v4.11.6/source/include/linux/compiler.h#L314
#define __READ_ONCE(x, check)						\
({									\
	union { typeof(x) __val; char __c[1]; } __u;			\
	if (check)							\
		__read_once_size(&(x), __u.__c, sizeof(x));		\
	else								\
		__read_once_size_nocheck(&(x), __u.__c, sizeof(x));	\
	__u.__val;							\
})
#define READ_ONCE(x) __READ_ONCE(x, 1)

#define WRITE_ONCE(x, val) \
({                          \
    union { typeof(x) __val; char __c[1]; } __u = { .__val = (val) }; \
    __write_once_size(&(x), __u.__c, sizeof(x)); \
    __u.__val;              \
})

/* 错误：编译器可能把 flag 缓存进寄存器，永远读不到更新 */
while (!flag) ;
use(data);

/* 生产者 */
obj->a = 1;
obj->b = 2;
smp_wmb();            /* 保证上面的初始化先于指针发布可见 */
WRITE_ONCE(gp, obj);  /* 发布指针 */

/* 消费者 */
p = READ_ONCE(gp);
if (p) {
    smp_rmb();        /* 保证先读到指针，再读它指向的内容 */
    use(p->a, p->b);
}

//https://elixir.bootlin.com/linux/v4.11.6/source/include/linux/rcupdate.h#L594
#define rcu_assign_pointer(p, v)					      \
({									      \
	uintptr_t _r_a_p__v = (uintptr_t)(v);				      \
									      \
	if (__builtin_constant_p(v) && (_r_a_p__v) == (uintptr_t)NULL)	      \
		WRITE_ONCE((p), (typeof(p))(_r_a_p__v));		      \
	else								      \
		smp_store_release(&p, RCU_INITIALIZER((typeof(p))_r_a_p__v)); \
	_r_a_p__v;							      \
})

static __always_inline void atomic_inc(atomic_t *v)
{
    asm volatile(LOCK_PREFIX "incl %0" : "+m" (v->counter));
}

//https://elixir.bootlin.com/linux/v4.11.6/source/arch/x86/include/asm/thread_info.h#L55
struct thread_info {
	unsigned long		flags;		/* low level flags */
};

spinlock_t lock;
spin_lock_init(&lock);
spin_lock(&lock);
/* 极短临界区 */
spin_unlock(&lock);

static __always_inline void queued_spin_lock(struct qspinlock *lock)
{
    u32 val;
    val = atomic_cmpxchg_acquire(&lock->val, 0, _Q_LOCKED_VAL);
    if (likely(val == 0))
        return;                 /* 无竞争，直接拿到 */
    queued_spin_lock_slowpath(lock, val);
}

rwlock_t lock;
read_lock(&lock);   /* 多读者可同时进入 */
/* 只读临界区 */
read_unlock(&lock);

write_lock(&lock);  /* 写者独占 */
/* 读写临界区 */
write_unlock(&lock);

//https://elixir.bootlin.com/linux/v4.11.6/source/include/linux/sched/task.h#L20
extern rwlock_t tasklist_lock;

seqlock_t sl;
/* 写者：更新序号（奇数=写进行中） */
write_seqlock(&sl);
/* 修改数据 */
write_sequnlock(&sl);

/* 读者：读前后各取一次序号，若变化或为奇则重试 */
unsigned seq;
do {
    seq = read_seqbegin(&sl);
    /* 读数据到局部变量 */
} while (read_seqretry(&sl, seq));

//https://elixir.bootlin.com/linux/v4.11.6/source/include/linux/semaphore.h#L16
struct semaphore {
    raw_spinlock_t      lock;
    unsigned int        count;
    struct list_head    wait_list;
};

void down(struct semaphore *sem);   /* count-- ，为 0 则睡眠等待 */
void up(struct semaphore *sem);     /* count++ ，唤醒一个等待者 */

struct mutex m;
mutex_lock(&m);
/* 可以较长、可以睡眠的临界区 */
mutex_unlock(&m);

//https://elixir.bootlin.com/linux/v4.11.6/source/include/linux/mutex.h#L53
struct mutex {
    atomic_long_t       owner;      /* 指向 owner task_struct，低位存标志 */
    spinlock_t          wait_lock;
#ifdef CONFIG_MUTEX_SPIN_ON_OWNER
    struct optimistic_spin_queue osq; /* MCS 乐观自旋队列 */
#endif
    struct list_head    wait_list;
    ......
};

#define MUTEX_FLAG_WAITERS  0x01  /* 有等待者，unlock 必须唤醒 */
#define MUTEX_FLAG_HANDOFF  0x02  /* unlock 需把锁直接交给队首（防饥饿） */

/* __mutex_lock_common 主循环（简化自 v4.11.6） */
for (;;) {
    if (__mutex_trylock(lock, first))   /* first: 是否已是队首, 处理 handoff */
        goto acquired;
    if (unlikely(signal_pending_state(state, task))) { ret = -EINTR; goto err; }
    spin_unlock_mutex(&lock->wait_lock, flags);
    schedule_preempt_disabled();        /* 睡眠 */
    spin_lock_mutex(&lock->wait_lock, flags);
    if (!first && __mutex_waiter_is_first(lock, &waiter)) {
        first = true;
        __mutex_set_flag(lock, MUTEX_FLAG_HANDOFF);  /* 我成了队首，要求直接移交 */
    }
}

struct rw_semaphore rwsem;
down_read(&rwsem);   /* 读者，可多个 */
/* 可睡眠的只读临界区 */
up_read(&rwsem);

down_write(&rwsem);  /* 写者独占 */
up_write(&rwsem);

DEFINE_PER_CPU(int, my_counter);        /* 定义 */

/* 访问：关抢占以保证"取本 CPU id"和"访问副本"之间不被迁移 */
int *p = get_cpu_var(my_counter);       /* = this_cpu_ptr + preempt_disable */
(*p)++;
put_cpu_var(my_counter);                /* preempt_enable */

/* 或用 this_cpu_* 原子式访问（自身保证本 CPU 原子） */
this_cpu_inc(my_counter);

//https://elixir.bootlin.com/linux/v4.11.6/source/include/linux/percpu-defs.h#L262
/*
 * Must be an lvalue. Since @var must be a simple identifier,
 * we force a syntax error here if it isn't.
 */
#define get_cpu_var(var)						\
(*({									\
	preempt_disable();						\
	this_cpu_ptr(&var);						\
}))


/*
 * The weird & is necessary because sparse considers (void)(var) to be
 * a direct dereference of percpu variable (var).
 */
#define put_cpu_var(var)						\
do {									\
	(void)&(var);							\
	preempt_enable();						\
} while (0)

DECLARE_WAIT_QUEUE_HEAD(wq);
/* 等待方：wait_event 会循环检查 condition，为假则睡眠 */
wait_event_interruptible(wq, condition);
/* 唤醒方：改变条件后唤醒 */
condition = true;
wake_up_interruptible(&wq);

/* 手写版，等价于 wait_event 的展开 */
DEFINE_WAIT(wait);
for (;;) {
    prepare_to_wait(&wq, &wait, TASK_INTERRUPTIBLE); /* 挂入队列，置睡眠态 */
    if (condition)          /* 关键：必须先置态再判条件，避免丢唤醒 */
        break;
    if (signal_pending(current)) { ... }
    schedule();             /* 让出 CPU */
}
finish_wait(&wq, &wait);    /* 出队，置 TASK_RUNNING */

rcu_read_lock();
p = rcu_dereference(gp);        /* 订阅：安全读取被 RCU 保护的指针 */
if (p)
    do_something(p->field);     /* 保证在本临界区内 p 不会被释放 */
rcu_read_unlock();

/* 读侧内层：仅在 RCU 临界区(或持 tasklist_lock)内有效 */
struct task_struct *pid_task(struct pid *pid, enum pid_type type)
{
    struct task_struct *result = NULL;
    if (pid) {
        struct hlist_node *first;
        /* rcu_dereference_check: 带依赖序地读链表首节点；
         * lockdep 断言:调用者要么持 rcu_read_lock,要么持 tasklist_lock */
        first = rcu_dereference_check(hlist_first_rcu(&pid->tasks[type]),
                                      lockdep_tasklist_lock_is_held());
        if (first)
            result = hlist_entry(first, struct task_struct, pids[(type)].node);
    }
    return result;   /* 出了保护区,这个指针就不保证有效 */
}

/* 调用层：把"临界区内有效"升级为"跨临界区长期有效" */
struct task_struct *get_pid_task(struct pid *pid, enum pid_type type)
{
    struct task_struct *result;
    rcu_read_lock();                 /* STEP1： 进入 RCU 读侧:保证 result 在本区间内不被 kfree */
    result = pid_task(pid, type);
    if (result)
        get_task_struct(result);     /* STEP2： 关键:还在临界区内就把引用计数 +1 */
    rcu_read_unlock();               /* STEP3： 离开临界区:此后 result 靠引用计数(而非RCU)存活 */
    return result;                   /* 调用者用完必须 put_task_struct() 配平 */
}

//get_task_struct：本质是累加引用计数
#define get_task_struct(tsk) do { atomic_inc(&(tsk)->usage); } while(0)

extern rwlock_t tasklist_lock;

//https://elixir.bootlin.com/linux/v4.11.6/source/kernel/pid.c#L388
/* 写侧-加入:必须在 tasklist_lock 写锁保护下调用 */
void attach_pid(struct task_struct *task, enum pid_type type)
{
    struct pid_link *link = &task->pids[type];
    /* hlist_add_head_rcu: 内含发布语义(smp)把新节点挂链,读者可无锁并发遍历 */
    hlist_add_head_rcu(&link->node, &link->pid->tasks[type]);
}

/* 写侧-摘除并回收(简化自 release_task) */
void release_task(struct task_struct *p)
{
    write_lock_irq(&tasklist_lock);   /* STEP1： 写者互斥:防并发增删撕裂链表 */
    __exit_signal(p);                 /*    内部 __unhash_process → detach_pid(→hlist_del_rcu)/list_del_rcu */
    write_unlock_irq(&tasklist_lock);
    ......
    call_rcu(&p->rcu, delayed_put_task_struct); /* STEP2： 延迟回收:等宽限期后才真正 put/free */
}

static void delayed_put_task_struct(struct rcu_head *rhp)
{
    struct task_struct *tsk = container_of(rhp, struct task_struct, rcu);
    ......
    put_task_struct(tsk);             /* STEP3：宽限期已过,读者都退出,安全递减最终引用 */
}

//https://elixir.bootlin.com/linux/v4.11.6/source/fs/inode.c#L1527
void iput(struct inode *inode)
{
    if (!inode)
        return;
    BUG_ON(inode->i_state & I_CLEAR);
retry:
    /* atomic_dec_and_lock: 原子递减 i_count
     *   - 递减后 != 0: 不拿锁,直接返回（快路径,绝大多数 put）
     *   - 递减后 == 0: 拿住 i_lock 并返回真（慢路径,进入销毁）*/
    if (atomic_dec_and_lock(&inode->i_count, &inode->i_lock)) {
        if (inode->i_nlink && (inode->i_state & I_DIRTY_TIME)) {
            atomic_inc(&inode->i_count);
            inode->i_state &= ~I_DIRTY_TIME;
            spin_unlock(&inode->i_lock);
            mark_inode_dirty_sync(inode);
            goto retry;
        }
        iput_final(inode);            /* 持 i_lock 进入最终销毁 */
    }
}

//igrab 的作用是安全地获取并增加一个现有 inode 的引用计数
//https://elixir.bootlin.com/linux/v4.11.6/source/fs/inode.c#L1218
struct inode *igrab(struct inode *inode)
{
    spin_lock(&inode->i_lock);
    /* 只有 inode 未进入销毁流程时才允许 +1;
     * i_lock 保证这里读到的 i_state 与 iput 里的"判零/置 I_FREEING"互斥 */
    if (!(inode->i_state & (I_FREEING | I_WILL_FREE))) {
        __iget(inode);             /* = atomic_inc(&inode->i_count) */
        spin_unlock(&inode->i_lock);
    } else {
        spin_unlock(&inode->i_lock);
        inode = NULL;              /* 正在被释放,拿不到,返回 NULL */
    }
    return inode;
}

struct __wait_queue_head {
    spinlock_t        lock;          /* 队列自带的锁 */
    struct list_head  task_list;
};

/* 等待方：入队与置睡眠态在同一把锁下 */
void prepare_to_wait(wait_queue_head_t *q, wait_queue_t *wait, int state)
{
    unsigned long flags;
    wait->flags &= ~WQ_FLAG_EXCLUSIVE;
    spin_lock_irqsave(&q->lock, flags);
    if (list_empty(&wait->task_list))
        __add_wait_queue(q, wait);   /* 挂入队列 */
    set_current_state(state);        /* 置睡眠态(内含屏障) */
    spin_unlock_irqrestore(&q->lock, flags);
}

/* 唤醒方：遍历唤醒也拿同一把 q->lock */
void __wake_up(wait_queue_head_t *q, unsigned int mode, int nr, void *key)
{
    unsigned long flags;
    spin_lock_irqsave(&q->lock, flags);
    __wake_up_common(q, mode, nr, 0, key);
    spin_unlock_irqrestore(&q->lock, flags);
}

#define wait_event(wq, condition)                            \
do {                                                         \
    DEFINE_WAIT(__wait);                                     \
    for (;;) {                                               \
        prepare_to_wait(&wq, &__wait, TASK_UNINTERRUPTIBLE); \
        if (condition)          /* 先置睡眠态,再判条件 */     \
            break;                                           \
        schedule();             /* 让出CPU,被 wake_up 唤醒后重判 */ \
    }                                                        \
    finish_wait(&wq, &__wait);  /* 出队 + 置 TASK_RUNNING */ \
} while (0)

//https://elixir.bootlin.com/linux/v4.11.6/source/kernel/sched/completion.c#L18
/* 通知方：可能在硬中断里被调用 */
void complete(struct completion *x)
{
    unsigned long flags;
    /* irqsave: 关本CPU中断，防"进程上下文持 x->wait.lock 时被同CPU中断重入抢同一把锁"自锁死 */
    spin_lock_irqsave(&x->wait.lock, flags);
    x->done++;                               /* 更新条件 */
    __wake_up_locked(&x->wait, TASK_NORMAL, 1); /* 已持锁,唤醒一个等待者 */
    spin_unlock_irqrestore(&x->wait.lock, flags);
}

/* 等待方:进程上下文 */
//https://elixir.bootlin.com/linux/v4.11.6/source/kernel/sched/completion.c#L113
void wait_for_completion(struct completion *x)
{
    /* 内部 spin_lock_irq(&x->wait.lock); 若 x->done==0 则挂入 x->wait 睡眠,
     * 被 complete() 唤醒后 x->done-- 返回 */
    wait_for_common(x, MAX_SCHEDULE_TIMEOUT, TASK_UNINTERRUPTIBLE);
}

static inline long __sched
__wait_for_common(struct completion *x,
		  long (*action)(long), long timeout, int state)
{
	might_sleep();

	spin_lock_irq(&x->wait.lock);
	timeout = do_wait_for_common(x, action, timeout, state);
	spin_unlock_irq(&x->wait.lock);
	return timeout;
}

static long __sched
do_wait_for_common(struct completion *x,
                   long (*action)(long), long timeout, int state)
{
    if (!x->done) {
        DECLARE_WAITQUEUE(wait, current);
        __add_wait_queue_tail_exclusive(&x->wait, &wait); /* 独占入队,避免惊群 */
        do {
            if (signal_pending_state(state, current)) { /* ... 处理信号 */ }
            __set_current_state(state);        /* 置睡眠态 */
            spin_unlock_irq(&x->wait.lock);    /* 放锁再睡 */
            timeout = action(timeout);         /* 通常是 schedule_timeout */
            spin_lock_irq(&x->wait.lock);      /* 醒来重新拿锁 */
        } while (!x->done && timeout);         /* 复检条件:防丢唤醒/伪唤醒 */
        __remove_wait_queue(&x->wait, &wait);
        if (!x->done)
            return timeout;
    }
    x->done--;                                 /* 消费一次完成事件 */
    return timeout ?: 1;
}

//https://elixir.bootlin.com/linux/v4.11.6/source/mm/memory.c#L614
int __pte_alloc(struct mm_struct *mm, pmd_t *pmd, unsigned long address)
{
    spinlock_t *ptl;
    pgtable_t new = pte_alloc_one(mm, address);   /* 先在锁外分配、清零、初始化 */
    if (!new)
        return -ENOMEM;

    /* 关键屏障:保证"页表页的初始化"先于"指针被挂入 pmd"对其他CPU可见;
     * 否则无锁遍历页表的 CPU 可能看到指针却读到未初始化内容 */
    smp_wmb();

    ptl = pmd_lock(mm, pmd);                       /* 拿页表锁 */
    if (likely(pmd_none(*pmd))) {                  /* 再检一次:别的CPU是否已经装好了? */
        atomic_long_inc(&mm->nr_ptes);
        pmd_populate(mm, pmd, new);                /* 发布:把新页表挂入 pmd */
        new = NULL;
    }
    spin_unlock(ptl);
    if (new)                                       /* 输给了别人,释放自己多分配的那份 */
        pte_free(mm, new);
    return 0;
}

/* 读者:无锁 pointer chasing(简化) */
pmd_t *pmd = pmd_offset(pud, addr);
if (pmd_none(*pmd))            /* 读到"发布后的指针" */
    return NULL;
pte = pte_offset_map(pmd, addr); /* 依赖"先读指针,再读其内容"的数据依赖序 */

/* 读端：无锁遍历(调用者持 rcu_read_lock) */
//https://elixir.bootlin.com/linux/v4.11.6/source/include/linux/netdevice.h#L2343
#define for_each_netdev_rcu(net, d) \
    list_for_each_entry_rcu(d, &(net)->dev_base_head, dev_list)

struct net_device *dev_getbyhwaddr_rcu(struct net *net, unsigned short type,
                                       const char *ha)
{
    struct net_device *dev;
    for_each_netdev_rcu(net, dev)          /* rcu_dereference 跟 next 指针 */
        if (dev->type == type && !memcmp(dev->dev_addr, ha, dev->addr_len))
            return dev;
    return NULL;
}

//https://elixir.bootlin.com/linux/v4.11.6/source/include/linux/netdevice.h#L2337
//全局锁
extern rwlock_t				dev_base_lock;		/* Device list lock */

/* 写端：加入（持 RTNL + dev_base_lock）*/
//https://elixir.bootlin.com/linux/v4.11.6/source/net/core/dev.c#L226
static int list_netdevice(struct net_device *dev)
{
    struct net *net = dev_net(dev);
    ASSERT_RTNL();                          /* 写者之间用 RTNL 互斥 */
    write_lock_bh(&dev_base_lock);          /* 只用来挡住老式 read_lock 纯读者 */
    list_add_tail_rcu(&dev->dev_list, &net->dev_base_head); /* 发布语义挂链 */
    hlist_add_head_rcu(&dev->index_hlist, dev_index_hash(net, dev->ifindex));
    write_unlock_bh(&dev_base_lock);
    return 0;
}

/* 写端：删除(只摘链；调用者随后须过一个 RCU 宽限期才能 free/复用 dev) */
//https://elixir.bootlin.com/linux/v4.11.6/source/net/core/dev.c#L245
static void unlist_netdevice(struct net_device *dev)
{
    ASSERT_RTNL();
    write_lock_bh(&dev_base_lock);
    list_del_rcu(&dev->dev_list);           /* 只摘链,不释放 */
    hlist_del_rcu(&dev->index_hlist);
    write_unlock_bh(&dev_base_lock);

	dev_base_seq_inc(dev_net(dev));
}

//https://elixir.bootlin.com/linux/v4.11.6/source/net/ipv4/inet_hashtables.c#L267
/* 读端：软中断里无锁查找 socket（调用者持 rcu_read_lock）*/
struct sock *__inet_lookup_established(struct net *net,
                struct inet_hashinfo *hashinfo, ...)
{
    unsigned int slot = hash & hashinfo->ehash_mask;
    struct inet_ehash_bucket *head = &hashinfo->ehash[slot];
    const struct hlist_nulls_node *node;
    struct sock *sk;
begin:
    sk_nulls_for_each_rcu(sk, node, &head->chain) {   /* RCU 遍历该桶 */
        if (sk->sk_hash != hash) continue;
        if (likely(INET_MATCH(sk, net, acookie, saddr, daddr, ports, dif))) {
            if (unlikely(!atomic_inc_not_zero(&sk->sk_refcnt)))
                goto out;                     /* 已在销毁,放弃 */
            if (unlikely(!INET_MATCH(sk, net, acookie, saddr, daddr, ports, dif))) {
                sock_gen_put(sk); goto begin; /* 拿到引用后复检,防窗口内被改 */
            }
            goto found;
        }
    }
    /* nulls 值 != 起始 slot：说明遍历中元素被移到别的桶，整体重试 */
    if (get_nulls_value(node) != slot)
        goto begin;
out:   sk = NULL;
found: return sk;
}

/* 写端：插入（拿该桶的 per-bucket spinlock） */
//https://elixir.bootlin.com/linux/v4.11.6/source/net/ipv4/inet_hashtables.c#L392
bool inet_ehash_insert(struct sock *sk, struct sock *osk)
{
    struct hlist_nulls_head *list;
    struct inet_ehash_bucket *head;
    spinlock_t *lock;

    sk->sk_hash = sk_ehashfn(sk);
    head = inet_ehash_bucket(hashinfo, sk->sk_hash);
    list = &head->chain;
    lock = inet_ehash_lockp(hashinfo, sk->sk_hash);   /* 按 hash 取该桶的锁 */
    spin_lock(lock);
    if (osk)                                  /* 需替换（如 TIMEWAIT）时先原子摘除 */
        sk_nulls_del_node_init_rcu(osk);
    __sk_nulls_add_node_rcu(sk, list);        /* 发布语义挂入桶 */
    spin_unlock(lock);
    return true;
}

struct inet_hashinfo {
	/* This is for sockets with full identity only.  Sockets here will
	 * always be without wildcards and will have the following invariant:
	 *
	 *          TCP_ESTABLISHED <= sk->sk_state < TCP_CLOSE
	 *
	 */
	struct inet_ehash_bucket	*ehash;
	spinlock_t			*ehash_locks;   //每个bucket一个spin_lock
    ......
}


//https://elixir.bootlin.com/linux/v4.11.6/source/include/net/inet_hashtables.h#L163
static inline spinlock_t *inet_ehash_lockp(
	struct inet_hashinfo *hashinfo,
	unsigned int hash)
{
	return &hashinfo->ehash_locks[hash & hashinfo->ehash_locks_mask];
}

/* 读端:查找覆盖 addr 的 VMA(调用者持 down_read(&mm->mmap_sem);省略 vmacache 快查) */
struct vm_area_struct *find_vma(struct mm_struct *mm, unsigned long addr)
{
    struct rb_node *rb_node = mm->mm_rb.rb_node;
    struct vm_area_struct *vma = NULL;
    while (rb_node) {
        struct vm_area_struct *tmp =
            rb_entry(rb_node, struct vm_area_struct, vm_rb);
        if (tmp->vm_end > addr) {
            vma = tmp;
            if (tmp->vm_start <= addr) break;    /* 命中 */
            rb_node = rb_node->rb_left;
        } else {
            rb_node = rb_node->rb_right;
        }
    }
    return vma;
}

/* 写端:插入新 VMA(调用者持 down_write(&mm->mmap_sem)) */
void __vma_link_rb(struct mm_struct *mm, struct vm_area_struct *vma,
                   struct rb_node **rb_link, struct rb_node *rb_parent)
{
    rb_link_node(&vma->vm_rb, rb_parent, rb_link); /* 挂上新节点 */
    vma->rb_subtree_gap = 0;
    vma_gap_update(vma);
    vma_rb_insert(vma, &mm->mm_rb);                /* 旋转 + 着色 rebalance */
}

static noinline void
__do_page_fault(struct pt_regs *regs, unsigned long error_code,
		unsigned long address)
{
	struct vm_area_struct *vma;
	struct task_struct *tsk;
	struct mm_struct *mm;
    ......
    if (unlikely(!down_read_trylock(&mm->mmap_sem))) {
		if ((error_code & PF_USER) == 0 &&
		    !search_exception_tables(regs->ip)) {
			bad_area_nosemaphore(regs, error_code, address, NULL);
			return;
		}
retry:  
		down_read(&mm->mmap_sem);
	} else {
		might_sleep();
	}

	vma = find_vma(mm, address);
	if (unlikely(!vma)) {
		bad_area(regs, error_code, address);
		return;
	}
    ......
	if (unlikely(expand_stack(vma, address))) {
		bad_area(regs, error_code, address);
		return;
	}

good_area:
	if (unlikely(access_error(error_code, vma))) {
		bad_area_access_error(regs, error_code, address, vma);
		return;
	}

	/*
	 * If for any reason at all we couldn't handle the fault,
	 * make sure we exit gracefully rather than endlessly redo
	 * the fault.  Since we never set FAULT_FLAG_RETRY_NOWAIT, if
	 * we get VM_FAULT_RETRY back, the mmap_sem has been unlocked.
	 */
	fault = handle_mm_fault(vma, address, flags);
	major |= fault & VM_FAULT_MAJOR;
    ......

	up_read(&mm->mmap_sem);

    ......
}

//https://elixir.bootlin.com/linux/v4.11.6/source/fs/dcache.c#L2064
struct dentry *__d_lookup_rcu(const struct dentry *parent,
                const struct qstr *name, unsigned *seqp)
{
    ......
    hlist_bl_for_each_entry_rcu(dentry, node, head, d_hash) {  /* RCU 遍历哈希链 */
        unsigned seq;
seqretry:
        seq = raw_seqcount_begin(&dentry->d_seq);   /* 取该 dentry 的 seqcount 快照 */
        if (dentry->d_parent != parent)
            continue;
        if (d_unhashed(dentry))
            continue;
        *seqp = seq;
        if (dentry->d_name.hash_len != hashlen)
            continue;
        if (dentry_cmp(dentry, str, hashlen_len(hashlen)) != 0)
            continue;
        return dentry;
    }
    return NULL;
}

//https://elixir.bootlin.com/linux/v4.11.6/source/fs/namei.c#L1537
static int lookup_fast(struct nameidata *nd,
		       struct path *path, struct inode **inode,
		       unsigned *seqp)
{
	struct vfsmount *mnt = nd->path.mnt;
	struct dentry *dentry, *parent = nd->path.dentry;
	int status = 1;
	int err;

	/*
	 * Rename seqlock is not required here because in the off chance
	 * of a false negative due to a concurrent rename, the caller is
	 * going to fall back to non-racy lookup.
	 */
	if (nd->flags & LOOKUP_RCU) {
		unsigned seq;
		bool negative;
        //外侧无任何锁
		dentry = __d_lookup_rcu(parent, &nd->last, &seq);
		if (unlikely(!dentry)) {
			if (unlazy_walk(nd))
				return -ECHILD;
			return 0;
		}

		......
	} else {
		dentry = __d_lookup(parent, &nd->last);
		if (unlikely(!dentry))
			return 0;
		status = d_revalidate(dentry, nd->flags);
	}
	......

	path->mnt = mnt;
	path->dentry = dentry;
	err = follow_managed(path, nd);
	if (likely(err > 0))
		*inode = d_backing_inode(path->dentry);
	return err;
}

//https://elixir.bootlin.com/linux/v4.11.6/source/fs/dcache.c#L2203
struct dentry *__d_lookup(const struct dentry *parent, const struct qstr *name)
{
	unsigned int hash = name->hash;
	struct hlist_bl_head *b = d_hash(hash);
	struct hlist_bl_node *node;
	struct dentry *found = NULL;
	struct dentry *dentry;

	rcu_read_lock();
	
	hlist_bl_for_each_entry_rcu(dentry, node, b, d_hash) {

		if (dentry->d_name.hash != hash)
			continue;

		spin_lock(&dentry->d_lock);
		if (dentry->d_parent != parent)
			goto next;
		if (d_unhashed(dentry))
			goto next;

		if (!d_same_name(dentry, parent, name))
			goto next;

		dentry->d_lockref.count++;
		found = dentry;
		spin_unlock(&dentry->d_lock);
		break;
next:
		spin_unlock(&dentry->d_lock);
 	}
 	rcu_read_unlock();

 	return found;
}

//https://elixir.bootlin.com/linux/v4.11.6/source/fs/namei.c#L1638
/* Fast lookup failed, do it the slow way */
static struct dentry *lookup_slow(const struct qstr *name,
				  struct dentry *dir,
				  unsigned int flags)
{
	struct dentry *dentry = ERR_PTR(-ENOENT), *old;
	struct inode *inode = dir->d_inode;
	DECLARE_WAIT_QUEUE_HEAD_ONSTACK(wq);

	inode_lock_shared(inode);
	/* Don't go there if it's already dead */
	if (unlikely(IS_DEADDIR(inode)))
		goto out;
again:
	dentry = d_alloc_parallel(dir, name, &wq);
	if (IS_ERR(dentry))
		goto out;
	if (unlikely(!d_in_lookup(dentry))) {
		if (!(flags & LOOKUP_NO_REVAL)) {
			int error = d_revalidate(dentry, flags);
			if (unlikely(error <= 0)) {
				if (!error) {
					d_invalidate(dentry);
					dput(dentry);
					goto again;
				}
				dput(dentry);
				dentry = ERR_PTR(error);
			}
		}
	} else {
		old = inode->i_op->lookup(inode, dentry, flags);
		d_lookup_done(dentry);
		if (unlikely(old)) {
			dput(dentry);
			dentry = old;
		}
	}
out:
	inode_unlock_shared(inode);
	return dentry;
}

struct dentry *d_alloc_parallel(struct dentry *parent,
				const struct qstr *name,
				wait_queue_head_t *wq)
{
	unsigned int hash = name->hash;
	struct hlist_bl_head *b = in_lookup_hash(parent, hash);
	struct hlist_bl_node *node;
	struct dentry *new = d_alloc(parent, name);
	struct dentry *dentry;
	unsigned seq, r_seq, d_seq;

	if (unlikely(!new))
		return ERR_PTR(-ENOMEM);

retry:
	rcu_read_lock();
	seq = smp_load_acquire(&parent->d_inode->i_dir_seq) & ~1;
	r_seq = read_seqbegin(&rename_lock);
    //call __d_lookup_rcu
	dentry = __d_lookup_rcu(parent, name, &d_seq);
	if (unlikely(dentry)) {
		if (!lockref_get_not_dead(&dentry->d_lockref)) {
			rcu_read_unlock();
			goto retry;
		}
		if (read_seqcount_retry(&dentry->d_seq, d_seq)) {
			rcu_read_unlock();
			dput(dentry);
			goto retry;
		}
		rcu_read_unlock();
		dput(new);
		return dentry;
	}
	if (unlikely(read_seqretry(&rename_lock, r_seq))) {
		rcu_read_unlock();
		goto retry;
	}
	hlist_bl_lock(b);
	if (unlikely(parent->d_inode->i_dir_seq != seq)) {
		hlist_bl_unlock(b);
		rcu_read_unlock();
		goto retry;
	}
	/*
	 * No changes for the parent since the beginning of d_lookup().
	 * Since all removals from the chain happen with hlist_bl_lock(),
	 * any potential in-lookup matches are going to stay here until
	 * we unlock the chain.  All fields are stable in everything
	 * we encounter.
	 */
	hlist_bl_for_each_entry(dentry, node, b, d_u.d_in_lookup_hash) {
		if (dentry->d_name.hash != hash)
			continue;
		if (dentry->d_parent != parent)
			continue;
		if (!d_same_name(dentry, parent, name))
			continue;
		hlist_bl_unlock(b);
		/* now we can try to grab a reference */
		if (!lockref_get_not_dead(&dentry->d_lockref)) {
			rcu_read_unlock();
			goto retry;
		}

		rcu_read_unlock();
		/*
		 * somebody is likely to be still doing lookup for it;
		 * wait for them to finish
		 */
		spin_lock(&dentry->d_lock);
		d_wait_lookup(dentry);
		/*
		 * it's not in-lookup anymore; in principle we should repeat
		 * everything from dcache lookup, but it's likely to be what
		 * d_lookup() would've found anyway.  If it is, just return it;
		 * otherwise we really have to repeat the whole thing.
		 */
		if (unlikely(dentry->d_name.hash != hash))
			goto mismatch;
		if (unlikely(dentry->d_parent != parent))
			goto mismatch;
		if (unlikely(d_unhashed(dentry)))
			goto mismatch;
		if (unlikely(!d_same_name(dentry, parent, name)))
			goto mismatch;
		/* OK, it *is* a hashed match; return it */
		spin_unlock(&dentry->d_lock);
		dput(new);
		return dentry;
	}
	rcu_read_unlock();
	/* we can't take ->d_lock here; it's OK, though. */
	new->d_flags |= DCACHE_PAR_LOOKUP;
	new->d_wait = wq;
	hlist_bl_add_head_rcu(&new->d_u.d_in_lookup_hash, b);
	hlist_bl_unlock(b);
	return new;
mismatch:
	spin_unlock(&dentry->d_lock);
	dput(dentry);
	goto retry;
}
EXPORT_SYMBOL(d_alloc_parallel);