WebRTC源码分析-线程基础之消息循环，消息投递

前言

如之前的总述文章所述，rtc::Thread类封装了WebRTC中线程的一般功能，比如设置线程名称，启动线程执行用户代码，线程的join，sleep，run，stop等方法；同时也提供了线程内部的消息循环，以及线程之间以同步、异步方式投递消息，同步方式在目标线程执行方法并返回结果等线程之间交互的方式；另外，每个线程均持有SocketServer类成员对象，该类实现了IO多路复用功能。

本文将针对rtc::Thread类所提供消息循环，消息投递的功能进行介绍。由于Thread类是通过继承MessageQueue才具有此类功能，因此，在介绍Thread相关API实现之前应先介绍MessageQueue相关的知识：消息队里管理(MessageQueueManager)，消息队列(MessageQueue)，消息(Message，DelayedMessage)，消息数据(MessageData，TypedMessageData，ScopedMessageData，DisposeData)的相关知识。

Thread类在rtc_base/thread.h中声明，定义在rtc_base/thread.c中（只保留了消息循环以及消息投递相关的API）：

class RTC_LOCKABLE Thread : public MessageQueue {
 public:
  virtual void Run();

  virtual void Send(const Location& posted_from,
                    MessageHandler* phandler,
                    uint32_t id = 0,
                    MessageData* pdata = nullptr);

  template <class ReturnT, class FunctorT>
  ReturnT Invoke(const Location& posted_from, FunctorT&& functor) {
    FunctorMessageHandler<ReturnT, FunctorT> handler(
        std::forward<FunctorT>(functor));
    InvokeInternal(posted_from, &handler);
    return handler.MoveResult();
  }

  bool IsProcessingMessagesForTesting() override;
  void Clear(MessageHandler* phandler,
             uint32_t id = MQID_ANY,
             MessageList* removed = nullptr) override;
  void ReceiveSends() override;

  bool ProcessMessages(int cms);

 protected:
  friend class ScopedDisallowBlockingCalls;

 private:
  void ReceiveSendsFromThread(const Thread* source);
  bool PopSendMessageFromThread(const Thread* source, _SendMessage* msg);
  void InvokeInternal(const Location& posted_from, MessageHandler* handler);

  std::list<_SendMessage> sendlist_;
  bool blocking_calls_allowed_ = true;

  friend class ThreadManager;

  RTC_DISALLOW_COPY_AND_ASSIGN(Thread);
};

消息循环的建立

由上一篇文章WebRTC源码分析-线程基础之线程基本功能的线程启动分析知道，用户在没有传入自己的Runnable对象时，新的线程上会执行Thread.Run()方法，该方法源码如下，内部会调用ProcessMessages(kForever)去运行消息循环。而用户如果实现了自己的Runnable对象时，也想要运行消息循环，那咋办嘛？ 下面的注释就告知你了，在Runnable.Run()中适时的调用ProcessMessages()方法就行。

  // By default, Thread::Run() calls ProcessMessages(kForever).  To do other
  // work, override Run().  To receive and dispatch messages, call
  // ProcessMessages occasionally.
  void Thread::Run() {
      ProcessMessages(kForever);
  }

下面看看这个ProcessMessages()怎么建立消息循环的。分两种情形：
1）默认地，ProcessMessages(kForever)，告知无限期进行处理。此时的情形就是函数内部的while循环不停的调用Get()去获取消息，然后处理消息Dispatch()。循环能够退出的条件就是Get方法返回false。由
WebRTC源码分析-线程基础之MessageQueue 分析Get()方法分析可知，无限期处理的情况下，只有循环停止工作或者IO处理出错才会导致Get()返回false。
2）如果ProcessMessages(int cmsLoop)，有限期进行处理。那么退出循环的方式有两个，一个是使用时间已经到了，返回true；另外一个是Get()方法返回false，有限期处理的情况下，Get()返回false的条件有三：循环停止工作；IO处理出错；已经耗完所有处理时间也还未找到一个MSG。

