refactor coroutine (#56)

* refactor coroutine

* add Cancelable

* add example/use-anyflow-with-brpc

.bazelrc

@@ -1,6 +1,5 @@
 common --registry=https://bcr.bazel.build
 common --registry=file://%workspace%/registry
-common --registry=https://raw.githubusercontent.com/bazelboost/registry/main
 
 build --cxxopt=-std=c++20
 

.bazelversion

@@ -1 +1 @@
-7.3.1
+7.3.2

.github/workflows/ci.yml

@@ -12,26 +12,34 @@ jobs:
       fail-fast: false
       matrix:
         compiler: [{name: gcc, flag: --action_env=CC=gcc-14}, {name: clang, flag: --action_env=CC=clang-18}]
-        std: [{name: c++20}, {name: c++14, flag: --cxxopt=-std=c++14 --cxxopt=-faligned-new}]
+        std: [{name: c++20}, {name: c++14, flag: --cxxopt=-std=c++14 --cxxopt=-faligned-new}, {name: c++14-coroutine, flag: --cxxopt=-faligned-new --cxxopt=-fconcepts --cxxopt=-fcoroutines}]
         stdlib: [{name: stdlibc++}, {name: libc++, flag: --cxxopt=-stdlib=libc++ --linkopt=-stdlib=libc++}]
         feature: [{name: asan, flag: --config=asan}, {name: tsan, flag: --features=tsan}]
         exclude:
         - compiler: {name: gcc}
           stdlib: {name: libc++}
+        - compiler: {name: clang}
+          std: {name: c++14-coroutine}
         - std: {name: c++14}
           feature: {name: tsan}
+        - std: {name: c++14-coroutine}
+          feature: {name: tsan}
     runs-on: ubuntu-24.04
     name: basic-${{matrix.compiler.name}}-${{matrix.std.name}}-${{matrix.stdlib.name}}-${{matrix.feature.name}}
     steps:
     - uses: actions/checkout@v4
-    - uses: actions/cache@v4
+    - uses: actions/cache/restore@v4
       with:
         path: bazel-disk
         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: sudo apt install libc++-18-dev libc++abi-18-dev
     - run: bazel test --compilation_mode=dbg --disk_cache=bazel-disk --verbose_failures --test_output=errors ${{matrix.compiler.flag}} ${{matrix.std.flag}} ${{matrix.stdlib.flag}} ${{matrix.feature.flag}} test/...
+    - uses: actions/cache/save@v4
+      if: always()
+      with:
+        path: bazel-disk
+        key: bazel-disk-basic-${{matrix.compiler.name}}-${{matrix.std.name}}-${{matrix.stdlib.name}}-${{matrix.feature.name}}-${{github.sha}}
 
   arenastring:
     strategy:
@@ -51,60 +59,76 @@ jobs:
     name: arenastring-${{matrix.compiler.name}}-${{matrix.stdlib.name}}-${{matrix.mutable.name}}
     steps:
     - uses: actions/checkout@v4
-    - uses: actions/cache@v4
+    - uses: actions/cache/restore@v4
       with:
         path: bazel-disk
         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: sudo apt install libc++-18-dev libc++abi-18-dev
     - run: sed -i "/single_version_override.*protobuf/s/version = '[^']*'/version = '27.5.arenastring'/" MODULE.bazel
     - run: bazel test --compilation_mode=opt --disk_cache=bazel-disk --verbose_failures --test_output=errors --config=asan ${{matrix.compiler.flag}} ${{matrix.stdlib.flag}} ${{matrix.mutable.flag}} test/...
+    - uses: actions/cache/save@v4
+      if: always()
+      with:
+        path: bazel-disk
+        key: bazel-disk-arenastring-${{matrix.compiler.name}}-${{matrix.stdlib.name}}-${{matrix.mutable.name}}-${{github.sha}}
 
   aarch64:
     runs-on: ubuntu-24.04
     steps:
     - uses: actions/checkout@v4
-    - uses: actions/cache@v4
+    - uses: actions/cache/restore@v4
       with:
         path: bazel-disk
         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 --verbose_failures --test_output=errors --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/... -test/logging:test_log_statically
+    - uses: actions/cache/save@v4
+      if: always()
+      with:
+        path: bazel-disk
+        key: bazel-disk-aarch64-${{github.sha}}
 
   workspace:
     runs-on: ubuntu-24.04
     steps:
     - uses: actions/checkout@v4
-    - uses: actions/cache@v4
+    - uses: actions/cache/restore@v4
       with:
         path: bazel-disk
         key: bazel-disk-workspace-${{github.sha}}
         restore-keys: bazel-disk-workspace-
-        save-always: true
     - run: bazel test --disk_cache=bazel-disk --verbose_failures --test_output=errors --enable_bzlmod=false test/...
+    - uses: actions/cache/save@v4
+      if: always()
+      with:
+        path: bazel-disk
+        key: bazel-disk-workspace-${{github.sha}}
 
   coverage:
     runs-on: ubuntu-24.04
     steps:
     - uses: actions/checkout@v4
-    - uses: actions/cache@v4
+    - uses: actions/cache/restore@v4
       with:
         path: bazel-disk
         key: bazel-disk-coverage-${{github.sha}}
         restore-keys: bazel-disk-coverage-
-        save-always: true
     - run: bazel coverage --disk_cache=bazel-disk --verbose_failures --test_output=errors --combined_report=lcov --instrumentation_filter='src/babylon,-src/babylon/reusable/patch' test/...
     - run: .github/workflows/filter_lcov.sh
     - uses: coverallsapp/github-action@v2
       with:
         github-token: ${{github.token}}
         file: bazel-out/_coverage/_coverage_report.lcov
+    - uses: actions/cache/save@v4
+      if: always()
+      with:
+        path: bazel-disk
+        key: bazel-disk-coverage-${{github.sha}}
 
   cmake:
     runs-on: ubuntu-24.04

.github/workflows/example.yml

@@ -15,6 +15,7 @@ jobs:
         - depend-use-cmake-find
         - depend-use-cmake-subdir
         - depend-use-workspace
+        - use-anyflow-with-brpc
         - use-arena-with-brpc
         - use-async-logger
         - use-counter-with-bvar

BUILD

@@ -75,6 +75,11 @@ alias(
   actual = '//src/babylon/concurrent:counter',
 )
 
+alias(
+  name = 'concurrent_deposit_box',
+  actual = '//src/babylon/concurrent:deposit_box',
+)
+
 alias(
   name = 'concurrent_epoch',
   actual = '//src/babylon/concurrent:epoch',
@@ -127,7 +132,7 @@ alias(
 
 alias(
   name = 'coroutine',
-  actual = '//src/babylon:coroutine',
+  actual = '//src/babylon/coroutine',
 )
 
 alias(

MODULE.bazel

@@ -13,11 +13,9 @@ module(
 # --registry=https://bcr.bazel.build
 bazel_dep(name = 'abseil-cpp', version = '20211102.0', repo_name = 'com_google_absl')
 bazel_dep(name = 'bazel_skylib', version = '1.0.3')
+bazel_dep(name = 'boost.preprocessor', version = '1.83.0')
+bazel_dep(name = 'boost.spirit', version = '1.83.0')
 bazel_dep(name = 'protobuf', version = '3.19.6', repo_name = 'com_google_protobuf')
-
-# --registry=https://raw.githubusercontent.com/bazelboost/registry/main
-bazel_dep(name = 'boost.preprocessor', version = '1.83.0.bzl.1')
-bazel_dep(name = 'boost.spirit', version = '1.83.0.bzl.2')
 ################################################################################
 
 ################################################################################

example/use-anyflow-with-brpc/.bazelrc

@@ -0,0 +1,5 @@
+common --registry=https://bcr.bazel.build
+common --registry=https://baidu.github.io/babylon/registry
+common --registry=https://raw.githubusercontent.com/bazelboost/registry/main
+
+build --cxxopt=-std=c++17

example/use-anyflow-with-brpc/.bazelversion

@@ -0,0 +1 @@
+7.3.1

example/use-anyflow-with-brpc/BUILD

@@ -0,0 +1,68 @@
+proto_library(
+  name = 'search_proto',
+  srcs = ['search.proto'],
+)
+
+cc_proto_library(
+  name = 'cc_search_proto',
+  deps = [':search_proto'],
+)
+
+cc_library(
+  name = 'butex_interface',
+  hdrs = ['butex_interface.h'],
+  deps = [
+    '@babylon',
+    '@brpc',
+  ],
+)
+
+cc_library(
+  name = 'bthread_graph_executor',
+  srcs = ['bthread_graph_executor.cpp'],
+  hdrs = ['bthread_graph_executor.h'],
+  deps = [
+    ':butex_interface',
+  ],
+)
+
+cc_library(
+  name = 'graph_configurator',
+  hdrs = ['graph_configurator.h'],
+  deps = [
+    ':bthread_graph_executor',
+    '@yaml-cpp',
+  ],
+)
+
+cc_library(
+  name = 'processors',
+  srcs = ['parse.cpp', 'rank.cpp', 'recall.cpp', 'response.cpp', 'user_profile.cpp'],
+  deps = [
+    ':cc_search_proto',
+    '@babylon',
+    '@yaml-cpp',
+  ],
+  # BABYLON_REGISTER_COMPONENT need alwayslink
+  alwayslink = True,
+)
+
+cc_binary(
+  name = 'client',
+  srcs = ['client.cpp'],
+  deps = [
+    ':cc_search_proto',
+    '@brpc',
+  ],
+)
+
+cc_binary(
+  name = 'server',
+  srcs = ['server.cpp'],
+  deps = [
+    ':cc_search_proto',
+    ':graph_configurator',
+    ':processors',
+    '@brpc',
+  ],
+)

example/use-anyflow-with-brpc/MODULE.bazel

@@ -0,0 +1,3 @@
+bazel_dep(name = 'babylon', version = '1.3.2')
+bazel_dep(name = 'brpc', version = '1.9.0')
+bazel_dep(name = 'yaml-cpp', version = '0.8.0')

example/use-anyflow-with-brpc/README.md

@@ -0,0 +1,10 @@
+# Use anyflow with brpc
+
+结合brpc，使用anyflow进行服务模块化的简单样例
+
+## 示例构成
+- `:graph_configurator`: anyflow本身只提供了基础API，一般业务实际场景需要结合自己所使用的配置机制，做一层包装，这里用yaml为例展示了这个包装的基础做法
+  - `dag.yaml`: 对应的配置文件
+- `:processors`: 模块化分割的搜索服务实现拆分样例，采用babylon::ApplicationContext的注册机制管理组件构造
+  - 包含['parse.cpp', 'rank.cpp', 'recall.cpp', 'response.cpp', 'user_profile.cpp']
+- `:server`: 顶层服务封装，只包含初始化`graph_configurator`，以及将service的request和response注入到graph内作为初始节点

example/use-anyflow-with-brpc/bthread_graph_executor.cpp

@@ -0,0 +1,70 @@
+#include "bthread_graph_executor.h"
+
+#include "babylon/logging/logger.h"
+
+#include "butex_interface.h"
+
+#include <tuple>
+
+BABYLON_NAMESPACE_BEGIN
+namespace anyflow {
+
+static void* execute_invoke_vertex(void* args) {
+  auto param =
+      reinterpret_cast<::std::tuple<GraphVertex*, GraphVertexClosure>*>(args);
+  auto vertex = ::std::get<0>(*param);
+  auto& closure = ::std::get<1>(*param);
+  vertex->run(::std::move(closure));
+  delete param;
+  return NULL;
+}
+
+static void* execute_invoke_closure(void* args) {
+  auto param =
+      reinterpret_cast<::std::tuple<ClosureContext*, Closure::Callback*>*>(
+          args);
+  auto closure = ::std::get<0>(*param);
+  auto callback = ::std::get<1>(*param);
+  closure->run(callback);
+  delete param;
+  return NULL;
+}
+
+BthreadGraphExecutor& BthreadGraphExecutor::instance() {
+  static BthreadGraphExecutor executor;
+  return executor;
+}
+
+Closure BthreadGraphExecutor::create_closure() noexcept {
+  return Closure::create<ButexInterface>(*this);
+}
+
+int BthreadGraphExecutor::run(GraphVertex* vertex,
+                              GraphVertexClosure&& closure) noexcept {
+  bthread_t th;
+  auto param = new ::std::tuple<GraphVertex*, GraphVertexClosure>(
+      vertex, ::std::move(closure));
+  if (0 != bthread_start_background(&th, NULL, execute_invoke_vertex, param)) {
+    LOG(WARNING) << "start bthread to run vertex failed";
+    closure = ::std::move(::std::get<1>(*param));
+    delete param;
+    return -1;
+  }
+  return 0;
+}
+
+int BthreadGraphExecutor::run(ClosureContext* closure,
+                              Closure::Callback* callback) noexcept {
+  bthread_t th;
+  auto param =
+      new ::std::tuple<ClosureContext*, Closure::Callback*>(closure, callback);
+  if (0 != bthread_start_background(&th, NULL, execute_invoke_closure, param)) {
+    LOG(WARNING) << "start bthread to run closure failed";
+    delete param;
+    return -1;
+  }
+  return 0;
+}
+
+} // namespace anyflow
+BABYLON_NAMESPACE_END

example/use-anyflow-with-brpc/bthread_graph_executor.h

@@ -0,0 +1,21 @@
+#include "babylon/anyflow/closure.h"
+#include "babylon/anyflow/executor.h"
+#include "babylon/anyflow/vertex.h"
+
+#include "bthread/bthread.h"
+
+BABYLON_NAMESPACE_BEGIN
+namespace anyflow {
+
+class BthreadGraphExecutor : public GraphExecutor {
+ public:
+  static BthreadGraphExecutor& instance();
+  virtual Closure create_closure() noexcept override;
+  virtual int run(GraphVertex* vertex,
+                  GraphVertexClosure&& closure) noexcept override;
+  virtual int run(ClosureContext* closure,
+                  Closure::Callback* callback) noexcept override;
+};
+
+} // namespace anyflow
+BABYLON_NAMESPACE_END

example/use-anyflow-with-brpc/build.sh

@@ -0,0 +1,4 @@
+#!/bin/sh
+set -ex
+
+bazel build server client