Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -26,7 +26,7 @@
#include <sstream>
#include <iomanip>
#include <thread>

#include <memory>

namespace hdfs {

Expand Down Expand Up @@ -73,18 +73,20 @@ int DelimitedPBMessageSize(const ::google::protobuf::MessageLite *msg) {
return ::google::protobuf::io::CodedOutputStream::VarintSize32(size) + size;
}

std::string GetRandomClientName() {
std::shared_ptr<std::string> GetRandomClientName() {
std::vector<unsigned char>buf(8);
RAND_pseudo_bytes(&buf[0], 8);
if (RAND_bytes(&buf[0], static_cast<int>(buf.size())) != 1) {
return nullptr;
}

std::ostringstream oss;
oss << "DFSClient_" << getpid() << "_" <<
std::this_thread::get_id() << "_" <<
std::setw(2) << std::hex << std::uppercase << std::setfill('0');
for (unsigned char b: buf)
for (auto b : buf) {
oss << static_cast<unsigned>(b);

return oss.str();
}
return std::make_shared<std::string>(oss.str());
}

std::string Base64Encode(const std::string &src) {
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,7 @@
#include "common/logging.h"

#include <mutex>
#include <memory>
#include <string>

#include <boost/asio/ip/tcp.hpp>
Expand Down Expand Up @@ -61,7 +62,7 @@ std::string SerializeDelimitedProtobufMessage(const ::google::protobuf::MessageL
std::string Base64Encode(const std::string &src);

// Return a new high-entropy client name
std::string GetRandomClientName();
std::shared_ptr<std::string> GetRandomClientName();

// Returns true if _someone_ is holding the lock (not necessarily this thread,
// but a std::mutex doesn't track which thread is holding the lock)
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -24,6 +24,8 @@
#include "hdfspp/events.h"

#include <future>
#include <memory>
#include <string>
#include <tuple>

#include <boost/asio/buffer.hpp>
Expand All @@ -38,7 +40,7 @@ FileHandle::~FileHandle() {}

FileHandleImpl::FileHandleImpl(const std::string & cluster_name,
const std::string & path,
std::shared_ptr<IoService> io_service, const std::string &client_name,
std::shared_ptr<IoService> io_service, const std::shared_ptr<std::string> &client_name,
const std::shared_ptr<const struct FileInfo> file_info,
std::shared_ptr<BadDataNodeTracker> bad_data_nodes,
std::shared_ptr<LibhdfsEvents> event_handlers)
Expand Down Expand Up @@ -191,6 +193,11 @@ void FileHandleImpl::AsyncPreadSome(
return;
}

if (client_name_ == nullptr) {
handler(Status::Error("AsyncPreadSome: Unable to generate random client name"), "", 0);
return;
}

/**
* Note: block and chosen_dn will end up pointing to things inside
* the blocks_ vector. They shouldn't be directly deleted.
Expand Down Expand Up @@ -245,7 +252,7 @@ void FileHandleImpl::AsyncPreadSome(
// steal the FileHandle's dn and put it back when we're done
std::shared_ptr<DataNodeConnection> dn = CreateDataNodeConnection(io_service_, chosen_dn, &block->blocktoken());
std::string dn_id = dn->uuid_;
std::string client_name = client_name_;
std::string client_name = *client_name_;

// Wrap the DN in a block reader to handle the state and logic of the
// block request protocol
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -29,7 +29,9 @@
#include "bad_datanode_tracker.h"
#include "ClientNamenodeProtocol.pb.h"

#include <memory>
#include <mutex>
#include <string>

namespace hdfs {

Expand All @@ -51,10 +53,11 @@ class FileHandleImpl : public FileHandle {
MEMCHECKED_CLASS(FileHandleImpl)
FileHandleImpl(const std::string & cluster_name,
const std::string & path,
std::shared_ptr<IoService> io_service, const std::string &client_name,
const std::shared_ptr<const struct FileInfo> file_info,
std::shared_ptr<BadDataNodeTracker> bad_data_nodes,
std::shared_ptr<LibhdfsEvents> event_handlers);
std::shared_ptr<IoService> io_service,
const std::shared_ptr<std::string> &client_name,
const std::shared_ptr<const struct FileInfo> file_info,
std::shared_ptr<BadDataNodeTracker> bad_data_nodes,
std::shared_ptr<LibhdfsEvents> event_handlers);

/*
* Reads the file at the specified offset into the buffer.
Expand Down Expand Up @@ -129,7 +132,7 @@ class FileHandleImpl : public FileHandle {
const std::string cluster_name_;
const std::string path_;
std::shared_ptr<IoService> io_service_;
const std::string client_name_;
const std::shared_ptr<std::string> client_name_;
const std::shared_ptr<const struct FileInfo> file_info_;
std::shared_ptr<BadDataNodeTracker> bad_node_tracker_;
bool CheckSeekBounds(ssize_t desired_position);
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -23,6 +23,7 @@
#include "hdfspp/hdfspp.h"
#include "reader/fileinfo.h"

#include <memory>
#include <thread>

namespace hdfs {
Expand Down Expand Up @@ -217,7 +218,7 @@ class FileSystemImpl : public FileSystem {
**/
std::shared_ptr<IoService> io_service_;
const Options options_;
const std::string client_name_;
const std::shared_ptr<std::string> client_name_;
std::string cluster_name_;
NameNodeOperations nn_;
std::shared_ptr<BadDataNodeTracker> bad_node_tracker_;
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -26,6 +26,8 @@
#include "ClientNamenodeProtocol.pb.h"
#include "ClientNamenodeProtocol.hrpc.inl"

#include <memory>
#include <string>

namespace hdfs {

Expand All @@ -43,7 +45,7 @@ class NameNodeOperations {
public:
MEMCHECKED_CLASS(NameNodeOperations)
NameNodeOperations(std::shared_ptr<IoService> io_service, const Options &options,
const std::string &client_name, const std::string &user_name,
const std::shared_ptr<std::string> &client_name, const std::string &user_name,
const char *protocol_name, int protocol_version) :
io_service_(io_service),
engine_(std::make_shared<RpcEngine>(io_service, options, client_name, user_name, protocol_name, protocol_version)),
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -306,13 +306,19 @@ std::shared_ptr<std::string> RpcConnection::PrepareContextPacket() {
return std::make_shared<std::string>();
}

const auto& client_name = pinnedEngine->client_name();
if (client_name == nullptr) {
LOG_ERROR(kRPC, << "RpcConnection@" << this << " unable to generate random client name");
return std::make_shared<std::string>();
}

std::shared_ptr<std::string> serializedPacketBuffer = std::make_shared<std::string>();

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());
headerProto.set_clientid(*client_name);

IpcConnectionContextProto handshakeContextProto;
handshakeContextProto.set_protocol(pinnedEngine->protocol_name());
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -24,6 +24,7 @@

#include <algorithm>
#include <memory>
#include <string>

#include <boost/date_time/posix_time/posix_time_duration.hpp>
#include <openssl/rand.h>
Expand All @@ -36,7 +37,7 @@ using optional = std::experimental::optional<T>;


RpcEngine::RpcEngine(std::shared_ptr<IoService> io_service, const Options &options,
const std::string &client_name, const std::string &user_name,
const std::shared_ptr<std::string> &client_name, const std::string &user_name,
const char *protocol_name, int protocol_version)
: io_service_(io_service),
options_(options),
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -38,6 +38,7 @@
#include <memory>
#include <vector>
#include <mutex>
#include <string>

namespace hdfs {

Expand Down Expand Up @@ -79,7 +80,7 @@ class LockFreeRpcEngine {
virtual const RetryPolicy *retry_policy() = 0;
virtual int NextCallId() = 0;

virtual const std::string &client_name() = 0;
virtual const std::shared_ptr<std::string> &client_name() = 0;
virtual const std::unique_ptr<std::string> &client_id() = 0;
virtual const std::string &user_name() = 0;
virtual const std::string &protocol_name() = 0;
Expand Down Expand Up @@ -109,7 +110,7 @@ class RpcEngine : public LockFreeRpcEngine, public std::enable_shared_from_this<
};

RpcEngine(std::shared_ptr<IoService> service, const Options &options,
const std::string &client_name, const std::string &user_name,
const std::shared_ptr<std::string> &client_name, const std::string &user_name,
const char *protocol_name, int protocol_version);

void Connect(const std::string & cluster_name,
Expand Down Expand Up @@ -141,7 +142,7 @@ class RpcEngine : public LockFreeRpcEngine, public std::enable_shared_from_this<
void TEST_SetRetryPolicy(std::unique_ptr<const RetryPolicy> policy);
std::unique_ptr<const RetryPolicy> TEST_GenerateRetryPolicyUsingOptions();

const std::string &client_name() override { return client_name_; }
const std::shared_ptr<std::string> &client_name() override { return client_name_; }
const std::unique_ptr<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_; }
Expand All @@ -165,7 +166,7 @@ class RpcEngine : public LockFreeRpcEngine, public std::enable_shared_from_this<
private:
mutable std::shared_ptr<IoService> io_service_;
const Options options_;
const std::string client_name_;
const std::shared_ptr<std::string> client_name_;
const std::unique_ptr<std::string> client_id_;
const std::string protocol_name_;
const int protocol_version_;
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -23,11 +23,16 @@
#include "fs/bad_datanode_tracker.h"
#include "reader/block_reader.h"

#include <gtest/gtest.h>
#include <gmock/gmock.h>
#include <gmock/gmock-spec-builders.h>
#include <gmock/gmock-generated-actions.h>

#include <boost/asio/buffer.hpp>
#include <boost/asio/error.hpp>



using hadoop::common::TokenProto;
using hadoop::hdfs::DatanodeInfoProto;
using hadoop::hdfs::DatanodeIDProto;
Expand Down Expand Up @@ -139,7 +144,10 @@ TEST(BadDataNodeTest, TestNoNodes) {
auto monitors = std::make_shared<LibhdfsEvents>();
bad_node_tracker->AddBadNode("foo");

PartialMockFileHandle is("cluster", "file", io_service, GetRandomClientName(), file_info, bad_node_tracker, monitors);
const auto client_name = GetRandomClientName();
ASSERT_NE(client_name, nullptr);

PartialMockFileHandle is("cluster", "file", io_service, client_name, file_info, bad_node_tracker, monitors);
Status stat;
size_t read = 0;

Expand Down Expand Up @@ -195,7 +203,11 @@ TEST(BadDataNodeTest, NNEventCallback) {

return event_response::make_ok();
});
PartialMockFileHandle is("cluster", "file", io_service, GetRandomClientName(), file_info, tracker, monitors);

const auto client_name = GetRandomClientName();
ASSERT_NE(client_name, nullptr);

PartialMockFileHandle is("cluster", "file", io_service, client_name, file_info, tracker, monitors);
Status stat;
size_t read = 0;

Expand Down Expand Up @@ -241,7 +253,11 @@ TEST(BadDataNodeTest, RecoverableError) {
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, GetRandomClientName(), file_info, tracker, monitors);

const auto client_name = GetRandomClientName();
ASSERT_NE(client_name, nullptr);

PartialMockFileHandle is("cluster", "file", io_service, client_name, file_info, tracker, monitors);
Status stat;
size_t read = 0;
EXPECT_CALL(*is.mock_reader_, AsyncReadBlock(_,_,_,_,_))
Expand Down Expand Up @@ -292,7 +308,11 @@ TEST(BadDataNodeTest, InternalError) {
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, GetRandomClientName(), file_info, tracker, monitors);

const auto client_name = GetRandomClientName();
ASSERT_NE(client_name, nullptr);

PartialMockFileHandle is("cluster", "file", io_service, client_name, file_info, tracker, monitors);
Status stat;
size_t read = 0;
EXPECT_CALL(*is.mock_reader_, AsyncReadBlock(_,_,_,_,_))
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -27,8 +27,10 @@

#include <google/protobuf/io/coded_stream.h>
#include <google/protobuf/io/zero_copy_stream_impl.h>
#include <gmock/gmock.h>
#include <gtest/gtest.h>
#include <gmock/gmock.h>
#include <gmock/gmock-spec-builders.h>
#include <gmock/gmock-generated-actions.h>
#include <boost/system/error_code.hpp>
#include <boost/asio/buffer.hpp>
#include <boost/asio/io_service.hpp>
Expand Down Expand Up @@ -165,8 +167,10 @@ TEST(RemoteBlockReaderTest, TestReadSingleTrunk) {
EXPECT_CALL(reader, AsyncReadPacket(_, _))
.WillOnce(InvokeArgument<1>(Status::OK(), sizeof(buf)));

const auto client_name = GetRandomClientName();
ASSERT_NE(client_name, nullptr);
reader.AsyncReadBlock(
GetRandomClientName(), block, 0, boost::asio::buffer(buf, sizeof(buf)),
*client_name, block, 0, boost::asio::buffer(buf, sizeof(buf)),
[&stat, &read](const Status &status, size_t transferred) {
stat = status;
read = transferred;
Expand All @@ -192,8 +196,10 @@ TEST(RemoteBlockReaderTest, TestReadMultipleTrunk) {
.Times(4)
.WillRepeatedly(InvokeArgument<1>(Status::OK(), sizeof(buf) / 4));

const auto client_name = GetRandomClientName();
ASSERT_NE(client_name, nullptr);
reader.AsyncReadBlock(
GetRandomClientName(), block, 0, boost::asio::buffer(buf, sizeof(buf)),
*client_name, block, 0, boost::asio::buffer(buf, sizeof(buf)),
[&stat, &read](const Status &status, size_t transferred) {
stat = status;
read = transferred;
Expand All @@ -220,8 +226,10 @@ TEST(RemoteBlockReaderTest, TestReadError) {
.WillOnce(InvokeArgument<1>(Status::OK(), sizeof(buf) / 4))
.WillOnce(InvokeArgument<1>(Status::Error("error"), 0));

const auto client_name = GetRandomClientName();
ASSERT_NE(client_name, nullptr);
reader.AsyncReadBlock(
GetRandomClientName(), block, 0, boost::asio::buffer(buf, sizeof(buf)),
*client_name, block, 0, boost::asio::buffer(buf, sizeof(buf)),
[&stat, &read](const Status &status, size_t transferred) {
stat = status;
read = transferred;
Expand Down
Loading