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

HDFS-10543: libhdfs++: hdfsRead stops at block boundary. Contributed by Xiaowei Zhu.

Browse Source
This commit is contained in:
James 2016-06-24 12:15:18 -04:00 committed by James Clampffer
parent d4c3cfbf47
commit a0c7aee3b6
1 changed files with 23 additions and 15 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -78,26 +78,34 @@ Status FileHandleImpl::PositionRead(void *buf, size_t *nbyte, off_t offset) {
LOG_TRACE(kFileHandle, << "FileHandleImpl::[sync]PositionRead(" LOG_TRACE(kFileHandle, << "FileHandleImpl::[sync]PositionRead("
<< FMT_THIS_ADDR << ", buf=" << buf << FMT_THIS_ADDR << ", buf=" << buf
<< ", nbyte=" << *nbyte << ") called"); << ", nbyte=" << *nbyte << ") called");
size_t totalBytesRead = 0;
Status stat = Status::OK();
while (*nbyte != 0 && offset < (off_t)(file_info_->file_length_)) {
auto callstate = std::make_shared<std::promise<std::tuple<Status, size_t>>>();
std::future<std::tuple<Status, size_t>> future(callstate->get_future());
auto callstate = std::make_shared<std::promise<std::tuple<Status, size_t>>>(); /* wrap async call with promise/future to make it blocking */
std::future<std::tuple<Status, size_t>> future(callstate->get_future()); auto callback = [callstate](const Status &s, size_t bytes) {
callstate->set_value(std::make_tuple(s,bytes));
};
/* wrap async call with promise/future to make it blocking */ PositionRead(buf, *nbyte, offset, callback);
auto callback = [callstate](const Status &s, size_t bytes) {
callstate->set_value(std::make_tuple(s,bytes));
};
PositionRead(buf, *nbyte, offset, callback); /* wait for async to finish */
auto returnstate = future.get();
stat = std::get<0>(returnstate);
/* wait for async to finish */ if (!stat.ok()) {
auto returnstate = future.get(); return stat;
auto stat = std::get<0>(returnstate); }
if (!stat.ok()) { size_t bytesRead = std::get<1>(returnstate);
return stat; *nbyte = *nbyte - bytesRead;
totalBytesRead += bytesRead;
offset += bytesRead;
} }
/* Update the bytes read for return */
*nbyte = std::get<1>(returnstate); *nbyte = totalBytesRead;
return stat; return stat;
} }
@ -110,8 +118,8 @@ Status FileHandleImpl::Read(void *buf, size_t *nbyte) {
if(!stat.ok()) { if(!stat.ok()) {
return stat; return stat;
} }
offset_ += *nbyte; offset_ += *nbyte;
return Status::OK(); return Status::OK();
} }