bool Thread::ProcessMessages(int cmsLoop) {
  // Using ProcessMessages with a custom clock for testing and a time greater
  // than 0 doesn't work, since it's not guaranteed to advance the custom
  // clock's time, and may get stuck in an infinite loop.
  RTC_DCHECK(GetClockForTesting() == nullptr || cmsLoop == 0 ||
             cmsLoop == kForever);
  // 计算终止处理消息的时间
  int64_t msEnd = (kForever == cmsLoop) ? 0 : TimeAfter(cmsLoop);
  // 下次可以进行消息获取的时间长度
  int cmsNext = cmsLoop;

  while (true) {
#if defined(WEBRTC_MAC)
    ScopedAutoReleasePool pool;
#endif
    // 获取消息
    Message msg;
    if (!Get(&msg, cmsNext))
      return !IsQuitting();
    // 处理消息
    Dispatch(&msg);
    // 若不是无限期，计算下次可以进行消息获取的时间。
    if (cmsLoop != kForever) {
      cmsNext = static_cast<int>(TimeUntil(msEnd));
      // 若使用时间已经到了，那么退出循环
      if (cmsNext < 0)
        return true;
    }
  }
}

Post消息

向一个线程Post消息，只是简单地向消息循环的队列中插入一条待处理的消息，然后Post方法就会返回，不会引发当前线程的阻塞。Thread方法并没有重写MQ的Post方法，因此，关于Post方法的细节分析见 WebRTC源码分析-线程基础之MessageQueue

Send消息

向一个线程Send消息，会阻塞当前线程的运行，直到该消息被目标线程消费完后才会解除阻塞，从Send方法返回。算法流程如源码及其注释如下（分9个步骤）：

void Thread::Send(const Location& posted_from,
                  MessageHandler* phandler,
                  uint32_t id,
                  MessageData* pdata) {
  // 目标线程的消息循环是否还在处理消息？                                        // 步骤1
  if (IsQuitting())
    return;
  // 创建需要处理的消息                                                                       // 步骤2
  Message msg;
  msg.posted_from = posted_from;
  msg.phandler = phandler;
  msg.message_id = id;
  msg.pdata = pdata;
  // 若目标线程就是自己，那么直接在此处处理完消息就ok                // 步骤3
  if (IsCurrent()) {
    phandler->OnMessage(&msg);
    return;
  }

  // 断言当前线程是否具有阻塞权限，无阻塞权限，                          // 步骤4
  // 那么向别的线程Send消息就是个非法操作
  AssertBlockingIsAllowedOnCurrentThread();

  // 确保当前线程有一个Thread对象与之绑定                                     // 步骤5
  AutoThread thread;
  Thread* current_thread = Thread::Current();
  RTC_DCHECK(current_thread != nullptr);  // AutoThread ensures this

  // 创建一个SendMessage对象，放置到目标线程对象的sendlist_              // 步骤6
  // ready表征该消息是否已经处理完。
  bool ready = false;
  {
    CritScope cs(&crit_);
    _SendMessage smsg;
    smsg.thread = current_thread;
    smsg.msg = msg;
    smsg.ready = &ready;
    sendlist_.push_back(smsg);
  }

  // 将目标线程从IO处理中唤醒，赶紧处理消息啦~                                      // 步骤7
  // 目标线程将在其消息循环中，调用ReceiveSends()处理Send消息~~
  WakeUpSocketServer();

  // 同步等待消息被处理                                                                              // 步骤8
  bool waited = false;
  crit_.Enter();
  while (!ready) {
    crit_.Leave();
    // 对方也可能向我Send了消息，可不能都互相阻塞住了
    // 处理对方可能Send给我的消息。
    current_thread->ReceiveSendsFromThread(this);
    // 处理完对方的Send消息后，阻塞等待对方处理完我Send的消息，然后来唤醒我吧
    // 但这儿会有个意外，这就是waited存在的意义了
    current_thread->socketserver()->Wait(kForever, false);
    waited = true;
    crit_.Enter();
  }
  crit_.Leave();

  // 如果出现过waited，那么再唤醒一次当前线程去处理Post消息。       // 步骤9
  if (waited) {
    current_thread->socketserver()->WakeUp();
  }
}

