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 xy...@apache.org on 2018/04/14 00:09:42 UTC
[14/44] hadoop git commit: HDFS-13403: libhdfs++ Use hdfs::IoService
object rather than asio::io_service. Contributed by James Clampffer.
http://git-wip-us.apache.org/repos/asf/hadoop/blob/eefe2a14/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 8e579a2..1dd43af 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
@@ -26,6 +26,7 @@
#include "common/logging.h"
#include "common/util.h"
#include "common/libhdfs_events_impl.h"
+#include "hdfspp/ioservice.h"
#include <asio/connect.hpp>
#include <asio/read.hpp>
@@ -76,8 +77,8 @@ template <class Socket>
RpcConnectionImpl<Socket>::RpcConnectionImpl(std::shared_ptr<RpcEngine> engine)
: RpcConnection(engine),
options_(engine->options()),
- socket_(engine->io_service()),
- connect_timer_(engine->io_service())
+ socket_(engine->io_service()->GetRaw()),
+ connect_timer_(engine->io_service()->GetRaw())
{
LOG_TRACE(kRPC, << "RpcConnectionImpl::RpcConnectionImpl called &" << (void*)this);
}
@@ -353,7 +354,7 @@ void RpcConnectionImpl<Socket>::FlushPendingRequests() {
OnSendCompleted(ec, size);
});
} else { // Nothing to send for this request, inform the handler immediately
- ::asio::io_service *service = GetIoService();
+ std::shared_ptr<IoService> 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
@@ -361,7 +362,7 @@ void RpcConnectionImpl<Socket>::FlushPendingRequests() {
return;
}
- service->post(
+ service->PostTask(
// Never hold locks when calling a callback
[req]() { req->OnResponseArrived(nullptr, Status::OK()); }
);
http://git-wip-us.apache.org/repos/asf/hadoop/blob/eefe2a14/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 0ca7c6a..ad6c9b9 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
@@ -30,7 +30,7 @@ template <class T>
using optional = std::experimental::optional<T>;
-RpcEngine::RpcEngine(::asio::io_service *io_service, const Options &options,
+RpcEngine::RpcEngine(std::shared_ptr<IoService> io_service, const Options &options,
const std::string &client_name, const std::string &user_name,
const char *protocol_name, int protocol_version)
: io_service_(io_service),
@@ -40,7 +40,7 @@ RpcEngine::RpcEngine(::asio::io_service *io_service, const Options &options,
protocol_name_(protocol_name),
protocol_version_(protocol_version),
call_id_(0),
- retry_timer(*io_service),
+ retry_timer(io_service->GetRaw()),
event_handlers_(std::make_shared<LibhdfsEvents>()),
connect_canceled_(false)
{
@@ -86,7 +86,7 @@ bool RpcEngine::CancelPendingConnect() {
void RpcEngine::Shutdown() {
LOG_DEBUG(kRPC, << "RpcEngine::Shutdown called");
- io_service_->post([this]() {
+ io_service_->PostLambda([this]() {
std::lock_guard<std::mutex> state_lock(engine_state_lock_);
conn_.reset();
});
@@ -154,7 +154,7 @@ void RpcEngine::AsyncRpc(
// In case user-side code isn't checking the status of Connect before doing RPC
if(connect_canceled_) {
- io_service_->post(
+ io_service_->PostLambda(
[handler](){ handler(Status::Canceled()); }
);
return;
@@ -190,7 +190,7 @@ void RpcEngine::AsyncRpcCommsError(
std::vector<std::shared_ptr<Request>> pendingRequests) {
LOG_ERROR(kRPC, << "RpcEngine::AsyncRpcCommsError called; status=\"" << status.ToString() << "\" conn=" << failedConnection.get() << " reqs=" << std::to_string(pendingRequests.size()));
- io_service().post([this, status, failedConnection, pendingRequests]() {
+ io_service_->PostLambda([this, status, failedConnection, pendingRequests]() {
RpcCommsError(status, failedConnection, pendingRequests);
});
}
@@ -238,7 +238,7 @@ void RpcEngine::RpcCommsError(
// on. There might be a good argument for caching the first error
// rather than the last one, that gets messy
- io_service().post([req, status]() {
+ io_service()->PostLambda([req, status]() {
req->OnResponseArrived(nullptr, status); // Never call back while holding a lock
});
it = pendingRequests.erase(it);
@@ -283,7 +283,7 @@ void RpcEngine::RpcCommsError(
for(unsigned int i=0; i<pendingRequests.size(); i++) {
std::shared_ptr<Request> sharedCurrentRequest = pendingRequests[i];
- io_service().post([sharedCurrentRequest, badEndpointStatus]() {
+ io_service()->PostLambda([sharedCurrentRequest, badEndpointStatus]() {
sharedCurrentRequest->OnResponseArrived(nullptr, badEndpointStatus); // Never call back while holding a lock
});
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/eefe2a14/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 9f45fcf..845eaf5 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
@@ -60,6 +60,7 @@ class RpcConnection;
class SaslProtocol;
class RpcConnection;
class Request;
+class IoService;
/*
* These methods of the RpcEngine will never acquire locks, and are safe for
@@ -83,7 +84,7 @@ public:
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 std::shared_ptr<IoService> io_service() const = 0;
virtual const Options &options() = 0;
};
@@ -107,7 +108,7 @@ class RpcEngine : public LockFreeRpcEngine, public std::enable_shared_from_this<
kCallIdSasl = -33
};
- RpcEngine(::asio::io_service *io_service, const Options &options,
+ RpcEngine(std::shared_ptr<IoService> service, const Options &options,
const std::string &client_name, const std::string &user_name,
const char *protocol_name, int protocol_version);
@@ -145,7 +146,7 @@ class RpcEngine : public LockFreeRpcEngine, public std::enable_shared_from_this<
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_; }
+ std::shared_ptr<IoService> io_service() const override { return io_service_; }
const Options &options() override { return options_; }
static std::string GetRandomClientName();
@@ -162,7 +163,7 @@ protected:
std::vector<::asio::ip::tcp::endpoint> last_endpoints_;
private:
- ::asio::io_service * const io_service_;
+ mutable std::shared_ptr<IoService> io_service_;
const Options options_;
const std::string client_name_;
const std::string client_id_;
http://git-wip-us.apache.org/repos/asf/hadoop/blob/eefe2a14/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/tests/bad_datanode_test.cc
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/tests/bad_datanode_test.cc b/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/tests/bad_datanode_test.cc
index 00bbf3d..23de015 100644
--- a/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/tests/bad_datanode_test.cc
+++ b/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/tests/bad_datanode_test.cc
@@ -16,11 +16,12 @@
* limitations under the License.
*/
-#include "fs/filesystem.h"
-#include "fs/bad_datanode_tracker.h"
#include "common/libhdfs_events_impl.h"
-
#include "common/util.h"
+#include "fs/filesystem.h"
+#include "fs/filehandle.h"
+#include "fs/bad_datanode_tracker.h"
+#include "reader/block_reader.h"
#include <gmock/gmock.h>
@@ -54,7 +55,7 @@ public:
const std::string & client_name,
const hadoop::hdfs::LocatedBlockProto &block,
size_t offset,
- const MutableBuffers &buffers,
+ const MutableBuffer &buffer,
const std::function<void(const Status &, size_t)> handler));
virtual void CancelOperation() override {
@@ -67,14 +68,14 @@ class MockDNConnection : public DataNodeConnection, public std::enable_shared_fr
handler(Status::OK(), shared_from_this());
}
- void async_read_some(const MutableBuffers &buf,
+ void async_read_some(const MutableBuffer &buf,
std::function<void (const asio::error_code & error,
std::size_t bytes_transferred) > handler) override {
(void)buf;
handler(asio::error::fault, 0);
}
- void async_write_some(const ConstBuffers &buf,
+ void async_write_some(const ConstBuffer &buf,
std::function<void (const asio::error_code & error,
std::size_t bytes_transferred) > handler) override {
(void)buf;
@@ -101,7 +102,7 @@ protected:
return mock_reader_;
}
std::shared_ptr<DataNodeConnection> CreateDataNodeConnection(
- ::asio::io_service *io_service,
+ std::shared_ptr<IoService> io_service,
const ::hadoop::hdfs::DatanodeInfoProto & dn,
const hadoop::common::TokenProto * token) override {
(void) io_service; (void) dn; (void) token;
@@ -130,12 +131,12 @@ TEST(BadDataNodeTest, TestNoNodes) {
char buf[4096] = {
0,
};
- IoServiceImpl io_service;
+ std::shared_ptr<IoService> io_service = IoService::MakeShared();
auto bad_node_tracker = std::make_shared<BadDataNodeTracker>();
auto monitors = std::make_shared<LibhdfsEvents>();
bad_node_tracker->AddBadNode("foo");
- PartialMockFileHandle is("cluster", "file", &io_service.io_service(), GetRandomClientName(), file_info, bad_node_tracker, monitors);
+ PartialMockFileHandle is("cluster", "file", io_service, GetRandomClientName(), file_info, bad_node_tracker, monitors);
Status stat;
size_t read = 0;
@@ -170,7 +171,7 @@ TEST(BadDataNodeTest, NNEventCallback) {
char buf[4096] = {
0,
};
- IoServiceImpl io_service;
+ std::shared_ptr<IoService> io_service = IoService::MakeShared();
auto tracker = std::make_shared<BadDataNodeTracker>();
@@ -191,7 +192,7 @@ TEST(BadDataNodeTest, NNEventCallback) {
return event_response::make_ok();
});
- PartialMockFileHandle is("cluster", "file", &io_service.io_service(), GetRandomClientName(), file_info, tracker, monitors);
+ PartialMockFileHandle is("cluster", "file", io_service, GetRandomClientName(), file_info, tracker, monitors);
Status stat;
size_t read = 0;
@@ -234,10 +235,10 @@ TEST(BadDataNodeTest, RecoverableError) {
char buf[4096] = {
0,
};
- IoServiceImpl io_service;
+ std::shared_ptr<IoService> io_service = IoService::MakeShared();
auto tracker = std::make_shared<BadDataNodeTracker>();
auto monitors = std::make_shared<LibhdfsEvents>();
- PartialMockFileHandle is("cluster", "file", &io_service.io_service(), GetRandomClientName(), file_info, tracker, monitors);
+ PartialMockFileHandle is("cluster", "file", io_service, GetRandomClientName(), file_info, tracker, monitors);
Status stat;
size_t read = 0;
EXPECT_CALL(*is.mock_reader_, AsyncReadBlock(_,_,_,_,_))
@@ -285,10 +286,10 @@ TEST(BadDataNodeTest, InternalError) {
char buf[4096] = {
0,
};
- IoServiceImpl io_service;
+ std::shared_ptr<IoService> io_service = IoService::MakeShared();
auto tracker = std::make_shared<BadDataNodeTracker>();
auto monitors = std::make_shared<LibhdfsEvents>();
- PartialMockFileHandle is("cluster", "file", &io_service.io_service(), GetRandomClientName(), file_info, tracker, monitors);
+ PartialMockFileHandle is("cluster", "file", io_service, GetRandomClientName(), file_info, tracker, monitors);
Status stat;
size_t read = 0;
EXPECT_CALL(*is.mock_reader_, AsyncReadBlock(_,_,_,_,_))
http://git-wip-us.apache.org/repos/asf/hadoop/blob/eefe2a14/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/tests/hdfs_ioservice_test.cc
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/tests/hdfs_ioservice_test.cc b/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/tests/hdfs_ioservice_test.cc
index 5ee9789..2fdbd80 100644
--- a/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/tests/hdfs_ioservice_test.cc
+++ b/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/tests/hdfs_ioservice_test.cc
@@ -16,13 +16,15 @@
* limitations under the License.
*/
-#include "common/hdfs_ioservice.h"
+#include "hdfspp/ioservice.h"
#include <future>
#include <functional>
#include <thread>
#include <string>
+
+#include <google/protobuf/stubs/common.h>
#include <gmock/gmock.h>
using ::testing::_;
@@ -34,7 +36,7 @@ using namespace hdfs;
// Make sure IoService spins up specified number of threads
TEST(IoServiceTest, InitThreads) {
#ifndef DISABLE_CONCURRENT_WORKERS
- std::shared_ptr<IoServiceImpl> service = std::static_pointer_cast<IoServiceImpl>(IoService::MakeShared());
+ std::shared_ptr<IoService> service = IoService::MakeShared();
EXPECT_NE(service, nullptr);
unsigned int thread_count = 4;
@@ -50,7 +52,7 @@ TEST(IoServiceTest, InitThreads) {
// Make sure IoService defaults to logical thread count
TEST(IoServiceTest, InitDefaultThreads) {
#ifndef DISABLE_CONCURRENT_WORKERS
- std::shared_ptr<IoServiceImpl> service = std::static_pointer_cast<IoServiceImpl>(IoService::MakeShared());
+ std::shared_ptr<IoService> service = IoService::MakeShared();
EXPECT_NE(service, nullptr);
unsigned int thread_count = std::thread::hardware_concurrency();
@@ -66,7 +68,7 @@ TEST(IoServiceTest, InitDefaultThreads) {
// Check IoService::PostTask
TEST(IoServiceTest, SimplePost) {
- std::shared_ptr<IoServiceImpl> service = std::static_pointer_cast<IoServiceImpl>(IoService::MakeShared());
+ std::shared_ptr<IoService> service = IoService::MakeShared();
EXPECT_NE(service, nullptr);
unsigned int thread_count = std::thread::hardware_concurrency();
http://git-wip-us.apache.org/repos/asf/hadoop/blob/eefe2a14/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/tests/mock_connection.h
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/tests/mock_connection.h b/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/tests/mock_connection.h
index cd1fc12..de234ef 100644
--- a/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/tests/mock_connection.h
+++ b/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/tests/mock_connection.h
@@ -49,7 +49,7 @@ public:
virtual ~MockConnectionBase();
typedef std::pair<asio::error_code, std::string> ProducerResult;
- void async_read_some(const MutableBuffers &buf,
+ void async_read_some(const MutableBuffer &buf,
std::function<void (const asio::error_code & error,
std::size_t bytes_transferred) > handler) override {
if (produced_.size() == 0) {
@@ -72,7 +72,7 @@ public:
io_service_->post(std::bind(handler, asio::error_code(), len));
}
- void async_write_some(const ConstBuffers &buf,
+ void async_write_some(const ConstBuffer &buf,
std::function<void (const asio::error_code & error,
std::size_t bytes_transferred) > handler) override {
// CompletionResult res = OnWrite(buf);
http://git-wip-us.apache.org/repos/asf/hadoop/blob/eefe2a14/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/tests/remote_block_reader_test.cc
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/tests/remote_block_reader_test.cc b/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/tests/remote_block_reader_test.cc
index 80127f3..4b909b2 100644
--- a/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/tests/remote_block_reader_test.cc
+++ b/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/tests/remote_block_reader_test.cc
@@ -69,14 +69,14 @@ public:
/* event handler to trigger side effects */
std::function<void(void)> OnRead;
- void async_read_some(const MutableBuffers &buf,
+ void async_read_some(const MutableBuffer &buf,
std::function<void (const asio::error_code & error,
std::size_t bytes_transferred) > handler) override {
this->OnRead();
this->MockConnectionBase::async_read_some(buf, handler);
}
- void async_write_some(const ConstBuffers &buf,
+ void async_write_some(const ConstBuffer &buf,
std::function<void (const asio::error_code & error,
std::size_t bytes_transferred) > handler) override {
this->MockConnectionBase::async_write_some(buf, handler);
http://git-wip-us.apache.org/repos/asf/hadoop/blob/eefe2a14/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 f998c7f..6bbe725 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
@@ -16,6 +16,8 @@
* limitations under the License.
*/
+#include "hdfspp/ioservice.h"
+
#include "mock_connection.h"
#include "test.pb.h"
#include "RpcHeader.pb.h"
@@ -23,7 +25,6 @@
#include "common/namenode_info.h"
#include <google/protobuf/io/coded_stream.h>
-
#include <gmock/gmock.h>
using ::hadoop::common::RpcResponseHeaderProto;
@@ -104,9 +105,10 @@ static inline std::pair<error_code, string> RpcResponse(
using namespace hdfs;
TEST(RpcEngineTest, TestRoundTrip) {
- ::asio::io_service io_service;
+
+ std::shared_ptr<IoService> io_service = IoService::MakeShared();
Options options;
- std::shared_ptr<RpcEngine> engine = std::make_shared<RpcEngine>(&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);
conn->TEST_set_connected(true);
@@ -129,20 +131,20 @@ TEST(RpcEngineTest, TestRoundTrip) {
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;
- io_service.stop();
+ io_service->Stop();
});
- io_service.run();
+ io_service->Run();
ASSERT_TRUE(complete);
}
TEST(RpcEngineTest, TestConnectionResetAndFail) {
- ::asio::io_service io_service;
+ std::shared_ptr<IoService> io_service = IoService::MakeShared();
Options options;
- std::shared_ptr<RpcEngine> engine = std::make_shared<RpcEngine>(&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);
conn->TEST_set_connected(true);
@@ -164,23 +166,23 @@ TEST(RpcEngineTest, TestConnectionResetAndFail) {
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();
+ io_service->Stop();
ASSERT_FALSE(stat.ok());
});
- io_service.run();
+ io_service->Run();
ASSERT_TRUE(complete);
}
TEST(RpcEngineTest, TestConnectionResetAndRecover) {
- ::asio::io_service io_service;
+ std::shared_ptr<IoService> io_service = IoService::MakeShared();
Options options;
options.max_rpc_retries = 1;
options.rpc_retry_delay_ms = 0;
std::shared_ptr<SharedConnectionEngine> engine
- = std::make_shared<SharedConnectionEngine>(&io_service, options, "foo", "", "protocol", 1);
+ = 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.
@@ -206,22 +208,22 @@ 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();
+ io_service->Stop();
ASSERT_TRUE(stat.ok());
});
- io_service.run();
+ io_service->Run();
ASSERT_TRUE(complete);
}
TEST(RpcEngineTest, TestConnectionResetAndRecoverWithDelay) {
- ::asio::io_service io_service;
+ std::shared_ptr<IoService> io_service = IoService::MakeShared();
Options options;
options.max_rpc_retries = 1;
options.rpc_retry_delay_ms = 1;
std::shared_ptr<SharedConnectionEngine> engine =
- std::make_shared<SharedConnectionEngine>(&io_service, options, "foo", "", "protocol", 1);
+ 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.
@@ -246,17 +248,17 @@ 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();
+ io_service->Stop();
ASSERT_TRUE(stat.ok());
});
- ::asio::deadline_timer timer(io_service);
+ ::asio::deadline_timer timer(io_service->GetRaw());
timer.expires_from_now(std::chrono::hours(100));
timer.async_wait([](const asio::error_code & err){(void)err; ASSERT_FALSE("Timed out"); });
- io_service.run();
+ io_service->Run();
ASSERT_TRUE(complete);
}
@@ -267,7 +269,7 @@ TEST(RpcEngineTest, TestConnectionFailure)
SharedMockConnection::SetSharedConnectionData(producer);
// Error and no retry
- ::asio::io_service io_service;
+ std::shared_ptr<IoService> io_service = IoService::MakeShared();
bool complete = false;
@@ -275,16 +277,16 @@ TEST(RpcEngineTest, TestConnectionFailure)
options.max_rpc_retries = 0;
options.rpc_retry_delay_ms = 0;
std::shared_ptr<SharedConnectionEngine> engine
- = std::make_shared<SharedConnectionEngine>(&io_service, options, "foo", "", "protocol", 1);
+ = 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();
+ io_service->Stop();
ASSERT_FALSE(stat.ok());
});
- io_service.run();
+ io_service->Run();
ASSERT_TRUE(complete);
}
@@ -294,7 +296,7 @@ TEST(RpcEngineTest, TestConnectionFailureRetryAndFailure)
producer->checkProducerForConnect = true;
SharedMockConnection::SetSharedConnectionData(producer);
- ::asio::io_service io_service;
+ std::shared_ptr<IoService> io_service = IoService::MakeShared();
bool complete = false;
@@ -302,18 +304,18 @@ TEST(RpcEngineTest, TestConnectionFailureRetryAndFailure)
options.max_rpc_retries = 2;
options.rpc_retry_delay_ms = 0;
std::shared_ptr<SharedConnectionEngine> engine =
- std::make_shared<SharedConnectionEngine>(&io_service, options, "foo", "", "protocol", 1);
+ 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();
+ io_service->Stop();
ASSERT_FALSE(stat.ok());
});
- io_service.run();
+ io_service->Run();
ASSERT_TRUE(complete);
}
@@ -323,7 +325,7 @@ TEST(RpcEngineTest, TestConnectionFailureAndRecover)
producer->checkProducerForConnect = true;
SharedMockConnection::SetSharedConnectionData(producer);
- ::asio::io_service io_service;
+ std::shared_ptr<IoService> io_service = IoService::MakeShared();
bool complete = false;
@@ -331,29 +333,30 @@ TEST(RpcEngineTest, TestConnectionFailureAndRecover)
options.max_rpc_retries = 1;
options.rpc_retry_delay_ms = 0;
std::shared_ptr<SharedConnectionEngine> engine =
- std::make_shared<SharedConnectionEngine>(&io_service, options, "foo", "", "protocol", 1);
+ 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();
+ io_service->Stop();
ASSERT_TRUE(stat.ok());
});
- io_service.run();
+ io_service->Run();
ASSERT_TRUE(complete);
}
TEST(RpcEngineTest, TestEventCallbacks)
{
- ::asio::io_service io_service;
+ std::shared_ptr<IoService> io_service = IoService::MakeShared();
+
Options options;
options.max_rpc_retries = 99;
options.rpc_retry_delay_ms = 0;
std::shared_ptr<SharedConnectionEngine> engine =
- std::make_shared<SharedConnectionEngine>(&io_service, options, "foo", "", "protocol", 1);
+ 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.
@@ -399,17 +402,18 @@ 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();
+ io_service->Stop();
ASSERT_TRUE(stat.ok());
});
// If you're adding event hooks you'll most likely need to update this.
// It's a brittle test but makes it hard to miss control flow changes in RPC retry.
- for(const auto& m : callbacks)
+ for(const auto& m : callbacks) {
std::cerr << m << std::endl;
- io_service.run();
+ }
+ io_service->Run();
ASSERT_TRUE(complete);
ASSERT_EQ(9, callbacks.size());
ASSERT_EQ(FS_NN_CONNECT_EVENT, callbacks[0]); // error
@@ -430,7 +434,7 @@ TEST(RpcEngineTest, TestConnectionFailureAndAsyncRecover)
producer->checkProducerForConnect = true;
SharedMockConnection::SetSharedConnectionData(producer);
- ::asio::io_service io_service;
+ std::shared_ptr<IoService> io_service = IoService::MakeShared();
bool complete = false;
@@ -438,31 +442,31 @@ TEST(RpcEngineTest, TestConnectionFailureAndAsyncRecover)
options.max_rpc_retries = 1;
options.rpc_retry_delay_ms = 1;
std::shared_ptr<SharedConnectionEngine> engine =
- std::make_shared<SharedConnectionEngine>(&io_service, options, "foo", "", "protocol", 1);
+ 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();
+ io_service->Stop();
ASSERT_TRUE(stat.ok());
});
- ::asio::deadline_timer timer(io_service);
+ ::asio::deadline_timer timer(io_service->GetRaw());
timer.expires_from_now(std::chrono::hours(100));
timer.async_wait([](const asio::error_code & err){(void)err; ASSERT_FALSE("Timed out"); });
- io_service.run();
+ io_service->Run();
ASSERT_TRUE(complete);
}
TEST(RpcEngineTest, TestTimeout) {
- ::asio::io_service io_service;
+ std::shared_ptr<IoService> io_service = IoService::MakeShared();
Options options;
options.rpc_timeout = 1;
- std::shared_ptr<RpcEngine> engine = std::make_shared<RpcEngine>(&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);
conn->TEST_set_connected(true);
@@ -481,15 +485,15 @@ TEST(RpcEngineTest, TestTimeout) {
std::shared_ptr<EchoResponseProto> resp(new EchoResponseProto());
engine->AsyncRpc("test", &req, resp, [resp, &complete,&io_service](const Status &stat) {
complete = true;
- io_service.stop();
+ io_service->Stop();
ASSERT_FALSE(stat.ok());
});
- ::asio::deadline_timer timer(io_service);
+ ::asio::deadline_timer timer(io_service->GetRaw());
timer.expires_from_now(std::chrono::hours(100));
timer.async_wait([](const asio::error_code & err){(void)err; ASSERT_FALSE("Timed out"); });
- io_service.run();
+ io_service->Run();
ASSERT_TRUE(complete);
}
---------------------------------------------------------------------
To unsubscribe, e-mail: common-commits-unsubscribe@hadoop.apache.org
For additional commands, e-mail: common-commits-help@hadoop.apache.org