Chromium线程模型、消息循环

StartupTask create_threads =
base::Bind(&BrowserMainLoop::CreateThreads, base::Unretained(this));
startup_task_runner_->AddTask(create_threads);

staticbool PostTask(ID identifier,
                     consttracked_objects::Location& from_here,
                     constbase::Closure& task);
staticbool PostDelayedTask(ID identifier,
                            consttracked_objects::Location& from_here,
                            constbase::Closure& task,
                            base::TimeDelta delay);
staticbool PostNonNestableTask(ID identifier,
                                consttracked_objects::Location& from_here,
                                constbase::Closure& task);

第一个参数是一个线程的标识符，定义如下：

enumID {
    // The main thread in the browser.
    UI,
    // This is the thread that interacts with the database.
    DB,
    // This is the thread that interacts with the file system.
    FILE,
    // Used for file system operations that block user interactions.
    // Responsiveness of this thread affect users.
    FILE_USER_BLOCKING,
    // Used to launch and terminate Chrome processes.
    PROCESS_LAUNCHER,
    // This is the thread to handle slow HTTP cache operations.
    CACHE,
    // This is the thread that processes non-blocking IO, i.e. IPC and network.
    // Blocking IO should happen on other threads like DB, FILE,
    // FILE_USER_BLOCKING and CACHE depending on the usage.
    IO,
    // NOTE: do not add new threads here that are only used by a small number of
    // files. Instead you should just use a Thread class and pass its
    // MessageLoopProxy around. Named threads there are only for threads that
    // are used in many places.
    // This identifier does not represent a thread.  Instead it counts the
    // number of well-known threads.  Insert new well-known threads before this
    // identifier.
    ID_COUNT
  };

BrowserThread::PostTask(
    BrowserThread::DB, FROM_HERE,
        base::Bind(
            &GetUrlThumbnailTask, url_string, top_sites,
            base::Owned(j_callback), lookup_success_callback,
            lookup_failed_callback));

BrowserThread维护着一个全局变量g_globals，它的类型是BrowserThreadGlobals，定义如下：

structBrowserThreadGlobals {
  BrowserThreadGlobals()
      : blocking_pool(newbase::SequencedWorkerPool(3, "BrowserBlocking")) {
    memset(threads, 0, BrowserThread::ID_COUNT * sizeof(threads[0]));
    memset(thread_delegates, 0,
           BrowserThread::ID_COUNT * sizeof(thread_delegates[0]));
  }
  // This lock protects |threads|. Do not read or modify that array
  // without holding this lock. Do not block while holding this lock.
  base::Lock lock;
  // This array is protected by |lock|. The threads are not owned by this
  // array. Typically, the threads are owned on the UI thread by
  // BrowserMainLoop. BrowserThreadImpl objects remove themselves from this
  // array upon destruction.
  BrowserThreadImpl* threads[BrowserThread::ID_COUNT];
  // Only atomic operations are used on this array. The delegates are not owned
  // by this array, rather by whoever calls BrowserThread::SetDelegate.
  BrowserThreadDelegate* thread_delegates[BrowserThread::ID_COUNT];
  constscoped_refptr<base::SequencedWorkerPool> blocking_pool;
};
base::LazyInstance<BrowserThreadGlobals>::Leaky
    g_globals = LAZY_INSTANCE_INITIALIZER;
} // namespace

其中的threads数组保存了创建的BrowserThread实例的指针。在初始化的时候会给threads赋值：

voidBrowserThreadImpl::Initialize() {
  BrowserThreadGlobals& globals = g_globals.Get();
  base::AutoLock lock(globals.lock);
  DCHECK(identifier_ >= 0 && identifier_ < ID_COUNT);
  DCHECK(globals.threads[identifier_] == NULL);
  globals.threads[identifier_] = this;
}

BrowserThreadGlobals& globals = g_globals.Get();
  if(!target_thread_outlives_current)
    globals.lock.Acquire();
  base::MessageLoop* message_loop =
      globals.threads[identifier] ? globals.threads[identifier]->message_loop()
                                  : NULL;
  if(message_loop) {
    if(nestable) {
      message_loop->PostDelayedTask(from_here, task, delay);
    }else {
      message_loop->PostNonNestableDelayedTask(from_here, task, delay);
    }
  }

消息循环工作原理

MessageLoop实际上就是一个循环，不断从任务队列中取出任务，并执行，那么当所有任务都执行完毕之后呢？最直接的做法是，采用忙等待，不断检查任务队列中是否有新的任务。Chromium当然不会采用这么拙劣的方法。以MessagePumpDefault的实现为例，当所有任务执行完毕之后，会执行event对象的wait()函数，等待事件或信号唤醒继续循环执行。

ThreadRestrictions::ScopedAllowWait allow_wait;
if(delayed_work_time_.is_null()) {
  event_.Wait();
}else {
  TimeDelta delay = delayed_work_time_ - TimeTicks::Now();
  if(delay > TimeDelta()) {
    event_.TimedWait(delay);
  }else {
    // It looks like delayed_work_time_ indicates a time in the past, so we
    // need to call DoDelayedWork now.
    delayed_work_time_ = TimeTicks();
  }
}

唤醒的信号是怎么发送的呢？我们来看PostTask*的执行过程：

添加任务到队列时，如果发现任务队列为空，就会调用ScheduleWork启动消息循环，ScheduleWork具体的实现与采用的系统，以及采用的事件模型有关。还是以MessagePumpDefault为例，它的实现如下：

boolIncomingTaskQueue::PostPendingTask(PendingTask* pending_task) {
  ……
  boolwas_empty = incoming_queue_.empty();
  incoming_queue_.push(*pending_task);
  pending_task->task.Reset();
  if(always_schedule_work_ || (!message_loop_scheduled_ && was_empty)) {
    // Wake up the message loop.
    message_loop_->ScheduleWork();
    // After we've scheduled the message loop, we do not need to do so again
    // until we know it has processed all of the work in our queue and is
    // waiting for more work again. The message loop will always attempt to
    // reload from the incoming queue before waiting again so we clear this flag
    // in ReloadWorkQueue().
    message_loop_scheduled_ = true;
  }
  returntrue;
}
voidMessagePumpDefault::ScheduleWork() {
  // Since this can be called on any thread, we need to ensure that our Run
  // loop wakes up.
  event_.Signal();
}

为了减少锁的使用和锁的范围，Chromium采用了一个比较巧妙的方法：简单来讲，MessageLoop维护有两个队列，一个work_queue，一个incoming_queue。消息循环不断从work_queue取任务并执行，新加入任务放入incoming_queue。当work_queue中的任务都执行完后，再把incoming_queue拷贝到work_queue（需要加锁）。这样避免了每执行一个任务都要去加锁。

参考

理解WebKit和Chromium: 消息循环(Message Loop)

Chromium on Android: Android系统上Chromium主消息循环的实现分析

Chrome学习笔记（一）：线程模型，消息循环