要理解上述算法，需要搞清楚Send方法的代码是在当前线程执行的，而调用的是目标线程对象Thread的Send方法，即Send方法中的this，是目标线程线程对象Thread。捋清楚这点非常重要！！！这儿我一步步分析上述算法过程：

  1. 判断目标线程的消息循环是否仍在工作：IsQuitting()是目标线程对象Thread的方法，但是是在当前线程中执行的！若消息循环停止工作，那么会拒绝处理消息，Send会直接返回，但是调用方是无法获知的。一般建议是在向线程发送消息之前调用IsProcessingMessagesForTesting()判断下该消息循环是否还在正常运行。
  2. 创建需要消费的消息对象：此处没有什么可以多说的
  3. 判断目标线程是否就是当前线程： 通过Thread.IsCurrent()可以判别这点，如果目标线程就是当前线程，那就是自己给自己Send消息了，直接在此处消费消息并返回。
  4. 断言当前线程是否允许阻塞： 注意，这儿不是断言目标线程。因为，向另外一个线程Send消息时，当前线程需要阻塞地等待目标线程处理完消息后才返回。如果，当前线程没有阻塞权限的话，那就是非法操作了。
  5. 确保当前线程有一个关联的Thread对象： 为什么？因为后续的阻塞唤醒操作都要通过Thread对象的方法来实现，如果当前线程没有关联Thread对象，那么这些操作就无法完成。怎么做？通过创建一个局部对象AutoThread thread来实现。源码如下，注意两点：
    1）只有当当前线程无Thread关联时，才会将AutoThread作为当前线程的关联Thread；
    2）由于AutoThread thread是局部对象，当Send函数结束时该对象生命周期走到尾声，可以利用其析构函数中需要恢复当前对象无Thread对象绑定的状态(当然，前提是之前就无Thread对象关联)。
AutoThread::AutoThread()
    : Thread(SocketServer::CreateDefault(), /*do_init=*/false) {
  DoInit();
  if (!ThreadManager::Instance()->CurrentThread()) {
    ThreadManager::Instance()->SetCurrentThread(this);
  }
}

AutoThread::~AutoThread() {
  Stop();
  DoDestroy();
  if (ThreadManager::Instance()->CurrentThread() == this) {
    ThreadManager::Instance()->SetCurrentThread(nullptr);
  }
}
  1. 创建_SendMessage实例，并入队： _ SendMessage结构体的声明如下，该对象被创建后，会进入目标线程的sendlist_。其中Thread* thread存储的是主动投放消息的当前线程。当目标线程在消费msg之后，会将ready标志置为true，并且通过thread->WakeUp()解除当前线程的阻塞，从而判断ready后获知Send消息已经被消费了。
struct _SendMessage {
  _SendMessage() {}
  Thread* thread;   // 当前线程对象
  Message msg;     // 消息
  bool* ready;         // 消息是否处理完毕的标志
};
  1. 唤醒目标线程处理消息： 当_ SendMessage消息已经进入目标线程的sendlist_队列了，当然是要唤醒目标线程去处理啦，MQ的WakeUpSocketServer()就干这个事。好像很简单？这里代码没有体现的一点是：目标线程是如何处理这个_ SendMessage消息的，WebRTC源码分析-线程基础之MessageQueue 提到过 MQ的Get()方法最开始就是优先地，阻塞地调用它的ReceiveSends()处理Send消息，可惜的是MQ的该方法是个vitural方法，并且啥也没干，刚好Thread对象就重写了ReceiveSends()方法，这正是处理Send消息最佳之处。由于ReceiveSends()是在目标线程干的事，为了不打乱节奏，此处不展开描述ReceiveSends()。
  2. 同步等待Send消息被处理：虽然代码没几行，但是整个过程中最难理解的地方，慢慢展开来说，把这块儿的代码再次贴出来。
  bool waited = false;
  crit_.Enter();
  while (!ready) {
    crit_.Leave();
    current_thread->ReceiveSendsFromThread(this);
    current_thread->socketserver()->Wait(kForever, false);
    waited = true;
    crit_.Enter();
  }
  crit_.Leave();

