Chromium多线程模型设计和实现分析

标签：file   mpio   net   poi   并发   写文件   选择   lld   tno   

       Chromium除了远近闻名的多进程架构之外，它的多线程模型也相当引人注目的。Chromium的多进程架构是为了解决网页的稳定性问题，而多线程模型则是为了解决网页的卡顿问题。为了达到这个目的，Chromium的多线程模型是基于异步通信的。也就是说，一个线程请求另外一个线程执行一个任务的时候，不需要等待该任务完成就可以去做其它事情，从而避免了卡顿。本文就分析Chromium的多线程模型的设计和实现。

老罗的新浪微博：http://weibo.com/shengyangluo，欢迎关注！

《Android系统源代码情景分析》一书正在进击的程序员网（http://0xcc0xcd.com）中连载，点击进入！

       有同学看到这里可能会有疑问，如果一个线程请求另外一个线程执行一个任务的时候，就是要等该任务完成之后才能做其它事情，那么使用异步通信有什么用呢？的确如此，但是Chromium提供这种基于异步通信的多线程模型，就是希望作为开发者的你在实现一个模块的时候，尽最大努力地设计好各个子模块及其对应的数据结构，使得它们在协作时可以最大程度地进行异步通信。因此，Chromium基于异步通信的多线程模型更多的是体现一种设计哲学。

       一个典型的异步通信过程如图1所示：

图1  线程异步通信过程

       Task-1被分解成三个子任务Task-1(1)、Task-1(2)和Task-1(3)。其中，Task-1(1)由Thread-1执行。Task-1(1)执行完成后，Thread-1通过我们在前面Chromium多线程通信的Closure机制分析一文分析的Closure请求Thread-2执行Task-1(2)。Task-1(2)执行完成后，Thread-2又通过一个Closure请求Thread-1执行Task-1(3)。至此，Task-1就执行完成。我们可以将第一个Closure看作是一个Request操作，而第二个Closure是一个Reply操作。这是一个典型的异步通信过程。当然，如果不需要知道异步通信结果，那么第二个Closure和Task-1(3)就是不需要的。

       假设Thread-1需要知道异步通信的结果，那么在图1中我们可以看到一个非常关键的点：Thread-1并不是什么也不干就只是等着Thread-2执行完成Task-1(2)，它趁着这个等待的空隙，干了另外一件事情——Task-2。如果我们将Thread-1看作是一个UI线程，那么就意味着这种异步通信模式是可以提高它的响应性的。

       为了能够完成上述的异步通信过程，一个线程的生命周期如图2所示：

图2 线程生命周期

       线程经过短暂的启动之后（Start），就围绕着一个任务队列（TaskQueue）不断地进行循环，直到被通知停止为止（Stop）。在围绕任务队列循环期间，它会不断地检查任务队列是否为空。如果不为空，那么就会将里面的任务（Task）取出来，并且进行处理。这样，一个线程如果要请求另外一个线程执行某一个操作，那么只需要将该操作封装成一个任务，并且发送到目标线程的任务队列去即可。

       为了更好地理解这种基于任务队列的线程运行模式，我们脑补一下另外一种常用的基于锁的线程运行模式。一个线程要执行某一个操作的时候，就直接调用一个代表该操作的一个函数。如果该函数需要访问全局数据或者共享数据，那么就需要进行加锁，避免其它线程也正在访问这些全局数据或者共享数据。这样做的一个好处是我们只需要关心问题的建模，而不需要关心问题是由谁来执行的，只要保证逻辑正确并且数据完整即可。当然坏处也是显然的。首先是为了保持数据完整性，也就是避免访问数据时出现竞争条件，代码里面充斥着各种锁。其次，如果多个线程同时获取同一个锁，那么就会产生竞争。这种锁竞争会带来额外的开销，从而降低线程的响应性。

       基于任务队列的线程运行模式，要求在对问题进行建模时，要提前知道谁是执行者。也就是说，在对问题进行建模时，需要指派好每一个子问题的执行者。这样我们为子问题设计数据结构时，就规定这些数据结构仅仅会被子问题的执行者访问。这样执行者在解决指派给它的问题时，就不需要进行加锁操作，因为在解决问题过程中需要访问的数据不会同时被其它执行者访问。这就是通过任务队列来实现异步通信的多线程模型的设计哲学。

       当然，这并不是说，基于任务队列的线程运行模式可以完全避免使用锁，因为任务队列本身就是一个线程间的共享资源。想象一下，一个线程要往里面添加任务，另一个线程要从里面将任务提取出来处理。因此，所有涉及到任务队列访问的地方都是需要加锁的。但是如果我们再仔细想想，那么就会发现，任务队列只是一个基础设施，它与具体的问题是无关的。因此，只要我们遵循上述设计哲学，就可以将代码里面需要加锁的地方仅限于访问任务队列的地方，从而就可以减少锁竞争带来的额外的开销。

       这样说来，似乎基于任务队列的线程运行模式很好，但是实际上它对问题建模提出了更高的要求，也就是进行子问题划分时，要求划分出来的子问题是正交的，这样我们才有可能为这些子问题设计出不会同时被访问的数据结构。看到“正交”两个字，是不是想起高数里面的向量空间的正交基了？或者傅里叶变换用到的一组三角函数了？其实道理就是一样一样的。

       好了，说了这么多，我们就步入到正题，分析Chromium多线程模型的设计和实现，也就是基于任务队列的线程运行模式涉及到核心类图，如图3所示：

