|
|
|
@ -19,6 +19,7 @@ |
|
|
|
|
#ifndef TEST_CPP_MICROBENCHMARKS_CALLBACK_TEST_SERVICE_H |
|
|
|
|
#define TEST_CPP_MICROBENCHMARKS_CALLBACK_TEST_SERVICE_H |
|
|
|
|
|
|
|
|
|
#include <benchmark/benchmark.h> |
|
|
|
|
#include <condition_variable> |
|
|
|
|
#include <memory> |
|
|
|
|
#include <mutex> |
|
|
|
@ -47,24 +48,30 @@ class CallbackStreamingTestService |
|
|
|
|
class BidiClient |
|
|
|
|
: public grpc::experimental::ClientBidiReactor<EchoRequest, EchoResponse> { |
|
|
|
|
public: |
|
|
|
|
BidiClient(EchoTestService::Stub* stub, EchoRequest* request, |
|
|
|
|
EchoResponse* response, ClientContext* context, |
|
|
|
|
int num_msgs_to_send) |
|
|
|
|
: request_{request}, |
|
|
|
|
BidiClient(benchmark::State& state, EchoTestService::Stub* stub, |
|
|
|
|
EchoRequest* request, EchoResponse* response, std::mutex& mu, |
|
|
|
|
std::condition_variable& cv, bool& done) |
|
|
|
|
: state_{state}, |
|
|
|
|
stub_{stub}, |
|
|
|
|
request_{request}, |
|
|
|
|
response_{response}, |
|
|
|
|
context_{context}, |
|
|
|
|
msgs_to_send_{num_msgs_to_send} { |
|
|
|
|
stub->experimental_async()->BidiStream(context_, this); |
|
|
|
|
MaybeWrite(); |
|
|
|
|
StartCall(); |
|
|
|
|
mu_{mu}, |
|
|
|
|
cv_{cv}, |
|
|
|
|
done_(done) { |
|
|
|
|
gpr_log(GPR_INFO, "client enter"); |
|
|
|
|
msgs_size_ = state.range(0); |
|
|
|
|
msgs_to_send_ = state.range(1); |
|
|
|
|
cli_ctx_ = new ClientContext(); |
|
|
|
|
cli_ctx_->AddMetadata(kServerFinishAfterNReads, |
|
|
|
|
grpc::to_string(msgs_to_send_)); |
|
|
|
|
cli_ctx_->AddMetadata(kServerMessageSize, grpc::to_string(msgs_size_)); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
void OnReadDone(bool ok) override { |
|
|
|
|
if (!ok) { |
|
|
|
|
return; |
|
|
|
|
} |
|
|
|
|
if (ok && reads_complete_ < msgs_to_send_) { |
|
|
|
|
reads_complete_++; |
|
|
|
|
if (writes_complete_ < msgs_to_send_) { |
|
|
|
|
MaybeWrite(); |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
@ -79,9 +86,19 @@ class BidiClient |
|
|
|
|
|
|
|
|
|
void OnDone(const Status& s) override { |
|
|
|
|
GPR_ASSERT(s.ok()); |
|
|
|
|
std::unique_lock<std::mutex> l(mu_); |
|
|
|
|
done_ = true; |
|
|
|
|
cv_.notify_one(); |
|
|
|
|
if (state_.KeepRunning()) { |
|
|
|
|
count++; |
|
|
|
|
gpr_log(GPR_INFO, "client start %d rpc", count); |
|
|
|
|
BidiClient* test = |
|
|
|
|
new BidiClient(state_, stub_, request_, response_, mu_, cv_, done_); |
|
|
|
|
test->StartNewRpc(); |
|
|
|
|
} else { |
|
|
|
|
gpr_log(GPR_INFO, "client done"); |
|
|
|
|
std::unique_lock<std::mutex> l(mu_); |
|
|
|
|
done_ = true; |
|
|
|
|
cv_.notify_one(); |
|
|
|
|
} |
|
|
|
|
delete cli_ctx_; |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
void Await() { |
|
|
|
@ -91,24 +108,36 @@ class BidiClient |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
void StartNewRpc() { |
|
|
|
|
gpr_log(GPR_INFO, "%d rpc start", count); |
|
|
|
|
stub_->experimental_async()->BidiStream(cli_ctx_, this); |
|
|
|
|
gpr_log(GPR_INFO, "%d write start", count); |
|
|
|
|
MaybeWrite(); |
|
|
|
|
StartCall(); |
|
|
|
|
gpr_log(GPR_INFO, "%d call start", count); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
private: |
|
|
|
|
void MaybeWrite() { |
|
|
|
|
if (writes_complete_ == msgs_to_send_) { |
|
|
|
|
StartWritesDone(); |
|
|
|
|
} else { |
|
|
|
|
if (writes_complete_ < msgs_to_send_) { |
|
|
|
|
StartWrite(request_); |
|
|
|
|
} else { |
|
|
|
|
StartWritesDone(); |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
ClientContext* cli_ctx_; |
|
|
|
|
benchmark::State& state_; |
|
|
|
|
EchoTestService::Stub* stub_; |
|
|
|
|
EchoRequest* request_; |
|
|
|
|
EchoResponse* response_; |
|
|
|
|
ClientContext* context_; |
|
|
|
|
int reads_complete_{0}; |
|
|
|
|
int writes_complete_{0}; |
|
|
|
|
const int msgs_to_send_; |
|
|
|
|
std::mutex mu_; |
|
|
|
|
std::condition_variable cv_; |
|
|
|
|
bool done_ = false; |
|
|
|
|
int msgs_to_send_; |
|
|
|
|
int msgs_size_; |
|
|
|
|
int count{0}; |
|
|
|
|
std::mutex& mu_; |
|
|
|
|
std::condition_variable& cv_; |
|
|
|
|
bool& done_; |
|
|
|
|
}; |
|
|
|
|
|
|
|
|
|
} // namespace testing
|
|
|
|
|