multiplexing gRPC callback and cds events (#85)

Author: Shikugawa

source/tracer_impl.cc

@@ -28,7 +28,7 @@ namespace cpp2sky {
 TracerImpl::TracerImpl(TracerConfig& config,
                        std::shared_ptr<grpc::ChannelCredentials> cred)
     : config_(config),
-      grpc_callback_thread_([this] { this->run(); }),
+      evloop_thread_([this] { this->run(); }),
       segment_factory_(config) {
   init(config, cred);
 }
@@ -38,7 +38,7 @@ TracerImpl::TracerImpl(
     AsyncClientPtr<TracerRequestType, TracerResponseType> reporter_client)
     : config_(config),
       reporter_client_(std::move(reporter_client)),
-      grpc_callback_thread_([this] { this->run(); }),
+      evloop_thread_([this] { this->run(); }),
       segment_factory_(config) {
   init(config, nullptr);
 }
@@ -47,11 +47,7 @@ TracerImpl::~TracerImpl() {
   reporter_client_.reset();
   cds_client_.reset();
   cq_.Shutdown();
-  grpc_callback_thread_.join();
-
-  if (cds_thread_.joinable()) {
-    cds_thread_.join();
-  }
+  evloop_thread_.join();
 }
 
 TracingContextPtr TracerImpl::newContext() { return segment_factory_.create(); }
@@ -80,23 +76,28 @@ void TracerImpl::run() {
   void* got_tag;
   bool ok = false;
   while (true) {
-    grpc::CompletionQueue::NextStatus status =
-        cq_.AsyncNext(&got_tag, &ok, gpr_inf_future(GPR_CLOCK_REALTIME));
-    if (status == grpc::CompletionQueue::SHUTDOWN) {
-      return;
+    // TODO(shikugawa): cleanup evloop handler.
+    if (cds_timer_ != nullptr && cds_timer_->check()) {
+      cdsRequest();
+    }
+
+    grpc::CompletionQueue::NextStatus status = cq_.AsyncNext(
+        &got_tag, &ok, gpr_time_from_nanos(0, GPR_CLOCK_REALTIME));
+    switch (status) {
+      case grpc::CompletionQueue::TIMEOUT:
+        continue;
+      case grpc::CompletionQueue::SHUTDOWN:
+        return;
     }
     static_cast<StreamCallbackTag*>(got_tag)->callback(!ok);
   }
 }
 
-void TracerImpl::startCds(std::chrono::seconds seconds) {
-  while (true) {
-    skywalking::v3::ConfigurationSyncRequest request;
-    request.set_service(config_.tracerConfig().service_name());
-    request.set_uuid(config_.uuid());
-    cds_client_->sendMessage(request);
-    std::this_thread::sleep_for(seconds);
-  }
+void TracerImpl::cdsRequest() {
+  skywalking::v3::ConfigurationSyncRequest request;
+  request.set_service(config_.tracerConfig().service_name());
+  request.set_uuid(config_.uuid());
+  cds_client_->sendMessage(request);
 }
 
 void TracerImpl::init(TracerConfig& config,
@@ -124,10 +125,8 @@ void TracerImpl::init(TracerConfig& config,
         config.address(), cq_,
         std::make_unique<GrpcAsyncConfigDiscoveryServiceStreamBuilder>(config_),
         cred);
-    cds_thread_ = std::thread([this] {
-      this->startCds(
-          std::chrono::seconds(config_.tracerConfig().cds_request_interval()));
-    });
+    cds_timer_ =
+        std::make_unique<Timer>(config_.tracerConfig().cds_request_interval());
   }
 }
 

source/tracer_impl.h

@@ -23,6 +23,7 @@
 #include "source/cds_impl.h"
 #include "source/grpc_async_client_impl.h"
 #include "source/tracing_context_impl.h"
+#include "source/utils/timer.h"
 
 namespace cpp2sky {
 
@@ -50,14 +51,14 @@ class TracerImpl : public Tracer {
   void init(TracerConfig& config,
             std::shared_ptr<grpc::ChannelCredentials> cred);
   void run();
-  void startCds(std::chrono::seconds seconds);
+  void cdsRequest();
 
+  std::unique_ptr<Timer> cds_timer_;
   DynamicConfig config_;
   AsyncClientPtr<TracerRequestType, TracerResponseType> reporter_client_;
   AsyncClientPtr<CdsRequest, CdsResponse> cds_client_;
   grpc::CompletionQueue cq_;
-  std::thread grpc_callback_thread_;
-  std::thread cds_thread_;
+  std::thread evloop_thread_;
   TracingContextFactory segment_factory_;
   std::list<MatcherPtr> op_name_matchers_;
 };

source/utils/BUILD

@@ -6,6 +6,7 @@ cc_library(
     "base64.h",
     "random_generator.h",
     "circular_buffer.h",
+    "timer.h",
   ],
   srcs = [
     "random_generator.cc",

source/utils/timer.h

@@ -0,0 +1,43 @@
+// Copyright 2021 SkyAPM
+
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+
+//     http://apache.org/licenses/LICENSE-2.0
+
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#pragma once
+
+#include <chrono>
+
+#include "cpp2sky/time.h"
+
+namespace cpp2sky {
+class Timer {
+ public:
+  Timer(int64_t interval_sec) {
+    interval_ = interval_sec * 1000;
+    prev_time_ = TimePoint<SteadyTime>().fetch();
+  }
+
+  bool check() {
+    const auto current_time = TimePoint<SteadyTime>().fetch();
+    bool result = current_time - prev_time_ > interval_;
+    if (result) {
+      prev_time_ = current_time;
+    }
+
+    return result;
+  }
+
+ private:
+  int64_t prev_time_;
+  int64_t interval_;
+};
+}  // namespace cpp2sky