add ConcurrentExecutionQueue (#45)

* add ConcurrentExecutionQueue

* fix tsan test for  ConcurrentExecutionQueue

* re-oganize pipeline

* Update execution_queue.md

Author: oathdruid

.github/workflows/ci.yml

@@ -2,139 +2,76 @@ name: CI
 
 on:
   push:
-    branches: ["main"]
+    branches: [main]
   pull_request:
-    branches: ["main"]
+    branches: [main]
 
 jobs:
-  gcc12-basic-asan:
+  basic:
+    strategy:
+      matrix:
+        compiler: [{name: gcc, flag: --action_env=CC=gcc-12}, {name: clang, flag: --action_env=CC=clang-14}]
+        std: [{name: c++17}, {name: c++14, flag: --cxxopt=-std=c++14 --cxxopt=-faligned-new}]
+        stdlib: [{name: stdlibc++}, {name: libc++, flag: --cxxopt=-stdlib=libc++ --linkopt=-stdlib=libc++}]
+        feature: [{name: asan, flag: --features=asan}, {name: tsan, flag: --features=tsan}]
+        exclude:
+        - compiler: {name: gcc}
+          stdlib: {name: libc++}
+        - std: {name: c++14}
+          feature: {name: tsan}
     runs-on: ubuntu-latest
+    name: basic-${{matrix.compiler.name}}-${{matrix.std.name}}-${{matrix.stdlib.name}}-${{matrix.feature.name}}
     steps:
     - uses: actions/checkout@v4
     - uses: actions/cache@v4
       with:
         path: bazel-disk
-        key: bazel-disk-gcc12-basic-asan-${{github.sha}}
-        restore-keys: bazel-disk-gcc12-basic-asan-
-    - run: bazel test --disk_cache=bazel-disk --action_env=CC=gcc-12 --features=asan test/...
+        key: bazel-disk-basic-${{matrix.compiler.name}}-${{matrix.std.name}}-${{matrix.stdlib.name}}-${{matrix.feature.name}}-${{github.sha}}
+        restore-keys: bazel-disk-basic-${{matrix.compiler.name}}-${{matrix.std.name}}-${{matrix.stdlib.name}}-${{matrix.feature.name}}-
+        save-always: true
+    - run: bazel test --disk_cache=bazel-disk --verbose_failures ${{matrix.compiler.flag}} ${{matrix.std.flag}} ${{matrix.stdlib.flag}} ${{matrix.feature.flag}} test/...
 
-  gcc12-basic-tsan:
+  arenastring:
+    strategy:
+      matrix:
+        compiler: [{name: gcc, flag: --action_env=CC=gcc-12}, {name: clang, flag: --action_env=CC=clang-14}]
+        stdlib: [{name: stdlibc++}, {name: libc++, flag: --cxxopt=-stdlib=libc++ --linkopt=-stdlib=libc++}]
+        mutable: [{name: default}, {name: half, flag: --config=arenastring}, {name: full, flag: --config=mutable-donated-string}]
+        exclude:
+        - compiler: {name: gcc}
+          stdlib: {name: libc++}
+        - compiler: {name: clang}
+          mutable: {name: half}
+        - compiler: {name: clang}
+          mutable: {name: full}
     runs-on: ubuntu-latest
+    name: arenastring-${{matrix.compiler.name}}-${{matrix.stdlib.name}}-${{matrix.mutable.name}}
     steps:
     - uses: actions/checkout@v4
     - uses: actions/cache@v4
       with:
         path: bazel-disk
-        key: bazel-disk-gcc12-basic-tsan-${{github.sha}}
-        restore-keys: bazel-disk-gcc12-basic-tsan-
-    - run: bazel test --disk_cache=bazel-disk --action_env=CC=gcc-12 --features=tsan test/...
+        key: bazel-disk-arenastring-${{matrix.compiler.name}}-${{matrix.stdlib.name}}-${{matrix.mutable.name}}-${{github.sha}}
+        restore-keys: bazel-disk-arenastring-${{matrix.compiler.name}}-${{matrix.stdlib.name}}-${{matrix.mutable.name}}-
+        save-always: true
+    - run: sed -i "/single_version_override.*protobuf/s/version = '[^']*'/version = '27.3.arenastring'/" MODULE.bazel
+    - run: bazel test --disk_cache=bazel-disk --verbose_failures --features=asan ${{matrix.compiler.flag}} ${{matrix.stdlib.flag}} ${{matrix.mutable.flag}} test/...
 
-  gcc12-arenastring-asan:
+  aarch64:
     runs-on: ubuntu-latest
     steps:
     - uses: actions/checkout@v4
     - uses: actions/cache@v4
       with:
         path: bazel-disk
-        key: bazel-disk-gcc12-arenastring-${{github.sha}}
-        restore-keys: bazel-disk-gcc12-arenastring-
-    - run: sed -i "/single_version_override.*protobuf/s/version = '[^']*'/version = '25.3.arenastring'/" MODULE.bazel
-    - run: bazel test --disk_cache=bazel-disk --action_env=CC=gcc-12 --config=arenastring --features=asan test/...
-
-  gcc12-mutable-donated-string-asan:
-    runs-on: ubuntu-latest
-    steps:
-    - uses: actions/checkout@v4
-    - uses: actions/cache@v4
-      with:
-        path: bazel-disk
-        key: bazel-disk-gcc12-mutable-donated-string-${{github.sha}}
-        restore-keys: bazel-disk-gcc12-mutable-donated-string-
-    - run: sed -i "/single_version_override.*protobuf/s/version = '[^']*'/version = '25.3.arenastring'/" MODULE.bazel
-    - run: bazel test --disk_cache=bazel-disk --action_env=CC=gcc-12 --config=mutable-donated-string --features=asan test/...
-
-  gcc12-cxx14-asan:
-    runs-on: ubuntu-latest
-    steps:
-    - uses: actions/checkout@v4
-    - uses: actions/cache@v4
-      with:
-        path: bazel-disk
-        key: bazel-disk-gcc12-cxx14-asan-${{github.sha}}
-        restore-keys: bazel-disk-gcc12-cxx14-asan-
-    - run: bazel test --disk_cache=bazel-disk --action_env=CC=gcc-12 --cxxopt=-std=c++14 --cxxopt=-faligned-new --features=asan test/...
-
-  clang14-basic-asan:
-    runs-on: ubuntu-latest
-    steps:
-    - uses: actions/checkout@v4
-    - uses: actions/cache@v4
-      with:
-        path: bazel-disk
-        key: bazel-disk-clang14-basic-asan-${{github.sha}}
-        restore-keys: bazel-disk-clang14-basic-asan-
-    - run: bazel test --disk_cache=bazel-disk --action_env=CC=clang-14 --cxxopt=-stdlib=libc++ --linkopt=-stdlib=libc++ --features=asan test/...
-
-  clang14-basic-tsan:
-    runs-on: ubuntu-latest
-    steps:
-    - uses: actions/checkout@v4
-    - uses: actions/cache@v4
-      with:
-        path: bazel-disk
-        key: bazel-disk-clang14-basic-tsan-${{github.sha}}
-        restore-keys: bazel-disk-clang14-basic-tsan-
-    - run: bazel test --disk_cache=bazel-disk --action_env=CC=clang-14 --cxxopt=-stdlib=libc++ --linkopt=-stdlib=libc++ --features=tsan test/...
-
-  clang14-arenastring-asan:
-    runs-on: ubuntu-latest
-    steps:
-    - uses: actions/checkout@v4
-    - uses: actions/cache@v4
-      with:
-        path: bazel-disk
-        key: bazel-disk-clang14-arenastring-${{github.sha}}
-        restore-keys: bazel-disk-clang14-arenastring-
-    - run: sed -i "/single_version_override.*protobuf/s/version = '[^']*'/version = '25.3.arenastring'/" MODULE.bazel
-    - run: bazel test --disk_cache=bazel-disk --action_env=CC=clang-14 --cxxopt=-stdlib=libc++ --linkopt=-stdlib=libc++ --config=arenastring --features=asan test/...
-
-  clang14-mutable-donated-string-asan:
-    runs-on: ubuntu-latest
-    steps:
-    - uses: actions/checkout@v4
-    - uses: actions/cache@v4
-      with:
-        path: bazel-disk
-        key: bazel-disk-clang14-mutable-donated-string-${{github.sha}}
-        restore-keys: bazel-disk-clang14-mutable-donated-string-
-    - run: sed -i "/single_version_override.*protobuf/s/version = '[^']*'/version = '25.3.arenastring'/" MODULE.bazel
-    - run: bazel test --disk_cache=bazel-disk --action_env=CC=clang-14 --cxxopt=-stdlib=libc++ --linkopt=-stdlib=libc++ --config=mutable-donated-string --features=asan test/...
-
-  clang14-cxx14-asan:
-    runs-on: ubuntu-latest
-    steps:
-    - uses: actions/checkout@v4
-    - uses: actions/cache@v4
-      with:
-        path: bazel-disk
-        key: bazel-disk-clang14-cxx14-asan-${{github.sha}}
-        restore-keys: bazel-disk-clang14-cxx14-asan-
-    - run: bazel test --disk_cache=bazel-disk --action_env=CC=clang-14 --cxxopt=-stdlib=libc++ --linkopt=-stdlib=libc++ --cxxopt=-std=c++14 --cxxopt=-faligned-new --features=asan test/...
-
-  gcc12-aarch64-asan:
-    runs-on: ubuntu-latest
-    steps:
-    - uses: actions/checkout@v4
-    - uses: actions/cache@v4
-      with:
-        path: bazel-disk
-        key: bazel-disk-gcc12-aarch64-asan-${{github.sha}}
-        restore-keys: bazel-disk-gcc12-aarch64-asan-
+        key: bazel-disk-aarch64-${{github.sha}}
+        restore-keys: bazel-disk-aarch64-
+        save-always: true
     - run: sudo apt update
     - run: sudo apt install g++-12-aarch64-linux-gnu
     - run: sudo apt install qemu-user
     - run: sudo ln -s /usr/aarch64-linux-gnu/lib/ld-linux-aarch64.so.1 /lib/
-    - run: bazel test --disk_cache=bazel-disk --platforms='@cross_config_toolchain//:cross' --action_env=CROSS_CC=/usr/bin/aarch64-linux-gnu-gcc-12 --features=-default_link_flags --features=asan --linkopt=-fuse-ld=gold --test_env=LD_LIBRARY_PATH=/usr/aarch64-linux-gnu/lib --test_env=ASAN_OPTIONS=detect_odr_violation=1:detect_leaks=0 test/...
+    - run: bazel test --disk_cache=bazel-disk --verbose_failures --platforms='@cross_config_toolchain//:cross' --action_env=CROSS_CC=/usr/bin/aarch64-linux-gnu-gcc-12 --features=-default_link_flags --test_env=LD_LIBRARY_PATH=/usr/aarch64-linux-gnu/lib test/...
 
   workspace:
     runs-on: ubuntu-latest
@@ -145,7 +82,8 @@ jobs:
         path: bazel-disk
         key: bazel-disk-workspace-${{github.sha}}
         restore-keys: bazel-disk-workspace-
-    - run: bazel test --enable_bzlmod=false --disk_cache=bazel-disk --action_env=CC=gcc-12 --features=asan test/...
+        save-always: true
+    - run: bazel test --disk_cache=bazel-disk --enable_bzlmod=false test/...
 
   coverage:
     runs-on: ubuntu-latest
@@ -156,7 +94,8 @@ jobs:
         path: bazel-disk
         key: bazel-disk-coverage-${{github.sha}}
         restore-keys: bazel-disk-coverage-
-    - run: bazel coverage --disk_cache=bazel-disk --combined_report=lcov --instrumentation_filter='src/babylon,-src/babylon/reusable/patch' test/...
+        save-always: true
+    - run: bazel coverage --disk_cache=bazel-disk --verbose_failures --combined_report=lcov --instrumentation_filter='src/babylon,-src/babylon/reusable/patch' test/...
     - run: .github/workflows/filter_lcov.sh
     - uses: coverallsapp/github-action@v2
       with:

BUILD

@@ -75,6 +75,11 @@ alias(
   actual = '//src/babylon/concurrent:counter',
 )
 
+alias(
+  name = 'concurrent_execution_queue',
+  actual = '//src/babylon/concurrent:execution_queue',
+)
+
 alias(
   name = 'concurrent_id_allocator',
   actual = '//src/babylon/concurrent:id_allocator',

MODULE.bazel

@@ -25,7 +25,7 @@ bazel_dep(name = 'boost.spirit', version = '1.83.0.bzl.1')
 
 # --registry=https://bcr.bazel.build
 single_version_override(module_name = 'abseil-cpp', version = '20240116.2')
-single_version_override(module_name = 'protobuf', version = '27.2')
+single_version_override(module_name = 'protobuf', version = '27.3')
 bazel_dep(name = 'googletest', version = '1.14.0', repo_name = 'com_google_googletest', dev_dependency = True)
 bazel_dep(name = 'platforms', version = '0.0.10', dev_dependency = True)
 bazel_dep(name = 'rules_cc', version = '0.0.9', dev_dependency = True)

WORKSPACE.bzlmod

@@ -0,0 +1,47 @@
+# execution_queue
+
+## 原理
+
+包装[ConcurrentBoundedQueue](bounded_queue.md)，实现按需激活MPSC消费者的模式
+- 每次生产数据时对待处理计数器原子自增，如果自增前为0，边沿触发，启动消费者线程
+- 消费者线程持续消费数据，当无数据可消费时，将计数器值和0交换
+- 如果交换得到的值和最近一次消费前记录的值无变化，退出消费者线程
+- 否则，进入下一轮消费循环
+
+主要用于支持大量低活队列的情况，节省不活跃的监听消费线程，功能和[bthread::ExecutionQueue](https://github.com/apache/brpc/blob/master/docs/cn/execution_queue.md)类似，但是
+- 消费者通过[Executor](../executor.md)接口提交，支持使用自定义线程/协程机制
+- 保持生产者wait-free提交的同时，不会在竞争时引起消费者阻塞
+
+![](images/bthread_execution_queue.png)
+
+## 用法示例
+
+```c++
+#include "babylon/concurrent/execution_queue.h"
+
+using ::babylon::ConcurrentExecutionQueue;
+
+// 显式定义一个队列
+using Queue = ConcurrentExecutionQueue<T>;
+Queue queue;
+
+// 设置队列容量到N
+// 消费者采用some_executor执行
+// 注册lambda消费函数
+queue.initialize(N, some_executor, [] (Queue::Iterator iter, Queue::Iterator end) {
+  // 消费区间的数据
+  while (iter != end) {
+    T& item = *iter;
+    do_sth_with(item);
+    ++iter;
+  }
+});
+
+// 生产一个数据，并按需启动后台消费
+queue.execute("10086");
+...
+
+// 等待当前发布数据全部消费完成
+// 注意不包含stop语义，可以反复execute & join
+queue.join();
+```

docs/concurrent/execution_queue.md

@@ -4,6 +4,7 @@
 
 - [bounded_queue](bounded_queue.md)
 - [counter](counter.md)
+- [execution_queue](execution_queue.md)
 - [id_allocator](id_allocator.md)
 - [object_pool](object_pool.md)
 - [thread_local](thread_local.md)

docs/concurrent/images/bthread_execution_queue.png

@@ -0,0 +1,37 @@
+# TODO: migrate all dependencies from WORKSPACE to MODULE.bazel
+# https://github.com/protocolbuffers/protobuf/issues/14313
+module(
+    name = "protobuf",
+    version = "27.3.arenastring", # Automatically updated on release
+    compatibility_level = 1,
+    repo_name = "com_google_protobuf",
+)
+
+# LOWER BOUND dependency versions.
+# Bzlmod follows MVS:
+# https://bazel.build/versions/6.0.0/build/bzlmod#version-resolution
+# Thus the highest version in their module graph is resolved.
+bazel_dep(name = "abseil-cpp", version = "20230802.0.bcr.1", repo_name = "com_google_absl")
+bazel_dep(name = "bazel_skylib", version = "1.4.1")
+bazel_dep(name = "jsoncpp", version = "1.9.5")
+bazel_dep(name = "rules_cc", version = "0.0.9")
+bazel_dep(name = "rules_java", version = "5.3.5")
+bazel_dep(name = "rules_jvm_external", version = "5.1")
+bazel_dep(name = "rules_pkg", version = "0.7.0")
+bazel_dep(name = "rules_python", version = "0.10.2")
+bazel_dep(name = "platforms", version = "0.0.8")
+bazel_dep(name = "zlib", version = "1.2.11")
+
+# TODO: remove after toolchain types are moved to protobuf
+bazel_dep(name = "rules_proto", version = "4.0.0")
+
+bazel_dep(name = "googletest", version = "1.14.0", repo_name = "com_google_googletest", dev_dependency = True)
+bazel_dep(name = "rules_ruby", dev_dependency = True)
+archive_override(
+    module_name = "rules_ruby",
+    urls = ["https://github.com/protocolbuffers/rules_ruby/archive/b7f3e9756f3c45527be27bc38840d5a1ba690436.tar.gz"],
+    strip_prefix = "rules_ruby-b7f3e9756f3c45527be27bc38840d5a1ba690436",
+    integrity = "sha256-RNo21X/p9slOdF912FLprLnxvZHMXBTxaUDGFmm2bx8=",
+    patches = ["third_party/rules_ruby.patch"],
+    patch_strip = 1,
+)