Execution Queue

A high performance execution queue.

概述

类似于kylin的ExecMan, ExecutionQueue提供了异步串行执行的功能。ExecutionQueue的相关技术最早使用在RPC中实现多线程向同一个fd写数据. 在r31345之后加入到bthread。 ExecutionQueue 提供了如下基本功能:

  • 异步有序执行: 任务在另外一个单独的线程中执行, 并且执行顺序严格和提交顺序一致.
  • Multi Producer: 多个线程可以同时向一个ExecutionQueue提交任务
  • 支持cancel一个已经提交的任务
  • 支持stop
  • 支持高优任务插队

和ExecMan的主要区别:

  • ExecutionQueue的任务提交接口是wait-free的, ExecMan依赖了lock, 这意味着当机器整体比较繁忙的时候,使用ExecutionQueue不会因为某个进程被系统强制切换导致所有线程都被阻塞。
  • ExecutionQueue支持批量处理: 执行线程可以批量处理提交的任务, 获得更好的locality. ExecMan的某个线程处理完某个AsyncClient的AsyncContext之后下一个任务很可能是属于另外一个AsyncClient的AsyncContex, 这时候cpu cache会在不同AsyncClient依赖的资源间进行不停的切换。
  • ExecutionQueue的处理函数不会被绑定到固定的线程中执行, ExecMan中是根据AsyncClient hash到固定的执行线程,不同的ExecutionQueue之间的任务处理完全独立,当线程数足够多的情况下,所有非空闲的ExecutionQueue都能同时得到调度。同时也意味着当线程数不足的时候,ExecutionQueue无法保证公平性, 当发生这种情况的时候需要动态增加bthread的worker线程来增加整体的处理能力.
  • ExecutionQueue运行线程为bthread, 可以随意的使用一些bthread同步原语而不用担心阻塞pthread的执行. 而在ExecMan里面得尽量避免使用较高概率会导致阻塞的同步原语.

背景

在多核并发编程领域, Message passing作为一种解决竞争的手段得到了比较广泛的应用,它按照业务依赖的资源将逻辑拆分成若干个独立actor,每个actor负责对应资源的维护工作,当一个流程需要修改某个资源的时候, 就转化为一个消息发送给对应actor,这个actor(通常在另外的上下文中)根据命令内容对这个资源进行相应的修改,之后可以选择唤醒调用者(同步)或者提交到下一个actor(异步)的方式进行后续处理。

img

ExecutionQueue Vs Mutex

ExecutionQueue和mutex都可以用来在多线程场景中消除竞争. 相比较使用mutex, 使用ExecutionQueue有着如下几个优点:

  • 角色划分比较清晰, 概念理解上比较简单, 实现中无需考虑锁带来的问题(比如死锁)
  • 能保证任务的执行顺序,mutex的唤醒顺序不能得到严格保证.
  • 所有线程各司其职,都能在做有用的事情,不存在等待.
  • 在繁忙、卡顿的情况下能更好的批量执行,整体上获得较高的吞吐.

但是缺点也同样明显:

  • 一个流程的代码往往散落在多个地方,代码理解和维护成本高。
  • 为了提高并发度, 一件事情往往会被拆分到多个ExecutionQueue进行流水线处理,这样会导致在多核之间不停的进行切换,会付出额外的调度以及同步cache的开销, 尤其是竞争的临界区非常小的情况下, 这些开销不能忽略.
  • 同时原子的操作多个资源实现会变得复杂, 使用mutex可以同时锁住多个mutex, 用了ExeuctionQueue就需要依赖额外的dispatch queue了。
  • 由于所有操作都是单线程的,某个任务运行慢了就会阻塞同一个ExecutionQueue的其他操作。
  • 并发控制变得复杂,ExecutionQueue可能会由于缓存的任务过多占用过多的内存。

不考虑性能和复杂度,理论上任何系统都可以只使用mutex或者ExecutionQueue来消除竞争. 但是复杂系统的设计上,建议根据不同的场景灵活决定如何使用这两个工具:

  • 如果临界区非常小,竞争又不是很激烈,优先选择使用mutex, 之后可以结合contention profiler来判断mutex是否成为瓶颈。
  • 需要有序执行,或者无法消除的激烈竞争但是可以通过批量执行来提高吞吐, 可以选择使用ExecutionQueue。

总之,多线程编程没有万能的模型,需要根据具体的场景,结合丰富的profliling工具,最终在复杂度和性能之间找到合适的平衡。

特别指出一点,Linux中mutex无竞争的lock/unlock只有需要几条原子指令,在绝大多数场景下的开销都可以忽略不计.

使用方式

实现执行函数

  1. // Iterate over the given tasks
  2. //
  3. // Example:
  4. //
  5. // #include <bthread/execution_queue.h>
  6. //
  7. // int demo_execute(void* meta, TaskIterator<T>& iter) {
  8. // if (iter.is_stopped()) {
  9. // // destroy meta and related resources
  10. // return 0;
  11. // }
  12. // for (; iter; ++iter) {
  13. // // do_something(meta, *iter)
  14. // // or do_something(meta, iter->a_member_of_T)
  15. // }
  16. // return 0;
  17. // }
  18. template <typename T>
  19. class TaskIterator;

