simplify gRPC message pending/drain queue (#72)

Author: Shikugawa

cpp2sky/internal/async_client.h

@@ -20,6 +20,8 @@
 
 #include <memory>
 
+#include "source/utils/circular_buffer.h"
+
 using google::protobuf::Message;
 
 namespace cpp2sky {
@@ -30,14 +32,14 @@ class AsyncClient {
   virtual ~AsyncClient() = default;
 
   /**
-   * Send the specified protobuf message
+   * Send the specified protobuf message.
    */
   virtual void sendMessage(RequestType message) = 0;
 
   /**
-   * Drain pending message.
+   * Pending message queue reference.
    */
-  virtual void drainPendingMessage(RequestType message) = 0;
+  virtual CircularBuffer<RequestType>& pendingMessages() = 0;
 
   /**
    * Start stream if there is no living stream.

docker-compose.dev.yml

@@ -30,18 +30,18 @@ services:
       - '8600:8600'
       - '8600:8600/udp'
   oap:
-    image: apache/skywalking-oap-server:8.4.0-es7
+    image: apache/skywalking-oap-server:8.5.0-es7
     depends_on:
       - elasticsearch
     links:
       - elasticsearch
-    restart: always
     ports:
       - 11800:11800
       - 12800:12800
     environment:
       SW_STORAGE: elasticsearch7
       SW_STORAGE_ES_CLUSTER_NODES: elasticsearch:9200
+    restart: always
     healthcheck:
       test: ["CMD-SHELL", "/skywalking/bin/swctl"]
       interval: 30s
@@ -52,12 +52,11 @@ services:
       - consul
       - elasticsearch
   ui:
-    image: apache/skywalking-ui:8.4.0
+    image: apache/skywalking-ui:8.5.0
     depends_on:
       - oap
     links:
       - oap
-    restart: always
     ports:
       - 8080:8080
     environment:

source/BUILD

@@ -36,33 +36,6 @@ cc_library(
   visibility = ["//visibility:public"],
 )
 
-# TODO(shikugawa): This is a terrible hack to disable unexpected behavior for 
-# test that has introduced GRPC_TEST.
-cc_library(
-  name = "async_client_impl_test",
-  hdrs = [
-    "grpc_async_client_impl.h",
-  ],
-  srcs = [
-    "grpc_async_client_impl.cc",
-  ],
-  deps = [
-    "@skywalking_data_collect_protocol//language-agent:tracing_protocol_cc_proto",
-    "@skywalking_data_collect_protocol//language-agent:tracing_protocol_cc_grpc",
-    "@com_github_grpc_grpc//:grpc++",
-    "@com_github_gabime_spdlog//:spdlog",
-    "@com_google_absl//absl/strings:strings",
-    "//cpp2sky/internal:async_client_interface",
-    "//cpp2sky/internal:stream_builder_interface",
-    "//cpp2sky/internal:matcher_interface",
-    "//cpp2sky:cpp2sky_interface",
-    "//source/utils:util_lib",
-    "//source/matchers:suffix_matcher_lib",
-  ],
-  copts = ["-DGRPC_TEST"],
-  visibility = ["//visibility:public"],
-)
-
 cc_library(
   name = "cpp2sky_data_lib",
   hdrs  =[

source/cds_impl.h

@@ -37,8 +37,8 @@ class GrpcAsyncConfigDiscoveryServiceClient final
   ~GrpcAsyncConfigDiscoveryServiceClient();
 
   void sendMessage(CdsRequest request);
-  void drainPendingMessage(CdsRequest pending_message) override {}
   void startStream() override {}
+  CircularBuffer<CdsRequest>& pendingMessages() override { assert(false); }
   grpc::CompletionQueue& completionQueue() override { return cq_; }
   grpc::TemplatedGenericStub<CdsRequest, CdsResponse>& stub() override {
     return stub_;

source/dynamic_config.h

@@ -42,4 +42,4 @@ class DynamicConfig {
   std::set<std::string> ignore_fields_;
 };
 
-}  // namespace cpp2sky
+}  // namespace cpp2sky

source/grpc_async_client_impl.cc

@@ -41,37 +41,30 @@ GrpcAsyncSegmentReporterClient::GrpcAsyncSegmentReporterClient(
 }
 
 GrpcAsyncSegmentReporterClient::~GrpcAsyncSegmentReporterClient() {
-  // It will wait until there is no drained messages.
-  // There are no timeout option to handle this, so if you would like to stop
-  // them, you should send signals like SIGTERM.
-  // If server stopped with accidental issue, the event loop handle that it
-  // failed to send message and close stream, then recreate new stream and try
-  // to do it. This process will continue forever without sending explicit
-  // signal.
-  // TODO(shikugawa): Block to wait drained messages to be clear with createing
-  // condition variable wrapper.
-#ifndef GRPC_TEST
+  // It will wait until there is no drained messages with 5 second timeout.
   if (stream_) {
     std::unique_lock<std::mutex> lck(mux_);
-    while (!drained_messages_.empty()) {
-      cv_.wait(lck);
+    while (!pending_messages_.empty()) {
+      cv_.wait_for(lck, std::chrono::seconds(5));
+      pending_messages_.clear();
     }
   }
-#endif
 
   resetStream();
 }
 
 void GrpcAsyncSegmentReporterClient::sendMessage(TracerRequestType message) {
+  pending_messages_.push(message);
+
   if (!stream_) {
-    drained_messages_.push(message);
     info(
-        "[Reporter] No active stream, inserted message into draining message "
+        "[Reporter] No active stream, inserted message into pending message "
         "queue. "
         "pending message size: {}",
-        drained_messages_.size());
+        pending_messages_.size());
     return;
   }
+
   stream_->sendMessage(message);
 }
 
@@ -80,19 +73,6 @@ void GrpcAsyncSegmentReporterClient::startStream() {
 
   stream_ = factory_->create(*this, cv_);
   info("[Reporter] Stream {} had created.", fmt::ptr(stream_.get()));
-
-  const auto drained_messages_size = drained_messages_.size();
-
-  while (!drained_messages_.empty()) {
-    auto msg = drained_messages_.front();
-    drained_messages_.pop();
-    if (msg.has_value()) {
-      stream_->sendMessage(msg.value());
-    }
-  }
-
-  info("[Reporter] {} drained messages inserted into pending messages.",
-       drained_messages_size);
 }
 
 void GrpcAsyncSegmentReporterClient::resetStream() {
@@ -121,28 +101,15 @@ GrpcAsyncSegmentReporterStream::GrpcAsyncSegmentReporterStream(
   request_writer_->StartCall(reinterpret_cast<void*>(&ready_));
 }
 
-GrpcAsyncSegmentReporterStream::~GrpcAsyncSegmentReporterStream() {
-  const auto pending_messages_size = pending_messages_.size();
-  while (!pending_messages_.empty()) {
-    auto msg = pending_messages_.front();
-    pending_messages_.pop();
-    if (msg.has_value()) {
-      client_.drainPendingMessage(msg.value());
-    }
-  }
-  info("[Reporter] {} pending messages drained.", pending_messages_size);
-}
-
 void GrpcAsyncSegmentReporterStream::sendMessage(TracerRequestType message) {
-  pending_messages_.push(message);
   clearPendingMessage();
 }
 
 bool GrpcAsyncSegmentReporterStream::clearPendingMessage() {
-  if (state_ != StreamState::Idle || pending_messages_.empty()) {
+  if (state_ != StreamState::Idle || client_.pendingMessages().empty()) {
     return false;
   }
-  auto message = pending_messages_.front();
+  auto message = client_.pendingMessages().front();
   if (!message.has_value()) {
     return false;
   }
@@ -164,21 +131,19 @@ void GrpcAsyncSegmentReporterStream::onIdle() {
 
   // Release pending messages which are inserted when stream is not ready
   // to write.
-  clearPendingMessage();
-
-  if (pending_messages_.empty()) {
+  if (!clearPendingMessage()) {
     cv_.notify_all();
   }
 }
 
 void GrpcAsyncSegmentReporterStream::onWriteDone() {
   info("[Reporter] Write finished");
 
-  // Enqueue message after sending message finished.
+  // Dequeue message after sending message finished.
   // With this, messages which failed to sent never lost even if connection
   // was closed. because pending messages with messages which failed to send
   // will drained and resend another stream.
-  pending_messages_.pop();
+  client_.pendingMessages().pop();
   state_ = StreamState::Idle;
 
   onIdle();

source/grpc_async_client_impl.h

@@ -27,13 +27,13 @@
 #include "cpp2sky/internal/stream_builder.h"
 #include "language-agent/Tracing.grpc.pb.h"
 #include "language-agent/Tracing.pb.h"
-#include "source/utils/circular_buffer.h"
-
-#define DRAIN_BUFFER_SIZE 1024
-#define PENDING_MESSAGE_BUFFER_SIZE 1024
 
 namespace cpp2sky {
 
+namespace {
+static constexpr size_t pending_message_buffer_size = 1024;
+}
+
 using TracerRequestType = skywalking::v3::SegmentObject;
 using TracerResponseType = skywalking::v3::Commands;
 
@@ -51,8 +51,8 @@ class GrpcAsyncSegmentReporterClient final
 
   // AsyncClient
   void sendMessage(TracerRequestType message) override;
-  void drainPendingMessage(TracerRequestType pending_message) override {
-    drained_messages_.push(pending_message);
+  CircularBuffer<TracerRequestType>& pendingMessages() override {
+    return pending_messages_;
   }
   void startStream() override;
   grpc::TemplatedGenericStub<TracerRequestType, TracerResponseType>& stub()
@@ -61,7 +61,7 @@ class GrpcAsyncSegmentReporterClient final
   }
   grpc::CompletionQueue& completionQueue() override { return cq_; }
 
-  size_t numOfMessages() { return drained_messages_.size(); }
+  size_t numOfMessages() { return pending_messages_.size(); }
 
  private:
   void resetStream();
@@ -72,7 +72,8 @@ class GrpcAsyncSegmentReporterClient final
   grpc::CompletionQueue& cq_;
   grpc::TemplatedGenericStub<TracerRequestType, TracerResponseType> stub_;
   AsyncStreamPtr<TracerRequestType, TracerResponseType> stream_;
-  CircularBuffer<TracerRequestType> drained_messages_{DRAIN_BUFFER_SIZE};
+  CircularBuffer<TracerRequestType> pending_messages_{
+      pending_message_buffer_size};
 
   std::mutex mux_;
   std::condition_variable cv_;
@@ -85,7 +86,6 @@ class GrpcAsyncSegmentReporterStream final
   GrpcAsyncSegmentReporterStream(
       AsyncClient<TracerRequestType, TracerResponseType>& client,
       std::condition_variable& cv, const std::string& token);
-  ~GrpcAsyncSegmentReporterStream() override;
 
   // AsyncStream
   void sendMessage(TracerRequestType message) override;
@@ -106,8 +106,6 @@ class GrpcAsyncSegmentReporterStream final
   std::unique_ptr<
       grpc::ClientAsyncReaderWriter<TracerRequestType, TracerResponseType>>
       request_writer_;
-  CircularBuffer<TracerRequestType> pending_messages_{
-      PENDING_MESSAGE_BUFFER_SIZE};
   StreamState state_{StreamState::Initialized};
 
   StreamCallbackTag ready_{StreamState::Ready, this};

source/utils/circular_buffer.h

@@ -86,6 +86,14 @@ class CircularBuffer {
    */
   size_t size() const { return item_count_; }
 
+  /**
+   * Clear buffer
+   */
+  void clear() {
+    buf_.clear();
+    item_count_ = 0;
+  }
+
   // Used for test
   size_t frontIdx() { return front_; }
   size_t backIdx() { return back_; }

test/BUILD

@@ -47,7 +47,7 @@ cc_test(
   ],
   deps = [
     "@com_google_googletest//:gtest_main",
-    "//source:async_client_impl_test",
+    "//source:cpp2sky_lib",
     ":mocks",
   ],
   visibility = ["//visibility:public"],

test/grpc_async_client_test.cc

@@ -57,18 +57,4 @@ TEST_F(GrpcAsyncSegmentReporterClientTest, SendMessageTest) {
   client_->sendMessage(fake_message);
 }
 
-TEST_F(GrpcAsyncSegmentReporterClientTest, MessageDrainTest) {
-  std::queue<TracerRequestType> fake_pending_messages;
-  for (int i = 0; i < 3; ++i) {
-    fake_pending_messages.emplace(skywalking::v3::SegmentObject());
-  }
-  while (fake_pending_messages.size() != 0) {
-    auto msg = fake_pending_messages.front();
-    fake_pending_messages.pop();
-    client_->drainPendingMessage(msg);
-  }
-  EXPECT_EQ(fake_pending_messages.size(), 0);
-  EXPECT_EQ(client_->numOfMessages(), 3);
-}
-
 }  // namespace cpp2sky