diff --git a/.github/actions/init-ut-make-config/action.yml b/.github/actions/init-ut-make-config/action.yml index 9df8103652..8f770845f7 100644 --- a/.github/actions/init-ut-make-config/action.yml +++ b/.github/actions/init-ut-make-config/action.yml @@ -5,9 +5,6 @@ inputs: runs: using: "composite" steps: - - run: | - sudo apt-get update && sudo apt-get install -y clang-12 lldb-12 lld-12 libgtest-dev cmake gdb libstdc++6-11-dbg - shell: bash - run: | cd /usr/src/gtest && export CC=clang-12 && export CXX=clang++-12 && sudo cmake -DCMAKE_POLICY_VERSION_MINIMUM=3.5 . sudo make -j ${{env.proc_num}} && sudo mv lib/libgtest* /usr/lib/ diff --git a/.github/actions/install-all-dependencies/action.yml b/.github/actions/install-all-dependencies/action.yml index 5c1f673ff7..e104d10d21 100644 --- a/.github/actions/install-all-dependencies/action.yml +++ b/.github/actions/install-all-dependencies/action.yml @@ -2,11 +2,13 @@ runs: using: "composite" steps: - uses: ./.github/actions/install-essential-dependencies - - run: sudo apt-get update && sudo apt-get install -y libunwind-dev libgoogle-glog-dev automake bison flex libboost-all-dev libevent-dev libtool pkg-config libibverbs-dev - shell: bash + with: + extra-packages: >- + ccache libunwind-dev libgoogle-glog-dev automake bison flex + libboost-all-dev libevent-dev libtool pkg-config - run: | wget https://archive.apache.org/dist/thrift/0.11.0/thrift-0.11.0.tar.gz && tar -xf thrift-0.11.0.tar.gz && cd thrift-0.11.0/ - ./configure --prefix=/usr --with-rs=no --with-ruby=no --with-python=no --with-java=no --with-go=no --with-perl=no --with-php=no --with-csharp=no --with-erlang=no --with-lua=no --with-nodejs=no --with-haskell=no --with-dotnetcore=no CXXFLAGS="-Wno-unused-variable" + ./configure --prefix=/usr --disable-tests --with-rs=no --with-ruby=no --with-python=no --with-java=no --with-go=no --with-perl=no --with-php=no --with-csharp=no --with-erlang=no --with-lua=no --with-nodejs=no --with-haskell=no --with-dotnetcore=no CXXFLAGS="-Wno-unused-variable" make -j ${{env.proc_num}} && sudo make install shell: bash - run: | diff --git a/.github/actions/install-essential-dependencies/action.yml b/.github/actions/install-essential-dependencies/action.yml index 5c1944d68e..a9802337c8 100644 --- a/.github/actions/install-essential-dependencies/action.yml +++ b/.github/actions/install-essential-dependencies/action.yml @@ -1,9 +1,18 @@ +inputs: + extra-packages: + description: Additional apt packages to install in the same transaction + required: false + runs: using: "composite" steps: - run: ulimit -c unlimited -S && sudo bash -c "echo 'core.%e.%p' > /proc/sys/kernel/core_pattern" shell: bash - - run: sudo apt-get update && sudo apt-get install -y git g++ make libssl-dev libgflags-dev libprotobuf-dev libprotoc-dev protobuf-compiler libleveldb-dev redis-server mysql-server libibverbs-dev + - run: | + sudo apt-get update + sudo apt-get install -y git g++ make libssl-dev libgflags-dev \ + libprotobuf-dev libprotoc-dev protobuf-compiler libleveldb-dev \ + redis-server mysql-server libibverbs-dev ${{ inputs.extra-packages }} shell: bash - run: redis-server --version && mysqld --version shell: bash diff --git a/.github/actions/setup-build-cache/action.yml b/.github/actions/setup-build-cache/action.yml new file mode 100644 index 0000000000..6045ec891d --- /dev/null +++ b/.github/actions/setup-build-cache/action.yml @@ -0,0 +1,75 @@ +name: Setup build cache +description: Restore and configure the compiler cache used by CI builds + +inputs: + kind: + description: Cache backend to configure (ccache or bazel) + required: true + cache-key: + description: Cache namespace override for matrix jobs + required: false + +runs: + using: composite + steps: + - name: Restore ccache + if: inputs.kind == 'ccache' + uses: actions/cache@v6 + with: + path: ~/.cache/ccache + key: ${{ runner.os }}-ccache-${{ inputs.cache-key || github.job }}--${{ github.sha }} + restore-keys: | + ${{ runner.os }}-ccache-${{ inputs.cache-key || github.job }}-- + + - name: Configure ccache + if: inputs.kind == 'ccache' + shell: bash + run: | + if [[ "${{ runner.os }}" == "macOS" ]]; then + brew install ccache + fi + + mkdir -p "${HOME}/.cache/ccache" "${HOME}/.cache/ccache-bin" + for compiler in cc c++ gcc g++ clang clang++ clang-12 clang++-12; do + ln -sf "$(command -v ccache)" "${HOME}/.cache/ccache-bin/${compiler}" + done + CCACHE_DIR="${HOME}/.cache/ccache" ccache --max-size=2G + CCACHE_DIR="${HOME}/.cache/ccache" ccache --zero-stats + + echo "${HOME}/.cache/ccache-bin" >> "${GITHUB_PATH}" + echo "CCACHE_DIR=${HOME}/.cache/ccache" >> "${GITHUB_ENV}" + echo "CCACHE_BASEDIR=${GITHUB_WORKSPACE}" >> "${GITHUB_ENV}" + echo "CCACHE_COMPILERCHECK=content" >> "${GITHUB_ENV}" + echo "CCACHE_NOHASHDIR=true" >> "${GITHUB_ENV}" + + - name: Restore Bazel repository cache + if: inputs.kind == 'bazel' + uses: actions/cache@v6 + with: + path: | + ~/.cache/bazel-repository + ~/.cache/bazelisk + key: ${{ runner.os }}-bazel-repository-${{ hashFiles('.bazelversion', 'MODULE.bazel', 'MODULE.bazel.lock', 'WORKSPACE') }} + restore-keys: | + ${{ runner.os }}-bazel-repository- + + - name: Restore Bazel disk cache + if: inputs.kind == 'bazel' + uses: actions/cache@v6 + with: + path: ~/.cache/bazel-disk + key: ${{ runner.os }}-bazel-disk-${{ github.job }}-${{ github.sha }} + restore-keys: | + ${{ runner.os }}-bazel-disk-${{ github.job }}- + ${{ runner.os }}-bazel-disk- + + - name: Configure Bazel caches + if: inputs.kind == 'bazel' + shell: bash + run: | + mkdir -p "${HOME}/.cache/bazel-repository" "${HOME}/.cache/bazel-disk" + cat >> "${HOME}/.bazelrc" <- + --headers=/usr/include --libs=/usr/lib /usr/lib64 + --cc=gcc --cxx=g++ --werror + - name: gcc-all-options + make-options: >- + --headers=/usr/include --libs=/usr/lib /usr/lib64 + --cc=gcc --cxx=g++ --werror --with-thrift --with-glog + --with-rdma --with-debug-bthread-sche-safety --with-debug-lock + --with-bthread-tracer --with-asan + - name: clang-default + make-options: >- + --headers=/usr/include --libs=/usr/lib /usr/lib64 + --cc=clang --cxx=clang++ --werror + - name: clang-all-options + make-options: >- + --headers=/usr/include --libs=/usr/lib /usr/lib64 + --cc=clang --cxx=clang++ --werror --with-thrift --with-glog + --with-rdma --with-debug-bthread-sche-safety --with-debug-lock + --with-bthread-tracer --with-asan steps: - - uses: actions/checkout@v2 + - uses: actions/checkout@v7.0.1 - uses: ./.github/actions/install-all-dependencies - - - name: gcc with default options - uses: ./.github/actions/compile-with-make - with: - options: --headers=/usr/include --libs=/usr/lib /usr/lib64 --cc=gcc --cxx=g++ --werror - - - name: gcc with all options - uses: ./.github/actions/compile-with-make + - uses: ./.github/actions/setup-build-cache with: - options: --headers=/usr/include --libs=/usr/lib /usr/lib64 --cc=gcc --cxx=g++ --werror \ - --with-thrift --with-glog --with-rdma --with-debug-bthread-sche-safety \ - --with-debug-lock --with-bthread-tracer --with-asan + kind: ccache + cache-key: make-${{ matrix.name }} - - name: clang with default options + - name: Build uses: ./.github/actions/compile-with-make with: - options: --headers=/usr/include --libs=/usr/lib /usr/lib64 --cc=clang --cxx=clang++ --werror + options: ${{ matrix.make-options }} - - name: clang with all options - uses: ./.github/actions/compile-with-make - with: - options: --headers=/usr/include --libs=/usr/lib /usr/lib64 --cc=clang --cxx=clang++ --werror \ - --with-thrift --with-glog --with-rdma --with-debug-bthread-sche-safety \ - --with-debug-lock --with-bthread-tracer --with-asan + - name: Show ccache statistics + if: always() + run: ccache --show-stats compile-with-cmake: + name: compile-with-cmake (${{ matrix.name }}) runs-on: ubuntu-22.04 + strategy: + fail-fast: false + max-parallel: 2 + matrix: + include: + - name: gcc-default + cc: gcc + cxx: g++ + cmake-options: '' + - name: gcc-all-options + cc: gcc + cxx: g++ + cmake-options: >- + -DWITH_MESALINK=OFF -DWITH_GLOG=ON -DWITH_THRIFT=ON + -DWITH_RDMA=ON -DWITH_UBRING=ON + -DWITH_DEBUG_BTHREAD_SCHE_SAFETY=ON -DWITH_DEBUG_LOCK=ON + -DWITH_BTHREAD_TRACER=ON -DWITH_ASAN=ON + - name: clang-default + cc: clang + cxx: clang++ + cmake-options: '' + - name: clang-all-options + cc: clang + cxx: clang++ + cmake-options: >- + -DWITH_MESALINK=OFF -DWITH_GLOG=ON -DWITH_THRIFT=ON + -DWITH_RDMA=ON -DWITH_UBRING=ON + -DWITH_DEBUG_BTHREAD_SCHE_SAFETY=ON -DWITH_DEBUG_LOCK=ON + -DWITH_BTHREAD_TRACER=ON -DWITH_ASAN=ON steps: - - uses: actions/checkout@v2 + - uses: actions/checkout@v7.0.1 - uses: ./.github/actions/install-all-dependencies + - uses: ./.github/actions/setup-build-cache + with: + kind: ccache + cache-key: cmake-${{ matrix.name }} - - name: gcc with default options - run: | - export CC=gcc && export CXX=g++ - mkdir gcc_build && cd gcc_build && cmake -DCMAKE_POLICY_VERSION_MINIMUM=3.5 .. - make -j ${{env.proc_num}} && make clean - - - name: gcc with all options - run: | - export CC=gcc && export CXX=g++ - mkdir gcc_build_all && cd gcc_build_all - cmake -DWITH_MESALINK=OFF -DWITH_GLOG=ON -DWITH_THRIFT=ON -DWITH_RDMA=ON -DWITH_UBRING=ON \ - -DWITH_DEBUG_BTHREAD_SCHE_SAFETY=ON -DWITH_DEBUG_LOCK=ON -DWITH_BTHREAD_TRACER=ON \ - -DWITH_ASAN=ON -DCMAKE_POLICY_VERSION_MINIMUM=3.5 .. - make -j ${{env.proc_num}} && make clean - - - name: clang with default options + - name: Build + env: + CC: ${{ matrix.cc }} + CXX: ${{ matrix.cxx }} + CMAKE_OPTIONS: ${{ matrix.cmake-options }} run: | - export CC=clang && export CXX=clang++ - mkdir clang_build && cd clang_build && cmake -DCMAKE_POLICY_VERSION_MINIMUM=3.5 .. - make -j ${{env.proc_num}} && make clean + read -r -a cmake_options <<< "${CMAKE_OPTIONS}" + cmake -S . -B build -DCMAKE_POLICY_VERSION_MINIMUM=3.5 \ + "${cmake_options[@]}" + cmake --build build -j ${{env.proc_num}} - - name: clang with all options - run: | - export CC=clang && export CXX=clang++ - mkdir clang_build_all && cd clang_build_all - cmake -DWITH_MESALINK=OFF -DWITH_GLOG=ON -DWITH_THRIFT=ON -DWITH_RDMA=ON -DWITH_UBRING=ON \ - -DWITH_DEBUG_BTHREAD_SCHE_SAFETY=ON -DWITH_DEBUG_LOCK=ON -DWITH_BTHREAD_TRACER=ON \ - -DWITH_ASAN=ON -DCMAKE_POLICY_VERSION_MINIMUM=3.5 .. - make -j ${{env.proc_num}} && make clean + - name: Show ccache statistics + if: always() + run: ccache --show-stats - gcc-compile-with-make-protobuf: + compile-with-make-protobuf: + name: compile-with-make-protobuf (${{ matrix.compiler }}-${{ matrix.name }}) runs-on: ubuntu-22.04 + strategy: + fail-fast: false + max-parallel: 2 + matrix: + include: + - compiler: gcc + name: protobuf-3.5.1 + protobuf-version: 3.5.1 + protobuf-cpp-version: 3.5.1 + protobuf-install-dir: /protobuf-3.5.1 + cc: gcc + cxx: g++ + - compiler: gcc + name: protobuf-3.12.4 + protobuf-version: 3.12.4 + protobuf-cpp-version: 3.12.4 + protobuf-install-dir: /protobuf-3.12.4 + cc: gcc + cxx: g++ + - compiler: gcc + name: protobuf-21.12 + protobuf-version: '21.12' + protobuf-cpp-version: 3.21.12 + protobuf-install-dir: /protobuf-3.21.12 + cc: gcc + cxx: g++ + - compiler: clang + name: protobuf-3.5.1 + protobuf-version: 3.5.1 + protobuf-cpp-version: 3.5.1 + protobuf-install-dir: /protobuf-3.5.1 + cc: clang + cxx: clang++ + - compiler: clang + name: protobuf-3.12.4 + protobuf-version: 3.12.4 + protobuf-cpp-version: 3.12.4 + protobuf-install-dir: /protobuf-3.12.4 + cc: clang + cxx: clang++ + - compiler: clang + name: protobuf-21.12 + protobuf-version: '21.12' + protobuf-cpp-version: 3.21.12 + protobuf-install-dir: /protobuf-3.21.12 + cc: clang + cxx: clang++ steps: - - uses: actions/checkout@v2 + - uses: actions/checkout@v7.0.1 - uses: ./.github/actions/install-essential-dependencies - - - name: protobuf 3.5.1 - uses: ./.github/actions/compile-with-make-protobuf with: - protobuf-version: 3.5.1 - protobuf-cpp-version: 3.5.1 - protobuf-install-dir: /protobuf-3.5.1 - config-brpc-options: --cc=gcc --cxx=g++ --werror - - - name: protobuf 3.12.4 - uses: ./.github/actions/compile-with-make-protobuf + extra-packages: ccache + - uses: ./.github/actions/setup-build-cache with: - protobuf-version: 3.12.4 - protobuf-cpp-version: 3.12.4 - protobuf-install-dir: /protobuf-3.12.4 - config-brpc-options: --cc=gcc --cxx=g++ --werror + kind: ccache + cache-key: ${{ matrix.compiler }}-make-${{ matrix.name }} - - name: protobuf 21.12 + - name: Build uses: ./.github/actions/compile-with-make-protobuf with: - protobuf-version: 21.12 - protobuf-cpp-version: 3.21.12 - protobuf-install-dir: /protobuf-3.21.12 - config-brpc-options: --cc=gcc --cxx=g++ --werror + protobuf-version: ${{ matrix.protobuf-version }} + protobuf-cpp-version: ${{ matrix.protobuf-cpp-version }} + protobuf-install-dir: ${{ matrix.protobuf-install-dir }} + config-brpc-options: >- + --cc=${{ matrix.cc }} --cxx=${{ matrix.cxx }} --werror + + - name: Show ccache statistics + if: always() + run: ccache --show-stats gcc-unittest-with-bazel: runs-on: ubuntu-22.04 steps: - - uses: actions/checkout@v2 + - uses: actions/checkout@v7.0.1 + - uses: ./.github/actions/setup-build-cache + with: + kind: bazel # Install redis-server/mysql-server so the integration tests that fork a # real server (e.g. brpc_redis_unittest) actually run under bazel instead # of skipping. Same shared action the make-based unittest jobs use. @@ -128,7 +208,10 @@ jobs: gcc-compile-with-bazel-all-options: runs-on: ubuntu-22.04 steps: - - uses: actions/checkout@v2 + - uses: actions/checkout@v7.0.1 + - uses: ./.github/actions/setup-build-cache + with: + kind: bazel - run: sudo apt-get update && sudo apt-get install -y libibverbs-dev - name: root run: | @@ -162,40 +245,13 @@ jobs: --define with_babylon_counter=true \ -- //... - clang-compile-with-make-protobuf: - runs-on: ubuntu-22.04 - steps: - - uses: actions/checkout@v2 - - uses: ./.github/actions/install-essential-dependencies - - - name: protobuf 3.5.1 - uses: ./.github/actions/compile-with-make-protobuf - with: - protobuf-version: 3.5.1 - protobuf-cpp-version: 3.5.1 - protobuf-install-dir: /protobuf-3.5.1 - config-brpc-options: --cc=clang --cxx=clang++ --werror - - - name: protobuf 3.12.4 - uses: ./.github/actions/compile-with-make-protobuf - with: - protobuf-version: 3.12.4 - protobuf-cpp-version: 3.12.4 - protobuf-install-dir: /protobuf-3.12.4 - config-brpc-options: --cc=clang --cxx=clang++ --werror - - - name: protobuf 21.12 - uses: ./.github/actions/compile-with-make-protobuf - with: - protobuf-version: 21.12 - protobuf-cpp-version: 3.21.12 - protobuf-install-dir: /protobuf-3.21.12 - config-brpc-options: --cc=clang --cxx=clang++ --werror - clang-unittest-with-bazel: runs-on: ubuntu-22.04 steps: - - uses: actions/checkout@v2 + - uses: actions/checkout@v7.0.1 + - uses: ./.github/actions/setup-build-cache + with: + kind: bazel # Install redis-server/mysql-server so the forked-server integration tests # actually run under bazel (see gcc-unittest-with-bazel). - uses: ./.github/actions/install-essential-dependencies @@ -209,7 +265,10 @@ jobs: clang-compile-with-bazel-all-options: runs-on: ubuntu-22.04 steps: - - uses: actions/checkout@v2 + - uses: actions/checkout@v7.0.1 + - uses: ./.github/actions/setup-build-cache + with: + kind: bazel - run: sudo apt-get update && sudo apt-get install -y libibverbs-dev - name: root run: | @@ -248,8 +307,13 @@ jobs: clang-unittest: runs-on: ubuntu-22.04 steps: - - uses: actions/checkout@v2 + - uses: actions/checkout@v7.0.1 - uses: ./.github/actions/install-essential-dependencies + with: + extra-packages: ccache clang-12 lldb-12 lld-12 libgtest-dev cmake gdb libstdc++6-11-dbg + - uses: ./.github/actions/setup-build-cache + with: + kind: ccache - uses: ./.github/actions/init-ut-make-config with: options: --with-bthread-tracer --with-rdma @@ -262,12 +326,20 @@ jobs: run: | cd test sh ./run_tests.sh + - name: Show ccache statistics + if: always() + run: ccache --show-stats clang-unittest-asan: runs-on: ubuntu-22.04 steps: - - uses: actions/checkout@v2 + - uses: actions/checkout@v7.0.1 - uses: ./.github/actions/install-essential-dependencies + with: + extra-packages: ccache clang-12 lldb-12 lld-12 libgtest-dev cmake gdb libstdc++6-11-dbg + - uses: ./.github/actions/setup-build-cache + with: + kind: ccache - uses: ./.github/actions/init-ut-make-config with: options: --with-bthread-tracer --with-asan @@ -284,6 +356,9 @@ jobs: # too slowly, so they flake here (connection refused). Skip just those under ASan; the # redis codec/server tests still run, and the full suite runs in clang-unittest. GTEST_FILTER='-RedisTest.sanity:RedisTest.keys_with_spaces:RedisTest.incr_and_decr:RedisTest.by_components:RedisTest.auth' sh ./run_tests.sh + - name: Show ccache statistics + if: always() + run: ccache --show-stats clang-unittest-bazel-with-babylon-and-new-pb: runs-on: ubuntu-22.04 @@ -294,7 +369,10 @@ jobs: # honors USE_BAZEL_VERSION. USE_BAZEL_VERSION: "8.3.1" steps: - - uses: actions/checkout@v2 + - uses: actions/checkout@v7.0.1 + - uses: ./.github/actions/setup-build-cache + with: + kind: bazel # Install redis-server/mysql-server so the forked-server integration tests # actually run under bazel (see gcc-unittest-with-bazel). - uses: ./.github/actions/install-essential-dependencies diff --git a/.github/workflows/ci-macos.yml b/.github/workflows/ci-macos.yml index 36424708ca..0b595e27c0 100644 --- a/.github/workflows/ci-macos.yml +++ b/.github/workflows/ci-macos.yml @@ -22,7 +22,10 @@ jobs: runs-on: macos-latest # https://github.com/actions/runner-images steps: - - uses: actions/checkout@v2 + - uses: actions/checkout@v7.0.1 + - uses: ./.github/actions/setup-build-cache + with: + kind: ccache - name: install dependences run: | @@ -41,11 +44,18 @@ jobs: mkdir build && cd build && cmake -DCMAKE_POLICY_VERSION_MINIMUM=3.5 -DWITH_UBRING=ON -DCMAKE_PREFIX_PATH=$(brew --prefix protobuf@21) .. make -j ${{env.proc_num}} && make clean + - name: Show ccache statistics + if: always() + run: ccache --show-stats + compile-with-make-cmake-protobuf29: runs-on: macos-latest # https://github.com/actions/runner-images steps: - - uses: actions/checkout@v2 + - uses: actions/checkout@v7.0.1 + - uses: ./.github/actions/setup-build-cache + with: + kind: ccache - name: install dependences run: | @@ -63,8 +73,15 @@ jobs: mkdir build && cd build && cmake -DCMAKE_POLICY_VERSION_MINIMUM=3.5 -DWITH_UBRING=ON -DCMAKE_PREFIX_PATH=$(brew --prefix protobuf@29) .. make -j ${{env.proc_num}} && make clean + - name: Show ccache statistics + if: always() + run: ccache --show-stats + compile-with-bazel: runs-on: macos-latest # https://github.com/actions/runner-images steps: - - uses: actions/checkout@v2 + - uses: actions/checkout@v7.0.1 + - uses: ./.github/actions/setup-build-cache + with: + kind: bazel - run: bazel build --verbose_failures -- //:brpc -//example/... diff --git a/.github/workflows/license-eyes.yml b/.github/workflows/license-eyes.yml index 3969942453..4ecba6a2b1 100644 --- a/.github/workflows/license-eyes.yml +++ b/.github/workflows/license-eyes.yml @@ -33,7 +33,7 @@ jobs: runs-on: ubuntu-latest steps: - name: "Checkout ${{ github.ref }} ( ${{ github.sha }} )" - uses: actions/checkout@v2 + uses: actions/checkout@v7.0.1 - name: Check License uses: apache/skywalking-eyes@v0.4.0 env: diff --git a/CMakeLists.txt b/CMakeLists.txt index 9419ac3a76..93113b6fc4 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -227,6 +227,7 @@ if(CMAKE_CXX_COMPILER_ID STREQUAL "GNU") endif() find_package(Protobuf REQUIRED) +find_package(ZLIB REQUIRED) if(Protobuf_VERSION VERSION_GREATER 4.21) # required by absl set(BRPC_CXX_STANDARD 17) @@ -351,7 +352,7 @@ set(DYNAMIC_LIB ${CMAKE_THREAD_LIBS_INIT} ${THRIFT_LIB} dl - z) + ZLIB::ZLIB) if(WITH_BORINGSSL) list(APPEND DYNAMIC_LIB ${BORINGSSL_SSL_LIBRARY}) diff --git a/docs/cn/client.md b/docs/cn/client.md index a353114734..e659403c24 100755 --- a/docs/cn/client.md +++ b/docs/cn/client.md @@ -7,6 +7,7 @@ Echo的[client端代码](https://github.com/apache/brpc/blob/master/example/echo # 事实速查 - Channel.Init()是线程不安全的。 +- 一个Channel只能成功初始化一次。Init()失败后可以重试。 - Channel.CallMethod()是线程安全的，一个Channel可以被所有线程同时使用。 - Channel可以分配在栈上。 - Channel在发送异步请求后可以析构。 @@ -32,6 +33,8 @@ channel.Init(..., &options); ``` 注意Channel不会修改options，Init结束后不会再访问options。所以options一般就像上面代码中那样放栈上。Channel.options()可以获得channel在使用的所有选项。 +Init失败后可以重试；一旦成功，Channel的目标和选项即固定，之后的所有Init调用都会返回-1。需要使用不同的目标或配置时，请新建一个Channel。 + Init函数分为连接一台服务器和连接服务集群。 # 连接一台服务器 diff --git a/docs/en/client.md b/docs/en/client.md index 087c39b53c..266eb90289 100644 --- a/docs/en/client.md +++ b/docs/en/client.md @@ -7,6 +7,8 @@ # Quick facts - Channel.Init() is not thread-safe. +- A Channel can be initialized successfully only once. Failed Init() calls may + be retried. - Channel.CallMethod() is thread-safe and a Channel can be used by multiple threads simultaneously. - Channel can be put on stack. - Channel can be destructed just after sending asynchronous request. @@ -32,6 +34,10 @@ channel.Init(..., &options); ``` Note that Channel neither modifies `options` nor accesses `options` after completion of Init(), thus options can be put on stack safely as in above code. Channel.options() gets options being used by the Channel. +Init() may be retried after a failure. Once it succeeds, the Channel's target +and options are fixed and every later Init() call returns -1. Create a new +Channel to use a different target or configuration. + Init() can connect one server or a cluster(multiple servers). # Connect to a server diff --git a/src/brpc/channel.cpp b/src/brpc/channel.cpp index 01d8637564..ff81e521d5 100644 --- a/src/brpc/channel.cpp +++ b/src/brpc/channel.cpp @@ -254,6 +254,10 @@ int Channel::InitChannelOptions(const ChannelOptions* options) { int Channel::Init(const char* server_addr_and_port, const ChannelOptions* options) { + if (_server_id != INVALID_SOCKET_ID || _lb != NULL) { + LOG(ERROR) << "Channel=" << this << " has already been initialized"; + return -1; + } GlobalInitializeOrDie(); butil::EndPoint point; const AdaptiveProtocolType& ptype = (options ? options->protocol : _options.protocol); @@ -287,6 +291,10 @@ int Channel::Init(const char* server_addr_and_port, int Channel::Init(const char* server_addr, int port, const ChannelOptions* options) { + if (_server_id != INVALID_SOCKET_ID || _lb != NULL) { + LOG(ERROR) << "Channel=" << this << " has already been initialized"; + return -1; + } GlobalInitializeOrDie(); butil::EndPoint point; const AdaptiveProtocolType& ptype = (options ? options->protocol : _options.protocol); @@ -358,6 +366,10 @@ int Channel::InitSingle(const butil::EndPoint& server_addr_and_port, const char* raw_server_address, const ChannelOptions* options, int raw_port) { + if (_server_id != INVALID_SOCKET_ID || _lb != NULL) { + LOG(ERROR) << "Channel=" << this << " has already been initialized"; + return -1; + } GlobalInitializeOrDie(); if (InitChannelOptions(options) != 0) { return -1; @@ -410,6 +422,10 @@ int Channel::Init(const char* ns_url, // Treat ns_url as server_addr_and_port return Init(ns_url, options); } + if (_server_id != INVALID_SOCKET_ID || _lb != NULL) { + LOG(ERROR) << "Channel=" << this << " has already been initialized"; + return -1; + } GlobalInitializeOrDie(); if (InitChannelOptions(options) != 0) { return -1; diff --git a/src/brpc/channel.h b/src/brpc/channel.h index d13ae52df6..47f262f627 100644 --- a/src/brpc/channel.h +++ b/src/brpc/channel.h @@ -184,6 +184,9 @@ friend class SelectiveChannel; DISALLOW_COPY_AND_ASSIGN(Channel); + // Init() may be retried after failure, but a successful initialization is + // final: subsequent calls return -1. + // Connect this channel to a single server whose address is given by the // first parameter. Use default options if `options' is nullptr. int Init(butil::EndPoint server_addr_and_port, const ChannelOptions* options); diff --git a/src/brpc/controller.cpp b/src/brpc/controller.cpp index 629332f48a..4fff9fd2f4 100644 --- a/src/brpc/controller.cpp +++ b/src/brpc/controller.cpp @@ -1705,12 +1705,20 @@ void Controller::HandleStreamConnection(Socket *host_socket) { return; } size_t stream_num = _request_streams.size(); + const size_t expected_extra_streams = stream_num - 1; std::vector ptrs(stream_num); if (!FailedInline()) { if (_remote_stream_settings == nullptr) { if (!FailedInline()) { SetFailed(EREQUEST, "The server didn't accept the stream"); } + } else if (static_cast( + _remote_stream_settings->extra_stream_ids_size()) != + expected_extra_streams) { + SetFailed(ERESPONSE, "Server returned %d extra_stream_ids, " + "expected %zu", + _remote_stream_settings->extra_stream_ids_size(), + expected_extra_streams); } else { for (size_t i = 0; i < stream_num; ++i) { if (Stream::Address(_request_streams[i], &ptrs[i]) != 0) { diff --git a/src/brpc/stream.cpp b/src/brpc/stream.cpp index 8db9352320..7e032b690b 100644 --- a/src/brpc/stream.cpp +++ b/src/brpc/stream.cpp @@ -35,6 +35,7 @@ namespace brpc { DECLARE_bool(usercode_in_pthread); +DECLARE_uint64(max_body_size); DECLARE_int64(socket_max_streams_unconsumed_bytes); DEFINE_uint64(stream_write_max_segment_size, 512 * 1024 * 1024, "Stream message exceeding this size will be automatically split into smaller segments"); @@ -610,6 +611,19 @@ int Stream::OnReceived(const StreamFrameMeta& fm, butil::IOBuf *buf, Socket* soc CHECK(buf->empty()); break; case FRAME_TYPE_DATA: + if (buf->length() > FLAGS_max_body_size || + (_pending_buf != nullptr && + _pending_buf->length() > FLAGS_max_body_size - buf->length())) { + LOG(WARNING) << "Close stream=" << id() + << " whose pending message size=" + << (_pending_buf != nullptr ? _pending_buf->length() : 0) + << " plus frame size=" << buf->length() + << " exceeds max_body_size=" << FLAGS_max_body_size; + delete _pending_buf; + _pending_buf = nullptr; + Close(EMSGSIZE, "Reassembled stream message is too large"); + return -1; + } if (_pending_buf != nullptr) { _pending_buf->append(*buf); buf->clear(); diff --git a/src/brpc/ubshm/timer/timer_mgr.cpp b/src/brpc/ubshm/timer/timer_mgr.cpp index b5e0c9ef3b..d1ad0b8b54 100644 --- a/src/brpc/ubshm/timer/timer_mgr.cpp +++ b/src/brpc/ubshm/timer/timer_mgr.cpp @@ -14,455 +14,174 @@ // KIND, either express or implied. See the License for the // specific language governing permissions and limitations // under the License. - #define _GNU_SOURCE + #include -#include -#include -#include -#include -#include +#include +#include +#include #include -#include +#include +#include +#include #include "brpc/ubshm/timer/timer_mgr.h" namespace brpc { namespace ubring { +std::unordered_map > g_timer_ctx_map; +std::mutex g_timer_ctx_mutex; +std::atomic g_total_timer_num; -int32_t g_epoll_fd = -1; -std::atomic g_total_timer_num(0); -TimerFdCtx *g_timer_fd_ctx_map = nullptr; -uint32_t g_max_system_fd = 0; -static pthread_t g_epoll_execute_thread = 0; -static int32_t g_timer_module_initialized = 0; +static std::atomic g_timer_id_counter(1); -#if defined(OS_MACOSX) -static int timerfd_create_macosx(int clockid, int flags); -static int timerfd_settime_macosx(int fd, int flags, - const itimerspec *new_value, - itimerspec *old_value); -#endif - -static RETURN_CODE DeleteTimerInner(uint32_t fd) { - if (g_timer_fd_ctx_map == nullptr) { - return UBRING_OK; - } - - if (pthread_spin_lock(&g_timer_fd_ctx_map[fd].spin_lock) != 0) { - return UBRING_ERR; - } +static timespec get_current_realtime() { + timespec ts{}; + clock_gettime(CLOCK_REALTIME, &ts); + return ts; +} - if (g_timer_fd_ctx_map[fd].status == TIMER_CONTEXT_NOT_USING) { - pthread_spin_unlock(&g_timer_fd_ctx_map[fd].spin_lock); - return UBRING_OK; +static timespec add_timespec(const timespec &base, const timespec &offset) { + timespec result{}; + result.tv_sec = base.tv_sec + offset.tv_sec; + result.tv_nsec = base.tv_nsec + offset.tv_nsec; + if (result.tv_nsec >= NS_PER_SEC) { + result.tv_sec += result.tv_nsec / NS_PER_SEC; + result.tv_nsec %= NS_PER_SEC; } - - g_timer_fd_ctx_map[fd].status = TIMER_CONTEXT_NOT_USING; - g_timer_fd_ctx_map[fd].cb = nullptr; - g_timer_fd_ctx_map[fd].args = nullptr; - g_timer_fd_ctx_map[fd].periodical = 0; - g_timer_fd_ctx_map[fd].fd = 0; - - pthread_spin_unlock(&g_timer_fd_ctx_map[fd].spin_lock); - -#if defined(OS_LINUX) - epoll_ctl(g_epoll_fd, EPOLL_CTL_DEL, (int)fd, nullptr); -#elif defined(OS_MACOSX) - struct kevent evt; - EV_SET(&evt, fd, EVFILT_TIMER, EV_DELETE, 0, 0, nullptr); - kevent(g_epoll_fd, &evt, 1, nullptr, 0, nullptr); -#endif - - uint64_t exp = 0; - read((int)fd, &exp, sizeof(exp)); - - close((int)fd); - std::atomic_fetch_sub(&g_total_timer_num, 1U); - return UBRING_OK; + return result; } -static RETURN_CODE StartTimeEpoll(void) { -#if defined(OS_LINUX) - g_epoll_fd = epoll_create1(0); -#elif defined(OS_MACOSX) - g_epoll_fd = kqueue(); -#endif - if (UNLIKELY(g_epoll_fd == -1)) { - LOG(ERROR) << "Failed to create epoll/kqueue. errno=" << errno; - return UBRING_ERR; +static std::shared_ptr find_context(uint64_t timer_id) { + std::lock_guard lock(g_timer_ctx_mutex); + auto it = g_timer_ctx_map.find(timer_id); + if (it == g_timer_ctx_map.end()) { + return nullptr; } + return it->second; +} - int ret = pthread_create(&g_epoll_execute_thread, nullptr, TimerEpoll, nullptr); - if (UNLIKELY(ret != 0)) { - LOG(ERROR) << "Failed to create thread err=" << ret; - return UBRING_ERR; - } - return UBRING_OK; +int TimerInit() { + return 0; } -static RETURN_CODE TimerSpinLocksInit(void) { - if (g_timer_fd_ctx_map == nullptr) { - LOG(ERROR) << "Timer module is not fully initialized."; - return UBRING_ERR; - } +void TimerModuleDestroy() { + std::vector > contexts; + contexts.reserve(g_timer_ctx_map.size()); - for (uint32_t fd = 0; fd < g_max_system_fd; fd++) { - int ret = pthread_spin_init(&g_timer_fd_ctx_map[fd].spin_lock, - PTHREAD_PROCESS_PRIVATE); - if (ret != EOK) { - LOG(ERROR) << "Failed to initialize spin lock for fd=" << fd; - for (uint32_t cleanup_fd = 0; cleanup_fd < fd; cleanup_fd++) { - pthread_spin_destroy(&g_timer_fd_ctx_map[cleanup_fd].spin_lock); - } - return UBRING_ERR; + { + std::lock_guard lock(g_timer_ctx_mutex); + for (auto &pair: g_timer_ctx_map) { + pair.second->periodical = 0; + bthread_timer_del(pair.second->timer_id); + contexts.push_back(pair.second); } + g_timer_ctx_map.clear(); + g_total_timer_num.store(0); } - return UBRING_OK; -} - -static RETURN_CODE ExecuteCallback(int32_t timer_fd) { - UnifiedCallback((void *)(&g_timer_fd_ctx_map[timer_fd])); - return UBRING_OK; -} -static RETURN_CODE TimerCtxMapCompletion(void) { - memset(g_timer_fd_ctx_map, 0, sizeof(TimerFdCtx) * g_max_system_fd); - - RETURN_CODE ret = TimerSpinLocksInit(); - if (ret != UBRING_OK) { - LOG(ERROR) << "Failed to init spin locks for timer module."; - return UBRING_ERR; + for (auto &ctx: contexts) { + ctx->self_ref.reset(); } - return UBRING_OK; } -RETURN_CODE TimerInit(void) { - if (g_timer_module_initialized > 0) { - return UBRING_OK; +int32_t TimerStart(const itimerspec *time, TimerCallback cb, void *args) { + if (cb == nullptr) { + LOG(ERROR) << "Timer callback is nullptr"; + return -1; } - g_total_timer_num.store(0); + auto ctx = std::make_shared(); + ctx->cb = cb; + ctx->args = args; + ctx->periodical = (time->it_interval.tv_sec > 0 || time->it_interval.tv_nsec > 0) ? 1 : 0; + ctx->interval = time->it_interval; + ctx->self_ref = ctx; - struct rlimit rlim; - if (getrlimit(RLIMIT_NOFILE, &rlim) != UBRING_OK) { - LOG(ERROR) << "Failed to get fd"; - return UBRING_ERR; - } - g_max_system_fd = (uint32_t)rlim.rlim_cur; + uint64_t timer_id = g_timer_id_counter.fetch_add(1); - if (g_timer_fd_ctx_map == nullptr) { - g_timer_fd_ctx_map = (TimerFdCtx *)malloc(sizeof(TimerFdCtx) * g_max_system_fd); - if (UNLIKELY(!g_timer_fd_ctx_map)) { - LOG(ERROR) << "Fail to malloc space for timer modules. errno=%d", errno; - return UBRING_ERR; - } + timespec abstime = add_timespec(get_current_realtime(), time->it_value); - RETURN_CODE ret = TimerCtxMapCompletion(); - if (ret != UBRING_OK) { - LOG(ERROR) << "Failed to init main data structure of Time Module. ret=" << ret; - free(g_timer_fd_ctx_map); - g_timer_fd_ctx_map = nullptr; - return UBRING_ERR; + { + std::lock_guard lock(g_timer_ctx_mutex); + g_timer_ctx_map[timer_id] = ctx; + ++g_total_timer_num; + int ret = bthread_timer_add(&ctx->timer_id, abstime, TimerCallbackWrapper, reinterpret_cast(timer_id)); + if (ret != 0) { + LOG(ERROR) << "Failed to add bthread timer, ret=" << ret; + g_timer_ctx_map.erase(timer_id); + --g_total_timer_num; + return -1; } } - RETURN_CODE ret = StartTimeEpoll(); - if (ret != UBRING_OK) { - LOG(ERROR) << "Failed to start Timer Epoll. ret=" << ret; - if (LIKELY(g_timer_fd_ctx_map != nullptr)) { - FREE_PTR(g_timer_fd_ctx_map); - } - return UBRING_ERR; - } - g_timer_module_initialized = 1; - return UBRING_OK; + return static_cast(timer_id); } -void *UnifiedCallback(void *args) { - TimerFdCtx *ctx = (TimerFdCtx *)args; - if (pthread_spin_lock(&ctx->spin_lock) != 0) { - return nullptr; - } - - if (ctx->status == TIMER_CONTEXT_NOT_USING) { - pthread_spin_unlock(&ctx->spin_lock); - return nullptr; - } - - void *(*cb)(void *) = ctx->cb; - void *cb_args = ctx->args; - uint32_t fd = ctx->fd; - int is_periodical = ctx->periodical; - ctx->status = TIMER_CONTEXT_CALLBACK_ONGOING; - - pthread_spin_unlock(&ctx->spin_lock); - - cb(cb_args); - - if (!is_periodical) { - DeleteTimerInner(fd); - } - return nullptr; +uint32_t GetActiveTimerNum() { + return g_total_timer_num.load(); } -void *TimerEpoll(void *args) { - UNREFERENCE_PARAM(args); -#if defined(OS_LINUX) - struct epoll_event ready_events[MAX_TIMER]; -#elif defined(OS_MACOSX) - struct kevent ready_events[MAX_TIMER]; -#endif - - while (1) { - if (g_timer_module_initialized <= 0) { - LOG(ERROR) << "The Timer module is not initialized."; - break; +void DeleteTimerSafe(uint64_t timer_id) { + int ret = 0; + { + std::lock_guard lock(g_timer_ctx_mutex); + auto it = g_timer_ctx_map.find(timer_id); + if (it == g_timer_ctx_map.end()) { + return; } - -#if defined(OS_LINUX) - int32_t ready_num = epoll_wait(g_epoll_fd, ready_events, MAX_TIMER, - TIMER_EPOLL_WAIT_TIMEOUT); -#elif defined(OS_MACOSX) - struct timespec timeout = {0, TIMER_EPOLL_WAIT_TIMEOUT * 1000000}; - int32_t ready_num = kevent(g_epoll_fd, nullptr, 0, ready_events, MAX_TIMER, &timeout); -#endif - - if (UNLIKELY(ready_num == -1)) { - errno_t err = errno; - if (err == EINTR) { - LOG_EVERY_SECOND(WARNING) << "Epoll/Kqueue wait was interrupted. errno=" << err; - continue; - } else if (err == EBADF) { - LOG(WARNING) << "The Timer module is destroyed."; - break; - } - LOG(ERROR) << "Epoll/Kqueue wait internal error. errno=" << err; - break; - } - - for (int32_t i = 0; i < ready_num; i++) { -#if defined(OS_LINUX) - struct epoll_event *event = &ready_events[i]; - int32_t timer_fd = event->data.fd; -#elif defined(OS_MACOSX) - struct kevent *event = &ready_events[i]; - int32_t timer_fd = event->ident; -#endif - - uint64_t exp = 0; - if (read(timer_fd, &exp, sizeof(exp)) < 0) { - if (errno != EBADF) { - LOG(ERROR) << "Failed to read timerfd=" << timer_fd << " errno=" << errno; - } - continue; - } - if (TimerFdCtxValidate((uint32_t)timer_fd) != UBRING_OK) { - continue; - } - - RETURN_CODE ret = ExecuteCallback(timer_fd); - if (ret != UBRING_OK) { - LOG(ERROR) << "Failed execute callback ret=" << ret; - DeleteTimerInner((uint32_t)timer_fd); - continue; - } + auto ctx = it->second; + ctx->periodical = 0; + ret = bthread_timer_del(ctx->timer_id); + if (ret == 0) { + g_timer_ctx_map.erase(it); + ctx->self_ref.reset(); } } - return nullptr; -} - -void DeleteTimerSafe(uint32_t fd) { - if (g_timer_fd_ctx_map == nullptr) { - return; - } - - if (pthread_spin_lock(&g_timer_fd_ctx_map[fd].spin_lock) != 0) { - return; + if (ret == 0) { + --g_total_timer_num; } - - if (g_timer_fd_ctx_map[fd].status == TIMER_CONTEXT_NOT_USING) { - pthread_spin_unlock(&g_timer_fd_ctx_map[fd].spin_lock); - return; - } - - g_timer_fd_ctx_map[fd].status = TIMER_CONTEXT_NOT_USING; - g_timer_fd_ctx_map[fd].cb = nullptr; - g_timer_fd_ctx_map[fd].args = nullptr; - g_timer_fd_ctx_map[fd].periodical = 0; - g_timer_fd_ctx_map[fd].fd = 0; - - pthread_spin_unlock(&g_timer_fd_ctx_map[fd].spin_lock); - -#if defined(OS_LINUX) - epoll_ctl(g_epoll_fd, EPOLL_CTL_DEL, (int)fd, nullptr); -#elif defined(OS_MACOSX) - struct kevent evt; - EV_SET(&evt, fd, EVFILT_TIMER, EV_DELETE, 0, 0, nullptr); - kevent(g_epoll_fd, &evt, 1, nullptr, 0, nullptr); -#endif - - uint64_t exp = 0; - read((int)fd, &exp, sizeof(exp)); - - close((int)fd); - std::atomic_fetch_sub(&g_total_timer_num, 1U); } -void DeleteTimer(uint32_t fd) { - if (g_timer_fd_ctx_map == nullptr) { - LOG(WARNING) << "The timer is not initialized."; +void DeleteTimer(uint64_t timer_id) { + std::lock_guard lock(g_timer_ctx_mutex); + auto it = g_timer_ctx_map.find(timer_id); + if (it == g_timer_ctx_map.end()) { + LOG(WARNING) << "Timer id=" << timer_id << " not found"; return; } - - g_timer_fd_ctx_map[fd].periodical = 0; -} - -int32_t TimerStart(const itimerspec *time, void *(*cb)(void *), void *args) { - if (g_epoll_fd == -1) { - LOG(ERROR) << "Timer epoll/kqueue encountered internal error."; - return -1; - } - -#if defined(OS_LINUX) - int timer_fd = timerfd_create(CLOCK_MONOTONIC, 0); -#elif defined(OS_MACOSX) - int timer_fd = timerfd_create_macosx(CLOCK_MONOTONIC, 0); -#endif - - if (UNLIKELY(timer_fd >= (int)g_max_system_fd || timer_fd == -1)) { - LOG(ERROR) << "Failed to create timerfd=" << timer_fd << " errno=" << errno; - return -1; - } - - g_timer_fd_ctx_map[timer_fd].status = TIMER_CONTEXT_EPOLL_WAITING; - g_timer_fd_ctx_map[timer_fd].cb = cb; - g_timer_fd_ctx_map[timer_fd].args = args; - g_timer_fd_ctx_map[timer_fd].fd = (uint32_t)timer_fd; - - if (LIKELY(time->it_interval.tv_sec > 0 || time->it_interval.tv_nsec > 0)) { - g_timer_fd_ctx_map[timer_fd].periodical = 1; - } - -#if defined(OS_LINUX) - struct epoll_event event = { - .events = EPOLLIN, - .data = {.fd = timer_fd} - }; - - int32_t ret = epoll_ctl(g_epoll_fd, EPOLL_CTL_ADD, timer_fd, &event); -#elif defined(OS_MACOSX) - struct kevent event; - uint64_t timeout_nsec = time->it_value.tv_sec * 1000000000ULL + time->it_value.tv_nsec; - uint64_t interval_nsec = time->it_interval.tv_sec * 1000000000ULL + time->it_interval.tv_nsec; - EV_SET(&event, timer_fd, EVFILT_TIMER, EV_ADD | EV_ENABLE, 0, - timeout_nsec / 1000000, nullptr); - int32_t ret = kevent(g_epoll_fd, &event, 1, nullptr, 0, nullptr); -#endif - - if (UNLIKELY(ret != 0)) { - CloseTimerFd(timer_fd); - LOG(ERROR) << "Failed to add event to epoll/kqueue. errno=" << errno; - return -1; - } - - std::atomic_fetch_add(&g_total_timer_num, 1U); - -#if defined(OS_LINUX) - ret = timerfd_settime(timer_fd, 0, time, nullptr); -#elif defined(OS_MACOSX) - ret = timerfd_settime_macosx(timer_fd, 0, time, nullptr); -#endif - - if (UNLIKELY(ret != 0)) { -#if defined(OS_LINUX) - if (epoll_ctl(g_epoll_fd, EPOLL_CTL_DEL, timer_fd, nullptr) != 0) { -#elif defined(OS_MACOSX) - struct kevent evt; - EV_SET(&evt, timer_fd, EVFILT_TIMER, EV_DELETE, 0, 0, nullptr); - if (kevent(g_epoll_fd, &evt, 1, nullptr, 0, nullptr) != 0) { -#endif - LOG(ERROR) << "Failed to delete the timer fd=" << timer_fd << " with errno=" << errno; - } - CloseTimerFd(timer_fd); - std::atomic_fetch_sub(&g_total_timer_num, 1U); - LOG(ERROR) << "Failed to set timer"; - return -1; - } - - return timer_fd; + it->second->periodical = 0; } -uint32_t GetActiveTimerNum(void) { - return std::atomic_load(&g_total_timer_num); -} - -void CloseTimerFd(int fd) { - g_timer_fd_ctx_map[fd].cb = nullptr; - g_timer_fd_ctx_map[fd].args = nullptr; - g_timer_fd_ctx_map[fd].status = TIMER_CONTEXT_NOT_USING; - g_timer_fd_ctx_map[fd].fd = 0; - g_timer_fd_ctx_map[fd].periodical = 0; - if (close((int)fd) != 0) { - LOG(ERROR) << "Failed to close timer fd=" << fd << " errno=" << errno; +void TimerCallbackWrapper(void *arg) { + auto timer_id = reinterpret_cast(arg); + auto ctx = find_context(timer_id); + if (ctx == nullptr) { + LOG(ERROR) << "timer_id is not found, timer_id=" << timer_id; return; } -} - -void TimerModuleDestroy(void) { - uint32_t max_fd = g_max_system_fd; - if (g_timer_fd_ctx_map) { - for (uint32_t fd = 0; fd < max_fd; fd++) { - if (g_timer_fd_ctx_map[fd].status != TIMER_CONTEXT_NOT_USING) { - DeleteTimerSafe(fd); - } - } - } - close(g_epoll_fd); - g_epoll_fd = -1; - g_total_timer_num = 0; - g_timer_module_initialized = 0; - int32_t ret = pthread_join(g_epoll_execute_thread, nullptr); - if (ret != EOK) { - LOG(ERROR) << "Failed to join pthread, during destroying timer module. ret=" << ret; + if (ctx->cb == nullptr) { + LOG(ERROR) << "Timer callback is nullptr"; return; } -} -RETURN_CODE TimerFdCtxValidate(uint32_t fd) { - if (fd >= g_max_system_fd) { - LOG(ERROR) << "TimerFd=" << fd << " is out of range=" << g_max_system_fd; - return UBRING_ERR; - } - if (g_timer_fd_ctx_map[fd].status == TIMER_CONTEXT_NOT_USING) { - LOG(ERROR) << "TimerFd=" << fd << " has wrong status=" << g_timer_fd_ctx_map[fd].status; - return UBRING_ERR; - } - if (g_timer_fd_ctx_map[fd].cb == nullptr) { - LOG(ERROR) << "The callback is not set."; - return UBRING_ERR; - } - - return UBRING_OK; -} + void *cb_args = ctx->args; + timespec interval = ctx->interval; -#if defined(OS_MACOSX) -static int timerfd_create_macosx(int clockid, int flags) { - int pipefd[2]; - if (pipe(pipefd) == -1) { - return -1; + if (ctx->periodical == 0) { + { + std::lock_guard lock(g_timer_ctx_mutex); + g_timer_ctx_map.erase(timer_id); + --g_total_timer_num; + } + ctx->self_ref.reset(); } - return pipefd[0]; -} -static int timerfd_settime_macosx(int fd, int flags, - const itimerspec *new_value, - itimerspec *old_value) { - if (old_value != nullptr) { - memset(old_value, 0, sizeof(itimerspec)); + ctx->cb(cb_args); + + if (ctx->periodical == 1) { + timespec abstime = add_timespec(get_current_realtime(), interval); + bthread_timer_add(&ctx->timer_id, abstime, TimerCallbackWrapper, reinterpret_cast(timer_id)); } - return 0; } -#endif - -} // namespace ubring -} // namespace brpc \ No newline at end of file +} // namespace ubring +} // namespace brpc diff --git a/src/brpc/ubshm/timer/timer_mgr.h b/src/brpc/ubshm/timer/timer_mgr.h index 1b42caef04..4f1236c79d 100644 --- a/src/brpc/ubshm/timer/timer_mgr.h +++ b/src/brpc/ubshm/timer/timer_mgr.h @@ -18,21 +18,15 @@ #ifndef BRPC_TIMER_MGR_H #define BRPC_TIMER_MGR_H #include -#include +#include +#include +#include +#include +#include +#include "bthread/types.h" +#include "bthread/unstable.h" #include "brpc/ubshm/common/common.h" -#if defined(OS_LINUX) -#include -#include -#elif defined(OS_MACOSX) -#include -#include -#include -#endif - -#define MAX_TIMER 1024 -#define TIMER_EPOLL_WAIT_TIMEOUT 1000 - #if defined(OS_MACOSX) struct itimerspec { @@ -40,34 +34,37 @@ struct itimerspec struct timespec it_value; }; #endif + namespace brpc { namespace ubring { -typedef enum { - TIMER_CONTEXT_NOT_USING, - TIMER_CONTEXT_EPOLL_WAITING, - TIMER_CONTEXT_CALLBACK_ONGOING -} TimerFdCtxStatus; -typedef struct { - void *(*cb)(void*); +constexpr long long NS_PER_SEC = 1000000000LL; + +typedef void * (*TimerCallback)(void *); + +struct TimerContext{ + TimerCallback cb; void *args; - uint32_t fd; - TimerFdCtxStatus status; uint32_t periodical; - pthread_spinlock_t spin_lock; -} TimerFdCtx; + timespec interval; + bthread_timer_t timer_id; + std::shared_ptr self_ref; +}; + +extern std::unordered_map> g_timer_ctx_map; +extern std::mutex g_timer_ctx_mutex; +extern std::atomic g_total_timer_num; -RETURN_CODE TimerInit(void); + +int TimerInit(void); void TimerModuleDestroy(void); -void *UnifiedCallback(void *args); -void *TimerEpoll(void *args); -int32_t TimerStart(const itimerspec *time, void *(*cb)(void *), void *args); +int32_t TimerStart(const itimerspec *time, TimerCallback cb, void *args); uint32_t GetActiveTimerNum(void); -void CloseTimerFd(int fd); -void DeleteTimerSafe(uint32_t fd); -void DeleteTimer(uint32_t fd); -RETURN_CODE TimerFdCtxValidate(uint32_t fd); +void DeleteTimerSafe(uint64_t timer_id); +void DeleteTimer(uint64_t timer_id); + +void TimerCallbackWrapper(void *arg); } } #endif //BRPC_TIMER_MGR_H \ No newline at end of file diff --git a/src/brpc/ubshm/ub_endpoint.cpp b/src/brpc/ubshm/ub_endpoint.cpp index 31539fda85..f6e0ef7ba0 100644 --- a/src/brpc/ubshm/ub_endpoint.cpp +++ b/src/brpc/ubshm/ub_endpoint.cpp @@ -848,6 +848,7 @@ int UBShmEndpoint::PollingModeInitialize(bthread_tag_t tag, while (running->load(std::memory_order_relaxed)) { while (poller->op_queue.Dequeue(op)) { if (op.type == PollerSidOp::ADD) { + poller_sids.erase(op); poller_sids.emplace(op); } else if (op.type == PollerSidOp::REMOVE) { poller_sids.erase(op); diff --git a/test/brpc_channel_unittest.cpp b/test/brpc_channel_unittest.cpp index 997859ef76..b6125e26d8 100644 --- a/test/brpc_channel_unittest.cpp +++ b/test/brpc_channel_unittest.cpp @@ -2316,6 +2316,55 @@ TEST_F(ChannelTest, init_as_single_server) { } } +TEST_F(ChannelTest, reject_reinitialization_after_successful_init) { + butil::EndPoint first_endpoint; + butil::EndPoint second_endpoint; + ASSERT_EQ(0, str2endpoint("127.0.0.1:59347", &first_endpoint)); + ASSERT_EQ(0, str2endpoint("127.0.0.1:59348", &second_endpoint)); + const brpc::SocketMapKey first_key(first_endpoint); + const brpc::SocketMapKey second_key(second_endpoint); + + { + brpc::Channel channel; + ASSERT_EQ(0, channel.Init(first_endpoint, NULL)); + brpc::SocketId id; + ASSERT_EQ(0, brpc::SocketMapFind(first_key, &id)); + ASSERT_EQ(channel._server_id, id); + ASSERT_EQ(-1, channel.Init(first_endpoint, NULL)); + ASSERT_EQ(-1, channel.Init(second_endpoint, NULL)); + ASSERT_EQ(-1, channel.Init("unknown://unknown", "rr", NULL)); + } + + brpc::SocketId id; + EXPECT_NE(0, brpc::SocketMapFind(first_key, &id)); + EXPECT_NE(0, brpc::SocketMapFind(second_key, &id)); +} + +TEST_F(ChannelTest, retry_init_after_failed_init) { + butil::EndPoint endpoint; + ASSERT_EQ(0, str2endpoint("127.0.0.1:59349", &endpoint)); + const brpc::SocketMapKey key(endpoint); + + { + brpc::Channel channel; + brpc::ChannelOptions invalid_options; + invalid_options.client_host = "not a valid client host"; + ASSERT_EQ(-1, channel.Init(endpoint, &invalid_options)); + EXPECT_EQ(brpc::INVALID_SOCKET_ID, channel._server_id); + + brpc::ChannelOptions valid_options; + ASSERT_EQ(0, channel.Init(endpoint, &valid_options)); + EXPECT_NE(brpc::INVALID_SOCKET_ID, channel._server_id); + EXPECT_EQ(endpoint, channel._server_address); + brpc::SocketId id; + ASSERT_EQ(0, brpc::SocketMapFind(key, &id)); + ASSERT_EQ(channel._server_id, id); + } + + brpc::SocketId id; + EXPECT_NE(0, brpc::SocketMapFind(key, &id)); +} + TEST_F(ChannelTest, init_using_unknown_naming_service) { brpc::Channel channel; ASSERT_EQ(-1, channel.Init("unknown://unknown", "unknown", nullptr)); @@ -2391,73 +2440,138 @@ TEST_F(ChannelTest, parse_hostname) { brpc::ChannelOptions opt; opt.succeed_without_server = false; opt.protocol = brpc::PROTOCOL_HTTP; - brpc::Channel channel; - ASSERT_EQ(-1, channel.Init("", 8888, &opt)); - ASSERT_EQ("", channel._service_name); - ASSERT_EQ(-1, channel.Init("", &opt)); - ASSERT_EQ("", channel._service_name); - - ASSERT_EQ(0, channel.Init("http://127.0.0.1", 8888, &opt)); - ASSERT_EQ("127.0.0.1:8888", channel._service_name); - ASSERT_EQ(0, channel.Init("http://127.0.0.1:8888", &opt)); - ASSERT_EQ("127.0.0.1:8888", channel._service_name); - - ASSERT_EQ(0, channel.Init("localhost", 8888, &opt)); - ASSERT_EQ("localhost:8888", channel._service_name); - ASSERT_EQ(0, channel.Init("localhost:8888", &opt)); - ASSERT_EQ("localhost:8888", channel._service_name); - - ASSERT_EQ(0, channel.Init("http://www.baidu.com", &opt)); - ASSERT_EQ("www.baidu.com", channel._service_name); - ASSERT_EQ(0, channel.Init("http://www.baidu.com:80", &opt)); - ASSERT_EQ("www.baidu.com:80", channel._service_name); - ASSERT_EQ(0, channel.Init("http://www.baidu.com", 80, &opt)); - ASSERT_EQ("www.baidu.com:80", channel._service_name); - ASSERT_EQ(0, channel.Init("http://www.baidu.com:8888", &opt)); - ASSERT_EQ("www.baidu.com:8888", channel._service_name); - ASSERT_EQ(0, channel.Init("http://www.baidu.com", 8888, &opt)); - ASSERT_EQ("www.baidu.com:8888", channel._service_name); - ASSERT_EQ(0, channel.Init("http://www.baidu.com", "rr", &opt)); - ASSERT_EQ("www.baidu.com", channel._service_name); - ASSERT_EQ(0, channel.Init("http://www.baidu.com:80", "rr", &opt)); - ASSERT_EQ("www.baidu.com:80", channel._service_name); - ASSERT_EQ(0, channel.Init("http://www.baidu.com:8888", "rr", &opt)); - ASSERT_EQ("www.baidu.com:8888", channel._service_name); + { + brpc::Channel channel; + ASSERT_EQ(-1, channel.Init("", 8888, &opt)); + ASSERT_EQ("", channel._service_name); + } + { + brpc::Channel channel; + ASSERT_EQ(-1, channel.Init("", &opt)); + ASSERT_EQ("", channel._service_name); + } + + { + brpc::Channel channel; + ASSERT_EQ(0, channel.Init("http://127.0.0.1", 8888, &opt)); + ASSERT_EQ("127.0.0.1:8888", channel._service_name); + } + { + brpc::Channel channel; + ASSERT_EQ(0, channel.Init("http://127.0.0.1:8888", &opt)); + ASSERT_EQ("127.0.0.1:8888", channel._service_name); + } + + { + brpc::Channel channel; + ASSERT_EQ(0, channel.Init("localhost", 8888, &opt)); + ASSERT_EQ("localhost:8888", channel._service_name); + } + { + brpc::Channel channel; + ASSERT_EQ(0, channel.Init("localhost:8888", &opt)); + ASSERT_EQ("localhost:8888", channel._service_name); + } + + { + brpc::Channel channel; + ASSERT_EQ(0, channel.Init("http://www.baidu.com", &opt)); + ASSERT_EQ("www.baidu.com", channel._service_name); + } + { + brpc::Channel channel; + ASSERT_EQ(0, channel.Init("http://www.baidu.com:80", &opt)); + ASSERT_EQ("www.baidu.com:80", channel._service_name); + } + { + brpc::Channel channel; + ASSERT_EQ(0, channel.Init("http://www.baidu.com", 80, &opt)); + ASSERT_EQ("www.baidu.com:80", channel._service_name); + } + { + brpc::Channel channel; + ASSERT_EQ(0, channel.Init("http://www.baidu.com:8888", &opt)); + ASSERT_EQ("www.baidu.com:8888", channel._service_name); + } + { + brpc::Channel channel; + ASSERT_EQ(0, channel.Init("http://www.baidu.com", 8888, &opt)); + ASSERT_EQ("www.baidu.com:8888", channel._service_name); + } + { + brpc::Channel channel; + ASSERT_EQ(0, channel.Init("http://www.baidu.com", "rr", &opt)); + ASSERT_EQ("www.baidu.com", channel._service_name); + } + { + brpc::Channel channel; + ASSERT_EQ(0, channel.Init("http://www.baidu.com:80", "rr", &opt)); + ASSERT_EQ("www.baidu.com:80", channel._service_name); + } + { + brpc::Channel channel; + ASSERT_EQ(0, channel.Init("http://www.baidu.com:8888", "rr", &opt)); + ASSERT_EQ("www.baidu.com:8888", channel._service_name); + } opt.mutable_ssl_options()->verify.verify_mode = brpc::VerifyMode::VERIFY_PEER; opt.mutable_ssl_options()->verify.verify_depth = 1; opt.mutable_ssl_options()->verify.ca_file_path = "cert1.crt"; - ASSERT_EQ(0, channel.Init("https://www.baidu.com", &opt)); - ASSERT_EQ("www.baidu.com", channel._service_name); + { + brpc::Channel channel; + ASSERT_EQ(0, channel.Init("https://www.baidu.com", &opt)); + ASSERT_EQ("www.baidu.com", channel._service_name); #if defined(USE_MESALINK) || \ (!defined(OPENSSL_IS_BORINGSSL) && OPENSSL_VERSION_NUMBER < 0x10002000L) - ASSERT_TRUE(channel._options.ssl_options().verify.expected_peer_name.empty()); + ASSERT_TRUE(channel._options.ssl_options().verify.expected_peer_name.empty()); #else - ASSERT_EQ("www.baidu.com", - channel._options.ssl_options().verify.expected_peer_name); + ASSERT_EQ("www.baidu.com", + channel._options.ssl_options().verify.expected_peer_name); #endif - ASSERT_EQ(0, channel.Init("https://www.baidu.com:443", &opt)); - ASSERT_EQ("www.baidu.com:443", channel._service_name); + } + { + brpc::Channel channel; + ASSERT_EQ(0, channel.Init("https://www.baidu.com:443", &opt)); + ASSERT_EQ("www.baidu.com:443", channel._service_name); #if defined(USE_MESALINK) || \ (!defined(OPENSSL_IS_BORINGSSL) && OPENSSL_VERSION_NUMBER < 0x10002000L) - ASSERT_TRUE(channel._options.ssl_options().verify.expected_peer_name.empty()); + ASSERT_TRUE(channel._options.ssl_options().verify.expected_peer_name.empty()); #else - ASSERT_EQ("www.baidu.com", - channel._options.ssl_options().verify.expected_peer_name); + ASSERT_EQ("www.baidu.com", + channel._options.ssl_options().verify.expected_peer_name); #endif - ASSERT_EQ(0, channel.Init("https://www.baidu.com", 443, &opt)); - ASSERT_EQ("www.baidu.com:443", channel._service_name); - ASSERT_EQ(0, channel.Init("https://www.baidu.com:1443", &opt)); - ASSERT_EQ("www.baidu.com:1443", channel._service_name); - ASSERT_EQ(0, channel.Init("https://www.baidu.com", 1443, &opt)); - ASSERT_EQ("www.baidu.com:1443", channel._service_name); - ASSERT_EQ(0, channel.Init("https://www.baidu.com", "rr", &opt)); - ASSERT_EQ("www.baidu.com", channel._service_name); - ASSERT_EQ(0, channel.Init("https://www.baidu.com:443", "rr", &opt)); - ASSERT_EQ("www.baidu.com:443", channel._service_name); - ASSERT_EQ(0, channel.Init("https://www.baidu.com:1443", "rr", &opt)); - ASSERT_EQ("www.baidu.com:1443", channel._service_name); + } + { + brpc::Channel channel; + ASSERT_EQ(0, channel.Init("https://www.baidu.com", 443, &opt)); + ASSERT_EQ("www.baidu.com:443", channel._service_name); + } + { + brpc::Channel channel; + ASSERT_EQ(0, channel.Init("https://www.baidu.com:1443", &opt)); + ASSERT_EQ("www.baidu.com:1443", channel._service_name); + } + { + brpc::Channel channel; + ASSERT_EQ(0, channel.Init("https://www.baidu.com", 1443, &opt)); + ASSERT_EQ("www.baidu.com:1443", channel._service_name); + } + { + brpc::Channel channel; + ASSERT_EQ(0, channel.Init("https://www.baidu.com", "rr", &opt)); + ASSERT_EQ("www.baidu.com", channel._service_name); + } + { + brpc::Channel channel; + ASSERT_EQ(0, channel.Init("https://www.baidu.com:443", "rr", &opt)); + ASSERT_EQ("www.baidu.com:443", channel._service_name); + } + { + brpc::Channel channel; + ASSERT_EQ(0, channel.Init("https://www.baidu.com:1443", "rr", &opt)); + ASSERT_EQ("www.baidu.com:1443", channel._service_name); + } const char *address_list[] = { "10.127.0.1:1234", diff --git a/test/brpc_streaming_rpc_unittest.cpp b/test/brpc_streaming_rpc_unittest.cpp index 2e4c046a13..12c2ff3c31 100644 --- a/test/brpc_streaming_rpc_unittest.cpp +++ b/test/brpc_streaming_rpc_unittest.cpp @@ -27,6 +27,7 @@ #include "brpc/controller.h" #include "brpc/channel.h" #include "brpc/callback.h" +#include "brpc/details/controller_private_accessor.h" #include "brpc/socket.h" #include "brpc/stream_impl.h" #include "brpc/policy/streaming_rpc_protocol.h" @@ -367,6 +368,92 @@ static bool WaitForTrue(const std::atomic& f, int timeout_ms) { return WaitForTrue([&f]() { return f.load(std::memory_order_acquire); }, timeout_ms); } +class ReassemblyLimitHandler : public brpc::StreamInputHandler { +public: + int on_received_messages(brpc::StreamId, + butil::IOBuf* const messages[], + size_t size) override { + for (size_t i = 0; i < size; ++i) { + received_bytes.fetch_add(messages[i]->length(), + std::memory_order_relaxed); + } + received_messages.fetch_add(size, std::memory_order_release); + return 0; + } + + void on_idle_timeout(brpc::StreamId) override {} + void on_closed(brpc::StreamId) override {} + void on_failed(brpc::StreamId, int error_code, + const std::string&) override { + failure_code.store(error_code, std::memory_order_release); + } + + std::atomic received_bytes{0}; + std::atomic received_messages{0}; + std::atomic failure_code{0}; +}; + +TEST_F(StreamingRpcTest, limit_reassembled_message_size) { + std::string old_max_body_size; + std::string old_segment_size; + ASSERT_TRUE(GFLAGS_NAMESPACE::GetCommandLineOption( + "max_body_size", &old_max_body_size)); + ASSERT_TRUE(GFLAGS_NAMESPACE::GetCommandLineOption( + "stream_write_max_segment_size", &old_segment_size)); + + ReassemblyLimitHandler handler; + brpc::StreamOptions server_stream_options; + server_stream_options.handler = &handler; + brpc::Server server; + MyServiceWithStream service(server_stream_options); + ASSERT_EQ(0, server.AddService( + &service, brpc::SERVER_DOESNT_OWN_SERVICE)); + ASSERT_EQ(0, server.Start(0, nullptr)); + + brpc::Channel channel; + ASSERT_EQ(0, channel.Init(server.listen_address(), nullptr)); + brpc::Controller cntl; + brpc::StreamId request_stream; + ASSERT_EQ(0, brpc::StreamCreate(&request_stream, cntl, nullptr)); + brpc::ScopedStream stream_guard(request_stream); + test::EchoService_Stub stub(&channel); + stub.Echo(&cntl, &request, &response, nullptr); + ASSERT_FALSE(cntl.Failed()) << cntl.ErrorText(); + + ASSERT_FALSE(GFLAGS_NAMESPACE::SetCommandLineOption( + "max_body_size", "64").empty()); + ASSERT_FALSE(GFLAGS_NAMESPACE::SetCommandLineOption( + "stream_write_max_segment_size", "16").empty()); + BRPC_SCOPE_EXIT { + GFLAGS_NAMESPACE::SetCommandLineOption( + "max_body_size", old_max_body_size.c_str()); + GFLAGS_NAMESPACE::SetCommandLineOption( + "stream_write_max_segment_size", old_segment_size.c_str()); + }; + + butil::IOBuf at_limit; + at_limit.append(std::string(64, 'a')); + ASSERT_EQ(0, brpc::StreamWrite(request_stream, at_limit)); + ASSERT_TRUE(WaitForTrue([&handler]() { + return handler.received_messages.load(std::memory_order_acquire) == 1; + }, 2000)); + ASSERT_EQ(64u, handler.received_bytes.load(std::memory_order_relaxed)); + + butil::IOBuf over_limit; + over_limit.append(std::string(65, 'b')); + ASSERT_EQ(0, brpc::StreamWrite(request_stream, over_limit)); + ASSERT_TRUE(WaitForTrue([&handler]() { + return handler.failure_code.load(std::memory_order_acquire) != 0; + }, 2000)); + ASSERT_EQ(EMSGSIZE, + handler.failure_code.load(std::memory_order_relaxed)); + ASSERT_EQ(1u, + handler.received_messages.load(std::memory_order_relaxed)); + + server.Stop(0); + server.Join(); +} + TEST_F(StreamingRpcTest, sanity) { brpc::Server server; MyServiceWithStream service; @@ -1130,6 +1217,76 @@ class MyServiceWithExtraStream : public test::EchoService { int _n; }; +class MyServiceWithMismatchedExtraStreamIds : public test::EchoService { +public: + MyServiceWithMismatchedExtraStreamIds(size_t stream_count, int adjustment) + : _stream_count(stream_count), _adjustment(adjustment) {} + + void Echo(::google::protobuf::RpcController* controller, + const ::test::EchoRequest* request, + ::test::EchoResponse* response, + ::google::protobuf::Closure* done) override { + brpc::ClosureGuard done_guard(done); + brpc::Controller* cntl = static_cast(controller); + response->set_message(request->message()); + + brpc::ControllerPrivateAccessor accessor(cntl); + brpc::StreamSettings* settings = accessor.remote_stream_settings(); + if (_adjustment < 0) { + settings->mutable_extra_stream_ids()->RemoveLast(); + } else { + settings->add_extra_stream_ids(settings->extra_stream_ids(0)); + } + + brpc::StreamIds response_streams; + ASSERT_EQ(0, brpc::StreamAccept(response_streams, *cntl, nullptr)); + ASSERT_EQ((int)_stream_count + _adjustment, + (int)response_streams.size()); + } + +private: + size_t _stream_count; + int _adjustment; +}; + +TEST_F(StreamingRpcTest, reject_mismatched_returned_stream_identifiers) { + const size_t STREAM_COUNT = 3; + + for (int adjustment : {-1, 1}) { + brpc::Server server; + MyServiceWithMismatchedExtraStreamIds service(STREAM_COUNT, adjustment); + ASSERT_EQ(0, server.AddService( + &service, brpc::SERVER_DOESNT_OWN_SERVICE)); + ASSERT_EQ(0, server.Start(0, nullptr)); + + brpc::Channel channel; + ASSERT_EQ(0, channel.Init(server.listen_address(), nullptr)); + + brpc::Controller cntl; + brpc::StreamIds request_streams; + ASSERT_EQ(0, brpc::StreamCreate(request_streams, STREAM_COUNT, cntl, + nullptr)); + ASSERT_EQ(STREAM_COUNT, request_streams.size()); + + test::EchoService_Stub stub(&channel); + stub.Echo(&cntl, &request, &response, nullptr); + ASSERT_TRUE(cntl.Failed()); + ASSERT_EQ(brpc::ERESPONSE, cntl.ErrorCode()); + const std::string expected_error = + "extra_stream_ids, expected " + std::to_string(STREAM_COUNT - 1); + ASSERT_NE(std::string::npos, + cntl.ErrorText().find(expected_error)); + + for (brpc::StreamId stream_id : request_streams) { + brpc::StreamUniquePtr stream; + ASSERT_NE(0, brpc::Stream::Address(stream_id, &stream)); + } + + server.Stop(0); + server.Join(); + } +} + TEST_F(StreamingRpcTest, batch_create_extra_stream) { const size_t STREAM_COUNT = 3; // 1 first stream + 2 extra streams const int N = 1000; diff --git a/tools/parallel_http/parallel_http.cpp b/tools/parallel_http/parallel_http.cpp index c11c2158dd..c6864e3242 100644 --- a/tools/parallel_http/parallel_http.cpp +++ b/tools/parallel_http/parallel_http.cpp @@ -207,5 +207,7 @@ int main(int argc, char** argv) { usleep(10000); } } + + delete[] args; return 0; }