图3 基于任务队列的线程运行模式核心类关系图

       Thread是一个用来创建带消息循环的类。当我们创建一个Thread对象后，调用它的成员函数Start或者StartWithOptions就可以启动一个带消息循环的线程。其中，成员函数StartWithOptions可以指定线程创建参数。当我们不需要这个线程时，就可以调用之前创建的Thread对象的成员函数Stop。

       Thread类继承了PlatformThread::Delegate类，并且重写了它的成员函数ThreadMain。我们知道，Chromium是跨平台的，这样各个平台创建线程使用的API有可能是不一样的。不过，我们可以通过PlatformThread::Delegate类为各个平台创建的线程提供一个入口点。这个入口点就是PlatformThread::Delegate类的成员函数ThreadMain。由于Thread类重写了父类PlatformThread::Delegate的成员函数ThreadMain，因此无论是哪一个平台，当它创建完成一个线程后，都会以Thread类的成员函数ThreadMain作为线程的入口点。

       Thread类有一个重要的成员变量message_loop_，它指向的是一个MessageLoop对象。这个MessageLoop对象就是用来描述线程的消息循环的。MessageLoop类内部通过成员变量run_loop_指向的一个RunLoop对象和成员变量pump_指向的一个MessagePump对象来描述一个线程的消息循环。

       一个线程在运行的过程中，可以有若干个消息循环，也就是一个消息循环可以运行在另外一个消息循环里面。除了最外层的消息循环，其余的消息的消息循环称为嵌套消息循环。我们为什么需要嵌套消息循环呢？这主要是跟模式对话框有关。

       考虑一个情景，我们在一个窗口弹出一个文件选择对话框。窗口必须要等到用户在文件选择对话框选择了文件之后，才能去做其它事情。窗口是在消息循环过程中打开文件对话框的，它要等待用户在文件选择对话框中选择文件 ，就意味着消息循环被中止了。由于文件选择对话框也是通过消息循环来响应用户输入的，因此如果打开的它窗口中止了消息循环，就会导致它无法响应用户输入。为了解决这个问题，就要求打开文件选择的窗口不能中止消息循环。方法就是该窗口创建一个子消息循环，该子消息循环负责处理文件选择对应框的输入事件，直到用户选择了一个文件为止。

       MessageLoop类的成员变量run_loop_指向的一个RunLoop对象就是用来记录线程当使用的消息循环的。RunLoop类有三个重要的成员变量：

       1. message_loop_，记录一个RunLoop对象关联的MessageLoop对象。

       2. previous_loop_，记录前一个消息循环，当就是包含当前消息循环的消息循环。

       3. run_depth_，记录消息循环的嵌套深度。

       MessageLoop类的成员变量pump_指向的一个MessagePump对象是用来进行消息循环的，也就是说，Thread类描述的线程通过MessagePump类进入到消息循环中去。

       Thread类将消息划分为三类，分别通过以下三个成员变量来描述：

       1. work_queue_，指向一个TaskQueue对象，用来保存那些需要马上处理的消息。

       2. delayed_work_queue_，指向一个DelayedTaskQueue，用来保存那些需要延迟一段时间再处理的消息。

       3. deferred_non_nestable_work_queue_，指向一个TaskQueue对象，用来保存那些不能够在嵌套消息循环中处理的消息。

       一个MessagePump对象在进行消息循环时，如果发现消息队列中有消息，那么就需要通知关联的MessageLoop对象进行处理。通知使用的接口就通过MessagePump::Delegate类来描述。

       MessagePump::Delegate类定义了四个成员函数，如下所示：

       1. DoWork，用来通知MessageLoop类处理其成员变量work_queue_保存的消息。

       2. DoDelayedWork，用来通知MessageLoop类处理其成员变量delayed_work_queue_保存的消息。

       3. DoIdleWork，用来通知MessageLoop类当前无消息需要处理，MessageLoop类可以利用该间隙做一些Idle Work。

       4. GetQueueingInformation，用来获取MessageLoop类内部维护的消息队列的信息，例如消息队列的大小，以及下一个延迟消息的处理时间。

       有了前面的基础知识，接下来我们就可以大概描述Thread类描述的线程的执行过程。

       首先是线程的启动过程：

       1. 调用Thread类的成员函数Start或者StartWithOptions启动一个线程，并且以Thread类的成员函数ThreadMain作为入口点。

       2. Thread类的成员函数ThreadMain负责创建消息循环，也就是通过MessageLoop类创建消息循环。

       3. MessageLoop类在创建消息循环的过程中，会通过成员函数Init创建用来一个用来消息循环的MessagePump对象。

       4. 消息循环创建完成之后，调用MessageLoop类的成员函数Run进入消息循环。

       5. MessageLoop类的成员函数Run创建一个RunLoop对象，并且调用它的成员函数Run进入消息循环。注意，该RunLoop对象在创建的过程，会关联上当前线程使用的消息循环，也就是创建它的MessageLoop对象。

       6. RunLoop类的成员函数Run负责建立好消息循环的嵌套关系，也就是设置好它的成员变量previous_loop_和run_depth_等，然后就会调用其关联的MessageLoop对象的成员函数RunHandler进入消息循环。

       7. MessageLoop类的成员函数RunHandler调用成员变量pump_描述的一个MessagePump对象的成员函数Run进入消息循环。

       接下来是向线程的消息队列发送消息的过程。这是通过MessageLoop类的以下四个成员函数向消息队列发送消息的：

       1. PostTask，发送需要马上进行处理的并且可以在嵌套消息循环中处理的消息。

       2. PostDelayedTask，发送需要延迟处理的并且可以在嵌套消息循环中处理的消息。

       3. PostNonNestableTask，发送需要马上进行处理的并且不可以在嵌套消息循环中处理的消息。

       4. PostNonNestableDelayedTask，发送需要延迟处理的并且不可以在嵌套消息循环中处理的消息。

       向线程的消息队列发送了新的消息之后，需要唤醒线程，这是通过调用MessagePump类的成员函数Schedule进行的。线程被唤醒之后 ，就会分别调用MessageLoop类重写父类MessagePump::Delegate的两个成员函数DoWork和DoDelayedWork对消息队列的消息进行处理。如果没有消息可以处理，就调用MessageLoop类重写父类MessagePump::Delegate的成员函数DoIdleWork通知线程进入Idle状态，这时候线程就可以做一些Idle Work。

       MessageLoop类的成员函数DoWork在处理消息的过程中，按照以下三个类别进行处理：

       1. 对于可以马上处理的消息，即保存在成员变量work_queue_描述的消息队列的消息，执行它们的成员函数Run。

       2. 对于需要延迟处理的消息，将它们保存在成员变量delayed_work_queue_描述的消息队列中，并且调用成员变量pump_指向的一个MessagePump对象的成员函数ScheduleDelayedWork设置最早一个需要处理的延迟消息的处理时间，以便该MessagePump对象可以优化消息循环逻辑。

       3. 对于可以马上处理但是不可以在嵌套消息循环中处理的消息，如果线程是处理嵌套消息循环中，那么将它们保存在成员变量deferred_non_nestable_work_queue_描述的消息队列中，这些消息将会在线程进入Idle状态时，并且是处理最外层消息循环时，得到处理。

       以上就是Thread类描述的线程的大概执行过程，接下来我们通过源码分析详细描述这些过程。

       我们首先看线程的启动过程，即Thread类的成员函数Start的实现，如下所示：

