Java 程序员眼里的 Linux 内核 —— wait_event 源码分析

// ${linux_source}/include/linux/wait.h

/**
 * wait_event - sleep until a condition gets true
 * @wq: the waitqueue to wait on
 * @condition: a C expression for the event to wait for
 *
 * The process is put to sleep (TASK_UNINTERRUPTIBLE) until the
 * @condition evaluates to true. The @condition is checked each time
 * the waitqueue @wq is woken up.
 *
 * wake_up() has to be called after changing any variable that could
 * change the result of the wait condition.
 */
#define wait_event(wq, condition)       \
do {                                                           \
    if (condition)                                     \
        break;                                               \
    __wait_event(wq, condition);            \
} while (0)

// ${linux_source}/include/linux/wait.h
#define __wait_event(wq, condition)           \
do {                                                                   \
    DEFINE_WAIT(__wait);                                 \
                                                                          
    for (;;) {                                                     \
        prepare_to_wait(&wq, &__wait, TASK_UNINTERRUPTIBLE);    \
        if (condition)                                         \
            break;                                                   \
        // schedule 使用调度器调度另一个线程去执行。当前线程被重新      \
        // 调度时，schedule 函数才会返回           \
        schedule();                                            \
    }                                                                  \
    finish_wait(&wq, &__wait);                      \
} while (0)

// ${linux_source}/include/linux/wait.h
#define DEFINE_WAIT(name)                              \
    wait_queue_t name = {                                   \
        .private    = current,                               \
        .func       = autoremove_wake_function,   \
        .task_list  = LIST_HEAD_INIT((name).task_list),     \
    }

// ${linux_source}/kernel/wait.c
/*
 * Note: we use "set_current_state()" _after_ the wait-queue add,
 * because we need a memory barrier there on SMP, so that any
 * wake-function that tests for the wait-queue being active
 * will be guaranteed to see waitqueue addition _or_ subsequent
 * tests in this thread will see the wakeup having taken place.
 *
 * The spin_unlock() itself is semi-permeable and only protects
 * one way (it only protects stuff inside the critical region and
 * stops them from bleeding out - it would still allow subsequent
 * loads to move into the critical region).
 */
void fastcall
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);
    // wait 不存在于某个等待队列时，才把它加入 q
    // wait 是我们新定义的，list_empty 将会返回 true
    if (list_empty(&wait->task_list))
        __add_wait_queue(q, wait);
    /*
     * don't alter the task state if this is just going to
     * queue an async wait queue callback
     */
    // 根据 wait 的定义，is_sync_wait() 这里会返回 true
    if (is_sync_wait(wait))
        // 前面注释说使用 set_current_state() 作为屏障，对此不理解的读者可以暂时忽略，
        // 后面我们会举例说明相关的用法
        set_current_state(state);
    // 解锁
    spin_unlock_irqrestore(&q->lock, flags);
}

/*
 * Used to distinguish between sync and async io wait context:
 * sync i/o typically specifies a NULL wait queue entry or a wait
 * queue entry bound to a task (current task) to wake up.
 * aio specifies a wait queue entry with an async notification
 * callback routine, not associated with any task.
 */
#define is_sync_wait(wait)    (!(wait) || ((wait)->private))

void fastcall finish_wait(wait_queue_head_t *q, wait_queue_t *wait)
{
    unsigned long flags;

    __set_current_state(TASK_RUNNING);
    /*
     * We can check for list emptiness outside the lock
     * IFF:
     *  - we use the "careful" check that verifies both
     *    the next and prev pointers, so that there cannot
     *    be any half-pending updates in progress on other
     *    CPU's that we haven't seen yet (and that might
     *    still change the stack area.
     * and
     *  - all other users take the lock (ie we can only
     *    have _one_ other CPU that looks at or modifies
     *    the list).
     */
    if (!list_empty_careful(&wait->task_list)) {
        spin_lock_irqsave(&q->lock, flags);
        list_del_init(&wait->task_list);
        spin_unlock_irqrestore(&q->lock, flags);
    }
}

synchronized (this) {
    while (!condition) {
        wait();
    }
    // do your stuff
}

void wait() {
    add_to_condition_queue();
    unlock();
    schedule();
}

lock();
while (!condition) {
    add_to_condition_queue();
    unlock();
    schedule();
}
// do your stuff

void our_wait_event() {
    if (condition) return;

    for (;;) {
        // 如果你喜欢，换成 condition_queue 也可以
        add_to_wait_queue_if_not_added_yet();
        if (condition)
            break;
        schedule();
    }
    remove_from_wait_queue();
}

void our_wait_event2() {
    while (!condition) {
        add_to_wait_queue_if_not_added_yet();
        schedule();
    }
    remove_from_wait_queue();
}

