Handle errors better on client

pull/1240/head
Craig Tiller 10 years ago
parent aa2fca5a48
commit 8a5a666ad0
  1. 32
      test/cpp/qps/client.h
  2. 8
      test/cpp/qps/client_async.cc
  3. 7
      test/cpp/qps/client_sync.cc

@ -104,7 +104,7 @@ class Client {
void EndThreads() { threads_.clear(); }
virtual void ThreadFunc(Histogram* histogram, size_t thread_idx) = 0;
virtual bool ThreadFunc(Histogram* histogram, size_t thread_idx) = 0;
private:
class Thread {
@ -113,20 +113,24 @@ class Client {
: done_(false),
new_(nullptr),
impl_([this, idx, client]() {
for (;;) {
// run the loop body
client->ThreadFunc(&histogram_, idx);
// lock, see if we're done
std::lock_guard<std::mutex> g(mu_);
if (done_) {return;}
// check if we're marking, swap out the histogram if so
if (new_) {
new_->Swap(&histogram_);
new_ = nullptr;
cv_.notify_one();
for (;;) {
// run the loop body
bool thread_still_ok = client->ThreadFunc(&histogram_, idx);
// lock, see if we're done
std::lock_guard<std::mutex> g(mu_);
if (!thread_still_ok) {
gpr_log(GPR_ERROR, "Finishing client thread due to RPC error");
done_ = true;
}
if (done_) {return;}
// check if we're marking, swap out the histogram if so
if (new_) {
new_->Swap(&histogram_);
new_ = nullptr;
cv_.notify_one();
}
}
}
}) {}
}) {}
~Thread() {
{

@ -173,7 +173,7 @@ class AsyncUnaryClient GRPC_FINAL : public Client {
}
}
void ThreadFunc(Histogram* histogram, size_t thread_idx) GRPC_OVERRIDE {
bool ThreadFunc(Histogram* histogram, size_t thread_idx) GRPC_OVERRIDE {
void* got_tag;
bool ok;
cli_cqs_[thread_idx]->Next(&got_tag, &ok);
@ -185,6 +185,8 @@ class AsyncUnaryClient GRPC_FINAL : public Client {
ctx->StartNewClone();
delete ctx;
}
return true;
}
std::vector<std::unique_ptr<CompletionQueue>> cli_cqs_;
@ -301,7 +303,7 @@ class AsyncStreamingClient GRPC_FINAL : public Client {
}
}
void ThreadFunc(Histogram *histogram, size_t thread_idx) GRPC_OVERRIDE {
bool ThreadFunc(Histogram *histogram, size_t thread_idx) GRPC_OVERRIDE {
void *got_tag;
bool ok;
cli_cqs_[thread_idx]->Next(&got_tag, &ok);
@ -313,6 +315,8 @@ class AsyncStreamingClient GRPC_FINAL : public Client {
ctx->StartNewClone();
delete ctx;
}
return true;
}
std::vector<std::unique_ptr<CompletionQueue>> cli_cqs_;

@ -83,13 +83,14 @@ class SynchronousUnaryClient GRPC_FINAL : public SynchronousClient {
SynchronousClient(config) {StartThreads(num_threads_);}
~SynchronousUnaryClient() {}
void ThreadFunc(Histogram* histogram, size_t thread_idx) GRPC_OVERRIDE {
bool ThreadFunc(Histogram* histogram, size_t thread_idx) GRPC_OVERRIDE {
auto* stub = channels_[thread_idx % channels_.size()].get_stub();
double start = Timer::Now();
grpc::ClientContext context;
grpc::Status s =
stub->UnaryCall(&context, request_, &responses_[thread_idx]);
histogram->Add((Timer::Now() - start) * 1e9);
return s.IsOk();
}
};
@ -111,11 +112,13 @@ class SynchronousStreamingClient GRPC_FINAL : public SynchronousClient {
}
}
void ThreadFunc(Histogram* histogram, size_t thread_idx) GRPC_OVERRIDE {
bool ThreadFunc(Histogram* histogram, size_t thread_idx) GRPC_OVERRIDE {
double start = Timer::Now();
if (stream_->Write(request_) && stream_->Read(&responses_[thread_idx])) {
histogram->Add((Timer::Now() - start) * 1e9);
return true;
}
return false;
}
private:
grpc::ClientContext context_;

Loading…
Cancel
Save