多线程作业队列管理器

Question

我需要在交互式应用程序中管理占用大量 CPU 的多任务作业。作为背景，我的具体应用是一个工程设计界面。当用户调整模型的不同参数和选项时，会在后台运行多个模拟，并在完成时显示结果，即使用户仍在编辑值也是如此。由于多次模拟需要不同的时间（有些是毫秒，有些需要 5 秒，有些需要 10 分钟），因此基本上是尽快显示反馈的问题，但通常会中止以前开始但现在不再需要的作业，因为用户的更改已经使它们无效。不同的用户更改可能会使不同的计算无效，因此在任何时候我都可能运行 10 个不同的模拟。一些模拟有多个具有依赖关系的部分（模拟 A 和 B 可以单独计算，但我需要它们的结果作为模拟 C 的种子，所以我需要等待 A 和 B 先完成，然后再开始 C。）

我非常有信心处理这种应用程序的代码级方法是某种多线程作业队列。这将包括提交作业以执行、设置任务优先级、等待作业完成、指定依赖关系（执行此作业，但仅在作业 X 和作业 Y 完成后）、取消符合某些条件的作业子集、查询什么作业保留，设置工作线程计数和优先级，等等。多平台支持也非常有用。

这些不是软件中的新想法或新愿望，但我正处于应用程序的早期设计阶段，我需要选择使用哪个库来管理此类任务。我过去曾用 C 编写过自己的粗略线程管理器（我认为这是一种通过仪式），但我想使用现代工具来作为我工作的基础，而不是我自己以前的 hack。

首先想到的是跑到OpenMP，但我不确定这是否是我想要的。 OpenMP 非常适合精细并行化、自动展开循环等。虽然是多平台的，但它也会使用#pragmas 侵入您的代码。但大多数情况下，它不是为管理大型任务而设计的。尤其是取消挂起的作业或指定依赖项。可能，是的，但它并不优雅。

我注意到Google Chrome uses such a job manager for even the most trivial tasks. 的设计目标似乎是使用户交互线程尽可能轻巧灵活，因此任何可以异步生成的东西都应该是。从 Chrome 源代码来看，这似乎不是一个通用库，但看看设计如何使用异步启动来保持快速交互仍然很有趣。这与我正在做的事情越来越相似。

还有其他选择：

Surge.Act: 一个类似于 Boost 的库，用于定义作业。它建立在 OpenMP 之上，但确实允许链接依赖项，这很好。它似乎没有可以查询的经理，可以取消工作等。这是一个陈旧的项目，因此依赖它很可怕。

Job Queue 与我的想法非常接近，但这是一篇 5 年前的文章，而不是受支持的库。

Boost.threads 确实有很好的独立于平台的同步，但这不是作业管理器。 POCO 具有非常简洁的任务启动设计，但同样不是用于链接任务的完整管理器。 （也许我低估了 POCO）。

因此，虽然有可用的选项，但我并不满意，我有再次推出自己的图书馆的冲动。但我宁愿使用已经存在的东西。即使在搜索之后（在 SO 和网络上），我也没有找到任何感觉正确的东西，尽管我想这一定是一种经常需要的工具，所以肯定有一些社区库或至少是通用设计。 在 SO 上有一些 posts 关于 job queues，但似乎没有什么合适的。

我在这里的帖子是想问你我错过了哪些现有工具，和/或你是如何推出自己的这种多线程作业队列的。

Answer 1

我们必须构建自己的作业队列系统来满足与您类似的要求（UI 线程必须始终在 33 毫秒内响应，作业可以在 15-15000 毫秒内运行），因为真的没有什么能完全满足我们的需求，让一个人就很高效。

不幸的是，我们的代码与专有代码一样具有专有性，但我可以为您提供一些最显着的功能：

• 我们在程序开始时为每个内核启动一个线程。每个都从全局作业队列中提取工作。作业由一个函数对象和一组相关数据组成（实际上是对 func_ptr 和 void * 的详细说明）。线程 0，即快速客户端循环，不允许处理作业，但其余部分尽可能地抓取。
• 作业队列本身应该是无锁数据结构，例如lock-free singly linked list (Visual Studio comes with one)。避免使用互斥锁；队列的争用非常高，而且获取互斥体的成本很高。

• 将作业的所有必要数据打包到作业对象本身中——避免将作业中的指针返回到主堆，您必须在主堆中处理作业和锁之间的争用以及所有这些其他缓慢而烦人的东西。例如，所有模拟参数都应该进入作业的本地数据块。结果结构显然需要比作业更长的时间：您可以通过以下方式处理此问题：a）即使在作业对象完成运行后仍挂在作业对象上（因此您可以使用主线程中的内容），或 b）专门为每个作业分配一个结果结构并将一个指针填充到作业的数据对象中。即使结果本身不会存在于作业中，但这有效地为作业提供了对其输出内存的独占访问权限，因此您不必为锁而烦恼。

