序

本篇文章带着大家来看下CAF（C++ Actor Framwwork）的调度框架，也是算现阶段C++比较成熟的调度框架，大家如果自己完成一个比较大型的项目，任务调度也可以参照CAF。

鉴于篇幅较长，大家如果学习使用如何使用CAF可以参照 http://purecpp.cn/detail?id=2237 内容，如果想要直接学习调度框架，可以直接跳转到任务调度小节。

关于CAF用法

简单说一下CAF的用法或者说一个调用模型，它里边比较关键的概念是Actor，翻译过来就是角色，每个Actor中有beavior（行为）这个概念，beavior里边就是这个角色会做的事情（也就是一个函数）。Actor是基本的计算单元和通信的单元，两个actor相互通信，actor执行自己的behavior。如图所示：

再来看下官方的使用例子，我删了下注释：

using namespace caf;

behavior mirror(event_based_actor* self) {
  return {
    [=](const std::string& what) -> std::string {
      aout(self) << what << std::endl;
      return std::string{what.rbegin(), what.rend()};
    },
  };
}

void hello_world(event_based_actor* self, const actor& buddy) {
  self->request(buddy, std::chrono::seconds(10), "Hello World!")
    .then(
      [=](const std::string& what) {
        aout(self) << what << std::endl;
      });
}

void caf_main(actor_system& sys) {
  // create a new actor that calls 'mirror()'
  auto mirror_actor = sys.spawn(mirror);
  // create another actor that calls 'hello_world(mirror_actor)';
  sys.spawn(hello_world, mirror_actor);
  // the system will wait until both actors are done before exiting the program
}

CAF_MAIN()

我这里帮着大家顺一下这段代码：

• 函数入口在CAF_MAIN宏中，主要是做一些初始化工作，和我们主题相关的是会创建actor_system对象并调用caf_main函数。
  
• 然后caf_main中调用sys.spawn会创建出来一个actor，我们看mirror_actor的behavior，是在mirror函数中return了一个behavior对象，也就是spawn会调用mirror函数并获得了behavior，然后创建了actor对象。
  
• 再下一个spawn函数调用hello_world，传入mirror_actor对象，对应于buddy参数，并执行request函数，request传入buddy参数，表示会调用buddy的behavior函数，根据函数参数去找，，request参数的第二个表示该函数的超时时间，第三个之后才是要调用buddy这个actor的behavior的函数参数，那么这里正好可以调用到mirror_actor的behavior，当behavior返回时会自动执行then这个函数。
  
• 最后整体的执行顺序就是，sys调用spawn创建拥有behavior的actor对象mirror_actor, 第二个spawn会创建出来第二个actor对象，并调用自己的request函数向另外一个actor发送调用信息，也即会调用buddy的bahavior，调用behavior后执行then函数，整体结束。

调度器启动（初始化）

首先调度器的构造或者说初始化是在actor_system的构造函数进行：

// actor_system.cpp

actor_system::actor_system(actor_system_config& cfg) {
  // ...

  auto& sched = modules_[module::scheduler];
  using namespace scheduler;
  using policy::work_sharing;
  using policy::work_stealing;
  using share = coordinator<work_sharing>;
  using steal = coordinator<work_stealing>;
  if (!sched) {
    enum sched_conf {
      stealing = 0x0001,
      sharing = 0x0002,
      testing = 0x0003,
    };
    sched_conf sc = stealing;
    namespace sr = defaults::scheduler;
    auto sr_policy = get_or(cfg, "caf.scheduler.policy", sr::policy);
    if (sr_policy == "sharing")
      sc = sharing;
    else if (sr_policy == "testing")
      sc = testing;
    else if (sr_policy != "stealing")
      std::cerr << "[WARNING] " << deep_to_string(sr_policy)
                << " is an unrecognized scheduler pollicy, "
                   "falling back to 'stealing' (i.e. work-stealing)"
                << std::endl;
    switch (sc) {
      default: // any invalid configuration falls back to work stealing
        sched.reset(new steal(*this));
        break;
      case sharing:
        sched.reset(new share(*this));
        break;
      case testing:
        sched.reset(new test_coordinator(*this));
    }
  }

  // ...
}