bool Thread::Start() {
  Options options;
  ......
  return StartWithOptions(options);
}

       这个函数定义在文件external/chromium_org/base/threading/thread.cc中。

       Thread类的成员函数Start调用另外一个成员函数StartWithOptions来启动一个线程，后者可以通过一个类型为Options的参数指定线程的启动参数，这里没有指定，意味着采用默认参数启动一个线程。

       Thread类的成员函数StartWithOptions的实现如下所示：

bool Thread::StartWithOptions(const Options& options) {
  ......

  StartupData startup_data(options);
  startup_data_ = &startup_data;

  if (!PlatformThread::Create(options.stack_size, this, &thread_)) {
    ......
    return false;
  }

  // Wait for the thread to start and initialize message_loop_
  base::ThreadRestrictions::ScopedAllowWait allow_wait;
  startup_data.event.Wait();

  // set it to NULL so we don‘t keep a pointer to some object on the stack.
  startup_data_ = NULL;
  started_ = true;

  ......
  return true;
}

       Thread类的成员函数StartWithOptions首先是将线程启动参数封装一个在栈上分配的StartupData对象中，并且这个StartupData对象的地址会保存在Thread类的成员变量startup_data_中。接下来再调用由平台实现的PlatformThread类的静态成员函数Create创建一个线程。最后通过上面封装的StartupData对象的成员变量event描述的一个WaitableEvent对象等待上述创建的线程启动完成。

       一般情况下，线程是不可以进入等待状态的，因为这样会降低线程的响应性。但是有时候线程不得不进入等待状态，例如现在这个情况，当前线程必须要等新创建的线程启动完成之后才能返回，否则的话有可能新创建的线程还没有启动完成，前面在栈上分配的StartupData对象就已经被释放，这样会导致新创建的线程无法访问它的启动参数。

       当新创建的线程启动完成之后，就会通过上述的WaitableEvent对象唤醒当前线程，当前线程将Thread类的成员变量startup_data_置为NULL，避免它引用一个即将无效的在栈上分配的StartupData对象，并且将Thread类的成员变量started_的值设置为true，表示新创建的线程已经启动完毕。

       接下来我们继续分析PlatformThread类的静态成员函数Create的实现。以Android平台为例，它的实现如下所示：

bool PlatformThread::Create(size_t stack_size, Delegate* delegate,
                            PlatformThreadHandle* thread_handle) {
  base::ThreadRestrictions::ScopedAllowWait allow_wait;
  return CreateThread(stack_size, true /* joinable thread */,
                      delegate, thread_handle, kThreadPriority_Normal);
}

      PlatformThread类的静态成员函数Create调用了另外一个函数CreateThread来创建一个线程，后者的实现如下所示：

