alibaba · chenneal · Jan 7, 2021 · Jan 11, 2021 · Jan 11, 2021 · Jan 12, 2021
diff --git a/.github/workflows/main.yml b/.github/workflows/main.yml
@@ -0,0 +1,51 @@
+name: CI
+
+on:
+  push:
+    branches: master
+  pull_request:
+    branches: "*"
+
+jobs:
+  build:
+
+    runs-on: ${{ matrix.os }}
+    strategy:
+      matrix:
+        java: [8]
+        os: [ubuntu-18.04]
+
+    steps:
+      - uses: actions/checkout@v2
+
+      - name: Install bazel
+        run: |
+          wget https://github.com/bazelbuild/bazel/releases/download/3.7.0/bazel_3.7.0-linux-x86_64.deb
+          sudo dpkg -i bazel_3.7.0-linux-x86_64.deb
+      - name: Install clang8
+        run: |
+          echo "deb http://apt.llvm.org/bionic/ llvm-toolchain-bionic-8 main" | sudo tee -a /etc/apt/sources.list
+          echo "deb-src http://apt.llvm.org/bionic/ llvm-toolchain-bionic-8 main" | sudo tee -a /etc/apt/sources.list
+          wget -O - https://apt.llvm.org/llvm-snapshot.gpg.key | sudo apt-key add -
+          sudo apt-get update
+          sudo apt-get -y install clang-8 clang-format-8
+
+      - name: Clang format checker
+        run: |
+          bash format.sh
+
+      - name: Library build
+        run: |
+          bazel build //sentinel-core/... && bazel build //sentinel-datasource-extension/...
+
+      - name: Sentinel core unit tests
+        run: |
+          bazel test --test_filter=*-ParamMetricTest.TestOperateMetric --test_output=errors --strategy=TestRunner=standalone //sentinel-core/...
+
+      - name: Sentinel datasource extension tests
+        run: |
+          bazel build //sentinel-core/... && bazel build //sentinel-datasource-extension/...
+
+      - name: tsan for flow control
+        run: |
+          bazel build -c dbg --config=clang-tsan //tests/... && ./bazel-bin/tests/tsan-flow
diff --git a/.travis.yml b/.travis.yml
diff --git a/BUILD b/BUILD
@@ -8,6 +8,7 @@ configure_make(
         "--enable-shared=no",
         "--disable-libevent-regress",
         "--disable-openssl",
+        "--with-pic",
     ],
     lib_source = "@com_github_libevent//:all",
     out_lib_dir = "lib",

diff --git a/sentinel-core/circuitbreaker/slot_test.cc b/sentinel-core/circuitbreaker/slot_test.cc
@@ -3,10 +3,10 @@
 
 #include "sentinel-core/test/mock/statistic/node/mock.h"
 
-#include "sentinel-core/common/string_resource_wrapper.h"
-#include "sentinel-core/circuitbreaker/rule_manager.h"
 #include "sentinel-core/circuitbreaker/rule.h"
+#include "sentinel-core/circuitbreaker/rule_manager.h"
 #include "sentinel-core/circuitbreaker/slot.h"
+#include "sentinel-core/common/string_resource_wrapper.h"
 
 using testing::_;
 using testing::InSequence;