• 实际上我在上面进行了一些简化，因为我们需要准确编排哪些作业在哪些内核上运行，因此每个内核都有自己的作业队列，但这对您来说可能是不必要的。

Answer 2

我基于 Boost.threads 推出了自己的产品。我对编写这么少的代码所获得的巨大收益感到非常惊讶。如果你没有找到预制的东西，不要害怕自己动手。在 Boost.threads 和您编写自己的代码后的经验之间，这可能比您记忆中的要容易。

对于预制选项，不要忘记 Chromium 的许可非常友好，因此您可以围绕其代码滚动您自己的通用库。

Answer 3

Microsoft 正在为下一个版本的 Visual Studio 2010 开发一组技术，称为并发运行时、并行模式库和异步代理库，这可能会有所帮助。并发运行时将提供基于策略的调度，即允许您管理和组合多个调度程序实例（类似于线程池，但实例之间具有关联和负载平衡），并行模式库将提供基于任务的编程和具有类似 STL 的并行循环编程模型。 Agents 库提供基于参与者的编程模型，并支持构建并发数据流管道，即管理上述依赖项。不幸的是，它还没有发布，所以你可以在我们的team blog 上阅读它或观看一些视频on channel9，还有一个非常大的 CTP 也可供下载。

如果您现在正在寻找解决方案，英特尔的线程构建模块和 boost 的线程库都是很好的库，并且现在可用。 JustSoftwareSolutions 发布了一个与 C++0x 草案相匹配的 std::thread 实现，当然，如果您正在研究基于细粒度循环的并行性，那么 OpenMP 是广泛可用的。

正如其他人所暗示的那样，真正的挑战是正确识别工作并将其分解为适合并发执行的任务（即没有不受保护的共享状态），了解它们之间的依赖关系并最大限度地减少瓶颈上可能发生的争用（无论是瓶颈是protecting shared state 或确保工作队列的调度循环是低争用或无锁的）......并且在没有调度实现细节泄漏到其余代码的情况下做到这一点。

-瑞克

Answer 4

threadpool 这样的东西对你有用吗？它基于 boost::threads，基本上实现了一个简单的线程任务队列，将工作函数传递给池线程。

Answer 5

您可能想查看 Flow-Based Programming - 它基于异步组件之间的数据块流。有 Java 和 C# 版本的驱动程序，以及一些预编码的组件。它本质上是多线程的——事实上唯一的单线程代码在组件中，尽管您可以将时序约束添加到标准调度规则中。尽管对于您需要的东西来说，它的粒度可能太细了，但这里可能有一些您可以使用的东西。

Answer 6

我一直在寻找几乎相同的要求。我正在开发一款具有 4x-ish 机制的游戏，并且安排完成的不同部分几乎让我的大脑爆炸。我有一组复杂的工作需要在不同的时间分辨率下完成，并根据玩家主动加载的系统/区域进行不同程度的实际模拟。这意味着随着玩家从一个系统移动到另一个系统，我需要将一个系统加载到当前的高分辨率模拟，将最后一个系统卸载到较低分辨率的模拟，并对系统的活动/非活动区域执行相同的操作。不同的模拟是基于每个实体的概况的人口、政治、军事和经济行为的大列表。到目前为止，我将尝试描述我的问题和我的方法，我希望它有助于为您或其他人描述替代方案。我正在构建的结构的粗略轮廓将使用以下内容：

1. cpp-taskflow（现代 C++ 并行任务编程库）我将创建一个模块库，用作作业构造部分。每个条目都有一个用于初始化和销毁​​的 API 以及用于通信的指针。我希望以一种可以嵌套的方式编写它，使用 cpp-taskflow API 在创建作业时设置所有依赖项，但提供一种实时调整的方法并提供可用的终止开关。我制作的大部分内容将是状态机的决策树或行为树的状态机，因此作业数据结构将是时间分辨率标记数据的设置和状态，指向实际的统计数据和对象值。
2. FlatBuffers 我希望使用这个库来构建一个“工作列表条目”以及一个“对象包装器”系统。作业队列中的每个条目都是一个 flatbuffer 对象，描述需要完成的工作（模块的设置），以及包含需要完成的工作的数据（或指向数据的共享指针）。对象存储平面缓冲区将包含表示实体表的数据。对我来说，大多数实际数据都是需要决定/处理的数组。我还希望将平面缓冲区用作线程之间的通信/控制通道。我很难让所有其他人都通过一个主“路由器”线程进行通信，或者每个线程都包含自己的线程，并且有一些发现机制。
3. SQLite 由于只有活动区域/系统需要完成更高分辨率的工作，因此游戏将创建的一些后台作业列表（针对数千个系统及其实体）将非常庞大且寿命长。数以千计的数以百万计的工作（在我看来很大），每个工作都需要未知的时间才能完成。就我而言，我不在乎他们什么时候完成，只要他们都完成（长时间的活动）。我计划在每个线程上获取一个内存中的 sqlite 数据库表作为作业队列。每个条目将包含一个平面缓冲区工作块、一个指向完成后通知的缓冲区的指针、一个指向用于更新的控制缓冲区的指针，以及装饰作业项（位置、数据范围、优先级）的其他字段，这些字段将被填充为工作条目会产生新的工作，并且随着项目被消耗到数据库中。这给了我一种方法，我可以在作业之间创建关系联系，并且如果我需要重新工作/更新作业、删除它们及其依赖关系，或者更新/重新排序优先级或依赖关系，只需构造查询。在 sqlite db 中使用的所有这些也意味着我可以随时将整个内容转储到磁盘并稍后重新加载，或者切换到附加到磁盘并从磁盘处理它。此外，这使我可以访问大量搜索和排序算法工作，我通常需要一堆不同类型的容器。能够使用 SQL 查询为我提供了很多处理作业的选项。