bool CreateThread(size_t stack_size, bool joinable,
                  PlatformThread::Delegate* delegate,
                  PlatformThreadHandle* thread_handle,
                  ThreadPriority priority) {
  ......

  bool success = false;
  pthread_attr_t attributes;
  pthread_attr_init(&attributes);

  // Pthreads are joinable by default, so only specify the detached
  // attribute if the thread should be non-joinable.
  if (!joinable) {
    pthread_attr_setdetachstate(&attributes, PTHREAD_CREATE_DETACHED);
  }

  // Get a better default if available.
  if (stack_size == 0)
    stack_size = base::GetDefaultThreadStackSize(attributes);

  if (stack_size > 0)
    pthread_attr_setstacksize(&attributes, stack_size);

  ThreadParams params;
  params.delegate = delegate;
  params.joinable = joinable;
  params.priority = priority;
  params.handle = thread_handle;

  pthread_t handle;
  int err = pthread_create(&handle,
                           &attributes,
                           ThreadFunc,
                           &ms);
  success = !err;
  ......

  pthread_attr_destroy(&attributes);

  .....

  return success;
}

       从这里就可以看到，Android平台调用POSIX线程库中的函数pthread_create创建了一个线程，并且指定新创建的线程的入口点函数为ThreadFunc，同时传递给该入口点函数的参数为一个ThreadParams对象，该ThreadParams对象封装了线程启动过程中需要使用到的一系列参数。    

       新创建线程的入口点函数ThreadFunc的实现如下所示：

void* ThreadFunc(void* params) {
  ......
  ThreadParams* thread_params = static_cast<ThreadParams*>(params);

  PlatformThread::Delegate* delegate = thread_params->delegate;
  ......

  delegate->ThreadMain();

  ......
  return NULL;
}

       函数ThreadFunc首先将参数params转换为一个ThreadParams对象。有了这个ThreadParams对象之后，就可以通过它的成员变量delegate获得一个PlatformThread::Delegate对象。从前面的调用过程可以知道，这个PlatformThread::Delegate对象实际上是一个Thread对象，用来描述新创建的线程。得到了用来描述新创建线程的Thread对象之后，就可以调用它的成员函数ThreadMain继续启动线程了。

       Thread类的成员函数ThreadMain的实现如下所示：

void Thread::ThreadMain() {
  {
    ......
    scoped_ptr<MessageLoop> message_loop;
    if (!startup_data_->options.message_pump_factory.is_null()) {
      message_loop.reset(
          new MessageLoop(startup_data_->options.message_pump_factory.Run()));
    } else {
      message_loop.reset(
          new MessageLoop(startup_data_->options.message_loop_type));
    }

    ......

    message_loop_ = message_loop.get();

    Init();

    running_ = true;
    startup_data_->event.Signal();

    ......

    Run(message_loop_);
    running_ = false;

    ......

    message_loop_ = NULL;
  }
}

       回忆前面分析的Thread类的成员函数StartWithOptions，它已经将用来描述线程启动参数的一个Options对象保存在成员变量startup_data_描述的一个StartupData对象中，因此我们就可以重新获取这个Options对象。

       当Options类的成员变量message_pump_factory不等于NULL时，就表示新创建线程使用的Message Pump通过该成员变量描述的一个Callback对象来创建，也就是调用该Callback对象的成员函数Run来创建。关于Chromium的Callback机制，可以参考前面Chromium多线程通信的Closure机制分析一文。有了Message Pump之后，就可以创建一个Message Loop了。该Message Loop最终会保存在Thread类的成员变量message_loop_中。

       一般我们不通过Options类的成员变量message_pump_factory来创建Message Pump，而是通过另外一个成员变量message_loop_type来创建指定Message Loop的类型 ，从而确定要创建的Message Pump，这些逻辑都封装在MessageLoop类的构造函数中。

       创建好Message Loop之后，线程的启动工作就完成了，接下来新创建的线程就需要进入到初始化状态，这是通过调用Thread类的成员函数Init实现的。Thread类的成员函数Init一般由子类重写，这样子类就有机会执行一些线程初始化工作。

       再接下来，新创建的线程就需要进入运行状态，这是通过调用Thread类的成员函数Run实现的。不过在新创建线程进入运行状态之前，还会做两件事情。第一件事情是将Thread类的成员变量running_设置为true，表示新创建的线程正在运行。第二件事情是通过Thread类的成员变量startup_data_指向的一个StartupData对象的成员变量event描述的一个WaitableEvent唤醒请求创建新线程的线程。

       最后，当Thread类的成员函数Run执行完成返回后，需要将Thread类的成员变量running_和message_loop_分别重置为false和NULL，表示新创建的线程已经运行结束了，因此就不再需要Message Loop了。

       接下来我们首先分析线程的Message Loop的创建过程，也就是MessageLoop类的构造函数的实现，以完成线程的启动过程，然后再分析线程的运行过程，也就是Thread类的成员函数Run的实现。

       我们假设线程的Message Loop是通过Message Loop Type来创建的，对应的MessageLoop类构造函数的实现如下所示：

MessageLoop::MessageLoop(Type type)
    : type_(type),
      nestable_tasks_allowed_(true),
      ......
      run_loop_(NULL) {
  Init();

  pump_ = CreateMessagePumpForType(type).Pass();
}

       这个函数定义在文件external/chromium_org/base/message_loop/message_loop.cc。 

       MessageLoop类的成员变量type_描述的是消息循环的类型，nestable_tasks_allowed_描述当前是否允许处理嵌套消息，runn_loop_描述的是当前使用的消息循环。 

       MessageLoop类构造函数首先是调用成员函数Init执行初始化工作，接着再调用成员函数CreateMessagePumpForType根据消息循环的类型创建一个Message Pump。接下来我们就分别分析这两个成员函数的实现。

       MessageLoop类的成员函数Init的实现如下所示：

LazyInstance<base::ThreadLocalPointer<MessageLoop> >::Leaky lazy_tls_ptr =
    LAZY_INSTANCE_INITIALIZER;

......

