前言

image.png

webrtc下事件的发生

image.png

等待线程处理逻辑

image.png
image.png
内层循环，如果没有消息，则Wait，如果有则获取第一个，返回处理。

PhysicalSocketServer的Wait函数

image.png

源码分析

网络线程，PhysicalSocketServer，和事件
工作线程，使用的是NullSocketServer

H:\webrtc-20210315\webrtc-20210315\webrtc\webrtc-checkout\src\rtc_base\physical_socket_server.cc
PhysicalSocketServer::Wait

  1. bool PhysicalSocketServer::Wait(int cmsWait, bool process_io) {
  2.   // We don't support reentrant waiting.
  3.   RTC_DCHECK(!waiting_);
  4.   ScopedSetTrue s(&waiting_);
  6.   int64_t cmsTotal = cmsWait;
  7.   int64_t cmsElapsed = 0;
  8.   int64_t msStart = Time();
  10.   fWait_ = true;
  11.   while (fWait_) {
  12.     std::vector<WSAEVENT> events;
  13.     std::vector<uint64_t> event_owners;
  15.     // 将要监听的事件添加到数组中
  16.     events.push_back(socket_ev_);
  18.     {
  19.       CritScope cr(&crit_);
  20.       // Get a snapshot of all current dispatchers; this is used to avoid the
  21.       // ABA problem (see later comment) and avoids the dispatcher_by_key_
  22.       // iterator being invalidated by calling CheckSignalClose, which may
  23.       // remove the dispatcher from the list.
  24.       current_dispatcher_keys_.clear();
  25.       for (auto const& kv : dispatcher_by_key_) {
  26.         current_dispatcher_keys_.push_back(kv.first);
  27.       }
  28.       for (uint64_t key : current_dispatcher_keys_) {
  29.         if (!dispatcher_by_key_.count(key)) {
  30.           continue;
  31.         }
  32.         Dispatcher* disp = dispatcher_by_key_.at(key);
  33.         if (!disp)
  34.           continue;
  35.         if (!process_io && (disp != signal_wakeup_))
  36.           continue;
  37.         // 获取socket
  38.         SOCKET s = disp->GetSocket();
  39.         if (disp->CheckSignalClose()) {
  40.           // We just signalled close, don't poll this socket.
  41.         } else if (s != INVALID_SOCKET) { // 将socket和事件绑定，当事件发生时可以知道哪个socket触发的事件
  42.           WSAEventSelect(s, events[0],
  43.                          FlagsToEvents(disp->GetRequestedEvents()));
  44.         } else {  // 普通事件的socket
  45.           events.push_back(disp->GetWSAEvent());
  46.           event_owners.push_back(key);
  47.         }
  48.       }
  49.     }
  51.     // Which is shorter, the delay wait or the asked wait?
  53.     int64_t cmsNext;
  54.     if (cmsWait == kForever) {
  55.       cmsNext = cmsWait;
  56.     } else {
  57.       cmsNext = std::max<int64_t>(0, cmsTotal - cmsElapsed);
  58.     }
  60.     // Wait for one of the events to signal
  61.     // 开始事件等待
  62.     DWORD dw =
  63.         WSAWaitForMultipleEvents(static_cast<DWORD>(events.size()), &events[0],
  64.                                  false, static_cast<DWORD>(cmsNext), false);
  66.     // 事件发生后，开始判断并处理
  67.     if (dw == WSA_WAIT_FAILED) {
  68.       // Failed?
  69.       // TODO(pthatcher): need a better strategy than this!
  70.       WSAGetLastError();
  71.       RTC_NOTREACHED();
  72.       return false;
  73.     } else if (dw == WSA_WAIT_TIMEOUT) {
  74.       // Timeout?
  75.       return true;
  76.     } else {
  77.       // Figure out which one it is and call it
  78.       CritScope cr(&crit_);
  79.       int index = dw - WSA_WAIT_EVENT_0;
  80.       if (index > 0) {
  81.         --index;  // The first event is the socket event
  82.         uint64_t key = event_owners[index];
  83.         if (!dispatcher_by_key_.count(key)) {
  84.           // The dispatcher could have been removed while waiting for events.
  85.           continue;
  86.         }
  87.         Dispatcher* disp = dispatcher_by_key_.at(key);
  88.         disp->OnEvent(0, 0);
  89.       } else if (process_io) {
  90.         // Iterate only on the dispatchers whose sockets were passed into
  91.         // WSAEventSelect; this avoids the ABA problem (a socket being
  92.         // destroyed and a new one created with the same SOCKET handle).
  93.         for (uint64_t key : current_dispatcher_keys_) {
  94.           if (!dispatcher_by_key_.count(key)) {
  95.             continue;
  96.           }
  97.           Dispatcher* disp = dispatcher_by_key_.at(key);
  98.           SOCKET s = disp->GetSocket();
  99.           if (s == INVALID_SOCKET)
  100.             continue;
  102.           WSANETWORKEVENTS wsaEvents;
  103.           int err = WSAEnumNetworkEvents(s, events[0], &wsaEvents);
  104.           if (err == 0) {
  105.             {
  106.               if ((wsaEvents.lNetworkEvents & FD_READ) &&
  107.                   wsaEvents.iErrorCode[FD_READ_BIT] != 0) {
  108.                 RTC_LOG(WARNING)
  109.                     << "PhysicalSocketServer got FD_READ_BIT error "
  110.                     << wsaEvents.iErrorCode[FD_READ_BIT];
  111.               }
  112.               if ((wsaEvents.lNetworkEvents & FD_WRITE) &&
  113.                   wsaEvents.iErrorCode[FD_WRITE_BIT] != 0) {
  114.                 RTC_LOG(WARNING)
  115.                     << "PhysicalSocketServer got FD_WRITE_BIT error "
  116.                     << wsaEvents.iErrorCode[FD_WRITE_BIT];
  117.               }
  118.               if ((wsaEvents.lNetworkEvents & FD_CONNECT) &&
  119.                   wsaEvents.iErrorCode[FD_CONNECT_BIT] != 0) {
  120.                 RTC_LOG(WARNING)
  121.                     << "PhysicalSocketServer got FD_CONNECT_BIT error "
  122.                     << wsaEvents.iErrorCode[FD_CONNECT_BIT];
  123.               }
  124.               if ((wsaEvents.lNetworkEvents & FD_ACCEPT) &&
  125.                   wsaEvents.iErrorCode[FD_ACCEPT_BIT] != 0) {
  126.                 RTC_LOG(WARNING)
  127.                     << "PhysicalSocketServer got FD_ACCEPT_BIT error "
  128.                     << wsaEvents.iErrorCode[FD_ACCEPT_BIT];
  129.               }
  130.               if ((wsaEvents.lNetworkEvents & FD_CLOSE) &&
  131.                   wsaEvents.iErrorCode[FD_CLOSE_BIT] != 0) {
  132.                 RTC_LOG(WARNING)
  133.                     << "PhysicalSocketServer got FD_CLOSE_BIT error "
  134.                     << wsaEvents.iErrorCode[FD_CLOSE_BIT];
  135.               }
  136.             }
  137.             uint32_t ff = 0;
  138.             int errcode = 0;
  139.             if (wsaEvents.lNetworkEvents & FD_READ)
  140.               ff |= DE_READ;
  141.             if (wsaEvents.lNetworkEvents & FD_WRITE)
  142.               ff |= DE_WRITE;
  143.             if (wsaEvents.lNetworkEvents & FD_CONNECT) {
  144.               if (wsaEvents.iErrorCode[FD_CONNECT_BIT] == 0) {
  145.                 ff |= DE_CONNECT;
  146.               } else {
  147.                 ff |= DE_CLOSE;
  148.                 errcode = wsaEvents.iErrorCode[FD_CONNECT_BIT];
  149.               }
  150.             }
  151.             if (wsaEvents.lNetworkEvents & FD_ACCEPT)
  152.               ff |= DE_ACCEPT;
  153.             if (wsaEvents.lNetworkEvents & FD_CLOSE) {
  154.               ff |= DE_CLOSE;
  155.               errcode = wsaEvents.iErrorCode[FD_CLOSE_BIT];
  156.             }
  157.             if (ff != 0) {
  158.               disp->OnEvent(ff, errcode);
  159.             }
  160.           }
  161.         }
  162.       }
  164.       // Reset the network event until new activity occurs
  165.       WSAResetEvent(socket_ev_);
  166.     }
  168.     // Break?
  169.     if (!fWait_)
  170.       break;
  171.     cmsElapsed = TimeSince(msStart);
  172.     if ((cmsWait != kForever) && (cmsElapsed >= cmsWait)) {
  173.       break;
  174.     }
  175.   }
  177.   // Done
  178.   return true;
  179. }

触发事件

void Thread::WakeUpSocketServer() {
ss_->WakeUp();
}
image.png
image.png

向网络线程发送任务

image.png
这个事件，就是由Thread::WakeUpSocketServer触发的。

  1. void Thread::WakeUpSocketServer() {
  2.   ss_->WakeUp();
  3. }
  4. -》
  5. void PhysicalSocketServer::WakeUp() {
  6.   signal_wakeup_->Signal();
  7. }
  8. -》
  9.   virtual void Signal() {
  10.     if (hev_ != nullptr)
  11.       WSASetEvent(hev_);
  12.   }

事件触发后，又来执行
image.png