big-data/hadoop
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

HDFS-9890. libhdfs++: Add test suite to simulate network issues. Contributed by Xiaowei Zhu.

Browse Source
This commit is contained in:
James 2016-06-06 14:07:57 -04:00 committed by James Clampffer
parent 50aa4ea776
commit 11b68ecebe
18 changed files with 498 additions and 63 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
hadoop-hdfs-project/hadoop-hdfs-native-client/pom.xml
View File

@ -147,7 +147,7 @@ http://maven.apache.org/xsd/maven-4.0.0.xsd">
<mkdir dir="${project.build.directory}/native"/>
<exec executable="cmake" dir="${project.build.directory}/native"
failonerror="true">
<arg line="${basedir}/src/ -DGENERATED_JAVAH=${project.build.directory}/native/javah -DJVM_ARCH_DATA_MODEL=${sun.arch.data.model} -DHADOOP_BUILD=1 -DREQUIRE_FUSE=${require.fuse} -DREQUIRE_VALGRIND=${require.valgrind} -A '${env.PLATFORM}'"/>
<arg line="${basedir}/src/ -DCMAKE_BUILD_TYPE=Debug -DGENERATED_JAVAH=${project.build.directory}/native/javah -DJVM_ARCH_DATA_MODEL=${sun.arch.data.model} -DHADOOP_BUILD=1 -DREQUIRE_FUSE=${require.fuse} -DREQUIRE_VALGRIND=${require.valgrind} -A '${env.PLATFORM}'"/>
<arg line="${native_cmake_args}"/>
</exec>
<exec executable="msbuild" dir="${project.build.directory}/native"
@ -208,7 +208,7 @@ http://maven.apache.org/xsd/maven-4.0.0.xsd">
<target>
<mkdir dir="${project.build.directory}"/>
<exec executable="cmake" dir="${project.build.directory}" failonerror="true">
<arg line="${basedir}/src/ -DGENERATED_JAVAH=${project.build.directory}/native/javah -DJVM_ARCH_DATA_MODEL=${sun.arch.data.model} -DHADOOP_BUILD=1 -DREQUIRE_LIBWEBHDFS=${require.libwebhdfs} -DREQUIRE_FUSE=${require.fuse} -DREQUIRE_VALGRIND=${require.valgrind} "/>
<arg line="${basedir}/src/ -DCMAKE_BUILD_TYPE=Debug -DGENERATED_JAVAH=${project.build.directory}/native/javah -DJVM_ARCH_DATA_MODEL=${sun.arch.data.model} -DHADOOP_BUILD=1 -DREQUIRE_LIBWEBHDFS=${require.libwebhdfs} -DREQUIRE_FUSE=${require.fuse} -DREQUIRE_VALGRIND=${require.valgrind} "/>
<arg line="${native_cmake_args}"/>
</exec>
<exec executable="make" dir="${project.build.directory}" failonerror="true">

10
hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfs-tests/native_mini_dfs.c
View File

