mirror of https://github.com/grpc/grpc.git
Merge pull request #1241 from ctiller/local_worker
Allow RunScenarios to spawn in-process workerspull/1276/head
Vijay Pai
10 years ago
commit
4bfcf5b592
11 changed files with 385 additions and 236 deletions
@ -0,0 +1,233 @@ |
||||
/*
|
||||
* |
||||
* Copyright 2015, Google Inc. |
||||
* All rights reserved. |
||||
* |
||||
* Redistribution and use in source and binary forms, with or without |
||||
* modification, are permitted provided that the following conditions are |
||||
* met: |
||||
* |
||||
* * Redistributions of source code must retain the above copyright |
||||
* notice, this list of conditions and the following disclaimer. |
||||
* * Redistributions in binary form must reproduce the above |
||||
* copyright notice, this list of conditions and the following disclaimer |
||||
* in the documentation and/or other materials provided with the |
||||
* distribution. |
||||
* * Neither the name of Google Inc. nor the names of its |
||||
* contributors may be used to endorse or promote products derived from |
||||
* this software without specific prior written permission. |
||||
* |
||||
* THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS |
||||
* "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT |
||||
* LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR |
||||
* A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT |
||||
* OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, |
||||
* SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT |
||||
* LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, |
||||
* DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY |
||||
* THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT |
||||
* (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE |
||||
* OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. |
||||
* |
||||
*/ |
||||
|
||||
#include "qps_worker.h" |
||||
|
||||
#include <cassert> |
||||
#include <memory> |
||||
#include <mutex> |
||||
#include <string> |
||||
#include <thread> |
||||
#include <vector> |
||||
#include <sstream> |
||||
|
||||
#include <grpc/grpc.h> |
||||
#include <grpc/support/alloc.h> |
||||
#include <grpc/support/histogram.h> |
||||
#include <grpc/support/log.h> |
||||
#include <grpc/support/host_port.h> |
||||
#include <grpc++/client_context.h> |
||||
#include <grpc++/status.h> |
||||
#include <grpc++/server.h> |
||||
#include <grpc++/server_builder.h> |
||||
#include <grpc++/server_credentials.h> |
||||
#include <grpc++/stream.h> |
||||
#include "test/core/util/grpc_profiler.h" |
||||
#include "test/cpp/util/create_test_channel.h" |
||||
#include "test/cpp/qps/qpstest.pb.h" |
||||
#include "test/cpp/qps/client.h" |
||||
#include "test/cpp/qps/server.h" |
||||
|
||||
namespace grpc { |
||||
namespace testing { |
||||
|
||||
std::unique_ptr<Client> CreateClient(const ClientConfig& config) { |
||||
switch (config.client_type()) { |
||||
case ClientType::SYNCHRONOUS_CLIENT: |
||||
return (config.rpc_type() == RpcType::UNARY) ? |
||||
CreateSynchronousUnaryClient(config) : |
||||
CreateSynchronousStreamingClient(config); |
||||
case ClientType::ASYNC_CLIENT: |
||||
return (config.rpc_type() == RpcType::UNARY) ? |
||||
CreateAsyncUnaryClient(config) : CreateAsyncStreamingClient(config); |
||||
} |
||||
abort(); |
||||
} |
||||
|
||||
std::unique_ptr<Server> CreateServer(const ServerConfig& config, int server_port) { |
||||
switch (config.server_type()) { |
||||
case ServerType::SYNCHRONOUS_SERVER: |
||||
return CreateSynchronousServer(config, server_port); |
||||
case ServerType::ASYNC_SERVER: |
||||
return CreateAsyncServer(config, server_port); |
||||
} |
||||
abort(); |
||||
} |
||||
|
||||
class WorkerImpl GRPC_FINAL : public Worker::Service { |
||||
public: |
||||
explicit WorkerImpl(int server_port) : server_port_(server_port), acquired_(false) {} |
||||
|
||||
Status RunTest(ServerContext* ctx, |
||||
ServerReaderWriter<ClientStatus, ClientArgs>* stream) |
||||
GRPC_OVERRIDE { |
||||
InstanceGuard g(this); |
||||
if (!g.Acquired()) { |
||||
return Status(RESOURCE_EXHAUSTED); |
||||
} |
||||
|
||||
grpc_profiler_start("qps_client.prof"); |
||||
Status ret = RunTestBody(ctx,stream); |
||||
grpc_profiler_stop(); |
||||
return ret; |
||||
} |
||||
|
||||
Status RunServer(ServerContext* ctx, |
||||
ServerReaderWriter<ServerStatus, ServerArgs>* stream) |
||||
GRPC_OVERRIDE { |
||||
InstanceGuard g(this); |
||||
if (!g.Acquired()) { |
||||
return Status(RESOURCE_EXHAUSTED); |
||||
} |
||||
|
||||
grpc_profiler_start("qps_server.prof"); |
||||
Status ret = RunServerBody(ctx,stream); |
||||
grpc_profiler_stop(); |
||||
return ret; |
||||
} |
||||
|
||||
private: |
||||
// Protect against multiple clients using this worker at once.
|
||||
class InstanceGuard { |
||||
public: |
||||
InstanceGuard(WorkerImpl* impl) |
||||
: impl_(impl), acquired_(impl->TryAcquireInstance()) {} |
||||
~InstanceGuard() { |
||||
if (acquired_) { |
||||
impl_->ReleaseInstance(); |
||||
} |
||||
} |
||||
|
||||
bool Acquired() const { return acquired_; } |
||||
|
||||
private: |
||||
WorkerImpl* const impl_; |
||||
const bool acquired_; |
||||
}; |
||||
|
||||
bool TryAcquireInstance() { |
||||
std::lock_guard<std::mutex> g(mu_); |
||||
if (acquired_) return false; |
||||
acquired_ = true; |
||||
return true; |
||||
} |
||||
|
||||
void ReleaseInstance() { |
||||
std::lock_guard<std::mutex> g(mu_); |
||||
GPR_ASSERT(acquired_); |
||||
acquired_ = false; |
||||
} |
||||
|
||||
Status RunTestBody(ServerContext* ctx, |
||||
ServerReaderWriter<ClientStatus, ClientArgs>* stream) { |
||||
ClientArgs args; |
||||
if (!stream->Read(&args)) { |
||||
return Status(INVALID_ARGUMENT); |
||||
} |
||||
if (!args.has_setup()) { |
||||
return Status(INVALID_ARGUMENT); |
||||
} |
||||
auto client = CreateClient(args.setup()); |
||||
if (!client) { |
||||
return Status(INVALID_ARGUMENT); |
||||
} |
||||
ClientStatus status; |
||||
if (!stream->Write(status)) { |
||||
return Status(UNKNOWN); |
||||
} |
||||
while (stream->Read(&args)) { |
||||
if (!args.has_mark()) { |
||||
return Status(INVALID_ARGUMENT); |
||||
} |
||||
*status.mutable_stats() = client->Mark(); |
||||
stream->Write(status); |
||||
} |
||||
|
||||
return Status::OK; |
||||
} |
||||
|
||||
Status RunServerBody(ServerContext* ctx, |
||||
ServerReaderWriter<ServerStatus, ServerArgs>* stream) { |
||||
ServerArgs args; |
||||
if (!stream->Read(&args)) { |
||||
return Status(INVALID_ARGUMENT); |
||||
} |
||||
if (!args.has_setup()) { |
||||
return Status(INVALID_ARGUMENT); |
||||
} |
||||
auto server = CreateServer(args.setup(), server_port_); |
||||
if (!server) { |
||||
return Status(INVALID_ARGUMENT); |
||||
} |
||||
ServerStatus status; |
||||
status.set_port(server_port_); |
||||
if (!stream->Write(status)) { |
||||
return Status(UNKNOWN); |
||||
} |
||||
while (stream->Read(&args)) { |
||||
if (!args.has_mark()) { |
||||
return Status(INVALID_ARGUMENT); |
||||
} |
||||
*status.mutable_stats() = server->Mark(); |
||||
stream->Write(status); |
||||
} |
||||
|
||||
return Status::OK; |
||||
} |
||||
|
||||
const int server_port_; |
||||
|
||||
std::mutex mu_; |
||||
bool acquired_; |
||||
}; |
||||
|
||||
QpsWorker::QpsWorker(int driver_port, int server_port) { |
||||
impl_.reset(new WorkerImpl(server_port)); |
||||
|
||||
char* server_address = NULL; |
||||
gpr_join_host_port(&server_address, "::", driver_port); |
||||
|
||||
ServerBuilder builder; |
||||
builder.AddListeningPort(server_address, InsecureServerCredentials()); |
||||
builder.RegisterService(impl_.get()); |
||||
|
||||
gpr_free(server_address); |
||||
|
||||
server_ = std::move(builder.BuildAndStart()); |
||||
} |
||||
|
||||
QpsWorker::~QpsWorker() { |
||||
} |
||||
|
||||
} // namespace testing
|
||||
} // namespace grpc
|
||||
@ -0,0 +1,60 @@ |
||||
/*
|
||||
* |
||||
* Copyright 2015, Google Inc. |
||||
* All rights reserved. |
||||
* |
||||
* Redistribution and use in source and binary forms, with or without |
||||
* modification, are permitted provided that the following conditions are |
||||
* met: |
||||
* |
||||
* * Redistributions of source code must retain the above copyright |
||||
* notice, this list of conditions and the following disclaimer. |
||||
* * Redistributions in binary form must reproduce the above |
||||
* copyright notice, this list of conditions and the following disclaimer |
||||
* in the documentation and/or other materials provided with the |
||||
* distribution. |
||||
* * Neither the name of Google Inc. nor the names of its |
||||
* contributors may be used to endorse or promote products derived from |
||||
* this software without specific prior written permission. |
||||
* |
||||
* THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS |
||||
* "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT |
||||
* LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR |
||||
* A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT |
||||
* OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, |
||||
* SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT |
||||
* LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, |
||||
* DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY |
||||
* THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT |
||||
* (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE |
||||
* OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. |
||||
* |
||||
*/ |
||||
|
||||
#ifndef QPS_WORKER_H |
||||
#define QPS_WORKER_H |
||||
|
||||
#include <memory> |
||||
|
||||
namespace grpc { |
||||
|
||||
class Server; |
||||
|
||||
namespace testing { |
||||
|
||||
class WorkerImpl; |
||||
|
||||
class QpsWorker { |
||||
public: |
||||
QpsWorker(int driver_port, int server_port); |
||||
~QpsWorker(); |
||||
|
||||
private: |
||||
std::unique_ptr<WorkerImpl> impl_; |
||||
std::unique_ptr<Server> server_; |
||||
}; |
||||
|
||||
} // namespace testing
|
||||
} // namespace grpc
|
||||
|
||||
#endif |
||||
@ -1,28 +0,0 @@ |
||||
#!/bin/sh |
||||
|
||||
# performs a single qps run with one client and one server |
||||
|
||||
set -ex |
||||
|
||||
cd $(dirname $0)/../../.. |
||||
|
||||
killall qps_worker || true |
||||
|
||||
config=opt |
||||
|
||||
NUMCPUS=`python2.7 -c 'import multiprocessing; print multiprocessing.cpu_count()'` |
||||
|
||||
make CONFIG=$config qps_worker qps_smoke_test -j$NUMCPUS |
||||
|
||||
bins/$config/qps_worker -driver_port 10000 -server_port 10001 & |
||||
PID1=$! |
||||
bins/$config/qps_worker -driver_port 10010 -server_port 10011 & |
||||
PID2=$! |
||||
|
||||
export QPS_WORKERS="localhost:10000,localhost:10010" |
||||
|
||||
bins/$config/qps_smoke_test $* |
||||
|
||||
kill -2 $PID1 $PID2 |
||||
wait |
||||
|
||||
Loading…
Reference in new issue