Job System 初探

468 0 0

作者:i_dovelemon的博客

作者：i_dovelemon

日期：2023-08-24

主题：Fiber, Atomic Operation, MPMC Queue, Multiple thread, Job system

引言

现代 CPU 是多核处理器，为了充分利用 CPU 多核处理的特性，游戏引擎会大量使用多线程 (multiple thread) 进行任务处理。

而为了充分利用多线程，让开发变得简单，很多引擎会提供一个 job system 的系统，从而让开发人员将任务进行多线程并行处理，大大提高程序的性能。比如 unity 的 job system。

之前阅读 OurMachinery 相关博客的时候，有看到一篇 Fiber based job system ,讲述了它们参考 Naughty Dog 在 GDC 2015 上的演讲 Parallelizing the Naughty Dog Engine Using Fibers 设计它们的 job system 的相关经验。当时看到的时候，就想着以后要来实际编写下，看看这个系统是怎么样的一个情况。

最近刚好有了时间，所以抽空研究了下，简单做了一个 demo，这里和大家分享下一些心得体会。

目前的设计，还是非常基本的，不排除有重大 bug 在里面，性能方面也不是最优化的，只能说基本满足了我需要的功能，相关代码地址可以参考这里。

概念

这个系统的代码虽然很少，但是有很多以前不太接触的概念，这里先简单介绍下。

Fiber

和 Unity 的 job system 不太一样的地方，这里设计的 job system 利用了 fiber 来作为最终执行 job 的上下文。为什么要使用 fiber 了？这是因为 fiber 的执行能够随时中断，跳转到另外一个 fiber 上去，然后在切换回来，有点类似协程。这样就使得代码的编写变得任意，我们可以随时中断我们的代码进程，让系统跳转到其他的 fiber 去执行任务。同时 fiber 的切换相对于 thread 来说，比较的轻量，性能更好。对于 fiber 的详细介绍，可以参考 Fibers

MPMC Queue

Job system 里面大量的使用了 queue 这种数据结构，但是由于多线程需要同时多读多写(multi-producer multi-consumer)，所以普通的队列无法满足要求。这里使用了一个基于原子操作，长度固定的循环队列来实现，参考这里 Bounded MPMC queue。

实现

接口

接口设计的非常简单，除了系统的启动关闭之外，就只有两个函数，如下所示：

job_fence
job_kick(job_decal* decal, uint32_t count);

void
job_wait_for_complete(job_fence fence);

　　其中 job_kick 用于抛发指定数量的任务，并且返回一个 fence 对象，用于同步等待。job_wait_for_complete 的功能就是等待指定的抛发任务全部完成的函数。

job_decal 定义如下所示：

typedef void (*job_func)(void*);

typedef struct job_decal {
    job_func    job;
    void*       data;
} job_decal;

下面是一个简单的使用案例：

void
test_job(void* job_data) {
    TracyCZoneN(ctx, "test_job", true);

    Sleep(1);
    for (uint32_t i = 0; i < 100; i++) {
        *(uint8_t*)job_data = sinf(*(uint8_t*)job_data + 100);
        *(uint8_t*)job_data = cosf(*(uint8_t*)job_data);
        *(uint8_t*)job_data = cosf(*(uint8_t*)job_data);
        *(uint8_t*)job_data = sinf(*(uint8_t*)job_data + 100);
        *(uint8_t*)job_data = cosf(*(uint8_t*)job_data);
        *(uint8_t*)job_data = cosf(*(uint8_t*)job_data);
        *(uint8_t*)job_data = sinf(*(uint8_t*)job_data + 100);
        *(uint8_t*)job_data = cosf(*(uint8_t*)job_data);
        *(uint8_t*)job_data = cosf(*(uint8_t*)job_data);
        *(uint8_t*)job_data = sinf(*(uint8_t*)job_data + 100);
        *(uint8_t*)job_data = cosf(*(uint8_t*)job_data);
        *(uint8_t*)job_data = cosf(*(uint8_t*)job_data);
    }

    TracyCZoneEnd(ctx);
}

void
test_1() {
    TracyCZoneN(ctx, "test_1", true);

    SYSTEM_INFO info;
    GetSystemInfo(&info);
    job_init(512, 512, info.dwNumberOfProcessors - 1);

    constexpr uint32_t job_count = 100;

    char test_data[job_count];

    job_decal decal[job_count];
    for (uint32_t i = 0; i < job_count; i++) {
        test_data[i] = i;
        decal[i].job = test_job;
        decal[i].data = &test_data[i];
    }

    job_fence fence = job_kick(decal, job_count);
    job_wait_for_complete(fence);

    job_shutdown();

    TracyCZoneEnd(ctx);
}

核心数据

整个系统有如下几个重要的成员数据（省略了一些不重要的成员），这些数据在系统启动之后，就全部内存分配完毕，这样当我们整个 job system 运行的时候，就不再有相关的内存分配，提高性能：