void MessageLoop::Init() {
  ......
  lazy_tls_ptr.Pointer()->Set(this);

  incoming_task_queue_ = new internal::IncomingTaskQueue(this);
  message_loop_proxy_ =
      new internal::MessageLoopProxyImpl(incoming_task_queue_);
  thread_task_runner_handle_.reset(
      new ThreadTaskRunnerHandle(message_loop_proxy_));
}

      MessageLoop类的成员函数Init首先将当前创建的MessageLoop对象保存在全局变量lazy_tls_ptr指向一块线程局部存储中，这样我们就可以通过MessageLoop类的静态成员函数current获得当前线程的消息循环，如下所示：

MessageLoop* MessageLoop::current() {
  ......
  return lazy_tls_ptr.Pointer()->Get();
}

      回到MessageLoop类的成员函数Init中，接下来它创建了一个任务队列，并且保存在成员变量incoming_queue_中。这个任务队列通过IncomingQueue类来描述，它的定义如下所示：

class BASE_EXPORT IncomingTaskQueue
    : public RefCountedThreadSafe<IncomingTaskQueue> {
 public:
  ......

  bool AddToIncomingQueue(const tracked_objects::Location& from_here,
                          const Closure& task,
                          TimeDelta delay,
                          bool nestable);

  ......
 
  void ReloadWorkQueue(TaskQueue* work_queue);

  ......

  void WillDestroyCurrentMessageLoop();
 
  ......

 private:
  ......

  TaskQueue incoming_queue_;

  ......

  MessageLoop* message_loop_;

  ......
};

       IncomingQueue类有两个重要的成员变量：

       1. incoming_queue_，它描述的是一个TaskQueue，代表的是线程的消息队列，也就是所有发送给线程的消息都保存在这里。

       2. message_loop_，它指向一个MessageLoop对象，描述的是线程的消息循环。

       IncomingQueue类有三个重要的成员函数：

       1. AddToIncomingQueue，用来向成员变量incoming_queue_描述的消息队列发送一个消息，并且唤醒线程进行处理。

       2. ReloadWorkQueue，用来提取成员变量incoming_queue_描述的消息队列中的消息，并且保存在参数work_queue中。

       3. WillDestroyCurrentMessageLoop，当该函数被调用时，会将成员变量message_loop_的值设置为NULL，使得我们不能够再向线程发送消息，也就是请求线程执行某一个操作。

       IncomingQueue类的上述成员变量和成员函数我们后面分析消息的发送和处理再详细分析。现在返回到MessageLoop类的成员函数Init中，它接下来创建了一个MessageLoopProxyImpl对象和一个ThreadTaskRunnerHandle对象，分别保存在成员变量message_loop_proxy_和thread_task_runner_handle中，前者封装了当前线程的消息队列，后者又封装了前者。它们与MessageLoop类一样，都是可以用来向线程的消息队列发送消息，这意味着我们有三种方式向线程的消息队列发送消息，后面分析消息的发送过程时我们再详细分析。

       MessageLoop类的成员函数Init执行完成后，回到MessageLoop类的构造函数中，接下来它调用另外一个成员函数CreateMessagePumpForType根据消息循环的类型创建一个消息泵（Message Pump），并且保存在成员变量pump_中。

       MessageLoop类的成员函数CreateMessagePumpForType的实现如下所示：

#if defined(OS_IOS)
typedef MessagePumpIOSForIO MessagePumpForIO;
#elif defined(OS_NACL)
typedef MessagePumpDefault MessagePumpForIO;
#elif defined(OS_POSIX)
typedef MessagePumpLibevent MessagePumpForIO;
#endif
......

scoped_ptr<MessagePump> MessageLoop::CreateMessagePumpForType(Type type) {
  ......

#if defined(OS_IOS) || defined(OS_MACOSX)
#define MESSAGE_PUMP_UI scoped_ptr<MessagePump>(MessagePumpMac::Create())
#elif defined(OS_NACL)
// Currently NaCl doesn‘t have a UI MessageLoop.
// TODO(abarth): Figure out if we need this.
#define MESSAGE_PUMP_UI scoped_ptr<MessagePump>()
#else
#define MESSAGE_PUMP_UI scoped_ptr<MessagePump>(new MessagePumpForUI())
#endif

  if (type == MessageLoop::TYPE_UI) {
    if (message_pump_for_ui_factory_)
      return message_pump_for_ui_factory_();
    return MESSAGE_PUMP_UI;
  }
  if (type == MessageLoop::TYPE_IO)
    return scoped_ptr<MessagePump>(new MessagePumpForIO());

#if defined(OS_ANDROID)
  if (type == MessageLoop::TYPE_JAVA)
    return scoped_ptr<MessagePump>(new MessagePumpForUI());
#endif

  ......
  return scoped_ptr<MessagePump>(new MessagePumpDefault());
}

       这个函数定义在文件external/chromium_org/base/message_loop/message_loop.cc中。

       上面的代码通过一系列宏来适配不同的平台，这里我们只考虑Android平台，这意味着MessagePumpForIO定义为MessagePumpLibevent，MESSAGE_PUMP_UI定义为scoped_ptr<MessagePump>(new MessagePumpForUI())。

       从MessageLoop类的成员函数CreateMessagePumpForType的实现可以知道：

       1. 如果消息循环的类型为MessageLoop::TYPE_UI，那么对应的消息泵为MessagePumpForUI，或者由函数指针message_pump_for_ui_factory_指向的函数创建。但是一般不设置函数指针message_pump_for_ui_factory_，因此，类型为MessageLoop::TYPE_UI的消息循环对应的消息泵为MessagePumpForUI。在Chromium中，消息循环类型为MessageLoop::TYPE_UI的线程称为UI线程，也就是应用程序的主线程。

       2. 如果消息循环的类型为MessageLoop::TYPE_IO，那么对应的消息泵为MessagePumpForIO，即MessagePumpLibevent。在Chromium中，消息循环类型为MessageLoop::TYPE_IO的线程称为IO线程，但是这里的IO不是读写文件的意思，而是执行IPC的意思。

       3. 如果消息循环的类型为MessageLoop::TYPE_JAVA，那么对应的消息泵为MessagePumpForUI。在Chromium中，消息循环类型为MessageLoop::TYPE_JAVA的线程称为JAVA线程，它们与UI线程一样，在JAVA层具有自己的消息循环。

       4. 其余类型的消息循环，对应的消息泵为MessagePumpDefault。

       总结来说，就是在Android平台上，涉及到的消息泵有MessagePumpForUI、MessagePumpForIO和MessagePumpDefault三种，各自有不同的用途，其中MessagePumpForUI适用于在Java层具有自己的消息循环的UI线程和Java线程，MessagePumpLibevent适用于用来负责执行IPC的IO线程，MessagePumpDefault适用于其它的一般线程。我们先从一般性出发，分析MessagePumpDefault的实现，后面再分析MessagePumpForUI和MessagePumpForIO的实现。

       MessagePumpDefault类继承于MessagePump类，它的定义如下所示：