调度器的对象是从modules_这个数组中取出，我们看下module是什么：

// actor_system.hpp
class CAF_CORE_EXPORT actor_system {
  // ...

  class CAF_CORE_EXPORT module {
  public:
    enum id_t {
      scheduler,
      middleman,
      openssl_manager,
      network_manager,
      num_ids
    };

    virtual ~module();
    const char* name() const noexcept;
    virtual void start() = 0;
    virtual void stop() = 0;
    virtual void init(actor_system_config&) = 0;
    virtual id_t id() const = 0;
    virtual void* subtype_ptr() = 0;
  };

  // ...
};

module会提供一些公共接口，init，start，stop等等之类的。我们也看到调度器是属于module的。

然后再回到创建调度器那里，最开始调度器是没有的，然后可以看的出来有三种调度配置，steal（偷取任务），share（共享任务），test（测试模式），我们就忽略测试模式。截止到switch那里，我们知道从配置文件中读取到调度的配置是什么，就去创建相应的调度器。那么调度器的类就是coordinator，work_sharing和work_stealing分别是不同调度配置的实现类，coordinator传入模版参数为调度的策略类（work_sharing和work_stealing）来实现调度器的全部功能。

然后继续往下继续看，目前已经有了调度器的创建，我们大致看下实现调度器共的类大体继承关系：

// coordinator.hpp
template <class Policy>
class coordinator : public abstract_coordinator {
  // ...
};

// abstract_coordinator.hpp
class CAF_CORE_EXPORT abstract_coordinator : public actor_system::module {
  // ...
};

coordinator继承abstract_coordinator，abstract_coordinator然后继承actor_system::module，这就和我们上边说的调度器属于module联系上了。

然后我们继续回到actor_system继续往下走：

// actor_system.cpp
actor_system::actor_system(actor_system_config& cfg) {
  // ...

  for (auto& mod : modules_)
    if (mod)
      mod->init(cfg);

  // ...
}

这里我们看到会调用调度器实现类的init函数，init函数是由abstract_coordinator类实现：

// abstract_coordinator.cpp
void abstract_coordinator::init(actor_system_config& cfg) {
  namespace sr = defaults::scheduler;
  max_throughput_ = get_or(cfg, "caf.scheduler.max-throughput",
                           sr::max_throughput);
  num_workers_ = get_or(cfg, "caf.scheduler.max-threads",
                        default_thread_count());
}

这里实现比较简单，就是从配置获取最大吞吐量（max_throughput_）和worker的数量，worker的数量默认是线程个数。

继续看actor_system构造函数：

// actor_system.cpp

actor_system::actor_system(actor_system_config& cfg) {
  // ...

  for (auto& mod : modules_)
    if (mod)
      mod->start();

  // ...
}

这里就是会调用到coordinator的start函数：

// coordinator.hpp
template <class Policy>
class coordinator : public abstract_coordinator {
  // ...

using worker_type = worker<Policy>;
void start() override {
    // Create initial state for all workers.
    typename worker_type::policy_data init{this};
    // Prepare workers vector.
    auto num = num_workers();
    workers_.reserve(num);
    // Create worker instances.
    for (size_t i = 0; i < num; ++i)
      workers_.emplace_back(
        std::make_unique<worker_type>(i, this, init, max_throughput_));
    // Start all workers.
    for (auto& w : workers_)
      w->start();

    // ...
  }

  // ...
}

首先获取到worker个数，然后来构造worker放置到workers_的数据结构中。然后调用worker的start函数。我们进入到worker类中去看看顺便看下它的start函数：

// worker.hpp