1）ready这个参数会被当前线程访问，也会被目标线程访问，必然需要加锁，并且要达到两个线程的互斥，还必须要使用同一个锁。而这个锁就是目标线程的线程对象Thread的CriticalSection crit_成员。因此，while循环开头读取ready时进行了加锁解锁操作。
2）按理说，接下来当前线程阻塞等待目标线程完成操作之后通知我解除阻塞就行了。但是，考虑到一点，如果，两个线程同时互相Send消息，那岂不是二者都卡在等待对方通知这个地方，死锁了？为了避免这个情况，在阻塞等待前，先处理所有别人Send给我的消息吧。调用current_thread->ReceiveSendsFromThread(this)进行处理，注意current_thread是当前线程对象，因而是处理的当前线程所接收到的Send消息，而this是目标线程对象。 需要注意一点PopSendMessageFromThread()方法只会把source线程发送的消息从队列中取出，此时source为this，即目标线程对象。所以，此处ReceiveSendsFromThread()只会处理目标线程Send给当前线程的消息，此时，ready置为true，并调用目标线程的WakeUp()方法，唤醒了目标线程，使得目标线程不会阻塞在Send方法中，从而使得目标线程有机会去运行其消息循环，从而消费当前线程Send给它的消息。很绕很绕，我已经尽力了。

void Thread::ReceiveSendsFromThread(const Thread* source) {
  _SendMessage smsg;
  crit_.Enter();
  while (PopSendMessageFromThread(source, &smsg)) {
    crit_.Leave();
    Dispatch(&smsg.msg);
    crit_.Enter();
    *smsg.ready = true;
    smsg.thread->socketserver()->WakeUp();
  }
  crit_.Leave();
}

bool Thread::PopSendMessageFromThread(const Thread* source, _SendMessage* msg) {
  for (std::list<_SendMessage>::iterator it = sendlist_.begin();
       it != sendlist_.end(); ++it) {
    if (it->thread == source || source == nullptr) {
      *msg = *it;
      sendlist_.erase(it);
      return true;
    }
  }
  return false;
}

3）处理完目标线程可能Send给我的消息之后，我终于可以安心地阻塞在IO上等待了，current_thread->socketserver()->Wait(kForever, false)。这时，我们看看目标线程如何操作。之前在步骤7中，已经阐述过，目标线程被唤醒后在消息循环中优先阻塞地调用ReceiveSends()方法来处理Send消息，而且ReceiveSends()是被Thread重写过的方法。代码如下：咦，这不是上面调用的ReceiveSendsFromThread()方法嘛？不过传入的指针为空，此时PopSendMessageFromThread()会将目标线程上收到的所有Send消息都拿出来消费完，设置标志位，然后调用消息发送者线程的WakeUp()来唤醒消息发送者。到此处，终于把这个Send逻辑搞清楚了。完了？还没完呢，唤醒该线程的一定就是目标线程处理完Send消息之后嘛？还有这个waited干嘛用的啊？

void Thread::ReceiveSends() {
  ReceiveSendsFromThread(nullptr);
}