class MessagePumpDefault : public MessagePump {
 public:
  MessagePumpDefault();
  virtual ~MessagePumpDefault();

  // MessagePump methods:
  virtual void Run(Delegate* delegate) OVERRIDE;
  virtual void Quit() OVERRIDE;
  virtual void ScheduleWork() OVERRIDE;
  virtual void ScheduleDelayedWork(const TimeTicks& delayed_work_time) OVERRIDE;

 private:
  // This flag is set to false when Run should return.
  bool keep_running_;

  // Used to sleep until there is more work to do.
  WaitableEvent event_;

  // The time at which we should call DoDelayedWork.
  TimeTicks delayed_work_time_;

  DISALLOW_COPY_AND_ASSIGN(MessagePumpDefault);
};

       MessagePumpDefault类重写了父类MessagePump的成员函数Run、Quit、ScheduleWork和ScheduleDelayedWork，后面我们分析消息循环的执行过程和消息的发送过程时，会看到它们的具体实现。

       MessagePumpDefault类具有三个成员变量：

       1. keep_running_，类型为bool，表示消息循环是否需要继续执行。只要线程不退出，消息循环就要持续执行。

       2. event_，类型为WaitableEvent，表示一个可以进行Wait/Wake的事件变量。当线程的消息队列为空时，线程就通过它进入到Wait状态，而当向线程的消息队列发送了一个消息时，就通过它唤醒线程。

       3. delayed_work_time_，类型为TimeTicks，表示线程进入Wait状态的超时时间。达到超时时间之后，线程就会自动唤醒，然后处理那些延迟消息。

       这样，一个消息循环及其对应的消息泵就创建完毕，回到Thread类的成员函数ThreadMain中，接下来它调用成员函数Run使得线程进入到运行状态，也就是围绕消息队列进行不断的循环，直到线程退出为止。

       Thread类的成员函数Run的实现如下所示：

void Thread::Run(MessageLoop* message_loop) {
  message_loop->Run();
}

       Thread类的成员函数Run调用参数message_loop指向的一个MessageLoop对象的成员函数Run使得线程进入运行状态。

       MessageLoop类的成员函数Run的实现如下所示：

void MessageLoop::Run() {
  RunLoop run_loop;
  run_loop.Run();
}

       MessageLoop类的成员函数Run在栈上创建了一个RunLoop对象，然后通过调用该RunLoop对象的成员函数Run使得线程进入运行状态。

       前面提到，RunLoop的作用是用来建立消息循环的层次关系的，主要是通过它的两个成员变量previous_run_loop_和run_depth_来实现，此外，它还有一个成员变量loop_，用来关联它所对应的消息循环。

       RunLoop类的上述三个成员变量的定义如下所示：

class BASE_EXPORT RunLoop {
public:
  ......

 private:
  ......

  MessageLoop* loop_;

  // Parent RunLoop or NULL if this is the top-most RunLoop.
  RunLoop* previous_run_loop_;

  ......

  // Used to count how many nested Run() invocations are on the stack.
  int run_depth_;

  ......
};

      它们在RunLoop类的构造函数被初始化，如下所示：

RunLoop::RunLoop()
    : loop_(MessageLoop::current()),
      previous_run_loop_(NULL),
      run_depth_(0),
      ...... {
  ......
}

       从这里我们就可以看到，一个RunLoop关联的消息循环就是当前线程使用的消息循环。这个消息循环可以通过调用前面提到的MessageLoop类的静态成员函数current获得。

       RunLoop类的成员变量previous_run_loop_和run_depth_分别被初始化为NULL和0，表示还没有建立好层次关系，但是当RunLoop类的成员函数Run被调用时，它们就会被设置，从而形成层次关系。

       从前面的调用过程可以知道，RunLoop类的成员函数Run在MessageLoop类的成员函数Run中调用，它的实现如下所示：

