chromium源码阅读(4)-线程模型之IO线程

r囧r小猫 2022-08-03 05:19 291阅读 0赞

chromium的IO线程是典型的异步通讯模型，本文主要从udp的socket请求过程中分析IO线程的运作流程，讲述其实现的技巧和方法。  
1、平台适配  
对于不同的平台(android、linux、windows)，socket的平台各有差异，为了对应用层屏蔽平台差异，对外提供统一接口实现，chromium提供平台适配层。这种技巧在chromium中得到了普遍的应用。

#if defined(OS_WIN)
    #include "net/udp/udp_socket_win.h"
    #elif defined(OS_POSIX)
    #include "net/udp/udp_socket_libevent.h"
    #endif
    
    #if defined(OS_WIN)
    typedef UDPSocketWin UDPSocket;
    #elif defined(OS_POSIX)
    typedef UDPSocketLibevent UDPSocket;
    #endif

2、异步读写实现  
以前一直以为客户端的UDP流程：建立socket套接字—–sendto函数——close，在实际开发中发现，udp也可以调用connect方法，之后sendto可以不指定目的地址和端口直接发送数据。  
以udp socket异步读取数据为例，socket read函数直接调用系统socket的读函数，读取成功，直接同步返回；读取失败，如果是IO\_PENDING之外的错误，直接返回error。

int UDPSocketLibevent::Read(IOBuffer* buf,
                                int buf_len,
                                const CompletionCallback& callback)
    {
        return RecvFrom(buf, buf_len, NULL, callback);
    }
    
    int UDPSocketLibevent::RecvFrom(IOBuffer* buf,
                                    int buf_len,
                                    IPEndPoint* address,
                                    const CompletionCallback& callback)
    {
        int nread = InternalRecvFrom(buf, buf_len, address);
        if (nread != ERR_IO_PENDING)
            return nread;
    
        if (!base::MessageLoopForIO::current()->WatchFileDescriptor(
                    socket_, true, base::MessageLoopForIO::WATCH_READ,
                    &read_socket_watcher_, &read_watcher_))
        {
            int result = MapSystemError(errno);
            return result;
        }
    
        read_buf_ = buf;
        read_buf_len_ = buf_len;
        recv_from_address_ = address;
        read_callback_ = callback;
    
        return ERR_IO_PENDING;
    }

然后关键点来了，如果当前socket处于pending状态，直接把阻塞的socket调用注册给IO thread。  
base::MessageLoopForIO::current()->WatchFileDescriptor(  
socket\_, true, base::MessageLoopForIO::WATCH\_READ,  
&read\_socket\_watcher\_, &read\_watcher\_)  
WatchFileDescriptor又是如何实现的呢？

bool MessagePumpLibevent::WatchFileDescriptor(int fd,
                                                  bool persistent,
                                                  int mode,
                                                  FileDescriptorWatcher *controller,
                                                  Watcher *delegate) {
    // DCHECK_GE(fd, 0);
     // DCHECK(controller);
     // DCHECK(delegate);
     // DCHECK(mode == WATCH_READ || mode == WATCH_WRITE || mode == WATCH_READ_WRITE);
      // WatchFileDescriptor should be called on the pump thread. It is not
      // threadsafe, and your watcher may never be registered.
      //DCHECK(watch_file_descriptor_caller_checker_.CalledOnValidThread());
    
      int event_mask = persistent ? EV_PERSIST : 0;
      if (mode & WATCH_READ) {
        event_mask |= EV_READ;
      }
      if (mode & WATCH_WRITE) {
        event_mask |= EV_WRITE;
      }
    
      scoped_ptr<event> evt(controller->ReleaseEvent());
      if (evt.get() == NULL) {
        // Ownership is transferred to the controller.
        evt.reset(new event);
      } else {
        // Make sure we don't pick up any funky internal libevent masks.
        int old_interest_mask = evt.get()->ev_events &
            (EV_READ | EV_WRITE | EV_PERSIST);
    
        // Combine old/new event masks.
        event_mask |= old_interest_mask;
    
        // Must disarm the event before we can reuse it.
        event_del(evt.get());
    
        // It's illegal to use this function to listen on 2 separate fds with the
        // same |controller|.
        if (EVENT_FD(evt.get()) != fd) {
    // NOTREACHED() << "FDs don't match" << EVENT_FD(evt.get()) << "!=" << fd;
          return false;
        }
      }
    
      // Set current interest mask and message pump for this event.
      event_set(evt.get(), fd, event_mask, OnLibeventNotification, controller);
    
      // Tell libevent which message pump this socket will belong to when we add it.
      if (event_base_set(event_base_, evt.get())) {
        return false;
      }
    
      // Add this socket to the list of monitored sockets.
      if (event_add(evt.get(), NULL)) {
        return false;
      }
    
      // Transfer ownership of evt to controller.
      controller->Init(evt.release());
    
      controller->set_watcher(delegate);
      controller->set_pump(this);
    
      return true;
    }

主要调用了libevent网络库  
event\_set(evt.get(), fd, event\_mask, OnLibeventNotification, controller)，当网络数据返回时，调用OnLibeventNotification通知调用者。

如何检测网络数据返回呢？技巧主要在IO线程主函数上

bool did_work = delegate->DoWork();
        if (!keep_running_)
          break;
    
        event_base_loop(event_base_, EVLOOP_NONBLOCK);
        did_work |= processed_io_events_;
        processed_io_events_ = false;
        if (!keep_running_)
          break;
    
        did_work |= delegate->DoDelayedWork(&delayed_work_time_);
        if (!keep_running_)
          break;
    
        if (did_work)
          continue;

IO线程主线程调用event\_base\_loop(event\_base\_, EVLOOP\_NONBLOCK)执行网络回调，如果网络回调成功，会将processed\_io\_events变量标识为true，此时会反复检测直至所有的网络事件处理完毕。  
回调函数的实现如下：

void MessagePumpLibevent::OnLibeventNotification(int fd, short flags,void* context) {
        ....
      MessagePumpLibevent* pump = controller->pump();
      pump->processed_io_events_ = true;
    
      if (flags & EV_WRITE) {
        controller->OnFileCanWriteWithoutBlocking(fd, pump);
      }
      if (controller.get() && flags & EV_READ) {
        controller->OnFileCanReadWithoutBlocking(fd, pump);
      }
    }