@ -182,6 +182,16 @@ struct NativeMiniDfsCluster* nmdCreate(struct NativeMiniDfsConf *conf)
}
(*env)->DeleteLocalRef(env, val.l);
}
if (conf->numDataNodes) {
jthr = invokeMethod(env, &val, INSTANCE, bld, MINIDFS_CLUSTER_BUILDER,
"numDataNodes", "(I)L" MINIDFS_CLUSTER_BUILDER ";", conf->numDataNodes);
if (jthr) {
printExceptionAndFree(env, jthr, PRINT_EXC_ALL, "nmdCreate: "
"Builder::numDataNodes");
goto error;
}
}
(*env)->DeleteLocalRef(env, val.l);
jthr = invokeMethod(env, &val, INSTANCE, bld, MINIDFS_CLUSTER_BUILDER,
"build", "()L" MINIDFS_CLUSTER ";");
if (jthr) {

5
hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfs-tests/native_mini_dfs.h
View File

@ -51,6 +51,11 @@ struct NativeMiniDfsConf {
* Nonzero if we should configure short circuit.
*/
jboolean configureShortCircuit;
/**
* The number of datanodes in MiniDfsCluster
*/
jint numDataNodes;
};
/**

338
hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfs-tests/test_libhdfs_mini_stress.c Normal file
View File

@ -0,0 +1,338 @@
/**
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. The ASF licenses this file
* to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
#include "expect.h"
#include "hdfs/hdfs.h"
#include "hdfspp/hdfs_ext.h"
#include "native_mini_dfs.h"
#include "os/thread.h"
#include <errno.h>
#include <inttypes.h>
#include <stdint.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#define TO_STR_HELPER(X) #X
#define TO_STR(X) TO_STR_HELPER(X)
#define TLH_MAX_THREADS 10000
#define TLH_MAX_DNS 16
#define TLH_DEFAULT_BLOCK_SIZE 1048576
#define TLH_DEFAULT_DFS_REPLICATION 3
#define TLH_DEFAULT_IPC_CLIENT_CONNECT_MAX_RETRIES 100
#define TLH_DEFAULT_IPC_CLIENT_CONNECT_RETRY_INTERVAL_MS 5
#ifndef RANDOM_ERROR_RATIO
#define RANDOM_ERROR_RATIO 1000000000
#endif
struct tlhThreadInfo {
/** Thread index */
int threadIdx;
/** 0 = thread was successful; error code otherwise */
int success;
/** thread identifier */
thread theThread;
/** fs, shared with other threads **/
hdfsFS hdfs;
/** Filename */
const char *fileNm;
};
static int hdfsNameNodeConnect(struct NativeMiniDfsCluster *cl, hdfsFS *fs,
const char *username)
{
int ret;
tPort port;
hdfsFS hdfs;
struct hdfsBuilder *bld;
port = (tPort)nmdGetNameNodePort(cl);
if (port < 0) {
fprintf(stderr, "hdfsNameNodeConnect: nmdGetNameNodePort "
"returned error %d\n", port);
return port;
}
bld = hdfsNewBuilder();
if (!bld)
return -ENOMEM;
hdfsBuilderSetForceNewInstance(bld);
hdfsBuilderSetNameNode(bld, "localhost");
hdfsBuilderSetNameNodePort(bld, port);
hdfsBuilderConfSetStr(bld, "dfs.block.size",
TO_STR(TLH_DEFAULT_BLOCK_SIZE));
hdfsBuilderConfSetStr(bld, "dfs.blocksize",
TO_STR(TLH_DEFAULT_BLOCK_SIZE));
hdfsBuilderConfSetStr(bld, "dfs.replication",
TO_STR(TLH_DEFAULT_DFS_REPLICATION));
hdfsBuilderConfSetStr(bld, "ipc.client.connect.max.retries",
TO_STR(TLH_DEFAULT_IPC_CLIENT_CONNECT_MAX_RETRIES));
hdfsBuilderConfSetStr(bld, "ipc.client.connect.retry.interval",
TO_STR(TLH_DEFAULT_IPC_CLIENT_CONNECT_RETRY_INTERVAL_MS));
if (username) {
hdfsBuilderSetUserName(bld, username);
}
hdfs = hdfsBuilderConnect(bld);
if (!hdfs) {
ret = -errno;
return ret;
}
*fs = hdfs;
return 0;
}
static int hdfsWriteData(hdfsFS hdfs, const char *dirNm,
const char *fileNm, tSize fileSz)
{
hdfsFile file;
int ret, expected;
const char *content;
content = fileNm;
if (hdfsExists(hdfs, dirNm) == 0) {
EXPECT_ZERO(hdfsDelete(hdfs, dirNm, 1));
}
EXPECT_ZERO(hdfsCreateDirectory(hdfs, dirNm));
file = hdfsOpenFile(hdfs, fileNm, O_WRONLY, 0, 0, 0);
EXPECT_NONNULL(file);
expected = (int)strlen(content);
tSize sz = 0;
while (sz < fileSz) {
ret = hdfsWrite(hdfs, file, content, expected);
if (ret < 0) {
ret = errno;
fprintf(stderr, "hdfsWrite failed and set errno %d\n", ret);
return ret;
}
if (ret != expected) {
fprintf(stderr, "hdfsWrite was supposed to write %d bytes, but "
"it wrote %d\n", ret, expected);
return EIO;
}
sz += ret;
}
EXPECT_ZERO(hdfsFlush(hdfs, file));
EXPECT_ZERO(hdfsHSync(hdfs, file));
EXPECT_ZERO(hdfsCloseFile(hdfs, file));
return 0;
}
static int fileEventCallback(const char * event, const char * cluster, const char * file, int64_t value, int64_t cookie)
{
char * randomErrRatioStr = getenv("RANDOM_ERROR_RATIO");
int64_t randomErrRatio = RANDOM_ERROR_RATIO;
if (randomErrRatioStr) randomErrRatio = (int64_t)atoi(randomErrRatioStr);
if (randomErrRatio == 0) return DEBUG_SIMULATE_ERROR;
else if (randomErrRatio < 0) return LIBHDFSPP_EVENT_OK;
return random() % randomErrRatio == 0 ? DEBUG_SIMULATE_ERROR : LIBHDFSPP_EVENT_OK;
}
static int doTestHdfsMiniStress(struct tlhThreadInfo *ti)
{
char tmp[4096];
hdfsFile file;
int ret, expected;
hdfsFileInfo *fileInfo;
uint64_t readOps, nErrs=0;
tOffset seekPos;
const char *content;
content = ti->fileNm;
expected = (int)strlen(content);
fileInfo = hdfsGetPathInfo(ti->hdfs, ti->fileNm);
EXPECT_NONNULL(fileInfo);
file = hdfsOpenFile(ti->hdfs, ti->fileNm, O_RDONLY, 0, 0, 0);
EXPECT_NONNULL(file);
libhdfspp_file_event_callback callback = &fileEventCallback;
hdfsPreAttachFileMonitor(callback, 0);
fprintf(stderr, "testHdfsMiniStress(threadIdx=%d): starting read loop\n",
ti->threadIdx);
for (readOps=0; readOps < 1000; ++readOps) {
EXPECT_ZERO(hdfsCloseFile(ti->hdfs, file));
file = hdfsOpenFile(ti->hdfs, ti->fileNm, O_RDONLY, 0, 0, 0);
EXPECT_NONNULL(file);
seekPos = (((double)random()) / RAND_MAX) * (fileInfo->mSize - expected);
seekPos = (seekPos / expected) * expected;
ret = hdfsSeek(ti->hdfs, file, seekPos);
if (ret < 0) {
ret = errno;
fprintf(stderr, "hdfsSeek to %"PRIu64" failed and set"
" errno %d\n", seekPos, ret);
++nErrs;
continue;
}
ret = hdfsRead(ti->hdfs, file, tmp, expected);
if (ret < 0) {
ret = errno;
fprintf(stderr, "hdfsRead failed and set errno %d\n", ret);
++nErrs;
continue;
}
if (ret != expected) {
fprintf(stderr, "hdfsRead was supposed to read %d bytes, but "
"it read %d\n", ret, expected);
++nErrs;
continue;
}
ret = memcmp(content, tmp, expected);
if (ret) {
fprintf(stderr, "hdfsRead result (%.*s) does not match expected (%.*s)",
expected, tmp, expected, content);
++nErrs;
continue;
}
}
EXPECT_ZERO(hdfsCloseFile(ti->hdfs, file));
fprintf(stderr, "testHdfsMiniStress(threadIdx=%d): finished read loop\n",
ti->threadIdx);
EXPECT_ZERO(nErrs);
return 0;
}
static int testHdfsMiniStressImpl(struct tlhThreadInfo *ti)
{
fprintf(stderr, "testHdfsMiniStress(threadIdx=%d): starting\n",
ti->threadIdx);
EXPECT_NONNULL(ti->hdfs);
EXPECT_ZERO(doTestHdfsMiniStress(ti));
return 0;
}
static void testHdfsMiniStress(void *v)
{
struct tlhThreadInfo *ti = (struct tlhThreadInfo*)v;
int ret = testHdfsMiniStressImpl(ti);
ti->success = ret;
}
static int checkFailures(struct tlhThreadInfo *ti, int tlhNumThreads)
{
int i, threadsFailed = 0;
const char *sep = "";
for (i = 0; i < tlhNumThreads; i++) {
if (ti[i].success != 0) {
threadsFailed = 1;
}
}
if (!threadsFailed) {
fprintf(stderr, "testLibHdfsMiniStress: all threads succeeded. SUCCESS.\n");
return EXIT_SUCCESS;
}
fprintf(stderr, "testLibHdfsMiniStress: some threads failed: [");
for (i = 0; i < tlhNumThreads; i++) {
if (ti[i].success != 0) {
fprintf(stderr, "%s%d", sep, i);
sep = ", ";
}
}
fprintf(stderr, "]. FAILURE.\n");
return EXIT_FAILURE;
}
/**
* Test intended to stress libhdfs client with concurrent requests. Currently focused
* on concurrent reads.
*/
int main(void)
{
int i, tlhNumThreads;
char *dirNm, *fileNm;
tSize fileSz;
const char *tlhNumThreadsStr, *tlhNumDNsStr;
hdfsFS hdfs = NULL;
struct NativeMiniDfsCluster* tlhCluster;
struct tlhThreadInfo ti[TLH_MAX_THREADS];
struct NativeMiniDfsConf conf = {
1, /* doFormat */
};
dirNm = "/tlhMiniStressData";
fileNm = "/tlhMiniStressData/file";
fileSz = 2*1024*1024;
tlhNumDNsStr = getenv("TLH_NUM_DNS");
if (!tlhNumDNsStr) {
tlhNumDNsStr = "1";
}
conf.numDataNodes = atoi(tlhNumDNsStr);
if ((conf.numDataNodes <= 0) || (conf.numDataNodes > TLH_MAX_DNS)) {
fprintf(stderr, "testLibHdfsMiniStress: must have a number of datanodes "
"between 1 and %d inclusive, not %d\n",
TLH_MAX_DNS, conf.numDataNodes);
return EXIT_FAILURE;
}
tlhNumThreadsStr = getenv("TLH_NUM_THREADS");
if (!tlhNumThreadsStr) {
tlhNumThreadsStr = "8";
}
tlhNumThreads = atoi(tlhNumThreadsStr);
if ((tlhNumThreads <= 0) || (tlhNumThreads > TLH_MAX_THREADS)) {
fprintf(stderr, "testLibHdfsMiniStress: must have a number of threads "
"between 1 and %d inclusive, not %d\n",
TLH_MAX_THREADS, tlhNumThreads);
return EXIT_FAILURE;
}
memset(&ti[0], 0, sizeof(ti));
for (i = 0; i < tlhNumThreads; i++) {
ti[i].threadIdx = i;
}
tlhCluster = nmdCreate(&conf);
EXPECT_NONNULL(tlhCluster);
EXPECT_ZERO(nmdWaitClusterUp(tlhCluster));
EXPECT_ZERO(hdfsNameNodeConnect(tlhCluster, &hdfs, NULL));
// Single threaded writes for now.
EXPECT_ZERO(hdfsWriteData(hdfs, dirNm, fileNm, fileSz));
// Multi-threaded reads.
for (i = 0; i < tlhNumThreads; i++) {
ti[i].theThread.start = testHdfsMiniStress;
ti[i].theThread.arg = &ti[i];
ti[i].hdfs = hdfs;
ti[i].fileNm = fileNm;
EXPECT_ZERO(threadCreate(&ti[i].theThread));
}
for (i = 0; i < tlhNumThreads; i++) {
EXPECT_ZERO(threadJoin(&ti[i].theThread));
}
EXPECT_ZERO(hdfsDisconnect(hdfs));
EXPECT_ZERO(nmdShutdown(tlhCluster));
nmdFree(tlhCluster);
return checkFailures(ti, tlhNumThreads);
}