void RunLoop::Run() {
  if (!BeforeRun())
    return;
  loop_->RunHandler();
  AfterRun();
}

       在调用成员变量loop_指向的一个MessageLoop对象的成员函数RunHandler进入消息循环前后，RunLoop类的成员函数Run分别调用了BeforeRun和AfterRun两个成员函数，目的就是为了建立好消息循环的层次关系，它们的实现如下所示：

bool RunLoop::BeforeRun() {
  DCHECK(!run_called_);
  run_called_ = true;

  // Allow Quit to be called before Run.
  if (quit_called_)
    return false;

  // Push RunLoop stack:
  previous_run_loop_ = loop_->run_loop_;
  run_depth_ = previous_run_loop_? previous_run_loop_->run_depth_ + 1 : 1;
  loop_->run_loop_ = this;

  running_ = true;
  return true;
}

void RunLoop::AfterRun() {
  running_ = false;

  // Pop RunLoop stack:
  loop_->run_loop_ = previous_run_loop_;

  // Execute deferred QuitNow, if any:
  if (previous_run_loop_ && previous_run_loop_->quit_called_)
    loop_->QuitNow();
}

       MessageLoop类的成员变量run_loop_记录的是消息循环当前使用的Run Loop，因此，RunLoop类的成员函数BeforeRun会将当前正在处理的RunLoop对象记录在其成员变量loop_指向的一个MessageLoop对象的成员变量run_loop_中，而该MessageLoop对象的成员变量run_loop_原来指向的RunLoop对象则记录在当前正在处理的RunLoop对象的成员变量previous_run_loop_中，从而就形成一个Run Loop调用栈。此外，第一个Run Loop的Run Depth被设置为1，后面的Run Loop的Run Depth依次增加1。

       从上面的分析就可以看出，RunLoop类的成员函数BeforeRun执行的是一个Run Loop入栈操作，相应地，RunLoop类的成员函数AfterRun执行的是一个Run Loop出栈操作，它将消息循环当前使用的Run Loop恢复为前一个Run Loop。

       RunLoop类的成员变量running_描述的是一个Run Loop当前是否正在被消息循环使用，因此，在RunLoop类的成员函数BeforeRun和AfterRun中，它的值分别被设置为true和false。

       RunLoop类的成员变量quit_called_描述的是一个Run Loop是否收到退出请求。如果一个Run Loop当前正在消息循环使用，并且又收到了退出请求，那么就将会导致消息循环退出。这样就会导致以下两种情况：

       1. 一个Run Loop在即将被消息循环使用之前，就已经收到了退出请求，那么就不会被消息循环使用，表现就为在RunLoop类的成员函数BeforeRun中，如果当前正在处理的RunLoop对象的成员变量quit_called_的值等于true，那么就返回一个false值给调用者，表示当前正在处理的RunLoop对象不能够进入消息循环。

       2. 一个Run Loop在被消息循环使用期间，前一个Run Loop收到了退出请求，那么当前Run Loop结束使用之后，禁止返回到前一个Run Loop。这意味着要结束消息循环，表现就为在RunLoop类的成员函数AfterRun中，如果发现当前正在处理的RunLoop对象的成员变量previous_run_loop_不为NULL，并且它指向的一个RunLoop对象的成员变量quit_called的值被设置为true，那么就会调用当前正在处理的RunLoop对象的成员变量loop_指向的一个MessageLoop对象的成员函数QuitNow退出消息循环。

       回到RunLoop类的成员函数Run中，在调用成员函数BeforeRun成功建立好消息循环的层次关系之后，就通过当前正在处理的RunLoop对象进入到下一层消息循环中，这是通过调用当前正在处理的RunLoop对象的成员变量loop_指向的一个MessageLoop对象的成员函数RunHandler实现的。从前面的分析可以知道，该MessageLoop对象描述的是就是当前线程使用的消息循环。

       MessageLoop类的成员函数RunHandler的实现如下所示：

void MessageLoop::RunHandler() {
  ......

  pump_->Run(this);
}

      MessageLoop类的成员函数RunHandler通过调用成员变量pump_指向的一个MessagePump对象的成员函数Run进入消息循环。前面我们假设该MessagePump对象是一个MessagePumpDefault对象，因此接下来我们继续分析MessagePumpDefault类的成员函数Run的实现，如下所示：