启动一个ExecutionQueue:

  1. // Start a ExecutionQueue. If |options| is NULL, the queue will be created with
  2. // default options.
  3. // Returns 0 on success, errno otherwise
  4. // NOTE: type |T| can be non-POD but must be copy-constructible
  5. template <typename T>
  6. int execution_queue_start(
  7. ExecutionQueueId<T>* id,
  8. const ExecutionQueueOptions* options,
  9. int (*execute)(void* meta, TaskIterator<T>& iter),
  10. void* meta);

创建的返回值是一个64位的id, 相当于ExecutionQueue实例的一个弱引用, 可以wait-free的在O(1)时间内定位一个ExecutionQueue, 你可以到处拷贝这个id, 甚至可以放在RPC中,作为远端资源的定位工具。 你必须保证meta的生命周期,在对应的ExecutionQueue真正停止前不会释放.

停止一个ExecutionQueue:

  1. // Stop the ExecutionQueue.
  2. // After this function is called:
  3. // - All the following calls to execution_queue_execute would fail immediately.
  4. // - The executor will call |execute| with TaskIterator::is_queue_stopped() being
  5. // true exactly once when all the pending tasks have been executed, and after
  6. // this point it's ok to release the resource referenced by |meta|.
  7. // Returns 0 on success, errno othrwise
  8. template <typename T>
  9. int execution_queue_stop(ExecutionQueueId<T> id);
  10. // Wait until the the stop task (Iterator::is_queue_stopped() returns true) has
  11. // been executed
  12. template <typename T>
  13. int execution_queue_join(ExecutionQueueId<T> id);

stop和join都可以多次调用, 都会有合理的行为。stop可以随时调用而不用当心线程安全性问题。

和fd的close类似,如果stop不被调用, 相应的资源会永久泄露。

安全释放meta的时机: 可以在execute函数中收到iter.is_queue_stopped()==true的任务的时候释放,也可以等到join返回之后释放. 注意不要double-free

提交任务

  1. struct TaskOptions {
  2. TaskOptions();
  3. TaskOptions(bool high_priority, bool in_place_if_possible);
  4. // Executor would execute high-priority tasks in the FIFO order but before
  5. // all pending normal-priority tasks.
  6. // NOTE: We don't guarantee any kind of real-time as there might be tasks still
  7. // in process which are uninterruptible.
  8. //
  9. // Default: false
  10. bool high_priority;
  11. // If |in_place_if_possible| is true, execution_queue_execute would call
  12. // execute immediately instead of starting a bthread if possible
  13. //
  14. // Note: Running callbacks in place might cause the dead lock issue, you
  15. // should be very careful turning this flag on.
  16. //
  17. // Default: false
  18. bool in_place_if_possible;
  19. };
  20. const static TaskOptions TASK_OPTIONS_NORMAL = TaskOptions(/*high_priority=*/ false, /*in_place_if_possible=*/ false);
  21. const static TaskOptions TASK_OPTIONS_URGENT = TaskOptions(/*high_priority=*/ true, /*in_place_if_possible=*/ false);
  22. const static TaskOptions TASK_OPTIONS_INPLACE = TaskOptions(/*high_priority=*/ false, /*in_place_if_possible=*/ true);
  23. // Thread-safe and Wait-free.
  24. // Execute a task with defaut TaskOptions (normal task);
  25. template <typename T>
  26. int execution_queue_execute(ExecutionQueueId<T> id,
  27. typename butil::add_const_reference<T>::type task);
  28. // Thread-safe and Wait-free.
  29. // Execute a task with options. e.g
  30. // bthread::execution_queue_execute(queue, task, &bthread::TASK_OPTIONS_URGENT)
  31. // If |options| is NULL, we will use default options (normal task)
  32. // If |handle| is not NULL, we will assign it with the hanlder of this task.
  33. template <typename T>
  34. int execution_queue_execute(ExecutionQueueId<T> id,
  35. typename butil::add_const_reference<T>::type task,
  36. const TaskOptions* options);
  37. template <typename T>
  38. int execution_queue_execute(ExecutionQueueId<T> id,
  39. typename butil::add_const_reference<T>::type task,
  40. const TaskOptions* options,
  41. TaskHandle* handle);

high_priority的task之间的执行顺序也会严格按照提交顺序, 这点和ExecMan不同, ExecMan的QueueExecEmergent的AsyncContex执行顺序是undefined. 但是这也意味着你没有办法将任何任务插队到一个high priority的任务之前执行.

开启inplace_if_possible, 在无竞争的场景中可以省去一次线程调度和cache同步的开销. 但是可能会造成死锁或者递归层数过多(比如不停的ping-pong)的问题,开启前请确定你的代码中不存在这些问题。

取消一个已提交任务

  1. /// [Thread safe and ABA free] Cancel the corresponding task.
  2. // Returns:
  3. // -1: The task was executed or h is an invalid handle
  4. // 0: Success
  5. // 1: The task is executing
  6. int execution_queue_cancel(const TaskHandle& h);

返回非0仅仅意味着ExecutionQueue已经将对应的task递给过execute, 真实的逻辑中可能将这个task缓存在另外的容器中,所以这并不意味着逻辑上的task已经结束,你需要在自己的业务上保证这一点.


Last modified February 5, 2024: Release bRPC 1.8.0 (254a6bb)