boost.asio源码阅读(2) - task_io_service

测试账号 2021-01-12 06:20 615阅读 0赞

## 1.0 task\_io\_service ##

在boost.asio源码阅读(1)中，代码已经查看到task\_io\_service中。  
具体的操作调用void task\_io\_service::init\_task()  
本文从init\_task出发，查看到底basic\_socket\_acceptor的创建对task\_io\_service产生了什么影响.

## 1.1 init\_task ##

直接上代码

void task_io_service::init_task()
    {
      mutex::scoped_lock lock(mutex_);
      if (!shutdown_ && !task_)
      {
        1. task_ = &use_service<reactor>(this->get_io_service());
        2. op_queue_.push(&task_operation_);
        3. wake_one_thread_and_unlock(lock);
      }
    }

1.  创建task\_，其实task\_就是epoll\_reactor.
2.  将成员task\_operation\_放入队列
3.  唤醒一个线程，并对当前操作解锁。  
    这里能够明确io\_service使用了同步队列op\_queue(lock->入队->唤醒->unlock)，在basic\_socket\_acceptor  
    创建时，将task\_operation\_放入队列，在io\_service->run()中，这个变量将会被操作.

## 1.2 task\_io\_service::run ##

直奔主题

std::size_t task_io_service::run(boost::system::error_code& ec)
    {
      // ......
      thread_info this_thread;
      this_thread.private_outstanding_work = 0;
      // ......
      for (; do_run_one(lock, this_thread, ec); lock.lock())
      //......
    }

thread\_info 内部含有一个队列，也就是说在多线程调用io\_service::run的时候  
每一个线程将会具有自己的私有队列。

std::size_t task_io_service::do_run_one
    {
      while (!stopped_)
      {
        if (!op_queue_.empty())
        {
          // Prepare to execute first handler from queue.
          operation* o = op_queue_.front();
          op_queue_.pop();
          bool more_handlers = (!op_queue_.empty());
    
          1. if (o == &task_operation_)
          {
            task_interrupted_ = more_handlers;
    
            if (more_handlers && !one_thread_)
              wakeup_event_.unlock_and_signal_one(lock);
            else
              lock.unlock();
    
            2. task_cleanup on_exit = { this, &lock, &this_thread };
            (void)on_exit;
            task_->run(!more_handlers, this_thread.private_op_queue);
          }
          else
          {
            std::size_t task_result = o->task_result_;
    
            if (more_handlers && !one_thread_)
              wake_one_thread_and_unlock(lock);
            else
              lock.unlock();
    
            // Ensure the count of outstanding work is decremented on block exit.
            4. work_cleanup on_exit = { this, &lock, &this_thread };
            (void)on_exit;
    
            // Complete the operation. May throw an exception. Deletes the object.
            o->complete(*this, ec, task_result);
    
            return 1;
          }
        }
        else
        {
          wakeup_event_.clear(lock);
          5. wakeup_event_.wait(lock);
        }
      }
    
      return 0;
    }

1.  如果当前操作的队列内容是task\_operation\_则最后会进入到task\_->run也就是epoll\_wait。  
    同时，如果当前队列不为空且是多线程的时候，会继续唤醒线程来执行操作，避免epoll\_wait阻赛带来的延迟（虽然可以打断)。  
    到这里可以知道basic\_socket\_acceptor的创建使io\_service->run的时候可以进入epoll\_wait来处理相关的事件，
2.  和line4相同，都是在定义一个变量后，执行对应的析构操作。同时将线程的私有队列的内容，添加到公有队列op\_queue\_中,最后将刚才出队的task\_operation\_重新加到队尾(task\_cleanup)，使io\_service在处理完队列之前的事件后，再次进入到task->run中。
3.  line5表示在队列为空时，线程进入睡眠等待唤醒。

## 2. 小结 ##

1.  在本文中，可以知道basic\_socket\_acceptor的创建使io\_service在循环事件时能够进入到epoll\_wait。  
    当然如果这时使用者将acceptor删除也不能改变io\_service->run本身的逻辑，也就是说,如果acceptor不在需要，io\_service要么停止或者重新开始，避免epoll\_wait上面不必要的效率损失。
2.  task\_io\_service本身有一个公共队列，同时每一个执行线程有自己的私有队列，在每次处理事件之前，会将  
    私有队列的内容先加入到公有队列中。在某种程度上，公有队列的内容要比私有队列的优先级高些。
3.  asio的所有事件，最后都会回到公有队列中等待被执行。