@@ -38,7 +38,7 @@ TEST(CircuitBreakerSlotTest, CircuitBreakerErrorRatioTest) {
       std::make_shared<StringResourceWrapper>(resource_name, EntryType::OUT);
   auto entry = std::make_shared<Entry>(resource, context);
   entry->set_cur_node(node);
-  
+
   auto entry_error = std::make_shared<Entry>(resource, context);
   entry_error->set_cur_node(node);
   entry_error->set_error("test_error");
@@ -50,7 +50,8 @@ TEST(CircuitBreakerSlotTest, CircuitBreakerErrorRatioTest) {
 
   // Test breaker checking when no rule exists.
   for (int i = 0; i < 10; i++) {
-    Entry_And_Exit(slot_checker, slot_complete, entry, resource, node, 1, 0, myParams);
+    Entry_And_Exit(slot_checker, slot_complete, entry, resource, node, 1, 0,
+                   myParams);
   }
 
   Rule rule{resource_name};
@@ -68,7 +69,8 @@ TEST(CircuitBreakerSlotTest, CircuitBreakerErrorRatioTest) {
 
   // Test breaker checking when error entry happens.
   for (int i = 0; i < 10; i++) {
-    Entry_And_Exit(slot_checker, slot_complete, entry_error, resource, node, 1, 0, myParams);
+    Entry_And_Exit(slot_checker, slot_complete, entry_error, resource, node, 1,
+                   0, myParams);
   }
 
   EXPECT_EQ(cbs[0]->CurrentState(), State::kOpen);
@@ -89,7 +91,7 @@ TEST(CircuitBreakerSlotTest, CircuitBreakerErrorRatioTest) {
 }
 
 TEST(CircuitBreakerSlotTest, CircuitBreakerSlowRatioTest) {
-    std::string resource_name{"test_resource"};
+  std::string resource_name{"test_resource"};
   EntryContextSharedPtr context =
       std::make_shared<EntryContext>("test_context");
   Stat::NodeSharedPtr node = std::make_shared<Stat::MockNode>();
@@ -110,7 +112,8 @@ TEST(CircuitBreakerSlotTest, CircuitBreakerSlowRatioTest) {
 
   // Test breaker checking when no rule exists.
   for (int i = 0; i < 10; i++) {
-    Entry_And_Exit(slot_checker, slot_complete, entry, resource, node, 1, 0, myParams);
+    Entry_And_Exit(slot_checker, slot_complete, entry, resource, node, 1, 0,
+                   myParams);
   }
 
   Rule rule{resource_name};
@@ -129,7 +132,8 @@ TEST(CircuitBreakerSlotTest, CircuitBreakerSlowRatioTest) {
 
   // Test breaker checking when slow entry happens.
   for (int i = 0; i < 10; i++) {
-    Entry_And_Exit(slot_checker, slot_complete, entry_slow, resource, node, 1, 0, myParams);
+    Entry_And_Exit(slot_checker, slot_complete, entry_slow, resource, node, 1,
+                   0, myParams);
   }
 
   EXPECT_EQ(cbs[0]->CurrentState(), State::kOpen);
@@ -149,5 +153,5 @@ TEST(CircuitBreakerSlotTest, CircuitBreakerSlowRatioTest) {
   m.LoadRules({});
 }
 
-}
-}
+}  // namespace CircuitBreaker
+}  // namespace Sentinel
diff --git a/sentinel-core/log/block/block_log_task.cc b/sentinel-core/log/block/block_log_task.cc
@@ -81,7 +81,7 @@ void BlockLogTask::Log(const std::string& resource, const std::string& cause) {
   }
   auto key = absl::StrFormat("%s|%s", resource, cause);
   {
-    absl::ReaderMutexLock lck(&mtx_);
+    absl::WriterMutexLock lck(&mtx_);
     auto it = map_.find(key);
     if (it != map_.end()) {
       it->second.last_block_ = TimeUtils::CurrentTimeMillis().count();

diff --git a/sentinel-core/statistic/base/leap_array.h b/sentinel-core/statistic/base/leap_array.h
@@ -48,7 +48,7 @@ class LeapArray {
   const int32_t bucket_length_ms_;  // time length of each bucket
  private:
   const std::unique_ptr<WindowWrapSharedPtr<T>[]> array_;
-  std::mutex mtx_;
+  mutable std::mutex leap_array_mtx_;
 
   int32_t CalculateTimeIdx(/*@Valid*/ int64_t time_millis) const;
   int64_t CalculateWindowStart(/*@Valid*/ int64_t time_millis) const;
@@ -78,9 +78,12 @@ WindowWrapSharedPtr<T> LeapArray<T>::CurrentWindow(int64_t time_millis) {
   int64_t bucket_start = CalculateWindowStart(time_millis);
 
   while (true) {
+    std::unique_lock<std::mutex> lck(leap_array_mtx_, std::defer_lock);
+    // TODO: granularity too rough, need to be optimized.
+    leap_array_mtx_.lock();
     WindowWrapSharedPtr<T> old = array_[idx];
+    leap_array_mtx_.unlock();
     if (old == nullptr) {
-      std::unique_lock<std::mutex> lck(mtx_, std::defer_lock);
       if (lck.try_lock() && array_[idx] == nullptr) {
         WindowWrapSharedPtr<T> bucket = std::make_shared<WindowWrap<T>>(
             bucket_length_ms_, bucket_start, NewEmptyBucket(time_millis));
@@ -90,7 +93,7 @@ WindowWrapSharedPtr<T> LeapArray<T>::CurrentWindow(int64_t time_millis) {
     } else if (bucket_start == old->BucketStart()) {
       return old;
     } else if (bucket_start > old->BucketStart()) {
-      std::unique_lock<std::mutex> lck(mtx_, std::defer_lock);
+      std::unique_lock<std::mutex> lck(leap_array_mtx_, std::defer_lock);
       if (lck.try_lock()) {
         ResetWindowTo(old, bucket_start);
         return old;
@@ -148,7 +151,10 @@ std::vector<WindowWrapSharedPtr<T>> LeapArray<T>::Buckets(
   }
   int size = sample_count_;  // array_.size()
   for (int i = 0; i < size; i++) {
+    // TODO: granularity too rough, need to be optimized.
+    leap_array_mtx_.lock();
     auto w = array_[i];
+    leap_array_mtx_.unlock();
     if (w == nullptr || IsBucketDeprecated(time_millis, w)) {
       continue;
     }
@@ -166,7 +172,10 @@ std::vector<std::shared_ptr<T>> LeapArray<T>::Values(
   }
   int size = sample_count_;  // array_.size()
   for (int i = 0; i < size; i++) {
+    // TODO: granularity too rough, need to be optimized.
+    leap_array_mtx_.lock();
     WindowWrapSharedPtr<T> w = array_[i];
+    leap_array_mtx_.unlock();
     if (w == nullptr || IsBucketDeprecated(time_millis, w)) {
       continue;
     }

diff --git a/sentinel-core/statistic/base/metric_bucket.cc b/sentinel-core/statistic/base/metric_bucket.cc
@@ -19,7 +19,7 @@ int64_t MetricBucket::Get(const MetricEvent& event) const {
   return counters_[i].load();
 }
 
-int64_t MetricBucket::MinRt() const { return min_rt_; }
+int64_t MetricBucket::MinRt() const { return min_rt_.load(); }
 
 void MetricBucket::Add(const MetricEvent& event, int64_t n) {
   int i = (int)event;
@@ -29,12 +29,12 @@ void MetricBucket::Add(const MetricEvent& event, int64_t n) {
 void MetricBucket::AddRt(int64_t rt) {
   Add(MetricEvent::RT, rt);
   // Not thread-safe, but it's okay.
-  if (rt < min_rt_) {
-    min_rt_ = rt;
+  if (rt < min_rt_.load()) {
+    min_rt_.store(rt);
   }
 }
 
-void MetricBucket::InitMinRt() { min_rt_ = Constants::kMaxAllowedRt; }
+void MetricBucket::InitMinRt() { min_rt_.store(Constants::kMaxAllowedRt); }
 
 }  // namespace Stat
 }  // namespace Sentinel
diff --git a/sentinel-core/statistic/base/metric_bucket.h b/sentinel-core/statistic/base/metric_bucket.h
@@ -23,7 +23,7 @@ class MetricBucket {
   const std::unique_ptr<std::atomic<int64_t>[]> counters_ =
       std::make_unique<std::atomic<int64_t>[]>(
           static_cast<int>(MetricEvent::Count));
-  long min_rt_;
+  std::atomic<long> min_rt_;
 
   void InitMinRt();
 };

diff --git a/sentinel-core/statistic/base/window_wrap.h b/sentinel-core/statistic/base/window_wrap.h
@@ -1,5 +1,6 @@
 #pragma once
 
+#include <atomic>
 #include <memory>
 
 namespace Sentinel {
@@ -21,7 +22,7 @@ class WindowWrap {
   bool IsTimeInBucket(int64_t time_millis) const;
 
  private:
-  int64_t bucket_start_;
+  std::atomic<int64_t> bucket_start_;
   const int64_t bucket_length_ms_;
   const std::shared_ptr<T> value_;
 };
@@ -36,7 +37,7 @@ int64_t WindowWrap<T>::BucketLengthInMs() const {
 
 template <typename T>
 int64_t WindowWrap<T>::BucketStart() const {
-  return bucket_start_;
+  return bucket_start_.load();
 }
 
 template <typename T>
@@ -46,7 +47,7 @@ std::shared_ptr<T> WindowWrap<T>::Value() const {
 
 template <typename T>
 void WindowWrap<T>::ResetTo(int64_t start_time) {
-  this->bucket_start_ = start_time;
+  this->bucket_start_.store(start_time);
 }
 
 template <typename T>

diff --git a/sentinel-core/transport/command/http_server.cc b/sentinel-core/transport/command/http_server.cc
@@ -11,12 +11,7 @@ namespace Transport {
 HttpServer::HttpServer(http_request_callback_t callback)
     : request_callback_(callback) {}
 
-HttpServer::~HttpServer() {
-  if (http_) {
-    evhttp_free(http_);
-    http_ = nullptr;
-  }
-}
+HttpServer::~HttpServer() { Stop(); }
 
 bool HttpServer::Start(int port) {
   auto ret = event_loop_thread_.Start();
@@ -26,11 +21,9 @@ bool HttpServer::Start(int port) {
 
   port_ = port;
 
-  std::promise<bool> start_promise;
-  auto start_future = start_promise.get_future();
-  auto task = [&start_promise, this]() { InternalStart(start_promise); };
-
-  event_loop_thread_.RunTask(task);
+  auto start_future = start_promise_.get_future();
+  event_loop_thread_.RunTask(
+      [this](void) mutable { InternalStart(start_promise_); });
 
   return start_future.get();
 }

diff --git a/sentinel-core/transport/command/http_server.h b/sentinel-core/transport/command/http_server.h
@@ -33,6 +33,7 @@ class HttpServer {
 
   struct evhttp *http_ = nullptr;
   http_request_callback_t request_callback_;
+  std::promise<bool> start_promise_;
 
   static void HttpGenCallback(struct evhttp_request *req, void *arg);
 

diff --git a/sentinel-core/transport/common/BUILD b/sentinel-core/transport/common/BUILD
@@ -11,6 +11,7 @@ cc_library(
         ],
     deps = [
         "//:libevent",
+        "@com_google_absl//absl/synchronization",
     ],
     visibility = ["//visibility:public"],
 )