4）当前线程沉睡在current_thread->socketserver()->Wait(kForever, false)上时，唤醒该线程的一定就是目标线程处理完Send消息之后嘛？那可不一定，可能是别的线程Wake了它，也可能是目标线程Post了一条消息给当前线程(此时会唤醒当前线程，详见Post消息)。那么意味着当前线程Send消息可能还未处理完，自己就被醒了，那么又得进行While中的ready变量的访问了，又得加锁，解锁，把这个过程再走一遍。
5）这个waited变量干嘛用？就代码而言，我们发现waited变量进入了一次while循环就会变成true，表示我等待过至少一次。没有等待过的原因就是目标线程干活很麻利，在第一次判断ready值的时候就已经把活干完了，Send消息已被消费。具体记录这个等待过一次是做什么用呢？看步骤9

  1. 再唤醒一次 如果进入过一次循环等待，那么waited变量为true，需要再次唤醒当前线程一次current_thread->socketserver()->WakeUp()，让当前线程能处理消息。其实原因在源码的注释上写得很明白了，我就不翻译了。完毕。
  // Our Wait loop above may have consumed some WakeUp events for this
  // MessageQueue, that weren't relevant to this Send.  Losing these WakeUps can
  // cause problems for some SocketServers.
  //
  // Concrete example:
  // Win32SocketServer on thread A calls Send on thread B.  While processing the
  // message, thread B Posts a message to A.  We consume the wakeup for that
  // Post while waiting for the Send to complete, which means that when we exit
  // this loop, we need to issue another WakeUp, or else the Posted message
  // won't be processed in a timely manner.
  if (waited) {
    current_thread->socketserver()->WakeUp();
  }

Invoke跨线程同步执行方法

Invoke方法提供了一个方便的方式：阻塞当前线程，在另外一个线程上同步执行方法，并且返回执行结果。
本质上就是将需要执行的方法封装到消息处理器FunctorMessageHandler中，然后向目标线程Send这个携带特殊消息处理器FunctorMessageHandler的消息，该消息被消费后，阻塞结束，FunctorMessageHandler对象携带了方法执行的结果，当前线程可以从中获取到执行结果。其实，这里的重点有二：

  template <class ReturnT, class FunctorT>
  ReturnT Invoke(const Location& posted_from, FunctorT&& functor) {
    FunctorMessageHandler<ReturnT, FunctorT> handler(
        std::forward<FunctorT>(functor));
    InvokeInternal(posted_from, &handler);
    return handler.MoveResult();
  }

void Thread::InvokeInternal(const Location& posted_from,
                            MessageHandler* handler) {
  TRACE_EVENT2("webrtc", "Thread::Invoke", "src_file_and_line",
               posted_from.file_and_line(), "src_func",
               posted_from.function_name());
  Send(posted_from, handler);
}

总结

本文比较详细的介绍了Thread的消息循环，线程间Post消息，Send消息，跨线程执行方法等功能。由于Thread是通过继承MessageQueue才具有这些功能，因此，需要结合另外3篇文章来一起看。

本文对Post消息（非阻塞）只是一笔带过，因为在Thread并没有改写Post方法，而直接是MessageQueue的Post。

本文对Send消息（阻塞）重点拆解了一番，自己倒是理解了，但不知道是否描述的够清晰，是否有错误。博客的作用嘛，就是为了能够理清楚自己的思路，顺便也为他人做些贡献。如果有不对的地方，看到此处的同路人可以指出

本文还对跨线程执行方法提供了一种便捷的方式，本质就是封装一个特殊的消息处理器FunctorMessageHandler到消息中，并使用Send方法使得目标线程消费消息时执行该方法。