void MessagePumpDefault::Run(Delegate* delegate) {
  ......

  for (;;) {
    ......

    bool did_work = delegate->DoWork();
    if (!keep_running_)
      break;

    did_work |= delegate->DoDelayedWork(&delayed_work_time_);
    if (!keep_running_)
      break;

    if (did_work)
      continue;

    did_work = delegate->DoIdleWork();
    if (!keep_running_)
      break;

    if (did_work)
      continue;

    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();
      }
    }
    // Since event_ is auto-reset, we don‘t need to do anything special here
    // other than service each delegate method.
  }

  keep_running_ = true;
}

       这个函数定义在文件external/chromium_org/base/message_loop/message_pump_default.cc中。

       参数delegate是一个Delegate类型的指针，但是从上面的调用过程可以知道，它指向的是一个MessageLoop对象。

       MessagePumpDefault类的成员函数Run在一个for循环中不断地通过调用参数delegate指向的一个MessageLoop对象的成员函数DoWork和DoDelayedWork检查线程的消息队列是否有任务需要处理。如果没有，再调用该MessageLoop对象的成员函数DoIdleWork处理一些适用在线程空闲时进行的任务。

       MessageLoop类的成员函数DoWork、DoDelayedWork和DoIdleWork的返回值均为一个布尔值。当这个布尔值等于true的时候，就表示线程处理了一些任务。在这种情况下，就需要重新执行一遍for循环，这是因为上述三个函数在处理任务的过程中，可能又往线程的消息队列发送了新的任务，因此需要for循环检查进行检查，以及新发送的任务能够得到及时处理。

       另一方面，如果MessageLoop类的成员函数DoWork、DoDelayedWork和DoIdleWork的返回值均为false，那就表示线程当前实在是无事可做。这时候就不适合重新执行一遍for循环，因此这会使得线程在空转。在这种情况下，最好的方式就是让线程进入睡眠状态，以便将CPU释放出来。那么线程什么时候需要唤醒呢？

       在两种情况下，线程需要从睡眠状态唤醒过来。第一种情况是线程的消息队列有新的消息加入的时候，这时候由发送消息的线程进行唤醒。第二种情况是，线程有一个延时消息需要处理，那么当系统达到该消息的处理时间时，线程就需要自动唤醒过来。

       如果线程有一个延时消息需要处理，那么MessagePumpDefault类的成员变量delayed_work_time_就表示该消息在将来执行的时间点。注意，如果线程具有多个延时消息，那么MessagePumpDefault类的成员变量delayed_work_time_描述的是最早的延时点，这时候线程最多就只能睡眠到该时间点，然后自动唤醒过来。还有一点需要注意的是，如果最早的延时点小于系统的当前时间，那么线程就不可以睡眠，而要马上重新执行for循环，以便可以对已经超过了时间点处理的消息进行处理。如果线程没有延时消息需要处理，那么线程就不会设置自动唤醒时间，而是一直处理睡眠状态，直到被其它线程唤醒为止。

       无论线程是通过哪一种情况下进行睡眠状态，都是通过MessagePumpDefault类的成员变量event_描述的一个WaitableEvent对象进行，即通过调用它的成员函数Wait和TimedWait进行的。

       WaitableEvent是有效地实现线程消息循环的一个重要类。通过WaitableEvent类，线程可以在无消息处理时进入睡眠状态，并且在有消息处理时从睡眠状态唤醒过来，从而避免了不断地轮循消息队列是否有消息处理的操作。因为消息队列可能在大多数情况下都是空的，对它进行不断轮循将会浪费CPU周期。

       因为WaitableEvent类是如此重要，因此接下来我们先分析它的实现，然后再继续分析线程处理消息的过程，也就是MessageLoop类的成员函数DoWork、DoDelayedWork和DoIdleWork的实现。

       WaitableEvent类的定义如下所示：

class BASE_EXPORT WaitableEvent {
 public:
  ......
  WaitableEvent(bool manual_reset, bool initially_signaled);
  ......

  void Signal();
  ......
  void Wait();
  ......
  bool TimedWait(const TimeDelta& max_time);

  ......

  class Waiter {
   public:
    ......
    virtual bool Fire(WaitableEvent* signaling_event) = 0;
    ......
    virtual bool Compare(void* tag) = 0;
    ......
  };

  ......

  struct WaitableEventKernel :
      public RefCountedThreadSafe<WaitableEventKernel> {
   public:
    WaitableEventKernel(bool manual_reset, bool initially_signaled);

    bool Dequeue(Waiter* waiter, void* tag);

    base::Lock lock_;
    const bool manual_reset_;
    bool signaled_;
    std::list<Waiter*> waiters_;

    ......
  };

  ......

  bool SignalAll();
  bool SignalOne();
  void Enqueue(Waiter* waiter);

  scoped_refptr<WaitableEventKernel> kernel_;
  ......
};

       这个类定义在文件external/chromium_org/base/synchronization/waitable_event.h中。

       这里我们只讨论Android平台相关的实现。WaitableEvent类提供两个最基本的功能：Wait和Signal。Wait操作使得线程进入睡眠状态，而Signal操作使得线程从睡眠状态唤醒过来。

       在WaitableEvent类中，Wait操作对应的两个成员函数为Wait和TimedWait。前者使得线程一直处理唤醒状态，直到被其它线程唤醒为止，而后者使得线程进入到睡眠状态的时间为有限时间，并且在超过该时间后，线程自动唤醒。

       在WaitableEvent类中，Signal操作对应的成员函数为Signal，内部通成员函数SignalAll和SignalOne实现。前者唤醒所有的等待者，而后者只唤醒其中一个等待者。

       等待者通过内部类Waiter描述，它有Fire和Compare两个成员函数。一个Waiter需要唤醒时，它的成员函数Fire就会被调用。Waiter类的成员函数Compare用来比较一个Waiter与另外一个Waiter是否相同。

       一个WaitableEvent可以有若干个Waiter，这些Waiter通过WaitableEvent类的成员函数Enqueue加入到成员变量kernel_指向的一个WaitableEventKernel对象的成员变量waiters_描述的一个列表中。

       WaitableEventKernel类除了上述的成员变量waiters_之外，还具有以下三个成员变量：

       1. lock_，一个互斥锁，用来保护成员变量waiters_的并发访问。

       2. manual_reset_，一个布尔变量，用来表示一个WaitableEvent被唤醒的时候，是否需要手动设置才变为Signaled状态。

       3. signaled_，一个布尔变量，用来表示一个WaitableEvent是否处于Signaled状态。

Chromium多线程模型设计和实现分析

标签：file   mpio   net   poi   并发   写文件   选择   lld   tno   

原文地址：http://blog.csdn.net/luoshengyang/article/details/46855395