grpc: improve stream more reliable (#32)

Author: Shikugawa

cpp2sky/internal/async_client.h

@@ -17,8 +17,8 @@
 #include <google/protobuf/message.h>
 #include <grpcpp/grpcpp.h>
 
+#include <condition_variable>
 #include <memory>
-#include <queue>
 
 using google::protobuf::Message;
 
@@ -62,10 +62,9 @@ class AsyncClient {
   virtual std::string peerAddress() = 0;
 
   /**
-   * Drain pending messages
+   * Drain pending message.
    */
-  virtual void drainPendingMessages(
-      std::queue<RequestType>& pending_messages) = 0;
+  virtual void drainPendingMessage(RequestType message) = 0;
 
   /**
    * Reset stream if it is living.
@@ -109,6 +108,11 @@ class AsyncStream {
    */
   virtual void sendMessage(RequestType message) = 0;
 
+  /**
+   * Restore drained message.
+   */
+  virtual void undrainMessage(RequestType message) = 0;
+
   /**
    * Handle incoming event related to this stream.
    */
@@ -128,7 +132,7 @@ class AsyncStreamFactory {
    */
   virtual AsyncStreamPtr<RequestType> create(
       AsyncClient<RequestType, ResponseType>* client,
-      std::queue<RequestType>& drained_messages) = 0;
+      std::condition_variable& cv) = 0;
 };
 
 template <class RequestType, class ResponseType>

example/sample_client.cc

@@ -57,6 +57,5 @@ int main() {
 
   // 5. Send span data
   tracer->sendSegment(std::move(current_segment));
-
   return 0;
 }

source/grpc_async_client_impl.cc

@@ -14,11 +14,16 @@
 
 #include "grpc_async_client_impl.h"
 
+#include <chrono>
 #include <string_view>
+#include <thread>
 
 #include "source/utils/exception.h"
 #include "utils/grpc_status.h"
 
+#define DEFAULT_CONNECTION_ACTIVE_RETRY_TIMES 5
+#define DEFAULT_CONNECTION_ACTIVE_RETRY_SLEEP_SEC 3
+
 namespace cpp2sky {
 
 namespace {
@@ -51,9 +56,45 @@ GrpcAsyncSegmentReporterClient::GrpcAsyncSegmentReporterClient(
   startStream();
 }
 
+GrpcAsyncSegmentReporterClient::~GrpcAsyncSegmentReporterClient() {
+  // If connection is inactive, it dispose all drained messages even if it has
+  // tons of messages.
+  uint8_t retry_times = DEFAULT_CONNECTION_ACTIVE_RETRY_TIMES;
+  while (channel_->GetState(false) !=
+         grpc_connectivity_state::GRPC_CHANNEL_READY) {
+    if (retry_times <= 0) {
+      gpr_log(GPR_INFO,
+              "All %ld pending messages have disposed because of no active "
+              "connection",
+              drained_messages_.size());
+      resetStream();
+      return;
+    }
+    retry_times--;
+    std::this_thread::sleep_for(
+        std::chrono::seconds(DEFAULT_CONNECTION_ACTIVE_RETRY_SLEEP_SEC));
+  }
+
+  // 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.
+  {
+    std::unique_lock<std::mutex> lck(mux_);
+    while (!drained_messages_.empty()) {
+      cv_.wait(lck);
+    }
+  }
+
+  resetStream();
+}
+
 void GrpcAsyncSegmentReporterClient::sendMessage(TracerRequestType message) {
   if (!stream_) {
-    drained_messages_.emplace(message);
+    drained_messages_.push(message);
     gpr_log(GPR_INFO,
             "No active stream, inserted message into draining message queue. "
             "pending message size: %ld",
@@ -76,42 +117,39 @@ GrpcAsyncSegmentReporterClient::createWriter(grpc::ClientContext* ctx,
 void GrpcAsyncSegmentReporterClient::startStream() {
   resetStream();
 
-  // Try to establish connection.
-  channel_->GetState(true);
-  stream_ = factory_.create(this, drained_messages_);
-  stream_->startStream();
-}
+  stream_ = factory_.create(this, cv_);
 
-void GrpcAsyncSegmentReporterClient::drainPendingMessages(
-    std::queue<TracerRequestType>& pending_messages) {
-  const auto pending_messages_size = pending_messages.size();
-  while (!pending_messages.empty()) {
-    auto msg = pending_messages.front();
-    pending_messages.pop();
-    drained_messages_.emplace(msg);
+  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_->undrainMessage(msg.value());
+    }
   }
-  gpr_log(GPR_INFO, "%ld pending messages drained.", pending_messages_size);
+  gpr_log(GPR_INFO, "%ld drained messages inserted into pending messages.",
+          drained_messages_size);
+
+  stream_->startStream();
+  gpr_log(GPR_INFO, "Stream %p had created.", stream_.get());
 }
 
 GrpcAsyncSegmentReporterStream::GrpcAsyncSegmentReporterStream(
     AsyncClient<TracerRequestType, TracerResponseType>* client,
-    std::queue<TracerRequestType>& drained_messages)
-    : client_(client) {
-  const auto drained_messages_size = drained_messages.size();
-  while (!drained_messages.empty()) {
-    auto msg = drained_messages.front();
-    pending_messages_.emplace(msg);
-    drained_messages.pop();
-  }
-  gpr_log(GPR_INFO, "%ld drained messages inserted into pending messages.",
-          drained_messages_size);
-}
+    std::condition_variable& cv)
+    : client_(client), cv_(cv) {}
 
 GrpcAsyncSegmentReporterStream::~GrpcAsyncSegmentReporterStream() {
-  {
-    std::unique_lock<std::mutex> lck_(mux_);
-    cond_.wait(lck_, [this] { return pending_messages_.empty(); });
+  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());
+    }
   }
+  gpr_log(GPR_INFO, "%ld pending messages drained.", pending_messages_size);
+
   ctx_.TryCancel();
   request_writer_->Finish(&status_, toTag(&finish_));
 }
@@ -130,17 +168,19 @@ bool GrpcAsyncSegmentReporterStream::startStream() {
 }
 
 void GrpcAsyncSegmentReporterStream::sendMessage(TracerRequestType message) {
-  pending_messages_.emplace(message);
+  pending_messages_.push(message);
   clearPendingMessages();
 }
 
 bool GrpcAsyncSegmentReporterStream::clearPendingMessages() {
   if (state_ != Operation::Idle || pending_messages_.empty()) {
     return false;
   }
-  auto message = pending_messages_.back();
-  pending_messages_.pop();
-  request_writer_->Write(message, toTag(&write_done_));
+  auto message = pending_messages_.front();
+  if (!message.has_value()) {
+    return false;
+  }
+  request_writer_->Write(message.value(), toTag(&write_done_));
   return true;
 }
 
@@ -152,6 +192,11 @@ bool GrpcAsyncSegmentReporterStream::handleOperation(Operation incoming_op) {
     state_ = Operation::Idle;
   } else if (state_ == Operation::WriteDone) {
     gpr_log(GPR_INFO, "Write finished");
+    // Enqueue 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();
     state_ = Operation::Idle;
   }
 
@@ -161,16 +206,13 @@ bool GrpcAsyncSegmentReporterStream::handleOperation(Operation incoming_op) {
     // to write.
     clearPendingMessages();
 
-    // Release if lock with condition variable has been acquired.
-    // It will blocked if stream has notified to be closed.
     if (pending_messages_.empty()) {
-      cond_.notify_all();
+      cv_.notify_all();
     }
     return true;
   } else if (state_ == Operation::Finished) {
     gpr_log(GPR_INFO, "Stream closed with http status: %d",
             grpcStatusToGenericHttpStatus(status_.error_code()));
-    client_->drainPendingMessages(pending_messages_);
     if (!status_.ok()) {
       gpr_log(GPR_ERROR, "%s", status_.error_message().c_str());
     }
@@ -181,12 +223,11 @@ bool GrpcAsyncSegmentReporterStream::handleOperation(Operation incoming_op) {
 
 AsyncStreamPtr<TracerRequestType> GrpcAsyncSegmentReporterStreamFactory::create(
     AsyncClient<TracerRequestType, TracerResponseType>* client,
-    std::queue<TracerRequestType>& drained_messages) {
+    std::condition_variable& cv) {
   if (client == nullptr) {
     return nullptr;
   }
-  return std::make_shared<GrpcAsyncSegmentReporterStream>(client,
-                                                          drained_messages);
+  return std::make_shared<GrpcAsyncSegmentReporterStream>(client, cv);
 }
 
 }  // namespace cpp2sky

source/grpc_async_client_impl.h

@@ -24,6 +24,10 @@
 #include "cpp2sky/internal/async_client.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 {
 
@@ -54,19 +58,21 @@ class GrpcAsyncSegmentReporterClient final
       AsyncStreamFactory<TracerRequestType, TracerResponseType>& factory,
       std::shared_ptr<grpc::ChannelCredentials> cred, std::string address,
       std::string token);
+  ~GrpcAsyncSegmentReporterClient();
 
   // AsyncClient
   void sendMessage(TracerRequestType message) override;
   std::string peerAddress() override { return address_; }
   std::unique_ptr<grpc::ClientAsyncWriter<TracerRequestType>> createWriter(
       grpc::ClientContext* ctx, TracerResponseType* response,
       void* tag) override;
-  void drainPendingMessages(
-      std::queue<TracerRequestType>& pending_messages) override;
+  void drainPendingMessage(TracerRequestType pending_message) override {
+    drained_messages_.push(pending_message);
+  }
   void resetStream() override {
     if (stream_) {
+      gpr_log(GPR_INFO, "Stream %p had destroyed.", stream_.get());
       stream_.reset();
-      stream_ = nullptr;
     }
   }
   void startStream() override;
@@ -80,7 +86,10 @@ class GrpcAsyncSegmentReporterClient final
   grpc::CompletionQueue* cq_;
   std::shared_ptr<grpc::Channel> channel_;
   AsyncStreamPtr<TracerRequestType> stream_;
-  std::queue<TracerRequestType> drained_messages_;
+  CircularBuffer<TracerRequestType> drained_messages_{DRAIN_BUFFER_SIZE};
+
+  std::mutex mux_;
+  std::condition_variable cv_;
 };
 
 struct TaggedStream {
@@ -96,13 +105,16 @@ class GrpcAsyncSegmentReporterStream final
  public:
   GrpcAsyncSegmentReporterStream(
       AsyncClient<TracerRequestType, TracerResponseType>* client,
-      std::queue<TracerRequestType>& drained_messages);
+      std::condition_variable& cv);
   ~GrpcAsyncSegmentReporterStream() override;
 
   // AsyncStream
   bool startStream() override;
   void sendMessage(TracerRequestType message) override;
   bool handleOperation(Operation incoming_op) override;
+  void undrainMessage(TracerRequestType message) override {
+    pending_messages_.push(message);
+  }
 
  private:
   bool clearPendingMessages();
@@ -112,15 +124,15 @@ class GrpcAsyncSegmentReporterStream final
   grpc::Status status_;
   grpc::ClientContext ctx_;
   std::unique_ptr<grpc::ClientAsyncWriter<TracerRequestType>> request_writer_;
-  std::queue<TracerRequestType> pending_messages_;
+  CircularBuffer<TracerRequestType> pending_messages_{
+      PENDING_MESSAGE_BUFFER_SIZE};
   Operation state_{Operation::Initialized};
 
   TaggedStream connected_{Operation::Connected, this};
   TaggedStream write_done_{Operation::WriteDone, this};
   TaggedStream finish_{Operation::Finished, this};
 
-  std::mutex mux_;
-  std::condition_variable cond_;
+  std::condition_variable& cv_;
 };
 
 class GrpcAsyncSegmentReporterStreamFactory final
@@ -129,7 +141,7 @@ class GrpcAsyncSegmentReporterStreamFactory final
   // AsyncStreamFactory
   AsyncStreamPtr<TracerRequestType> create(
       AsyncClient<TracerRequestType, TracerResponseType>* client,
-      std::queue<TracerRequestType>& drained_messages) override;
+      std::condition_variable& cv) override;
 };
 
 }  // namespace cpp2sky

source/utils/BUILD

@@ -7,6 +7,7 @@ cc_library(
     "exception.h",
     "random_generator.h",
     "grpc_status.h",
+    "circular_buffer.h",
   ],
   deps = [
     "//cpp2sky:random_interface",