最后编辑于 ：
©著作权归作者所有,转载或内容合作请联系作者
  • 人面猴
    序言：七十年代末，一起剥皮案震惊了整个滨河市，随后出现的几起案子，更是在滨河造成了极大的恐慌，老刑警刘岩，带你破解...
    沈念sama阅读 211,948评论 6赞 492
  • 死咒
    序言：滨河连续发生了三起死亡事件，死亡现场离奇诡异，居然都是意外死亡，警方通过查阅死者的电脑和手机，发现死者居然都...
    沈念sama阅读 90,371评论 3赞 385
  • 救了他两次的神仙让他今天三更去死
    文/潘晓璐 我一进店门，熙熙楼的掌柜王于贵愁眉苦脸地迎上来，“玉大人，你说我怎么就摊上这事。” “怎么了？”我有些...
    开封第一讲书人阅读 157,490评论 0赞 348
  • 道士缉凶录：失踪的卖姜人
    文/不坏的土叔 我叫张陵，是天一观的道长。 经常有香客问我，道长，这世上最难降的妖魔是什么？ 我笑而不...
    开封第一讲书人阅读 56,521评论 1赞 284
  • 港岛之恋（遗憾婚礼）
    正文 为了忘掉前任，我火速办了婚礼，结果婚礼上，老公的妹妹穿的比我还像新娘。我一直安慰自己，他们只是感情好，可当我...
    茶点故事阅读 65,627评论 6赞 386
  • 恶毒庶女顶嫁案：这布局不是一般人想出来的
    文/花漫 我一把揭开白布。 她就那样静静地躺着，像睡着了一般。 火红的嫁衣衬着肌肤如雪。 梳的纹丝不乱的头发上，一...
    开封第一讲书人阅读 49,842评论 1赞 290
  • 城市分裂传说
    那天，我揣着相机与录音，去河边找鬼。 笑死，一个胖子当着我的面吹牛，可吹牛的内容都是我干的。 我是一名探鬼主播，决...
    沈念sama阅读 38,997评论 3赞 408
  • 双鸳鸯连环套：你想象不到人心有多黑
    文/苍兰香墨 我猛地睁开眼，长吁一口气：“原来是场噩梦啊……” “哼！你这毒妇竟也来了？” 一声冷哼从身侧响起，我...
    开封第一讲书人阅读 37,741评论 0赞 268
  • 万荣杀人案实录
    序言：老挝万荣一对情侣失踪，失踪者是张志新（化名）和其女友刘颖，没想到半个月后，有当地人在树林里发现了一具尸体，经...
    沈念sama阅读 44,203评论 1赞 303
  • 护林员之死
    正文 独居荒郊野岭守林人离奇死亡，尸身上长有42处带血的脓包…… 初始之章·张勋 以下内容为张勋视角 年9月15日...
    茶点故事阅读 36,534评论 2赞 327
  • 白月光启示录
    正文 我和宋清朗相恋三年，在试婚纱的时候发现自己被绿了。 大学时的朋友给我发了我未婚夫和他白月光在一起吃饭的照片。...
    茶点故事阅读 38,673评论 1赞 341
  • 活死人
    序言：一个原本活蹦乱跳的男人离奇死亡，死状恐怖，灵堂内的尸体忽然破棺而出，到底是诈尸还是另有隐情，我是刑警宁泽，带...
    沈念sama阅读 34,339评论 4赞 330
  • 日本核电站爆炸内幕
    正文 年R本政府宣布，位于F岛的核电站，受9级特大地震影响，放射性物质发生泄漏。R本人自食恶果不足惜，却给世界环境...
    茶点故事阅读 39,955评论 3赞 313
  • 男人毒药：我在死后第九天来索命
    文/蒙蒙 一、第九天 我趴在偏房一处隐蔽的房顶上张望。 院中可真热闹，春花似锦、人声如沸。这庄子的主人今日做“春日...
    开封第一讲书人阅读 30,770评论 0赞 21
  • 一桩弑父案，背后竟有这般阴谋
    文/苍兰香墨 我抬头看了看天上的太阳。三九已至，却和暖如春，着一层夹袄步出监牢的瞬间，已是汗流浃背。 一阵脚步声响...
    开封第一讲书人阅读 32,000评论 1赞 266
  • 情欲美人皮
    我被黑心中介骗来泰国打工， 没想到刚下飞机就差点儿被人妖公主榨干…… 1. 我叫王不留，地道东北人。 一个月前我还...
    沈念sama阅读 46,394评论 2赞 360
  • 代替公主和亲
    正文 我出身青楼，却偏偏与公主长得像，于是被迫代替她去往敌国和亲。 传闻我的和亲对象是个残疾皇子，可洞房花烛夜当晚...
    茶点故事阅读 43,562评论 2赞 349