typedef struct job_counter {
    volatile atomic_word count;
    volatile atomic_word gen;
} job_counter;

typedef struct job_queue_node {
    job_decal               decal;
    uint32_t                counter_index;
    job_counter*            counter;
} job_queue_node;

typedef struct job_system {
    ......
    atomic_queue<job_queue_node>*   job_queue;

    fiber*                          fiber_pool;
    fiber*                          switch_fiber_pool;  // which fiber switch to fiber in fiber_pool
    job_queue_node*                 fiber_exec_job_pool;  // which job will be executed by fiber
    uint32_t                        fiber_count;
    atomic_queue<uint32_t>*         free_fiber_indices_queue;

    job_counter*                    counter_pool;
    uint32_t                        counter_count;
    atomic_queue<uint32_t>*         free_counter_indices_queue;
} job_system;

成员主要分成 3 个大部分：用于管理抛发任务的任务队列；用于管理 fiber 的相关数据；用于管理 job_counter，形成同步点的相关数据；

Job Queue

job queue 是一个 mpmc queue 的队列，用于多线程的从这里面 enqueue/dequeue 任务，没有太复杂的地方。

Job Counter

Job counter 是一组原子计数器，当我们抛发一组任务的时候，会从空闲队列中拿出一个计数器，用来保存当前抛发了多少任务。当每一个任务执行完毕的时候，这个计数器的值就减一。当这个计数器为 0 的时候，就表示所有的任务已经完成。

我们看到 job_counter 的数据结构，不是只有一个 count，还有一个 gen 成员。这是因为原子计数器会被重复使用，一旦所有抛发的任务执行完毕之后，我们就会将计数器返回队列，那样就有机会被别人使用。这样就有可能出现，明明任务已经完成，但是计数器又不为 0 的情况。为了解决这个问题，特意添加了一个 generation 成员，用于表示当前的计数器是第多少代。这样当我们抛发任务，产生了一个 job_fence 的时候，可以将当前计数器的 gen 保存下来。当出现之前说的同一个计数器被多次使用的时候，我们就可以通过 gen 成员来判断是否为同一个计数器，从而正确的进行任务的同步。

Fiber Pool

fiber pool 和 counter pool 类似，预先创建好了指定数量的 fiber，同时准备了一个空闲队列，用于获取空闲的 fiber。

之前说过，这个系统执行任务的单元是 fiber，而不是线程，所以 fiber_exec_job_pool 用于保存对应位置的 fiber 所需要执行的 job 相关信息，这样我们就能够在 fiber 中获取当前需要执行的 job 相关信息，从而执行。

了解过 fiber 相关概念之后，我们就知道，一个 fiber a 想要执行，就需要让 fiber b 通过调用 switch fiber 来将当前线程的执行核心转移到 fiber a。而一旦我们执行完毕了 fiber a，就需要将执行权转移回 fiber b，从而让 fiber b 的代码继续执行，实现类似协程的效果。所以 switch_fiber_pool 用于保存从哪个 fiber 切换到了当前 fiber，从而当 fiber 执行完毕之后，能够正确的 switch 回去。

重要流程

绑定线程核心

job system 需要一些 woker thread，这些 worker thread 需要绑定到实际的 cpu 核心上去，这样才能够避免操作系统的调度，导致线程的执行发生中断，影响性能。而这个操作可以通过设置线程的 affinity mask 来完成，windows 下可以参考 SetThreadAffinityMask 。

我们可以通过在线程中执行一个无限循环函数，来判断线程是否正确的绑定到了 cpu 核心上去。当我们正确的绑定了之后，执行无限循环函数会将对应的 cpu 资源全部沾满，从而能够在 windows 资源管理器中看到如下图所示的样子：

工作线程主循环

工作线程，除了每个循环检测下是否需要退出线程之外，主要的任务就是判断是否有空闲的 job 需要执行。如果有空闲的 job 需要执行，那就拿取一个空闲的 job，并再拿取一个空闲的 fiber，设置相关的信息，然后将执行权交给对应的 fiber，让它去执行任务。当 fiber 执行完毕任务之后，将执行权再转交给之前的 fiber。

job_wait_for_complete

这个同步函数，并不是一直阻塞等待任务完成。它在等待的过程中，除了判断对应的 job counter 是否为 0，即所有任务都执行完毕之外，也运行了工作线程的主循环函数，从而能够充分利用等待时间去执行更多的任务。

多线程 profiler

有了多线程功能之后，需要进行调试，需要知道任务是否正确的并行处理了，依赖关系是否正确的建立了。所以就自然的想到了类似 unity 那样的 profiler ，它能够直观的看到 job 的执行情况。所以搜索了一些 profiler，看看有没有好用的。但是找下来，大部分都是性能分析相关的 profiler，支持多线程 timeline 的大部分都是收费的产品。最终找到了一个名为 Tracy Profiler 的开源工具，只要简单的接入，就能够实现 job 的 timeline profiler，题图就是来自于这个工具的截图，推荐大家使用。