template <class Policy>
class worker : public execution_unit {
public:
  using coordinator_ptr = coordinator<Policy>*;
  using policy_data = typename Policy::worker_data;
  worker(size_t worker_id, coordinator_ptr worker_parent,
         const policy_data& init, size_t throughput)
    : execution_unit(&worker_parent->system()),
      max_throughput_(throughput),
      id_(worker_id),
      parent_(worker_parent),
      data_(init) {
    // nop
  }

  void start() {
    CAF_ASSERT(this_thread_.get_id() == std::thread::id{});
    this_thread_ = system().launch_thread("caf.worker", thread_owner::scheduler,
                                          [this] { run(); });
  }

  // ...
};

首先构造函数是传入吞吐量，id，coordinator和policy_data对象。然后当调用start函数时其实就是开启了一个线程执行run函数，那也就是说每个worker其实就是开启一个线程来执行任务而已。再去看下run函数：

// worker.hpp

template <class Policy>
class worker : public execution_unit {
  // ...

private:
  void run() {
    // scheduling loop
    for (;;) {
      auto job = policy_.dequeue(this);
      policy_.before_resume(this, job);
      auto res = job->resume(this, max_throughput_);
      policy_.after_resume(this, job);

      switch (res) {
        case resumable::resume_later: {
          policy_.resume_job_later(this, job);
          break;
        }
        case resumable::done: {
          policy_.after_completion(this, job);
          intrusive_ptr_release(job);
          break;
        }
        case resumable::awaiting_message: {
          intrusive_ptr_release(job);
          break;
        }
        case resumable::shutdown_execution_unit: {
          policy_.after_completion(this, job);
          policy_.before_shutdown(this);
          return;
        }
      }
    }
  }

  Policy policy_;

  // ...
};

我省略了下代码，后边还会回来看。run函数主要就是做几件事，从Policy中获取一个job，执行job（job->resume）, 然后判断结果做出不同的行为，这里要说一下的是当执行完job，返回结果时resumable::resume_later时，会继续将这个job丢入到队列中，之后某个时刻会再次调用到这个job

以上就是整个调度器的初始化过程，首先actor_system的构造函数中会根据不同调度策略，创建调度器类（coordinator<Policy>），并调用coordinator的init函数和start函数。其中start函数会构造worker并调用worker的start函数。一个worker<Policy>在start时会开一个线程调用run函数，run函数负责执行被调度过来的任务。

actor的创建

我们的例子中CAF创建actor通过actor_system的spawn函数来创建一个actor，简单看下创建的actor的过程：

// actor_system.hpp
infer_handle_from_fun_t<F> spawn(F fun, Ts&&... xs) {
  using impl = infer_impl_from_fun_t<F>;
  check_invariants<impl>();
  static constexpr bool spawnable = detail::spawnable<F, impl, Ts...>();
  static_assert(spawnable,
                "cannot spawn function-based actor with given arguments");
  actor_config cfg;
  return spawn_functor<Os>(detail::bool_token<spawnable>{}, cfg, fun,
                          std::forward<Ts>(xs)...);
}

infer_handle_from_fun_t这个我们跟踪进去会发现是actor这个类型，函数内容的第一行的impl是event_based_actor，spawn_functor函数就会创建actor并把相应的的behavior保存下来。然后太多这里先不管关注了，这里大家需要理解的是有两个点，第一个我们这里会创建一个event_based_actor对象，然后使用actor对象来接收并返回，那也就是说CAF的actor中使用actor_control_block对象来持有创建出来的对象，因为创建出来有可能是别的actor类型，那看下所有类型的actor的类图：

第二个点是event_based_actor创建behavior是会调用到event_based_actor的make_behavior函数：

// event_based_actor.cpp
void event_based_actor::initialize() {
  // ...
  auto bhvr = make_behavior();
  if (bhvr) {
    become(std::move(bhvr));
  }
}


