Revision control
Copy as Markdown
// Copyright (C) 2024 The Qt Company Ltd.
// SPDX-License-Identifier: LicenseRef-Qt-Commercial OR GPL-3.0-only
#include "rpcjobs.h"
#include <qrpcbench_common.h>
#include <grpcpp/completion_queue.h>
#include <grpcpp/security/server_credentials.h>
#include <grpcpp/security/tls_certificate_provider.h>
#include <grpcpp/security/tls_credentials_options.h>
#include <grpcpp/server.h>
#include <grpcpp/server_builder.h>
#include <atomic>
#include <cassert>
#include <memory>
#include <string>
#include <string_view>
#include <vector>
#include <thread>
class AsyncServer
{
public:
enum class State { Created, Running, Finished };
void run(const QList<QString> &transports)
{
constexpr int numCompletionQueues = 1; // scale to ~numCPUs/2 for better throughput
constexpr int threadsPerQueue = 1; // Increase if CQ processing is the bottleneck
constexpr int initialInstancesPerQueue = 1; // raise for high concurrency
State expected = State::Created;
while (!mState.compare_exchange_weak(expected, State::Running)) {
if (expected != State::Created)
return;
}
{
grpc::ServerBuilder builder;
std::shared_ptr<grpc::ServerCredentials> tcpCreds = grpc::InsecureServerCredentials();
std::shared_ptr<grpc::ServerCredentials> tlsCreds;
{
auto provider = std::make_shared<grpc::experimental::InMemoryCertificateProvider>();
std::string sslRootKey(Bench::SslRootKey);
auto isOk = provider->UpdateRoot(sslRootKey).ok();
if (!isOk) {
std::cerr << "failed to update SslRootKey\n";
std::terminate();
}
std::vector<grpc::experimental::IdentityKeyOrSignerCertPair> identityPairs;
identityPairs.emplace_back(grpc::experimental::IdentityKeyOrSignerCertPair{
.private_key = std::string(Bench::SslKey.data(), Bench::SslKey.length()),
.certificate_chain = std::string(Bench::SslCert.data(),
Bench::SslCert.length()),
});
isOk = provider->UpdateIdentityKeyCertPair(std::move(identityPairs)).ok();
if (!isOk) {
std::cerr << "Failed to update identity key\n";
std::terminate();
}
auto tlsOpts = grpc::experimental::TlsServerCredentialsOptions::Create(provider)
.value();
tlsOpts.set_root_certificate_provider(provider);
// Needed for TLS debugging in wireshark (Edit > Preferences >
// Protocol > TLS > Master-Secret log filename)
tlsOpts.set_tls_session_key_log_file_path("sslkeylog.log");
tlsOpts.set_cert_request_type(GRPC_SSL_REQUEST_CLIENT_CERTIFICATE_AND_VERIFY);
tlsCreds = grpc::experimental::TlsServerCredentials(tlsOpts);
}
for (const auto &t : transports) {
const auto address = Bench::getTransportAddress(t);
if (t == "https")
builder.AddListeningPort(address, tlsCreds);
else
builder.AddListeningPort(address, tcpCreds);
std::cout << std::format("Server listening on: {}, {}\n", t.toStdString(), address);
}
builder.RegisterService(&mService);
for (int i = 0; i < numCompletionQueues; ++i)
mCompletionQueues.emplace_back(builder.AddCompletionQueue());
mServer = builder.BuildAndStart();
}
for (auto &cq : mCompletionQueues) {
for (int i = 0; i < initialInstancesPerQueue; ++i) {
new UnaryCall(cq.get(), &mService);
new ServerStreaming(cq.get(), &mService);
new ClientStreaming(cq.get(), &mService);
new BiDiStreaming(cq.get(), &mService);
}
}
for (auto &cq : mCompletionQueues) {
for (int i = 0; i < threadsPerQueue; ++i)
mThreads.emplace_back([cq = cq.get()] { processRPCs(cq); });
}
}
private:
qt::bench::BenchmarkService::AsyncService mService;
std::unique_ptr<grpc::Server> mServer;
std::vector<std::unique_ptr<grpc::ServerCompletionQueue>> mCompletionQueues;
std::vector<std::jthread> mThreads;
std::atomic<State> mState = { State::Created };
static_assert(std::atomic<State>::is_always_lock_free);
// A single thread works on the completion queue.
static void processRPCs(grpc::ServerCompletionQueue *cq)
{
assert(cq);
void *rawTag = nullptr;
bool ok = false;
while (cq->Next(&rawTag, &ok))
static_cast<RpcJob *>(rawTag)->onProcess(ok);
}
};
int main(int argc, char *argv[])
{
QStringList args;
for (int i = 0; i < argc; ++i)
args.push_back(argv[i]);
QCommandLineParser parser;
parser.setApplicationDescription("Asyncbenchserver");
parser.addHelpOption();
QCommandLineOption transport({ "t", "transport" }, "Use Transport(s)", "http|https");
#ifdef Q_OS_WINDOWS
transport.setDefaultValues({ "http", "https" });
#else
transport.setValueName("http|https|unix");
transport.setDefaultValues({ "http", "https", "unix" });
#endif
parser.addOption(transport);
parser.process(args);
AsyncServer server;
server.run(parser.values(transport));
}