3
hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/bindings/c/hdfs.cc
View File

@ -298,6 +298,9 @@ hdfsFile hdfsOpenFile(hdfsFS fs, const char *path, int flags, int bufferSize,
Error(stat);
return nullptr;
}
if (f && fileEventCallback) {
f->SetFileEventCallback(fileEventCallback.value());
}
return new hdfsFile_internal(f);
} catch (const std::exception & e) {
ReportException(e);

11
hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/fs/filehandle.cc
View File

@ -231,7 +231,7 @@ void FileHandleImpl::AsyncPreadSome(
// Wrap the DN in a block reader to handle the state and logic of the
// block request protocol
std::shared_ptr<BlockReader> reader;
reader = CreateBlockReader(BlockReaderOptions(), dn);
reader = CreateBlockReader(BlockReaderOptions(), dn, event_handlers_);
// Lambdas cannot capture copies of member variables so we'll make explicit
// copies for it
@ -240,7 +240,7 @@ void FileHandleImpl::AsyncPreadSome(
auto cluster_name = cluster_name_;
auto read_handler = [reader, event_handlers, cluster_name, path, dn_id, handler](const Status & status, size_t transferred) {
auto event_resp = event_handlers->call(FILE_DN_READ_EVENT, cluster_name.c_str(), path.c_str(), transferred);
event_response event_resp = event_handlers->call(FILE_DN_READ_EVENT, cluster_name.c_str(), path.c_str(), transferred);
#ifndef NDEBUG
if (event_resp.response() == event_response::kTest_Error) {
handler(event_resp.status(), dn_id, transferred);
@ -254,7 +254,7 @@ void FileHandleImpl::AsyncPreadSome(
auto connect_handler = [handler,event_handlers,cluster_name,path,read_handler,block,offset_within_block,size_within_block, buffers, reader, dn_id, client_name]
(Status status, std::shared_ptr<DataNodeConnection> dn) {
(void)dn;
auto event_resp = event_handlers->call(FILE_DN_CONNECT_EVENT, cluster_name.c_str(), path.c_str(), 0);
event_response event_resp = event_handlers->call(FILE_DN_CONNECT_EVENT, cluster_name.c_str(), path.c_str(), 0);
#ifndef NDEBUG
if (event_resp.response() == event_response::kTest_Error) {
status = event_resp.status();
@ -276,9 +276,10 @@ void FileHandleImpl::AsyncPreadSome(
}
std::shared_ptr<BlockReader> FileHandleImpl::CreateBlockReader(const BlockReaderOptions &options,
std::shared_ptr<DataNodeConnection> dn)
std::shared_ptr<DataNodeConnection> dn,
std::shared_ptr<LibhdfsEvents> event_handlers)
{
std::shared_ptr<BlockReader> reader = std::make_shared<BlockReaderImpl>(options, dn, cancel_state_);
std::shared_ptr<BlockReader> reader = std::make_shared<BlockReaderImpl>(options, dn, cancel_state_, event_handlers);
LOG_TRACE(kFileHandle, << "FileHandleImpl::CreateBlockReader(" << FMT_THIS_ADDR
<< ", ..., dnconn=" << dn.get()

3
hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/fs/filehandle.h
View File

@ -119,7 +119,8 @@ public:
protected:
virtual std::shared_ptr<BlockReader> CreateBlockReader(const BlockReaderOptions &options,
std::shared_ptr<DataNodeConnection> dn);
std::shared_ptr<DataNodeConnection> dn,
std::shared_ptr<hdfs::LibhdfsEvents> event_handlers);
virtual std::shared_ptr<DataNodeConnection> CreateDataNodeConnection(
::asio::io_service *io_service,
const ::hadoop::hdfs::DatanodeInfoProto & dn,

2
hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/fs/filesystem.cc
View File

@ -96,7 +96,7 @@ FileSystemImpl::FileSystemImpl(IoService *&io_service, const std::string &user_n
io_service = nullptr;
/* spawn background threads for asio delegation */
unsigned int threads = 1 /* options.io_threads_, pending HDFS-9117 */;
unsigned int threads = 2 /* options.io_threads_, pending HDFS-9117 */;
for (unsigned int i = 0; i < threads; i++) {
AddWorkerThread();
}

62
hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/reader/block_reader.cc
View File

@ -24,7 +24,6 @@
#include <future>
namespace hdfs {
#define FMT_CONT_AND_PARENT_ADDR "this=" << (void*)this << ", parent=" << (void*)parent_
@ -105,7 +104,17 @@ void BlockReaderImpl::AsyncRequestBlock(
m->Run([this, handler, offset](const Status &status, const State &s) { Status stat = status;
if (stat.ok()) {
const auto &resp = s.response;
if (resp.status() == ::hadoop::hdfs::Status::SUCCESS) {
if(this->event_handlers_) {
event_response event_resp = this->event_handlers_->call(FILE_DN_READ_EVENT, "", "", 0);
#ifndef NDEBUG
if (stat.ok() && event_resp.response() == event_response::kTest_Error) {
stat = Status::Error("Test error");
}
#endif
}
if (stat.ok() && resp.status() == ::hadoop::hdfs::Status::SUCCESS) {
if (resp.has_readopchecksuminfo()) {
const auto &checksum_info = resp.readopchecksuminfo();
chunk_padding_bytes_ = offset - checksum_info.chunkoffset();
@ -162,6 +171,14 @@ struct BlockReaderImpl::ReadPacketHeader
assert(v && "Failed to parse the header");
parent_->state_ = kReadChecksum;
}
if(parent_->event_handlers_) {
event_response event_resp = parent_->event_handlers_->call(FILE_DN_READ_EVENT, "", "", 0);
#ifndef NDEBUG
if (status.ok() && event_resp.response() == event_response::kTest_Error) {
status = Status::Error("Test error");
}
#endif
}
next(status);
};
@ -214,7 +231,7 @@ struct BlockReaderImpl::ReadChecksum : continuation::Continuation {
return;
}
auto handler = [parent, next](const asio::error_code &ec, size_t) {
auto handler = [parent, next, this](const asio::error_code &ec, size_t) {
Status status;
if (ec) {
status = Status(ec.value(), ec.message().c_str());
@ -222,6 +239,14 @@ struct BlockReaderImpl::ReadChecksum : continuation::Continuation {
parent->state_ =
parent->chunk_padding_bytes_ ? kReadPadding : kReadData;
}
if(parent->event_handlers_) {
event_response event_resp = parent->event_handlers_->call(FILE_DN_READ_EVENT, "", "", 0);
#ifndef NDEBUG
if (status.ok() && event_resp.response() == event_response::kTest_Error) {
status = Status::Error("Test error");
}
#endif
}
next(status);
};
parent->checksum_.resize(parent->packet_len_ - sizeof(int) -
@ -248,7 +273,6 @@ struct BlockReaderImpl::ReadData : continuation::Continuation {
virtual void Run(const Next &next) override {
LOG_TRACE(kBlockReader, << "BlockReaderImpl::ReadData::Run("
<< FMT_CONT_AND_PARENT_ADDR << ") called");
auto handler =
[next, this](const asio::error_code &ec, size_t transferred) {
Status status;
@ -261,6 +285,14 @@ struct BlockReaderImpl::ReadData : continuation::Continuation {
if (parent_->packet_data_read_bytes_ >= parent_->header_.datalen()) {
parent_->state_ = kReadPacketHeader;
}
if(parent_->event_handlers_) {
event_response event_resp = parent_->event_handlers_->call(FILE_DN_READ_EVENT, "", "", 0);
#ifndef NDEBUG
if (status.ok() && event_resp.response() == event_response::kTest_Error) {
status = Status::Error("Test error");
}
#endif
}
next(status);
};
@ -292,13 +324,22 @@ struct BlockReaderImpl::ReadPadding : continuation::Continuation {
return;
}
auto h = [next, this](const Status &status) {
auto h = [next, this](const Status &stat) {
Status status = stat;
if (status.ok()) {
assert(reinterpret_cast<const int &>(*bytes_transferred_) ==
parent_->chunk_padding_bytes_);
parent_->chunk_padding_bytes_ = 0;
parent_->state_ = kReadData;
}
if(parent_->event_handlers_) {
event_response event_resp = parent_->event_handlers_->call(FILE_DN_READ_EVENT, "", "", 0);
#ifndef NDEBUG
if (status.ok() && event_resp.response() == event_response::kTest_Error) {
status = Status::Error("Test error");
}
#endif
}
next(status);
};
read_data_->Run(h);
@ -334,11 +375,20 @@ struct BlockReaderImpl::AckRead : continuation::Continuation {
m->Push(
continuation::WriteDelimitedPBMessage(parent_->dn_, &m->state()));
m->Run([this, next](const Status &status,
m->Run([this, next](const Status &stat,
const hadoop::hdfs::ClientReadStatusProto &) {
Status status = stat;
if (status.ok()) {
parent_->state_ = BlockReaderImpl::kFinished;
}
if(parent_->event_handlers_) {
event_response event_resp = parent_->event_handlers_->call(FILE_DN_READ_EVENT, "", "", 0);
#ifndef NDEBUG
if (status.ok() && event_resp.response() == event_response::kTest_Error) {
status = Status::Error("Test error");
}
#endif
}
next(status);
});
}

5
hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/reader/block_reader.h
View File

@ -93,9 +93,9 @@ class BlockReaderImpl
: public BlockReader, public std::enable_shared_from_this<BlockReaderImpl> {
public:
explicit BlockReaderImpl(const BlockReaderOptions &options, std::shared_ptr<DataNodeConnection> dn,
CancelHandle cancel_state)
CancelHandle cancel_state, std::shared_ptr<LibhdfsEvents> event_handlers=nullptr)
: dn_(dn), state_(kOpen), options_(options),
chunk_padding_bytes_(0), cancel_state_(cancel_state) {}
chunk_padding_bytes_(0), cancel_state_(cancel_state), event_handlers_(event_handlers.get()) {}
virtual void AsyncReadPacket(
const MutableBuffers &buffers,
@ -152,6 +152,7 @@ private:
long long bytes_to_read_;
std::vector<char> checksum_;
CancelHandle cancel_state_;
LibhdfsEvents* event_handlers_;
};
}

11
hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/rpc/rpc_connection.cc
View File

@ -274,7 +274,16 @@ void RpcConnection::HandleRpcResponse(std::shared_ptr<Response> response) {
}
Status status;
if (h.has_exceptionclassname()) {
if(event_handlers_) {
event_response event_resp = event_handlers_->call(FS_NN_READ_EVENT, cluster_name_.c_str(), 0);
#ifndef NDEBUG
if (event_resp.response() == event_response::kTest_Error) {
status = event_resp.status();
}
#endif
}
if (status.ok() && h.has_exceptionclassname()) {
status =
Status::Exception(h.exceptionclassname().c_str(), h.errormsg().c_str());
}

26
hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/rpc/rpc_connection.h
View File

@ -30,6 +30,8 @@
#include <asio/read.hpp>
#include <asio/write.hpp>
#include <system_error>
namespace hdfs {
template <class NextLayer>
@ -63,6 +65,10 @@ public:
NextLayer next_layer_;
void ConnectComplete(const ::asio::error_code &ec);
// Hide default ctors.
RpcConnectionImpl();
RpcConnectionImpl(const RpcConnectionImpl &other);
};
template <class NextLayer>
@ -70,7 +76,7 @@ RpcConnectionImpl<NextLayer>::RpcConnectionImpl(RpcEngine *engine)
: RpcConnection(engine),
options_(engine->options()),
next_layer_(engine->io_service()) {
LOG_TRACE(kRPC, << "RpcConnectionImpl::RpcConnectionImpl called");
LOG_TRACE(kRPC, << "RpcConnectionImpl::RpcConnectionImpl called &" << (void*)this);
}
template <class NextLayer>
@ -84,7 +90,6 @@ RpcConnectionImpl<NextLayer>::~RpcConnectionImpl() {
LOG_WARN(kRPC, << "RpcConnectionImpl::~RpcConnectionImpl called with items in the requests_on_fly queue");
}
template <class NextLayer>
void RpcConnectionImpl<NextLayer>::Connect(
const std::vector<::asio::ip::tcp::endpoint> &server,
@ -145,7 +150,7 @@ void RpcConnectionImpl<NextLayer>::ConnectComplete(const ::asio::error_code &ec)
Status status = ToStatus(ec);
if(event_handlers_) {
auto event_resp = event_handlers_->call(FS_NN_CONNECT_EVENT, cluster_name_.c_str(), 0);
event_response event_resp = event_handlers_->call(FS_NN_CONNECT_EVENT, cluster_name_.c_str(), 0);
#ifndef NDEBUG
if (event_resp.response() == event_response::kTest_Error) {
status = event_resp.status();
@ -310,27 +315,28 @@ void RpcConnectionImpl<NextLayer>::FlushPendingRequests() {
template <class NextLayer>
void RpcConnectionImpl<NextLayer>::OnRecvCompleted(const ::asio::error_code &asio_ec,
void RpcConnectionImpl<NextLayer>::OnRecvCompleted(const ::asio::error_code &original_ec,
size_t) {
using std::placeholders::_1;
using std::placeholders::_2;
std::lock_guard<std::mutex> state_lock(connection_state_lock_);
::asio::error_code my_ec(original_ec);
LOG_TRACE(kRPC, << "RpcConnectionImpl::OnRecvCompleted called");
std::shared_ptr<RpcConnection> shared_this = shared_from_this();
::asio::error_code ec = asio_ec;
if(event_handlers_) {
auto event_resp = event_handlers_->call(FS_NN_READ_EVENT, cluster_name_.c_str(), 0);
event_response event_resp = event_handlers_->call(FS_NN_READ_EVENT, cluster_name_.c_str(), 0);
#ifndef NDEBUG
if (event_resp.response() == event_response::kTest_Error) {
ec = std::make_error_code(std::errc::network_down);
my_ec = std::make_error_code(std::errc::network_down);
}
#endif
}
switch (ec.value()) {
switch (my_ec.value()) {
case 0:
// No errors
break;
@ -338,8 +344,8 @@ void RpcConnectionImpl<NextLayer>::OnRecvCompleted(const ::asio::error_code &asi
// The event loop has been shut down. Ignore the error.
return;
default:
LOG_WARN(kRPC, << "Network error during RPC read: " << ec.message());
CommsError(ToStatus(ec));
LOG_WARN(kRPC, << "Network error during RPC read: " << my_ec.message());
CommsError(ToStatus(my_ec));
return;
}

1
hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/rpc/rpc_engine.h
View File

@ -229,7 +229,6 @@ class RpcConnection : public std::enable_shared_from_this<RpcConnection> {
std::shared_ptr<LibhdfsEvents> event_handlers_;
std::string cluster_name_;
// Lock for mutable parts of this class that need to be thread safe
std::mutex connection_state_lock_;

4
hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/tests/CMakeLists.txt
View File

@ -138,6 +138,10 @@ build_libhdfs_test(hdfspp_mini_dfs_smoke hdfspp_test_shim_static ${CMAKE_CURRENT
link_libhdfs_test (hdfspp_mini_dfs_smoke hdfspp_test_shim_static fs reader rpc proto common connection gmock_main ${PROTOBUF_LIBRARIES} ${OPENSSL_LIBRARIES} native_mini_dfs ${JAVA_JVM_LIBRARY} ${SASL_LIBRARIES})
add_libhdfs_test (hdfspp_mini_dfs_smoke hdfspp_test_shim_static)
build_libhdfs_test(libhdfs_mini_stress hdfspp_test_shim_static expect.c test_libhdfs_mini_stress.c ${OS_DIR}/thread.c)
link_libhdfs_test(libhdfs_mini_stress hdfspp_test_shim_static fs reader rpc proto common connection ${PROTOBUF_LIBRARIES} ${OPENSSL_LIBRARIES} native_mini_dfs ${JAVA_JVM_LIBRARY} ${SASL_LIBRARIES})
add_libhdfs_test(libhdfs_mini_stress hdfspp_test_shim_static)
build_libhdfs_test(hdfs_ext hdfspp_test_shim_static ${CMAKE_CURRENT_LIST_DIR}/hdfs_ext_test.cc)
link_libhdfs_test (hdfs_ext hdfspp_test_shim_static hdfspp_static gmock_main native_mini_dfs ${JAVA_JVM_LIBRARY} ${SASL_LIBRARIES})
add_libhdfs_test (hdfs_ext hdfspp_test_shim_static)

5
hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/tests/bad_datanode_test.cc
View File

@ -93,9 +93,10 @@ public:
std::shared_ptr<MockReader> mock_reader_ = std::make_shared<MockReader>();
protected:
std::shared_ptr<BlockReader> CreateBlockReader(const BlockReaderOptions &options,
std::shared_ptr<DataNodeConnection> dn) override
std::shared_ptr<DataNodeConnection> dn,
std::shared_ptr<hdfs::LibhdfsEvents> event_handlers) override
{
(void) options; (void) dn;
(void) options; (void) dn; (void) event_handlers;
assert(mock_reader_);
return mock_reader_;
}

48
hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/tests/hdfs_shim.c
View File

@ -24,15 +24,9 @@
#include <stdlib.h>
#include <string.h>
/* Cheat for now and use the same hdfsBuilder as libhdfs */
/* (libhdfspp doesn't have an hdfsBuilder yet). */
struct hdfsBuilder {
int forceNewInstance;
const char *nn;
tPort port;
const char *kerbTicketCachePath;
const char *userName;
struct hdfsBuilderConfOpt *opts;
struct libhdfs_hdfsBuilder * libhdfsBuilder;
struct libhdfspp_hdfsBuilder * libhdfsppBuilder;
};
/* Shim structs and functions that delegate to libhdfspp and libhdfs. */
@ -98,13 +92,13 @@ hdfsFS hdfsConnectNewInstance(const char* nn, tPort port) {
hdfsFS hdfsBuilderConnect(struct hdfsBuilder *bld) {
hdfsFS ret = calloc(1, sizeof(struct hdfs_internal));
ret->libhdfsppRep = libhdfspp_hdfsConnect(bld->nn, bld->port);
ret->libhdfsppRep = libhdfspp_hdfsBuilderConnect(bld->libhdfsppBuilder);
if (!ret->libhdfsppRep) {
free(ret);
ret = NULL;
} else {
/* Destroys bld object. */
ret->libhdfsRep = libhdfs_hdfsBuilderConnect(bld);
ret->libhdfsRep = libhdfs_hdfsBuilderConnect(bld->libhdfsBuilder);
if (!ret->libhdfsRep) {
libhdfspp_hdfsDisconnect(ret->libhdfsppRep);
free(ret);
@ -115,49 +109,61 @@ hdfsFS hdfsBuilderConnect(struct hdfsBuilder *bld) {
}
struct hdfsBuilder *hdfsNewBuilder(void) {
return libhdfs_hdfsNewBuilder();
struct hdfsBuilder * ret = calloc(1, sizeof(struct hdfsBuilder));
ret->libhdfsppBuilder = libhdfspp_hdfsNewBuilder();
ret->libhdfsBuilder = libhdfs_hdfsNewBuilder();
return ret;
}
void hdfsBuilderSetForceNewInstance(struct hdfsBuilder *bld) {
libhdfs_hdfsBuilderSetForceNewInstance(bld);
libhdfs_hdfsBuilderSetForceNewInstance(bld->libhdfsBuilder);
// libhdfspp_hdfsBuilderSetForceNewInstance(bld->libhdfsppBuilder);
}
void hdfsBuilderSetNameNode(struct hdfsBuilder *bld, const char *nn) {
libhdfs_hdfsBuilderSetNameNode(bld, nn);
libhdfs_hdfsBuilderSetNameNode(bld->libhdfsBuilder, nn);
libhdfspp_hdfsBuilderSetNameNode(bld->libhdfsppBuilder, nn);
}
void hdfsBuilderSetNameNodePort(struct hdfsBuilder *bld, tPort port) {
libhdfs_hdfsBuilderSetNameNodePort(bld, port);
libhdfs_hdfsBuilderSetNameNodePort(bld->libhdfsBuilder, port);
libhdfspp_hdfsBuilderSetNameNodePort(bld->libhdfsppBuilder, port);
}
void hdfsBuilderSetUserName(struct hdfsBuilder *bld, const char *userName) {
libhdfs_hdfsBuilderSetUserName(bld, userName);
libhdfs_hdfsBuilderSetUserName(bld->libhdfsBuilder, userName);
libhdfspp_hdfsBuilderSetUserName(bld->libhdfsppBuilder, userName);
}
void hdfsBuilderSetKerbTicketCachePath(struct hdfsBuilder *bld,
const char *kerbTicketCachePath) {
libhdfs_hdfsBuilderSetKerbTicketCachePath(bld, kerbTicketCachePath);
libhdfs_hdfsBuilderSetKerbTicketCachePath(bld->libhdfsBuilder, kerbTicketCachePath);
// libhdfspp_hdfsBuilderSetKerbTicketCachePath(bld->libhdfsppBuilder, kerbTicketCachePath);
}
void hdfsFreeBuilder(struct hdfsBuilder *bld) {
libhdfs_hdfsFreeBuilder(bld);
libhdfs_hdfsFreeBuilder(bld->libhdfsBuilder);
libhdfspp_hdfsFreeBuilder(bld->libhdfsppBuilder);
free(bld);
}
int hdfsBuilderConfSetStr(struct hdfsBuilder *bld, const char *key,
const char *val) {
return libhdfs_hdfsBuilderConfSetStr(bld, key, val);
fprintf(stderr, "hdfs_shim::hdfsBuilderConfSetStr) key=%s val=%s\n", key, val);
libhdfs_hdfsBuilderConfSetStr(bld->libhdfsBuilder, key, val);
return libhdfspp_hdfsBuilderConfSetStr(bld->libhdfsppBuilder, key, val);
}
int hdfsConfGetStr(const char *key, char **val) {
return libhdfs_hdfsConfGetStr(key, val);
return libhdfspp_hdfsConfGetStr(key, val);
}
int hdfsConfGetInt(const char *key, int32_t *val) {
return libhdfs_hdfsConfGetInt(key, val);
return libhdfspp_hdfsConfGetInt(key, val);
}
void hdfsConfStrFree(char *val) {
libhdfs_hdfsConfStrFree(val);
libhdfspp_hdfsConfStrFree(val);
}
int hdfsDisconnect(hdfsFS fs) {

1
hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/tests/libhdfspp_wrapper_defines.h
View File

@ -81,6 +81,7 @@
#define hadoopRzBufferLength libhdfspp_hadoopRzBufferLength
#define hadoopRzBufferGet libhdfspp_hadoopRzBufferGet
#define hadoopRzBufferFree libhdfspp_hadoopRzBufferFree
#define hdfsBuilder libhdfspp_hdfsBuilder
#define hdfs_internal libhdfspp_hdfs_internal
#define hdfsFS libhdfspp_hdfsFS
#define hdfsFile_internal libhdfspp_hdfsFile_internal

2
hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/tests/rpc_engine_test.cc
View File

@ -386,7 +386,7 @@ TEST(RpcEngineTest, TestEventCallbacks)
});
io_service.run();
ASSERT_TRUE(complete);
ASSERT_EQ(7, callbacks.size());
ASSERT_EQ(8, callbacks.size());
ASSERT_EQ(FS_NN_CONNECT_EVENT, callbacks[0]); // error
ASSERT_EQ(FS_NN_CONNECT_EVENT, callbacks[1]); // reconnect
ASSERT_EQ(FS_NN_READ_EVENT, callbacks[2]); // makes an error