You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@impala.apache.org by jo...@apache.org on 2018/12/07 04:30:17 UTC
[2/4] impala git commit: IMPALA-7738: Implement timeouts for HDFS
open calls
IMPALA-7738: Implement timeouts for HDFS open calls
This is part 1 of a push to add timeouts for all HDFS operations.
It adds timeouts for opening an HDFS file handle.
It introduces a new SynchronousThreadPool, which executes
an operation in a thread pool and waits up to a specified
timeout for the operation to complete. This type of thread
pool can accept any subclass of SynchronousWorkItem, and
a single thread pool can process different types of work
items. It is tested by a new test case in thread-pool-test.
This also introduces a new HdfsMonitor which implements
timeouts for HDFS operations, currently limited to
hdfsOpenFile(). This is implemented using a SynchronousThreadPool.
The timeout for hdfs operations is specified by
hdfs_operation_timeout_sec, which defaults to 5 minutes.
Testing:
1. Added a test to thread-pool-test for the new
SynchronousThreadPool.
2. Core tests
3. Added a custom cluster test that does "kill -STOP"
for the NameNode and verifies that a subsequent
hdfsOpenFile operation times out.
Change-Id: Ia14403ca5f3f19c6d5f61b9ab2306b0ad3267454
Reviewed-on: http://gerrit.cloudera.org:8080/11874
Reviewed-by: Joe McDonnell <jo...@cloudera.com>
Tested-by: Impala Public Jenkins <im...@cloudera.com>
Project: http://git-wip-us.apache.org/repos/asf/impala/repo
Commit: http://git-wip-us.apache.org/repos/asf/impala/commit/938be0e8
Tree: http://git-wip-us.apache.org/repos/asf/impala/tree/938be0e8
Diff: http://git-wip-us.apache.org/repos/asf/impala/diff/938be0e8
Branch: refs/heads/master
Commit: 938be0e840c84263a2b47fb89e655d998363b819
Parents: 2572a78
Author: Joe McDonnell <jo...@cloudera.com>
Authored: Fri Nov 2 18:03:34 2018 -0700
Committer: Joe McDonnell <jo...@cloudera.com>
Committed: Fri Dec 7 03:50:11 2018 +0000
----------------------------------------------------------------------
be/src/common/status.h | 6 ++
be/src/exec/base-sequence-scanner.cc | 3 +-
be/src/runtime/io/CMakeLists.txt | 1 +
be/src/runtime/io/disk-io-mgr.cc | 48 +++++-----
be/src/runtime/io/disk-io-mgr.h | 31 +++---
be/src/runtime/io/handle-cache.h | 32 ++++---
be/src/runtime/io/handle-cache.inline.h | 45 +++++----
be/src/runtime/io/hdfs-file-reader.cc | 25 ++---
be/src/runtime/io/hdfs-file-reader.h | 2 +-
be/src/runtime/io/hdfs-monitored-ops.cc | 99 +++++++++++++++++++
be/src/runtime/io/hdfs-monitored-ops.h | 62 ++++++++++++
be/src/runtime/runtime-state.cc | 4 +-
be/src/util/thread-pool-test.cc | 87 +++++++++++++++++
be/src/util/thread-pool.h | 127 +++++++++++++++++++++++++
common/thrift/generate_error_codes.py | 6 ++
tests/custom_cluster/test_hdfs_timeout.py | 90 ++++++++++++++++++
16 files changed, 582 insertions(+), 86 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/impala/blob/938be0e8/be/src/common/status.h
----------------------------------------------------------------------
diff --git a/be/src/common/status.h b/be/src/common/status.h
index d60efa7..6c14ad4 100644
--- a/be/src/common/status.h
+++ b/be/src/common/status.h
@@ -227,6 +227,12 @@ class NODISCARD Status {
&& msg_->error() == TErrorCode::DISK_IO_ERROR;
}
+ bool IsThreadPoolError() const {
+ return msg_ != NULL
+ && (msg_->error() == TErrorCode::THREAD_POOL_SUBMIT_FAILED ||
+ msg_->error() == TErrorCode::THREAD_POOL_TASK_TIMED_OUT);
+ }
+
/// Returns the error message associated with a non-successful status.
const ErrorMsg& msg() const {
DCHECK(msg_ != NULL);
http://git-wip-us.apache.org/repos/asf/impala/blob/938be0e8/be/src/exec/base-sequence-scanner.cc
----------------------------------------------------------------------
diff --git a/be/src/exec/base-sequence-scanner.cc b/be/src/exec/base-sequence-scanner.cc
index 6757ec3..b06602a 100644
--- a/be/src/exec/base-sequence-scanner.cc
+++ b/be/src/exec/base-sequence-scanner.cc
@@ -185,7 +185,8 @@ Status BaseSequenceScanner::GetNextInternal(RowBatch* row_batch) {
if (!status.IsCancelled() &&
!status.IsMemLimitExceeded() &&
!status.IsInternalError() &&
- !status.IsDiskIoError()) {
+ !status.IsDiskIoError() &&
+ !status.IsThreadPoolError()) {
state_->LogError(ErrorMsg(TErrorCode::SEQUENCE_SCANNER_PARSE_ERROR,
stream_->filename(), stream_->file_offset(),
(stream_->eof() ? "(EOF)" : "")));
http://git-wip-us.apache.org/repos/asf/impala/blob/938be0e8/be/src/runtime/io/CMakeLists.txt
----------------------------------------------------------------------
diff --git a/be/src/runtime/io/CMakeLists.txt b/be/src/runtime/io/CMakeLists.txt
index 675bd6e..2215678 100644
--- a/be/src/runtime/io/CMakeLists.txt
+++ b/be/src/runtime/io/CMakeLists.txt
@@ -31,6 +31,7 @@ add_library(Io
scan-range.cc
hdfs-file-reader.cc
local-file-reader.cc
+ hdfs-monitored-ops.cc
)
add_dependencies(Io gen-deps)
http://git-wip-us.apache.org/repos/asf/impala/blob/938be0e8/be/src/runtime/io/disk-io-mgr.cc
----------------------------------------------------------------------
diff --git a/be/src/runtime/io/disk-io-mgr.cc b/be/src/runtime/io/disk-io-mgr.cc
index fb0c876..ce6fd45 100644
--- a/be/src/runtime/io/disk-io-mgr.cc
+++ b/be/src/runtime/io/disk-io-mgr.cc
@@ -183,7 +183,7 @@ DiskIoMgr::DiskIoMgr() :
file_handle_cache_(min(FLAGS_max_cached_file_handles,
FileSystemUtil::MaxNumFileHandles()),
FLAGS_num_file_handle_cache_partitions,
- FLAGS_unused_file_handle_timeout_sec) {
+ FLAGS_unused_file_handle_timeout_sec, &hdfs_monitor_) {
DCHECK_LE(READ_SIZE_MIN_VALUE, FLAGS_read_size);
int num_local_disks = DiskInfo::num_disks();
if (FLAGS_num_disks < 0 || FLAGS_num_disks > DiskInfo::num_disks()) {
@@ -207,7 +207,7 @@ DiskIoMgr::DiskIoMgr(int num_local_disks, int threads_per_rotational_disk,
file_handle_cache_(min(FLAGS_max_cached_file_handles,
FileSystemUtil::MaxNumFileHandles()),
FLAGS_num_file_handle_cache_partitions,
- FLAGS_unused_file_handle_timeout_sec) {
+ FLAGS_unused_file_handle_timeout_sec, &hdfs_monitor_) {
if (num_local_disks == 0) num_local_disks = DiskInfo::num_disks();
disk_queues_.resize(num_local_disks + REMOTE_NUM_DISKS);
CheckSseSupport();
@@ -259,6 +259,9 @@ Status DiskIoMgr::Init() {
disk_thread_group_.AddThread(move(t));
}
}
+ // The file handle cache depends on the HDFS monitor, so initialize it first.
+ // Use the same number of threads for the HDFS monitor as there are Disk IO threads.
+ RETURN_IF_ERROR(hdfs_monitor_.Init(disk_thread_group_.Size()));
RETURN_IF_ERROR(file_handle_cache_.Init());
cached_read_options_ = hadoopRzOptionsAlloc();
@@ -478,36 +481,35 @@ int DiskIoMgr::AssignQueue(const char* file, int disk_id, bool expected_local) {
return disk_id % num_local_disks();
}
-ExclusiveHdfsFileHandle* DiskIoMgr::GetExclusiveHdfsFileHandle(const hdfsFS& fs,
- std::string* fname, int64_t mtime, RequestContext *reader) {
+Status DiskIoMgr::GetExclusiveHdfsFileHandle(const hdfsFS& fs,
+ std::string* fname, int64_t mtime, RequestContext *reader,
+ unique_ptr<ExclusiveHdfsFileHandle>& fid_out) {
SCOPED_TIMER(reader->open_file_timer_);
- ExclusiveHdfsFileHandle* fid = new ExclusiveHdfsFileHandle(fs, fname->data(), mtime);
- if (!fid->ok()) {
- VLOG_FILE << "Opening the file " << fname << " failed.";
- delete fid;
- return nullptr;
- }
+ unique_ptr<ExclusiveHdfsFileHandle> fid;
+ fid.reset(new ExclusiveHdfsFileHandle(fs, fname, mtime));
+ RETURN_IF_ERROR(fid->Init(&hdfs_monitor_));
+ fid_out.swap(fid);
ImpaladMetrics::IO_MGR_NUM_FILE_HANDLES_OUTSTANDING->Increment(1L);
// Every exclusive file handle is considered a cache miss
ImpaladMetrics::IO_MGR_CACHED_FILE_HANDLES_HIT_RATIO->Update(0L);
ImpaladMetrics::IO_MGR_CACHED_FILE_HANDLES_MISS_COUNT->Increment(1L);
reader->cached_file_handles_miss_count_.Add(1L);
- return fid;
+ return Status::OK();
}
-void DiskIoMgr::ReleaseExclusiveHdfsFileHandle(ExclusiveHdfsFileHandle* fid) {
+void DiskIoMgr::ReleaseExclusiveHdfsFileHandle(unique_ptr<ExclusiveHdfsFileHandle> fid) {
DCHECK(fid != nullptr);
ImpaladMetrics::IO_MGR_NUM_FILE_HANDLES_OUTSTANDING->Increment(-1L);
- delete fid;
+ fid.reset();
}
-CachedHdfsFileHandle* DiskIoMgr::GetCachedHdfsFileHandle(const hdfsFS& fs,
- std::string* fname, int64_t mtime, RequestContext *reader) {
+Status DiskIoMgr::GetCachedHdfsFileHandle(const hdfsFS& fs,
+ std::string* fname, int64_t mtime, RequestContext *reader,
+ CachedHdfsFileHandle** handle_out) {
bool cache_hit;
SCOPED_TIMER(reader->open_file_timer_);
- CachedHdfsFileHandle* fh = file_handle_cache_.GetFileHandle(fs, fname, mtime, false,
- &cache_hit);
- if (fh == nullptr) return nullptr;
+ RETURN_IF_ERROR(file_handle_cache_.GetFileHandle(fs, fname, mtime, false, handle_out,
+ &cache_hit));
ImpaladMetrics::IO_MGR_NUM_FILE_HANDLES_OUTSTANDING->Increment(1L);
if (cache_hit) {
ImpaladMetrics::IO_MGR_CACHED_FILE_HANDLES_HIT_RATIO->Update(1L);
@@ -518,7 +520,7 @@ CachedHdfsFileHandle* DiskIoMgr::GetCachedHdfsFileHandle(const hdfsFS& fs,
ImpaladMetrics::IO_MGR_CACHED_FILE_HANDLES_MISS_COUNT->Increment(1L);
reader->cached_file_handles_miss_count_.Add(1L);
}
- return fh;
+ return Status::OK();
}
void DiskIoMgr::ReleaseCachedHdfsFileHandle(std::string* fname,
@@ -534,12 +536,12 @@ Status DiskIoMgr::ReopenCachedHdfsFileHandle(const hdfsFS& fs, std::string* fnam
ImpaladMetrics::IO_MGR_CACHED_FILE_HANDLES_REOPENED->Increment(1L);
file_handle_cache_.ReleaseFileHandle(fname, *fid, true);
// The old handle has been destroyed, so *fid must be overwritten before returning.
- *fid = file_handle_cache_.GetFileHandle(fs, fname, mtime, true,
+ *fid = nullptr;
+ Status status = file_handle_cache_.GetFileHandle(fs, fname, mtime, true, fid,
&cache_hit);
- if (*fid == nullptr) {
+ if (!status.ok()) {
ImpaladMetrics::IO_MGR_NUM_FILE_HANDLES_OUTSTANDING->Increment(-1L);
- return Status(TErrorCode::DISK_IO_ERROR,
- GetHdfsErrorMsg("Failed to open HDFS file ", fname->data()));
+ return status;
}
DCHECK(!cache_hit);
return Status::OK();
http://git-wip-us.apache.org/repos/asf/impala/blob/938be0e8/be/src/runtime/io/disk-io-mgr.h
----------------------------------------------------------------------
diff --git a/be/src/runtime/io/disk-io-mgr.h b/be/src/runtime/io/disk-io-mgr.h
index d36906d..ac93c75 100644
--- a/be/src/runtime/io/disk-io-mgr.h
+++ b/be/src/runtime/io/disk-io-mgr.h
@@ -27,6 +27,7 @@
#include "common/status.h"
#include "runtime/bufferpool/buffer-pool.h"
#include "runtime/io/handle-cache.h"
+#include "runtime/io/hdfs-monitored-ops.h"
#include "runtime/io/local-file-system.h"
#include "runtime/io/request-ranges.h"
#include "util/aligned-new.h"
@@ -300,21 +301,24 @@ class DiskIoMgr : public CacheLineAligned {
bool Validate() const;
/// Given a FS handle, name and last modified time of the file, construct a new
- /// ExclusiveHdfsFileHandle. This records the time spent opening the handle in
- /// 'reader' and counts this as a cache miss. In the case of an error, returns nullptr.
- ExclusiveHdfsFileHandle* GetExclusiveHdfsFileHandle(const hdfsFS& fs,
- std::string* fname, int64_t mtime, RequestContext* reader);
+ /// ExclusiveHdfsFileHandle and return it via 'fid'. This records the time spent
+ /// opening the handle in 'reader' and counts this as a cache miss. In case of an
+ /// error, returns status and 'fid' is untouched.
+ Status GetExclusiveHdfsFileHandle(const hdfsFS& fs,
+ std::string* fname, int64_t mtime, RequestContext* reader,
+ std::unique_ptr<ExclusiveHdfsFileHandle>& fid) WARN_UNUSED_RESULT;
/// Releases an exclusive file handle, destroying it
- void ReleaseExclusiveHdfsFileHandle(ExclusiveHdfsFileHandle* fid);
+ void ReleaseExclusiveHdfsFileHandle(std::unique_ptr<ExclusiveHdfsFileHandle> fid);
/// Given a FS handle, name and last modified time of the file, gets a
- /// CachedHdfsFileHandle from the file handle cache. Records the time spent
- /// opening the handle in 'reader'. On success, records statistics
- /// about whether this was a cache hit or miss in the 'reader' as well as at the
- /// system level. In case of an error returns nullptr.
- CachedHdfsFileHandle* GetCachedHdfsFileHandle(const hdfsFS& fs,
- std::string* fname, int64_t mtime, RequestContext* reader);
+ /// CachedHdfsFileHandle from the file handle cache and returns it via 'fid'.
+ /// Records the time spent opening the handle in 'reader'. On success, records
+ /// statistics about whether this was a cache hit or miss in the 'reader' as well as
+ /// at the system level. In case of an error, returns status and 'fid' is untouched.
+ Status GetCachedHdfsFileHandle(const hdfsFS& fs,
+ std::string* fname, int64_t mtime, RequestContext* reader,
+ CachedHdfsFileHandle** fid) WARN_UNUSED_RESULT;
/// Releases a file handle back to the file handle cache when it is no longer in use.
void ReleaseCachedHdfsFileHandle(std::string* fname, CachedHdfsFileHandle* fid);
@@ -323,7 +327,7 @@ class DiskIoMgr : public CacheLineAligned {
/// file handle from the cache. Records the time spent reopening the handle
/// in 'reader'. Returns an error if the file could not be reopened.
Status ReopenCachedHdfsFileHandle(const hdfsFS& fs, std::string* fname, int64_t mtime,
- RequestContext* reader, CachedHdfsFileHandle** fid);
+ RequestContext* reader, CachedHdfsFileHandle** fid) WARN_UNUSED_RESULT;
// Function to change the underlying LocalFileSystem object used for disk I/O.
// DiskIoMgr will also take responsibility of the received LocalFileSystem pointer.
@@ -408,6 +412,9 @@ class DiskIoMgr : public CacheLineAligned {
/// round-robin assignment for that case.
static AtomicInt32 next_disk_id_;
+ /// Thread pool used to implement timeouts for HDFS operations.
+ HdfsMonitor hdfs_monitor_;
+
// Caching structure that maps file names to cached file handles. The cache has an upper
// limit of entries defined by FLAGS_max_cached_file_handles. Evicted cached file
// handles are closed.
http://git-wip-us.apache.org/repos/asf/impala/blob/938be0e8/be/src/runtime/io/handle-cache.h
----------------------------------------------------------------------
diff --git a/be/src/runtime/io/handle-cache.h b/be/src/runtime/io/handle-cache.h
index 232b630..d4cff37 100644
--- a/be/src/runtime/io/handle-cache.h
+++ b/be/src/runtime/io/handle-cache.h
@@ -35,6 +35,8 @@
namespace impala {
namespace io {
+class HdfsMonitor;
+
/// This abstract class is a small wrapper around the hdfsFile handle and the file system
/// instance which is needed to close the file handle. The handle incorporates
/// the last modified time of the file when it was opened. This is used to distinguish
@@ -47,17 +49,21 @@ class HdfsFileHandle {
/// Destructor will close the file handle
~HdfsFileHandle();
+ /// Init opens the file handle
+ Status Init(HdfsMonitor* monitor);
+
hdfsFile file() const { return hdfs_file_; }
int64_t mtime() const { return mtime_; }
- bool ok() const { return hdfs_file_ != nullptr; }
protected:
- /// Constructor will open the file
- HdfsFileHandle(const hdfsFS& fs, const char* fname, int64_t mtime);
+ HdfsFileHandle(const hdfsFS& fs, const std::string* fname, int64_t mtime)
+ : fs_(fs), fname_(fname), mtime_(mtime) {}
private:
hdfsFS fs_;
- hdfsFile hdfs_file_;
+ // fname_ has a limited lifetime. It is only valid from construction until Init().
+ const std::string* fname_;
+ hdfsFile hdfs_file_ = nullptr;
int64_t mtime_;
};
@@ -65,7 +71,7 @@ class HdfsFileHandle {
/// other purpose.
class CachedHdfsFileHandle : public HdfsFileHandle {
public:
- CachedHdfsFileHandle(const hdfsFS& fs, const char* fname, int64_t mtime);
+ CachedHdfsFileHandle(const hdfsFS& fs, const std::string* fname, int64_t mtime);
~CachedHdfsFileHandle();
};
@@ -73,7 +79,7 @@ class CachedHdfsFileHandle : public HdfsFileHandle {
/// is not appropriate.
class ExclusiveHdfsFileHandle : public HdfsFileHandle {
public:
- ExclusiveHdfsFileHandle(const hdfsFS& fs, const char* fname, int64_t mtime)
+ ExclusiveHdfsFileHandle(const hdfsFS& fs, const std::string* fname, int64_t mtime)
: HdfsFileHandle(fs, fname, mtime) {}
};
@@ -110,7 +116,7 @@ class FileHandleCache {
/// up. The cache will age out any file handle that is unused for
/// `unused_handle_timeout_secs` seconds. Age out is disabled if this is set to zero.
FileHandleCache(size_t capacity, size_t num_partitions,
- uint64_t unused_handle_timeout_secs);
+ uint64_t unused_handle_timeout_secs, HdfsMonitor* hdfs_monitor);
/// Destructor is only called for backend tests
~FileHandleCache();
@@ -132,8 +138,9 @@ class FileHandleCache {
///
/// This obtains exclusive control over the returned file handle. It must be paired
/// with a call to ReleaseFileHandle to release exclusive control.
- CachedHdfsFileHandle* GetFileHandle(const hdfsFS& fs, std::string* fname,
- int64_t mtime, bool require_new_handle, bool* cache_hit);
+ Status GetFileHandle(const hdfsFS& fs, std::string* fname,
+ int64_t mtime, bool require_new_handle, CachedHdfsFileHandle** handle_out,
+ bool* cache_hit) WARN_UNUSED_RESULT;
/// Release the exclusive hold on the specified file handle (which was obtained
/// by calling GetFileHandle). The cache may evict a file handle if the cache is
@@ -154,8 +161,8 @@ class FileHandleCache {
typedef std::list<LruListEntry> LruListType;
struct FileHandleEntry {
- FileHandleEntry(CachedHdfsFileHandle* fh_in, LruListType& lru_list)
- : fh(fh_in), lru_entry(lru_list.end()) {}
+ FileHandleEntry(std::unique_ptr<CachedHdfsFileHandle> fh_in, LruListType& lru_list)
+ : fh(std::move(fh_in)), lru_entry(lru_list.end()) {}
std::unique_ptr<CachedHdfsFileHandle> fh;
/// in_use is true for a file handle checked out via GetFileHandle() that has not
@@ -210,6 +217,9 @@ class FileHandleCache {
/// the shut_down_promise_ is set.
std::unique_ptr<Thread> eviction_thread_;
Promise<bool> shut_down_promise_;
+
+ /// Thread pool used to implement timeouts for HDFS operations
+ HdfsMonitor* hdfs_monitor_;
};
}
}
http://git-wip-us.apache.org/repos/asf/impala/blob/938be0e8/be/src/runtime/io/handle-cache.inline.h
----------------------------------------------------------------------
diff --git a/be/src/runtime/io/handle-cache.inline.h b/be/src/runtime/io/handle-cache.inline.h
index 99b106c..8db7730 100644
--- a/be/src/runtime/io/handle-cache.inline.h
+++ b/be/src/runtime/io/handle-cache.inline.h
@@ -18,6 +18,7 @@
#include <tuple>
#include "runtime/io/handle-cache.h"
+#include "runtime/io/hdfs-monitored-ops.h"
#include "util/hash-util.h"
#include "util/time.h"
@@ -27,22 +28,25 @@
namespace impala {
namespace io {
-HdfsFileHandle::HdfsFileHandle(const hdfsFS& fs, const char* fname,
- int64_t mtime)
- : fs_(fs), hdfs_file_(hdfsOpenFile(fs, fname, O_RDONLY, 0, 0, 0)), mtime_(mtime) {
- VLOG_FILE << "hdfsOpenFile() file=" << fname << " fid=" << hdfs_file_;
-}
-
HdfsFileHandle::~HdfsFileHandle() {
if (hdfs_file_ != nullptr && fs_ != nullptr) {
VLOG_FILE << "hdfsCloseFile() fid=" << hdfs_file_;
- hdfsCloseFile(fs_, hdfs_file_);
+ hdfsCloseFile(fs_, hdfs_file_); // TODO: check return code
}
fs_ = nullptr;
+ fname_ = nullptr;
hdfs_file_ = nullptr;
}
-CachedHdfsFileHandle::CachedHdfsFileHandle(const hdfsFS& fs, const char* fname,
+Status HdfsFileHandle::Init(HdfsMonitor* monitor) {
+ Status status = monitor->OpenHdfsFileWithTimeout(fs_, fname_, O_RDONLY, 0,
+ &hdfs_file_);
+ // fname_ is no longer needed, null it out
+ fname_ = nullptr;
+ return status;
+}
+
+CachedHdfsFileHandle::CachedHdfsFileHandle(const hdfsFS& fs, const string* fname,
int64_t mtime)
: HdfsFileHandle(fs, fname, mtime) {
ImpaladMetrics::IO_MGR_NUM_CACHED_FILE_HANDLES->Increment(1L);
@@ -53,9 +57,10 @@ CachedHdfsFileHandle::~CachedHdfsFileHandle() {
}
FileHandleCache::FileHandleCache(size_t capacity,
- size_t num_partitions, uint64_t unused_handle_timeout_secs)
+ size_t num_partitions, uint64_t unused_handle_timeout_secs, HdfsMonitor* hdfs_monitor)
: cache_partitions_(num_partitions),
- unused_handle_timeout_secs_(unused_handle_timeout_secs) {
+ unused_handle_timeout_secs_(unused_handle_timeout_secs),
+ hdfs_monitor_(hdfs_monitor) {
DCHECK_GT(num_partitions, 0);
size_t remainder = capacity % num_partitions;
size_t base_capacity = capacity / num_partitions;
@@ -80,9 +85,9 @@ Status FileHandleCache::Init() {
&FileHandleCache::EvictHandlesLoop, this, &eviction_thread_);
}
-CachedHdfsFileHandle* FileHandleCache::GetFileHandle(
+Status FileHandleCache::GetFileHandle(
const hdfsFS& fs, std::string* fname, int64_t mtime, bool require_new_handle,
- bool* cache_hit) {
+ CachedHdfsFileHandle** handle_out, bool* cache_hit) {
// Hash the key and get appropriate partition
int index = HashUtil::Hash(fname->data(), fname->size(), 0) % cache_partitions_.size();
FileHandleCachePartition& p = cache_partitions_[index];
@@ -112,7 +117,8 @@ CachedHdfsFileHandle* FileHandleCache::GetFileHandle(
elem->lru_entry = p.lru_list.end();
*cache_hit = true;
elem->in_use = true;
- return elem->fh.get();
+ *handle_out = elem->fh.get();
+ return Status::OK();
}
++range.first;
}
@@ -123,11 +129,9 @@ CachedHdfsFileHandle* FileHandleCache::GetFileHandle(
// the file handle without holding the lock to reduce contention.
*cache_hit = false;
// Create a new file handle
- CachedHdfsFileHandle* new_fh = new CachedHdfsFileHandle(fs, fname->data(), mtime);
- if (!new_fh->ok()) {
- delete new_fh;
- return nullptr;
- }
+ std::unique_ptr<CachedHdfsFileHandle> new_fh;
+ new_fh.reset(new CachedHdfsFileHandle(fs, fname, mtime));
+ RETURN_IF_ERROR(new_fh->Init(hdfs_monitor_));
// Get the lock and create/move the new entry into the map
// This entry is new and will be immediately used. Place it as the first entry
@@ -137,7 +141,7 @@ CachedHdfsFileHandle* FileHandleCache::GetFileHandle(
boost::lock_guard<SpinLock> g(p.lock);
pair<typename MapType::iterator, typename MapType::iterator> range =
p.cache.equal_range(*fname);
- FileHandleEntry entry(new_fh, p.lru_list);
+ FileHandleEntry entry(std::move(new_fh), p.lru_list);
typename MapType::iterator new_it = p.cache.emplace_hint(range.first,
*fname, std::move(entry));
++p.size;
@@ -145,7 +149,8 @@ CachedHdfsFileHandle* FileHandleCache::GetFileHandle(
FileHandleEntry* new_elem = &new_it->second;
DCHECK(!new_elem->in_use);
new_elem->in_use = true;
- return new_elem->fh.get();
+ *handle_out = new_elem->fh.get();
+ return Status::OK();
}
void FileHandleCache::ReleaseFileHandle(std::string* fname,
http://git-wip-us.apache.org/repos/asf/impala/blob/938be0e8/be/src/runtime/io/hdfs-file-reader.cc
----------------------------------------------------------------------
diff --git a/be/src/runtime/io/hdfs-file-reader.cc b/be/src/runtime/io/hdfs-file-reader.cc
index c6a6cec..6fb0508 100644
--- a/be/src/runtime/io/hdfs-file-reader.cc
+++ b/be/src/runtime/io/hdfs-file-reader.cc
@@ -55,16 +55,12 @@ Status HdfsFileReader::Open(bool use_file_handle_cache) {
if (use_file_handle_cache && expected_local_) return Status::OK();
auto io_mgr = scan_range_->io_mgr_;
// Get a new exclusive file handle.
- exclusive_hdfs_fh_ = io_mgr->GetExclusiveHdfsFileHandle(hdfs_fs_,
- scan_range_->file_string(), scan_range_->mtime(), scan_range_->reader_);
- if (exclusive_hdfs_fh_ == nullptr) {
- return Status(TErrorCode::DISK_IO_ERROR,
- GetHdfsErrorMsg("Failed to open HDFS file ", *scan_range_->file_string()));
- }
+ RETURN_IF_ERROR(io_mgr->GetExclusiveHdfsFileHandle(hdfs_fs_,
+ scan_range_->file_string(), scan_range_->mtime(), scan_range_->reader_,
+ exclusive_hdfs_fh_));
if (hdfsSeek(hdfs_fs_, exclusive_hdfs_fh_->file(), scan_range_->offset_) != 0) {
// Destroy the file handle
- io_mgr->ReleaseExclusiveHdfsFileHandle(exclusive_hdfs_fh_);
- exclusive_hdfs_fh_ = nullptr;
+ io_mgr->ReleaseExclusiveHdfsFileHandle(std::move(exclusive_hdfs_fh_));
return Status(TErrorCode::DISK_IO_ERROR,
Substitute("Error seeking to $0 in file: $1 $2", scan_range_->offset(),
*scan_range_->file_string(), GetHdfsErrorMsg("")));
@@ -99,13 +95,9 @@ Status HdfsFileReader::ReadFromPos(int64_t file_offset, uint8_t* buffer,
if (exclusive_hdfs_fh_ != nullptr) {
hdfs_file = exclusive_hdfs_fh_->file();
} else {
- borrowed_hdfs_fh = io_mgr->GetCachedHdfsFileHandle(hdfs_fs_,
+ RETURN_IF_ERROR(io_mgr->GetCachedHdfsFileHandle(hdfs_fs_,
scan_range_->file_string(),
- scan_range_->mtime(), request_context);
- if (borrowed_hdfs_fh == nullptr) {
- return Status(TErrorCode::DISK_IO_ERROR,
- GetHdfsErrorMsg("Failed to open HDFS file ", *scan_range_->file_string()));
- }
+ scan_range_->mtime(), request_context, &borrowed_hdfs_fh));
hdfs_file = borrowed_hdfs_fh->file();
}
@@ -228,8 +220,7 @@ void HdfsFileReader::Close() {
}
// Destroy the file handle.
- scan_range_->io_mgr_->ReleaseExclusiveHdfsFileHandle(exclusive_hdfs_fh_);
- exclusive_hdfs_fh_ = nullptr;
+ scan_range_->io_mgr_->ReleaseExclusiveHdfsFileHandle(std::move(exclusive_hdfs_fh_));
ImpaladMetrics::IO_MGR_NUM_OPEN_FILES->Increment(-1L);
}
@@ -287,7 +278,7 @@ void HdfsFileReader::ResetState() {
string HdfsFileReader::DebugString() const {
return FileReader::DebugString() + Substitute(
" exclusive_hdfs_fh=$0 num_remote_bytes=$1",
- exclusive_hdfs_fh_, num_remote_bytes_);
+ exclusive_hdfs_fh_.get(), num_remote_bytes_);
}
}
http://git-wip-us.apache.org/repos/asf/impala/blob/938be0e8/be/src/runtime/io/hdfs-file-reader.h
----------------------------------------------------------------------
diff --git a/be/src/runtime/io/hdfs-file-reader.h b/be/src/runtime/io/hdfs-file-reader.h
index ba95095..f8dbf93 100644
--- a/be/src/runtime/io/hdfs-file-reader.h
+++ b/be/src/runtime/io/hdfs-file-reader.h
@@ -57,7 +57,7 @@ private:
/// 3. The hdfs file is expected to be remote (expected_local_ == false)
/// In each case, the scan range gets a new ExclusiveHdfsFileHandle at Open(),
/// owns it exclusively, and destroys it in Close().
- ExclusiveHdfsFileHandle* exclusive_hdfs_fh_ = nullptr;
+ std::unique_ptr<ExclusiveHdfsFileHandle> exclusive_hdfs_fh_;
/// If true, we expect the reads to be a local read. Note that if this is false,
/// it does not necessarily mean we expect the read to be remote, and that we never
http://git-wip-us.apache.org/repos/asf/impala/blob/938be0e8/be/src/runtime/io/hdfs-monitored-ops.cc
----------------------------------------------------------------------
diff --git a/be/src/runtime/io/hdfs-monitored-ops.cc b/be/src/runtime/io/hdfs-monitored-ops.cc
new file mode 100644
index 0000000..3ea5c86
--- /dev/null
+++ b/be/src/runtime/io/hdfs-monitored-ops.cc
@@ -0,0 +1,99 @@
+// 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 "gutil/strings/substitute.h"
+
+#include "common/names.h"
+#include "common/status.h"
+#include "runtime/io/hdfs-monitored-ops.h"
+#include "util/hdfs-util.h"
+#include "util/time.h"
+
+namespace impala {
+
+namespace io {
+
+// Timeout for HDFS operations. This defaults to 5 minutes
+DEFINE_uint64(hdfs_operation_timeout_sec, 300, "Maximum time, in seconds, that an "
+ "HDFS operation should wait before timing out and failing.");
+
+Status HdfsMonitor::Init(int32_t num_threads) {
+ // The thread pool sets its queue size to be equal to the number of threads.
+ // TODO: is there a better queue size?
+ hdfs_worker_pool_.reset(new SynchronousThreadPool("hdfs monitor",
+ "hdfs_monitor_", num_threads, num_threads));
+ return hdfs_worker_pool_->Init();
+}
+
+class OpenHdfsFileOp : public SynchronousWorkItem {
+ public:
+ // To guarantee the appropriate lifetime, the 'fname' argument is copied. The 'fs'
+ // is safe, because it is kept open until the process terminates (see hdfs-fs-cache.h).
+ OpenHdfsFileOp(const hdfsFS& fs, const std::string* fname, int flags,
+ uint64_t blocksize)
+ : fs_(fs), fname_(*fname), flags_(flags), blocksize_(blocksize) {}
+
+ // Run hdfsOpenFile() and return error status if hdfsOpenFile() fails. If hdfsOpenFile()
+ // succeeds, it is valid to call GetFileHandle().
+ virtual Status Execute() override;
+
+ virtual std::string GetDescription() override;
+
+ // If the handle is successfully opened, GetHandle() will return the handle.
+ hdfsFile GetFileHandle() { return hdfs_file_; }
+
+ private:
+ const hdfsFS& fs_;
+ const std::string fname_;
+ int flags_;
+ uint64_t blocksize_;
+ hdfsFile hdfs_file_;
+};
+
+Status OpenHdfsFileOp::Execute() {
+ hdfs_file_ = hdfsOpenFile(fs_, fname_.data(), flags_, 0, 0, blocksize_);
+ if (hdfs_file_ == nullptr) {
+ // GetHdfsErrorMsg references thread local state to get error information, so it
+ // must happen in the same thread as the hdfsOpenFile().
+ return Status(TErrorCode::DISK_IO_ERROR,
+ GetHdfsErrorMsg("Failed to open HDFS file ", fname_));
+ }
+ return Status::OK();
+}
+
+std::string OpenHdfsFileOp::GetDescription() {
+ return Substitute("hdfsOpenFile() for $0", fname_);
+}
+
+Status HdfsMonitor::OpenHdfsFileWithTimeout(const hdfsFS& fs, const std::string* fname,
+ int flags, uint64_t blocksize, hdfsFile* fid_out) {
+ std::shared_ptr<OpenHdfsFileOp> op(new OpenHdfsFileOp(fs, fname, flags, blocksize));
+
+ // Submit the operation to the synchronous thread pool. If this times out, it returns
+ // an error status.
+ Status status = hdfs_worker_pool_->SynchronousOffer(op,
+ FLAGS_hdfs_operation_timeout_sec * MILLIS_PER_SEC);
+ RETURN_IF_ERROR(status);
+
+ // The operation didn't time out, and there wasn't an error. Get the handle.
+ *fid_out = op->GetFileHandle();
+ return Status::OK();
+}
+
+}
+
+}
http://git-wip-us.apache.org/repos/asf/impala/blob/938be0e8/be/src/runtime/io/hdfs-monitored-ops.h
----------------------------------------------------------------------
diff --git a/be/src/runtime/io/hdfs-monitored-ops.h b/be/src/runtime/io/hdfs-monitored-ops.h
new file mode 100644
index 0000000..d20341a
--- /dev/null
+++ b/be/src/runtime/io/hdfs-monitored-ops.h
@@ -0,0 +1,62 @@
+// 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.
+
+#ifndef IMPALA_RUNTIME_IO_HDFS_MONITORED_OPS_H
+#define IMPALA_RUNTIME_IO_HDFS_MONITORED_OPS_H
+
+#include "runtime/io/handle-cache.h"
+#include "util/thread-pool.h"
+
+namespace impala {
+
+namespace io {
+
+/// The HdfsMonitor implements timeouts on HDFS operations that otherwise would block
+/// indefinitely. It submits the operations to a thread pool and waits with a timeout
+/// for a response.
+class HdfsMonitor {
+ public:
+ HdfsMonitor() {}
+
+ // Initialize the thread pool with 'num_threads'
+ Status Init(int32_t num_threads) WARN_UNUSED_RESULT;
+
+ // Open the specified HDFS file. If the operation times out, returns an error status
+ // and leaves 'hdfs_file_out' untouched. If the operation does not time out, then
+ // the operation can still fail if the file does not exist. In this case,
+ // 'hdfs_file_out' is null and an error status is returned. If the operation succeeds,
+ // then 'hdfs_file_out' contains the file handle.
+ //
+ // This is a thin wrapper around hdfsOpenFile() that implements a timeout.
+ // The differences from hdfsOpenFile() are:
+ // - The filename is taken as a const string* rather than a const char*
+ // - The bufferSize and replication are omitted, since no Impala code sets
+ // these to anything other than 0.
+ // - It has be rearranged to return Status.
+ Status OpenHdfsFileWithTimeout(const hdfsFS& fs, const std::string* fname, int flags,
+ uint64_t blocksize, hdfsFile* hdfs_file_out) WARN_UNUSED_RESULT;
+
+ private:
+ // Pool of threads handling HDFS operations.
+ std::unique_ptr<SynchronousThreadPool> hdfs_worker_pool_;
+};
+
+}
+
+}
+
+#endif
http://git-wip-us.apache.org/repos/asf/impala/blob/938be0e8/be/src/runtime/runtime-state.cc
----------------------------------------------------------------------
diff --git a/be/src/runtime/runtime-state.cc b/be/src/runtime/runtime-state.cc
index 9681aac..0899964 100644
--- a/be/src/runtime/runtime-state.cc
+++ b/be/src/runtime/runtime-state.cc
@@ -219,7 +219,9 @@ Status RuntimeState::LogOrReturnError(const ErrorMsg& message) {
|| message.error() == TErrorCode::CANCELLED_INTERNALLY
|| message.error() == TErrorCode::MEM_LIMIT_EXCEEDED
|| message.error() == TErrorCode::INTERNAL_ERROR
- || message.error() == TErrorCode::DISK_IO_ERROR) {
+ || message.error() == TErrorCode::DISK_IO_ERROR
+ || message.error() == TErrorCode::THREAD_POOL_SUBMIT_FAILED
+ || message.error() == TErrorCode::THREAD_POOL_TASK_TIMED_OUT) {
return Status(message);
}
// Otherwise, add the error to the error log and continue.
http://git-wip-us.apache.org/repos/asf/impala/blob/938be0e8/be/src/util/thread-pool-test.cc
----------------------------------------------------------------------
diff --git a/be/src/util/thread-pool-test.cc b/be/src/util/thread-pool-test.cc
index a7e4df7..d0e9287 100644
--- a/be/src/util/thread-pool-test.cc
+++ b/be/src/util/thread-pool-test.cc
@@ -68,6 +68,93 @@ TEST(ThreadPoolTest, BasicTest) {
EXPECT_EQ(expected_count, count);
}
+class SleepWorkItem : public SynchronousWorkItem {
+public:
+ SleepWorkItem(int64_t timeout_ms, bool* destructor_called)
+ : timeout_ms_(timeout_ms), destructor_called_ptr_(destructor_called) {
+ *destructor_called_ptr_ = false;
+ }
+
+ ~SleepWorkItem() {
+ *destructor_called_ptr_ = true;
+ }
+
+ virtual Status Execute() override {
+ if (timeout_ms_ > 0) SleepForMs(timeout_ms_);
+ message_ = "Done";
+ return Status::OK();
+ }
+ virtual std::string GetDescription() override {
+ return Substitute("Simple task with $0 millisecond timeout", timeout_ms_);
+ }
+
+ std::string GetMessage() { return message_; }
+private:
+ std::string message_ = "Not done";
+ int64_t timeout_ms_;
+ bool* destructor_called_ptr_;
+};
+
+TEST(ThreadPoolTest, SynchronousThreadPoolTest) {
+ // Create a synchronous pool with one thread and a queue size of one.
+ SynchronousThreadPool pool("sync-thread-pool", "worker", 1, 1);
+ ASSERT_OK(pool.Init());
+
+ // Base case: work item takes no time, run it with a timeout of 5 milliseconds
+ unique_ptr<bool> no_sleep_destroyed(new bool);
+ std::shared_ptr<SleepWorkItem> no_sleep(new SleepWorkItem(0, no_sleep_destroyed.get()));
+ ASSERT_OK(pool.SynchronousOffer(no_sleep, 5));
+ ASSERT_EQ(no_sleep->GetMessage(), "Done");
+ // If the SynchronousOffer() completed successfully, the thread pool does not have any
+ // shared_ptr to the work item. The caller is the only holder, so when it calls
+ // reset, the destructor must be called.
+ no_sleep.reset();
+ ASSERT_TRUE(*no_sleep_destroyed);
+
+ // Timeout case #1: Submit one task that takes 100 milliseconds. Offer it with a timeout
+ // of 1 millisecond so that the caller immediately times out.
+ unique_ptr<bool> long_sleep_destroyed(new bool);
+ std::shared_ptr<SleepWorkItem> long_sleep(
+ new SleepWorkItem(100, long_sleep_destroyed.get()));
+ Status timeout_status = pool.SynchronousOffer(long_sleep, 1);
+ ASSERT_EQ(timeout_status.code(), TErrorCode::THREAD_POOL_TASK_TIMED_OUT);
+ // The work item is still running, and even if the caller releases its shared_ptr,
+ // the work item is not destroyed.
+ long_sleep.reset();
+ ASSERT_FALSE(*long_sleep_destroyed);
+
+ // The single thread in the thread pool is still running. Submit another task
+ // that will queue. The task doesn't matter.
+ unique_ptr<bool> queued_task_destroyed(new bool);
+ std::shared_ptr<SleepWorkItem> queued_task(
+ new SleepWorkItem(0, queued_task_destroyed.get()));
+ Status queued_task_status = pool.SynchronousOffer(queued_task, 1);
+ ASSERT_EQ(queued_task_status.code(), TErrorCode::THREAD_POOL_TASK_TIMED_OUT);
+ // The work item is queued, and even if the caller releases its shared_ptr, the work
+ // item is not destroyed.
+ queued_task.reset();
+ ASSERT_FALSE(*queued_task_destroyed);
+
+ // Now, the queue is full. Any new task will fail to submit.
+ unique_ptr<bool> fail_to_submit_destroyed(new bool);
+ std::shared_ptr<SleepWorkItem> fail_to_submit(
+ new SleepWorkItem(0, fail_to_submit_destroyed.get()));
+ Status fail_to_submit_status = pool.SynchronousOffer(fail_to_submit, 1);
+ ASSERT_EQ(fail_to_submit_status.code(), TErrorCode::THREAD_POOL_SUBMIT_FAILED);
+ // When the submit fails, the thread pool does not keep any shared_ptr to the work
+ // item. When the caller releases its shared_ptr, the work item is immediately
+ // destroyed.
+ fail_to_submit.reset();
+ ASSERT_TRUE(*fail_to_submit_destroyed);
+
+ // The tasks will still complete
+ pool.DrainAndShutdown();
+ // The work items that the thread pool had running and in the queue are destroyed
+ // when they complete (even though the caller long since released its shared_ptr).
+ ASSERT_TRUE(*long_sleep_destroyed);
+ ASSERT_TRUE(*queued_task_destroyed);
+}
+
}
IMPALA_TEST_MAIN();
http://git-wip-us.apache.org/repos/asf/impala/blob/938be0e8/be/src/util/thread-pool.h
----------------------------------------------------------------------
diff --git a/be/src/util/thread-pool.h b/be/src/util/thread-pool.h
index fedba58..c64285c 100644
--- a/be/src/util/thread-pool.h
+++ b/be/src/util/thread-pool.h
@@ -105,6 +105,20 @@ class ThreadPool : public CacheLineAligned {
return work_queue_.BlockingPut(std::forward<V>(work));
}
+ /// Blocks until the work item is placed on the queue or the timeout expires. The
+ /// ThreadPool must be initialized before calling this method. The same requirements
+ /// about the lifetime of 'work' applies as in Offer() above. If the operation times
+ /// out, the work item can be safely freed.
+ ///
+ /// Returns true if the work item was successfully added to the queue, false otherwise
+ /// (which means the operation timed out or the thread pool has already been shut down).
+ template <typename V>
+ bool Offer(V&& work, int64_t timeout_millis) {
+ DCHECK(initialized_);
+ int64_t timeout_micros = timeout_millis * MICROS_PER_MILLI;
+ return work_queue_.BlockingPutWithTimeout(std::forward<V>(work), timeout_micros);
+ }
+
/// Shuts the thread pool down, causing the work queue to cease accepting offered work
/// and the worker threads to terminate once they have processed their current work item.
/// Returns once the shutdown flag has been set, does not wait for the threads to
@@ -219,6 +233,119 @@ class CallableThreadPool : public ThreadPool<boost::function<void()>> {
}
};
+/// Parent class for all synchronous work items
+///
+/// Important note for all subclasses:
+/// All fields need to have a lifetime that matches this operation's lifetime.
+/// In particular, caller-provided arguments need to be moved, copied, or need to have a
+/// lifetime independent of the caller. The monitored operation must survive the caller
+/// timing out and potentially deallocating memory.
+class SynchronousWorkItem {
+ public:
+ virtual ~SynchronousWorkItem() {}
+
+ /// Customized implementation for each operation. Subclasses must override this.
+ /// The status is conveyed back to the original caller.
+ virtual Status Execute() {
+ DCHECK(false) << "Execute() must be implemented";
+ return Status("Execute() must be implemented");
+ }
+
+ virtual std::string GetDescription() {
+ DCHECK(false) << "GetDescription() must be implemented";
+ return "";
+ }
+
+ private:
+ friend class SynchronousThreadPool;
+
+ /// This is called by the worker thread and handles the mechanics of notifying
+ /// the caller and conveying status when Execute() completes.
+ void WorkerExecute() {
+ Status status = Execute();
+ DCHECK(!done_promise_.IsSet());
+ discard_result(done_promise_.Set(status));
+ }
+
+ /// Wait for the operation to complete or time out with the specified limit.
+ /// If the operation times out, it returns TErrorCode::THREAD_POOL_TASK_TIMED_OUT.
+ /// Otherwise, it returns the status returned by Execute().
+ Status Wait(int64_t timeout_millis) {
+ bool timed_out;
+ Status status = done_promise_.Get(timeout_millis, &timed_out);
+ if (timed_out) {
+ Status timeout_status = Status(TErrorCode::THREAD_POOL_TASK_TIMED_OUT,
+ GetDescription(), timeout_millis / MILLIS_PER_SEC);
+ LOG(WARNING) << timeout_status.GetDetail();
+ return timeout_status;
+ }
+ return status;
+ }
+
+ // Set to the return status of ExecuteImpl() upon completion
+ Promise<Status> done_promise_;
+};
+
+/// Synchronous thread pool can run any subclass of SynchronousWorkItem
+/// Ownership is shared between the caller side and the worker side:
+/// 1. The caller accesses the operation to wait for its completion (or timeout) and
+/// retrieve any result.
+/// 2. The ThreadPool worker calls Execute() on the operation and needs to maintain
+/// ownership until the operation completes. The blocking queue inside the ThreadPool
+/// also needs to maintain ownership until a thread removes the operation for
+/// processing.
+/// This is an awkward circumstance to have exclusive ownership. The caller can time
+/// out and leave while the worker is processing. When the worker completes and could
+/// release ownership, the caller might still need to retrieve the result or the caller
+/// might be gone. shared_ptr does what we want: the HdfsMonitorOp will survive until
+/// neither thread needs it anymore.
+class SynchronousThreadPool : public ThreadPool<std::shared_ptr<SynchronousWorkItem>> {
+ public:
+ SynchronousThreadPool(const std::string& group, const std::string& thread_prefix,
+ uint32_t num_threads, uint32_t queue_size) :
+ ThreadPool<std::shared_ptr<SynchronousWorkItem>>(group, thread_prefix, num_threads,
+ queue_size, &SynchronousThreadPool::Worker) {}
+
+ /// Run the provided work item and wait up to 'timeout_milliseconds' for the
+ /// operation to complete. If it completes, return the status from the work
+ /// item's Execute() function. Otherwise, return an error status:
+ /// - THREAD_POOL_TASK_TIMED_OUT if the individual task timed out
+ /// - THREAD_POOL_SUBMIT_FAILED if all the threads are busy and the task did not
+ /// even start
+ Status SynchronousOffer(std::shared_ptr<SynchronousWorkItem> work,
+ int64_t timeout_milliseconds) {
+ MonotonicStopWatch offer_timer;
+ offer_timer.Start();
+ bool offer_success = Offer(work, timeout_milliseconds);
+ offer_timer.Stop();
+ if (!offer_success) {
+ // This scenario only happens when all threads are occupied and the queue
+ // is full. This means the system is in a catastrophic state. Log to ERROR.
+ Status failed_to_submit_status =
+ Status(TErrorCode::THREAD_POOL_SUBMIT_FAILED, work->GetDescription(),
+ timeout_milliseconds / MILLIS_PER_SEC);
+ LOG(ERROR) << failed_to_submit_status.GetDetail();
+ return failed_to_submit_status;
+ }
+
+ int64_t offer_time_millis =
+ offer_timer.ElapsedTime() / (NANOS_PER_MICRO * MICROS_PER_MILLI);
+ int64_t timeout_left = timeout_milliseconds - offer_time_millis;
+ if (timeout_left <= 0) {
+ Status timeout_status = Status(TErrorCode::THREAD_POOL_TASK_TIMED_OUT,
+ work->GetDescription(), timeout_milliseconds / MILLIS_PER_SEC);
+ LOG(WARNING) << timeout_status.GetDetail();
+ return timeout_status;
+ }
+ return work->Wait(timeout_left);
+ }
+
+ private:
+ static void Worker(int thread_id, const std::shared_ptr<SynchronousWorkItem>& work) {
+ work->WorkerExecute();
+ }
+};
+
}
#endif
http://git-wip-us.apache.org/repos/asf/impala/blob/938be0e8/common/thrift/generate_error_codes.py
----------------------------------------------------------------------
diff --git a/common/thrift/generate_error_codes.py b/common/thrift/generate_error_codes.py
index d8d40b9..dd8f4f6 100755
--- a/common/thrift/generate_error_codes.py
+++ b/common/thrift/generate_error_codes.py
@@ -373,6 +373,12 @@ error_codes = (
("PARQUET_CORRUPT_BOOL_VALUE", 122, "File '$0' is corrupt: error decoding BOOLEAN "
"value with encoding $1 at offset $2"),
+
+ ("THREAD_POOL_SUBMIT_FAILED", 123,
+ "Failed to submit $0 to thread pool after waiting $1 seconds"),
+
+ ("THREAD_POOL_TASK_TIMED_OUT", 124,
+ "$0 failed to finish before the $1 second timeout"),
)
import sys
http://git-wip-us.apache.org/repos/asf/impala/blob/938be0e8/tests/custom_cluster/test_hdfs_timeout.py
----------------------------------------------------------------------
diff --git a/tests/custom_cluster/test_hdfs_timeout.py b/tests/custom_cluster/test_hdfs_timeout.py
new file mode 100644
index 0000000..a20285a
--- /dev/null
+++ b/tests/custom_cluster/test_hdfs_timeout.py
@@ -0,0 +1,90 @@
+# 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.
+
+import pytest
+import re
+import time
+
+from tests.common.custom_cluster_test_suite import CustomClusterTestSuite
+from tests.common.skip import SkipIfNotHdfsMinicluster
+from subprocess import check_call, check_output
+
+
+@SkipIfNotHdfsMinicluster.tuned_for_minicluster
+class TestHdfsTimeouts(CustomClusterTestSuite):
+ """Test to verify that HDFS operations time out correctly."""
+
+ @classmethod
+ def get_workload(self):
+ return 'functional-query'
+
+ @pytest.mark.execute_serially
+ @CustomClusterTestSuite.with_args(
+ impalad_args="--hdfs_operation_timeout_sec=5 --max_cached_file_handles=0")
+ def test_hdfs_open_timeout(self, vector):
+ """This verifies that hdfsOpenFile times out appropriately. It tests this by
+ halting the NameNode, running a query that needs to do hdfsOpenFile,
+ and verifying that it times out and throws an error."""
+
+ # Find the NameNode's pid via pgrep. This would raise an error if it did not
+ # find a pid, so there is at least one match.
+ pgrep_output = check_output(["pgrep", "-f", "namenode.NameNode"])
+ # In our test environment, this should only match one pid
+ assert(pgrep_output.count("\n") == 1)
+ namenode_pid = pgrep_output.strip()
+
+ # Run a query successfully. This fetches metadata from the NameNode,
+ # and since this will be cached, a subsequent run will not ask the NameNode
+ # for metadata. This means a subsequent execution will only talk to the NameNode
+ # for file open.
+ self.execute_query_expect_success(self.client,
+ "select count(*) from functional.alltypes", vector=vector)
+
+ # Stop the NameNode and execute the query again. Since the file handle cache is off,
+ # the query will do hdfsOpenFile calls and talk to the NameNode. Since the NameNode
+ # is stopped, those calls will hang, testing the timeout functionality.
+ ex = None
+ result = None
+ try:
+ # Stop the NameNode
+ check_call(["kill", "-STOP", namenode_pid])
+ start_time = time.time()
+ result = self.execute_query("select count(*) from functional.alltypes",
+ vector=vector)
+ end_time = time.time()
+ except Exception, e:
+ ex = e
+ finally:
+ end_time = time.time()
+ # Always resume the NameNode
+ check_call(["kill", "-CONT", namenode_pid])
+
+ # The query should have failed, which raises an exception
+ if ex is None:
+ assert False, "Query should have failed, but instead returned {0}".format(result)
+
+ # The exception should contain the appropriate error message
+ error_pattern = "hdfsOpenFile\(\) for.*failed to finish before the 5 second timeout"
+ assert len(re.findall(error_pattern, str(ex))) > 0
+
+ # The timeout is 5 seconds and seems to be enforced within about 20 seconds, so it
+ # would be unusual if this query does not finish in 60 seconds.
+ assert (end_time - start_time) < 60.0
+
+ # Execute the query a final time to verify that the system recovers.
+ self.execute_query_expect_success(self.client,
+ "select count(*) from functional.alltypes", vector=vector)