You are viewing a plain text version of this content. The canonical link for it is here.
Posted to common-commits@hadoop.apache.org by jh...@apache.org on 2018/03/22 21:21:54 UTC
[51/52] [abbrv] hadoop git commit: HDFS-12427: libhdfs++: Prevent
requests from holding dangling pointer to RpcEngine. Contributed by James
Clampffer.
HDFS-12427: libhdfs++: Prevent requests from holding dangling pointer to RpcEngine. Contributed by James Clampffer.
Project: http://git-wip-us.apache.org/repos/asf/hadoop/repo
Commit: http://git-wip-us.apache.org/repos/asf/hadoop/commit/fc5e44d9
Tree: http://git-wip-us.apache.org/repos/asf/hadoop/tree/fc5e44d9
Diff: http://git-wip-us.apache.org/repos/asf/hadoop/diff/fc5e44d9
Branch: refs/heads/HDFS-8707
Commit: fc5e44d9edeaad9c1381a46fb370b167ac6250b1
Parents: 48db24a
Author: James <jh...@apache.org>
Authored: Mon Oct 30 13:54:05 2017 -0400
Committer: James Clampffer <ja...@hp.com>
Committed: Thu Mar 22 17:19:47 2018 -0400
----------------------------------------------------------------------
.../main/native/libhdfspp/lib/fs/filesystem.h | 2 +
.../libhdfspp/lib/fs/namenode_operations.cc | 6 +-
.../libhdfspp/lib/fs/namenode_operations.h | 14 +-
.../libhdfspp/lib/proto/protoc_gen_hrpc.cc | 4 +-
.../main/native/libhdfspp/lib/rpc/request.cc | 61 ++++-----
.../src/main/native/libhdfspp/lib/rpc/request.h | 10 +-
.../native/libhdfspp/lib/rpc/rpc_connection.h | 17 +--
.../libhdfspp/lib/rpc/rpc_connection_impl.cc | 131 +++++++++++++------
.../libhdfspp/lib/rpc/rpc_connection_impl.h | 54 +++++---
.../main/native/libhdfspp/lib/rpc/rpc_engine.cc | 2 +-
.../main/native/libhdfspp/lib/rpc/rpc_engine.h | 30 ++---
.../native/libhdfspp/tests/rpc_engine_test.cc | 69 +++++-----
12 files changed, 240 insertions(+), 160 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/hadoop/blob/fc5e44d9/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/fs/filesystem.h
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/fs/filesystem.h b/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/fs/filesystem.h
index ac6c3d8..f2e9abd 100644
--- a/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/fs/filesystem.h
+++ b/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/fs/filesystem.h
@@ -212,6 +212,8 @@ private:
* The IoService must be the first member variable to ensure that it gets
* destroyed last. This allows other members to dequeue things from the
* service in their own destructors.
+ * A side effect of this is that requests may outlive the RpcEngine they
+ * reference.
**/
std::shared_ptr<IoServiceImpl> io_service_;
const Options options_;
http://git-wip-us.apache.org/repos/asf/hadoop/blob/fc5e44d9/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/fs/namenode_operations.cc
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/fs/namenode_operations.cc b/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/fs/namenode_operations.cc
index b5a6564..e46faad 100644
--- a/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/fs/namenode_operations.cc
+++ b/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/fs/namenode_operations.cc
@@ -42,11 +42,11 @@ namespace hdfs {
void NameNodeOperations::Connect(const std::string &cluster_name,
const std::vector<ResolvedNamenodeInfo> &servers,
std::function<void(const Status &)> &&handler) {
- engine_.Connect(cluster_name, servers, handler);
+ engine_->Connect(cluster_name, servers, handler);
}
bool NameNodeOperations::CancelPendingConnect() {
- return engine_.CancelPendingConnect();
+ return engine_->CancelPendingConnect();
}
void NameNodeOperations::GetBlockLocations(const std::string & path, uint64_t offset, uint64_t length,
@@ -678,7 +678,7 @@ void NameNodeOperations::DisallowSnapshot(const std::string & path, std::functio
}
void NameNodeOperations::SetFsEventCallback(fs_event_callback callback) {
- engine_.SetFsEventCallback(callback);
+ engine_->SetFsEventCallback(callback);
}
void NameNodeOperations::HdfsFileStatusProtoToStatInfo(
http://git-wip-us.apache.org/repos/asf/hadoop/blob/fc5e44d9/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/fs/namenode_operations.h
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/fs/namenode_operations.h b/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/fs/namenode_operations.h
index 97c0e18..f4caa18 100644
--- a/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/fs/namenode_operations.h
+++ b/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/fs/namenode_operations.h
@@ -46,8 +46,9 @@ public:
const std::string &client_name, const std::string &user_name,
const char *protocol_name, int protocol_version) :
io_service_(io_service),
- engine_(io_service, options, client_name, user_name, protocol_name, protocol_version),
- namenode_(& engine_), options_(options) {}
+ engine_(std::make_shared<RpcEngine>(io_service, options, client_name, user_name, protocol_name, protocol_version)),
+ namenode_(engine_), options_(options) {}
+
void Connect(const std::string &cluster_name,
const std::vector<ResolvedNamenodeInfo> &servers,
@@ -119,7 +120,14 @@ private:
static void GetFsStatsResponseProtoToFsInfo(hdfs::FsInfo & fs_info, const std::shared_ptr<::hadoop::hdfs::GetFsStatsResponseProto> & fs);
::asio::io_service * io_service_;
- RpcEngine engine_;
+
+ // This is the only permanent owner of the RpcEngine, however the RPC layer
+ // needs to reference count it prevent races during FileSystem destruction.
+ // In order to do this they hold weak_ptrs and promote them to shared_ptr
+ // when calling non-blocking RpcEngine methods e.g. get_client_id().
+ std::shared_ptr<RpcEngine> engine_;
+
+ // Automatically generated methods for RPC calls. See protoc_gen_hrpc.cc
ClientNamenodeProtocol namenode_;
const Options options_;
};
http://git-wip-us.apache.org/repos/asf/hadoop/blob/fc5e44d9/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/proto/protoc_gen_hrpc.cc
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/proto/protoc_gen_hrpc.cc b/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/proto/protoc_gen_hrpc.cc
index bb0d8d8..e7355c0 100644
--- a/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/proto/protoc_gen_hrpc.cc
+++ b/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/proto/protoc_gen_hrpc.cc
@@ -64,11 +64,11 @@ void StubGenerator::EmitService(const ServiceDescriptor *service,
out->Print("\n// GENERATED AUTOMATICALLY. DO NOT MODIFY.\n"
"class $service$ {\n"
"private:\n"
- " ::hdfs::RpcEngine *const engine_;\n"
+ " std::shared_ptr<::hdfs::RpcEngine> engine_;\n"
"public:\n"
" typedef std::function<void(const ::hdfs::Status &)> Callback;\n"
" typedef ::google::protobuf::MessageLite Message;\n"
- " inline $service$(::hdfs::RpcEngine *engine)\n"
+ " inline $service$(std::shared_ptr<::hdfs::RpcEngine> engine)\n"
" : engine_(engine) {}\n",
"service", service->name());
for (int i = 0; i < service->method_count(); ++i) {
http://git-wip-us.apache.org/repos/asf/hadoop/blob/fc5e44d9/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/rpc/request.cc
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/rpc/request.cc b/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/rpc/request.cc
index 119962a..356411e 100644
--- a/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/rpc/request.cc
+++ b/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/rpc/request.cc
@@ -38,6 +38,11 @@ using namespace ::std::placeholders;
static const int kNoRetry = -1;
// Protobuf helper functions.
+// Note/todo: Using the zero-copy protobuf API here makes the simple procedures
+// below tricky to read and debug while providing minimal benefit. Reducing
+// allocations in BlockReader (HDFS-11266) and smarter use of std::stringstream
+// will have a much larger impact according to cachegrind profiles on common
+// workloads.
static void AddHeadersToPacket(std::string *res,
std::initializer_list<const pb::MessageLite *> headers,
const std::string *payload) {
@@ -82,50 +87,33 @@ static void ConstructPayload(std::string *res, const pb::MessageLite *header) {
buf = header->SerializeWithCachedSizesToArray(buf);
}
-static void ConstructPayload(std::string *res, const std::string *request) {
- int len =
- pbio::CodedOutputStream::VarintSize32(request->size()) + request->size();
- res->reserve(len);
- pbio::StringOutputStream ss(res);
- pbio::CodedOutputStream os(&ss);
- uint8_t *buf = os.GetDirectBufferForNBytesAndAdvance(len);
- assert(buf);
- buf = pbio::CodedOutputStream::WriteVarint32ToArray(request->size(), buf);
- buf = os.WriteStringToArray(*request, buf);
-}
-
-static void SetRequestHeader(LockFreeRpcEngine *engine, int call_id,
+static void SetRequestHeader(std::weak_ptr<LockFreeRpcEngine> weak_engine, int call_id,
const std::string &method_name, int retry_count,
RpcRequestHeaderProto *rpc_header,
- RequestHeaderProto *req_header) {
+ RequestHeaderProto *req_header)
+{
+ // Ensure the RpcEngine is live. If it's not then the FileSystem is being destructed.
+ std::shared_ptr<LockFreeRpcEngine> counted_engine = weak_engine.lock();
+ if(!counted_engine) {
+ LOG_ERROR(kRPC, << "SetRequestHeader attempted to access an invalid RpcEngine");
+ return;
+ }
+
rpc_header->set_rpckind(RPC_PROTOCOL_BUFFER);
rpc_header->set_rpcop(RpcRequestHeaderProto::RPC_FINAL_PACKET);
rpc_header->set_callid(call_id);
- if (retry_count != kNoRetry)
+ if (retry_count != kNoRetry) {
rpc_header->set_retrycount(retry_count);
- rpc_header->set_clientid(engine->client_id());
-
+ }
+ rpc_header->set_clientid(counted_engine->client_id());
req_header->set_methodname(method_name);
- req_header->set_declaringclassprotocolname(engine->protocol_name());
- req_header->set_clientprotocolversion(engine->protocol_version());
+ req_header->set_declaringclassprotocolname(counted_engine->protocol_name());
+ req_header->set_clientprotocolversion(counted_engine->protocol_version());
}
// Request implementation
-Request::Request(LockFreeRpcEngine *engine, const std::string &method_name, int call_id,
- const std::string &request, Handler &&handler)
- : engine_(engine),
- method_name_(method_name),
- call_id_(call_id),
- timer_(engine->io_service()),
- handler_(std::move(handler)),
- retry_count_(engine->retry_policy() ? 0 : kNoRetry),
- failover_count_(0) {
- ConstructPayload(&payload_, &request);
-}
-
-
-Request::Request(LockFreeRpcEngine *engine, const std::string &method_name, int call_id,
+Request::Request(std::shared_ptr<LockFreeRpcEngine> engine, const std::string &method_name, int call_id,
const pb::MessageLite *request, Handler &&handler)
: engine_(engine),
method_name_(method_name),
@@ -133,13 +121,14 @@ Request::Request(LockFreeRpcEngine *engine, const std::string &method_name, int
timer_(engine->io_service()),
handler_(std::move(handler)),
retry_count_(engine->retry_policy() ? 0 : kNoRetry),
- failover_count_(0) {
+ failover_count_(0)
+{
ConstructPayload(&payload_, request);
}
-Request::Request(LockFreeRpcEngine *engine, Handler &&handler)
+Request::Request(std::shared_ptr<LockFreeRpcEngine> engine, Handler &&handler)
: engine_(engine),
- call_id_(-1),
+ call_id_(-1/*Handshake ID*/),
timer_(engine->io_service()),
handler_(std::move(handler)),
retry_count_(engine->retry_policy() ? 0 : kNoRetry),
http://git-wip-us.apache.org/repos/asf/hadoop/blob/fc5e44d9/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/rpc/request.h
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/rpc/request.h b/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/rpc/request.h
index d265a4c..f195540 100644
--- a/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/rpc/request.h
+++ b/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/rpc/request.h
@@ -23,6 +23,7 @@
#include "common/new_delete.h"
#include <string>
+#include <memory>
#include <google/protobuf/message_lite.h>
#include <google/protobuf/io/coded_stream.h>
@@ -48,14 +49,13 @@ class Request {
typedef std::function<void(::google::protobuf::io::CodedInputStream *is,
const Status &status)> Handler;
- Request(LockFreeRpcEngine *engine, const std::string &method_name, int call_id,
- const std::string &request, Handler &&callback);
- Request(LockFreeRpcEngine *engine, const std::string &method_name, int call_id,
+ // Constructors will not make any blocking calls while holding the shared_ptr<RpcEngine>
+ Request(std::shared_ptr<LockFreeRpcEngine> engine, const std::string &method_name, int call_id,
const ::google::protobuf::MessageLite *request, Handler &&callback);
// Null request (with no actual message) used to track the state of an
// initial Connect call
- Request(LockFreeRpcEngine *engine, Handler &&handler);
+ Request(std::shared_ptr<LockFreeRpcEngine> engine, Handler &&handler);
int call_id() const { return call_id_; }
std::string method_name() const { return method_name_; }
@@ -71,7 +71,7 @@ class Request {
std::string GetDebugString() const;
private:
- LockFreeRpcEngine *const engine_;
+ std::weak_ptr<LockFreeRpcEngine> engine_;
const std::string method_name_;
const int call_id_;
http://git-wip-us.apache.org/repos/asf/hadoop/blob/fc5e44d9/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/rpc/rpc_connection.h
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/rpc/rpc_connection.h b/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/rpc/rpc_connection.h
index 449a1b6..9e54983 100644
--- a/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/rpc/rpc_connection.h
+++ b/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/rpc/rpc_connection.h
@@ -52,7 +52,7 @@ class SaslProtocol;
class RpcConnection : public std::enable_shared_from_this<RpcConnection> {
public:
MEMCHECKED_CLASS(RpcConnection)
- RpcConnection(LockFreeRpcEngine *engine);
+ RpcConnection(std::shared_ptr<LockFreeRpcEngine> engine);
virtual ~RpcConnection();
// Note that a single server can have multiple endpoints - especially both
@@ -82,8 +82,8 @@ class RpcConnection : public std::enable_shared_from_this<RpcConnection> {
void SetClusterName(std::string cluster_name);
void SetAuthInfo(const AuthInfo& auth_info);
- LockFreeRpcEngine *engine() { return engine_; }
- ::asio::io_service &io_service();
+ std::weak_ptr<LockFreeRpcEngine> engine() { return engine_; }
+ ::asio::io_service *GetIoService();
protected:
struct Response {
@@ -139,7 +139,7 @@ class RpcConnection : public std::enable_shared_from_this<RpcConnection> {
void ClearAndDisconnect(const ::asio::error_code &ec);
std::shared_ptr<Request> RemoveFromRunningQueue(int call_id);
- LockFreeRpcEngine *const engine_;
+ std::weak_ptr<LockFreeRpcEngine> engine_;
std::shared_ptr<Response> current_response_state_;
AuthInfo auth_info_;
@@ -158,16 +158,17 @@ class RpcConnection : public std::enable_shared_from_this<RpcConnection> {
// State machine for performing a SASL handshake
std::shared_ptr<SaslProtocol> sasl_protocol_;
- // The request being sent over the wire; will also be in requests_on_fly_
- std::shared_ptr<Request> request_over_the_wire_;
+ // The request being sent over the wire; will also be in sent_requests_
+ std::shared_ptr<Request> outgoing_request_;
// Requests to be sent over the wire
std::deque<std::shared_ptr<Request>> pending_requests_;
// Requests to be sent over the wire during authentication; not retried if
// there is a connection error
std::deque<std::shared_ptr<Request>> auth_requests_;
// Requests that are waiting for responses
- typedef std::unordered_map<int, std::shared_ptr<Request>> RequestOnFlyMap;
- RequestOnFlyMap requests_on_fly_;
+ typedef std::unordered_map<int, std::shared_ptr<Request>> SentRequestMap;
+ SentRequestMap sent_requests_;
+
std::shared_ptr<LibhdfsEvents> event_handlers_;
std::string cluster_name_;
http://git-wip-us.apache.org/repos/asf/hadoop/blob/fc5e44d9/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/rpc/rpc_connection_impl.cc
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/rpc/rpc_connection_impl.cc b/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/rpc/rpc_connection_impl.cc
index 947d121..ee9b704 100644
--- a/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/rpc/rpc_connection_impl.cc
+++ b/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/rpc/rpc_connection_impl.cc
@@ -66,17 +66,29 @@ static void AddHeadersToPacket(
RpcConnection::~RpcConnection() {}
-RpcConnection::RpcConnection(LockFreeRpcEngine *engine)
+RpcConnection::RpcConnection(std::shared_ptr<LockFreeRpcEngine> engine)
: engine_(engine),
connected_(kNotYetConnected) {}
-::asio::io_service &RpcConnection::io_service() {
- return engine_->io_service();
+::asio::io_service *RpcConnection::GetIoService() {
+ std::shared_ptr<LockFreeRpcEngine> pinnedEngine = engine_.lock();
+ if(!pinnedEngine) {
+ LOG_ERROR(kRPC, << "RpcConnection@" << this << " attempted to access invalid RpcEngine");
+ return nullptr;
+ }
+
+ return &pinnedEngine->io_service();
}
void RpcConnection::StartReading() {
auto shared_this = shared_from_this();
- io_service().post([shared_this, this] () {
+ ::asio::io_service *service = GetIoService();
+ if(!service) {
+ LOG_ERROR(kRPC, << "RpcConnection@" << this << " attempted to access invalid IoService");
+ return;
+ }
+
+ service->post([shared_this, this] () {
OnRecvCompleted(::asio::error_code(), 0);
});
}
@@ -151,12 +163,19 @@ void RpcConnection::ContextComplete(const Status &s) {
void RpcConnection::AsyncFlushPendingRequests() {
std::shared_ptr<RpcConnection> shared_this = shared_from_this();
- io_service().post([shared_this, this]() {
+
+ ::asio::io_service *service = GetIoService();
+ if(!service) {
+ LOG_ERROR(kRPC, << "RpcConnection@" << this << " attempted to access invalid IoService");
+ return;
+ }
+
+ service->post([shared_this, this]() {
std::lock_guard<std::mutex> state_lock(connection_state_lock_);
LOG_TRACE(kRPC, << "RpcConnection::AsyncFlushPendingRequests called (connected=" << ToString(connected_) << ")");
- if (!request_over_the_wire_) {
+ if (!outgoing_request_) {
FlushPendingRequests();
}
});
@@ -209,7 +228,13 @@ Status RpcConnection::HandleRpcResponse(std::shared_ptr<Response> response) {
return status;
}
- io_service().post([req, response, status]() {
+ ::asio::io_service *service = GetIoService();
+ if(!service) {
+ LOG_ERROR(kRPC, << "RpcConnection@" << this << " attempted to access invalid IoService");
+ return Status::Error("RpcConnection attempted to access invalid IoService");
+ }
+
+ service->post([req, response, status]() {
req->OnResponseArrived(response->in.get(), status); // Never call back while holding a lock
});
@@ -267,23 +292,29 @@ std::shared_ptr<std::string> RpcConnection::PrepareContextPacket() {
// after the SASL handshake (if any)
assert(lock_held(connection_state_lock_)); // Must be holding lock before calling
- auto res = std::make_shared<std::string>();
+ std::shared_ptr<LockFreeRpcEngine> pinnedEngine = engine_.lock();
+ if(!pinnedEngine) {
+ LOG_ERROR(kRPC, << "RpcConnection@" << this << " attempted to access invalid RpcEngine");
+ return std::make_shared<std::string>();
+ }
+
+ std::shared_ptr<std::string> serializedPacketBuffer = std::make_shared<std::string>();
- RpcRequestHeaderProto h;
- h.set_rpckind(RPC_PROTOCOL_BUFFER);
- h.set_rpcop(RpcRequestHeaderProto::RPC_FINAL_PACKET);
- h.set_callid(RpcEngine::kCallIdConnectionContext);
- h.set_clientid(engine_->client_name());
+ RpcRequestHeaderProto headerProto;
+ headerProto.set_rpckind(RPC_PROTOCOL_BUFFER);
+ headerProto.set_rpcop(RpcRequestHeaderProto::RPC_FINAL_PACKET);
+ headerProto.set_callid(RpcEngine::kCallIdConnectionContext);
+ headerProto.set_clientid(pinnedEngine->client_name());
- IpcConnectionContextProto handshake;
- handshake.set_protocol(engine_->protocol_name());
+ IpcConnectionContextProto handshakeContextProto;
+ handshakeContextProto.set_protocol(pinnedEngine->protocol_name());
const std::string & user_name = auth_info_.getUser();
if (!user_name.empty()) {
- *handshake.mutable_userinfo()->mutable_effectiveuser() = user_name;
+ *handshakeContextProto.mutable_userinfo()->mutable_effectiveuser() = user_name;
}
- AddHeadersToPacket(res.get(), {&h, &handshake}, nullptr);
+ AddHeadersToPacket(serializedPacketBuffer.get(), {&headerProto, &handshakeContextProto}, nullptr);
- return res;
+ return serializedPacketBuffer;
}
void RpcConnection::AsyncRpc(
@@ -310,11 +341,22 @@ void RpcConnection::AsyncRpc_locked(
handler(status);
};
- int call_id = (method_name != SASL_METHOD_NAME ? engine_->NextCallId() : RpcEngine::kCallIdSasl);
- auto r = std::make_shared<Request>(engine_, method_name, call_id, req,
- std::move(wrapped_handler));
- auto r_vector = std::vector<std::shared_ptr<Request> > (1, r);
- SendRpcRequests(r_vector);
+
+ std::shared_ptr<Request> rpcRequest;
+ { // Scope to minimize how long RpcEngine's lifetime may be extended
+ std::shared_ptr<LockFreeRpcEngine> pinnedEngine = engine_.lock();
+ if(!pinnedEngine) {
+ LOG_ERROR(kRPC, << "RpcConnection@" << this << " attempted to access invalid RpcEngine");
+ handler(Status::Error("Invalid RpcEngine access."));
+ return;
+ }
+
+ int call_id = (method_name != SASL_METHOD_NAME ? pinnedEngine->NextCallId() : RpcEngine::kCallIdSasl);
+ rpcRequest = std::make_shared<Request>(pinnedEngine, method_name, call_id,
+ req, std::move(wrapped_handler));
+ }
+
+ SendRpcRequests({rpcRequest});
}
void RpcConnection::AsyncRpc(const std::vector<std::shared_ptr<Request> > & requests) {
@@ -330,13 +372,20 @@ void RpcConnection::SendRpcRequests(const std::vector<std::shared_ptr<Request> >
// Oops. The connection failed _just_ before the engine got a chance
// to send it. Register it as a failure
Status status = Status::ResourceUnavailable("RpcConnection closed before send.");
- engine_->AsyncRpcCommsError(status, shared_from_this(), requests);
+
+ std::shared_ptr<LockFreeRpcEngine> pinnedEngine = engine_.lock();
+ if(!pinnedEngine) {
+ LOG_ERROR(kRPC, << "RpcConnection@" << this << " attempted to access invalid RpcEngine");
+ return;
+ }
+
+ pinnedEngine->AsyncRpcCommsError(status, shared_from_this(), requests);
} else {
- for (auto r: requests) {
- if (r->method_name() != SASL_METHOD_NAME)
- pending_requests_.push_back(r);
+ for (auto request : requests) {
+ if (request->method_name() != SASL_METHOD_NAME)
+ pending_requests_.push_back(request);
else
- auth_requests_.push_back(r);
+ auth_requests_.push_back(request);
}
if (connected_ == kConnected || connected_ == kHandshaking || connected_ == kAuthenticating) { // Dont flush if we're waiting or handshaking
FlushPendingRequests();
@@ -395,26 +444,32 @@ void RpcConnection::CommsError(const Status &status) {
// Anything that has been queued to the connection (on the fly or pending)
// will get dinged for a retry
std::vector<std::shared_ptr<Request>> requestsToReturn;
- std::transform(requests_on_fly_.begin(), requests_on_fly_.end(),
+ std::transform(sent_requests_.begin(), sent_requests_.end(),
std::back_inserter(requestsToReturn),
- std::bind(&RequestOnFlyMap::value_type::second, _1));
- requests_on_fly_.clear();
+ std::bind(&SentRequestMap::value_type::second, _1));
+ sent_requests_.clear();
requestsToReturn.insert(requestsToReturn.end(),
std::make_move_iterator(pending_requests_.begin()),
std::make_move_iterator(pending_requests_.end()));
pending_requests_.clear();
- engine_->AsyncRpcCommsError(status, shared_from_this(), requestsToReturn);
+ std::shared_ptr<LockFreeRpcEngine> pinnedEngine = engine_.lock();
+ if(!pinnedEngine) {
+ LOG_ERROR(kRPC, << "RpcConnection@" << this << " attempted to access an invalid RpcEngine");
+ return;
+ }
+
+ pinnedEngine->AsyncRpcCommsError(status, shared_from_this(), requestsToReturn);
}
void RpcConnection::ClearAndDisconnect(const ::asio::error_code &ec) {
Disconnect();
std::vector<std::shared_ptr<Request>> requests;
- std::transform(requests_on_fly_.begin(), requests_on_fly_.end(),
+ std::transform(sent_requests_.begin(), sent_requests_.end(),
std::back_inserter(requests),
- std::bind(&RequestOnFlyMap::value_type::second, _1));
- requests_on_fly_.clear();
+ std::bind(&SentRequestMap::value_type::second, _1));
+ sent_requests_.clear();
requests.insert(requests.end(),
std::make_move_iterator(pending_requests_.begin()),
std::make_move_iterator(pending_requests_.end()));
@@ -426,13 +481,13 @@ void RpcConnection::ClearAndDisconnect(const ::asio::error_code &ec) {
std::shared_ptr<Request> RpcConnection::RemoveFromRunningQueue(int call_id) {
assert(lock_held(connection_state_lock_)); // Must be holding lock before calling
- auto it = requests_on_fly_.find(call_id);
- if (it == requests_on_fly_.end()) {
+ auto it = sent_requests_.find(call_id);
+ if (it == sent_requests_.end()) {
return std::shared_ptr<Request>();
}
auto req = it->second;
- requests_on_fly_.erase(it);
+ sent_requests_.erase(it);
return req;
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/fc5e44d9/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/rpc/rpc_connection_impl.h
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/rpc/rpc_connection_impl.h b/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/rpc/rpc_connection_impl.h
index 678dab3..8e579a2 100644
--- a/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/rpc/rpc_connection_impl.h
+++ b/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/rpc/rpc_connection_impl.h
@@ -40,7 +40,7 @@ class RpcConnectionImpl : public RpcConnection {
public:
MEMCHECKED_CLASS(RpcConnectionImpl)
- RpcConnectionImpl(RpcEngine *engine);
+ RpcConnectionImpl(std::shared_ptr<RpcEngine> engine);
virtual ~RpcConnectionImpl() override;
virtual void Connect(const std::vector<::asio::ip::tcp::endpoint> &server,
@@ -73,7 +73,7 @@ public:
};
template <class Socket>
-RpcConnectionImpl<Socket>::RpcConnectionImpl(RpcEngine *engine)
+RpcConnectionImpl<Socket>::RpcConnectionImpl(std::shared_ptr<RpcEngine> engine)
: RpcConnection(engine),
options_(engine->options()),
socket_(engine->io_service()),
@@ -88,8 +88,8 @@ RpcConnectionImpl<Socket>::~RpcConnectionImpl() {
if (pending_requests_.size() > 0)
LOG_WARN(kRPC, << "RpcConnectionImpl::~RpcConnectionImpl called with items in the pending queue");
- if (requests_on_fly_.size() > 0)
- LOG_WARN(kRPC, << "RpcConnectionImpl::~RpcConnectionImpl called with items in the requests_on_fly queue");
+ if (sent_requests_.size() > 0)
+ LOG_WARN(kRPC, << "RpcConnectionImpl::~RpcConnectionImpl called with items in the sent_requests queue");
}
template <class Socket>
@@ -101,13 +101,23 @@ void RpcConnectionImpl<Socket>::Connect(
this->auth_info_ = auth_info;
- auto connectionSuccessfulReq = std::make_shared<Request>(
- engine_, [handler](::google::protobuf::io::CodedInputStream *is,
- const Status &status) {
- (void)is;
- handler(status);
- });
- pending_requests_.push_back(connectionSuccessfulReq);
+ std::shared_ptr<Request> connectionRequest;
+ { // Scope to minimize how long RpcEngine's lifetime may be extended
+ std::shared_ptr<LockFreeRpcEngine> pinned_engine = engine_.lock();
+ if(!pinned_engine) {
+ LOG_ERROR(kRPC, << "RpcConnectionImpl@" << this << " attempted to access invalid RpcEngine");
+ handler(Status::Error("Invalid RpcEngine access."));
+ return;
+ }
+
+ connectionRequest = std::make_shared<Request>(pinned_engine,
+ [handler](::google::protobuf::io::CodedInputStream *is,const Status &status) {
+ (void)is;
+ handler(status);
+ });
+ }
+
+ pending_requests_.push_back(connectionRequest);
this->ConnectAndFlush(server); // need "this" so compiler can infer type of CAF
}
@@ -263,7 +273,7 @@ void RpcConnectionImpl<Socket>::OnSendCompleted(const ::asio::error_code &ec,
LOG_TRACE(kRPC, << "RpcConnectionImpl::OnSendCompleted called");
- request_over_the_wire_.reset();
+ outgoing_request_.reset();
if (ec) {
LOG_WARN(kRPC, << "Network error during RPC write: " << ec.message());
CommsError(ToStatus(ec));
@@ -283,7 +293,7 @@ void RpcConnectionImpl<Socket>::FlushPendingRequests() {
LOG_TRACE(kRPC, << "RpcConnectionImpl::FlushPendingRequests called");
// Don't send if we don't need to
- if (request_over_the_wire_) {
+ if (outgoing_request_) {
return;
}
@@ -324,9 +334,9 @@ void RpcConnectionImpl<Socket>::FlushPendingRequests() {
std::shared_ptr<std::string> payload = std::make_shared<std::string>();
req->GetPacket(payload.get());
if (!payload->empty()) {
- assert(requests_on_fly_.find(req->call_id()) == requests_on_fly_.end());
- requests_on_fly_[req->call_id()] = req;
- request_over_the_wire_ = req;
+ assert(sent_requests_.find(req->call_id()) == sent_requests_.end());
+ sent_requests_[req->call_id()] = req;
+ outgoing_request_ = req;
req->timer().expires_from_now(
std::chrono::milliseconds(options_.rpc_timeout));
@@ -343,7 +353,15 @@ void RpcConnectionImpl<Socket>::FlushPendingRequests() {
OnSendCompleted(ec, size);
});
} else { // Nothing to send for this request, inform the handler immediately
- io_service().post(
+ ::asio::io_service *service = GetIoService();
+ if(!service) {
+ LOG_ERROR(kRPC, << "RpcConnectionImpl@" << this << " attempted to access null IoService");
+ // No easy way to bail out of this context, but the only way to get here is when
+ // the FileSystem is being destroyed.
+ return;
+ }
+
+ service->post(
// Never hold locks when calling a callback
[req]() { req->OnResponseArrived(nullptr, Status::OK()); }
);
@@ -433,7 +451,7 @@ void RpcConnectionImpl<Socket>::Disconnect() {
LOG_INFO(kRPC, << "RpcConnectionImpl::Disconnect called");
- request_over_the_wire_.reset();
+ outgoing_request_.reset();
if (connected_ == kConnecting || connected_ == kHandshaking || connected_ == kAuthenticating || connected_ == kConnected) {
// Don't print out errors, we were expecting a disconnect here
SafeDisconnect(get_asio_socket_ptr(&socket_));
http://git-wip-us.apache.org/repos/asf/hadoop/blob/fc5e44d9/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/rpc/rpc_engine.cc
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/rpc/rpc_engine.cc b/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/rpc/rpc_engine.cc
index 34af423..0ca7c6a 100644
--- a/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/rpc/rpc_engine.cc
+++ b/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/rpc/rpc_engine.cc
@@ -171,7 +171,7 @@ std::shared_ptr<RpcConnection> RpcEngine::NewConnection()
{
LOG_DEBUG(kRPC, << "RpcEngine::NewConnection called");
- return std::make_shared<RpcConnectionImpl<::asio::ip::tcp::socket>>(this);
+ return std::make_shared<RpcConnectionImpl<::asio::ip::tcp::socket>>(shared_from_this());
}
std::shared_ptr<RpcConnection> RpcEngine::InitializeConnection()
http://git-wip-us.apache.org/repos/asf/hadoop/blob/fc5e44d9/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/rpc/rpc_engine.h
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/rpc/rpc_engine.h b/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/rpc/rpc_engine.h
index dc4054e..9f45fcf 100644
--- a/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/rpc/rpc_engine.h
+++ b/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/rpc/rpc_engine.h
@@ -75,16 +75,16 @@ public:
std::vector<std::shared_ptr<Request>> pendingRequests) = 0;
- virtual const RetryPolicy * retry_policy() const = 0;
+ virtual const RetryPolicy *retry_policy() = 0;
virtual int NextCallId() = 0;
- virtual const std::string &client_name() const = 0;
- virtual const std::string &client_id() const = 0;
- virtual const std::string &user_name() const = 0;
- virtual const std::string &protocol_name() const = 0;
- virtual int protocol_version() const = 0;
+ virtual const std::string &client_name() = 0;
+ virtual const std::string &client_id() = 0;
+ virtual const std::string &user_name() = 0;
+ virtual const std::string &protocol_name() = 0;
+ virtual int protocol_version() = 0;
virtual ::asio::io_service &io_service() = 0;
- virtual const Options &options() const = 0;
+ virtual const Options &options() = 0;
};
@@ -95,7 +95,7 @@ public:
* Threading model: thread-safe. All callbacks will be called back from
* an asio pool and will not hold any internal locks
*/
-class RpcEngine : public LockFreeRpcEngine {
+class RpcEngine : public LockFreeRpcEngine, public std::enable_shared_from_this<RpcEngine> {
public:
MEMCHECKED_CLASS(RpcEngine)
enum { kRpcVersion = 9 };
@@ -133,20 +133,20 @@ class RpcEngine : public LockFreeRpcEngine {
std::vector<std::shared_ptr<Request>> pendingRequests);
- const RetryPolicy * retry_policy() const override { return retry_policy_.get(); }
+ const RetryPolicy * retry_policy() override { return retry_policy_.get(); }
int NextCallId() override { return ++call_id_; }
void TEST_SetRpcConnection(std::shared_ptr<RpcConnection> conn);
void TEST_SetRetryPolicy(std::unique_ptr<const RetryPolicy> policy);
std::unique_ptr<const RetryPolicy> TEST_GenerateRetryPolicyUsingOptions();
- const std::string &client_name() const override { return client_name_; }
- const std::string &client_id() const override { return client_id_; }
- const std::string &user_name() const override { return auth_info_.getUser(); }
- const std::string &protocol_name() const override { return protocol_name_; }
- int protocol_version() const override { return protocol_version_; }
+ const std::string &client_name() override { return client_name_; }
+ const std::string &client_id() override { return client_id_; }
+ const std::string &user_name() override { return auth_info_.getUser(); }
+ const std::string &protocol_name() override { return protocol_name_; }
+ int protocol_version() override { return protocol_version_; }
::asio::io_service &io_service() override { return *io_service_; }
- const Options &options() const override { return options_; }
+ const Options &options() override { return options_; }
static std::string GetRandomClientName();
void SetFsEventCallback(fs_event_callback callback);
http://git-wip-us.apache.org/repos/asf/hadoop/blob/fc5e44d9/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/tests/rpc_engine_test.cc
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/tests/rpc_engine_test.cc b/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/tests/rpc_engine_test.cc
index 776894e..f998c7f 100644
--- a/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/tests/rpc_engine_test.cc
+++ b/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/tests/rpc_engine_test.cc
@@ -71,7 +71,7 @@ protected:
// Stuff in some dummy endpoints so we don't error out
last_endpoints_ = make_endpoint()[0].endpoints;
- return std::make_shared<RpcConnectionImpl<SharedMockRPCConnection>>(this);
+ return std::make_shared<RpcConnectionImpl<SharedMockRPCConnection>>(shared_from_this());
}
};
@@ -106,9 +106,9 @@ using namespace hdfs;
TEST(RpcEngineTest, TestRoundTrip) {
::asio::io_service io_service;
Options options;
- RpcEngine engine(&io_service, options, "foo", "", "protocol", 1);
+ std::shared_ptr<RpcEngine> engine = std::make_shared<RpcEngine>(&io_service, options, "foo", "", "protocol", 1);
auto conn =
- std::make_shared<RpcConnectionImpl<MockRPCConnection> >(&engine);
+ std::make_shared<RpcConnectionImpl<MockRPCConnection> >(engine);
conn->TEST_set_connected(true);
conn->StartReading();
@@ -122,14 +122,14 @@ TEST(RpcEngineTest, TestRoundTrip) {
.WillOnce(Return(RpcResponse(h, server_resp.SerializeAsString())));
std::shared_ptr<RpcConnection> conn_ptr(conn);
- engine.TEST_SetRpcConnection(conn_ptr);
+ engine->TEST_SetRpcConnection(conn_ptr);
bool complete = false;
EchoRequestProto req;
req.set_message("foo");
std::shared_ptr<EchoResponseProto> resp(new EchoResponseProto());
- engine.AsyncRpc("test", &req, resp, [resp, &complete,&io_service](const Status &stat) {
+ engine->AsyncRpc("test", &req, resp, [resp, &complete,&io_service](const Status &stat) {
ASSERT_TRUE(stat.ok());
ASSERT_EQ("foo", resp->message());
complete = true;
@@ -142,9 +142,9 @@ TEST(RpcEngineTest, TestRoundTrip) {
TEST(RpcEngineTest, TestConnectionResetAndFail) {
::asio::io_service io_service;
Options options;
- RpcEngine engine(&io_service, options, "foo", "", "protocol", 1);
+ std::shared_ptr<RpcEngine> engine = std::make_shared<RpcEngine>(&io_service, options, "foo", "", "protocol", 1);
auto conn =
- std::make_shared<RpcConnectionImpl<MockRPCConnection> >(&engine);
+ std::make_shared<RpcConnectionImpl<MockRPCConnection> >(engine);
conn->TEST_set_connected(true);
conn->StartReading();
@@ -158,13 +158,13 @@ TEST(RpcEngineTest, TestConnectionResetAndFail) {
h, "", make_error_code(::asio::error::connection_reset))));
std::shared_ptr<RpcConnection> conn_ptr(conn);
- engine.TEST_SetRpcConnection(conn_ptr);
+ engine->TEST_SetRpcConnection(conn_ptr);
EchoRequestProto req;
req.set_message("foo");
std::shared_ptr<EchoResponseProto> resp(new EchoResponseProto());
- engine.AsyncRpc("test", &req, resp, [&complete, &io_service](const Status &stat) {
+ engine->AsyncRpc("test", &req, resp, [&complete, &io_service](const Status &stat) {
complete = true;
io_service.stop();
ASSERT_FALSE(stat.ok());
@@ -179,11 +179,12 @@ TEST(RpcEngineTest, TestConnectionResetAndRecover) {
Options options;
options.max_rpc_retries = 1;
options.rpc_retry_delay_ms = 0;
- SharedConnectionEngine engine(&io_service, options, "foo", "", "protocol", 1);
+ std::shared_ptr<SharedConnectionEngine> engine
+ = std::make_shared<SharedConnectionEngine>(&io_service, options, "foo", "", "protocol", 1);
// Normally determined during RpcEngine::Connect, but in this case options
// provides enough info to determine policy here.
- engine.TEST_SetRetryPolicy(engine.TEST_GenerateRetryPolicyUsingOptions());
+ engine->TEST_SetRetryPolicy(engine->TEST_GenerateRetryPolicyUsingOptions());
EchoResponseProto server_resp;
@@ -205,7 +206,7 @@ TEST(RpcEngineTest, TestConnectionResetAndRecover) {
req.set_message("foo");
std::shared_ptr<EchoResponseProto> resp(new EchoResponseProto());
- engine.AsyncRpc("test", &req, resp, [&complete, &io_service](const Status &stat) {
+ engine->AsyncRpc("test", &req, resp, [&complete, &io_service](const Status &stat) {
complete = true;
io_service.stop();
ASSERT_TRUE(stat.ok());
@@ -219,11 +220,12 @@ TEST(RpcEngineTest, TestConnectionResetAndRecoverWithDelay) {
Options options;
options.max_rpc_retries = 1;
options.rpc_retry_delay_ms = 1;
- SharedConnectionEngine engine(&io_service, options, "foo", "", "protocol", 1);
+ std::shared_ptr<SharedConnectionEngine> engine =
+ std::make_shared<SharedConnectionEngine>(&io_service, options, "foo", "", "protocol", 1);
// Normally determined during RpcEngine::Connect, but in this case options
// provides enough info to determine policy here.
- engine.TEST_SetRetryPolicy(engine.TEST_GenerateRetryPolicyUsingOptions());
+ engine->TEST_SetRetryPolicy(engine->TEST_GenerateRetryPolicyUsingOptions());
EchoResponseProto server_resp;
server_resp.set_message("foo");
@@ -244,7 +246,7 @@ TEST(RpcEngineTest, TestConnectionResetAndRecoverWithDelay) {
req.set_message("foo");
std::shared_ptr<EchoResponseProto> resp(new EchoResponseProto());
- engine.AsyncRpc("test", &req, resp, [&complete, &io_service](const Status &stat) {
+ engine->AsyncRpc("test", &req, resp, [&complete, &io_service](const Status &stat) {
complete = true;
io_service.stop();
ASSERT_TRUE(stat.ok());
@@ -272,11 +274,12 @@ TEST(RpcEngineTest, TestConnectionFailure)
Options options;
options.max_rpc_retries = 0;
options.rpc_retry_delay_ms = 0;
- SharedConnectionEngine engine(&io_service, options, "foo", "", "protocol", 1);
+ std::shared_ptr<SharedConnectionEngine> engine
+ = std::make_shared<SharedConnectionEngine>(&io_service, options, "foo", "", "protocol", 1);
EXPECT_CALL(*producer, Produce())
.WillOnce(Return(std::make_pair(make_error_code(::asio::error::connection_reset), "")));
- engine.Connect("", make_endpoint(), [&complete, &io_service](const Status &stat) {
+ engine->Connect("", make_endpoint(), [&complete, &io_service](const Status &stat) {
complete = true;
io_service.stop();
ASSERT_FALSE(stat.ok());
@@ -298,13 +301,14 @@ TEST(RpcEngineTest, TestConnectionFailureRetryAndFailure)
Options options;
options.max_rpc_retries = 2;
options.rpc_retry_delay_ms = 0;
- SharedConnectionEngine engine(&io_service, options, "foo", "", "protocol", 1);
+ std::shared_ptr<SharedConnectionEngine> engine =
+ std::make_shared<SharedConnectionEngine>(&io_service, options, "foo", "", "protocol", 1);
EXPECT_CALL(*producer, Produce())
.WillOnce(Return(std::make_pair(make_error_code(::asio::error::connection_reset), "")))
.WillOnce(Return(std::make_pair(make_error_code(::asio::error::connection_reset), "")))
.WillOnce(Return(std::make_pair(make_error_code(::asio::error::connection_reset), "")));
- engine.Connect("", make_endpoint(), [&complete, &io_service](const Status &stat) {
+ engine->Connect("", make_endpoint(), [&complete, &io_service](const Status &stat) {
complete = true;
io_service.stop();
ASSERT_FALSE(stat.ok());
@@ -326,13 +330,14 @@ TEST(RpcEngineTest, TestConnectionFailureAndRecover)
Options options;
options.max_rpc_retries = 1;
options.rpc_retry_delay_ms = 0;
- SharedConnectionEngine engine(&io_service, options, "foo", "", "protocol", 1);
+ std::shared_ptr<SharedConnectionEngine> engine =
+ std::make_shared<SharedConnectionEngine>(&io_service, options, "foo", "", "protocol", 1);
EXPECT_CALL(*producer, Produce())
.WillOnce(Return(std::make_pair(make_error_code(::asio::error::connection_reset), "")))
.WillOnce(Return(std::make_pair(::asio::error_code(), "")))
.WillOnce(Return(std::make_pair(::asio::error::would_block, "")));
- engine.Connect("", make_endpoint(), [&complete, &io_service](const Status &stat) {
+ engine->Connect("", make_endpoint(), [&complete, &io_service](const Status &stat) {
complete = true;
io_service.stop();
ASSERT_TRUE(stat.ok());
@@ -347,16 +352,17 @@ TEST(RpcEngineTest, TestEventCallbacks)
Options options;
options.max_rpc_retries = 99;
options.rpc_retry_delay_ms = 0;
- SharedConnectionEngine engine(&io_service, options, "foo", "", "protocol", 1);
+ std::shared_ptr<SharedConnectionEngine> engine =
+ std::make_shared<SharedConnectionEngine>(&io_service, options, "foo", "", "protocol", 1);
// Normally determined during RpcEngine::Connect, but in this case options
// provides enough info to determine policy here.
- engine.TEST_SetRetryPolicy(engine.TEST_GenerateRetryPolicyUsingOptions());
+ engine->TEST_SetRetryPolicy(engine->TEST_GenerateRetryPolicyUsingOptions());
// Set up event callbacks
int calls = 0;
std::vector<std::string> callbacks;
- engine.SetFsEventCallback([&calls, &callbacks] (const char * event,
+ engine->SetFsEventCallback([&calls, &callbacks] (const char * event,
const char * cluster,
int64_t value) {
(void)cluster; (void)value;
@@ -393,7 +399,7 @@ TEST(RpcEngineTest, TestEventCallbacks)
std::shared_ptr<EchoResponseProto> resp(new EchoResponseProto());
bool complete = false;
- engine.AsyncRpc("test", &req, resp, [&complete, &io_service](const Status &stat) {
+ engine->AsyncRpc("test", &req, resp, [&complete, &io_service](const Status &stat) {
complete = true;
io_service.stop();
ASSERT_TRUE(stat.ok());
@@ -431,13 +437,14 @@ TEST(RpcEngineTest, TestConnectionFailureAndAsyncRecover)
Options options;
options.max_rpc_retries = 1;
options.rpc_retry_delay_ms = 1;
- SharedConnectionEngine engine(&io_service, options, "foo", "", "protocol", 1);
+ std::shared_ptr<SharedConnectionEngine> engine =
+ std::make_shared<SharedConnectionEngine>(&io_service, options, "foo", "", "protocol", 1);
EXPECT_CALL(*producer, Produce())
.WillOnce(Return(std::make_pair(make_error_code(::asio::error::connection_reset), "")))
.WillOnce(Return(std::make_pair(::asio::error_code(), "")))
.WillOnce(Return(std::make_pair(::asio::error::would_block, "")));
- engine.Connect("", make_endpoint(), [&complete, &io_service](const Status &stat) {
+ engine->Connect("", make_endpoint(), [&complete, &io_service](const Status &stat) {
complete = true;
io_service.stop();
ASSERT_TRUE(stat.ok());
@@ -455,9 +462,9 @@ TEST(RpcEngineTest, TestTimeout) {
::asio::io_service io_service;
Options options;
options.rpc_timeout = 1;
- RpcEngine engine(&io_service, options, "foo", "", "protocol", 1);
+ std::shared_ptr<RpcEngine> engine = std::make_shared<RpcEngine>(&io_service, options, "foo", "", "protocol", 1);
auto conn =
- std::make_shared<RpcConnectionImpl<MockRPCConnection> >(&engine);
+ std::make_shared<RpcConnectionImpl<MockRPCConnection> >(engine);
conn->TEST_set_connected(true);
conn->StartReading();
@@ -465,14 +472,14 @@ TEST(RpcEngineTest, TestTimeout) {
.WillOnce(Return(std::make_pair(::asio::error::would_block, "")));
std::shared_ptr<RpcConnection> conn_ptr(conn);
- engine.TEST_SetRpcConnection(conn_ptr);
+ engine->TEST_SetRpcConnection(conn_ptr);
bool complete = false;
EchoRequestProto req;
req.set_message("foo");
std::shared_ptr<EchoResponseProto> resp(new EchoResponseProto());
- engine.AsyncRpc("test", &req, resp, [resp, &complete,&io_service](const Status &stat) {
+ engine->AsyncRpc("test", &req, resp, [resp, &complete,&io_service](const Status &stat) {
complete = true;
io_service.stop();
ASSERT_FALSE(stat.ok());
---------------------------------------------------------------------
To unsubscribe, e-mail: common-commits-unsubscribe@hadoop.apache.org
For additional commands, e-mail: common-commits-help@hadoop.apache.org