behavior event_based_actor::make_behavior() {
  behavior res;
  if (initial_behavior_fac_) {
    res = initial_behavior_fac_(this);
    initial_behavior_fac_ = nullptr;
  }
  return res;
}

后调用become函数，最终会调用到他的父类scheduled_actor的do_become函数；

// scheduled_actor.cpp
void scheduled_actor::do_become(behavior bhvr, bool discard_old) {
  // ...
  if (bhvr)
    bhvr_stack_.push_back(std::move(bhvr));
  // ...
}

这里可以看到会讲behavior放置到scheduled_actor的bhvr_stack_这个成员变量里。

任务调度

创建一个任务

我们上边也看到event_based_actor是继承scheduled_actor这个类，但是实际的继承看起来还有有点绕的：

// event_based_actor.hpp
class CAF_CORE_EXPORT event_based_actor
  : public extend<scheduled_actor, event_based_actor>::
           with<mixin::sender,
                mixin::requester,
                mixin::subscriber,
                mixin::behavior_changer> {

                  // ...
};

这里看到实际上event_based_actor继承extend的with类型，最后大家会发现实际是继承mixin::sender，mixin::requester，mixin::subscriber，mixin::behavior_changer，这四个类型，这四个类型又继承scheduled_actor，这段代码还是有点意思，大家可以去实际去看下。

然后我们回到我们最初的例子，会通过event_based_actor的request函数调用到另外一个actor的behavior里。正好我们request函数是在继承的mixin::requester类里：

// requester.hpp
auto request(const Handle& dest, timeout, Ts&&... xs) {
  // ...
  auto self = static_cast<Subtype*>(this);
  auto req_id = self->new_request_id(P);
  auto pending_msg = disposable{};
  if (dest) {
    detail::profiled_send(self, self->ctrl(), dest, req_id, {},
                          self->context(), std::forward<Ts>(xs)...);
    pending_msg = self->request_response_timeout(timeout, req_id);
  } else {
    //...
  }
  using response_type
    = response_type_t<typename Handle::signatures,
                      detail::implicit_conversions_t<detail::decay_t<Ts>>...>;
  using handle_type
    = response_handle<Subtype, policy::single_response<response_type>>;
  return handle_type{self, req_id.response_id(), std::move(pending_msg)};
}

生成一个req_id，通过profiled_send发送消息到另外一个actor，并返回response_type。我们后边会调用then函数，也即调用response_type的then，最终会调用到scheduled_actor的add_multiplexed_response_handler函数中：

// scheduled_actor.cpp
void scheduled_actor::add_multiplexed_response_handler(message_id response_id, behavior bhvr) {
  if (bhvr.timeout() != infinite)
    request_response_timeout(bhvr.timeout(), response_id);
  multiplexed_responses_.emplace(response_id, std::move(bhvr));
}

可以看到会把then后边的函数存放在multiplexed_responses_，与response_id相对应。那这里大家应该可以猜到，是发送消息的actor收到回应就会调用到这个函数。


让我们回过头看下profiled_send函数会做什么：

// profiled_send.hpp
void profiled_send(Self* self, SelfHandle&& src, const Handle& dst,
                   message_id msg_id, std::vector<strong_actor_ptr> stages,
                   execution_unit* context, Ts&&... xs) {
  if (dst) {
    auto element = make_mailbox_element(std::forward<SelfHandle>(src), msg_id, std::move(stages), std::forward<Ts>(xs)...);
    dst->enqueue(std::move(element), context);
  } else {
    // ...
  }
}

首先通过make_mailbox_element构造一个message，然后就是直接放到dst的enqueue函数，也把消息放置在目的actor的队列里。

同样这个函数依然是调用到scheduled_actor的enqueue函数。

