Add various C++ concurrency and asynchronous programming notes

posts/数据结构与算法/单调队列.md

@@ -1,3 +1,8 @@
+---
+tags:
+  - 数据结构与算法
+---
+
 # 单调队列
 
 一个队列, 但是单调. ( 我是懂废话的
@@ -58,4 +63,4 @@ vector<int> maxSlidingWindow(vector<int>& nums, int k)
 
 上文提到了——**区间最值**, 也就是说我们需要维护一个动态区间最值的时候可以考虑使用单调队列来进行实现(当然也要考虑左右边界是不是基本单向移动, 然后旧元素是不是按照时间顺序过期之类的事情) —— 比如什么优化dp, 滑动窗口打什么 Patch 之类的……
 
-能干求区间最值这活的, 什么线段树, ST 表之类的也能干, 所以看实际情况来选择吧.
+能干求区间最值这活的, 什么线段树, ST 表之类的也能干, 所以看实际情况来选择吧.

posts/游戏开发/ChikaEngine/Job System/ParallelFor.md

@@ -0,0 +1,205 @@
+---
+tags:
+  - ChikaEngine
+  - 并发编程
+  - 异步与调度
+---
+
+# ParallelFor
+
+简单说下这个方法的设计, 其实非常的简单, 就是上游业务调用的时候, 可以把一个大的 Range 拆分成若干小的 Chunk 然后进行并行执行.
+
+比如说要做可见性剔除的操作, 我们有 1b 个物体
+
+```C++
+for (auto go : gameobjects)
+	operate(go);
+```
+
+这显然是一个非常贵的操作, 于是就可以调用 `ParallelFor` 来把这个大的数据拆成若干小的部分执行操作.
+
+## 数据流动
+
+我们借助一次 ChikaEngine 的真实调用来说 —— `BuildVisibilityParallel` 可见性剔除, 对于每一个 GO 的可见性剔除操作是彼此无关的, 所以与其使用一个循环进行串型运算, 不如直接交给 `ParallelFor` 进行并行运算.
+
+### 准备 ParallelFor 参数
+
+查看 `BuildVisibilityParallel` 的实现
+
+```C++
+const uint32_t count = static_cast<uint32_t>(instances.size());
+const uint32_t safeGrain = std::max(1u, config.grainSize);
+const uint32_t maximumChunks = std::max(1u, jobs.GetMaximumParallelChunks());
+const uint32_t chunkCapacity = std::min((count + safeGrain - 1u) / safeGrain, maximumChunks);
+std::vector<VisibilityResult> chunks(chunkCapacity);
+```
+
+- `count` 当前这一帧需要检查多少个 `RenderInstance` —— 所以 ParallelFor 的工作空间就是 `[0, count)`
+- `safeGrain` 大概分割每个 Chunk 的大小
+- `chunks` 提前创建每个 `chunk` 操作的输出存储位置
+
+### 调用
+
+```C++
+const Jobs::JobHandle parallel = Jobs::ParallelFor(
+	jobs,
+	count,
+	safeGrain,
+	"Renderer.Visibility.Chunk",
+	[&](Jobs::ParallelForRange range) { ... }
+)
+```
+
+传入 `jobs` 表示当前的方法要把任务拆分到哪个 `JobSystem` 中, 正常传入参数(还是一如既往的喜欢说废话), 最后的 lambda 表示对于每一个 Chunk 我们要执行什么操作.
+
+### 进入 ParallelFor
+
+进行一波运算 (比如 Range 其实比较小就直接打包提交给 JobSystem 之类的)
+
+然后拆分 Range 成若干 Chunk 并且给丢给 JobSystem ——
+
+```C++
+[sharedFunction]()
+{
+    (*sharedFunction)(
+        {0, 128, 0});
+}
+
+[sharedFunction]()
+{
+    (*sharedFunction)(
+        {128, 256, 1});
+}
+
+...
+```
+
+类似这样的东西
+
+### 总结
+
+那么请 G 老师重新展示 ASCII 艺术 ——
+
+```
+                   RenderSceneView
+                         │
+                         ▼
+                scene.GetInstances()
+                         │
+                         ▼
+                  instances[]
+                         │
+                         │ shared read
+                         ▼
+              BuildVisibilityParallel
+                         │
+                         ▼
+     ParallelFor(count, grain, lambda)
+                         │
+                Split [0,count)
+                         │
+         ┌───────────────┼───────────────┐
+         ▼               ▼               ▼
+      Chunk0          Chunk1          Chunk2
+      [0,128)        [128,256)       [256,384)
+         │               │               │
+         ▼               ▼               ▼
+       Job0             Job1            Job2
+         │               │               │
+         └───────────────┼───────────────┘
+                         ▼
+                      JobSystem
+                         │
+                         ▼
+                 Worker Scheduling
+                         │
+       ┌─────────────────┼─────────────────┐
+       ▼                 ▼                 ▼
+    Worker0           Worker1           Worker2
+       │                 │                 │
+       ▼                 ▼                 ▼
+ functor(range0)   functor(range1)   functor(range2)
+       │                 │                 │
+       ▼                 ▼                 ▼
+ instances[0..]    instances[128..]  instances[256..]
+       │                 │                 │
+       ▼                 ▼                 ▼
+  Visibility Test    Visibility Test  Visibility Test
+       │                 │                 │
+       ▼                 ▼                 ▼
+   chunks[0]          chunks[1]          chunks[2]
+       │                 │                 │
+       └─────────────────┼─────────────────┘
+                         │
+                         ▼
+                    Join / Wait
+                         │
+                         ▼
+                  Merge chunks[]
+                         │
+                         ▼
+                 VisibilityResult
+```
+
+做了一个简单示意图, 大体确乎如此, 不过真实情况需要看 JobSystem 的调度结果
+
+## ParallelFor
+
+现在回到 “正题” 稍微说说这个函数的设计.
+
+首先传进来的 Callable 要是可以说明对于 `ParallelForRange` 来说要怎么进行操作.
+
+```C++
+using StoredFunction = std::remove_cvref_t<Function>;
+
+auto sharedFunction = std::make_shared<StoredFunction>(std::forward<Function>(function));
+
+```
+
+先获取数传入的 Callable 类型, 然后做一个引用转发, 用 `make_shared` 进行存储这个方法. 使用 `make_shared` 是因为后面拆分成若干 Chunk 丢给 JobSystem 的时候, 这份 Functor 不是谁独占所有权, 没有谁可以保证自己最后被执行完并且来做内存释放. 所以使用 `shared_ptr` 来进行实现, 让这个 Function 的生命周期可以覆盖所有的 Chunk 执行生命周期.
+
+接着我们进行拆分 ——
+
+```C++
+for (uint32_t chunkIndex = 0; chunkIndex < chunkCount; ++chunkIndex)
+{
+	...
+}
+```
+
+此处除了创建 Job 之外, 我们还记录其 Handle.
+
+拆分结束之后
+
+```C++
+JobHandle join = jobs.ScheduleAfter(chunks, "ParallelFor.Join", [] {}, JobFailurePolicy::Propagate);
+```
+
+借助一次空任务来建立所有 chunk 的依赖关系, 这样返回这个 `join` 后的任务的 `Job Handle`, 之后在业务代码调用 `Wait` 的时候就是对所有数据的 `Wait` (其他操作同理).
+
+## 异常分析
+
+首先上游业务代码调用 `ParallelFor`, 然后它一路执行, 如果此时在构造 Chunk 丢给 JobSystem 的时候遇到错误. 那么
+
+```C++
+JobHandle chunk = jobs.Schedule(name, [sharedFunction, begin, end, chunkIndex]() { (*sharedFunction)(ParallelForRange{ begin, end, chunkIndex }); });
+if (!chunk.IsValid())
+{
+	Detail::CleanupParallelChunks(jobs, chunks);
+	throw JobCapacityError("failed to schedule ParallelFor chunk");
+}
+```
+
+抛出异常, 同时说明是在 `ParallelFor` 当中构造 Chunk 的时候的异常, 原因是 `JobCapacityError`
+
+同时执行一次 `Cleanup` —— 等待之前已经创建过的 Chunk 全部执行结束并且 Release, 消费掉之前创建的 Chunk —— 那么在 `wait` 的时候也有可能抛出异常, 此时我们把 `catch(...)` 留空, 保证不会在消费 Chunk 的时候被打断, 这样才能正确的消费掉所有的 Chunk. 同时不会打断上层调用的时候抛出的问题 `JobCapacityError`, 这才是主要错误.
+
+这个逻辑在 `join` 这个依赖任务创建的时候也是同理.
+
+以及如果在 JobSystem 内部出现了问题, 不好意思, 此时 `ParallelFor` 可能早就 return 返回了, 所以这个错误不会传递到 `ParallelFor` 中, 以及就算出现了什么玄学错误被传播到了这个方法中, 我们也没有对应的 catch —— 非常简单, 因为这只是一个 “中间件”, 也不应当有什么处理异常的能力.
+
+## 总结
+
+爽点, 底层数据不需要做所有权等的移交, 上层只是一直不断的在借用而已(传递一个引用); ParallelFor 不需要知道具体的“数据”是什么, 只需要知道范围. 也不需要知道怎么调度, 这不是它的职责; JobSystem 只管每个 Job 的调度, 然后执行这个 Job 即可, 不知道底层数据, 不需要知道 Chunk 的概念.
+
+这很 “单一职责”, 不赖.

posts/游戏开发/ChikaEngine/Job System/Wait.md

@@ -1,3 +1,10 @@
+---
+tags:
+  - ChikaEngine
+  - 并发编程
+  - 异步与调度
+---
+
 # Wait
 
 假设一个线程调用了 `Wait()` 方法, 那么首先它确乎是会阻止代码继续往下运行, 而是等待任务执行完毕.
@@ -22,4 +29,4 @@ slot->condition.wait(...)
 
 那么允许参与 `Help` 的线程只有 main thread 以及**当前**(发起 Wait 的线程)中 JobSystem 中的 Workers —— 我们可以假设没有 help 机制, 全部睡眠或者阻塞.
 
-那么有一种极端情况 —— Worker 0 等待一个 Worker 1 的任务, 于是 Worker 0 阻塞; Worker 1 等待 Worker 2 的任务, 于是 Worker 1 阻塞 ...... 直到最后所有 Worker 全部阻塞, 没有人继续干活(. 或者不形成环, 单纯是 Worker 0 等待 Worker 1; Worker 1 等待 Worker 0 而形成的死锁也有可能发生. ( 不是说可以解决逻辑死锁, 而是说可以解决饥饿死锁 —— 明明有处于 ready 的任务, 但是依旧所有线程挂起而不是恢复去执行)
+那么有一种极端情况 —— Worker 0 等待一个 Worker 1 的任务, 于是 Worker 0 阻塞; Worker 1 等待 Worker 2 的任务, 于是 Worker 1 阻塞 ...... 直到最后所有 Worker 全部阻塞, 没有人继续干活(. 或者不形成环, 单纯是 Worker 0 等待 Worker 1; Worker 1 等待 Worker 0 而形成的死锁也有可能发生. ( 不是说可以解决逻辑死锁, 而是说可以解决饥饿死锁 —— 明明有处于 ready 的任务, 但是依旧所有线程挂起而不是恢复去执行)

posts/游戏开发/ChikaEngine/Job System/Work Stealing.md

@@ -1,3 +1,10 @@
+---
+tags:
+  - ChikaEngine
+  - 并发编程
+  - 异步与调度
+---
+
 # Work Stealing
 
 当前的实现是, 优先执行自己队列中的任务, 如果空闲的话, 先尝试从全局的 InjectionQueue 中尝试接单, 如果还是空闲, 则尝试从其他 Worker 的队列中领任务. 如果还是空, 则进入休眠,交出资源,等待唤醒.
@@ -66,4 +73,4 @@ Worker 对于自己的 LocalQueue 的操作是 —— 不管是 Push (自己产
 					  Queued → Running
 ```
 
-差不多比较清晰. 稍作补充, `Main Thread` 是通过 `PumpMainThreadJobs` 来进行消费的
+差不多比较清晰. 稍作补充, `Main Thread` 是通过 `PumpMainThreadJobs` 来进行消费的

posts/游戏开发/ChikaEngine/Job System/线程局部调度上下文.md

@@ -1,3 +1,10 @@
+---
+tags:
+  - ChikaEngine
+  - 并发编程
+  - 异步与调度
+---
+
 # 线程局部调度上下文
 
 打算稍微说说 `Job System` 中的 `thread_local` 声明的变量的用途
@@ -115,4 +122,3 @@ const bool internalSubmission = g_scheduler == this && g_currentJob.IsValid();
 internalSubmission &&
 m_acceptingWorkerSubmissions.load(std::memory_order_acquire)
 ```
-

posts/编程语言/C++/并发模型/Condition Variable.md

@@ -1,3 +1,10 @@
+---
+tags:
+  - C++
+  - 并发编程
+  - 同步机制
+---
+
 # Condition Variable
 
 条件变量, 用于阻塞等待的同步元语.
@@ -131,4 +138,3 @@ if (success == false)
 `notify_one()` 唤醒至多一个正在该 Condition Variable 上等待的线程. 可以表示一些一对一的关系, 比如每次提交了一个任务, 那么也只需要唤醒一个 worker 即可.
 
 `notify_all()` 对应的, 意思是唤醒在该 CV 上等待的所有线程, 不过这可能涉及被唤醒的多个线程之间继续抢占锁的情况. 不过可以用来广播所有的线程状态变化 —— 比如让所有 worker shutdown, 就可以使用
-

posts/编程语言/C++/并发模型/CurrentWorker.md

@@ -1,3 +1,9 @@
+---
+tags:
+  - C++
+  - 并发编程
+---
+
 # CurrentWorker
 
 这是一个编外篇, 简单说明清楚一下使用一个 `local_thread` 来记录当前线程的设计.
@@ -49,4 +55,4 @@ int main()
 
 不过之前 `thread_local` 中的多 thread 最后做聚合, 其实倒是和当前这个例子有点像
 
-或许可以在脑子里有一个小建模 —— 就是一个哈希表, 是 current thread -> 对应的 local_thread 变量的映射. 那么每次看到访问这个变量的时候, 要想清楚这是哪个 thread 的变量.
+或许可以在脑子里有一个小建模 —— 就是一个哈希表, 是 current thread -> 对应的 local_thread 变量的映射. 那么每次看到访问这个变量的时候, 要想清楚这是哪个 thread 的变量.

posts/编程语言/C++/并发模型/thread_local.md

@@ -1,3 +1,10 @@
+---
+tags:
+  - C++
+  - 并发编程
+  - 内存管理
+---
+
 # thread_local
 
 由这个关键字声明的变量, 在每一个线程内都有一个自己独立的实例.
@@ -140,4 +147,4 @@ void Foo()
 
 不提供例子了, 自己看 ChikaEngine JobSystem 中的 `thread_local` 用法(
 
-与此同时, `thread_local` 在调用链比较深的时候也比较好用. 比如传递上下文的时候, 假设我们有 `layer1(ctx) -> layer2(ctx) -> layer3(ctx) -> layer4(ctx)`, 那么就可以使用 `thread_local` 进行化简, 是的 layer 1-4 不再需要传递 ctx, 而是直接用这个 `thread_local` 声明的变量即可.
+与此同时, `thread_local` 在调用链比较深的时候也比较好用. 比如传递上下文的时候, 假设我们有 `layer1(ctx) -> layer2(ctx) -> layer3(ctx) -> layer4(ctx)`, 那么就可以使用 `thread_local` 进行化简, 是的 layer 1-4 不再需要传递 ctx, 而是直接用这个 `thread_local` 声明的变量即可.

posts/编程语言/C++/异常安全.md

@@ -1,3 +1,10 @@
+---
+tags:
+  - C++
+  - 错误处理
+  - 资源管理
+---
+
 # 异常安全
 
 之前说过了`try-catch`模型, 那么异常安全实际上在思考一件事情 —— 在发生了异常之后, 整个系统现在是什么状态
@@ -148,4 +155,4 @@ void Foo()
 
 ## 总结
 
-Strong Guarantee 确乎很帅, 但是也带来了额外的空间, 心智负担等, 所以看业务是否需要吧. 不过确乎是有意思的. 当然, 其实在实践当中也会发现, 我们调用的外部方法产生的副作用对 Strong Guarantee 的影响非常大, 在分析的时候注意点吧.
+Strong Guarantee 确乎很帅, 但是也带来了额外的空间, 心智负担等, 所以看业务是否需要吧. 不过确乎是有意思的. 当然, 其实在实践当中也会发现, 我们调用的外部方法产生的副作用对 Strong Guarantee 的影响非常大, 在分析的时候注意点吧.