通信队列（作为数据库）是我是否应该通过相应线程进行访问的问题（每个线程都包含它自己的消息数据库，并且模块 API 具有抽象的锁/互斥锁），或者通过某个主路由器线程将所有更新、添加/删除和通信放入一个大型数据库中。就互斥和锁定而言，我不知道哪个最让我头疼。我花了几天时间来制作一个共享指针的怪物意大利面野兽，这些指针指向 sbuffer 池和查找表，所以每个线程都有自己的缓冲区，并分离出缓冲区。那时我决定将庞大的列表卸载到 sqlite。然后我想，为什么不直接将其他所有内容的 flatbuffer 对象放入表中。

数据库中的几乎所有内容都来自每个模块，我可以编写 sql 语句来表示我需要处理的数据的视图，以及动态地了解数据的处理方式。将工作本身放在数据库中意味着我也可以为它们做同样的事情。 SQLite 具有多线程访问，因此将其用作多线程作业队列管理器应该不会太费力。

总之，Cpp-Taskflow 将允许您使用依赖链和作业池多线程设置复杂的嵌套循环。开箱即用，它带有您需要的大部分结构。 FlatBuffers 将允许您创建作业声明和对象包装器，这些作业声明和对象包装器易于作为一个工作单元输入流缓冲区并在作业线程之间传递它们，SQLite 将允许您以某种方式标记流缓冲区作业并将其排队到 blob 条目中这应该允许添加、搜索、排序、更新和删除，而您的工作量最少。它还使保存和重新加载变得轻而易举。快照和回滚也应该是可行的，您只需要牢记数据库事件的顺序和解决方案。

编辑：尽管对此持保留态度，但我发现了您的问题，因为我正在尝试完成 Crashworks 所描述的内容。我正在考虑使用亲和力来打开长期存在的线程并让主线程运行大部分 Cpp-Taskflow 层次结构工作，将作业提供给其他线程。我还没有使用作业队列/控制通信的sqlite方法，这只是我目前的计划。

我希望有人觉得这很有帮助。

Answer 7

看看boost::future（也可以看看discussion和proposal），它看起来像是一个非常好的并行基础（特别是它似乎为C-depends-on-A-提供了极好的支持） and-B 类型的情况）。

我看了一点 OpenMP，但（像您一样）不相信它适用于除 Fortran/C 数字代码之外的任何东西。英特尔的Threading Building Blocks 对我来说看起来更有趣。

如果涉及到它，在 boost::thread 之上 roll your own 并不难。 [解释：线程farm（大多数人会称之为池）从函子（任务或作业）的线程安全queue 中提取工作。有关使用示例，请参阅 tests 和 benchmark。我有一些额外的复杂性来（可选地）支持具有优先级的任务，以及执行任务可以在工作队列中产生更多任务的情况（这使得知道所有工作何时实际完成有点问题；对“待定”的引用是可以处理案件的人）。无论如何可能会给你一些想法。]

Answer 8

你可能想看看Intel Thread Building Blocks。我相信它可以满足您的需求，并且版本 2 是开源的。

Answer 9

那里有很多分布式资源管理器。满足您几乎所有要求的软件是Sun Grid Engine。 SGE 用于一些世界上最大的超级计算机，并且正在积极开发中。

Torque、Platform LSF、Condor也有类似的解决方案。

听起来您可能想自己动手，但上述所有功能都有很多。

Answer 10

我不知道您是否正在寻找 C++ 库（我认为您是），但是 Doug Lea 的 Java 7 的 Fork/Join 框架非常漂亮，并且完全符合您的要求。您可能能够在 C++ 中实现它或找到一个预先实现的库。

更多信息在这里： http://artisans-serverintellect-com.si-eioswww6.com/default.asp?W1

Answer 11

也许有点晚了，但也看看 ThreadWeaver： http://en.wikipedia.org/wiki/ThreadWeaver