// scheduled_actor.cpp
bool scheduled_actor::enqueue(mailbox_element_ptr ptr, execution_unit* eu) {
  // ...
  auto mid = ptr->mid;
  auto sender = ptr->sender;
  // ...
  switch (mailbox().push_back(std::move(ptr))) {
    case intrusive::inbox_result::unblocked_reader: {
      // ...
      if (private_thread_)
        private_thread_->resume(this);
      else if (eu != nullptr)
        eu->exec_later(this);
      else
        // ...
      return true;
    }
    case intrusive::inbox_result::success:
      return true;
    default: { // intrusive::inbox_result::queue_closed
      // ...
      return false;
    }
  }
}

这里稍微有点复杂，首先就是需要把message放到scheduled_actor的信箱里，这个信箱也是scheduled_actor的成员，然后会返回结果。

需要比对下返回结果，unblocked_reader的意思是actor未被调度执行behavior，最开始actor没有任务执行那肯定是这样的。success就是表明actor现在正在执行其他的behavior然后直接返回true就可以了。

那为什么要这样区分呢，我们看下没有actor没有调度执行behavior这条分支后边会做什么。首先如果scheduled_actor有自己的私有的线程，一般都是没有的，这里不是重点。然后看下eu是execution_unit对象，那么这个execution_unit是什么呢，大家看代码就会发现execution_unit是被worker继承的。篇幅原因我省略了一些代码讲解，我直接说下原理，当actor在创建的时候会设定下他的上下文（ctx）是某个worker，而这里eu其实是发送方actor的上下文（ctx），也就是worker。这里eu就是worker，我们直接定位到函数：

  void exec_later(job_ptr job) override {
    policy_.internal_enqueue(this, job);
  }

这里其实很关键，我们能发现worker最终消费的job其实是actor，然后把job也就是actor会放到policy的队列中。

现在我们知道为什么消息放到信箱中后需要区分现在actor的状态了，如果actor现在没有执行behavior，需要将actor放到队列中被调度执行，如果现在actor正在执行behavior那么不用管，把message放到信箱里，actor执行完当前的behavior，自然会从信箱拿出新的message从而去执行相应的behavior。

任务调度执行

这一小节算是这篇文章的重点了，限于一些CAF的概念不得不讲了前边很大的一坨。

到目前为止我们明白了actor是被worker调度执行的单元，且actor是存放在policy的队列中。然后被worker消费并执行actor的behavior。
我们钱包讲到了policy其实是有两种，share和steal，顾名思义就是一种是各个policy的任务是共享的，另外一种是一个policy会去别的policy去偷任务执行。具体代码是怎样的呢？我们来看下两个policy的入队列和出队列的函数；
先看worker_stealing这个类的函数：

// worker_stealing.hpp
template <class Worker>
void internal_enqueue(Worker* self, resumable* job) {
  d(self).queue.prepend(job);
}

入队列的函数很简单，仅仅是将job放到self(worker)的worker_data中的队列中。

// worker_stealing.hpp
template <class Worker>
resumable* dequeue(Worker* self) {
  // ...
  auto* job = d(self).queue.try_take_head();
  if (job)
    return job;
  for (size_t k = 0; k < 2; ++k) { // iterate over the first two strategies
    for (size_t i = 0; i < strategies[k].attempts;
          i += strategies[k].step_size) {
      // try to steal every X poll attempts
      if ((i % strategies[k].steal_interval) == 0) {
        job = try_steal(self);
        if (job)
          return job;
      }
      // wait for some work to appear
      job = d(self).queue.try_take_head(strategies[k].sleep_duration);
      if (job)
        return job;
    }
  }
  // ...
}

template <class Worker>
resumable* try_steal(Worker* self) {
  auto p = self->parent();
  if (p->num_workers() < 2) {
    return nullptr;
  }
  // roll the dice to pick a victim other than ourselves
  auto victim = d(self).uniform(d(self).rengine);
  if (victim == self->id())
    victim = p->num_workers() - 1;

  return d(p->worker_by_id(victim)).queue.try_take_tail();
}

