aboutsummaryrefslogtreecommitdiffhomepage
path: root/test/cpp/qps
diff options
context:
space:
mode:
Diffstat (limited to 'test/cpp/qps')
-rw-r--r--test/cpp/qps/client_async.cc19
-rw-r--r--test/cpp/qps/qps_worker.cc2
-rw-r--r--test/cpp/qps/report.cc16
-rw-r--r--test/cpp/qps/report.h24
-rw-r--r--test/cpp/qps/server_async.cc35
-rw-r--r--test/cpp/qps/worker.cc2
6 files changed, 59 insertions, 39 deletions
diff --git a/test/cpp/qps/client_async.cc b/test/cpp/qps/client_async.cc
index 1b7a8d26b2..8c8d927d15 100644
--- a/test/cpp/qps/client_async.cc
+++ b/test/cpp/qps/client_async.cc
@@ -168,7 +168,7 @@ class AsyncClient : public Client {
if (!closed_loop_) {
rpc_deadlines_.emplace_back();
next_channel_.push_back(i % channel_count_);
- issue_allowed_.push_back(true);
+ issue_allowed_.emplace_back(true);
grpc_time next_issue;
NextIssueTime(i, &next_issue);
@@ -234,12 +234,6 @@ class AsyncClient : public Client {
GPR_ASSERT(false);
break;
}
- if ((closed_loop_ || !rpc_deadlines_[thread_idx].empty()) &&
- grpc_time_source::now() > deadline) {
- // we have missed some 1-second deadline, which is worth noting
- gpr_log(GPR_INFO, "Missed an RPC deadline");
- // Don't give up, as there might be some truly heavy tails
- }
if (got_event) {
ClientRpcContext* ctx = ClientRpcContext::detag(got_tag);
if (ctx->RunNextState(ok, histogram) == false) {
@@ -313,11 +307,20 @@ class AsyncClient : public Client {
}
private:
+ class boolean { // exists only to avoid data-race on vector<bool>
+ public:
+ boolean(): val_(false) {}
+ boolean(bool b): val_(b) {}
+ operator bool() const {return val_;}
+ boolean& operator=(bool b) {val_=b; return *this;}
+ private:
+ bool val_;
+ };
std::vector<std::unique_ptr<CompletionQueue>> cli_cqs_;
std::vector<deadline_list> rpc_deadlines_; // per thread deadlines
std::vector<int> next_channel_; // per thread round-robin channel ctr
- std::vector<bool> issue_allowed_; // may this thread attempt to issue
+ std::vector<boolean> issue_allowed_; // may this thread attempt to issue
std::vector<grpc_time> next_issue_; // when should it issue?
std::vector<std::mutex> channel_lock_;
diff --git a/test/cpp/qps/qps_worker.cc b/test/cpp/qps/qps_worker.cc
index 423275ee85..f1cea5ee66 100644
--- a/test/cpp/qps/qps_worker.cc
+++ b/test/cpp/qps/qps_worker.cc
@@ -31,7 +31,7 @@
*
*/
-#include "qps_worker.h"
+#include "test/cpp/qps/qps_worker.h"
#include <cassert>
#include <memory>
diff --git a/test/cpp/qps/report.cc b/test/cpp/qps/report.cc
index 678ea080d1..94aacdbd1c 100644
--- a/test/cpp/qps/report.cc
+++ b/test/cpp/qps/report.cc
@@ -43,39 +43,39 @@ void CompositeReporter::add(std::unique_ptr<Reporter> reporter) {
reporters_.emplace_back(std::move(reporter));
}
-void CompositeReporter::ReportQPS(const ScenarioResult& result) const {
+void CompositeReporter::ReportQPS(const ScenarioResult& result) {
for (size_t i = 0; i < reporters_.size(); ++i) {
reporters_[i]->ReportQPS(result);
}
}
-void CompositeReporter::ReportQPSPerCore(const ScenarioResult& result) const {
+void CompositeReporter::ReportQPSPerCore(const ScenarioResult& result) {
for (size_t i = 0; i < reporters_.size(); ++i) {
reporters_[i]->ReportQPSPerCore(result);
}
}
-void CompositeReporter::ReportLatency(const ScenarioResult& result) const {
+void CompositeReporter::ReportLatency(const ScenarioResult& result) {
for (size_t i = 0; i < reporters_.size(); ++i) {
reporters_[i]->ReportLatency(result);
}
}
-void CompositeReporter::ReportTimes(const ScenarioResult& result) const {
+void CompositeReporter::ReportTimes(const ScenarioResult& result) {
for (size_t i = 0; i < reporters_.size(); ++i) {
reporters_[i]->ReportTimes(result);
}
}
-void GprLogReporter::ReportQPS(const ScenarioResult& result) const {
+void GprLogReporter::ReportQPS(const ScenarioResult& result) {
gpr_log(GPR_INFO, "QPS: %.1f",
result.latencies.Count() /
average(result.client_resources,
[](ResourceUsage u) { return u.wall_time; }));
}
-void GprLogReporter::ReportQPSPerCore(const ScenarioResult& result) const {
+void GprLogReporter::ReportQPSPerCore(const ScenarioResult& result) {
auto qps =
result.latencies.Count() /
average(result.client_resources,
@@ -85,7 +85,7 @@ void GprLogReporter::ReportQPSPerCore(const ScenarioResult& result) const {
qps / result.server_config.threads());
}
-void GprLogReporter::ReportLatency(const ScenarioResult& result) const {
+void GprLogReporter::ReportLatency(const ScenarioResult& result) {
gpr_log(GPR_INFO,
"Latencies (50/90/95/99/99.9%%-ile): %.1f/%.1f/%.1f/%.1f/%.1f us",
result.latencies.Percentile(50) / 1000,
@@ -95,7 +95,7 @@ void GprLogReporter::ReportLatency(const ScenarioResult& result) const {
result.latencies.Percentile(99.9) / 1000);
}
-void GprLogReporter::ReportTimes(const ScenarioResult& result) const {
+void GprLogReporter::ReportTimes(const ScenarioResult& result) {
gpr_log(GPR_INFO, "Server system time: %.2f%%",
100.0 * sum(result.server_resources,
[](ResourceUsage u) { return u.system_time; }) /
diff --git a/test/cpp/qps/report.h b/test/cpp/qps/report.h
index 0cce08816a..b1cf83fc23 100644
--- a/test/cpp/qps/report.h
+++ b/test/cpp/qps/report.h
@@ -59,16 +59,16 @@ class Reporter {
string name() const { return name_; }
/** Reports QPS for the given \a result. */
- virtual void ReportQPS(const ScenarioResult& result) const = 0;
+ virtual void ReportQPS(const ScenarioResult& result) = 0;
/** Reports QPS per core as (YYY/server core). */
- virtual void ReportQPSPerCore(const ScenarioResult& result) const = 0;
+ virtual void ReportQPSPerCore(const ScenarioResult& result) = 0;
/** Reports latencies for the 50, 90, 95, 99 and 99.9 percentiles, in ms. */
- virtual void ReportLatency(const ScenarioResult& result) const = 0;
+ virtual void ReportLatency(const ScenarioResult& result) = 0;
/** Reports system and user time for client and server systems. */
- virtual void ReportTimes(const ScenarioResult& result) const = 0;
+ virtual void ReportTimes(const ScenarioResult& result) = 0;
private:
const string name_;
@@ -82,10 +82,10 @@ class CompositeReporter : public Reporter {
/** Adds a \a reporter to the composite. */
void add(std::unique_ptr<Reporter> reporter);
- void ReportQPS(const ScenarioResult& result) const GRPC_OVERRIDE;
- void ReportQPSPerCore(const ScenarioResult& result) const GRPC_OVERRIDE;
- void ReportLatency(const ScenarioResult& result) const GRPC_OVERRIDE;
- void ReportTimes(const ScenarioResult& result) const GRPC_OVERRIDE;
+ void ReportQPS(const ScenarioResult& result) GRPC_OVERRIDE;
+ void ReportQPSPerCore(const ScenarioResult& result) GRPC_OVERRIDE;
+ void ReportLatency(const ScenarioResult& result) GRPC_OVERRIDE;
+ void ReportTimes(const ScenarioResult& result) GRPC_OVERRIDE;
private:
std::vector<std::unique_ptr<Reporter> > reporters_;
@@ -97,10 +97,10 @@ class GprLogReporter : public Reporter {
GprLogReporter(const string& name) : Reporter(name) {}
private:
- void ReportQPS(const ScenarioResult& result) const GRPC_OVERRIDE;
- void ReportQPSPerCore(const ScenarioResult& result) const GRPC_OVERRIDE;
- void ReportLatency(const ScenarioResult& result) const GRPC_OVERRIDE;
- void ReportTimes(const ScenarioResult& result) const GRPC_OVERRIDE;
+ void ReportQPS(const ScenarioResult& result) GRPC_OVERRIDE;
+ void ReportQPSPerCore(const ScenarioResult& result) GRPC_OVERRIDE;
+ void ReportLatency(const ScenarioResult& result) GRPC_OVERRIDE;
+ void ReportTimes(const ScenarioResult& result) GRPC_OVERRIDE;
};
} // namespace testing
diff --git a/test/cpp/qps/server_async.cc b/test/cpp/qps/server_async.cc
index 210aef4fd6..f5251e961b 100644
--- a/test/cpp/qps/server_async.cc
+++ b/test/cpp/qps/server_async.cc
@@ -64,7 +64,7 @@ namespace testing {
class AsyncQpsServerTest : public Server {
public:
- AsyncQpsServerTest(const ServerConfig &config, int port) : shutdown_(false) {
+ AsyncQpsServerTest(const ServerConfig &config, int port) {
char *server_address = NULL;
gpr_join_host_port(&server_address, "::", port);
@@ -97,6 +97,9 @@ class AsyncQpsServerTest : public Server {
}
}
for (int i = 0; i < config.threads(); i++) {
+ shutdown_state_.emplace_back(new PerThreadShutdownState());
+ }
+ for (int i = 0; i < config.threads(); i++) {
threads_.push_back(std::thread([=]() {
// Wait until work is available or we are shutting down
bool ok;
@@ -105,11 +108,9 @@ class AsyncQpsServerTest : public Server {
ServerRpcContext *ctx = detag(got_tag);
// The tag is a pointer to an RPC context to invoke
bool still_going = ctx->RunNextState(ok);
- std::unique_lock<std::mutex> g(shutdown_mutex_);
- if (!shutdown_) {
+ if (!shutdown_state_[i]->shutdown()) {
// this RPC context is done, so refresh it
if (!still_going) {
- g.unlock();
ctx->Reset();
}
} else {
@@ -122,9 +123,8 @@ class AsyncQpsServerTest : public Server {
}
~AsyncQpsServerTest() {
server_->Shutdown();
- {
- std::lock_guard<std::mutex> g(shutdown_mutex_);
- shutdown_ = true;
+ for (auto ss = shutdown_state_.begin(); ss != shutdown_state_.end(); ++ss) {
+ (*ss)->set_shutdown();
}
for (auto thr = threads_.begin(); thr != threads_.end(); thr++) {
thr->join();
@@ -316,8 +316,25 @@ class AsyncQpsServerTest : public Server {
TestService::AsyncService async_service_;
std::forward_list<ServerRpcContext *> contexts_;
- std::mutex shutdown_mutex_;
- bool shutdown_;
+ class PerThreadShutdownState {
+ public:
+ PerThreadShutdownState() : shutdown_(false) {}
+
+ bool shutdown() const {
+ std::lock_guard<std::mutex> lock(mutex_);
+ return shutdown_;
+ }
+
+ void set_shutdown() {
+ std::lock_guard<std::mutex> lock(mutex_);
+ shutdown_ = true;
+ }
+
+ private:
+ mutable std::mutex mutex_;
+ bool shutdown_;
+ };
+ std::vector<std::unique_ptr<PerThreadShutdownState>> shutdown_state_;
};
std::unique_ptr<Server> CreateAsyncServer(const ServerConfig &config,
diff --git a/test/cpp/qps/worker.cc b/test/cpp/qps/worker.cc
index dfc102fc17..14a8b0b089 100644
--- a/test/cpp/qps/worker.cc
+++ b/test/cpp/qps/worker.cc
@@ -40,7 +40,7 @@
#include <grpc/support/time.h>
#include <gflags/gflags.h>
-#include "qps_worker.h"
+#include "test/cpp/qps/qps_worker.h"
#include "test/cpp/util/test_config.h"
DEFINE_int32(driver_port, 0, "Driver server port.");
generated by cgit v1.2.3 (git 2.39.1) at 2025-02-08 14:00:42 +0000