thread1                                          thread2
-----------------------           --------------
check condition => false
add_to_wait_queue()

                                                  alter_condition()
                                                  notify_all()

state = TASK_UNINTERRUPTABLE
schedule()

public static SomeClass getInstance() {
    if (sInstance == null) {
        synchronized (SomeClass.class) {
            if (sInstance != null) {
                sInstance = new SomeClass();
            }
        }
    }
    return sInstance;
}

interface DownloadCallback {
    void onSuccess(File file);
}

public void download(String url, DownloadCallback callback) {
    // ...
}

interface DownloadCallback {
    void onSuccess(File file);
}

private class DownloadTask {
    // guarded by itself
    private final List<DownloadCallback> mCallbacks = new ArrayList<>();

    public DownloadTask(String url, DownloadCallback callback) {
        mCallbacks.add(callback);
    }

    public void download() {
        // downloading ...
        File file = new File("downloaded-file");

        // bonus: 为什么需要拷贝 callback 列表？
        List<DownloadCallback> copy;
        synchronized (mCallbacks) {
            copy = new ArrayList<>(mCallbacks);
            mCallbacks.clear();
        }
        for (DownloadCallback callback : copy) {
            callback.onSuccess(file);
        }
    }

    public void addCallback(DownloadCallback callback) {
        synchronized (mCallbacks) {
            mCallbacks.add(callback);
        }
    }
}


private final ConcurrentMap<String, DownloadTask> mTasks = new ConcurrentHashMap<>();

public void download(String url, DownloadCallback callback) {
    File file = new File(getDestFilePath(url));
    if (file.exists()) {
        // 已经存在则不需要下载了
        callback.onSuccess(file);
        return;
    }
    DownloadTask task = new DownloadTask(url, callback);
    DownloadTask downloadingTask = mTasks.putIfAbsent(url, task);
    if (downloadingTask == null) {
        // 没有正在下载的任务时才需要下载
        task.download();
        return;
    }
    // 加入正在下载的任务的 callback 列表，
    downloadingTask.addCallback(callback);
}

private String getDestFilePath(String url) {
    return "url-to-file-path...";
}

/*
 * Note: we use "set_current_state()" _after_ the wait-queue add,
 * because we need a memory barrier there on SMP, so that any
 * wake-function that tests for the wait-queue being active
 * will be guaranteed to see waitqueue addition _or_ subsequent
 * tests in this thread will see the wakeup having taken place.
 *
 * The spin_unlock() itself is semi-permeable and only protects
 * one way (it only protects stuff inside the critical region and
 * stops them from bleeding out - it would still allow subsequent
 * loads to move into the critical region).
 */

static inline int waitqueue_active(wait_queue_head_t *q)
{
    return !list_empty(&q->task_list);
}

/*
 * set_current_state() includes a barrier so that the write of current->state
 * is correctly serialised wrt the caller's subsequent test of whether to
 * actually sleep:
 *
 *    set_current_state(TASK_UNINTERRUPTIBLE);
 *    if (do_i_need_to_sleep())
 *        schedule();
 *
 * If the caller does not need such serialisation then use __set_current_state()
 */
#define __set_current_state(state_value)                 \
    do { current->state = (state_value); } while (0)
#define set_current_state(state_value)                    \
    set_mb(current->state, (state_value))

 (*) set_mb(var, value)

     This assigns the value to the variable and then inserts a full memory
     barrier after it, depending on the function.  It isn't guaranteed to
     insert anything more than a compiler barrier in a UP compilation.

for_each_process_in_wait_queue_without_lock:
    if process.state is sleeping:
        wake it up
        remove it from wait-queue

CPU0                                CPU1
----------------        ------------------
                                    check_condition() => false
condition = true
wake_up()
                                      __set_current_state()
                                        schedule()

CPU0                              CPU1
----------------        ------------------
condition = true
wake_up()
                                    __set_current_state()
                                    mb();
                                     check_condition() => true

private int mSomeData;          // guarded by mDataSet
private int mSomeOtherData;     // guarded by mDataSet
private volatile boolean mDataSet;

public void foo() {
    if (mDataSet) {
        // you can now use mSomeData, mSomeOtherData
    }
}

public void bar() {
    mSomeData = 1;
    mSomeOtherData = 2;
    mDataSet = true;
}

private int mSomeData;
private int mSomeOtherData;
private boolean mDataSet;

public void foo() {
    if (mDataSet) {
        // 这个读内存屏障保证我们读到 mDataSet 后，也能读到 mSomeData/mSomeOtherData
        // 的最新值
        rmb();
        // you can now use mSomeData, mSomeOtherData
    }
}

public void bar() {
    mSomeData = 1;
    mSomeOtherData = 2;
    // 写内存屏障保证对 mSomeData/mSomeOtherData 的写在 mDataSet = true 之前执行
    wmb();
    mDataSet = true;
}