简单理解下，当self（worker调用policy的dequeue函数取数据传入worker对象指针）的worker_data的队列中有数据就会从该队列中取得数据，当自己的队列中没有后且满足某些策略后调用try_steal去偷取别的worker下的worker_data的任务。利用worker的parent获取到其他的worker，然后再从其worker_data的队尾取出数据，worker的parent就是调度器的类coodinator，coodinator会持有所有的worker的对象。

接下来看work_sharing的相关函数：

//work_sharing.hpp
template <class Coordinator>
bool enqueue(Coordinator* self, resumable* job) {
  queue_type l;
  l.push_back(job);
  std::unique_lock<std::mutex> guard(d(self).lock);
  d(self).queue.splice(d(self).queue.end(), l);
  d(self).cv.notify_one();
  return true;
}

template <class Worker>
resumable* dequeue(Worker* self) {
  auto& parent_data = d(self->parent());
  std::unique_lock<std::mutex> guard(parent_data.lock);
  parent_data.cv.wait(guard, [&] { return !parent_data.queue.empty(); });
  resumable* job = parent_data.queue.front();
  parent_data.queue.pop_front();
  return job;
}

share模式的policy就比较简单，入队列就是将数据放置在coordinator的队列中，然后出队列取数据时，多个worker从其parent也就是coordinator的队列获取数据并返回。

让我们再次回到worker的执行体中：

// worker.hpp
void run() {
    CAF_SET_LOGGER_SYS(&system());
    // scheduling loop
    for (;;) {
      auto job = policy_.dequeue(this);
      policy_.before_resume(this, job);
      auto res = job->resume(this, max_throughput_);
      policy_.after_resume(this, job);

      //...
    }
  }

关键点是我们会从policy调用dequeue函数，传入参数this，然后获取到job，也就是actor，然后调用actor的resume就可以执行相应的behavior了。

我们也简单过一下resume函数：

// scheduled_actor.cpp
resumable::resume_result scheduled_actor::resume(execution_unit* ctx, size_t max_throughput) {
  // ...
  auto handle_async = [this, max_throughput, &consumed](mailbox_element& x) {
    return run_with_metrics(x, [this, max_throughput, &consumed, &x] {
      switch (reactivate(x)) {
        // ...
      }
    });
  };

  while (consumed < max_throughput) {
    mailbox_.fetch_more();
  
    auto& hq = get_urgent_queue();
    auto& nq = get_normal_queue();
    if (hq.new_round(quantum * 3, handle_async).consumed_items > 0) {
      
      nq.flush_cache();
    }
    if (nq.new_round(quantum, handle_async).consumed_items > 0) {
      hq.flush_cache();
    }

    auto delta = consumed - prev;
    if (delta > 0) {
      // ...
    } else {
      reset_timeouts_if_needed();
      if (mailbox().try_block())
        return resumable::awaiting_message;
    }
  }
  // ...
}

我只列了下重要的代码，handle_async是实际上会执行behavior的地方，首先mailbox_执行fetch_more会标注mailbox_中可以被消费的消息，然后hq和nq会相继取出消息去调用handle_async执行behavior，最后消费完全后，会使得mailbox设定其状态为block，也就是我们前边说到的actor会认为时当前没有behavior正在执行。

总结

这里来总结下CAF这个是如何调度任务的：

1. 首先调度器（coordinator）会被初始化，调度器初始化多个worker，每个worker则为单独运行在一个线程上，然后worker获取任务会有两个policy来选择，共享模式和偷窃模式，共享模式是多个worker共享coordinator中的队列，偷窃模式是当一个worker使用完自己的队列就会去偷取别的worker的队列中的任务。
2. 关于CAF特有的则是，创建任务时将消息直接放到actor的信箱里，将actor放到worker的队列中，当worker消费actor并调用其resume函数，actor就可以执行所有的从信箱中获取所有的消息所匹配的behavior。

ref

https://actor-framework.readthedocs.io/en/stable/index.html
http://purecpp.cn/detail?id=2237
https://github.com/actor-framework/actor-framework