You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@impala.apache.org by st...@apache.org on 2021/02/20 03:03:52 UTC
[impala] 08/08: IMPALA-9867: Add Support for Spilling to S3:
Milestone 1
This is an automated email from the ASF dual-hosted git repository.
stigahuang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/impala.git
commit 85e39e170fab6ceb579dac442995239857efca42
Author: Yida Wu <wy...@gmail.com>
AuthorDate: Mon Aug 10 06:41:03 2020 -0700
IMPALA-9867: Add Support for Spilling to S3: Milestone 1
Major Features
1) Local files as buffers for spilling to S3.
2) Async Upload for remote files.
3) Sync remote files deletion after query ends.
4) Local buffer files management.
5) Compatibility of spilling to local and remote.
6) All the errors from hdfs/s3 should terminate the query.
Changes on TmpFile:
* TmpFile is separated into two types of implementation, TmpFileLocal
and TmpFileRemote.
TmpFileLocal is used for Spilling to local file system.
TmpFileRemote is a new type for Spilling to the remote. It contains
two DiskFiles, one for local buffer, the other for the remote file.
* The DiskFile is an object that contains the information of a pysical
file for passing to the DiskIOMgr to execute the IO operations on
that specific file. The DiskFile also contains status information of
the file,includes DiskFileStatus::INWRITING/PERSISTED/DELETED.
When the DiskFile is initialized, it is in INWRITING status. If the
file is persisted into the file system, it would become PERSISTED
status. If the file is deleted, for example, the local buffer is
evicted, so the DiskFile status of the buffer file would become
deleted. After that, if the file is fetching from the remote, the
DiskFile status of the buffer file would become INWRITING, and then
PERSISTED if the fetching finishes successfully.
Implementation Details:
1) A new enum type is added to specify the disk type of files,
indicating where the file physically locates.
The types include DiskFileType::LOCAL/LOCAL_BUFFER/DFS/S3.
DiskFileType::LOCAL indicates the file is in the local file system.
DiskFileType::LOCAL_BUFFER indicates the file is in the local file
system, and it is the buffer of a remote scratch file.
DiskFileType::DFS/S3 indicates the file is in the HDFS/S3.
The local buffer allows the buffer pool to pin(read), but mainly
for remote files, buffer pool would pin(read) the page from the
remote file system.
2) Two disk queues have been added to do the file operation jobs.
Queue name: RemoteS3DiskFileOper/RemoteDfsDiskFileOper
File operations on the remote disk like upload and fetch should
be done in these queues. The purpose of the queues is to isolate
the file operations from normal read/write IO operations in different
queues. It could increase the efficiency of the file operations by
not being interrupted during a relatively long execution time, and
also provide a more accurate control on the thread number working on
file operation jobs.
RemoteOperRange is the new type to carry the file operation jobs.
Previously,we have request types of READ and WRITE.
Now FILE_FETCH/FILE_UPLOAD are added.
3) The tmp files are physically deleted when the tmp file group is
deconstructing. For remote files, the entire directory would be
deleted.
4) The local buffer files management is to control the total size
of local buffer files and evict files if needed.
A local buffer file can be evicted if the temporary file has uploaded
a copy to the remote disk or the query ends.
There are two modes to decide the sequence of choosing files to be
evicted first. Default is LIFO, the other is FIFO. It can be
controlled by startup option remote_tmp_files_avail_pool_lifo.
Also, a thread TmpFileSpaceReserveThreadLoop in TmpFileMgr is
running to allow to reserve buffer file space in an async way to
avoid deadlocks.
Startup option allow_spill_to_hdfs is added. By default the HDFS path
is not allowed, but for testcases, the option can be set true to
allow the use of HDFS path as scratch space for testing only.
Startup option wait_for_spill_buffer_timeout_s is added to control
the maximum duration waiting for the buffer in the TmpFileBufferPool.
Default value is 60, stands for 60 seconds.
5) Spilling to local has higher priority than spilling to remote.
If no local scratch space is available, temporary data will be
spilled to remote.
The first available local directory is used for the local buffer
for spilling to remote if any remote directory is configured.
If remote directory is configured without any available local
scratch space, an error will be returned during initialization.
The purpose of the design is to simplify the implementation in
milestone 1 with less changes on the configuration.
Example (setting remote scratch space):
Assume that the directories we have for scratch space:
* Local dir: /tmp/local_buffer, /tmp/local, /tmp/local_sec
* Remote dir: s3a://tmp/remote
The scratch space path is configured in the startup options, and could
have three types of configurations:
1. Pure local scratch space
--scratch_dirs="/tmp/local"
2. Pure remote scratch space
--scratch_dirs="s3a://tmp/remote,/tmp/local_buffer:16GB"
3. Mixed local and remote scratch space
--scratch_dirs="s3a://tmp/romote:200GB,/tmp/local_buffer:1GB,
/tmp/local:2GB, /tmp/local_sec:16GB"
* Type 1: a pure local scratch space with unlimited size.
* Type 2: a pure remote scratch space with a 16GB local buffer.
* Type 3: a mixed local and remote scratch space, the size of the local
buffer for the remote directory is 1GB, while local scratch spaces are
2GB and 16GB, remote scratch space bytes limit is 200GB. Remote scratch
space is used only when all of the local spaces are at capacity.
* Note: The first local directory would be the local buffer path, if a
remote scratch space is registered.
Limitations:
* Only one remote scratch dir is supported.
* The first local scratch dir is used for the buffer of remote scratch
space if remote scratch dir exists.
Testcases:
* Ran pre-review-test
* Unit Tests added to
tmp-file-mgr-test/disk-io-mgr-test/buffer-pool-test.
* E2E Tests added to custom_cluster/test_scratch_disk.py.
* Ran Unit Tests:
$IMPALA_HOME/be/build/debug/runtime/buffered-tuple-stream-test
$IMPALA_HOME/be/build/debug/runtime/tmp-file-mgr-test
$IMPALA_HOME/be/build/debug/runtime/bufferpool/buffer-pool-test
$IMPALA_HOME/be/build/debug/runtime/io/disk-io-mgr-test
* Ran E2E Tests:
custom_cluster/test_scratch_disk.py
Change-Id: I419b1d5dbbfe35334d9f964c4b65e553579fdc89
Reviewed-on: http://gerrit.cloudera.org:8080/16318
Reviewed-by: Impala Public Jenkins <im...@cloudera.com>
Tested-by: Impala Public Jenkins <im...@cloudera.com>
---
be/src/runtime/bufferpool/buffer-pool-test.cc | 58 ++
be/src/runtime/hdfs-fs-cache.cc | 20 +-
be/src/runtime/hdfs-fs-cache.h | 8 +-
be/src/runtime/io/CMakeLists.txt | 2 +
be/src/runtime/io/disk-file.cc | 69 ++
be/src/runtime/io/disk-file.h | 261 ++++++
be/src/runtime/io/disk-io-mgr-test.cc | 715 +++++++++++++++-
be/src/runtime/io/disk-io-mgr.cc | 237 ++++-
be/src/runtime/io/disk-io-mgr.h | 37 +-
be/src/runtime/io/file-writer.h | 80 ++
be/src/runtime/io/local-file-system.cc | 45 +-
be/src/runtime/io/local-file-system.h | 18 +-
be/src/runtime/io/local-file-writer.cc | 96 +++
be/src/runtime/io/local-file-writer.h | 45 +
be/src/runtime/io/request-context.cc | 67 +-
be/src/runtime/io/request-context.h | 17 +-
be/src/runtime/io/request-ranges.h | 144 +++-
be/src/runtime/io/scan-range.cc | 92 +-
be/src/runtime/tmp-file-mgr-internal.h | 339 +++++++-
be/src/runtime/tmp-file-mgr-test.cc | 582 ++++++++++++-
be/src/runtime/tmp-file-mgr.cc | 1143 ++++++++++++++++++++++---
be/src/runtime/tmp-file-mgr.h | 254 +++++-
be/src/util/hdfs-util.cc | 53 +-
be/src/util/hdfs-util.h | 21 +-
common/thrift/metrics.json | 20 +
tests/custom_cluster/test_scratch_disk.py | 188 +++-
26 files changed, 4334 insertions(+), 277 deletions(-)
diff --git a/be/src/runtime/bufferpool/buffer-pool-test.cc b/be/src/runtime/bufferpool/buffer-pool-test.cc
index 949ac14..23009f9 100644
--- a/be/src/runtime/bufferpool/buffer-pool-test.cc
+++ b/be/src/runtime/bufferpool/buffer-pool-test.cc
@@ -59,10 +59,18 @@ using std::uniform_real_distribution;
DECLARE_bool(disk_spill_encryption);
DECLARE_string(disk_spill_compression_codec);
DECLARE_bool(disk_spill_punch_holes);
+DECLARE_string(remote_tmp_file_block_size);
+DECLARE_string(remote_tmp_file_size);
+DECLARE_bool(allow_spill_to_hdfs);
// This suffix is appended to a tmp dir
const string SCRATCH_SUFFIX = "/impala-scratch";
+/// For testing spill to remote.
+static const string HDFS_LOCAL_URL = "hdfs://localhost";
+static const string REMOTE_URL = HDFS_LOCAL_URL;
+static const string LOCAL_BUFFER_PATH = "/tmp/buffer-pool-test-buffer";
+
namespace impala {
using BufferHandle = BufferPool::BufferHandle;
@@ -73,6 +81,9 @@ class BufferPoolTest : public ::testing::Test {
public:
virtual void SetUp() {
test_env_.reset(new TestEnv);
+ FLAGS_allow_spill_to_hdfs = true;
+ FLAGS_remote_tmp_file_size = "512KB";
+
// Don't create global buffer pool in 'test_env_' - we'll create a buffer pool in
// each test function.
test_env_->DisableBufferPool();
@@ -140,6 +151,37 @@ class BufferPoolTest : public ::testing::Test {
return tmp_dirs;
}
+ /// Init a new tmp file manager with spill to remote options.
+ vector<string> InitTmpFileMgrSpillToRemote(
+ int num_local_dirs, int64_t local_spill_limit, int64_t local_buffer_limit) {
+ vector<string> tmp_dirs;
+ int expected_tmp_dirs_num = num_local_dirs + 2;
+ int expected_active_dev = num_local_dirs + 1;
+ if (local_buffer_limit == -1) {
+ tmp_dirs.push_back(LOCAL_BUFFER_PATH);
+ } else {
+ tmp_dirs.push_back(Substitute(LOCAL_BUFFER_PATH + ":$0", local_buffer_limit));
+ }
+ tmp_dirs.push_back(REMOTE_URL);
+ EXPECT_OK(FileSystemUtil::RemoveAndCreateDirectory(LOCAL_BUFFER_PATH));
+ for (int i = 0; i < num_local_dirs; ++i) {
+ const string& create_dir = Substitute("/tmp/buffer-pool-test.$0", i);
+ // Fix permissions in case old directories were left from previous runs of test.
+ chmod((create_dir + SCRATCH_SUFFIX).c_str(), S_IRWXU);
+ EXPECT_OK(FileSystemUtil::RemoveAndCreateDirectory(create_dir));
+ created_tmp_dirs_.push_back(create_dir);
+ const string& dir = Substitute("/tmp/buffer-pool-test.$0:$1", i, local_spill_limit);
+ tmp_dirs.push_back(dir);
+ }
+ EXPECT_EQ(expected_tmp_dirs_num, tmp_dirs.size());
+ test_env_.reset(new TestEnv);
+ test_env_->DisableBufferPool();
+ test_env_->SetTmpFileMgrArgs(tmp_dirs, false, "", false);
+ EXPECT_OK(test_env_->Init());
+ EXPECT_EQ(expected_active_dev, test_env_->tmp_file_mgr()->NumActiveTmpDevices());
+ return tmp_dirs;
+ }
+
static int64_t QueryId(int hi, int lo) { return static_cast<int64_t>(hi) << 32 | lo; }
/// Helper function to create one reservation tracker per query.
@@ -2018,6 +2060,22 @@ TEST_F(BufferPoolTest, Multi8Random) {
TestRandomInternalMulti(8, 8 * 1024, false);
}
+// Test with remote scratch space mixed with two local dirs.
+TEST_F(BufferPoolTest, Multi8RandomSpillToRemoteMix) {
+ // 4M buffer size.
+ InitTmpFileMgrSpillToRemote(2, 8 * 1024, 4 * 1024 * 1024);
+ TestRandomInternalMulti(8, 8 * 1024, true);
+ TestRandomInternalMulti(8, 8 * 1024, false);
+}
+
+// Test with remote scratch space only.
+TEST_F(BufferPoolTest, Multi8RandomSpillToRemote) {
+ // 4M buffer size.
+ InitTmpFileMgrSpillToRemote(0, -1, 4 * 1024 * 1024);
+ TestRandomInternalMulti(8, 8 * 1024, true);
+ TestRandomInternalMulti(8, 8 * 1024, false);
+}
+
// Sanity test with hole punching and no compression.
// This will be run with and without encryption because those flags are toggled for the
// entire test suite.
diff --git a/be/src/runtime/hdfs-fs-cache.cc b/be/src/runtime/hdfs-fs-cache.cc
index d3acd0c..7f8fcb7 100644
--- a/be/src/runtime/hdfs-fs-cache.cc
+++ b/be/src/runtime/hdfs-fs-cache.cc
@@ -68,8 +68,8 @@ Status HdfsFsCache::Init() {
return Status::OK();
}
-Status HdfsFsCache::GetConnection(const string& path, hdfsFS* fs,
- HdfsFsMap* local_cache) {
+Status HdfsFsCache::GetConnection(const string& path, hdfsFS* fs, HdfsFsMap* local_cache,
+ const HdfsConnOptions* options) {
string err;
const string& namenode = GetNameNodeFromPath(path, &err);
if (!err.empty()) return Status(err);
@@ -90,15 +90,25 @@ Status HdfsFsCache::GetConnection(const string& path, hdfsFS* fs,
if (i == fs_map_.end()) {
hdfsBuilder* hdfs_builder = hdfsNewBuilder();
hdfsBuilderSetNameNode(hdfs_builder, namenode.c_str());
- if (!s3a_access_key_.empty()) {
+ if (!s3a_access_key_.empty() || (options != nullptr && !options->empty())) {
// Use a new instance of the filesystem object to be sure that it picks up the
// configuration changes we're going to make. Without this call, a cached
// filesystem object is used which is unaffected by calls to
// hdfsBuilderConfSetStr(). This is unexpected behavior in the HDFS API, but is
// unlikely to change.
hdfsBuilderSetForceNewInstance(hdfs_builder);
- hdfsBuilderConfSetStr(hdfs_builder, "fs.s3a.access.key", s3a_access_key_.c_str());
- hdfsBuilderConfSetStr(hdfs_builder, "fs.s3a.secret.key", s3a_secret_key_.c_str());
+ if (!s3a_access_key_.empty()) {
+ hdfsBuilderConfSetStr(
+ hdfs_builder, "fs.s3a.access.key", s3a_access_key_.c_str());
+ hdfsBuilderConfSetStr(
+ hdfs_builder, "fs.s3a.secret.key", s3a_secret_key_.c_str());
+ }
+ if (options != nullptr) {
+ for (auto option : *options) {
+ hdfsBuilderConfSetStr(
+ hdfs_builder, option.first.c_str(), option.second.c_str());
+ }
+ }
}
*fs = hdfsBuilderConnect(hdfs_builder);
if (*fs == NULL) {
diff --git a/be/src/runtime/hdfs-fs-cache.h b/be/src/runtime/hdfs-fs-cache.h
index 9f283b8..2f37815 100644
--- a/be/src/runtime/hdfs-fs-cache.h
+++ b/be/src/runtime/hdfs-fs-cache.h
@@ -40,6 +40,7 @@ namespace impala {
class HdfsFsCache {
public:
typedef boost::unordered_map<std::string, hdfsFS> HdfsFsMap;
+ typedef std::vector<std::pair<string, string>> HdfsConnOptions;
static HdfsFsCache* instance() { return HdfsFsCache::instance_.get(); }
@@ -51,10 +52,11 @@ class HdfsFsCache {
/// Get connection to specific fs by specifying a path. Optionally, a local cache can
/// be provided so that the process-wide lock can be avoided on subsequent calls for
- /// the same filesystem. The caller is responsible for synchronizing the local cache
+ /// the same filesystem. Also, options can be provided to create the hdfs connection.
+ /// The caller is responsible for synchronizing the local cache
/// (e.g. by passing a thread-local cache).
- Status GetConnection(const std::string& path, hdfsFS* fs,
- HdfsFsMap* local_cache = NULL);
+ Status GetConnection(const std::string& path, hdfsFS* fs, HdfsFsMap* local_cache = NULL,
+ const HdfsConnOptions* options = NULL);
/// Get NameNode info from path, set error message if path is not valid.
/// Exposed as a static method for testing purpose.
diff --git a/be/src/runtime/io/CMakeLists.txt b/be/src/runtime/io/CMakeLists.txt
index 7c2637b..9d4ae98 100644
--- a/be/src/runtime/io/CMakeLists.txt
+++ b/be/src/runtime/io/CMakeLists.txt
@@ -25,6 +25,7 @@ add_library(Io
data-cache.cc
disk-io-mgr.cc
disk-io-mgr-stress.cc
+ disk-file.cc
local-file-system.cc
local-file-system-with-fault-injection.cc
error-converter.cc
@@ -32,6 +33,7 @@ add_library(Io
scan-range.cc
hdfs-file-reader.cc
local-file-reader.cc
+ local-file-writer.cc
hdfs-monitored-ops.cc
data-cache-trace.cc
)
diff --git a/be/src/runtime/io/disk-file.cc b/be/src/runtime/io/disk-file.cc
new file mode 100644
index 0000000..fc4f5fe
--- /dev/null
+++ b/be/src/runtime/io/disk-file.cc
@@ -0,0 +1,69 @@
+// 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 "runtime/io/disk-file.h"
+#include <iostream>
+#include <mutex>
+#include <thread>
+#include "common/names.h"
+#include "runtime/io/local-file-writer.h"
+#include "util/filesystem-util.h"
+#include "util/spinlock.h"
+
+using namespace impala;
+using namespace impala::io;
+
+static const Status& DISK_FILE_DELETE_FAILED_INCORRECT_STATUS = Status(ErrorMsg::Init(
+ TErrorCode::GENERAL, "DiskFile::Delete() failed with incorrect status"));
+
+Status DiskFile::Delete(const unique_lock<shared_mutex>& lock) {
+ DCHECK(lock.mutex() == &physical_file_lock_ && lock.owns_lock());
+ Status status = Status::OK();
+ unique_lock<SpinLock> status_l(status_lock_);
+ // No support for remote file deletion yet.
+ if (disk_type_ == DiskFileType::LOCAL_BUFFER || disk_type_ == DiskFileType::LOCAL) {
+ if (is_deleted(status_l)) return DISK_FILE_DELETE_FAILED_INCORRECT_STATUS;
+ RETURN_IF_ERROR(FileSystemUtil::RemovePaths({path_}));
+ SetStatusLocked(io::DiskFileStatus::DELETED, status_l);
+ }
+ return status;
+}
+
+DiskFile::DiskFile(const string& path, DiskIoMgr* io_mgr)
+ : path_(path),
+ disk_type_(DiskFileType::LOCAL),
+ file_status_(DiskFileStatus::INWRITING),
+ file_writer_(new LocalFileWriter(io_mgr, path.c_str())),
+ space_reserved_(true) {}
+
+DiskFile::DiskFile(const string& path, DiskIoMgr* io_mgr, int64_t file_size,
+ DiskFileType disk_type, const hdfsFS* hdfs_conn)
+ : path_(path),
+ file_size_(file_size),
+ disk_type_(disk_type),
+ file_status_(DiskFileStatus::INWRITING) {
+ DCHECK(disk_type != DiskFileType::LOCAL);
+ if (disk_type == DiskFileType::LOCAL_BUFFER) {
+ file_writer_.reset(new LocalFileWriter(io_mgr, path.c_str(), file_size));
+ hdfs_conn_ = nullptr;
+ space_reserved_.Store(false);
+ } else {
+ DCHECK(hdfs_conn != nullptr);
+ hdfs_conn_ = *hdfs_conn;
+ space_reserved_.Store(true);
+ }
+}
diff --git a/be/src/runtime/io/disk-file.h b/be/src/runtime/io/disk-file.h
new file mode 100644
index 0000000..2febae7
--- /dev/null
+++ b/be/src/runtime/io/disk-file.h
@@ -0,0 +1,261 @@
+// 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.
+
+#pragma once
+
+#include <string>
+
+#include <boost/thread/shared_mutex.hpp>
+#include "common/atomic.h"
+#include "common/status.h"
+#include "runtime/hdfs-fs-cache.h"
+#include "runtime/io/file-writer.h"
+
+#include "util/spinlock.h"
+
+namespace impala {
+class TmpFileMgr;
+class TmpFileRemote;
+namespace io {
+class RemoteOperRange;
+class ScanRange;
+class WriteRange;
+
+/// DiskFileType indicates the type of the file handled by the DiskFile.
+/// LOCAL indicates the file is in the local filesystem.
+/// LOCAL_BUFFER indicates the file is used as a buffer in the local filesystem.
+/// DFS indicates the file is an HDFS file.
+/// S3 indicates the file is in the Amazon S3 filesystem.
+/// DUMMY is a type for a fake file.
+enum class DiskFileType { LOCAL, LOCAL_BUFFER, DFS, S3, DUMMY };
+
+/// DiskFileStatus indicates the status of the file handled by the DiskFile.
+/// INWRITING indicates the file is allowed to write, and it is the initial status.
+/// PERSISTED indicates the file is closed with all of the content written.
+/// DELETED indicates the file is deleted from the filesystem.
+enum class DiskFileStatus { INWRITING, PERSISTED, DELETED };
+
+/// DiskFile is a handle to a physical file, mainly contains the information
+/// of the file, control the status changing, and provide the file writer for
+/// writing the file.
+///
+/// Some methods of DiskFile are not thread-safe.
+class DiskFile {
+ public:
+ /// Constructor for a LOCAL file only.
+ DiskFile(const std::string& path, DiskIoMgr* io_mgr);
+
+ /// Constructor for a file with detail settings.
+ DiskFile(const std::string& path, DiskIoMgr* io_mgr, int64_t file_size,
+ DiskFileType disk_type, const hdfsFS* hdfs_conn = nullptr);
+
+ virtual ~DiskFile() {}
+
+ // Delete the physical file. Caller should hold the exclusive file lock.
+ Status Delete(const std::unique_lock<boost::shared_mutex>& lock);
+
+ /// Returns the path of the DiskFile.
+ /// If it is a remote temporary file, the path should be a remote scratch space path.
+ /// The same, if it is a local one, the path should be a local path either.
+ const std::string& path() const { return path_; }
+
+ /// Returns the disk type of the DiskFile.
+ DiskFileType disk_type() { return disk_type_; }
+
+ /// Return the default size of the file.
+ int64_t file_size() const { return file_size_; }
+
+ /// Return the actual size of the file.
+ int64_t actual_file_size() const { return actual_file_size_.Load(); }
+
+ /// If return True, the file is persisted.
+ /// The caller should hold the status lock.
+ bool is_persisted(const std::unique_lock<SpinLock>& l) {
+ return GetFileStatusLocked(l) == DiskFileStatus::PERSISTED;
+ }
+
+ /// If return True, the file is in writing.
+ /// The caller should hold the status lock.
+ bool is_writing(const std::unique_lock<SpinLock>& l) {
+ return GetFileStatusLocked(l) == DiskFileStatus::INWRITING;
+ }
+
+ /// If return True, the file is deleted.
+ /// The caller should hold the status lock.
+ bool is_deleted(const std::unique_lock<SpinLock>& l) {
+ return GetFileStatusLocked(l) == DiskFileStatus::DELETED;
+ }
+
+ /// Set the status of the DiskFile. Caller should not hold the status lock.
+ void SetStatus(DiskFileStatus status) {
+ std::unique_lock<SpinLock> l(status_lock_);
+ SetStatusLocked(status, l);
+ }
+
+ /// Same as SetStatus, but the caller should hold the status lock.
+ void SetStatusLocked(
+ DiskFileStatus status, const std::unique_lock<SpinLock>& status_lock) {
+ DCHECK(status_lock.mutex() == &status_lock_ && status_lock.owns_lock());
+ SetInternalStatus(status);
+ }
+
+ /// Returns the status of the file.
+ /// The caller should not hold the status lock.
+ DiskFileStatus GetFileStatus() {
+ std::unique_lock<SpinLock> l(status_lock_);
+ return GetFileStatusLocked(l);
+ }
+
+ DiskFileStatus GetFileStatusLocked(const std::unique_lock<SpinLock>& status_lock) {
+ DCHECK(status_lock.mutex() == &status_lock_ && status_lock.owns_lock());
+ return file_status_;
+ }
+
+ /// Return the writer of the file.
+ io::FileWriter* GetFileWriter() { return file_writer_.get(); }
+
+ /// Getter and setter for space reserved flag.
+ void SetSpaceReserved() { space_reserved_.Store(true); }
+ bool IsSpaceReserved() { return space_reserved_.Load(); }
+
+ /// Set actual file size. Should only be called by the TmpFileRemote::AllocateSpace()
+ /// right after the allocation is at capacity, and the function should only be called
+ /// once during the lifetime of the DiskFile.
+ void SetActualFileSize(int64_t size) {
+ DCHECK_EQ(0, actual_file_size_.Load());
+ DCHECK_LE(file_size_, size);
+ actual_file_size_.Store(size);
+ }
+
+ private:
+ friend class RemoteOperRange;
+ friend class ScanRange;
+ friend class WriteRange;
+ friend class impala::TmpFileMgr;
+ friend class impala::TmpFileRemote;
+
+ /// Path of the physical file in the filesystem.
+ const std::string path_;
+
+ /// The default file size of the disk file.
+ const int64_t file_size_ = 0;
+
+ /// Specify the type of the disk where the file locates.
+ const DiskFileType disk_type_;
+
+ /// The status of the file.
+ /// It includes IN_WRITING/PERSISTED/DELETED.
+ DiskFileStatus file_status_;
+
+ /// The FileWriter of the DiskFile.
+ /// Caller can use the writer to write a range to the disk file.
+ std::unique_ptr<FileWriter> file_writer_;
+
+ /// Every time to query or modify the status, or need to guarantee working under
+ /// certain status, the caller should own the lock.
+ /// If a thread needs to hold both status lock of two disk files, the sequence of
+ /// acquiring locks must be from local file to remote file to avoid a deadlock.
+ /// The lock order of file lock (below) and status lock should be file lock acquired
+ /// first.
+ SpinLock status_lock_;
+
+ /// Protect the physical file from deleting. For removing the file, the caller should
+ /// hold the unique lock of the mutex. In other cases, like writing to or reading from
+ /// the file, a shared lock will be held.
+ /// If a thread needs to hold the lock of two disk files for doing an operation like
+ /// upload, the sequence of acquiring locks must be from local file to remote file to
+ /// avoid a deadlock.
+ /// The lock order of file lock and status lock (above) should be file lock acquired
+ /// first.
+ boost::shared_mutex physical_file_lock_;
+
+ /// The hdfs connection used to connect to the remote scratch path.
+ hdfsFS hdfs_conn_;
+
+ /// Specify if the file's space is reserved to be allowed to write to the filesystem
+ /// because the filesystem may reach the size limit and needs some time before it can
+ /// release space for new writes to the filesystem, so the space reserved indicator is
+ /// a way to gain the permission to write.
+ /// It is mainly used for a LOCAL_BUFFER file to control multiple write ranges to
+ /// guarantee sending them to the DiskQueues with space reserved.
+ /// For simplicity, the space_reserved_ only allows one-way transition, from false to
+ /// true, once the space_reserved_ is set true, won't be changed back.
+ /// For a LOCAL_BUFFER file, space_reserved_ is primarily accessed in TmpFileBufferPool
+ /// under the exclusive TmpFileBufferPool::lock_ to guarantee the correctness of
+ /// accessing. One special case is that space_reserved_ can be set true if quick space
+ /// reservation is successful in TmpFileGroup::AllocateRemoteSpace(), but because the
+ /// "set" only works on the first range of the file, and the operation is before all the
+ /// ranges of the same file arriving the TmpFileBufferPool (it is guaranteed by
+ /// TmpFileGroup::lock_ gained in the TmpFileGroup::AllocateSpace()), the correctness is
+ /// guaranteed. One other special "read" on the indicator is in TmpFileRemote::Remove(),
+ /// when the function is called in TmpFileGroup::Close() and the order should be after
+ /// all of the "set" of the indicator, so the correctness of the "read" is guaranteed.
+ /// If any above condition is changed, we may need to reconsider the safety.
+ /// For type LOCAL, the space reserved is always true, because the query would fail
+ /// immediately before creating the DiskFile if it exceeds the total size limit.
+ AtomicBool space_reserved_{false};
+
+ /// The actual file size is mainly used for a LOCAL_BUFFER file, the value could be
+ /// slightly over the default file size. Should only be set by the
+ /// TmpFileRemote::AllocateSpace() right after the allocation is at capacity, and the
+ /// value would be checked by the DiskQueues when writing the range to the file system.
+ /// If the current bytes written is equals to the actual file size, it indicates the
+ /// range is the last range of the file, the file can be closed and starts to upload.
+ /// The reason of identifying the last range by the actual file size is that, the order
+ /// of writing to the file system doesn't have to be the same as the order assigned to
+ /// them in TmpFileMgr because the ranges of a file arrive in different disk queues and
+ /// are written separately, and for efficiency consider, we do a sequential write for
+ /// the same LOCAL_BUFFER file instead of seeking for a specific offset, so the last
+ /// range being written into the filesystem is unknown util it is written. The safety is
+ /// guaranteed because the actual_file_size_ is only set once when the allocation of a
+ /// file is at capacity in TmpFileMgr, and it is guaranteed to be set before the last
+ /// range arrives the DiskQueue. When actual_file_size_ is used for a file in remote FS,
+ /// like S3, it is set after a successful upload.
+ AtomicInt64 actual_file_size_{0};
+
+ /// Internal setter to set the status.
+ /// The status is from INWRITING -> PERSISTED -> DELETED, which should not be a
+ /// reverse transition.
+ void SetInternalStatus(DiskFileStatus new_status) {
+ DCHECK(disk_type_ != DiskFileType::LOCAL);
+ switch (new_status) {
+ case DiskFileStatus::INWRITING: {
+ DCHECK(file_status_ != DiskFileStatus::DELETED
+ && file_status_ != DiskFileStatus::PERSISTED);
+ break;
+ }
+ case DiskFileStatus::PERSISTED: {
+ DCHECK(file_status_ != DiskFileStatus::DELETED);
+ break;
+ }
+ case DiskFileStatus::DELETED: {
+ break;
+ }
+ default:
+ DCHECK(false) << "Invalid disk file status: " << static_cast<int>(new_status);
+ }
+ file_status_ = new_status;
+ }
+
+ /// Return the lock of the file.
+ boost::shared_mutex* GetFileLock() { return &physical_file_lock_; }
+
+ /// Return the status lock of the file.
+ SpinLock* GetStatusLock() { return &status_lock_; }
+};
+} // namespace io
+} // namespace impala
diff --git a/be/src/runtime/io/disk-io-mgr-test.cc b/be/src/runtime/io/disk-io-mgr-test.cc
index 96291fb..e9ece41 100644
--- a/be/src/runtime/io/disk-io-mgr-test.cc
+++ b/be/src/runtime/io/disk-io-mgr-test.cc
@@ -29,6 +29,8 @@
#include "runtime/io/local-file-system-with-fault-injection.h"
#include "runtime/io/request-context.h"
#include "runtime/test-env.h"
+#include "runtime/tmp-file-mgr-internal.h"
+#include "runtime/tmp-file-mgr.h"
#include "testutil/gtest-util.h"
#include "testutil/rand-util.h"
#include "testutil/scoped-flag-setter.h"
@@ -51,15 +53,28 @@ DECLARE_int32(num_s3_io_threads);
DECLARE_int32(num_adls_io_threads);
DECLARE_int32(num_abfs_io_threads);
DECLARE_int32(num_ozone_io_threads);
+DECLARE_int32(num_remote_hdfs_file_oper_io_threads);
+DECLARE_int32(num_s3_file_oper_io_threads);
+
#ifndef NDEBUG
DECLARE_int32(stress_disk_read_delay_ms);
#endif
+DECLARE_string(remote_tmp_file_size);
+DECLARE_string(remote_tmp_file_block_size);
+DECLARE_bool(allow_spill_to_hdfs);
+
const int MIN_BUFFER_SIZE = 128;
const int MAX_BUFFER_SIZE = 1024;
const int64_t LARGE_RESERVATION_LIMIT = 4L * 1024L * 1024L * 1024L;
const int64_t LARGE_INITIAL_RESERVATION = 128L * 1024L * 1024L;
const int64_t BUFFER_POOL_CAPACITY = LARGE_RESERVATION_LIMIT;
+
+/// For testing spill to remote.
+static const string HDFS_LOCAL_URL = "hdfs://localhost:20500";
+static const string REMOTE_URL = HDFS_LOCAL_URL;
+static const string LOCAL_BUFFER_PATH = "/tmp/tmp-file-mgr-test-buffer";
+
namespace impala {
namespace io {
@@ -69,10 +84,14 @@ static Status CONTEXT_CANCELLED_STATUS =
class DiskIoMgrTest : public testing::Test {
public:
virtual void SetUp() {
+ FLAGS_remote_tmp_file_block_size = "1K";
+ FLAGS_allow_spill_to_hdfs = true;
+
test_env_.reset(new TestEnv);
// Tests try to allocate arbitrarily small buffers. Ensure Buffer Pool allows it.
test_env_->SetBufferPoolArgs(1, BUFFER_POOL_CAPACITY);
ASSERT_OK(test_env_->Init());
+ metrics_.reset(new MetricGroup("disk-io-mgr-test"));
RandTestUtil::SeedRng("DISK_IO_MGR_TEST_SEED", &rng_);
}
@@ -137,6 +156,36 @@ class DiskIoMgrTest : public testing::Test {
}
}
+ RuntimeProfile* NewProfile() { return RuntimeProfile::Create(&pool_, "test profile"); }
+
+ /// Create a new file group with the default configs.
+ TmpFileGroup* NewFileGroup(DiskIoMgr* io_mgr = nullptr) {
+ if (io_mgr == nullptr) io_mgr = test_env_->exec_env()->disk_io_mgr();
+ TmpFileGroup* file_group = pool_.Add(
+ new TmpFileGroup(test_env_->tmp_file_mgr(), io_mgr, NewProfile(), TUniqueId()));
+ return file_group;
+ }
+
+ /// Create a new file group with spiiling to remote setting.
+ TmpFileGroup* NewRemoteFileGroup(TmpFileMgr* tmp_file_mgr, DiskIoMgr* io_mgr) {
+ if (io_mgr == nullptr) io_mgr = test_env_->exec_env()->disk_io_mgr();
+ vector<string> tmp_dirs({LOCAL_BUFFER_PATH});
+ RemoveAndCreateDirs(tmp_dirs);
+ tmp_dirs.push_back(REMOTE_URL);
+ Status status = tmp_file_mgr->InitCustom(tmp_dirs, true, "", false, metrics_.get());
+ if (!status.ok()) return nullptr;
+ return pool_.Add(new TmpFileGroup(tmp_file_mgr, io_mgr, NewProfile(), TUniqueId()));
+ }
+
+ bool FileExist(const std::string& name) {
+ struct stat buffer;
+ return (stat(name.c_str(), &buffer) == 0);
+ }
+
+ bool HdfsFileExist(const std::string& name) {
+ return (hdfsExists(hdfsConnect("default", 0), name.c_str()) == 0);
+ }
+
protected:
void CreateTempFile(const char* filename, const char* data) {
CreateTempFile(filename, data, strlen(data));
@@ -160,6 +209,12 @@ class DiskIoMgrTest : public testing::Test {
return truncate(filename, file_size);
}
+ void RemoveAndCreateDirs(const vector<string>& dirs) {
+ for (const string& dir : dirs) {
+ ASSERT_OK(FileSystemUtil::RemoveAndCreateDirectory(dir));
+ }
+ }
+
// Validates that buffer[i] is \0 or expected[i]
static void ValidateEmptyOrCorrect(const char* expected, const char* buffer, int len) {
for (int i = 0; i < len; ++i) {
@@ -259,9 +314,8 @@ class DiskIoMgrTest : public testing::Test {
// Auxiliary function to ExecuteWriteFailureTest(). Handles the
// AddWriteRange() call.
- void AddWriteRange(int num_of_writes, int32_t* data,
- const string& tmp_file, int offset, RequestContext* writer,
- const string& expected_output);
+ void AddWriteRange(int num_of_writes, int32_t* data, const string& tmp_file, int offset,
+ RequestContext* writer, const string& expected_output, TmpFileGroup* file_group);
void SingleReaderTestBody(const char* data, const char* expected_result,
vector<ScanRange::SubRange> sub_ranges = {});
@@ -279,13 +333,69 @@ class DiskIoMgrTest : public testing::Test {
ObjectPool pool_;
+ scoped_ptr<MetricGroup> metrics_;
+
+ /// The file groups created - closed at end of each test.
+ vector<TmpFileGroup*> file_groups_;
+
ReservationTracker root_reservation_;
mutex written_mutex_;
ConditionVariable writes_done_;
int num_ranges_written_;
+
+ mutex oper_mutex_;
+ ConditionVariable oper_done_;
+ int num_oper_;
};
+TEST_F(DiskIoMgrTest, TestDisk) {
+ InitRootReservation(LARGE_RESERVATION_LIMIT);
+ num_ranges_written_ = 0;
+ string tmp_file = "/tmp/disk_io_mgr_test.txt";
+ int num_ranges = 1;
+ int64_t cur_offset = 0;
+
+ scoped_ptr<DiskIoMgr> read_io_mgr(new DiskIoMgr(1, 1, 1, 1, 10));
+ BufferPool::ClientHandle read_client;
+ RegisterBufferPoolClient(
+ LARGE_RESERVATION_LIMIT, LARGE_INITIAL_RESERVATION, &read_client);
+ ASSERT_OK(read_io_mgr->Init());
+ unique_ptr<RequestContext> reader = read_io_mgr->RegisterContext();
+ TmpFileGroup* tmp_file_grp = NewFileGroup();
+ // Pool for temporary objects from this iteration only.
+ ObjectPool tmp_pool;
+ DiskIoMgr io_mgr(1, 1, 1, 1, 10);
+ ASSERT_OK(io_mgr.Init());
+ unique_ptr<RequestContext> writer = io_mgr.RegisterContext();
+ for (int i = 0; i < 1; ++i) {
+ int32_t* data = tmp_pool.Add(new int32_t);
+ *data = 5;
+ WriteRange** new_range = tmp_pool.Add(new WriteRange*);
+ WriteRange::WriteDoneCallback callback =
+ bind(mem_fn(&DiskIoMgrTest::WriteValidateCallback), this, num_ranges, new_range,
+ read_io_mgr.get(), reader.get(), &read_client, data, Status::OK(), _1);
+ *new_range = tmp_pool.Add(new WriteRange(tmp_file, cur_offset, 0, callback));
+ (*new_range)->SetData(reinterpret_cast<uint8_t*>(data), sizeof(int32_t));
+ TmpFile** new_tmp_file_obj = tmp_pool.Add(new TmpFile*);
+ *new_tmp_file_obj = tmp_pool.Add(new TmpFileLocal(tmp_file_grp, 0, tmp_file));
+ (*new_range)->SetDiskFile((*new_tmp_file_obj)->GetWriteFile());
+ EXPECT_OK(writer->AddWriteRange(*new_range));
+ cur_offset += sizeof(int32_t);
+ }
+
+ {
+ unique_lock<mutex> lock(written_mutex_);
+ while (num_ranges_written_ < 1) writes_done_.Wait(lock);
+ }
+ num_ranges_written_ = 0;
+ io_mgr.UnregisterContext(writer.get());
+
+ tmp_file_grp->Close();
+ read_io_mgr->UnregisterContext(reader.get());
+ buffer_pool()->DeregisterClient(&read_client);
+}
+
// Test a single writer with multiple disks and threads per disk. Each WriteRange
// writes random 4-byte integers, and upon completion, the written data is validated
// by reading the data back via a separate IoMgr instance. All writes are expected to
@@ -310,6 +420,7 @@ TEST_F(DiskIoMgrTest, SingleWriter) {
LARGE_RESERVATION_LIMIT, LARGE_INITIAL_RESERVATION, &read_client);
ASSERT_OK(read_io_mgr->Init());
unique_ptr<RequestContext> reader = read_io_mgr->RegisterContext();
+ TmpFileGroup* tmp_file_grp = NewFileGroup();
for (int num_threads_per_disk = 1; num_threads_per_disk <= 5; ++num_threads_per_disk) {
for (int num_disks = 1; num_disks <= 5; num_disks += 2) {
// Pool for temporary objects from this iteration only.
@@ -327,6 +438,9 @@ TEST_F(DiskIoMgrTest, SingleWriter) {
*new_range = tmp_pool.Add(
new WriteRange(tmp_file, cur_offset, num_ranges % num_disks, callback));
(*new_range)->SetData(reinterpret_cast<uint8_t*>(data), sizeof(int32_t));
+ TmpFile** new_tmp_file_obj = tmp_pool.Add(new TmpFile*);
+ *new_tmp_file_obj = tmp_pool.Add(new TmpFileLocal(tmp_file_grp, 0, tmp_file));
+ (*new_range)->SetDiskFile((*new_tmp_file_obj)->GetWriteFile());
EXPECT_OK(writer->AddWriteRange(*new_range));
cur_offset += sizeof(int32_t);
}
@@ -340,6 +454,7 @@ TEST_F(DiskIoMgrTest, SingleWriter) {
}
}
+ tmp_file_grp->Close();
read_io_mgr->UnregisterContext(reader.get());
buffer_pool()->DeregisterClient(&read_client);
}
@@ -356,9 +471,11 @@ TEST_F(DiskIoMgrTest, InvalidWrite) {
unique_ptr<RequestContext> writer = io_mgr.RegisterContext();
int32_t* data = pool_.Add(new int32_t);
*data = rand();
+ TmpFileGroup* tmp_file_grp = NewFileGroup();
// Write to file in non-existent directory.
WriteRange** new_range = pool_.Add(new WriteRange*);
+ TmpFile** new_tmp_file_obj = pool_.Add(new TmpFile*);
WriteRange::WriteDoneCallback callback =
bind(mem_fn(&DiskIoMgrTest::WriteValidateCallback), this, num_of_writes, new_range,
nullptr, nullptr, nullptr, data,
@@ -368,6 +485,8 @@ TEST_F(DiskIoMgrTest, InvalidWrite) {
*new_range = pool_.Add(new WriteRange(tmp_file, rand(), 0, callback));
(*new_range)->SetData(reinterpret_cast<uint8_t*>(data), sizeof(int32_t));
+ *new_tmp_file_obj = pool_.Add(new TmpFileLocal(tmp_file_grp, 0, tmp_file));
+ (*new_range)->SetDiskFile((*new_tmp_file_obj)->GetWriteFile());
EXPECT_OK(writer->AddWriteRange(*new_range));
// Write to a bad location in a file that exists.
@@ -379,6 +498,7 @@ TEST_F(DiskIoMgrTest, InvalidWrite) {
}
new_range = pool_.Add(new WriteRange*);
+ new_tmp_file_obj = pool_.Add(new TmpFile*);
callback = bind(mem_fn(&DiskIoMgrTest::WriteValidateCallback), this, num_of_writes,
new_range, nullptr, nullptr, nullptr, data,
Status(TErrorCode::DISK_IO_ERROR, GetBackendString(),
@@ -386,6 +506,8 @@ TEST_F(DiskIoMgrTest, InvalidWrite) {
"Invalid inputs. errno=22, offset=-1"), _1);
*new_range = pool_.Add(new WriteRange(tmp_file, -1, 0, callback));
+ *new_tmp_file_obj = pool_.Add(new TmpFileLocal(tmp_file_grp, 0, tmp_file));
+ (*new_range)->SetDiskFile((*new_tmp_file_obj)->GetWriteFile());
(*new_range)->SetData(reinterpret_cast<uint8_t*>(data), sizeof(int32_t));
EXPECT_OK(writer->AddWriteRange(*new_range));
@@ -394,6 +516,7 @@ TEST_F(DiskIoMgrTest, InvalidWrite) {
while (num_ranges_written_ < num_of_writes) writes_done_.Wait(lock);
}
num_ranges_written_ = 0;
+ tmp_file_grp->Close();
io_mgr.UnregisterContext(writer.get());
}
@@ -448,31 +571,37 @@ void DiskIoMgrTest::ExecuteWriteFailureTest(DiskIoMgr* io_mgr, const string& fil
LOG(ERROR) << "Error creating temp file " << file_name << " of size 100";
EXPECT_TRUE(false);
}
+ TmpFileGroup* tmp_file_grp = NewFileGroup(io_mgr);
std::unique_ptr<LocalFileSystemWithFaultInjection> fs(
new LocalFileSystemWithFaultInjection());
fs->SetWriteFaultInjection(function_name, err_no);
// DiskIoMgr takes responsibility of fs from this point.
io_mgr->SetLocalFileSystem(std::move(fs));
- AddWriteRange(num_of_writes, &data, file_name, 0, writer.get(), expected_error);
+ AddWriteRange(
+ num_of_writes, &data, file_name, 0, writer.get(), expected_error, tmp_file_grp);
{
unique_lock<mutex> lock(written_mutex_);
while (num_ranges_written_ < num_of_writes) writes_done_.Wait(lock);
}
num_ranges_written_ = 0;
+ tmp_file_grp->Close();
io_mgr->UnregisterContext(writer.get());
}
void DiskIoMgrTest::AddWriteRange(int num_of_writes, int32_t* data,
const string& file_name, int offset, RequestContext* writer,
- const string& expected_output) {
+ const string& expected_output, TmpFileGroup* file_group) {
WriteRange::WriteDoneCallback callback =
bind(mem_fn(&DiskIoMgrTest::WriteValidateCallback), this, num_of_writes,
nullptr, nullptr, nullptr, nullptr, data,
Status(TErrorCode::DISK_IO_ERROR, GetBackendString(), expected_output), _1);
WriteRange* write_range = pool_.Add(new WriteRange(file_name, offset, 0, callback));
write_range->SetData(reinterpret_cast<uint8_t*>(data), sizeof(int32_t));
+ TmpFile** new_tmp_file_obj = pool_.Add(new TmpFile*);
+ *new_tmp_file_obj = pool_.Add(new TmpFileLocal(file_group, 0, file_name));
+ write_range->SetDiskFile((*new_tmp_file_obj)->GetWriteFile());
EXPECT_OK(writer->AddWriteRange(write_range));
}
@@ -493,6 +622,7 @@ TEST_F(DiskIoMgrTest, SingleWriterCancel) {
file_size;
EXPECT_TRUE(false);
}
+ TmpFileGroup* tmp_file_grp = NewFileGroup();
scoped_ptr<DiskIoMgr> read_io_mgr(new DiskIoMgr(1, 1, 1, 1, 10));
BufferPool::ClientHandle read_client;
@@ -523,6 +653,9 @@ TEST_F(DiskIoMgrTest, SingleWriterCancel) {
*new_range = tmp_pool.Add(
new WriteRange(tmp_file, cur_offset, num_ranges % num_disks, callback));
(*new_range)->SetData(reinterpret_cast<uint8_t*>(data), sizeof(int32_t));
+ TmpFile** new_tmp_file_obj = tmp_pool.Add(new TmpFile*);
+ *new_tmp_file_obj = tmp_pool.Add(new TmpFileLocal(tmp_file_grp, 0, tmp_file));
+ (*new_range)->SetDiskFile((*new_tmp_file_obj)->GetWriteFile());
cur_offset += sizeof(int32_t);
Status add_status = writer->AddWriteRange(*new_range);
EXPECT_TRUE(add_status.code() == validate_status.code());
@@ -537,6 +670,7 @@ TEST_F(DiskIoMgrTest, SingleWriterCancel) {
}
}
+ tmp_file_grp->Close();
read_io_mgr->UnregisterContext(reader.get());
buffer_pool()->DeregisterClient(&read_client);
}
@@ -1033,6 +1167,7 @@ TEST_F(DiskIoMgrTest, MultipleReaderWriter) {
unique_ptr<BufferPool::ClientHandle[]> clients(
new BufferPool::ClientHandle[num_contexts]);
Status status;
+ TmpFileGroup* tmp_file_grp = NewFileGroup();
for (int iteration = 0; iteration < ITERATIONS; ++iteration) {
for (int threads_per_disk = 1; threads_per_disk <= 5; ++threads_per_disk) {
for (int num_disks = 1; num_disks <= 5; num_disks += 2) {
@@ -1076,6 +1211,10 @@ TEST_F(DiskIoMgrTest, MultipleReaderWriter) {
new_range->SetData(
reinterpret_cast<const uint8_t*>(data + (write_offset % strlen(data))),
1);
+ TmpFile** new_tmp_file_obj = tmp_pool.Add(new TmpFile*);
+ *new_tmp_file_obj =
+ tmp_pool.Add(new TmpFileLocal(tmp_file_grp, 0, file_name));
+ new_range->SetDiskFile((*new_tmp_file_obj)->GetWriteFile());
status = contexts[context_index]->AddWriteRange(new_range);
++write_offset;
}
@@ -1095,6 +1234,7 @@ TEST_F(DiskIoMgrTest, MultipleReaderWriter) {
} // for (int num_disks
} // for (int threads_per_disk
} // for (int iteration
+ tmp_file_grp->Close();
}
// This test will test multiple concurrent reads each reading a different file.
@@ -1575,7 +1715,8 @@ TEST_F(DiskIoMgrTest, VerifyNumThreadsParameter) {
InitRootReservation(LARGE_RESERVATION_LIMIT);
const int num_io_threads_for_remote_disks = FLAGS_num_remote_hdfs_io_threads
+ FLAGS_num_s3_io_threads + FLAGS_num_adls_io_threads + FLAGS_num_abfs_io_threads
- + FLAGS_num_ozone_io_threads;
+ + FLAGS_num_ozone_io_threads + FLAGS_num_remote_hdfs_file_oper_io_threads
+ + FLAGS_num_s3_file_oper_io_threads;
// Verify num_io_threads_per_rotational_disk and num_io_threads_per_solid_state_disk.
// Since we do not have control over which disk is used, we check for either type
@@ -1685,6 +1826,7 @@ TEST_F(DiskIoMgrTest, MetricsOfWriteSizeAndLatency) {
// Issue a number of writes to the disks.
ObjectPool tmp_pool;
+ TmpFileGroup* tmp_file_grp = NewFileGroup();
DiskIoMgr io_mgr(num_disks, 1, 1, 1, 10);
ASSERT_OK(io_mgr.Init());
unique_ptr<RequestContext> writer = io_mgr.RegisterContext();
@@ -1695,6 +1837,9 @@ TEST_F(DiskIoMgrTest, MetricsOfWriteSizeAndLatency) {
*new_range =
tmp_pool.Add(new WriteRange(tmp_file, cur_offset, i % num_disks, callback));
(*new_range)->SetData(reinterpret_cast<uint8_t*>(data), sizeof(int32_t));
+ TmpFile** new_tmp_file_obj = tmp_pool.Add(new TmpFile*);
+ *new_tmp_file_obj = tmp_pool.Add(new TmpFileLocal(tmp_file_grp, 0, tmp_file));
+ (*new_range)->SetDiskFile((*new_tmp_file_obj)->GetWriteFile());
cur_offset += sizeof(int32_t);
ASSERT_OK(writer->AddWriteRange(*new_range));
}
@@ -1726,6 +1871,7 @@ TEST_F(DiskIoMgrTest, MetricsOfWriteSizeAndLatency) {
}
num_ranges_written_ = 0;
+ tmp_file_grp->Close();
io_mgr.UnregisterContext(writer.get());
}
@@ -1740,6 +1886,7 @@ TEST_F(DiskIoMgrTest, MetricsOfWriteIoError) {
auto write_io_err = ImpaladMetrics::IO_MGR_METRICS->FindMetricForTesting<IntCounter>(
"impala-server.io-mgr.queue-0.write-io-error");
if (write_io_err != nullptr) write_io_err->SetValue(0);
+ TmpFileGroup* tmp_file_grp = NewFileGroup();
vector<string> tmp_path;
tmp_path.push_back(tmp_file);
@@ -1760,6 +1907,9 @@ TEST_F(DiskIoMgrTest, MetricsOfWriteIoError) {
};
*new_range = tmp_pool.Add(new WriteRange(tmp_file, 0, 0, callback));
(*new_range)->SetData(reinterpret_cast<uint8_t*>(data), sizeof(int32_t));
+ TmpFile** new_tmp_file_obj = tmp_pool.Add(new TmpFile*);
+ *new_tmp_file_obj = tmp_pool.Add(new TmpFileLocal(tmp_file_grp, 0, tmp_file));
+ (*new_range)->SetDiskFile((*new_tmp_file_obj)->GetWriteFile());
EXPECT_OK(writer->AddWriteRange(*new_range));
{
unique_lock<mutex> lock(written_mutex_);
@@ -1769,7 +1919,560 @@ TEST_F(DiskIoMgrTest, MetricsOfWriteIoError) {
EXPECT_EQ(write_io_err->GetValue(), 1);
num_ranges_written_ = 0;
+ tmp_file_grp->Close();
io_mgr.UnregisterContext(writer.get());
}
+
+// Issue writing operations to a remote directory.
+// Test if the temporary file can be uploaded and read correctly.
+TEST_F(DiskIoMgrTest, WriteToRemoteSuccess) {
+ InitRootReservation(LARGE_RESERVATION_LIMIT);
+ num_ranges_written_ = 0;
+ string remote_file_path = REMOTE_URL + "/tmp/test";
+ string new_file_path_local_buffer = LOCAL_BUFFER_PATH + "/test";
+ int32_t file_size = 1024;
+ FLAGS_remote_tmp_file_size = "1K";
+
+ // Delete the file if it exists.
+ hdfsDelete(hdfsConnect("default", 0), remote_file_path.c_str(), 1);
+
+ TmpFileMgr tmp_file_mgr;
+ DiskIoMgr io_mgr(1, 1, 1, 1, 10);
+ ASSERT_OK(io_mgr.Init());
+ TmpFileGroup* tmp_file_grp = NewRemoteFileGroup(&tmp_file_mgr, &io_mgr);
+ ASSERT_TRUE(tmp_file_grp != nullptr);
+
+ string key_prefix = "impala-server.io-mgr.queue-";
+ string write_size_postfix = ".write-size";
+ string write_latency_postfix = ".write-latency";
+ string i_str = std::to_string(io_mgr.RemoteDfsDiskFileOperId());
+
+ auto write_size_remote =
+ ImpaladMetrics::IO_MGR_METRICS->FindMetricForTesting<HistogramMetric>(
+ key_prefix + i_str + write_size_postfix);
+ auto write_latency_remote =
+ ImpaladMetrics::IO_MGR_METRICS->FindMetricForTesting<HistogramMetric>(
+ key_prefix + i_str + write_latency_postfix);
+ if (write_size_remote != nullptr) write_size_remote->Reset();
+ if (write_latency_remote != nullptr) write_latency_remote->Reset();
+ size_t write_size_len = sizeof(int32_t);
+
+ ObjectPool tmp_pool;
+ unique_ptr<RequestContext> io_ctx = io_mgr.RegisterContext();
+
+ TmpFileRemote** new_tmp_file_obj = tmp_pool.Add(new TmpFileRemote*);
+ *new_tmp_file_obj = tmp_pool.Add(new TmpFileRemote(tmp_file_grp, 0, remote_file_path,
+ new_file_path_local_buffer, false, REMOTE_URL.c_str()));
+
+ vector<WriteRange*> ranges;
+ vector<int32_t> datas;
+
+ (*new_tmp_file_obj)->GetWriteFile()->SetActualFileSize(file_size);
+
+ for (int i = 0; i < file_size / write_size_len; i++) {
+ int32_t* data = tmp_pool.Add(new int32_t);
+ *data = rand();
+ datas.push_back(*data);
+ WriteRange** new_range = tmp_pool.Add(new WriteRange*);
+ WriteRange::WriteDoneCallback callback = [=](const Status& status) {
+ ASSERT_EQ(0, status.code());
+ lock_guard<mutex> l(written_mutex_);
+ num_ranges_written_ = 1;
+ writes_done_.NotifyOne();
+ };
+
+ *new_range = tmp_pool.Add(new WriteRange(remote_file_path, 0, 0, callback));
+ (*new_range)->SetData(reinterpret_cast<uint8_t*>(data), sizeof(int32_t));
+ (*new_range)->SetDiskFile((*new_tmp_file_obj)->GetWriteFile());
+ ranges.push_back(*new_range);
+ EXPECT_OK(io_ctx->AddWriteRange(*new_range));
+ {
+ unique_lock<mutex> lock(written_mutex_);
+ while (num_ranges_written_ < 1) writes_done_.Wait(lock);
+ }
+ num_ranges_written_ = 0;
+ if (i == file_size / write_size_len - 1) {
+ (*new_tmp_file_obj)->SetAtCapacity();
+ }
+ }
+
+ EXPECT_TRUE(FileExist(new_file_path_local_buffer));
+
+ // Do upload the file.
+ auto disk_id = io_mgr.RemoteDfsDiskFileOperId();
+ RemoteOperRange::RemoteOperDoneCallback u_callback = [](const Status& upload_status) {};
+ RemoteOperRange* upload_range = tmp_pool.Add(new RemoteOperRange(
+ (*new_tmp_file_obj)->DiskBufferFile(), (*new_tmp_file_obj)->DiskFile(), file_size,
+ disk_id, RequestType::FILE_UPLOAD, &io_mgr, u_callback));
+ EXPECT_OK(io_ctx->AddRemoteOperRange(upload_range));
+
+ int wait_times = 10;
+ while (true) {
+ if ((*new_tmp_file_obj)->DiskFile()->GetFileStatus() == DiskFileStatus::PERSISTED) {
+ break;
+ }
+ // Suppose the upload should be finished in two seconds.
+ ASSERT_TRUE(wait_times-- > 0);
+ usleep(200 * 1000);
+ }
+
+ EXPECT_TRUE(HdfsFileExist(remote_file_path));
+
+ auto exam_fuc = [&](HistogramMetric* metric, const string& keyname) {
+ uint64_t total_cnt = metric->TotalCount();
+ uint64_t min_value = metric->MinValue();
+ uint64_t max_value = metric->MaxValue();
+ if (keyname == "write_latency") {
+ // Write latency counts not only file writing, but also file close.
+ // Since it is supposed to be slow to close remote files, due to the
+ // close is doing at least a part of upload, so we expect the total
+ // count is 2 here.
+ EXPECT_EQ(total_cnt, 2);
+ } else {
+ // The block size is the same as the file size, so writing is finished
+ // by one time file writing.
+ EXPECT_EQ(total_cnt, 1);
+ }
+ // Check if the min and max of write size are the same as the written len.
+ if (keyname == "write_size") {
+ EXPECT_EQ(min_value, file_size);
+ EXPECT_EQ(max_value, file_size);
+ }
+ };
+
+ exam_fuc(write_size_remote, "write_size");
+ exam_fuc(write_latency_remote, "write_latency");
+
+ // Bogus value
+ int64_t mtime = 100000;
+
+ // Test reading from local buffer, since the scratch file is in
+ // the local file system.
+ for (int i = 0; i < ranges.size(); i++) {
+ ScanRange* scan_range = tmp_pool.Add(new ScanRange);
+ auto range = ranges.at(i);
+ auto data = datas.at(i);
+ size_t buffer_len = sizeof(int32_t);
+ vector<uint8_t> client_buffer(buffer_len);
+ scan_range->Reset(hdfsConnect("default", 0), range->file(), range->len(),
+ range->offset(), 0, false, mtime,
+ BufferOpts::ReadInto(client_buffer.data(), buffer_len, BufferOpts::NO_CACHING),
+ nullptr, (*new_tmp_file_obj)->DiskFile(), (*new_tmp_file_obj)->DiskBufferFile());
+ bool needs_buffers;
+ ASSERT_OK(io_ctx->StartScanRange(scan_range, &needs_buffers));
+ unique_ptr<BufferDescriptor> io_buffer;
+ ASSERT_OK(scan_range->GetNext(&io_buffer));
+ ASSERT_TRUE(io_buffer->eosr());
+ EXPECT_EQ(range->len(), io_buffer->len());
+ EXPECT_EQ(client_buffer.data(), io_buffer->buffer());
+ EXPECT_EQ(*(int32_t*)client_buffer.data(), data);
+ EXPECT_TRUE(scan_range->use_local_buffer());
+ scan_range->ReturnBuffer(move(io_buffer));
+ }
+
+ // Delete the local buffer file, try to read from the remote.
+ (*new_tmp_file_obj)->DiskBufferFile()->SetStatus(io::DiskFileStatus::DELETED);
+ EXPECT_OK(FileSystemUtil::RemovePaths({(*new_tmp_file_obj)->DiskBufferFile()->path()}));
+
+ for (int i = 0; i < ranges.size(); i++) {
+ ScanRange* scan_range = tmp_pool.Add(new ScanRange);
+ auto range = ranges.at(i);
+ auto data = datas.at(i);
+ size_t buffer_len = sizeof(int32_t);
+ vector<uint8_t> client_buffer(buffer_len);
+ scan_range->Reset(hdfsConnect("default", 0), range->file(), range->len(),
+ range->offset(), 0, false, mtime,
+ BufferOpts::ReadInto(client_buffer.data(), buffer_len, BufferOpts::NO_CACHING),
+ nullptr, (*new_tmp_file_obj)->DiskFile(), (*new_tmp_file_obj)->DiskBufferFile());
+ bool needs_buffers;
+ ASSERT_OK(io_ctx->StartScanRange(scan_range, &needs_buffers));
+ unique_ptr<BufferDescriptor> io_buffer;
+ ASSERT_OK(scan_range->GetNext(&io_buffer));
+ ASSERT_TRUE(io_buffer->eosr());
+ EXPECT_EQ(range->len(), io_buffer->len());
+ EXPECT_EQ(client_buffer.data(), io_buffer->buffer());
+ EXPECT_EQ(*(int32_t*)client_buffer.data(), data);
+ EXPECT_FALSE(scan_range->use_local_buffer());
+ scan_range->ReturnBuffer(move(io_buffer));
+ }
+
+ num_ranges_written_ = 0;
+ tmp_file_grp->Close();
+ io_mgr.UnregisterContext(io_ctx.get());
+}
+
+// Issue a writing operation to a remote directory.
+// Test if it can be read correctly when the temporary file is not full and not uploaded.
+TEST_F(DiskIoMgrTest, WriteToRemotePartialFileSuccess) {
+ InitRootReservation(LARGE_RESERVATION_LIMIT);
+ num_ranges_written_ = 0;
+ string remote_file_path = REMOTE_URL + "/tmp/test";
+ string new_file_path_local_buffer = LOCAL_BUFFER_PATH + "/test";
+ FLAGS_remote_tmp_file_size = "1K";
+
+ // Delete the file in hdfs if it exists.
+ hdfsDelete(hdfsConnect("default", 0), remote_file_path.c_str(), 1);
+
+ // Delete the file in local file system
+ vector<string> tmp_path;
+ tmp_path.push_back(new_file_path_local_buffer);
+ Status rm_status = FileSystemUtil::RemovePaths(tmp_path);
+
+ TmpFileMgr tmp_file_mgr;
+ DiskIoMgr io_mgr(1, 1, 1, 1, 10);
+ ASSERT_OK(io_mgr.Init());
+ TmpFileGroup* tmp_file_grp = NewRemoteFileGroup(&tmp_file_mgr, &io_mgr);
+ ASSERT_TRUE(tmp_file_grp != nullptr);
+
+ string metric_name = "tmp-file-mgr.local-buff-bytes-used.dir-0";
+
+ ObjectPool tmp_pool;
+ unique_ptr<RequestContext> io_ctx = io_mgr.RegisterContext();
+
+ TmpFileRemote** new_tmp_file_obj = tmp_pool.Add(new TmpFileRemote*);
+ *new_tmp_file_obj = tmp_pool.Add(new TmpFileRemote(tmp_file_grp, 0, remote_file_path,
+ new_file_path_local_buffer, false, REMOTE_URL.c_str()));
+
+ int32_t* data = tmp_pool.Add(new int32_t);
+ *data = rand();
+ WriteRange** new_range = tmp_pool.Add(new WriteRange*);
+ WriteRange::WriteDoneCallback callback = [=](const Status& status) {
+ ASSERT_EQ(0, status.code());
+ lock_guard<mutex> l(written_mutex_);
+ num_ranges_written_ = 1;
+ writes_done_.NotifyOne();
+ };
+
+ *new_range = tmp_pool.Add(new WriteRange(remote_file_path, 0, 0, callback));
+ (*new_range)->SetData(reinterpret_cast<uint8_t*>(data), sizeof(int32_t));
+ (*new_range)->SetDiskFile((*new_tmp_file_obj)->GetWriteFile());
+ EXPECT_OK(io_ctx->AddWriteRange(*new_range));
+ {
+ unique_lock<mutex> lock(written_mutex_);
+ while (num_ranges_written_ < 1) writes_done_.Wait(lock);
+ }
+ num_ranges_written_ = 0;
+ (*new_tmp_file_obj)->SetAtCapacity();
+
+ EXPECT_TRUE(FileExist(new_file_path_local_buffer));
+ EXPECT_FALSE(HdfsFileExist(remote_file_path));
+
+ // Bogus value
+ int64_t mtime = 100000;
+
+ // Test reading from remote scatch space.
+ ScanRange* scan_range = tmp_pool.Add(new ScanRange);
+ size_t buffer_len = sizeof(int32_t);
+ vector<uint8_t> client_buffer(buffer_len);
+ scan_range->Reset(hdfsConnect("default", 0), (*new_range)->file(), (*new_range)->len(),
+ (*new_range)->offset(), 0, false, mtime,
+ BufferOpts::ReadInto(client_buffer.data(), buffer_len, BufferOpts::NO_CACHING),
+ nullptr, (*new_tmp_file_obj)->DiskFile(), (*new_tmp_file_obj)->DiskBufferFile());
+ bool needs_buffers;
+ ASSERT_OK(io_ctx->StartScanRange(scan_range, &needs_buffers));
+ unique_ptr<BufferDescriptor> io_buffer;
+ ASSERT_OK(scan_range->GetNext(&io_buffer));
+ ASSERT_TRUE(io_buffer->eosr());
+ EXPECT_EQ((*new_range)->len(), io_buffer->len());
+ EXPECT_EQ(client_buffer.data(), io_buffer->buffer());
+ EXPECT_EQ(*(int32_t*)client_buffer.data(), *data);
+ scan_range->ReturnBuffer(move(io_buffer));
+
+ num_ranges_written_ = 0;
+ tmp_file_grp->Close();
+ io_mgr.UnregisterContext(io_ctx.get());
+}
+
+// Upload an non-existent file and failed.
+TEST_F(DiskIoMgrTest, WriteToRemoteUploadFailed) {
+ InitRootReservation(LARGE_RESERVATION_LIMIT);
+ num_oper_ = 0;
+ string remote_file_path = REMOTE_URL + "/tmp/test";
+ string non_existent_dir = "/non-existent-dir/test";
+ FLAGS_remote_tmp_file_size = "1K";
+ int64_t file_size = 1024;
+ int64_t block_size = 1024;
+
+ // Delete the file if it exists.
+ hdfsDelete(hdfsConnect("default", 0), remote_file_path.c_str(), 1);
+
+ TmpFileMgr tmp_file_mgr;
+ DiskIoMgr io_mgr(1, 1, 1, 1, 10);
+ ASSERT_OK(io_mgr.Init());
+ TmpFileGroup* tmp_file_grp = NewRemoteFileGroup(&tmp_file_mgr, &io_mgr);
+ ASSERT_TRUE(tmp_file_grp != nullptr);
+
+ ObjectPool tmp_pool;
+ unique_ptr<RequestContext> io_ctx = io_mgr.RegisterContext();
+
+ TmpFileRemote** new_tmp_file_obj = tmp_pool.Add(new TmpFileRemote*);
+ *new_tmp_file_obj = tmp_pool.Add(new TmpFileRemote(
+ tmp_file_grp, 0, remote_file_path, non_existent_dir, false, REMOTE_URL.c_str()));
+
+ DiskFile* remote_file = (*new_tmp_file_obj)->DiskFile();
+ DiskFile* local_buffer_file = (*new_tmp_file_obj)->DiskBufferFile();
+ local_buffer_file->SetStatus(DiskFileStatus::PERSISTED);
+ local_buffer_file->SetActualFileSize(file_size);
+
+ auto disk_id = io_mgr.RemoteDfsDiskFileOperId();
+ RemoteOperRange::RemoteOperDoneCallback callback = [=](const Status& status) {
+ ASSERT_NE(0, status.code());
+ lock_guard<mutex> l(oper_mutex_);
+ num_oper_ = 1;
+ oper_done_.NotifyOne();
+ };
+
+ auto oper_range = tmp_pool.Add(new RemoteOperRange(local_buffer_file, remote_file,
+ block_size, disk_id, RequestType::FILE_UPLOAD, &io_mgr, callback));
+ Status add_status = io_ctx->AddRemoteOperRange(oper_range);
+ ASSERT_OK(add_status);
+
+ {
+ unique_lock<mutex> lock(oper_mutex_);
+ while (num_oper_ < 1) oper_done_.Wait(lock);
+ }
+
+ // None of the files should exist.
+ EXPECT_FALSE(FileExist(non_existent_dir));
+ EXPECT_FALSE(HdfsFileExist(remote_file_path));
+
+ num_oper_ = 0;
+ tmp_file_grp->Close();
+ io_mgr.UnregisterContext(io_ctx.get());
+}
+
+// Upload file to the remote successfully, then evict the local buffer.
+TEST_F(DiskIoMgrTest, WriteToRemoteEvictLocal) {
+ InitRootReservation(LARGE_RESERVATION_LIMIT);
+ num_ranges_written_ = 0;
+ string remote_file_path = REMOTE_URL + "/tmp/test1";
+ string local_buffer_file_path = LOCAL_BUFFER_PATH + "/test1";
+ int32_t file_size = 1024;
+ FLAGS_remote_tmp_file_size = "1K";
+
+ // Delete the file if it exists.
+ hdfsDelete(hdfsConnect("default", 0), remote_file_path.c_str(), 1);
+
+ TmpFileMgr tmp_file_mgr;
+ DiskIoMgr io_mgr(1, 1, 1, 1, 10);
+ ASSERT_OK(io_mgr.Init());
+ TmpFileGroup* tmp_file_grp = NewRemoteFileGroup(&tmp_file_mgr, &io_mgr);
+ ASSERT_TRUE(tmp_file_grp != nullptr);
+
+ size_t write_size_len = sizeof(int32_t);
+
+ ObjectPool tmp_pool;
+ unique_ptr<RequestContext> io_ctx = io_mgr.RegisterContext();
+
+ TmpFileRemote* new_tmp_file_obj = new TmpFileRemote(tmp_file_grp, 0, remote_file_path,
+ local_buffer_file_path, false, REMOTE_URL.c_str());
+ shared_ptr<TmpFileRemote> shared_tmp_file;
+ shared_tmp_file.reset(move(new_tmp_file_obj));
+
+ vector<WriteRange*> ranges;
+ vector<int32_t> datas;
+
+ // Set the actual file size which is needed by the upload process.
+ shared_tmp_file->GetWriteFile()->SetActualFileSize(file_size);
+
+ for (int i = 0; i < file_size / write_size_len; i++) {
+ int32_t* data = tmp_pool.Add(new int32_t);
+ *data = rand();
+ datas.push_back(*data);
+ WriteRange** new_range = tmp_pool.Add(new WriteRange*);
+ WriteRange::WriteDoneCallback callback = [=](const Status& status) {
+ ASSERT_EQ(0, status.code());
+ lock_guard<mutex> l(written_mutex_);
+ num_ranges_written_ = 1;
+ writes_done_.NotifyOne();
+ };
+
+ *new_range = tmp_pool.Add(new WriteRange(remote_file_path, 0, 0, callback));
+ (*new_range)->SetData(reinterpret_cast<uint8_t*>(data), sizeof(int32_t));
+ (*new_range)->SetDiskFile(shared_tmp_file->GetWriteFile());
+ ranges.push_back(*new_range);
+ EXPECT_OK(io_ctx->AddWriteRange(*new_range));
+ {
+ unique_lock<mutex> lock(written_mutex_);
+ while (num_ranges_written_ < 1) writes_done_.Wait(lock);
+ }
+ num_ranges_written_ = 0;
+ if (i == file_size / write_size_len - 1) shared_tmp_file->SetAtCapacity();
+ }
+
+ EXPECT_TRUE(FileExist(local_buffer_file_path));
+
+ // Do upload the file.
+ RemoteOperRange::RemoteOperDoneCallback u_callback = [&](const Status& upload_status) {
+ if (upload_status.ok()) {
+ auto shared_tmp_file_ptr = std::dynamic_pointer_cast<TmpFile>(shared_tmp_file);
+ tmp_file_mgr.EnqueueTmpFilesPool(shared_tmp_file_ptr, true);
+ }
+ };
+ RemoteOperRange* upload_range = tmp_pool.Add(
+ new RemoteOperRange(shared_tmp_file->DiskBufferFile(), shared_tmp_file->DiskFile(),
+ file_size, 0, RequestType::FILE_UPLOAD, &io_mgr, u_callback));
+ Status add_status = io_ctx->AddRemoteOperRange(upload_range);
+
+ int wait_times = 10;
+ while (true) {
+ if (shared_tmp_file->DiskFile()->GetFileStatus() == DiskFileStatus::PERSISTED) {
+ break;
+ }
+ // Suppose the upload should be finished in two seconds.
+ ASSERT_TRUE(wait_times-- > 0);
+ usleep(200 * 1000);
+ }
+
+ // TryEvictFile and the local buffer file should be evicted for releasing local
+ // scratch space.
+ std::shared_ptr<TmpFile> tmp_file;
+ ASSERT_OK(tmp_file_mgr.DequeueTmpFilesPool(&tmp_file, false));
+ ASSERT_TRUE(tmp_file != nullptr);
+ Status try_evict_status = tmp_file_mgr.TryEvictFile(tmp_file.get());
+ ASSERT_TRUE(try_evict_status.ok());
+ EXPECT_TRUE(HdfsFileExist(remote_file_path));
+ EXPECT_FALSE(FileExist(local_buffer_file_path));
+ EXPECT_EQ(shared_tmp_file->DiskBufferFile()->GetFileStatus(), DiskFileStatus::DELETED);
+
+ num_ranges_written_ = 0;
+ tmp_file_grp->Close();
+ io_mgr.UnregisterContext(io_ctx.get());
+}
+
+// Use an invalid block size to emulate the case when memory allocation failed.
+TEST_F(DiskIoMgrTest, WriteToRemoteFailMallocBlock) {
+ num_ranges_written_ = 0;
+ string remote_file_path = REMOTE_URL + "/tmp/test1";
+ string local_buffer_file_path = LOCAL_BUFFER_PATH + "/test1";
+ int64_t invalid_block_size = -1;
+
+ TmpFileMgr tmp_file_mgr;
+ DiskIoMgr io_mgr(1, 1, 1, 1, 10);
+ ASSERT_OK(io_mgr.Init());
+ TmpFileGroup* tmp_file_grp = NewRemoteFileGroup(&tmp_file_mgr, &io_mgr);
+ ASSERT_TRUE(tmp_file_grp != nullptr);
+
+ ObjectPool tmp_pool;
+ unique_ptr<RequestContext> io_ctx = io_mgr.RegisterContext();
+
+ TmpFileRemote** new_tmp_file_obj = tmp_pool.Add(new TmpFileRemote*);
+ *new_tmp_file_obj = tmp_pool.Add(new TmpFileRemote(tmp_file_grp, 0, remote_file_path,
+ local_buffer_file_path, false, REMOTE_URL.c_str()));
+
+ RemoteOperRange::RemoteOperDoneCallback u_callback = [this](
+ const Status& upload_status) {
+ // Assert the upload failed due to a failed malloc with an invalid block size.
+ ASSERT_TRUE(!upload_status.ok());
+ lock_guard<mutex> l(oper_mutex_);
+ num_ranges_written_ = 1;
+ oper_done_.NotifyOne();
+ };
+
+ auto disk_id = io_mgr.RemoteDfsDiskFileOperId();
+ std::unique_ptr<io::RemoteOperRange> upload_range;
+ upload_range.reset(new RemoteOperRange(nullptr, nullptr, invalid_block_size, disk_id,
+ RequestType::FILE_UPLOAD, &io_mgr, u_callback));
+ Status add_status = io_ctx->AddRemoteOperRange(upload_range.get());
+ ASSERT_TRUE(add_status.ok());
+
+ {
+ unique_lock<mutex> lock(oper_mutex_);
+ while (num_ranges_written_ < 1) oper_done_.Wait(lock);
+ }
+
+ num_ranges_written_ = 0;
+ tmp_file_grp->Close();
+ io_mgr.UnregisterContext(io_ctx.get());
+}
+
+// Issue writing operations with different page sizes, the actual file size is
+// over the default file size. Test if the upload is successful.
+TEST_F(DiskIoMgrTest, WriteToRemoteDiffPagesSuccess) {
+ InitRootReservation(LARGE_RESERVATION_LIMIT);
+ num_ranges_written_ = 0;
+ string remote_file_path = REMOTE_URL + "/tmp/test";
+ string new_file_path_local_buffer = LOCAL_BUFFER_PATH + "/test";
+ int32_t block_size = 1024;
+ FLAGS_remote_tmp_file_size = "1K";
+ const int32_t page_size[2] = {768, 512};
+ char page_1[page_size[0]];
+ char page_2[page_size[1]];
+ char* page[2] = {page_1, page_2};
+ int32_t actual_size = page_size[0] + page_size[1];
+
+ // Delete the file if it exists.
+ hdfsDelete(hdfsConnect("default", 0), remote_file_path.c_str(), 1);
+
+ TmpFileMgr tmp_file_mgr;
+ DiskIoMgr io_mgr(1, 1, 1, 1, 10);
+ ASSERT_OK(io_mgr.Init());
+ TmpFileGroup* tmp_file_grp = NewRemoteFileGroup(&tmp_file_mgr, &io_mgr);
+ ASSERT_TRUE(tmp_file_grp != nullptr);
+
+ ObjectPool tmp_pool;
+ unique_ptr<RequestContext> io_ctx = io_mgr.RegisterContext();
+
+ TmpFileRemote** new_tmp_file_obj = tmp_pool.Add(new TmpFileRemote*);
+ *new_tmp_file_obj = tmp_pool.Add(new TmpFileRemote(tmp_file_grp, 0, remote_file_path,
+ new_file_path_local_buffer, false, REMOTE_URL.c_str()));
+
+ WriteRange::WriteDoneCallback callback = [=](const Status& status) {
+ ASSERT_EQ(0, status.code());
+ lock_guard<mutex> l(written_mutex_);
+ num_ranges_written_ = 1;
+ writes_done_.NotifyOne();
+ };
+
+ // Add pages, and writes them to the local buffer.
+ for (int i = 0; i < 2; i++) {
+ if (i == 1) {
+ (*new_tmp_file_obj)->SetAtCapacity();
+ (*new_tmp_file_obj)->GetWriteFile()->SetActualFileSize(actual_size);
+ }
+ WriteRange** new_range = tmp_pool.Add(new WriteRange*);
+ *new_range = tmp_pool.Add(new WriteRange(remote_file_path, 0, 0, callback));
+ (*new_range)->SetData(reinterpret_cast<uint8_t*>(page[i]), page_size[i]);
+ (*new_range)->SetDiskFile((*new_tmp_file_obj)->GetWriteFile());
+ EXPECT_OK(io_ctx->AddWriteRange(*new_range));
+ {
+ unique_lock<mutex> lock(written_mutex_);
+ while (num_ranges_written_ < 1) writes_done_.Wait(lock);
+ }
+ num_ranges_written_ = 0;
+ }
+
+ EXPECT_TRUE(FileExist(new_file_path_local_buffer));
+ struct stat stat_val;
+ int rc = stat(new_file_path_local_buffer.c_str(), &stat_val);
+ ASSERT_TRUE(rc == 0);
+ EXPECT_EQ(stat_val.st_size, actual_size);
+
+ // Do upload the local buffer file.
+ auto disk_id = io_mgr.RemoteDfsDiskFileOperId();
+ RemoteOperRange::RemoteOperDoneCallback u_callback = [](const Status& upload_status) {};
+ RemoteOperRange* upload_range = tmp_pool.Add(new RemoteOperRange(
+ (*new_tmp_file_obj)->DiskBufferFile(), (*new_tmp_file_obj)->DiskFile(), block_size,
+ disk_id, RequestType::FILE_UPLOAD, &io_mgr, u_callback));
+ EXPECT_OK(io_ctx->AddRemoteOperRange(upload_range));
+
+ int wait_times = 10;
+ while (true) {
+ if ((*new_tmp_file_obj)->DiskFile()->GetFileStatus() == DiskFileStatus::PERSISTED) {
+ break;
+ }
+ // Suppose the upload should be finished in two seconds.
+ ASSERT_TRUE(wait_times-- > 0);
+ usleep(200 * 1000);
+ }
+
+ // Assert remote file actual size is the same as the local actual size.
+ EXPECT_TRUE(HdfsFileExist(remote_file_path));
+ ASSERT_EQ((*new_tmp_file_obj)->DiskFile()->actual_file_size(), actual_size);
+
+ num_ranges_written_ = 0;
+ tmp_file_grp->Close();
+ io_mgr.UnregisterContext(io_ctx.get());
+}
}
}
diff --git a/be/src/runtime/io/disk-io-mgr.cc b/be/src/runtime/io/disk-io-mgr.cc
index 30369bc..f963de4 100644
--- a/be/src/runtime/io/disk-io-mgr.cc
+++ b/be/src/runtime/io/disk-io-mgr.cc
@@ -18,14 +18,22 @@
#include "runtime/io/disk-io-mgr.h"
#include "common/global-flags.h"
+#include "common/names.h"
#include "common/thread-debug-info.h"
#include "runtime/exec-env.h"
+#include "runtime/hdfs-fs-cache.h"
#include "runtime/io/data-cache.h"
+#include "runtime/io/disk-file.h"
#include "runtime/io/disk-io-mgr-internal.h"
-#include "runtime/io/handle-cache.inline.h"
#include "runtime/io/error-converter.h"
+#include "runtime/io/file-writer.h"
+#include "runtime/io/handle-cache.inline.h"
#include <boost/algorithm/string.hpp>
+#include <boost/filesystem.hpp>
+#include <boost/lexical_cast.hpp>
+#include <boost/uuid/random_generator.hpp>
+#include <boost/uuid/uuid_io.hpp>
#include "gutil/strings/substitute.h"
#include "util/bit-util.h"
@@ -35,6 +43,7 @@
#include "util/hdfs-util.h"
#include "util/histogram-metric.h"
#include "util/metrics.h"
+#include "util/os-util.h"
#include "util/test-info.h"
#include "util/time.h"
@@ -48,6 +57,9 @@ using namespace impala;
using namespace impala::io;
using namespace strings;
+using boost::shared_mutex;
+using boost::filesystem::path;
+using boost::uuids::random_generator;
using std::to_string;
// Control the number of disks on the machine. If 0, this comes from the system
@@ -94,12 +106,21 @@ DEFINE_int32(num_io_threads_per_solid_state_disk, 0,
// across the local disk queues.
DEFINE_int32(num_remote_hdfs_io_threads, 8, "Number of remote HDFS I/O threads");
+// The maximum number of HDFS I/O threads for doing file operations, such as file
+// uploading. The default value of 8 was chosen empirically to maximize HDFS throughput.
+DEFINE_int32(num_remote_hdfs_file_oper_io_threads, 8,
+ "Number of remote HDFS file operations I/O threads");
+
// The maximum number of S3 I/O threads. The default value of 16 was chosen emperically
// to maximize S3 throughput. Maximum throughput is achieved with multiple connections
// open to S3 and use of multiple CPU cores since S3 reads are relatively compute
// expensive (SSL and JNI buffer overheads).
DEFINE_int32(num_s3_io_threads, 16, "Number of S3 I/O threads");
+// The maximum number of S3 I/O threads for doing file operations, such as file
+// uploading. The default value of 16 was chosen empirically to maximize S3 throughput.
+DEFINE_int32(num_s3_file_oper_io_threads, 16, "Number of S3 file operations I/O threads");
+
// The maximum number of ABFS I/O threads. TODO: choose the default empirically.
DEFINE_int32(num_abfs_io_threads, 16, "Number of ABFS I/O threads");
@@ -203,6 +224,161 @@ void WriteRange::SetData(const uint8_t* buffer, int64_t len) {
len_ = len;
}
+void WriteRange::SetOffset(int64_t file_offset) {
+ offset_ = file_offset;
+}
+
+Status WriteRange::DoWrite() {
+ Status ret_status = Status::OK();
+ Status close_status = Status::OK();
+ DiskQueue* queue = io_ctx_->parent_->disk_queues_[disk_id_];
+ int64_t written_bytes = 0;
+ FileWriter* file_writer = disk_file_->GetFileWriter();
+ /// DoWrite is used for Spilling.
+ /// For spilling to local, we will open and close the file handle for
+ /// each write range, which is the WriteOne() function, since it could
+ /// be a random write in each file.
+ /// For spilling to remote, we will keep the file handle open, until the
+ /// write range is the last one of the file to do the sequential write.
+ if (disk_file_->disk_type() == DiskFileType::LOCAL) {
+ return file_writer->WriteOne(this);
+ }
+ {
+ ScopedHistogramTimer write_timer(queue->write_latency());
+ shared_lock<shared_mutex> lock(disk_file_->physical_file_lock_);
+ ret_status = file_writer->Open();
+ if (!ret_status.ok()) return DoWriteEnd(queue, ret_status);
+ ret_status = file_writer->Write(this, &written_bytes);
+ int64_t actual_file_size = disk_file_->actual_file_size();
+ // actual_file_size is only set once, otherwise it is 0 by default. If it is still
+ // not set, it is impossible to be full.
+ if (actual_file_size != 0) {
+ DCHECK_LE(written_bytes, disk_file_->actual_file_size());
+ is_full_ = written_bytes == actual_file_size;
+ } else {
+ // If the actual size hasn't been set, the written bytes must be less than the
+ // default file size.
+ DCHECK_LT(written_bytes, disk_file_->file_size());
+ }
+ if (is_full_) close_status = file_writer->Close();
+ if (ret_status.ok() && !close_status.ok()) ret_status = close_status;
+ if (ret_status.ok() && is_full_) {
+ // If the file is full, the file handle should be closed,
+ // so set the file to persisted status.
+ disk_file_->SetStatus(io::DiskFileStatus::PERSISTED);
+ }
+ }
+ return DoWriteEnd(queue, ret_status);
+}
+
+Status WriteRange::DoWriteEnd(DiskQueue* queue, const Status& ret_status) {
+ if (ret_status.ok()) {
+ queue->write_size()->Update(len());
+ } else {
+ queue->write_io_err()->Increment(1);
+ }
+ return ret_status;
+}
+
+RemoteOperRange::RemoteOperRange(DiskFile* src_file, DiskFile* dst_file,
+ int64_t block_size, int disk_id, RequestType::type type, DiskIoMgr* io_mgr,
+ RemoteOperDoneCallback callback)
+ : RequestRange(type, disk_id),
+ callback_(callback),
+ io_mgr_(io_mgr),
+ disk_file_src_(src_file),
+ disk_file_dst_(dst_file),
+ block_size_(block_size) {}
+
+Status RemoteOperRange::DoOper(uint8_t* buffer, int64_t buffer_size) {
+ DCHECK(request_type() == RequestType::FILE_UPLOAD);
+ return DoUpload(buffer, buffer_size);
+}
+
+Status RemoteOperRange::DoUpload(uint8_t* buffer, int64_t buffer_size) {
+ DCHECK(disk_file_src_ != nullptr);
+ DCHECK(disk_file_dst_ != nullptr);
+ hdfsFS hdfs_conn = disk_file_dst_->hdfs_conn_;
+ int64_t file_size = disk_file_src_->actual_file_size_.Load();
+ DCHECK(hdfs_conn != nullptr);
+ DCHECK(file_size != 0);
+ const char* remote_file_path = disk_file_dst_->path().c_str();
+ const char* local_file_path = disk_file_src_->path().c_str();
+ DiskQueue* queue = io_mgr_->disk_queues_[disk_id_];
+ Status status = Status::OK();
+ int64_t ret, offset = 0;
+ FILE* local_file = nullptr;
+
+ // To get the shared lock to protect the physical files from deleting during the
+ // upload. The sequence is to acquire the local file lock, then remote file lock,
+ // or it might cause deadlocks.
+ shared_lock<shared_mutex> srcl(disk_file_src_->physical_file_lock_);
+ shared_lock<shared_mutex> dstl(disk_file_dst_->physical_file_lock_);
+
+ // If it is not persisted, the cases could be the query is cancelled or finished.
+ auto src_status = disk_file_src_->GetFileStatus();
+ if (src_status != io::DiskFileStatus::PERSISTED) {
+ DCHECK(src_status == io::DiskFileStatus::DELETED);
+ return Status::OK();
+ }
+
+ RETURN_IF_ERROR(io_mgr_->local_file_system_->OpenForRead(
+ local_file_path, O_RDONLY, S_IRUSR | S_IWUSR, &local_file));
+ hdfsFile remote_hdfs_file =
+ hdfsOpenFile(hdfs_conn, remote_file_path, O_WRONLY, 0, 0, buffer_size);
+
+ if (remote_hdfs_file == nullptr) {
+ status = Status(TErrorCode::DISK_IO_ERROR, GetBackendString(),
+ Substitute("Could not open file: $0: $1", remote_file_path, GetStrErrMsg()));
+ goto end;
+ }
+
+ /// Read the blocks from the local buffer file and write the blocks
+ /// to the remote file.
+ while (file_size != offset) {
+ int bytes = min(file_size - offset, buffer_size);
+ status =
+ io_mgr_->local_file_system_->Fread(local_file, buffer, bytes, local_file_path);
+ if (!status.ok()) goto end;
+ {
+ ScopedHistogramTimer write_timer(queue->write_latency());
+ // Write local buffer to the remote file.
+ ret = hdfsWrite(hdfs_conn, remote_hdfs_file, buffer, bytes);
+ }
+ if (ret == -1 || ret != bytes) {
+ queue->write_io_err()->Increment(1);
+ status = Status(TErrorCode::DISK_IO_ERROR, GetBackendString(),
+ Substitute("Error writing to $0 in file: $1 $2", offset, remote_file_path,
+ GetHdfsErrorMsg("")));
+ goto end;
+ }
+ offset += bytes;
+ queue->write_size()->Update(bytes);
+ }
+
+end:
+ if (remote_hdfs_file != nullptr) {
+ // We count file close as write latency, as it takes time for
+ // S3 to close to finish the upload procedure.
+ ScopedHistogramTimer write_timer(queue->write_latency());
+ if (hdfsCloseFile(hdfs_conn, remote_hdfs_file) != 0) {
+ // Try to close the local file if error happens.
+ RETURN_IF_ERROR(io_mgr_->local_file_system_->Fclose(local_file, local_file_path));
+ return Status(TErrorCode::DISK_IO_ERROR, GetBackendString(),
+ Substitute(
+ "Failed to close HDFS file: $0", remote_file_path, GetHdfsErrorMsg("")));
+ }
+ }
+ RETURN_IF_ERROR(io_mgr_->local_file_system_->Fclose(local_file, local_file_path));
+ if (status.ok()) {
+ disk_file_dst_->SetStatus(io::DiskFileStatus::PERSISTED);
+ disk_file_dst_->SetActualFileSize(file_size);
+ } else {
+ LOG(WARNING) << "File upload failed, msg:" << status.msg().msg();
+ }
+ return status;
+}
+
static void CheckSseSupport() {
if (!CpuInfo::IsSupported(CpuInfo::SSE4_2)) {
LOG(WARNING) << "This machine does not support sse4_2. The default IO system "
@@ -292,6 +468,12 @@ Status DiskIoMgr::Init() {
} else if (i == RemoteOzoneDiskId()) {
num_threads_per_disk = FLAGS_num_ozone_io_threads;
device_name = "Ozone remote";
+ } else if (i == RemoteDfsDiskFileOperId()) {
+ num_threads_per_disk = FLAGS_num_remote_hdfs_file_oper_io_threads;
+ device_name = "HDFS remote file operations";
+ } else if (i == RemoteS3DiskFileOperId()) {
+ num_threads_per_disk = FLAGS_num_s3_file_oper_io_threads;
+ device_name = "S3 remote file operations";
} else if (DiskInfo::is_rotational(i)) {
num_threads_per_disk = num_io_threads_per_rotational_disk_;
// During tests, i may not point to an existing disk.
@@ -496,7 +678,8 @@ int64_t DiskIoMgr::ComputeIdealBufferReservation(int64_t scan_range_len) {
// work is available or the thread is shut down.
// Work is available if there is a RequestContext with
// - A ScanRange with a buffer available, or
-// - A WriteRange in unstarted_write_ranges_.
+// - A WriteRange in unstarted_write_ranges_ or
+// - A RemoteOperRange in unstarted_remote_upload_ranges_
RequestRange* DiskQueue::GetNextRequestRange(RequestContext** request_context) {
// This loops returns either with work to do or when the disk IoMgr shuts down.
while (true) {
@@ -545,13 +728,41 @@ void DiskQueue::DiskThreadLoop(DiskIoMgr* io_mgr) {
ScopedThreadContext tdi_scope(GetThreadDebugInfo(), worker_context->query_id(),
worker_context->instance_id());
- if (range->request_type() == RequestType::READ) {
- ScanRange* scan_range = static_cast<ScanRange*>(range);
- ReadOutcome outcome = scan_range->DoRead(this, disk_id_);
- worker_context->ReadDone(disk_id_, outcome, scan_range);
- } else {
- DCHECK(range->request_type() == RequestType::WRITE);
- io_mgr->Write(worker_context, static_cast<WriteRange*>(range));
+ switch (range->request_type()) {
+ case RequestType::READ: {
+ ScanRange* scan_range = static_cast<ScanRange*>(range);
+ ReadOutcome outcome = scan_range->DoRead(this, disk_id_);
+ worker_context->ReadDone(disk_id_, outcome, scan_range);
+ break;
+ }
+ case RequestType::WRITE: {
+ WriteRange* write_range = static_cast<WriteRange*>(range);
+ Status status = write_range->DoWrite();
+ worker_context->OperDone(write_range, status);
+ break;
+ }
+ case RequestType::FILE_UPLOAD: {
+ RemoteOperRange* oper_range = static_cast<RemoteOperRange*>(range);
+ int64_t size = oper_range->block_size();
+ // Use malloc to get the memory in case there is no available space
+ // in the buffer pool because spilling to disk happens when scarcity
+ // of memory in the buffer pool. Be better to preserve memory than
+ // malloc.
+ uint8_t* buffer = static_cast<uint8_t*>(malloc(size));
+ if (UNLIKELY(buffer == nullptr)) {
+ worker_context->OperDone(oper_range,
+ Status(Substitute("Couldn't allocate memory for remote file operations, "
+ "block size: '$0'",
+ size)));
+ } else {
+ Status oper_status = oper_range->DoOper(buffer, size);
+ worker_context->OperDone(oper_range, oper_status);
+ free(buffer);
+ }
+ break;
+ }
+ default:
+ DCHECK(false) << "Invalid request type: " << range->request_type();
}
}
}
@@ -570,7 +781,7 @@ void DiskIoMgr::Write(RequestContext* writer_context, WriteRange* write_range) {
ret_status = WriteRangeHelper(file_handle, write_range);
- close_status = local_file_system_->Fclose(file_handle, write_range);
+ close_status = local_file_system_->Fclose(file_handle, write_range->file());
if (ret_status.ok() && !close_status.ok()) ret_status = close_status;
}
@@ -580,13 +791,13 @@ end:
} else {
queue->write_io_err()->Increment(1);
}
- writer_context->WriteDone(write_range, ret_status);
+ writer_context->OperDone(write_range, ret_status);
}
Status DiskIoMgr::WriteRangeHelper(FILE* file_handle, WriteRange* write_range) {
// Seek to the correct offset and perform the write.
- RETURN_IF_ERROR(local_file_system_->Fseek(file_handle, write_range->offset(), SEEK_SET,
- write_range));
+ RETURN_IF_ERROR(local_file_system_->Fseek(
+ file_handle, write_range->offset(), SEEK_SET, write_range));
#ifndef NDEBUG
if (FLAGS_stress_scratch_write_delay_ms > 0) {
diff --git a/be/src/runtime/io/disk-io-mgr.h b/be/src/runtime/io/disk-io-mgr.h
index b26f910..32a50e4 100644
--- a/be/src/runtime/io/disk-io-mgr.h
+++ b/be/src/runtime/io/disk-io-mgr.h
@@ -37,7 +37,6 @@
#include "util/thread.h"
namespace impala {
-
namespace io {
class DataCache;
@@ -74,10 +73,20 @@ class DiskQueue;
/// are made. It is the responsibility of the client to ensure that the data to be
/// written is valid. The file to be written is created if not already present.
///
+/// For File Operations:
+/// In addition to operations for readers and writers, there is a special operation type
+/// which is the file operation, allows io operations (upload or fetch) on the entire file
+/// between local filesystem and remote filesystem. Each file operation range is enqueued
+/// in the specific file operation queue, and processed by a disk thread for doing the
+/// file operation. After the operation is done, a callback function of the file operation
+/// range is invoked. There is a memory allocation for a block to do the data
+/// transmission, and the memory is released immediately after the operation is done.
+///
/// There are several key methods for scanning data with the IoMgr.
/// 1. RequestContext::StartScanRange(): adds range to the IoMgr to start immediately.
/// 2. RequestContext::AddScanRanges(): adds ranges to the IoMgr that the reader wants to
-/// scan, but does not start them until RequestContext::GetNextUnstartedRange() is called.
+/// scan, but does not start them until RequestContext::GetNextUnstartedRange() is
+/// called.
/// 3. RequestContext::GetNextUnstartedRange(): returns to the caller the next scan range
/// it should process.
/// 4. ScanRange::GetNext(): returns the next buffer for this range, blocking until
@@ -98,7 +107,9 @@ class DiskQueue;
/// by a write, and a write is followed by a read, i.e. reads and writes alternate.
/// If multiple write ranges are enqueued for a single disk, they will be processed
/// by the disk threads in order, but may complete in any order. No guarantees are made
-/// on ordering of writes across disks.
+/// on ordering of writes across disks. The strategy of scheduling is the same for file
+/// operation ranges, but the file operation ranges are in a sperate queue compared to
+/// read(scan) or write ranges.
///
/// Resource Management: the IoMgr is designed to share the available disk I/O capacity
/// between many clients and to help use the available I/O capacity efficiently. The IoMgr
@@ -278,7 +289,7 @@ class DiskIoMgr : public CacheLineAligned {
/// disk_queues_. The disk_id is the volume ID for the local disk that holds the
/// files, or -1 if unknown. Flag expected_local is true iff this impalad is
/// co-located with the datanode for this file.
- int AssignQueue(const char* file, int disk_id, bool expected_local);
+ int AssignQueue(const char* file, int disk_id, bool expected_local = false);
int64_t min_buffer_size() const { return min_buffer_size_; }
int64_t max_buffer_size() const { return max_buffer_size_; }
@@ -295,9 +306,19 @@ class DiskIoMgr : public CacheLineAligned {
/// The disk ID (and therefore disk_queues_ index) used for DFS accesses.
int RemoteDfsDiskId() const { return num_local_disks() + REMOTE_DFS_DISK_OFFSET; }
+ /// The disk ID used for DFS File Operations (upload or fetch) accesses.
+ int RemoteDfsDiskFileOperId() const {
+ return num_local_disks() + REMOTE_DFS_DISK_FILE_OPER_OFFSET;
+ }
+
/// The disk ID (and therefore disk_queues_ index) used for S3 accesses.
int RemoteS3DiskId() const { return num_local_disks() + REMOTE_S3_DISK_OFFSET; }
+ /// The disk ID used for S3 File Operations (upload or fetch) accesses.
+ int RemoteS3DiskFileOperId() const {
+ return num_local_disks() + REMOTE_S3_DISK_FILE_OPER_OFFSET;
+ }
+
/// The disk ID (and therefore disk_queues_ index) used for ABFS accesses.
int RemoteAbfsDiskId() const { return num_local_disks() + REMOTE_ABFS_DISK_OFFSET; }
@@ -352,12 +373,17 @@ class DiskIoMgr : public CacheLineAligned {
/// "Disk" queue offsets for remote accesses. Offset 0 corresponds to
/// disk ID (i.e. disk_queue_ index) of num_local_disks().
+ /// DISK_FILE_OPER queues are for the file operations, such as file upload
+ /// or fetch, to be isolated from the range read/write operations in ordinary
+ /// queues mainly for efficiency consideration.
enum {
REMOTE_DFS_DISK_OFFSET = 0,
REMOTE_S3_DISK_OFFSET,
REMOTE_ADLS_DISK_OFFSET,
REMOTE_ABFS_DISK_OFFSET,
REMOTE_OZONE_DISK_OFFSET,
+ REMOTE_DFS_DISK_FILE_OPER_OFFSET,
+ REMOTE_S3_DISK_FILE_OPER_OFFSET,
REMOTE_NUM_DISKS
};
@@ -387,7 +413,10 @@ class DiskIoMgr : public CacheLineAligned {
/// BEGIN: private members that are accessed by other io:: classes
friend class DiskQueue;
friend class ScanRange;
+ friend class WriteRange;
+ friend class RemoteOperRange;
friend class HdfsFileReader;
+ friend class LocalFileWriter;
/// Write the specified range to disk and calls writer_context->WriteDone() when done.
/// Responsible for opening and closing the file that is written.
diff --git a/be/src/runtime/io/file-writer.h b/be/src/runtime/io/file-writer.h
new file mode 100644
index 0000000..fe86989
--- /dev/null
+++ b/be/src/runtime/io/file-writer.h
@@ -0,0 +1,80 @@
+// 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.
+
+#pragma once
+
+#include <string>
+
+#include "common/status.h"
+
+namespace impala {
+class TmpFile;
+namespace io {
+class DiskIoMgr;
+class WriteRange;
+
+/// Abstract class that provides interface for file writing operations.
+/// Child classes implement these operations for the local file system
+/// and for HDFS.
+/// A typical process is calling Open(), then Write() with certain
+/// WriteRange objects, after the writing finished, call the Close() to
+/// close the file handle. The writes are the sequential operations to
+/// the same file.
+class FileWriter {
+ public:
+ FileWriter(DiskIoMgr* io_mgr, const char* file_path, const int64_t file_size)
+ : io_mgr_(io_mgr), file_path_(file_path), file_size_(file_size) {}
+ virtual ~FileWriter() {}
+
+ /// The set of Open/Write/Close function is used for sequential range writing
+ /// for a specific file. The file handle is opened and closed only once per
+ /// file by calling Open() and Close(). By calling the Write(), the write
+ /// ranges are written into the file sequentially in different threads using
+ /// the same file handle.
+ /// The caller needs to call the Close() to close the file handle when is_full
+ /// is returned True in Write().
+ virtual Status Open() = 0;
+ virtual Status Write(WriteRange* range, int64_t* written_bytes) = 0;
+ virtual Status Close() = 0;
+
+ /// The WriteOne function is used for a random range writing. The caller would
+ /// expect the function to open a file handle of the file, write the range, and
+ /// close the file handle.
+ virtual Status WriteOne(WriteRange*) = 0;
+
+ protected:
+ /// DiskIoMgr used for the I/O operations and the statistics.
+ DiskIoMgr* io_mgr_ = nullptr;
+
+ /// The lock_ is to guarantee only one thread is using the file handle.
+ /// Ideally, Open() and Close() would be called once for each file writer, so
+ /// the lock_ is a exclusive lock mainly for running the Write().
+ /// WriteOne() doesn't need the lock_ because it opens its own file handle,
+ /// and doesn't share it with other WriteRanges.
+ std::mutex lock_;
+
+ /// The bytes have been written by the file writer.
+ int64_t written_bytes_ = 0;
+
+ /// The file path to be written.
+ const char* file_path_;
+
+ /// The size of the file.
+ const int64_t file_size_;
+};
+} // namespace io
+} // namespace impala
diff --git a/be/src/runtime/io/local-file-system.cc b/be/src/runtime/io/local-file-system.cc
index 63aae52..dc82d64 100644
--- a/be/src/runtime/io/local-file-system.cc
+++ b/be/src/runtime/io/local-file-system.cc
@@ -32,8 +32,18 @@ FILE* LocalFileSystem::FdopenAux(int file_desc, const char* options) {
return fdopen(file_desc, options);
}
-Status LocalFileSystem::OpenForWrite(const char* file_name, int oflag, int mode,
- FILE** file) {
+Status LocalFileSystem::OpenForRead(
+ const char* file_name, int oflag, int mode, FILE** file) {
+ return Open(file_name, oflag, mode, "rb", file);
+}
+
+Status LocalFileSystem::OpenForWrite(
+ const char* file_name, int oflag, int mode, FILE** file) {
+ return Open(file_name, oflag, mode, "wb", file);
+}
+
+Status LocalFileSystem::Open(
+ const char* file_name, int oflag, int mode, const char* fd_option, FILE** file) {
DCHECK(file_name != nullptr);
DCHECK(file != nullptr);
@@ -42,7 +52,7 @@ Status LocalFileSystem::OpenForWrite(const char* file_name, int oflag, int mode,
return ErrorConverter::GetErrorStatusFromErrno("open()", file_name, errno);
}
- *file = FdopenAux(file_desc, "wb");
+ *file = FdopenAux(file_desc, fd_option);
if (*file == nullptr) {
Status fdopen_status = ErrorConverter::GetErrorStatusFromErrno("fdopen()", file_name,
errno);
@@ -84,11 +94,25 @@ size_t LocalFileSystem::FwriteAux(FILE* file_handle, const WriteRange* write_ran
return fwrite(write_range->data(), 1, write_range->len(), file_handle);
}
-Status LocalFileSystem::Fclose(FILE* file_handle, const WriteRange* write_range) {
+Status LocalFileSystem::Fread(
+ FILE* file_handle, uint8_t* buffer, int64_t length, const char* file_path) {
+ DCHECK(file_handle != nullptr);
+ int64_t bytes_read = FreadAux(file_handle, buffer, length);
+ if (bytes_read < length) {
+ return ErrorConverter::GetErrorStatusFromErrno(
+ "fread()", file_path, errno, {{"range_length", SimpleItoa(length)}});
+ }
+ return Status::OK();
+}
+
+size_t LocalFileSystem::FreadAux(FILE* file_handle, uint8_t* buffer, int64_t length) {
+ return fread(buffer, 1, length, file_handle);
+}
+
+Status LocalFileSystem::Fclose(FILE* file_handle, const char* file_path) {
DCHECK(file_handle != nullptr);
if (FcloseAux(file_handle) != 0) {
- return ErrorConverter::GetErrorStatusFromErrno("fclose()", write_range->file(),
- errno);
+ return ErrorConverter::GetErrorStatusFromErrno("fclose()", file_path, errno);
}
return Status::OK();
}
@@ -97,5 +121,14 @@ int LocalFileSystem::FcloseAux(FILE* file_handle) {
return fclose(file_handle);
}
+Status LocalFileSystem::Write(int file_desc, const WriteRange* range) {
+ DCHECK(range != nullptr);
+ int64_t bytes_written = write(file_desc, range->data(), range->len());
+ if (bytes_written < range->len()) {
+ return ErrorConverter::GetErrorStatusFromErrno(
+ "write()", range->file(), errno, {{"range_length", SimpleItoa(range->len())}});
+ }
+ return Status::OK();
+}
}
}
diff --git a/be/src/runtime/io/local-file-system.h b/be/src/runtime/io/local-file-system.h
index e30edce..c14f932 100644
--- a/be/src/runtime/io/local-file-system.h
+++ b/be/src/runtime/io/local-file-system.h
@@ -32,14 +32,19 @@ class LocalFileSystem {
public:
// A wrapper around open() and fdopen(). For the possible values of oflag and mode
// see the documentation of open(). Sets 'file' to a FILE* returned from fdopen().
- Status OpenForWrite(const char* file_name, int oflag, int mode, FILE** file);
+ Status Open(
+ const char* file_name, int oflag, int mode, const char* fd_option, FILE** file);
+ Status OpenForRead(const char* file_name, int oflag, int mode, FILE** file);
+ Status OpenForWrite(const char* file_name, int oflag, int mode, FILE** file);
- Status Fseek(FILE* file_handle, off_t offset, int whence,
- const WriteRange* write_range);
- Status Fwrite(FILE* file_handle, const WriteRange* write_range);
- Status Fclose(FILE* file_handle, const WriteRange* write_range);
+ Status Fseek(FILE* file_handle, off_t offset, int whence, const WriteRange* write_range);
+ Status Fwrite(FILE* file_handle, const WriteRange* write_range);
+ Status Fread(FILE* file_handle, uint8_t* buffer, int64_t length, const char* file_path);
+ Status Fclose(FILE* file_handle, const char* file_path);
+ virtual ~LocalFileSystem() {}
- virtual ~LocalFileSystem() {}
+ // Wrapper function to use write() to write the bytes.
+ Status Write(int file_desc, const WriteRange* range);
protected:
// Wrapper functions around open(), fdopen(), fseek(), fwrite() and fclose().
@@ -48,6 +53,7 @@ protected:
virtual FILE* FdopenAux(int file_desc, const char* options);
virtual int FseekAux(FILE* file_handle, off_t offset, int whence);
virtual size_t FwriteAux(FILE* file_handle, const WriteRange* write_range);
+ virtual size_t FreadAux(FILE* file_handle, uint8_t* buffer, int64_t length);
virtual int FcloseAux(FILE* file_handle);
};
diff --git a/be/src/runtime/io/local-file-writer.cc b/be/src/runtime/io/local-file-writer.cc
new file mode 100644
index 0000000..0395fd3
--- /dev/null
+++ b/be/src/runtime/io/local-file-writer.cc
@@ -0,0 +1,96 @@
+// 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 <stdio.h>
+#include <algorithm>
+
+#include "runtime/io/disk-io-mgr-internal.h"
+#include "runtime/io/local-file-writer.h"
+#include "runtime/io/request-ranges.h"
+#include "util/histogram-metric.h"
+#include "util/impalad-metrics.h"
+#include "util/metrics.h"
+
+#include "common/names.h"
+
+namespace impala {
+namespace io {
+
+Status LocalFileWriter::Open() {
+ lock_guard<mutex> lock(lock_);
+ if (file_ != nullptr) return Status::OK();
+ return io_mgr_->local_file_system_->OpenForWrite(
+ file_path_, O_RDWR | O_CREAT, S_IRUSR | S_IWUSR, &file_);
+}
+
+Status LocalFileWriter::Write(WriteRange* range, int64_t* written_bytes) {
+ lock_guard<mutex> lock(lock_);
+ if (file_ == nullptr) {
+ return Status(Substitute("File handle of $0 has been closed.", file_path_));
+ }
+ // Use write() instead of fwrite(), because the file handle is shared by multiple write
+ // ranges before the file is full and closed, during the period, if using fwrite(), it
+ // is not safe to read the file because the buffer fwrite() is using may not flush to
+ // the disk. But write() is safe to read immediately after the write because it writes
+ // without a buffer.
+ RETURN_IF_ERROR(io_mgr_->local_file_system_->Write(fileno(file_), range));
+ ImpaladMetrics::IO_MGR_BYTES_WRITTEN->Increment(range->len());
+ range->SetOffset(written_bytes_);
+ written_bytes_ += range->len();
+ *written_bytes = written_bytes_;
+ return Status::OK();
+}
+
+Status LocalFileWriter::Close() {
+ lock_guard<mutex> lock(lock_);
+ if (file_ == nullptr) return Status::OK();
+ RETURN_IF_ERROR(io_mgr_->local_file_system_->Fclose(file_, file_path_));
+ file_ = nullptr;
+ return Status::OK();
+}
+
+Status LocalFileWriter::WriteOne(WriteRange* write_range) {
+ DCHECK(write_range != nullptr);
+ Status ret_status = Status::OK();
+ // Do not need to acquire the lock_ because we open a new file handle in WriteOne
+ // for writing, instead of sharing the same file handle.
+ FILE* file_handle = nullptr;
+ Status close_status = Status::OK();
+ DiskQueue* queue = io_mgr_->disk_queues_[write_range->disk_id()];
+
+ {
+ ScopedHistogramTimer write_timer(queue->write_latency());
+ ret_status = io_mgr_->local_file_system_->OpenForWrite(
+ write_range->file(), O_RDWR | O_CREAT, S_IRUSR | S_IWUSR, &file_handle);
+ if (!ret_status.ok()) goto end;
+
+ ret_status = io_mgr_->WriteRangeHelper(file_handle, write_range);
+
+ close_status = io_mgr_->local_file_system_->Fclose(file_handle, write_range->file());
+ if (ret_status.ok() && !close_status.ok()) ret_status = close_status;
+ }
+
+end:
+ if (ret_status.ok()) {
+ queue->write_size()->Update(write_range->len());
+ } else {
+ queue->write_io_err()->Increment(1);
+ }
+ return ret_status;
+}
+} // namespace io
+} // namespace impala
diff --git a/be/src/runtime/io/local-file-writer.h b/be/src/runtime/io/local-file-writer.h
new file mode 100644
index 0000000..23bf66c
--- /dev/null
+++ b/be/src/runtime/io/local-file-writer.h
@@ -0,0 +1,45 @@
+// 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.
+
+#pragma once
+
+#include "runtime/io/file-writer.h"
+#include "runtime/tmp-file-mgr-internal.h"
+
+namespace impala {
+namespace io {
+
+/// File writer class for the local file system.
+class LocalFileWriter : public FileWriter {
+ public:
+ LocalFileWriter(DiskIoMgr* io_mgr, const char* file_path, int64_t file_size = 0)
+ : FileWriter(io_mgr, file_path, file_size) {}
+ ~LocalFileWriter() {}
+
+ virtual Status Open() override;
+ virtual Status Write(WriteRange* range, int64_t* written_bytes) override;
+ virtual Status Close() override;
+
+ // Open and close the file within the function.
+ virtual Status WriteOne(WriteRange* range) override;
+
+ private:
+ /// Points to a C FILE object between calls to Open() and Close(), otherwise nullptr.
+ FILE* file_ = nullptr;
+};
+} // namespace io
+} // namespace impala
diff --git a/be/src/runtime/io/request-context.cc b/be/src/runtime/io/request-context.cc
index 6c3b356..b805a54 100644
--- a/be/src/runtime/io/request-context.cc
+++ b/be/src/runtime/io/request-context.cc
@@ -58,6 +58,10 @@ class RequestContext::PerDiskState {
const InternalQueue<WriteRange>* unstarted_write_ranges() const {
return &unstarted_write_ranges_;
}
+ const InternalQueue<RemoteOperRange>* unstarted_remote_upload_ranges() const {
+ return &unstarted_remote_upload_ranges_;
+ }
+
const InternalQueue<RequestRange>* in_flight_ranges() const {
return &in_flight_ranges_;
}
@@ -66,6 +70,10 @@ class RequestContext::PerDiskState {
InternalQueue<WriteRange>* unstarted_write_ranges() {
return &unstarted_write_ranges_;
}
+ InternalQueue<RemoteOperRange>* unstarted_remote_upload_ranges() {
+ return &unstarted_remote_upload_ranges_;
+ }
+
InternalQueue<RequestRange>* in_flight_ranges() { return &in_flight_ranges_; }
/// Schedules the request context on this disk if it's not already on the queue.
@@ -188,6 +196,9 @@ class RequestContext::PerDiskState {
/// GetNextRequestRange() and GetNextUnstartedRange() may result in only reads being
/// processed)
InternalQueue<WriteRange> unstarted_write_ranges_;
+
+ /// A Queue for file operation ranges to process uploading operations to remote disks.
+ InternalQueue<RemoteOperRange> unstarted_remote_upload_ranges_;
};
void RequestContext::ReadDone(int disk_id, ReadOutcome outcome, ScanRange* range) {
@@ -218,15 +229,22 @@ void RequestContext::ReadDone(int disk_id, ReadOutcome outcome, ScanRange* range
DCHECK(Validate()) << endl << DebugString();
}
-void RequestContext::WriteDone(WriteRange* write_range, const Status& write_status) {
- // Copy disk_id before running callback: the callback may modify write_range.
- int disk_id = write_range->disk_id();
+void RequestContext::OperDone(RequestRange* range, const Status& status) {
+ DCHECK(range != nullptr);
+
+ // Copy disk_id before running callback: the callback may modify range.
+ int disk_id = range->disk_id();
// Execute the callback before decrementing the thread count. Otherwise
// RequestContext::Cancel() that waits for the disk ref count to be 0 will
// return, creating a race, e.g. see IMPALA-1890.
- // The status of the write does not affect the status of the writer context.
- write_range->callback()(write_status);
+ // The status of the operation does not affect the status of the request context.
+ if (range->request_type() == RequestType::WRITE) {
+ (static_cast<WriteRange*>(range))->callback()(status);
+ } else {
+ DCHECK(range->request_type() == RequestType::FILE_UPLOAD);
+ (static_cast<RemoteOperRange*>(range))->callback()(status);
+ }
{
unique_lock<mutex> lock(lock_);
DCHECK(Validate()) << endl << DebugString();
@@ -264,6 +282,7 @@ void RequestContext::WriteDone(WriteRange* write_range, const Status& write_stat
void RequestContext::Cancel() {
// Callbacks are collected in this vector and invoked while no lock is held.
vector<WriteRange::WriteDoneCallback> write_callbacks;
+ vector<RemoteOperRange::RemoteOperDoneCallback> remote_oper_callbacks;
{
unique_lock<mutex> lock(lock_);
DCHECK(Validate()) << endl << DebugString();
@@ -292,6 +311,12 @@ void RequestContext::Cancel() {
while ((write_range = disk_state.unstarted_write_ranges()->Dequeue()) != nullptr) {
write_callbacks.push_back(write_range->callback());
}
+
+ RemoteOperRange* oper_range;
+ while ((oper_range = disk_state.unstarted_remote_upload_ranges()->Dequeue())
+ != nullptr) {
+ remote_oper_callbacks.push_back(oper_range->callback());
+ }
}
// Clear out the lists of scan ranges.
while (ready_to_start_ranges_.Dequeue() != nullptr);
@@ -309,6 +334,11 @@ void RequestContext::Cancel() {
write_callback(CONTEXT_CANCELLED);
}
+ for (const RemoteOperRange::RemoteOperDoneCallback& oper_callback :
+ remote_oper_callbacks) {
+ oper_callback(CONTEXT_CANCELLED);
+ }
+
// Signal reader and unblock the GetNext/Read thread. That read will fail with
// a cancelled status.
ready_to_start_ranges_cv_.NotifyAll();
@@ -366,8 +396,7 @@ void RequestContext::AddRangeToDisk(const unique_lock<mutex>& lock,
disk_state->ScheduleContext(lock, this, range->disk_id());
}
}
- } else {
- DCHECK(range->request_type() == RequestType::WRITE);
+ } else if (range->request_type() == RequestType::WRITE) {
DCHECK(schedule_mode == ScheduleMode::IMMEDIATELY) << static_cast<int>(schedule_mode);
WriteRange* write_range = static_cast<WriteRange*>(range);
disk_state->unstarted_write_ranges()->Enqueue(write_range);
@@ -375,7 +404,14 @@ void RequestContext::AddRangeToDisk(const unique_lock<mutex>& lock,
// Ensure that the context is scheduled so that the write range gets picked up.
// ScheduleContext() has no effect if already scheduled, so this is safe to do always.
disk_state->ScheduleContext(lock, this, range->disk_id());
+ } else {
+ DCHECK(range->request_type() == RequestType::FILE_UPLOAD);
+ DCHECK(schedule_mode == ScheduleMode::IMMEDIATELY) << static_cast<int>(schedule_mode);
+ RemoteOperRange* oper_range = static_cast<RemoteOperRange*>(range);
+ disk_state->unstarted_remote_upload_ranges()->Enqueue(oper_range);
+ disk_state->ScheduleContext(lock, this, range->disk_id());
}
+
++disk_state->num_remaining_ranges();
}
@@ -530,10 +566,18 @@ Status RequestContext::TryReadFromCache(const unique_lock<mutex>& lock,
Status RequestContext::AddWriteRange(WriteRange* write_range) {
unique_lock<mutex> lock(lock_);
if (state_ == RequestContext::Cancelled) return CONTEXT_CANCELLED;
+ write_range->SetRequestContext(this);
AddRangeToDisk(lock, write_range, ScheduleMode::IMMEDIATELY);
return Status::OK();
}
+Status RequestContext::AddRemoteOperRange(RemoteOperRange* oper_range) {
+ unique_lock<mutex> lock(lock_);
+ if (state_ == RequestContext::Cancelled) return CONTEXT_CANCELLED;
+ AddRangeToDisk(lock, oper_range, ScheduleMode::IMMEDIATELY);
+ return Status::OK();
+}
+
void RequestContext::AddActiveScanRangeLocked(
const unique_lock<mutex>& lock, ScanRange* range) {
DCHECK(lock.mutex() == &lock_ && lock.owns_lock());
@@ -604,6 +648,15 @@ RequestRange* RequestContext::GetNextRequestRange(int disk_id) {
request_disk_state->in_flight_ranges()->Enqueue(write_range);
}
+ // Do remote temporary files related work.
+ if (!request_disk_state->unstarted_remote_upload_ranges()->empty()) {
+ RemoteOperRange* oper_range;
+ if (!request_disk_state->unstarted_remote_upload_ranges()->empty()) {
+ oper_range = request_disk_state->unstarted_remote_upload_ranges()->Dequeue();
+ request_disk_state->in_flight_ranges()->Enqueue(oper_range);
+ }
+ }
+
// Get the next scan range to work on from the reader. Only in_flight_ranges
// are eligible since the disk threads do not start new ranges on their own.
if (request_disk_state->in_flight_ranges()->empty()) {
diff --git a/be/src/runtime/io/request-context.h b/be/src/runtime/io/request-context.h
index e32c334..5424901 100644
--- a/be/src/runtime/io/request-context.h
+++ b/be/src/runtime/io/request-context.h
@@ -139,6 +139,10 @@ class RequestContext {
/// on the IoMgr disk queue.
Status AddWriteRange(WriteRange* write_range) WARN_UNUSED_RESULT;
+ /// Add a RemoteOperRange for the writer. This is non-blocking and schedules the context
+ /// on the IoMgr disk queue.
+ Status AddRemoteOperRange(RemoteOperRange* write_range) WARN_UNUSED_RESULT;
+
/// Cancel the context asynchronously. All outstanding requests are cancelled
/// asynchronously. This does not need to be called if the context finishes normally.
/// Calling GetNext() on any scan ranges belonging to this RequestContext will return
@@ -223,6 +227,9 @@ class RequestContext {
friend class DiskIoMgr;
friend class ScanRange;
friend class HdfsFileReader;
+ friend class WriteRange;
+ friend class RemoteOperRange;
+ friend class LocalFileWriter;
enum State {
/// Reader is initialized and maps to a client
@@ -279,11 +286,11 @@ class RequestContext {
/// Caller must not hold 'lock_'.
void ReadDone(int disk_id, ReadOutcome outcome, ScanRange* range);
- /// Invokes write_range->callback() after the range has been written and
- /// updates per-disk state and handle state. The status of the write OK/RUNTIME_ERROR
- /// etc. is passed via write_status and to the callback. A write error does not cancel
- /// the writer context - that decision is left to the callback handler.
- void WriteDone(WriteRange* write_range, const Status& write_status);
+ /// Invokes write_range->callback() or oper_range->callback() after the range has been
+ /// executed, and updates per-disk state and handle state. The status of the operation
+ /// OK/RUNTIME_ERROR etc. is passed via status and to the callback. An error status
+ /// does not cancel the request context - that decision is left to the callback handler.
+ void OperDone(RequestRange* range, const Status& status);
/// Cancel the context if not already cancelled, wait for all scan ranges to finish
/// and mark the context as inactive, after which it cannot be used.
diff --git a/be/src/runtime/io/request-ranges.h b/be/src/runtime/io/request-ranges.h
index 817151e..dca4398 100644
--- a/be/src/runtime/io/request-ranges.h
+++ b/be/src/runtime/io/request-ranges.h
@@ -22,12 +22,14 @@
#include <functional>
#include <mutex>
+#include <boost/thread/shared_mutex.hpp>
#include <gtest/gtest_prod.h> // for FRIEND_TEST
#include "common/atomic.h"
#include "common/hdfs.h"
#include "common/status.h"
#include "runtime/bufferpool/buffer-pool.h"
+#include "runtime/io/disk-file.h"
#include "util/condition-variable.h"
#include "util/internal-queue.h"
@@ -37,6 +39,7 @@ class DiskIoMgr;
class DiskQueue;
class ExclusiveHdfsFileHandle;
class FileReader;
+class FileWriter;
class RequestContext;
class ScanRange;
@@ -105,10 +108,13 @@ class BufferDescriptor {
};
/// The request type, read or write associated with a request range.
+/// Ohter than those, request type file_upload is the type for remote file operation
+/// ranges, for doing file uploading to the remote.
struct RequestType {
enum type {
READ,
WRITE,
+ FILE_UPLOAD,
};
};
@@ -140,8 +146,12 @@ class RequestRange : public InternalQueue<RequestRange>::Node {
RequestType::type request_type() const { return request_type_; }
protected:
- RequestRange(RequestType::type request_type)
- : fs_(nullptr), offset_(-1), len_(-1), disk_id_(-1), request_type_(request_type) {}
+ RequestRange(RequestType::type request_type, int disk_id = -1)
+ : fs_(nullptr),
+ offset_(-1),
+ len_(-1),
+ disk_id_(disk_id),
+ request_type_(request_type) {}
/// Hadoop filesystem that contains file_, or set to nullptr for local filesystem.
hdfsFS fs_;
@@ -155,7 +165,7 @@ class RequestRange : public InternalQueue<RequestRange>::Node {
/// Length of data read or written.
int64_t len_;
- /// Id of disk containing byte range.
+ /// Id of disk queue containing byte range.
int disk_id_;
/// The type of IO request, READ or WRITE.
@@ -264,19 +274,25 @@ class ScanRange : public RequestRange {
/// 'buffer_opts' specifies buffer management options - see the DiskIoMgr class comment
/// and the BufferOpts comments for details.
/// 'meta_data' is an arbitrary client-provided pointer for any auxiliary data.
- ///
+ /// 'disk_file' and 'disk_buffer_file' provides methods to confirm and guarantee the
+ /// status of the physical file is available to access for the scanning. They are used
+ /// for scanning a file related to spilling to a remote filesystem, both should be
+ /// non-null, for other cases, like scanning a file related to spilling to local
+ /// filesytem, both would not be used and should be null.
/// TODO: IMPALA-4249: clarify if a ScanRange can be reused after Reset(). Currently
/// it is not generally safe to do so, but some unit tests reuse ranges after
/// successfully reading to eos.
void Reset(hdfsFS fs, const char* file, int64_t len, int64_t offset, int disk_id,
bool expected_local, int64_t mtime, const BufferOpts& buffer_opts,
- void* meta_data = nullptr);
+ void* meta_data = nullptr, DiskFile* disk_file = nullptr,
+ DiskFile* disk_buffer_file = nullptr);
/// Same as above, but it also adds sub-ranges. No need to merge contiguous sub-ranges
/// in advance, as this method will do the merge.
void Reset(hdfsFS fs, const char* file, int64_t len, int64_t offset, int disk_id,
bool expected_local, int64_t mtime, const BufferOpts& buffer_opts,
- std::vector<SubRange>&& sub_ranges, void* meta_data = nullptr);
+ std::vector<SubRange>&& sub_ranges, void* meta_data = nullptr,
+ DiskFile* disk_file = nullptr, DiskFile* disk_buffer_file = nullptr);
void* meta_data() const { return meta_data_; }
int cache_options() const { return cache_options_; }
@@ -285,6 +301,7 @@ class ScanRange : public RequestRange {
bool read_in_flight() const { return read_in_flight_; }
bool expected_local() const { return expected_local_; }
int64_t bytes_to_read() const { return bytes_to_read_; }
+ bool use_local_buffer() const { return use_local_buffer_; }
/// Returns the next buffer for this scan range. buffer is an output parameter.
/// This function blocks until a buffer is ready or an error occurred. If this is
@@ -357,6 +374,9 @@ class ScanRange : public RequestRange {
/// the underlying filesystem. Caller must not hold 'lock_'.
ReadOutcome DoRead(DiskQueue* queue, int disk_id);
+ /// The function runs the actual read logic to read content with the specific reader.
+ ReadOutcome DoReadInternal(DiskQueue* queue, int disk_id, FileReader* file_reader);
+
/// Cleans up a buffer that was not returned to the client.
/// Either ReturnBuffer() or CleanUpBuffer() is called for every BufferDescriptor.
/// The caller must hold 'lock_' via 'scan_range_lock'.
@@ -444,7 +464,8 @@ class ScanRange : public RequestRange {
/// If cached data is available, then memcpy() from it instead of actually reading the
/// files. 'queue' is updated with the latencies and sizes of reads from the underlying
/// filesystem.
- Status ReadSubRanges(DiskQueue* queue, BufferDescriptor* buffer, bool* eof);
+ Status ReadSubRanges(
+ DiskQueue* queue, BufferDescriptor* buffer, bool* eof, FileReader* file_reader);
/// Validates the internal state of this range. lock_ must be taken
/// before calling this.
@@ -557,9 +578,24 @@ class ScanRange : public RequestRange {
/// Polymorphic object that is responsible for doing file operations.
std::unique_ptr<FileReader> file_reader_;
+ /// Polymorphic object that is responsible for doing file operations if the path
+ /// is a remote url and the file is in the local buffer.
+ std::unique_ptr<FileReader> local_buffer_reader_;
+
+ /// If set to true, the scan range is using local_buffer_reader_ to do scan operations.
+ /// The flag is set during DoRead(). If the path is a remote path and the file has
+ /// a local buffer, the flag is set to true, otherwise the flag is false.
+ bool use_local_buffer_{false};
+
/// If not empty, the ScanRange will only read these parts from the file.
std::vector<SubRange> sub_ranges_;
+ /// The file handle of the physical file to be accessed.
+ DiskFile* disk_file_ = nullptr;
+
+ /// The file handle of the physical buffer file to be accessed.
+ DiskFile* disk_buffer_file_ = nullptr;
+
// Read position in the sub-ranges.
struct SubRangePosition {
/// Index of SubRange in 'ScanRange::sub_ranges_' to read next
@@ -604,17 +640,111 @@ class WriteRange : public RequestRange {
/// is called or after the write callback was called).
void SetData(const uint8_t* buffer, int64_t len);
+ /// Set the offset of this WriteRange.
+ /// Caller should guarantee the thread safe of calling the function.
+ void SetOffset(int64_t file_offset);
+
+ /// Set the DiskFile pointer which the WriteRange belongs to.
+ /// Can only be called when the write is not in flight (i.e. before AddWriteRange()
+ /// is called or after the write callback was called).
+ void SetDiskFile(DiskFile* disk_file) { disk_file_ = disk_file; }
+
+ /// Set the IO context to this WriteRange.
+ /// Can only be called when the write is not in flight (i.e. before AddWriteRange()
+ /// is called or after the write callback was called).
+ void SetRequestContext(RequestContext* io_ctx) { io_ctx_ = io_ctx; }
+
+ /// Execute writing the this range to the corresponding file.
+ Status DoWrite();
+
+ /// Handle the status of Function DoWrite().
+ Status DoWriteEnd(DiskQueue* queue, const Status& ret_status);
+
+ /// Return the data to be written.
const uint8_t* data() const { return data_; }
+
+ /// Return the disk file pointer of the write range.
+ DiskFile* disk_file() const { return disk_file_; }
+
+ RequestContext* io_ctx() const { return io_ctx_; }
+
+ /// Return if the disk file that the write range belongs to is completed.
+ bool is_full() const { return is_full_; }
+
WriteDoneCallback callback() const { return callback_; }
private:
DISALLOW_COPY_AND_ASSIGN(WriteRange);
+
+ friend class LocalFileWriter;
+
/// Data to be written. RequestRange::len_ contains the length of data
/// to be written.
const uint8_t* data_;
/// Callback to invoke after the write is complete.
WriteDoneCallback callback_;
+
+ /// IO context is to help to find out the disk queue and corresponding metrics.
+ RequestContext* io_ctx_;
+
+ /// The DiskFile is which the WriteRange writes into.
+ /// A DiskFile can have multiple WriteRanges.
+ DiskFile* disk_file_ = nullptr;
+
+ /// Indicate if the file which the write range belongs to is full after writing.
+ bool is_full_ = false;
+};
+
+class RemoteOperRange : public RequestRange {
+ public:
+ /// This callback is invoked on each RemoteOperRange after the operation is complete
+ /// or the context is cancelled. The status returned by the callback parameter indicates
+ /// if the operation was successful (i.e. Status::OK), if there was an error
+ /// TErrorCode::RUNTIME_ERROR) or if the context was cancelled
+ /// (TErrorCode::CANCELLED_INTERNALLY). The callback is only invoked if this
+ /// RemoteOperRange was successfully added (i.e. AddRemoteOperRange() succeeded).
+ /// No locks are held while the callback is invoked.
+ typedef std::function<void(const Status&)> RemoteOperDoneCallback;
+ RemoteOperRange(DiskFile* src_file, DiskFile* dst_file, int64_t file_offset,
+ int disk_id, RequestType::type type, DiskIoMgr* io_mgr,
+ RemoteOperDoneCallback callback);
+
+ /// Called from a disk I/O thread to do the file operation of this range. The
+ /// returned Status describes what the result of the read was. 'buff' is the
+ /// block buffer which is used for file operations. 'buff_size' is the size of the
+ /// block buffer. Caller must not hold 'lock_'.
+ Status DoOper(uint8_t* buff, int64_t buff_size);
+
+ int64_t block_size() { return block_size_; }
+
+ RemoteOperDoneCallback callback() const { return callback_; }
+
+ private:
+ DISALLOW_COPY_AND_ASSIGN(RemoteOperRange);
+
+ friend class LocalFileWriter;
+
+ /// Callback to invoke after the operation is complete.
+ RemoteOperDoneCallback callback_;
+
+ /// IO manager is to help to find out the disk queue and corresponding metrics.
+ DiskIoMgr* io_mgr_;
+
+ /// disk_file_src_ contains the information of the file which is the source to
+ /// do the file operation. For example, if the operation is file upload, this
+ /// is the handle for the source physical file to be uploaded.
+ DiskFile* disk_file_src_;
+
+ /// disk_file_dst_ contains the information of the file which is the destination
+ /// of the file operation.
+ DiskFile* disk_file_dst_;
+
+ /// block size to do the file operation.
+ int64_t block_size_;
+
+ /// Execute the upload file operation.
+ Status DoUpload(uint8_t* buff, int64_t buff_size);
};
inline bool BufferDescriptor::is_cached() const {
diff --git a/be/src/runtime/io/scan-range.cc b/be/src/runtime/io/scan-range.cc
index 9eeb788..569ef57 100644
--- a/be/src/runtime/io/scan-range.cc
+++ b/be/src/runtime/io/scan-range.cc
@@ -16,8 +16,8 @@
// under the License.
#include "runtime/exec-env.h"
-#include "runtime/io/disk-io-mgr.h"
#include "runtime/io/disk-io-mgr-internal.h"
+#include "runtime/io/disk-io-mgr.h"
#include "runtime/io/hdfs-file-reader.h"
#include "runtime/io/local-file-reader.h"
#include "util/error-util.h"
@@ -27,6 +27,9 @@
using namespace impala;
using namespace impala::io;
+using namespace std;
+using std::mutex;
+using std::unique_lock;
DECLARE_bool(cache_remote_file_handles);
DECLARE_bool(cache_s3_file_handles);
@@ -160,7 +163,8 @@ unique_ptr<BufferDescriptor> ScanRange::GetUnusedBuffer(
return result;
}
-ReadOutcome ScanRange::DoRead(DiskQueue* queue, int disk_id) {
+ReadOutcome ScanRange::DoReadInternal(
+ DiskQueue* queue, int disk_id, FileReader* file_reader) {
int64_t bytes_remaining = bytes_to_read_ - bytes_read_;
DCHECK_GT(bytes_remaining, 0);
@@ -202,7 +206,7 @@ ReadOutcome ScanRange::DoRead(DiskQueue* queue, int disk_id) {
(FLAGS_cache_abfs_file_handles && disk_id_ == io_mgr_->RemoteAbfsDiskId()))) {
use_file_handle_cache = true;
}
- Status read_status = file_reader_->Open(use_file_handle_cache);
+ Status read_status = file_reader->Open(use_file_handle_cache);
bool eof = false;
if (read_status.ok()) {
COUNTER_ADD_IF_NOT_NULL(reader_->active_read_thread_counter_, 1L);
@@ -210,12 +214,12 @@ ReadOutcome ScanRange::DoRead(DiskQueue* queue, int disk_id) {
if (sub_ranges_.empty()) {
DCHECK(cache_.data == nullptr);
- read_status = file_reader_->ReadFromPos(queue, offset_ + bytes_read_,
- buffer_desc->buffer_,
- min(bytes_to_read() - bytes_read_, buffer_desc->buffer_len_),
- &buffer_desc->len_, &eof);
+ read_status =
+ file_reader->ReadFromPos(queue, offset_ + bytes_read_, buffer_desc->buffer_,
+ min(bytes_to_read() - bytes_read_, buffer_desc->buffer_len_),
+ &buffer_desc->len_, &eof);
} else {
- read_status = ReadSubRanges(queue, buffer_desc.get(), &eof);
+ read_status = ReadSubRanges(queue, buffer_desc.get(), &eof, file_reader);
}
COUNTER_ADD_IF_NOT_NULL(reader_->bytes_read_counter_, buffer_desc->len_);
@@ -223,8 +227,8 @@ ReadOutcome ScanRange::DoRead(DiskQueue* queue, int disk_id) {
}
DCHECK(buffer_desc->buffer_ != nullptr);
- DCHECK(!buffer_desc->is_cached()) <<
- "Pure HDFS cache reads don't go through this code path.";
+ DCHECK(!buffer_desc->is_cached())
+ << "Pure HDFS cache reads don't go through this code path.";
if (!read_status.ok()) {
// Free buffer to release resources before we cancel the range so that all buffers
// are freed at cancellation.
@@ -252,7 +256,7 @@ ReadOutcome ScanRange::DoRead(DiskQueue* queue, int disk_id) {
// Store the state we need before calling EnqueueReadyBuffer().
bool eosr = buffer_desc->eosr();
// No more reads for this scan range - we can close it.
- if (eosr) file_reader_->Close();
+ if (eosr) file_reader->Close();
// Read successful - enqueue the buffer and return the appropriate outcome.
if (!EnqueueReadyBuffer(move(buffer_desc))) return ReadOutcome::CANCELLED;
// At this point, if eosr=true, then we cannot touch the state of this scan range
@@ -260,8 +264,41 @@ ReadOutcome ScanRange::DoRead(DiskQueue* queue, int disk_id) {
return eosr ? ReadOutcome::SUCCESS_EOSR : ReadOutcome::SUCCESS_NO_EOSR;
}
+ReadOutcome ScanRange::DoRead(DiskQueue* queue, int disk_id) {
+ FileReader* file_reader = file_reader_.get();
+ if (disk_file_ != nullptr && disk_file_->disk_type() != DiskFileType::LOCAL) {
+ // The sequence for acquiring the locks should always be from the local to
+ // the remote to avoid deadlocks.
+ shared_lock<shared_mutex> local_file_lock(*(disk_buffer_file_->GetFileLock()));
+ shared_lock<shared_mutex> remote_file_lock(*(disk_file_->GetFileLock()));
+ {
+ unique_lock<SpinLock> buffer_file_lock(*(disk_buffer_file_->GetStatusLock()));
+ unique_lock<SpinLock> file_lock(*(disk_file_->GetStatusLock()));
+ if (disk_buffer_file_->is_deleted(buffer_file_lock)
+ && disk_file_->is_deleted(file_lock)) {
+ // If both of the local buffer file and the remote file have been deleted,
+ // the only case could be the query is cancelled, so that both files are deleted.
+ return ReadOutcome::CANCELLED;
+ }
+ // If the local buffer exists, we can read from the local buffer, otherwise,
+ // we will read from the remote file system.
+ if (!disk_buffer_file_->is_deleted(buffer_file_lock)) {
+ file_reader = local_buffer_reader_.get();
+ file_ = disk_buffer_file_->path();
+ use_local_buffer_ = true;
+ } else {
+ // Read from the remote file. The remote file must be in persisted status.
+ DCHECK(disk_file_->is_persisted(file_lock));
+ use_local_buffer_ = false;
+ }
+ }
+ return DoReadInternal(queue, disk_id, file_reader);
+ }
+ return DoReadInternal(queue, disk_id, file_reader);
+}
+
Status ScanRange::ReadSubRanges(
- DiskQueue* queue, BufferDescriptor* buffer_desc, bool* eof) {
+ DiskQueue* queue, BufferDescriptor* buffer_desc, bool* eof, FileReader* file_reader) {
buffer_desc->len_ = 0;
while (buffer_desc->len() < buffer_desc->buffer_len()
&& sub_range_pos_.index < sub_ranges_.size()) {
@@ -275,7 +312,7 @@ Status ScanRange::ReadSubRanges(
cache_.data + offset, bytes_to_read);
} else {
int64_t current_bytes_read;
- Status read_status = file_reader_->ReadFromPos(queue, offset,
+ Status read_status = file_reader->ReadFromPos(queue, offset,
buffer_desc->buffer_ + buffer_desc->len(), bytes_to_read, ¤t_bytes_read,
eof);
if (!read_status.ok()) return read_status;
@@ -313,7 +350,11 @@ void ScanRange::CleanUpBuffer(const unique_lock<mutex>& scan_range_lock,
// Close the scan range if there are no more buffers in the reader and no more buffers
// will be returned to readers in future. Close() is idempotent so it is ok to call
// multiple times during cleanup so long as the range is actually finished.
- file_reader_->Close();
+ if (!use_local_buffer_) {
+ file_reader_->Close();
+ } else {
+ local_buffer_reader_->Close();
+ }
}
}
@@ -341,12 +382,14 @@ void ScanRange::Cancel(const Status& status) {
void ScanRange::CancelInternal(const Status& status, bool read_error) {
DCHECK(io_mgr_ != nullptr);
DCHECK(!status.ok());
+ FileReader* file_reader =
+ use_local_buffer_ ? local_buffer_reader_.get() : file_reader_.get();
{
// Grab both locks to make sure that we don't change 'cancel_status_' while other
// threads are in critical sections.
unique_lock<mutex> scan_range_lock(lock_);
{
- unique_lock<SpinLock> fs_lock(file_reader_->lock());
+ unique_lock<SpinLock> fs_lock(file_reader->lock());
DCHECK(Validate()) << DebugString();
// If already cancelled, preserve the original reason for cancellation. Most of the
// cleanup is not required if already cancelled, but we need to set
@@ -375,7 +418,7 @@ void ScanRange::CancelInternal(const Status& status, bool read_error) {
// TODO: IMPALA-4249 - this Close() call makes it unsafe to reuse a cancelled scan
// range, because there is no synchronisation between this Close() call and the
// client adding the ScanRange back into the IoMgr.
- if (external_buffer_tag_ != ExternalBufferTag::CACHED_BUFFER) file_reader_->Close();
+ if (external_buffer_tag_ != ExternalBufferTag::CACHED_BUFFER) file_reader->Close();
}
void ScanRange::WaitForInFlightRead() {
@@ -443,9 +486,9 @@ ScanRange::~ScanRange() {
void ScanRange::Reset(hdfsFS fs, const char* file, int64_t len, int64_t offset,
int disk_id, bool expected_local, int64_t mtime, const BufferOpts& buffer_opts,
- void* meta_data) {
- Reset(fs, file, len, offset, disk_id, expected_local, mtime, buffer_opts, {},
- meta_data);
+ void* meta_data, DiskFile* disk_file, DiskFile* disk_buffer_file) {
+ Reset(fs, file, len, offset, disk_id, expected_local, mtime, buffer_opts, {}, meta_data,
+ disk_file, disk_buffer_file);
}
ScanRange* ScanRange::AllocateScanRange(ObjectPool* obj_pool, hdfsFS fs, const char* file,
@@ -464,7 +507,8 @@ ScanRange* ScanRange::AllocateScanRange(ObjectPool* obj_pool, hdfsFS fs, const c
void ScanRange::Reset(hdfsFS fs, const char* file, int64_t len, int64_t offset,
int disk_id, bool expected_local, int64_t mtime, const BufferOpts& buffer_opts,
- vector<SubRange>&& sub_ranges, void* meta_data) {
+ vector<SubRange>&& sub_ranges, void* meta_data, DiskFile* disk_file,
+ DiskFile* disk_buffer_file) {
DCHECK(ready_buffers_.empty());
DCHECK(!read_in_flight_);
DCHECK(file != nullptr);
@@ -473,8 +517,9 @@ void ScanRange::Reset(hdfsFS fs, const char* file, int64_t len, int64_t offset,
DCHECK(buffer_opts.client_buffer_ == nullptr ||
buffer_opts.client_buffer_len_ >= len_);
fs_ = fs;
- if (fs_) {
- file_reader_ = make_unique<HdfsFileReader>(this, fs_, expected_local);
+ if (fs != nullptr) {
+ file_reader_ = make_unique<HdfsFileReader>(this, fs_, false);
+ local_buffer_reader_ = make_unique<LocalFileReader>(this);
} else {
file_reader_ = make_unique<LocalFileReader>(this);
}
@@ -484,6 +529,8 @@ void ScanRange::Reset(hdfsFS fs, const char* file, int64_t len, int64_t offset,
offset_ = offset;
disk_id_ = disk_id;
cache_options_ = buffer_opts.cache_options_;
+ disk_file_ = disk_file;
+ disk_buffer_file_ = disk_buffer_file;
// HDFS ranges must have an mtime > 0. Local ranges do not use mtime.
if (fs_) DCHECK_GT(mtime, 0);
@@ -566,6 +613,7 @@ void ScanRange::InitInternal(DiskIoMgr* io_mgr, RequestContext* reader) {
bytes_read_ = 0;
sub_range_pos_ = {};
file_reader_->ResetState();
+ if (local_buffer_reader_ != nullptr) local_buffer_reader_->ResetState();
DCHECK(Validate()) << DebugString();
}
diff --git a/be/src/runtime/tmp-file-mgr-internal.h b/be/src/runtime/tmp-file-mgr-internal.h
index 7412841..bd75239 100644
--- a/be/src/runtime/tmp-file-mgr-internal.h
+++ b/be/src/runtime/tmp-file-mgr-internal.h
@@ -19,8 +19,13 @@
#define IMPALA_RUNTIME_TMP_FILE_MGR_INTERNAL_H
#include <string>
+#include <unordered_map>
+#include <unordered_set>
+#include <boost/thread/shared_mutex.hpp>
#include "common/atomic.h"
+#include "runtime/hdfs-fs-cache.h"
+#include "runtime/io/file-writer.h"
#include "runtime/tmp-file-mgr.h"
namespace impala {
@@ -36,7 +41,9 @@ namespace impala {
class TmpFile {
public:
TmpFile(TmpFileGroup* file_group, TmpFileMgr::DeviceId device_id,
- const std::string& path);
+ const std::string& path, bool expected_local = true);
+
+ virtual ~TmpFile() {}
/// Allocates 'num_bytes' bytes in this file for a new block of data if there is
/// free capacity in this temporary directory. If there is insufficient capacity,
@@ -44,31 +51,67 @@ class TmpFile {
/// This function does not actually perform any file operations.
/// On success, sets 'offset' to the file offset of the first byte in the allocated
/// range on success.
- bool AllocateSpace(int64_t num_bytes, int64_t* offset);
+ virtual bool AllocateSpace(int64_t num_bytes, int64_t* offset) = 0;
/// Called when an IO error is encountered for this file. Logs the error and blacklists
/// the file. Returns true if the file just became blacklisted.
bool Blacklist(const ErrorMsg& msg);
- /// Delete the physical file on disk, if one was created.
+ /// Delete the physical file including buffer on disk, if one was created.
/// It is not valid to read or write to a file after calling Remove().
- Status Remove();
+ virtual Status Remove() = 0;
/// Get the disk ID that should be used for IO mgr queueing.
- int AssignDiskQueue() const;
+ int AssignDiskQueue(bool is_local_buffer = false) const;
/// Try to punch a hole in the file of size 'len' at 'offset'.
Status PunchHole(int64_t offset, int64_t len);
+ /// Returns the exact file handle for writing. For Spilling to remote, the default
+ /// is to return the local buffer file instead of the remote file.
+ virtual io::DiskFile* GetWriteFile() = 0;
+
+ /// Returns the path of the TmpFile.
+ /// If it is a remote temporary file, the path should be a remote scratch space path.
+ /// Similarly, if it is a local one, the path should be a local path.
const std::string& path() const { return path_; }
/// Caller must hold TmpFileMgr::FileGroup::lock_.
bool is_blacklisted() const { return blacklisted_; }
+ /// Returns the current length of the file.
+ int64_t len() const { return allocation_offset_; }
+
+ /// Returns the disk id of the temporary file.
+ int disk_id() const { return disk_id_; }
+
+ /// Returns if the temporary file is in local file system.
+ bool is_local() { return expected_local_; }
+
+ /// Returns the path of the local buffer of the TmpFile.
+ const string& LocalBuffPath() { return local_buffer_path_; }
+
std::string DebugString();
+ /// Helper to get the TmpDir that this file is associated with.
+ TmpFileMgr::TmpDir* GetDir();
+
+ /// Helper to get the TmpFileGroup that this file is associated with.
+ TmpFileGroup* FileGroup() const { return file_group_; }
+
+ /// Return the pointer of the disk file.
+ io::DiskFile* DiskFile() { return disk_file_.get(); }
+
+ /// Return the disk type where the file locates.
+ io::DiskFileType disk_type() { return disk_type_; }
+
private:
friend class TmpFileMgrTest;
+ friend class TmpFileLocal;
+ friend class TmpFileRemote;
+ friend class TmpFileDummy;
+ friend class TmpWriteHandle;
+
/// The name of the sub-directory that Impala creates within each configured scratch
/// directory.
const static std::string TMP_SUB_DIR_NAME;
@@ -87,7 +130,10 @@ class TmpFile {
const TmpFileMgr::DeviceId device_id_;
/// The id of the disk on which the physical file lies.
- const int disk_id_;
+ int disk_id_;
+
+ // If the file is expected to be in the local file system.
+ bool expected_local_;
/// Total bytes of the file that have been given out by AllocateSpace(). Note that
/// these bytes may not be actually using space on the filesystem, either because the
@@ -101,9 +147,286 @@ class TmpFile {
/// Protected by TmpFileMgr::FileGroup::lock_.
bool blacklisted_;
- /// Helper to get the TmpDir that this file is associated with.
- TmpFileMgr::TmpDir* GetDir();
+ /// Specify the type of the disk where the file locates.
+ io::DiskFileType disk_type_;
+
+ /// The file path of the local buffer of the TmpFile.
+ string local_buffer_path_;
+
+ /// The disk file of the temporary file which is a handle to manage the status and
+ /// operate on the physical file.
+ std::unique_ptr<io::DiskFile> disk_file_;
+};
+
+/// TmpFileLocal is a derived class of TmpFile to provide methods to handle a
+/// physical file in a temporary directory of local filesystem.
+class TmpFileLocal : public TmpFile {
+ public:
+ TmpFileLocal(TmpFileGroup* file_group, TmpFileMgr::DeviceId device_id,
+ const std::string& path, bool expected_local = true);
+
+ bool AllocateSpace(int64_t num_bytes, int64_t* offset);
+ io::DiskFile* GetWriteFile();
+ Status Remove();
+};
+
+/// TmpFileRemote is a derived class of TmpFile to provide methods to handle a
+/// physical file in a temporary directory of remote filesystem.
+///
+/// Locking:
+/// For remote temporary files, a locking mechanism is applied to guarantee a safe write,
+/// read or upload on the file.A remote temporary file can have two DiskFiles, a local
+/// buffer and a remote file.
+/// Each DiskFile owns two type of locks, a file lock and a status lock.
+/// DiskFile::lock_ -- file lock
+/// DiskFile::status_lock_ -- status lock
+/// For doing file deleting operation, a unique file lock is needed. For other types of
+/// operations on the file, like reading or writing, a shared file lock is needed to
+/// protect the file from deleting.
+/// For status transition, the thread needs to hold the status lock, details can be found
+/// in the header of TmpFileRemote.
+/// If an operation requires locks from two DiskFiles, such as an upload operation, the
+/// sequence of acquiring the lock must be from the local file to the remote file to
+/// avoid deadlocks, and file locks need to be acquired before status locks.
+///
+/// By default, there are two DiskFiles, local buffer and remote file, which are
+/// used to manage the status of the temporary file. For each DiskFile, there are
+/// three status, InWriting/Persisted/Deleted, details can be found in the header of
+/// DiskFile.
+/// Assume that:
+/// Local buffer: InWriting A, Persisted B, Deleted C
+/// Remote file: InWriting D, Persisted E, Deleted F
+/// The normal status transition procedure of a remote temporary file should be:
+/// AD ---> BD Local Buffer File Closed
+/// BD ---> BE Upload Complete
+/// BE ---> CE Local Buffer Evicted
+/// Any State ---> CF Temporary File Destoryed
+/// Most of the state transitions are done in the DiskIoMgr when an IO operation is done.
+/// For reading or changing the state, a status lock of the specific DiskFile should be
+/// acquired.
+class TmpFileRemote : public TmpFile {
+ public:
+ TmpFileRemote(TmpFileGroup* file_group, TmpFileMgr::DeviceId device_id,
+ const std::string& path, const std::string& local_buffer_path,
+ bool expected_local = false, const char* url = nullptr);
+ ~TmpFileRemote();
+
+ bool AllocateSpace(int64_t num_bytes, int64_t* offset);
+ io::DiskFile* GetWriteFile();
+ TmpFileMgr::TmpDir* GetLocalBufferDir() const;
+ Status Remove();
+
+ /// Returns the size of the file.
+ int64_t file_size() const { return file_size_; }
+
+ /// Returns if the disk buffer file pointer.
+ io::DiskFile* DiskBufferFile() { return disk_buffer_file_.get(); }
+
+ /// Set the at_capacity_ indicator to true.
+ void SetAtCapacity() {
+ DCHECK(!at_capacity_);
+ at_capacity_ = true;
+ }
+
+ /// Set the file is enqueued or not.
+ /// The function is thread-safe.
+ void SetEnqueued(bool is_enqueued) {
+ std::lock_guard<SpinLock> l(lock_);
+ DCHECK(is_enqueued != enqueued_);
+ enqueued_ = is_enqueued;
+ }
+
+ /// Set the buffer of the file is returned to the pool.
+ /// The function is thread-safe.
+ void SetBufferReturned() {
+ std::lock_guard<SpinLock> l(lock_);
+ DCHECK(!buffer_returned_);
+ buffer_returned_ = true;
+ }
+
+ /// Returns if the file is enqueued to the tmp file available pool.
+ /// The function is thread-safe.
+ bool is_enqueued() {
+ std::lock_guard<SpinLock> l(lock_);
+ return enqueued_;
+ }
+
+ /// Returns if the buffer is returned to the tmp file available pool.
+ /// The function is thread-safe.
+ bool is_buffer_returned() {
+ std::lock_guard<SpinLock> l(lock_);
+ return buffer_returned_;
+ }
+
+ private:
+ friend class TmpWriteHandle;
+ friend class TmpFileMgr;
+ friend class TmpFileGroup;
+ friend class TmpFileBufferPool;
+
+ /// The default file size of the temporary file, but the actual file size can be a
+ /// little over it if the size of the last page written to the file is over the
+ /// remaining space.
+ int64_t file_size_ = 0;
+
+ /// Bogus value of mtime for HDFS files.
+ const int64_t mtime_{100000};
+
+ /// The range for doing file uploading.
+ std::unique_ptr<io::RemoteOperRange> upload_range_;
+
+ // The pointer of the disk buffer file, which is the local buffer
+ // of the disk file when disk file is a remote disk file.
+ std::unique_ptr<io::DiskFile> disk_buffer_file_;
+
+ /// The hdfs connection used to connect to the remote scratch path.
+ hdfsFS hdfs_conn_;
+
+ /// at_capacity_ is set to true if the file can't assign space anymore when the
+ /// assigned space is equal to or just over the default file size.
+ bool at_capacity_ = false;
+
+ /// Protect below members.
+ SpinLock lock_;
+
+ /// Indicates if the file is enqueued to the pool. For debug use.
+ bool enqueued_ = false;
+
+ /// True if the buffer of the file is returned to the pool. We assume that the buffer
+ /// only returns once and only needs to be returned when the buffer space is reserved.
+ bool buffer_returned_ = false;
+};
+
+/// TmpFileDummy is a derived class of TmpFile for dummy allocation, used in
+/// TmpFileBufferPool only.
+class TmpFileDummy : public TmpFile {
+ public:
+ TmpFileDummy() : TmpFile(nullptr, -1, "") { disk_type_ = io::DiskFileType::DUMMY; }
+ bool AllocateSpace(int64_t num_bytes, int64_t* offset) { return true; }
+ io::DiskFile* GetWriteFile() { return nullptr; }
+ Status Remove() { return Status::OK(); }
+};
+
+/// Temporary file buffer pool allows the temporary files to return their buffer to the
+/// pool and can be evicted to make room for other files. The pool also provides an async
+/// way for the write ranges to wait until there is an available space to reserve before
+/// they are sent to the disk queues for writing.
+class TmpFileBufferPool {
+ public:
+ TmpFileBufferPool(TmpFileMgr* tmp_file_mgr);
+ ~TmpFileBufferPool();
+
+ // Loop to reserve space for the ranges which are put into the queue write_ranges_, the
+ // space is gained from tmp_files_avail_pool_ by calling DequeueTmpFilesPool(). Once
+ // the space is reserved, the write ranges which are belong to the same file will be
+ // sent to the disk queues to write. If an error happens during the sending to the disk
+ // queue, the callback function of the write ranges would be called with the status.
+ void TmpFileSpaceReserveThreadLoop();
+
+ /// Enqueue the write ranges to wait for buffer space reservation. All ranges of the
+ /// same file would be put into a map write_ranges_to_add_ to wait until the
+ /// reservation is done and being sent to the disk queue after that. Specially, the
+ /// first write range of a file (offset is 0) is used to doing the reservation.
+ Status EnqueueWriteRange(io::WriteRange* range, TmpFile* file);
+
+ /// Function is called to remove all enqueued write ranges belonging to the "io_ctx"
+ /// when the TmpFileGroup which the "io_ctx" belongs to is closing.
+ void RemoveWriteRanges(io::RequestContext* io_ctx);
+
+ // Enqueue the temporary file whose buffer file is available to be evicted.
+ void EnqueueTmpFilesPool(std::shared_ptr<TmpFile>& tmp_file, bool front);
+
+ // Dequeue a temporary file, whose buffer is supposed to be available being evicted,
+ // from the available pool and make room for other files' buffer.
+ Status DequeueTmpFilesPool(std::shared_ptr<TmpFile>* tmp_file, bool quick_return);
+
+ private:
+ friend class TmpFileMgr;
+ friend class TmpFileMgrTest;
+
+ /// The TmpFileMgr the TmpFileBufferPool belongs to.
+ TmpFileMgr* tmp_file_mgr_ = nullptr;
+
+ /// A pool stores the pointer of remote temporary files whose buffer are available to be
+ /// evicted.
+ /// The file is enqueued when:
+ /// 1. The file is uploaded to the remote directory.
+ /// 2. The file is deleted without unloading.
+ /// The file is enqueued by calling EnqueueTmpFilesPool(), and dequeued by
+ /// DequeueTmpFilesPool().
+ std::list<std::shared_ptr<TmpFile>> tmp_files_avail_pool_;
+
+ /// Condition variable for tmp files available pool to wait for available
+ /// tmp files to be evicted. Wait() in the DequeueTmpFilesPool() and NotifyOne()
+ /// in the EnqueueTmpFilesPool().
+ ConditionVariable tmp_files_available_cv_;
+
+ /// Protects tmp files available pool members.
+ std::mutex tmp_files_avail_pool_lock_;
+
+ /// Lock that protects below members.
+ std::mutex lock_;
+
+ /// Condition variable to signal the space reservation thread that there is work to do
+ /// or the thread should shut down. The space reservation thread will be woken up when
+ /// there is a write range added to the queue write_ranges_.
+ ConditionVariable work_available_;
+
+ /// The write ranges waiting for the buffer space reservation.
+ /// The key is the pointer of the DiskFile which the ranges are writing into.
+ std::unordered_map<io::DiskFile*, std::vector<io::WriteRange*>> write_ranges_to_add_;
+
+ /// Records the relationship of io_ctx and disk files, data inserts when add write
+ /// ranges for waiting the reservation, removes when the request context is
+ /// deconstructing when the TmpFileGroup is Closing.
+ std::unordered_map<io::RequestContext*, std::unordered_set<io::DiskFile*>>
+ io_ctx_to_file_set_map_;
+
+ /// The write ranges waiting for the buffer space reservation (only the first range of
+ /// a file).
+ std::list<io::WriteRange*> write_ranges_;
+
+ /// An index to bind the write range to its iterator in write_ranges_ and its TmpFile.
+ std::unordered_map<io::WriteRange*,
+ std::pair<std::list<io::WriteRange*>::const_iterator, TmpFile*>>
+ write_ranges_iterator_;
+
+ /// The current write range which is waiting for the reservation.
+ io::WriteRange* cur_write_range_ = nullptr;
+
+ /// The temporary file which the current write range waiting for reservation is
+ /// associated with.
+ std::shared_ptr<TmpFile> cur_tmp_file_;
+
+ /// Dummy TmpFile for the case that needs to return the space back to the pool without
+ /// having a TmpFile.
+ std::shared_ptr<TmpFile> tmp_file_dummy_;
+
+ /// True if the TmpFileMgr is destroying.
+ bool shut_down_ = false;
+
+ /// A timer metric for recording the dequeue waiting time.
+ HistogramMetric* dequeue_timer_metric_ = nullptr;
+
+ /// A helper function to add or remove all of the write ranges related to the same disk
+ /// file in the pool.
+ /// When the function is called after the reservation of a buffer file is done, it adds
+ /// all of the ranges of the file to the disk queue. If any adding fails, all other
+ /// writeback function of the ranges would be put to the vector "write_callbacks" for
+ /// the caller to callback with an error status. All ranges belonging to the file would
+ /// be removed from the pool.
+ /// When the function is called when the writing tasks are cancelled (happens when the
+ /// TmpFileGroup is closing), then is_cancelled is set to true, the function is used to
+ /// remove all the ranges belonging to the specific disk_file in the pool.
+ /// Caller should hold the lock_.
+ Status MoveWriteRangesHelper(io::DiskFile* disk_file,
+ std::vector<TmpFileMgr::WriteDoneCallback>* write_callbacks, bool is_cancelled);
+
+ /// Internal function for RemoveWriteRanges();
+ void RemoveWriteRangesInternal(io::RequestContext* io_ctx,
+ std::vector<TmpFileMgr::WriteDoneCallback>* write_callbacks);
};
+
} // namespace impala
#endif
diff --git a/be/src/runtime/tmp-file-mgr-test.cc b/be/src/runtime/tmp-file-mgr-test.cc
index 0a5ec60..22d0ce0e 100644
--- a/be/src/runtime/tmp-file-mgr-test.cc
+++ b/be/src/runtime/tmp-file-mgr-test.cc
@@ -52,6 +52,8 @@ DECLARE_bool(disk_spill_punch_holes);
#ifndef NDEBUG
DECLARE_int32(stress_scratch_write_delay_ms);
#endif
+DECLARE_string(remote_tmp_file_size);
+DECLARE_bool(allow_spill_to_hdfs);
namespace impala {
@@ -62,6 +64,11 @@ static const int64_t MEGABYTE = 1024L * KILOBYTE;
static const int64_t GIGABYTE = 1024L * MEGABYTE;
static const int64_t TERABYTE = 1024L * GIGABYTE;
+/// For testing spill to remote.
+static const string HDFS_LOCAL_URL = "hdfs://localhost";
+static const string REMOTE_URL = HDFS_LOCAL_URL;
+static const string LOCAL_BUFFER_PATH = "/tmp/tmp-file-mgr-test-buffer";
+
class TmpFileMgrTest : public ::testing::Test {
public:
static const int DEFAULT_PRIORITY = numeric_limits<int>::max();
@@ -73,6 +80,8 @@ class TmpFileMgrTest : public ::testing::Test {
#ifndef NDEBUG
FLAGS_stress_scratch_write_delay_ms = 0;
#endif
+ FLAGS_remote_tmp_file_size = "8MB";
+ FLAGS_allow_spill_to_hdfs = true;
metrics_.reset(new MetricGroup("tmp-file-mgr-test"));
profile_ = RuntimeProfile::Create(&obj_pool_, "tmp-file-mgr-test");
@@ -180,6 +189,10 @@ class TmpFileMgrTest : public ::testing::Test {
group->io_ctx_->Cancel();
}
+ static void SetWriteRangeContext(WriteRange* write_range, TmpFileGroup* group) {
+ write_range->SetRequestContext(group->io_ctx_.get());
+ }
+
/// Helper to get the # of bytes allocated by the group. Validates that the sum across
/// all files equals this total.
static int64_t BytesAllocated(TmpFileGroup* group) {
@@ -190,7 +203,7 @@ class TmpFileMgrTest : public ::testing::Test {
EXPECT_GE(allocated, reclaimed);
bytes_allocated += allocated - reclaimed;
}
- EXPECT_EQ(bytes_allocated, group->current_bytes_allocated_);
+ EXPECT_EQ(bytes_allocated, group->current_bytes_allocated_.Load());
return bytes_allocated;
}
@@ -238,7 +251,16 @@ class TmpFileMgrTest : public ::testing::Test {
void TestBlockVerification();
/// Implementation of TestCompressBufferManagment
- void TestCompressBufferManagement();
+ void TestCompressBufferManagement(bool spill_to_remote, bool wait_upload);
+
+ /// Helper function to do test on TmpFileBufferPool.
+ void TestTmpFileBufferPoolHelper(TmpFileMgr& tmp_file_mgr,
+ unique_ptr<TmpFileGroup>* tmp_file_group,
+ boost::scoped_ptr<io::WriteRange>& write_range,
+ WriteRange::WriteDoneCallback& callback);
+
+ /// Check the TmpFileBufferPool after the test case tears down.
+ void TestTmpFileBufferPoolTearDown(TmpFileMgr& tmp_file_mgr);
ObjectPool obj_pool_;
scoped_ptr<MetricGroup> metrics_;
@@ -944,15 +966,41 @@ TEST_F(TmpFileMgrTest, TestDirectoryLimitParsing) {
// Test compression buffer memory management for reads and writes.
TEST_F(TmpFileMgrTest, TestCompressBufferManagement) {
FLAGS_disk_spill_encryption = false;
- TestCompressBufferManagement();
+ TestCompressBufferManagement(false, false);
}
TEST_F(TmpFileMgrTest, TestCompressBufferManagementEncrypted) {
FLAGS_disk_spill_encryption = true;
- TestCompressBufferManagement();
+ TestCompressBufferManagement(false, false);
+}
+
+// Test spill to a remote directory with compression or encryption.
+TEST_F(TmpFileMgrTest, TestCompressBufferManagementRemote) {
+ FLAGS_disk_spill_encryption = false;
+ TestCompressBufferManagement(true, false);
+}
+
+TEST_F(TmpFileMgrTest, TestCompressBufferManagementEncryptedRemote) {
+ FLAGS_disk_spill_encryption = true;
+ TestCompressBufferManagement(true, false);
+}
+
+// Test spill to a remote directory with compression or encryption, and wait for the
+// upload finished.
+TEST_F(TmpFileMgrTest, TestCompressBufferManagementRemoteUpload) {
+ FLAGS_disk_spill_encryption = false;
+ FLAGS_remote_tmp_file_size = "8B";
+ TestCompressBufferManagement(true, true);
+}
+
+TEST_F(TmpFileMgrTest, TestCompressBufferManagementEncryptedRemoteUpload) {
+ FLAGS_disk_spill_encryption = true;
+ FLAGS_remote_tmp_file_size = "8B";
+ TestCompressBufferManagement(true, true);
}
-void TmpFileMgrTest::TestCompressBufferManagement() {
+void TmpFileMgrTest::TestCompressBufferManagement(
+ bool spill_to_remote, bool wait_upload) {
// Data string should be long and redundant enough to be compressible.
const string DATA = "the quick brown fox jumped over the lazy dog"
"the fast red fox leaped over the sleepy dog";
@@ -964,8 +1012,14 @@ void TmpFileMgrTest::TestCompressBufferManagement() {
// for 'data' (since compression buffers need to be slightly larger than
// uncompressed data).
TmpFileMgr tmp_file_mgr;
- ASSERT_OK(tmp_file_mgr.InitCustom(
- vector<string>{"/tmp/tmp-file-mgr-test.1"}, true, "lz4", true, metrics_.get()));
+ if (spill_to_remote) {
+ RemoveAndCreateDirs(vector<string>{LOCAL_BUFFER_PATH});
+ ASSERT_OK(tmp_file_mgr.InitCustom(vector<string>{REMOTE_URL, LOCAL_BUFFER_PATH}, true,
+ "lz4", true, metrics_.get()));
+ } else {
+ ASSERT_OK(tmp_file_mgr.InitCustom(
+ vector<string>{"/tmp/tmp-file-mgr-test.1"}, true, "lz4", true, metrics_.get()));
+ }
MemTracker* compressed_buffer_tracker = tmp_file_mgr.compressed_buffer_tracker();
TmpFileGroup file_group(&tmp_file_mgr, io_mgr(), profile_, TUniqueId());
MemRange data_mem_range(reinterpret_cast<uint8_t*>(&data[0]), data.size());
@@ -1002,6 +1056,25 @@ void TmpFileMgrTest::TestCompressBufferManagement() {
EXPECT_EQ(0, compressed_buffer_tracker->consumption())
<< "No memory should be consumed when no reads or writes in-flight";
+ if (spill_to_remote && wait_upload) {
+ EXPECT_NE(compressed_handle->file_, uncompressed_handle->file_);
+ auto wait_upload_func = [&](TmpFile* tmp_file) {
+ // Wait until the file has been uploaded to remote dir.
+ // Should be finished in 2 seconds.
+ int wait_times = 10;
+ while (wait_times-- > 0) {
+ if (tmp_file->DiskFile()->GetFileStatus() == io::DiskFileStatus::PERSISTED) break;
+ usleep(200 * 1000);
+ }
+ EXPECT_EQ(tmp_file->DiskFile()->GetFileStatus(), io::DiskFileStatus::PERSISTED);
+ // Remove the local buffer to enforce reading from the remote file.
+ tmp_file->GetWriteFile()->SetStatus(io::DiskFileStatus::DELETED);
+ EXPECT_OK(FileSystemUtil::RemovePaths({tmp_file->GetWriteFile()->path()}));
+ };
+ wait_upload_func(compressed_handle->file_);
+ wait_upload_func(uncompressed_handle->file_);
+ }
+
vector<uint8_t> tmp(data.size());
vector<uint8_t> big_tmp(big_data.size());
// Check behaviour when reading compressed range with available memory.
@@ -1189,4 +1262,499 @@ TEST_F(TmpFileMgrTest, TestPriorityBasedSpilling) {
file_group1.Close();
}
+
+/// Test the case we have one remote dir with one local dir as buffer.
+TEST_F(TmpFileMgrTest, TestRemoteOneDir) {
+ vector<string> tmp_dirs({LOCAL_BUFFER_PATH});
+ TmpFileMgr tmp_file_mgr;
+ RemoveAndCreateDirs(tmp_dirs);
+ tmp_dirs.push_back(REMOTE_URL);
+ ASSERT_OK(tmp_file_mgr.InitCustom(tmp_dirs, true, "", false, metrics_.get()));
+ TUniqueId id;
+ TmpFileGroup file_group(&tmp_file_mgr, io_mgr(), profile_, id);
+ int64_t alloc_size = 1024;
+
+ EXPECT_EQ(1, tmp_file_mgr.NumActiveTmpDevices());
+ vector<TmpFileMgr::DeviceId> devices = tmp_file_mgr.ActiveTmpDevices();
+ EXPECT_EQ(1, devices.size());
+ TmpFile* file = nullptr;
+ int64_t offset = -1;
+ ASSERT_OK(GroupAllocateSpace(&file_group, alloc_size, &file, &offset));
+ ASSERT_TRUE(file != nullptr);
+ EXPECT_EQ(0, offset);
+ // Check the prefix is the expected temporary directory.
+ EXPECT_EQ(0, file->path().find(tmp_dirs[1]));
+ file_group.Close();
+ CheckMetrics(&tmp_file_mgr);
+}
+
+/// Test that reporting a write error is possible but does not result in
+/// blacklisting the device.
+TEST_F(TmpFileMgrTest, TestRemoteDirReportError) {
+ int64_t alloc_size = 1024;
+ int64_t alloc_size2 = 512;
+ FLAGS_remote_tmp_file_size = "2K";
+ vector<string> tmp_dirs({LOCAL_BUFFER_PATH});
+ TmpFileMgr tmp_file_mgr;
+ RemoveAndCreateDirs(tmp_dirs);
+ tmp_dirs.push_back(REMOTE_URL);
+ ASSERT_OK(tmp_file_mgr.InitCustom(tmp_dirs, false, "", false, metrics_.get()));
+ TUniqueId id;
+ TmpFileGroup file_group(&tmp_file_mgr, io_mgr(), profile_, id);
+
+ EXPECT_EQ(1, tmp_file_mgr.NumActiveTmpDevices());
+ vector<TmpFileMgr::DeviceId> devices = tmp_file_mgr.ActiveTmpDevices();
+ EXPECT_EQ(1, devices.size());
+ CheckMetrics(&tmp_file_mgr);
+
+ TmpFile* good_file = nullptr;
+ int64_t offset = -1;
+ ASSERT_OK(GroupAllocateSpace(&file_group, alloc_size, &good_file, &offset));
+ ASSERT_TRUE(good_file != nullptr);
+ EXPECT_EQ(0, offset);
+
+ ASSERT_OK(GroupAllocateSpace(&file_group, alloc_size, &good_file, &offset));
+ ASSERT_TRUE(good_file != nullptr);
+ EXPECT_EQ(alloc_size, offset);
+
+ // A new file should be created.
+ TmpFile* bad_file = nullptr;
+ ASSERT_OK(GroupAllocateSpace(&file_group, alloc_size, &bad_file, &offset));
+ ASSERT_TRUE(bad_file != nullptr);
+ EXPECT_EQ(0, offset);
+
+ // Inject an error on one device so that we can validate it is handled correctly.
+ ErrorMsg errmsg(TErrorCode::GENERAL, "A fake error");
+ bad_file->Blacklist(errmsg);
+
+ // File-level blacklisting is enabled but not device-level.
+ EXPECT_TRUE(bad_file->is_blacklisted());
+ // The bad device should still be active.
+ EXPECT_EQ(1, tmp_file_mgr.NumActiveTmpDevices());
+ vector<TmpFileMgr::DeviceId> devices_after = tmp_file_mgr.ActiveTmpDevices();
+ EXPECT_EQ(1, devices_after.size());
+ CheckMetrics(&tmp_file_mgr);
+
+ // Attempts to expand bad file should change a file.
+ TmpFile* bad_file2;
+ TmpFile* good_file2;
+ ASSERT_OK(GroupAllocateSpace(&file_group, alloc_size2, &bad_file2, &offset));
+ EXPECT_EQ(0, offset);
+ EXPECT_NE(bad_file, bad_file2);
+ ASSERT_OK(GroupAllocateSpace(&file_group, alloc_size, &good_file2, &offset));
+ EXPECT_EQ(alloc_size2, offset);
+ EXPECT_EQ(bad_file2, good_file2);
+ file_group.Close();
+ CheckMetrics(&tmp_file_mgr);
+}
+
+TEST_F(TmpFileMgrTest, TestRemoteAllocateNonWritable) {
+ vector<string> tmp_dirs({LOCAL_BUFFER_PATH});
+ TmpFileMgr tmp_file_mgr;
+ RemoveAndCreateDirs(tmp_dirs);
+ tmp_dirs.push_back(REMOTE_URL);
+ ASSERT_OK(tmp_file_mgr.InitCustom(tmp_dirs, true, "", false, metrics_.get()));
+ TUniqueId id;
+ TmpFileGroup file_group(&tmp_file_mgr, io_mgr(), profile_, id);
+ int64_t alloc_size = 1024;
+
+ EXPECT_EQ(1, tmp_file_mgr.NumActiveTmpDevices());
+ vector<TmpFileMgr::DeviceId> devices = tmp_file_mgr.ActiveTmpDevices();
+ EXPECT_EQ(1, devices.size());
+ TmpFile* file = nullptr;
+ int64_t offset = -1;
+ ASSERT_OK(GroupAllocateSpace(&file_group, alloc_size, &file, &offset));
+ ASSERT_TRUE(file != nullptr);
+ EXPECT_EQ(0, offset);
+
+ // Make local buffer non-writable and test allocation at different stages:
+ // new file creation, files with no allocated blocks. files with allocated space.
+ // No errors should be encountered during allocation since allocation is purely logical.
+ chmod(LOCAL_BUFFER_PATH.c_str(), 0);
+ ASSERT_OK(GroupAllocateSpace(&file_group, alloc_size, &file, &offset));
+ ASSERT_TRUE(file != nullptr);
+ EXPECT_EQ(alloc_size, offset);
+ chmod(LOCAL_BUFFER_PATH.c_str(), S_IRWXU);
+ file_group.Close();
+}
+
+TEST_F(TmpFileMgrTest, TestRemoteScratchLimit) {
+ int64_t alloc_size = 512;
+ int64_t limit = alloc_size * 2;
+ FLAGS_remote_tmp_file_size = "1K";
+ vector<string> tmp_dirs({LOCAL_BUFFER_PATH});
+ TmpFileMgr tmp_file_mgr;
+ RemoveAndCreateDirs(tmp_dirs);
+ tmp_dirs.push_back(REMOTE_URL);
+
+ ASSERT_OK(tmp_file_mgr.InitCustom(tmp_dirs, true, "", false, metrics_.get()));
+ TUniqueId id;
+ TmpFileGroup file_group(&tmp_file_mgr, io_mgr(), profile_, id, limit);
+
+ // Test individual limit is enforced.
+ Status status;
+ int64_t offset;
+ TmpFile* alloc_file = nullptr;
+
+ // Alloc from file 1 should succeed.
+ ASSERT_OK(GroupAllocateSpace(&file_group, alloc_size, &alloc_file, &offset));
+ ASSERT_TRUE(alloc_file != nullptr);
+ ASSERT_EQ(0, offset);
+
+ // Allocate up to the max.
+ ASSERT_OK(GroupAllocateSpace(&file_group, alloc_size, &alloc_file, &offset));
+ ASSERT_TRUE(alloc_file != nullptr);
+ ASSERT_EQ(alloc_size, offset);
+
+ // Test aggregate limit is enforced.
+ status = GroupAllocateSpace(&file_group, 1, &alloc_file, &offset);
+ ASSERT_FALSE(status.ok());
+ ASSERT_EQ(status.code(), TErrorCode::SCRATCH_LIMIT_EXCEEDED);
+ ASSERT_NE(string::npos, status.msg().msg().find(GetBackendString()));
+
+ // Check HWM metrics
+ checkHWMMetrics(limit, limit);
+ file_group.Close();
+ checkHWMMetrics(0, limit);
+}
+
+/// Test writing a single record to the remote dir and read it back from
+/// the local buffer and verify the data correctness.
+/// There is a testcase to verify the data correctness in DiskIoMgrTest for
+/// reading from the remote file.
+TEST_F(TmpFileMgrTest, TestRemoteWriteRange) {
+ FLAGS_disk_spill_encryption = false;
+ vector<string> tmp_dirs({LOCAL_BUFFER_PATH});
+ TmpFileMgr tmp_file_mgr;
+ RemoveAndCreateDirs(tmp_dirs);
+ tmp_dirs.push_back(REMOTE_URL);
+
+ ASSERT_OK(tmp_file_mgr.InitCustom(tmp_dirs, true, "", false, metrics_.get()));
+ TUniqueId id;
+ TmpFileGroup file_group(&tmp_file_mgr, io_mgr(), profile_, id);
+ string data = "the quick brown fox jumped over the lazy dog";
+ MemRange data_mem_range(reinterpret_cast<uint8_t*>(&data[0]), data.size());
+
+ // Start a write in flight, which should write it to disk without encryption.
+ unique_ptr<TmpWriteHandle> handle;
+ WriteRange::WriteDoneCallback callback =
+ bind(mem_fn(&TmpFileMgrTest::SignalCallback), this, _1);
+ ASSERT_OK(file_group.Write(data_mem_range, callback, &handle));
+
+ WaitForWrite(handle.get());
+ WaitForCallbacks(1);
+
+ vector<uint8_t> tmp;
+ tmp.resize(data.size());
+ Status read_status = file_group.Read(handle.get(), MemRange(tmp.data(), tmp.size()));
+ EXPECT_TRUE(read_status.ok());
+ EXPECT_EQ(tmp.size(), data.size());
+ for (int i = 0; i < data.size(); i++) {
+ EXPECT_EQ(tmp[i], data[i]);
+ }
+ file_group.Close();
+}
+
+/// Test writing a single record with encryption to a remote dir with local
+/// buffer.
+TEST_F(TmpFileMgrTest, TestRemoteBlockVerification) {
+ FLAGS_disk_spill_encryption = true;
+ vector<string> tmp_dirs({LOCAL_BUFFER_PATH});
+ TmpFileMgr tmp_file_mgr;
+ RemoveAndCreateDirs(tmp_dirs);
+ tmp_dirs.push_back(REMOTE_URL);
+
+ ASSERT_OK(tmp_file_mgr.InitCustom(tmp_dirs, true, "", false, metrics_.get()));
+ TUniqueId id;
+ TmpFileGroup file_group(&tmp_file_mgr, io_mgr(), profile_, id);
+ string data = "the quick brown fox jumped over the lazy dog";
+ MemRange data_mem_range(reinterpret_cast<uint8_t*>(&data[0]), data.size());
+
+ // Start a write in flight, which should encrypt the data and write it to disk.
+ unique_ptr<TmpWriteHandle> handle;
+ WriteRange::WriteDoneCallback callback =
+ bind(mem_fn(&TmpFileMgrTest::SignalCallback), this, _1);
+ ASSERT_OK(file_group.Write(data_mem_range, callback, &handle));
+ string file_path = handle->TmpFileBufferPath();
+
+ WaitForWrite(handle.get());
+ WaitForCallbacks(1);
+
+ // Modify the data in the local buffer file and check that a read error occurs.
+ LOG(INFO) << "Corrupting " << file_path;
+ uint8_t corrupt_byte = data[0] ^ 1;
+ FILE* file = fopen(file_path.c_str(), "rb+");
+ ASSERT_TRUE(file != nullptr);
+ ASSERT_EQ(corrupt_byte, fputc(corrupt_byte, file));
+ ASSERT_EQ(0, fclose(file));
+ vector<uint8_t> tmp;
+ tmp.resize(data.size());
+ Status read_status = file_group.Read(handle.get(), MemRange(tmp.data(), tmp.size()));
+ LOG(INFO) << read_status.GetDetail();
+ EXPECT_EQ(TErrorCode::SCRATCH_READ_VERIFY_FAILED, read_status.code())
+ << read_status.GetDetail();
+
+ // Modify the data in memory. Restoring the data should fail.
+ LOG(INFO) << "Corrupting data in memory";
+ data[0] = corrupt_byte;
+ Status restore_status = file_group.RestoreData(move(handle), data_mem_range);
+ LOG(INFO) << restore_status.GetDetail();
+ EXPECT_EQ(TErrorCode::SCRATCH_READ_VERIFY_FAILED, restore_status.code())
+ << restore_status.GetDetail();
+
+ file_group.Close();
+ test_env_->TearDownQueries();
+}
+
+/// Test when the total file size reaches the remote directory limits.
+TEST_F(TmpFileMgrTest, TestRemoteDirectoryLimits) {
+ vector<string> tmp_dirs{{LOCAL_BUFFER_PATH}};
+ TmpFileMgr tmp_file_mgr;
+ RemoveAndCreateDirs(tmp_dirs);
+ tmp_dirs.push_back(REMOTE_URL);
+ int64_t alloc_size = 1024;
+ int64_t file_size = 1024;
+ FLAGS_remote_tmp_file_size = "1K";
+
+ ASSERT_OK(tmp_file_mgr.InitCustom(tmp_dirs, true, "", false, metrics_.get()));
+ TUniqueId id;
+ TmpFileGroup file_group_1(&tmp_file_mgr, io_mgr(), profile_, id);
+ TmpFileGroup file_group_2(&tmp_file_mgr, io_mgr(), profile_, id);
+
+ IntGauge* dir_usage = metrics_->FindMetricForTesting<IntGauge>(
+ "tmp-file-mgr.scratch-space-bytes-used.dir-0");
+ IntGauge* local_buff_usage = metrics_->FindMetricForTesting<IntGauge>(
+ "tmp-file-mgr.local-buff-bytes-used.dir-0");
+
+ int64_t offset;
+ TmpFile* alloc_file;
+
+ ASSERT_OK(GroupAllocateSpace(&file_group_1, alloc_size, &alloc_file, &offset));
+ EXPECT_EQ(file_size, dir_usage->GetValue());
+ EXPECT_EQ(file_size, local_buff_usage->GetValue());
+
+ // This time we should hit the limit on the first directory. Do this from a
+ // different file group to show that limits are enforced across file groups.
+ ASSERT_OK(GroupAllocateSpace(&file_group_2, alloc_size, &alloc_file, &offset));
+ EXPECT_EQ(2 * file_size, dir_usage->GetValue());
+ EXPECT_EQ(2 * file_size, local_buff_usage->GetValue());
+
+ file_group_2.Close();
+ // Metrics should be decremented when the file groups delete the underlying files.
+ EXPECT_EQ(file_size, dir_usage->GetValue());
+ EXPECT_EQ(file_size, local_buff_usage->GetValue());
+
+ // We should be able to reuse the space freed up.
+ ASSERT_OK(GroupAllocateSpace(&file_group_1, alloc_size, &alloc_file, &offset));
+ EXPECT_EQ(2 * file_size, dir_usage->GetValue());
+ EXPECT_EQ(2 * file_size, local_buff_usage->GetValue());
+
+ file_group_1.Close();
+ EXPECT_EQ(0, dir_usage->GetValue());
+ EXPECT_EQ(0, local_buff_usage->GetValue());
+}
+
+/// Test when both remote and local dir exist, if the directory limits
+/// work.
+TEST_F(TmpFileMgrTest, TestMixDirectoryLimits) {
+ vector<string> tmp_create_dirs{{LOCAL_BUFFER_PATH, "/tmp/tmp-file-mgr-test-local"}};
+ vector<string> tmp_dirs{{LOCAL_BUFFER_PATH, "/tmp/tmp-file-mgr-test-local:1024"}};
+ TmpFileMgr tmp_file_mgr;
+ RemoveAndCreateDirs(tmp_create_dirs);
+ tmp_dirs.push_back(REMOTE_URL);
+ int64_t alloc_size = 1024;
+ int64_t file_size = 1024;
+ FLAGS_remote_tmp_file_size = "1K";
+
+ ASSERT_OK(tmp_file_mgr.InitCustom(tmp_dirs, false, "", false, metrics_.get()));
+ TUniqueId id;
+ TmpFileGroup file_group_1(&tmp_file_mgr, io_mgr(), profile_, id);
+ TmpFileGroup file_group_2(&tmp_file_mgr, io_mgr(), profile_, id);
+
+ IntGauge* dir_usage_local = metrics_->FindMetricForTesting<IntGauge>(
+ "tmp-file-mgr.scratch-space-bytes-used.dir-0");
+ IntGauge* dir_usage_remote = metrics_->FindMetricForTesting<IntGauge>(
+ "tmp-file-mgr.scratch-space-bytes-used.dir-1");
+ IntGauge* local_buff_usage = metrics_->FindMetricForTesting<IntGauge>(
+ "tmp-file-mgr.local-buff-bytes-used.dir-0");
+
+ ASSERT_TRUE(dir_usage_remote != nullptr);
+ ASSERT_TRUE(dir_usage_local != nullptr);
+ ASSERT_TRUE(local_buff_usage != nullptr);
+
+ int64_t offset;
+ TmpFile* alloc_file;
+
+ // Allocate space from local.
+ ASSERT_OK(GroupAllocateSpace(&file_group_1, alloc_size, &alloc_file, &offset));
+ EXPECT_EQ(file_size, dir_usage_local->GetValue());
+ EXPECT_EQ(0, dir_usage_remote->GetValue());
+ EXPECT_EQ(0, local_buff_usage->GetValue());
+
+ // This time we should hit the limit on the local directory. Should allocate
+ // from the remote directory.
+ ASSERT_OK(GroupAllocateSpace(&file_group_2, alloc_size, &alloc_file, &offset));
+ EXPECT_EQ(file_size, dir_usage_local->GetValue());
+ EXPECT_EQ(file_size, dir_usage_remote->GetValue());
+ EXPECT_EQ(file_size, local_buff_usage->GetValue());
+
+ file_group_2.Close();
+ // Metrics of the remote directory should be decremented when the
+ // second file groups delete the underlying files.
+ EXPECT_EQ(file_size, dir_usage_local->GetValue());
+ EXPECT_EQ(0, dir_usage_remote->GetValue());
+ EXPECT_EQ(0, local_buff_usage->GetValue());
+
+ // Should allocate from the remote directory again.
+ ASSERT_OK(GroupAllocateSpace(&file_group_1, alloc_size, &alloc_file, &offset));
+ EXPECT_EQ(file_size, dir_usage_local->GetValue());
+ EXPECT_EQ(file_size, dir_usage_remote->GetValue());
+ EXPECT_EQ(file_size, local_buff_usage->GetValue());
+
+ file_group_1.Close();
+ EXPECT_EQ(0, dir_usage_local->GetValue());
+ EXPECT_EQ(0, dir_usage_remote->GetValue());
+ EXPECT_EQ(0, local_buff_usage->GetValue());
+}
+
+/// Config a value bigger than the max remote file size allowed, should be
+/// adjusted to the max size.
+TEST_F(TmpFileMgrTest, TestMixTmpFileLimits) {
+ vector<string> tmp_create_dirs{{LOCAL_BUFFER_PATH}};
+ vector<string> tmp_dirs{{LOCAL_BUFFER_PATH}};
+ TmpFileMgr tmp_file_mgr;
+ RemoveAndCreateDirs(tmp_create_dirs);
+ tmp_dirs.push_back(REMOTE_URL);
+ int64_t alloc_size = 1024;
+ int64_t file_size = 256 * 1024 * 1024;
+ int64_t offset;
+ TmpFile* alloc_file;
+ FLAGS_remote_tmp_file_size = "512MB";
+
+ ASSERT_OK(tmp_file_mgr.InitCustom(tmp_dirs, false, "", false, metrics_.get()));
+ TUniqueId id;
+ TmpFileGroup file_group(&tmp_file_mgr, io_mgr(), profile_, id);
+ EXPECT_EQ(tmp_file_mgr.GetRemoteTmpFileSize(), file_size);
+ IntGauge* dir_usage_local = metrics_->FindMetricForTesting<IntGauge>(
+ "tmp-file-mgr.scratch-space-bytes-used.dir-0");
+ ASSERT_TRUE(dir_usage_local != nullptr);
+ // Allocate space from local.
+ ASSERT_OK(GroupAllocateSpace(&file_group, alloc_size, &alloc_file, &offset));
+ EXPECT_EQ(file_size, dir_usage_local->GetValue());
+ file_group.Close();
+}
+
+void TmpFileMgrTest::TestTmpFileBufferPoolHelper(TmpFileMgr& tmp_file_mgr,
+ unique_ptr<TmpFileGroup>* tmp_file_group,
+ boost::scoped_ptr<io::WriteRange>& write_range,
+ WriteRange::WriteDoneCallback& callback) {
+ TUniqueId id;
+ vector<string> tmp_create_dirs{{LOCAL_BUFFER_PATH}};
+ vector<string> tmp_dirs{{Substitute(LOCAL_BUFFER_PATH + ":$0", 2048)}};
+ RemoveAndCreateDirs(tmp_create_dirs);
+ tmp_dirs.push_back(REMOTE_URL);
+ int64_t alloc_size = 1024;
+ int64_t file_size = 1024;
+ FLAGS_remote_tmp_file_size = "1KB";
+ int64_t offset;
+ TmpFile *file1, *file2, *file3;
+ cb_counter_ = 0;
+
+ ASSERT_OK(tmp_file_mgr.InitCustom(tmp_dirs, false, "", false, metrics_.get()));
+ unique_ptr<TmpFileGroup> unique_file_group =
+ make_unique<TmpFileGroup>(&tmp_file_mgr, io_mgr(), profile_, id);
+ auto file_group = unique_file_group.get();
+ EXPECT_EQ(tmp_file_mgr.GetRemoteTmpFileSize(), file_size);
+ IntGauge* dir_usage_local = metrics_->FindMetricForTesting<IntGauge>(
+ "tmp-file-mgr.scratch-space-bytes-used.dir-0");
+ ASSERT_TRUE(dir_usage_local != nullptr);
+ ASSERT_OK(GroupAllocateSpace(file_group, alloc_size, &file1, &offset));
+ EXPECT_EQ(file_size, dir_usage_local->GetValue());
+ ASSERT_OK(GroupAllocateSpace(file_group, alloc_size, &file2, &offset));
+ EXPECT_NE(file1, file2);
+ EXPECT_TRUE(file1->GetWriteFile()->IsSpaceReserved());
+ EXPECT_TRUE(file2->GetWriteFile()->IsSpaceReserved());
+ ASSERT_OK(GroupAllocateSpace(file_group, alloc_size, &file3, &offset));
+ EXPECT_NE(file2, file3);
+
+ write_range.reset(new WriteRange(
+ file3->path(), offset, file3->AssignDiskQueue(!file3->is_local()), callback));
+ const char* dummy_data = "dummy data";
+ write_range->SetData((const uint8_t*)dummy_data, strlen(dummy_data));
+ write_range->SetDiskFile(file3->GetWriteFile());
+ SetWriteRangeContext(write_range.get(), file_group);
+ // The space hasn't been reserved for file3 because the pool size is 2K, and there are
+ // two files in use, we use the AsyncWriteRange() function to wait untial a file is
+ // available to be evicted, then finish the writing of file3.
+ EXPECT_FALSE(file3->GetWriteFile()->IsSpaceReserved());
+ ASSERT_OK(tmp_file_mgr.AsyncWriteRange(write_range.get(), file3));
+ *tmp_file_group = move(unique_file_group);
+}
+
+void TmpFileMgrTest::TestTmpFileBufferPoolTearDown(TmpFileMgr& tmp_file_mgr) {
+ auto tmp_file_pool = tmp_file_mgr.tmp_dirs_remote_ctrl_.tmp_file_pool_.get();
+ ASSERT_TRUE(tmp_file_pool->io_ctx_to_file_set_map_.empty());
+ ASSERT_TRUE(tmp_file_pool->write_ranges_to_add_.empty());
+ ASSERT_TRUE(tmp_file_pool->write_ranges_.empty());
+ ASSERT_TRUE(tmp_file_pool->write_ranges_iterator_.empty());
+}
+
+// Test one write is waiting in the pool for the reservation, but file group is closed
+// before the reservation is done. The write's callback function should be called.
+TEST_F(TmpFileMgrTest, TestTmpFileBufferPoolOneWriteCancel) {
+ TmpFileMgr tmp_file_mgr;
+ unique_ptr<TmpFileGroup> file_group;
+ boost::scoped_ptr<io::WriteRange> write_range;
+ WriteRange::WriteDoneCallback callback =
+ bind(mem_fn(&TmpFileMgrTest::SignalCallback), this, _1);
+ TestTmpFileBufferPoolHelper(tmp_file_mgr, &file_group, write_range, callback);
+ file_group->Close();
+ WaitForCallbacks(1);
+ TestTmpFileBufferPoolTearDown(tmp_file_mgr);
+}
+
+// Test two writes are waiting in the pool for the reservation, but file group is closed
+// before the reservation is done. The writes' callback functions should be called.
+TEST_F(TmpFileMgrTest, TestTmpFileBufferPoolTwoWritesCancel) {
+ TmpFileMgr tmp_file_mgr;
+ unique_ptr<TmpFileGroup> file_group;
+ boost::scoped_ptr<io::WriteRange> write_range;
+ WriteRange::WriteDoneCallback callback =
+ bind(mem_fn(&TmpFileMgrTest::SignalCallback), this, _1);
+ TestTmpFileBufferPoolHelper(tmp_file_mgr, &file_group, write_range, callback);
+ TmpFile* file4;
+ int64_t alloc_size = 1024;
+ int64_t offset;
+ TmpFileGroup* tmp_file_group = file_group.get();
+ ASSERT_OK(GroupAllocateSpace(tmp_file_group, alloc_size, &file4, &offset));
+ boost::scoped_ptr<io::WriteRange> write_range2;
+ write_range2.reset(new WriteRange(
+ file4->path(), offset, file4->AssignDiskQueue(!file4->is_local()), callback));
+ const char* dummy_data = "dummy data";
+ write_range2->SetData((const uint8_t*)dummy_data, strlen(dummy_data));
+ write_range2->SetDiskFile(file4->GetWriteFile());
+ SetWriteRangeContext(write_range2.get(), tmp_file_group);
+ EXPECT_FALSE(file4->GetWriteFile()->IsSpaceReserved());
+ ASSERT_OK(tmp_file_mgr.AsyncWriteRange(write_range2.get(), file4));
+ file_group->Close();
+ WaitForCallbacks(2);
+ TestTmpFileBufferPoolTearDown(tmp_file_mgr);
+}
+
+// Test one write is waiting in the pool for the reservation, when a file is returned to
+// the buffer pool, the reservation should be done.
+TEST_F(TmpFileMgrTest, TestTmpFileBufferPoolOneWriteDone) {
+ TmpFileMgr tmp_file_mgr;
+ unique_ptr<TmpFileGroup> file_group;
+ boost::scoped_ptr<io::WriteRange> write_range;
+ WriteRange::WriteDoneCallback callback =
+ bind(mem_fn(&TmpFileMgrTest::SignalCallback), this, _1);
+ TestTmpFileBufferPoolHelper(tmp_file_mgr, &file_group, write_range, callback);
+ // Return a dummy file to the pool.
+ tmp_file_mgr.EnqueueTmpFilesPoolDummyFile();
+ WaitForCallbacks(1);
+ file_group->Close();
+ TestTmpFileBufferPoolTearDown(tmp_file_mgr);
+}
+
} // namespace impala
diff --git a/be/src/runtime/tmp-file-mgr.cc b/be/src/runtime/tmp-file-mgr.cc
index 49bc40e..676f646 100644
--- a/be/src/runtime/tmp-file-mgr.cc
+++ b/be/src/runtime/tmp-file-mgr.cc
@@ -33,8 +33,10 @@
#include "kudu/util/env.h"
#include "runtime/bufferpool/buffer-pool-counters.h"
#include "runtime/exec-env.h"
+#include "runtime/hdfs-fs-cache.h"
#include "runtime/io/disk-io-mgr.h"
#include "runtime/io/error-converter.h"
+#include "runtime/io/local-file-writer.h"
#include "runtime/io/request-context.h"
#include "runtime/mem-tracker.h"
#include "runtime/runtime-state.h"
@@ -45,7 +47,10 @@
#include "util/debug-util.h"
#include "util/disk-info.h"
#include "util/filesystem-util.h"
+#include "util/hdfs-util.h"
+#include "util/histogram-metric.h"
#include "util/kudu-status-util.h"
+#include "util/os-util.h"
#include "util/parse-util.h"
#include "util/pretty-printer.h"
#include "util/runtime-profile-counters.h"
@@ -91,6 +96,23 @@ DEFINE_string(scratch_dirs, "/tmp",
DEFINE_bool(allow_multiple_scratch_dirs_per_device, true,
"If false and --scratch_dirs contains multiple directories on the same device, "
"then only the first writable directory is used");
+DEFINE_string(remote_tmp_file_size, "16M",
+ "Specify the size of a remote temporary file. Upper bound is 256MB. Lower bound "
+ "is the block size. The size should be power of 2 and integer times of the block "
+ "size.");
+DEFINE_string(remote_tmp_file_block_size, "1M",
+ "Specify the size of the block for doing file uploading and fetching. The block "
+ "size should be power of 2 and less than the size of remote temporary file.");
+DEFINE_bool(remote_tmp_files_avail_pool_lifo, false,
+ "If true, lifo is the algo to evict the local buffer files during spilling "
+ "to the remote. Otherwise, fifo would be used.");
+DEFINE_bool(allow_spill_to_hdfs, false,
+ "Spill to HDFS is a test-only feature, only when set true, the user can configure "
+ "a HDFS scratch path.");
+DEFINE_int32(wait_for_spill_buffer_timeout_s, 60,
+ "Specify the timeout duration waiting for the buffer to write (second). If a spilling"
+ "opertion fails to get a buffer from the pool within the duration, the operation"
+ "fails.");
using boost::algorithm::is_any_of;
using boost::algorithm::join;
@@ -112,6 +134,13 @@ constexpr int64_t TmpFileMgr::HOLE_PUNCH_BLOCK_SIZE_BYTES;
const string TMP_SUB_DIR_NAME = "impala-scratch";
const uint64_t AVAILABLE_SPACE_THRESHOLD_MB = 1024;
+const uint64_t MAX_REMOTE_TMPFILE_SIZE_THRESHOLD_MB = 256;
+
+// Default path for Spill to HDFS.
+// The Spill to HDFS is only supported for testing, and a fixed default HDFS path is used,
+// To support a flexible HDFS path input, we may need some modifications, Jira
+// IMPALA-10429 is tracking the limitation.
+const string TMP_DIR_HDFS_LOCAL_DEFAULT = "hdfs://localhost:20500/tmp";
// Metric keys
const string TMP_FILE_MGR_ACTIVE_SCRATCH_DIRS = "tmp-file-mgr.active-scratch-dirs";
@@ -123,6 +152,15 @@ const string TMP_FILE_MGR_SCRATCH_SPACE_BYTES_USED =
"tmp-file-mgr.scratch-space-bytes-used";
const string SCRATCH_DIR_BYTES_USED_FORMAT =
"tmp-file-mgr.scratch-space-bytes-used.dir-$0";
+const string LOCAL_BUFF_BYTES_USED_FORMAT = "tmp-file-mgr.local-buff-bytes-used.dir-$0";
+const string TMP_FILE_BUFF_POOL_DEQUEUE_DURATIONS =
+ "tmp-file-mgr.tmp-file-buff-pool-dequeue-durations";
+
+static const Status& TMP_FILE_MGR_NO_AVAILABLE_FILE_TO_EVICT = Status(ErrorMsg::Init(
+ TErrorCode::GENERAL,
+ "TmpFileMgr::ReserveLocalBufferSpace() failed to find available files to evict."));
+static const Status& TMP_FILE_BUFFER_POOL_CONTEXT_CANCELLED =
+ Status::CancelledInternal("TmpFileBufferPool");
using DeviceId = TmpFileMgr::DeviceId;
using TmpDir = TmpFileMgr::TmpDir;
@@ -175,13 +213,72 @@ Status TmpFileMgr::InitCustom(const vector<string>& tmp_dir_specifiers,
ExecEnv::GetInstance()->process_mem_tracker()));
}
}
+
+ bool is_percent;
+ tmp_dirs_remote_ctrl_.remote_tmp_file_size_ =
+ ParseUtil::ParseMemSpec(FLAGS_remote_tmp_file_size, &is_percent, 0);
+ if (tmp_dirs_remote_ctrl_.remote_tmp_file_size_ <= 0) {
+ return Status(Substitute(
+ "Invalid value of remote_tmp_file_size '$0'", FLAGS_remote_tmp_file_size));
+ }
+ if (tmp_dirs_remote_ctrl_.remote_tmp_file_size_
+ > MAX_REMOTE_TMPFILE_SIZE_THRESHOLD_MB * 1024 * 1024) {
+ tmp_dirs_remote_ctrl_.remote_tmp_file_size_ =
+ MAX_REMOTE_TMPFILE_SIZE_THRESHOLD_MB * 1024 * 1024;
+ }
+ tmp_dirs_remote_ctrl_.remote_tmp_block_size_ =
+ ParseUtil::ParseMemSpec(FLAGS_remote_tmp_file_block_size, &is_percent,
+ tmp_dirs_remote_ctrl_.remote_tmp_file_size_);
+ if (tmp_dirs_remote_ctrl_.remote_tmp_block_size_ <= 0) {
+ return Status(Substitute(
+ "Invalid value of remote_tmp_block_size '$0'", FLAGS_remote_tmp_file_block_size));
+ }
+ tmp_dirs_remote_ctrl_.wait_for_spill_buffer_timeout_us_ =
+ FLAGS_wait_for_spill_buffer_timeout_s * MICROS_PER_SEC;
+ if (tmp_dirs_remote_ctrl_.wait_for_spill_buffer_timeout_us_ <= 0) {
+ return Status(Substitute("Invalid value of wait_for_spill_buffer_timeout_us '$0'",
+ FLAGS_wait_for_spill_buffer_timeout_s));
+ }
+ // Below options are using for test by setting different modes to implement the
+ // spilling to the remote.
+ tmp_dirs_remote_ctrl_.remote_tmp_files_avail_pool_lifo_ =
+ FLAGS_remote_tmp_files_avail_pool_lifo;
+ tmp_dirs_remote_ctrl_.allow_spill_to_hdfs_ = FLAGS_allow_spill_to_hdfs;
+
vector<std::unique_ptr<TmpDir>> tmp_dirs;
+ // need_local_buffer_dir indicates if currently we need to a directory in local scratch
+ // space for being the buffer of a remote directory.
+ bool need_local_buffer_dir = false;
+
// Parse the directory specifiers. Don't return an error on parse errors, just log a
// warning - we don't want to abort process startup because of misconfigured scratch,
// since queries will generally still be runnable.
for (const string& tmp_dir_spec : tmp_dir_specifiers) {
+ string tmp_dirs_without_prefix, prefix;
+ string tmp_dir_spec_trimmed(boost::algorithm::trim_left_copy(tmp_dir_spec));
+
+ if (IsHdfsPath(tmp_dir_spec_trimmed.c_str(), false)) {
+ if (!tmp_dirs_remote_ctrl_.allow_spill_to_hdfs_) {
+ return Status("HDFS scratch space is not allowed. Please set allow_spill_to_hdfs "
+ "if HDFS path is needed.");
+ }
+ prefix = FILESYS_PREFIX_HDFS;
+ tmp_dirs_without_prefix = tmp_dir_spec_trimmed.substr(strlen(FILESYS_PREFIX_HDFS));
+ need_local_buffer_dir = true;
+ } else if (IsS3APath(tmp_dir_spec.c_str(), false)) {
+ prefix = FILESYS_PREFIX_S3;
+ tmp_dirs_without_prefix = tmp_dir_spec_trimmed.substr(strlen(FILESYS_PREFIX_S3));
+ need_local_buffer_dir = true;
+ // Initialize the S3 options for later getting S3 connection.
+ s3a_options_ = {make_pair("fs.s3a.fast.upload", "true"),
+ make_pair("fs.s3a.fast.upload.buffer", "disk")};
+ } else {
+ prefix = "";
+ tmp_dirs_without_prefix = tmp_dir_spec_trimmed.substr(0);
+ }
+
vector<string> toks;
- split(toks, tmp_dir_spec, is_any_of(":"), token_compress_off);
+ split(toks, tmp_dirs_without_prefix, is_any_of(":"), token_compress_off);
if (toks.size() > 3) {
LOG(ERROR) << "Could not parse temporary dir specifier, too many colons: '"
<< tmp_dir_spec << "'";
@@ -193,7 +290,7 @@ Status TmpFileMgr::InitCustom(const vector<string>& tmp_dir_specifiers,
bytes_limit = ParseUtil::ParseMemSpec(toks[1], &is_percent, 0);
if (bytes_limit < 0 || is_percent) {
LOG(ERROR) << "Malformed scratch directory capacity configuration '"
- << tmp_dir_spec << "'";
+ << tmp_dirs_without_prefix << "'";
continue;
} else if (bytes_limit == 0) {
// Interpret -1, 0 or empty string as no limit.
@@ -210,83 +307,119 @@ Status TmpFileMgr::InitCustom(const vector<string>& tmp_dir_specifiers,
continue;
}
}
- IntGauge* bytes_used_metric = metrics->AddGauge(
- SCRATCH_DIR_BYTES_USED_FORMAT, 0, Substitute("$0", tmp_dirs.size()));
- tmp_dirs.emplace_back(new TmpDir(toks[0], bytes_limit, priority, bytes_used_metric));
+ tmp_dirs.emplace_back(
+ new TmpDir(prefix.append(toks[0]), bytes_limit, priority, nullptr));
}
- // Sort the tmp directories by priority.
- std::sort(tmp_dirs.begin(), tmp_dirs.end(),
- [](const std::unique_ptr<TmpDir>& a, const std::unique_ptr<TmpDir>& b) {
- return a->priority < b->priority;
- });
-
vector<bool> is_tmp_dir_on_disk(DiskInfo::num_disks(), false);
// For each tmp directory, find the disk it is on,
// so additional tmp directories on the same disk can be skipped.
for (int i = 0; i < tmp_dirs.size(); ++i) {
+ Status status;
path tmp_path(trim_right_copy_if(tmp_dirs[i]->path, is_any_of("/")));
- tmp_path = absolute(tmp_path);
- path scratch_subdir_path(tmp_path / TMP_SUB_DIR_NAME);
- // tmp_path must be a writable directory.
- Status status = FileSystemUtil::VerifyIsDirectory(tmp_path.string());
- if (!status.ok()) {
- LOG(WARNING) << "Cannot use directory " << tmp_path.string() << " for scratch: "
- << status.msg().msg();
- continue;
- }
- // Find the disk id of tmp_path. Add the scratch directory if there isn't another
- // directory on the same disk (or if we don't know which disk it is on).
- int disk_id = DiskInfo::disk_id(tmp_path.c_str());
- if (!one_dir_per_device || disk_id < 0 || !is_tmp_dir_on_disk[disk_id]) {
- uint64_t available_space;
- RETURN_IF_ERROR(FileSystemUtil::GetSpaceAvailable(tmp_path.string(),
- &available_space));
- if (available_space < AVAILABLE_SPACE_THRESHOLD_MB * 1024 * 1024) {
- LOG(WARNING) << "Filesystem containing scratch directory " << tmp_path
- << " has less than " << AVAILABLE_SPACE_THRESHOLD_MB
- << "MB available.";
+ bool is_hdfs_path = IsHdfsPath(tmp_path.c_str(), false);
+ bool is_s3a_path = IsS3APath(tmp_path.c_str(), false);
+ if (is_hdfs_path || is_s3a_path) {
+ // Only support one remote dir.
+ if (tmp_dirs_remote_ != nullptr) {
+ LOG(WARNING) << "Only one remote directory is supported. Directory "
+ << tmp_path.c_str() << " is abandoned.";
+ continue;
}
- // Create the directory, destroying if already present. If this succeeds, we will
- // have an empty writable scratch directory.
- status = FileSystemUtil::RemoveAndCreateDirectory(scratch_subdir_path.string());
- if (status.ok()) {
- if (disk_id >= 0) is_tmp_dir_on_disk[disk_id] = true;
- LOG(INFO) << "Using scratch directory " << scratch_subdir_path.string() << " on "
- << "disk " << disk_id
- << " limit: " << PrettyPrinter::PrintBytes(tmp_dirs[i]->bytes_limit);
- tmp_dirs_.emplace_back(scratch_subdir_path.string(), tmp_dirs[i]->bytes_limit,
- tmp_dirs[i]->priority, tmp_dirs[i]->bytes_used_metric);
+ string scratch_subdir_path_str;
+ if (is_hdfs_path) {
+ hdfsFS hdfs_conn;
+ RETURN_IF_ERROR(HdfsFsCache::instance()->GetConnection(
+ TMP_DIR_HDFS_LOCAL_DEFAULT, &hdfs_conn, &hdfs_conns_));
+ scratch_subdir_path_str = TMP_DIR_HDFS_LOCAL_DEFAULT + "/" + TMP_SUB_DIR_NAME;
+ if (hdfsExists(hdfs_conn, scratch_subdir_path_str.c_str()) != 0) {
+ if (hdfsCreateDirectory(hdfs_conn, scratch_subdir_path_str.c_str()) != 0) {
+ return Status(
+ GetHdfsErrorMsg("HDFS create path failed: ", scratch_subdir_path_str));
+ }
+ }
} else {
- LOG(WARNING) << "Could not remove and recreate directory "
- << scratch_subdir_path.string() << ": cannot use it for scratch. "
- << "Error was: " << status.msg().msg();
+ DCHECK(is_s3a_path);
+ path scratch_subdir_path(tmp_path / TMP_SUB_DIR_NAME);
+ scratch_subdir_path_str = scratch_subdir_path.string();
+ }
+ tmp_dirs_remote_ =
+ std::make_unique<TmpDir>(scratch_subdir_path_str, tmp_dirs[i]->bytes_limit,
+ tmp_dirs[i]->priority, tmp_dirs[i]->bytes_used_metric, false);
+ } else {
+ tmp_path = absolute(tmp_path);
+ path scratch_subdir_path(tmp_path / TMP_SUB_DIR_NAME);
+ // tmp_path must be a writable directory.
+ status = FileSystemUtil::VerifyIsDirectory(tmp_path.string());
+ if (!status.ok()) {
+ LOG(WARNING) << "Cannot use directory " << tmp_path.string()
+ << " for scratch: " << status.msg().msg();
+ continue;
}
- if (punch_holes_) {
- // Make sure hole punching is supported for the directory.
- // IMPALA-9798: this file should *not* be created inside impala-scratch
- // subdirectory to avoid races with multiple impalads starting up.
- RETURN_IF_ERROR(FileSystemUtil::CheckHolePunch(tmp_path.string()));
+
+ // Find the disk id of tmp_path. Add the scratch directory if there isn't another
+ // directory on the same disk (or if we don't know which disk it is on).
+ int disk_id = DiskInfo::disk_id(tmp_path.c_str());
+ if (!one_dir_per_device || disk_id < 0 || !is_tmp_dir_on_disk[disk_id]) {
+ uint64_t available_space;
+ RETURN_IF_ERROR(
+ FileSystemUtil::GetSpaceAvailable(tmp_path.string(), &available_space));
+ if (available_space < AVAILABLE_SPACE_THRESHOLD_MB * 1024 * 1024) {
+ LOG(WARNING) << "Filesystem containing scratch directory " << tmp_path
+ << " has less than " << AVAILABLE_SPACE_THRESHOLD_MB
+ << "MB available.";
+ }
+ RETURN_IF_ERROR(CreateDirectory(scratch_subdir_path.string(), tmp_path.string(),
+ tmp_dirs[i], metrics, &is_tmp_dir_on_disk, &need_local_buffer_dir, disk_id));
}
}
}
+ // Sort the tmp directories by priority.
+ std::sort(tmp_dirs_.begin(), tmp_dirs_.end(),
+ [](const TmpDir& a, const TmpDir& b) { return a.priority < b.priority; });
+
+ if (tmp_dirs_remote_ != nullptr) {
+ if (local_buff_dir_ == nullptr) {
+ // Should at least have one local dir for the buffer. Later we might allow to use
+ // s3 fast upload directly without a buffer.
+ return Status(
+ Substitute("No local directory configured for remote scratch space: $0",
+ tmp_dirs_remote_->path));
+ } else {
+ LOG(INFO) << "Using scratch directory " << tmp_dirs_remote_->path
+ << " limit: " << PrettyPrinter::PrintBytes(tmp_dirs_remote_->bytes_limit);
+ IntGauge* bytes_used_metric = metrics->AddGauge(
+ SCRATCH_DIR_BYTES_USED_FORMAT, 0, Substitute("$0", tmp_dirs_.size()));
+ tmp_dirs_remote_->bytes_used_metric = bytes_used_metric;
+ }
+ }
+
DCHECK(metrics != nullptr);
num_active_scratch_dirs_metric_ =
metrics->AddGauge(TMP_FILE_MGR_ACTIVE_SCRATCH_DIRS, 0);
active_scratch_dirs_metric_ = SetMetric<string>::CreateAndRegister(
metrics, TMP_FILE_MGR_ACTIVE_SCRATCH_DIRS_LIST, set<string>());
- num_active_scratch_dirs_metric_->SetValue(tmp_dirs_.size());
+ if (tmp_dirs_remote_ == nullptr) {
+ num_active_scratch_dirs_metric_->SetValue(tmp_dirs_.size());
+ } else {
+ num_active_scratch_dirs_metric_->SetValue(tmp_dirs_.size() + 1);
+ }
for (int i = 0; i < tmp_dirs_.size(); ++i) {
active_scratch_dirs_metric_->Add(tmp_dirs_[i].path);
}
+ if (tmp_dirs_remote_ != nullptr) {
+ active_scratch_dirs_metric_->Add(tmp_dirs_remote_->path);
+ RETURN_IF_ERROR(CreateTmpFileBufferPoolThread(metrics));
+ }
+
scratch_bytes_used_metric_ =
metrics->AddHWMGauge(TMP_FILE_MGR_SCRATCH_SPACE_BYTES_USED_HIGH_WATER_MARK,
TMP_FILE_MGR_SCRATCH_SPACE_BYTES_USED, 0);
initialized_ = true;
- if (tmp_dirs_.empty() && !tmp_dirs.empty()) {
+ if ((tmp_dirs_.empty() && local_buff_dir_ == nullptr) && !tmp_dirs.empty()) {
LOG(ERROR) << "Running without spill to disk: could not use any scratch "
<< "directories in list: " << join(tmp_dir_specifiers, ",")
<< ". See previous warnings for information on causes.";
@@ -294,6 +427,69 @@ Status TmpFileMgr::InitCustom(const vector<string>& tmp_dir_specifiers,
return Status::OK();
}
+Status TmpFileMgr::CreateDirectory(const string& scratch_subdir_path,
+ const string& tmp_path, const std::unique_ptr<TmpDir>& tmp_dir, MetricGroup* metrics,
+ vector<bool>* is_tmp_dir_on_disk, bool* need_local_buffer_dir, int disk_id) {
+ // Create the directory, destroying if already present. If this succeeds, we will
+ // have an empty writable scratch directory.
+ Status status = FileSystemUtil::RemoveAndCreateDirectory(scratch_subdir_path);
+ if (status.ok()) {
+ if (*need_local_buffer_dir) {
+ *need_local_buffer_dir = false;
+ // Add the first local dir as local buffer, the dir is only served as the buffer for
+ // Spill to Remote FS. At least we need the dir has two default file size space.
+ if (tmp_dir->bytes_limit < tmp_dirs_remote_ctrl_.remote_tmp_file_size_ * 2) {
+ return Status(Substitute(
+ "Local buffer directory $0 configured for remote scratch "
+ "space has a size limit of $1 bytes, should be at least twice as the "
+ "temporary file size "
+ "$2 bytes",
+ tmp_dir->path, tmp_dir->bytes_limit,
+ tmp_dirs_remote_ctrl_.remote_tmp_file_size_));
+ }
+ IntGauge* local_buff_bytes_used_metric =
+ metrics->AddGauge(LOCAL_BUFF_BYTES_USED_FORMAT, 0, Substitute("$0", 0));
+ local_buff_dir_ = std::make_unique<TmpDir>(
+ scratch_subdir_path, tmp_dir->bytes_limit, 0, local_buff_bytes_used_metric);
+ return Status::OK();
+ }
+ if (disk_id >= 0) (*is_tmp_dir_on_disk)[disk_id] = true;
+ LOG(INFO) << "Using scratch directory " << scratch_subdir_path << " on "
+ << "disk " << disk_id
+ << " limit: " << PrettyPrinter::PrintBytes(tmp_dir->bytes_limit);
+ IntGauge* bytes_used_metric = metrics->AddGauge(
+ SCRATCH_DIR_BYTES_USED_FORMAT, 0, Substitute("$0", tmp_dirs_.size()));
+ tmp_dirs_.emplace_back(
+ scratch_subdir_path, tmp_dir->bytes_limit, tmp_dir->priority, bytes_used_metric);
+ } else {
+ LOG(WARNING) << "Could not remove and recreate directory " << scratch_subdir_path
+ << ": cannot use it for scratch. "
+ << "Error was: " << status.msg().msg();
+ }
+ if (punch_holes_) {
+ // Make sure hole punching is supported for the directory.
+ // IMPALA-9798: this file should *not* be created inside impala-scratch
+ // subdirectory to avoid races with multiple impalads starting up.
+ RETURN_IF_ERROR(FileSystemUtil::CheckHolePunch(tmp_path));
+ }
+ return Status::OK();
+}
+
+Status TmpFileMgr::CreateTmpFileBufferPoolThread(MetricGroup* metrics) {
+ DCHECK(metrics != nullptr);
+ tmp_dirs_remote_ctrl_.tmp_file_pool_.reset(new TmpFileBufferPool(this));
+ std::unique_ptr<Thread> t;
+ RETURN_IF_ERROR(Thread::Create("tmp-file-pool", "work-loop(TmpFileSpaceReserve Worker)",
+ &TmpFileBufferPool::TmpFileSpaceReserveThreadLoop,
+ tmp_dirs_remote_ctrl_.tmp_file_pool_.get(), &t));
+ tmp_dirs_remote_ctrl_.tmp_file_mgr_thread_group_.AddThread(move(t));
+ int64_t ONE_HOUR_IN_NS = 60L * 60L * NANOS_PER_SEC;
+ tmp_dirs_remote_ctrl_.tmp_file_pool_->dequeue_timer_metric_ =
+ metrics->RegisterMetric(new HistogramMetric(
+ MetricDefs::Get(TMP_FILE_BUFF_POOL_DEQUEUE_DURATIONS), ONE_HOUR_IN_NS, 3));
+ return Status::OK();
+}
+
void TmpFileMgr::NewFile(
TmpFileGroup* file_group, DeviceId device_id, unique_ptr<TmpFile>* new_file) {
DCHECK(initialized_);
@@ -307,53 +503,159 @@ void TmpFileMgr::NewFile(
path new_file_path(tmp_dirs_[device_id].path);
new_file_path /= file_name.str();
- new_file->reset(new TmpFile(file_group, device_id, new_file_path.string()));
+ new_file->reset(new TmpFileLocal(file_group, device_id, new_file_path.string()));
}
-string TmpFileMgr::GetTmpDirPath(DeviceId device_id) const {
+void TmpFileMgr::RemoveRemoteDir(TmpFileGroup* file_group, DeviceId device_id) {
+ if (tmp_dirs_remote_ == nullptr) return;
+ string dir = tmp_dirs_remote_->path;
+ stringstream files_dir;
+ files_dir << dir << "/" << PrintId(ExecEnv::GetInstance()->backend_id(), "_") << "_"
+ << PrintId(file_group->unique_id(), "_");
+ hdfsFS hdfs_conn;
+ Status status =
+ HdfsFsCache::instance()->GetConnection(files_dir.str(), &hdfs_conn, &hdfs_conns_);
+ DCHECK(hdfs_conn != nullptr);
+ hdfsDelete(hdfs_conn, files_dir.str().c_str(), 1);
+}
+
+Status TmpFileMgr::AsyncWriteRange(WriteRange* write_range, TmpFile* tmp_file) {
+ if (write_range->disk_file()->disk_type() == io::DiskFileType::LOCAL) {
+ DCHECK(write_range != nullptr);
+ DCHECK(write_range->io_ctx() != nullptr);
+ return write_range->io_ctx()->AddWriteRange(write_range);
+ }
+ // If spill to a remote directory, TmpFileBufferPool is helping to send the writes to
+ // the DiskQueue because the local buffer for the remote file may be used up and it may
+ // need to wait in the pool before the writes can be sent to the DiskQueue.
+ DCHECK(tmp_dirs_remote_ctrl_.tmp_file_pool_ != nullptr);
+ DCHECK(tmp_file != nullptr);
+ return tmp_dirs_remote_ctrl_.tmp_file_pool_->EnqueueWriteRange(write_range, tmp_file);
+}
+
+void TmpFileMgr::EnqueueTmpFilesPoolDummyFile() {
+ EnqueueTmpFilesPool(tmp_dirs_remote_ctrl_.tmp_file_pool_->tmp_file_dummy_, true);
+}
+
+void TmpFileMgr::EnqueueTmpFilesPool(shared_ptr<TmpFile>& tmp_file, bool front) {
+ tmp_dirs_remote_ctrl_.tmp_file_pool_->EnqueueTmpFilesPool(tmp_file, front);
+}
+
+Status TmpFileMgr::DequeueTmpFilesPool(shared_ptr<TmpFile>* tmp_file, bool quick_return) {
+ return tmp_dirs_remote_ctrl_.tmp_file_pool_->DequeueTmpFilesPool(
+ tmp_file, quick_return);
+}
+
+Status TmpFileMgr::TryEvictFile(TmpFile* tmp_file) {
+ DCHECK(tmp_file != nullptr);
+ if (tmp_file->disk_type() == io::DiskFileType::DUMMY) return Status::OK();
+
+ // Remove the buffer of the TmpFile.
+ // After deletion of the buffer, if the TmpFile doesn't exist in the remote file system
+ // either, that means the TmpFile shared pointer can be removed from the TmpFileMgr,
+ // because in this case, the physical file is considered no longer in the system.
+ // Fetch the unique locks of the files when doing the deletion.
+ TmpFileRemote* tmp_file_remote = static_cast<TmpFileRemote*>(tmp_file);
+ DiskFile* buffer_file = tmp_file_remote->DiskBufferFile();
+ Status status = Status::OK();
+ {
+ unique_lock<shared_mutex> buffer_lock(buffer_file->physical_file_lock_);
+ if (buffer_file->GetFileStatus() == io::DiskFileStatus::PERSISTED) {
+ status = buffer_file->Delete(buffer_lock);
+ }
+ }
+ return status;
+}
+
+Status TmpFileMgr::ReserveLocalBufferSpace(bool quick_return) {
+ int64_t file_size = GetRemoteTmpFileSize();
+
+ // The high water mark is used to record the total bytes which have been assigned, we
+ // can assume that all the assigned bytes will be finally returned to the pool.
+ // Before the high water mark reaches the bytes limit of the local buffer directory,
+ // the caller can gain space freely. But if the high water mark is over the bytes limit,
+ // the caller needs to gain space from the pool because all the available spaces are in
+ // the pool now.
+ TmpDir* dir = local_buff_dir_.get();
+ if (tmp_dirs_remote_ctrl_.local_buff_dir_bytes_high_water_mark_.Add(file_size)
+ > dir->bytes_limit) {
+ tmp_dirs_remote_ctrl_.local_buff_dir_bytes_high_water_mark_.Add(-file_size);
+ } else {
+ GetLocalBufferDir()->bytes_used_metric->Increment(file_size);
+ return Status::OK();
+ }
+
+ shared_ptr<TmpFile> tmp_file;
+ // If all of the space of the buffer directory has been assigned, gain a file which
+ // is available to be evicted from the TmpFileBufferPool. It can be a long wait if
+ // quick return is not set and there is no available file in the pool.
+ Status status = DequeueTmpFilesPool(&tmp_file, quick_return);
+ if (!status.ok()) {
+ DCHECK(tmp_file == nullptr);
+ return status;
+ }
+ // Evict the file to release the physical space.
+ // If error happens during eviction, we log an warning, and return status ok instead to
+ // keep the caller doing the writing since probably the physical file is already
+ // deleted.
+ status = TryEvictFile(tmp_file.get());
+ if (!status.ok()) {
+ LOG(WARNING) << "File Eviction Failed: " << tmp_file->GetWriteFile()->path();
+ }
+ return Status::OK();
+}
+
+TmpDir* TmpFileMgr::GetLocalBufferDir() const {
+ return local_buff_dir_.get();
+}
+
+int TmpFileMgr::NumActiveTmpDevicesLocal() {
DCHECK(initialized_);
- DCHECK_GE(device_id, 0);
- DCHECK_LT(device_id, tmp_dirs_.size());
- return tmp_dirs_[device_id].path;
+ return tmp_dirs_.size();
}
int TmpFileMgr::NumActiveTmpDevices() {
DCHECK(initialized_);
- return tmp_dirs_.size();
+ return tmp_dirs_.size() + ((tmp_dirs_remote_ == nullptr) ? 0 : 1);
}
vector<DeviceId> TmpFileMgr::ActiveTmpDevices() {
vector<DeviceId> devices;
- for (DeviceId device_id = 0; device_id < tmp_dirs_.size(); ++device_id) {
+ DeviceId device_id = 0;
+ for (; device_id < tmp_dirs_.size(); ++device_id) {
+ devices.push_back(device_id);
+ }
+ if (tmp_dirs_remote_ != nullptr) {
devices.push_back(device_id);
}
return devices;
}
-TmpFile::TmpFile(TmpFileGroup* file_group, DeviceId device_id, const string& path)
+string TmpFileMgr::GetTmpDirPath(DeviceId device_id) const {
+ DCHECK(initialized_);
+ DCHECK_GE(device_id, 0);
+ DCHECK_LT(device_id, tmp_dirs_.size() + ((tmp_dirs_remote_ == nullptr) ? 0 : 1));
+ if (device_id < tmp_dirs_.size()) {
+ return tmp_dirs_[device_id].path;
+ } else {
+ return tmp_dirs_remote_->path;
+ }
+}
+
+TmpFile::TmpFile(
+ TmpFileGroup* file_group, DeviceId device_id, const string& path, bool expected_local)
: file_group_(file_group),
path_(path),
device_id_(device_id),
disk_id_(DiskInfo::disk_id(path.c_str())),
- blacklisted_(false) {
- DCHECK(file_group != nullptr);
-}
+ expected_local_(expected_local),
+ blacklisted_(false) {}
-bool TmpFile::AllocateSpace(int64_t num_bytes, int64_t* offset) {
- DCHECK_GT(num_bytes, 0);
- TmpDir* dir = GetDir();
- // Increment optimistically and roll back if the limit is exceeded.
- if (dir->bytes_used_metric->Increment(num_bytes) > dir->bytes_limit) {
- dir->bytes_used_metric->Increment(-num_bytes);
- return false;
+int TmpFile::AssignDiskQueue(bool is_local_buffer) const {
+ if (is_local_buffer) {
+ return file_group_->io_mgr_->AssignQueue(local_buffer_path_.c_str(), -1, true);
}
- *offset = allocation_offset_;
- allocation_offset_ += num_bytes;
- return true;
-}
-
-int TmpFile::AssignDiskQueue() const {
- return file_group_->io_mgr_->AssignQueue(path_.c_str(), disk_id_, false);
+ return file_group_->io_mgr_->AssignQueue(path_.c_str(), disk_id_, expected_local_);
}
bool TmpFile::Blacklist(const ErrorMsg& msg) {
@@ -366,18 +668,14 @@ bool TmpFile::Blacklist(const ErrorMsg& msg) {
}
}
-Status TmpFile::Remove() {
- // Remove the file if present (it may not be present if no writes completed).
- Status status = FileSystemUtil::RemovePaths({path_});
- int64_t bytes_in_use = file_group_->tmp_file_mgr_->punch_holes() ?
- allocation_offset_ - bytes_reclaimed_.Load() :
- allocation_offset_;
- GetDir()->bytes_used_metric->Increment(-bytes_in_use);
- return status;
-}
-
TmpFileMgr::TmpDir* TmpFile::GetDir() {
- return &file_group_->tmp_file_mgr_->tmp_dirs_[device_id_];
+ auto tmp_file_mgr = file_group_->tmp_file_mgr_;
+ if (device_id_ >= tmp_file_mgr->tmp_dirs_.size()) {
+ // Only one remote directory supported.
+ DCHECK(device_id_ - tmp_file_mgr->tmp_dirs_.size() == 0);
+ return tmp_file_mgr->tmp_dirs_remote_.get();
+ }
+ return &tmp_file_mgr->tmp_dirs_[device_id_];
}
Status TmpFile::PunchHole(int64_t offset, int64_t len) {
@@ -402,6 +700,138 @@ string TmpFile::DebugString() {
this, path_, device_id_, disk_id_, allocation_offset_, blacklisted_);
}
+TmpFileLocal::TmpFileLocal(TmpFileGroup* file_group, TmpFileMgr::DeviceId device_id,
+ const std::string& path, bool expected_local)
+ : TmpFile(file_group, device_id, path, expected_local) {
+ DCHECK(file_group != nullptr);
+ disk_file_ = make_unique<io::DiskFile>(path_, file_group->io_mgr_);
+ disk_type_ = io::DiskFileType::LOCAL;
+}
+
+bool TmpFileLocal::AllocateSpace(int64_t num_bytes, int64_t* offset) {
+ DCHECK_GT(num_bytes, 0);
+ TmpDir* dir = GetDir();
+ // Increment optimistically and roll back if the limit is exceeded.
+ if (dir->bytes_used_metric->Increment(num_bytes) > dir->bytes_limit) {
+ dir->bytes_used_metric->Increment(-num_bytes);
+ return false;
+ }
+ *offset = allocation_offset_;
+ allocation_offset_ += num_bytes;
+ return true;
+}
+
+io::DiskFile* TmpFileLocal::GetWriteFile() {
+ return disk_file_.get();
+}
+
+Status TmpFileLocal::Remove() {
+ // Remove the file if present (it may not be present if no writes completed).
+ Status status = FileSystemUtil::RemovePaths({path_});
+ int64_t bytes_in_use = file_group_->tmp_file_mgr_->punch_holes() ?
+ allocation_offset_ - bytes_reclaimed_.Load() :
+ allocation_offset_;
+ GetDir()->bytes_used_metric->Increment(-bytes_in_use);
+ return status;
+}
+
+TmpFileRemote::TmpFileRemote(TmpFileGroup* file_group, TmpFileMgr::DeviceId device_id,
+ const std::string& path, const std::string& local_buffer_path, bool expected_local,
+ const char* hdfs_url)
+ : TmpFile(file_group, device_id, path, expected_local) {
+ DCHECK(hdfs_url != nullptr);
+ hdfs_conn_ = nullptr;
+ if (IsHdfsPath(hdfs_url, false)) {
+ hdfs_conn_ = hdfsConnect("default", 0);
+ disk_type_ = io::DiskFileType::DFS;
+ disk_id_ = file_group->io_mgr_->RemoteDfsDiskId();
+ } else if (IsS3APath(hdfs_url, false)) {
+ disk_type_ = io::DiskFileType::S3;
+ disk_id_ = file_group->io_mgr_->RemoteS3DiskId();
+ Status status = HdfsFsCache::instance()->GetConnection(hdfs_url, &hdfs_conn_,
+ &file_group_->tmp_file_mgr_->hdfs_conns_,
+ file_group_->tmp_file_mgr_->s3a_options());
+ }
+ file_size_ = file_group_->tmp_file_mgr_->GetRemoteTmpFileSize();
+ local_buffer_path_ = local_buffer_path;
+ disk_file_ = make_unique<io::DiskFile>(path_, file_group->io_mgr_,
+ file_group_->tmp_file_mgr_->GetRemoteTmpFileSize(), disk_type_, &hdfs_conn_);
+ disk_buffer_file_ = make_unique<io::DiskFile>(local_buffer_path_, file_group_->io_mgr_,
+ file_group_->tmp_file_mgr_->GetRemoteTmpFileSize(), io::DiskFileType::LOCAL_BUFFER);
+}
+
+TmpFileRemote::~TmpFileRemote() {
+ // Need to return the buffer before deconstruction if buffer space is reserved.
+ if (DiskBufferFile()->IsSpaceReserved()) DCHECK(is_buffer_returned());
+}
+
+bool TmpFileRemote::AllocateSpace(int64_t num_bytes, int64_t* offset) {
+ DCHECK_GT(num_bytes, 0);
+ if (at_capacity_) return false;
+ *offset = allocation_offset_;
+ allocation_offset_ += num_bytes;
+ // The actual size could be a little over the file size.
+ if (allocation_offset_ >= file_size_) {
+ // Set the actual file size of the disk file for the use of writing.
+ GetWriteFile()->SetActualFileSize(allocation_offset_);
+ at_capacity_ = true;
+ }
+ return true;
+}
+
+io::DiskFile* TmpFileRemote::GetWriteFile() {
+ return disk_buffer_file_.get();
+}
+
+TmpFileMgr::TmpDir* TmpFileRemote::GetLocalBufferDir() const {
+ return file_group_->tmp_file_mgr_->GetLocalBufferDir();
+}
+
+Status TmpFileRemote::Remove() {
+ Status status = Status::OK();
+ // If True, we need to enqueue the file back to the pool after deletion.
+ bool to_return_the_buffer = false;
+
+ // The order of acquiring the lock must be from local to remote to avoid deadlocks.
+ unique_lock<shared_mutex> buffer_file_lock(*(disk_buffer_file_->GetFileLock()));
+ unique_lock<shared_mutex> file_lock(*(disk_file_->GetFileLock()));
+
+ // Delete the local buffer file if exists.
+ if (disk_buffer_file_->GetFileStatus() != io::DiskFileStatus::DELETED) {
+ status = disk_buffer_file_->Delete(buffer_file_lock);
+ if (!status.ok()) {
+ // If the physical file is failed to delete, log a warning, and set a deleted flag
+ // anyway.
+ LOG(WARNING) << "Delete file: " << disk_buffer_file_->path() << " failed.";
+ disk_buffer_file_->SetStatus(io::DiskFileStatus::DELETED);
+ } else if (disk_file_->GetFileStatus() != io::DiskFileStatus::PERSISTED
+ && disk_buffer_file_->IsSpaceReserved()) {
+ // If the file is not uploaded and the buffer space is reserved, we need to return
+ // the buffer to the pool after deletion of the TmpFile. The buffer of a uploaded
+ // file should have been returned to the pool after upload operation completes.
+ to_return_the_buffer = true;
+ } else {
+ // Do nothing.
+ }
+ }
+
+ // Set the remote file status to deleted. The physical remote files would be deleted
+ // during deconstruction of TmpFileGroup by deleting the entire remote
+ // directory for efficiency consideration.
+ disk_file_->SetStatus(io::DiskFileStatus::DELETED);
+
+ // Update the metrics.
+ GetDir()->bytes_used_metric->Increment(-file_size_);
+
+ // Return the file to the pool if it hasn't been enqueued.
+ if (to_return_the_buffer) {
+ file_group_->tmp_file_mgr()->EnqueueTmpFilesPool(
+ file_group_->FindTmpFileSharedPtr(this), true);
+ }
+
+ return status;
+}
+
TmpFileGroup::TmpFileGroup(TmpFileMgr* tmp_file_mgr, DiskIoMgr* io_mgr,
RuntimeProfile* profile, const TUniqueId& unique_id, int64_t bytes_limit)
: tmp_file_mgr_(tmp_file_mgr),
@@ -425,6 +855,7 @@ TmpFileGroup::TmpFileGroup(TmpFileMgr* tmp_file_mgr, DiskIoMgr* io_mgr,
num_blacklisted_files_(0),
spilling_disk_faulty_(false),
current_bytes_allocated_(0),
+ current_bytes_allocated_remote_(0),
next_allocation_index_(0),
free_ranges_(64) {
DCHECK(tmp_file_mgr != nullptr);
@@ -456,17 +887,18 @@ Status TmpFileGroup::CreateFiles() {
lock_.DCheckLocked();
DCHECK(tmp_files_.empty());
vector<DeviceId> tmp_devices = tmp_file_mgr_->ActiveTmpDevices();
+ DCHECK(tmp_file_mgr_->NumActiveTmpDevicesLocal() <= tmp_devices.size());
int files_allocated = 0;
// Initialize the tmp files and the initial file to use.
- for (int i = 0; i < tmp_devices.size(); ++i) {
+ for (int i = 0; i < tmp_file_mgr_->NumActiveTmpDevicesLocal(); ++i) {
DeviceId device_id = tmp_devices[i];
unique_ptr<TmpFile> tmp_file;
tmp_file_mgr_->NewFile(this, device_id, &tmp_file);
tmp_files_.emplace_back(std::move(tmp_file));
++files_allocated;
}
- DCHECK_EQ(tmp_files_.size(), files_allocated);
- DCHECK_EQ(tmp_file_mgr_->tmp_dirs_.size(), tmp_files_.size());
+ DCHECK_EQ(tmp_file_mgr_->NumActiveTmpDevicesLocal(), files_allocated);
+ DCHECK_EQ(tmp_file_mgr_->NumActiveTmpDevicesLocal(), tmp_files_.size());
if (tmp_files_.size() == 0) return ScratchAllocationFailedStatus({});
// Initialize the next allocation index for each priority.
for (const auto& entry: tmp_files_index_range_) {
@@ -479,21 +911,36 @@ Status TmpFileGroup::CreateFiles() {
return Status::OK();
}
-void TmpFileGroup::Close() {
- // Cancel writes before deleting the files, since in-flight writes could re-create
- // deleted files.
- if (io_ctx_ != nullptr) io_mgr_->UnregisterContext(io_ctx_.get());
- for (std::unique_ptr<TmpFile>& file : tmp_files_) {
+template <typename T>
+void TmpFileGroup::CloseInternal(vector<T>& tmp_files) {
+ for (auto& file : tmp_files) {
Status status = file->Remove();
if (!status.ok()) {
LOG(WARNING) << "Error removing scratch file '" << file->path()
<< "': " << status.msg().msg();
}
}
+ tmp_files.clear();
+}
+
+void TmpFileGroup::Close() {
+ // Cancel writes before deleting the files, since in-flight writes could re-create
+ // deleted files.
+ if (io_ctx_ != nullptr) {
+ if (tmp_file_mgr_->HasRemoteDir()) {
+ // Remove all the writes using the io_ctx and waiting for buffer reservation in
+ // the pool.
+ DCHECK(tmp_file_mgr_->tmp_dirs_remote_ctrl_.tmp_file_pool_ != nullptr);
+ tmp_file_mgr_->tmp_dirs_remote_ctrl_.tmp_file_pool_->RemoveWriteRanges(
+ io_ctx_.get());
+ }
+ io_mgr_->UnregisterContext(io_ctx_.get());
+ }
+ CloseInternal<std::unique_ptr<TmpFile>>(tmp_files_);
+ CloseInternal<std::shared_ptr<TmpFile>>(tmp_files_remote_);
+ tmp_file_mgr_->RemoveRemoteDir(this, 0);
tmp_file_mgr_->scratch_bytes_used_metric_->Increment(
-1 * scratch_space_bytes_used_counter_->value());
-
- tmp_files_.clear();
}
// Rounds up to the smallest unit of allocation in a scratch file
@@ -509,9 +956,118 @@ static int64_t RoundUpToScratchRangeSize(bool punch_holes, int64_t bytes) {
}
}
-Status TmpFileGroup::AllocateSpace(
- int64_t num_bytes, TmpFile** tmp_file, int64_t* file_offset) {
- lock_guard<SpinLock> lock(lock_);
+void TmpFileGroup::UpdateScratchSpaceMetrics(int64_t num_bytes, bool is_remote) {
+ scratch_space_bytes_used_counter_->Add(num_bytes);
+ tmp_file_mgr_->scratch_bytes_used_metric_->Increment(num_bytes);
+ current_bytes_allocated_.Add(num_bytes);
+ if (is_remote) current_bytes_allocated_remote_.Add(num_bytes);
+}
+
+string TmpFileGroup::GenerateNewPath(string& dir, string& unique_name) {
+ stringstream file_name;
+ file_name << TMP_SUB_DIR_NAME << "-" << unique_name;
+ path new_file_path(dir);
+ new_file_path /= file_name.str();
+ return new_file_path.string();
+}
+
+std::shared_ptr<TmpFile>& TmpFileGroup::FindTmpFileSharedPtr(TmpFile* tmp_file) {
+ DCHECK(tmp_file != nullptr);
+ DCHECK(tmp_file->DiskFile()->disk_type() != io::DiskFileType::LOCAL);
+ lock_guard<SpinLock> lock(tmp_files_remote_ptrs_lock_);
+ auto shared_file_it = tmp_files_remote_ptrs_.find(tmp_file);
+ DCHECK(shared_file_it != tmp_files_remote_ptrs_.end());
+ return shared_file_it->second;
+}
+
+Status TmpFileGroup::AllocateRemoteSpace(int64_t num_bytes, TmpFile** tmp_file,
+ int64_t* file_offset, vector<int>* at_capacity_dirs) {
+ // Only one remote dir supported currently.
+ string dir = tmp_file_mgr_->tmp_dirs_remote_->path;
+ // It is not supposed to have a remote directory other than HDFS or S3.
+ DCHECK(IsHdfsPath(dir.c_str(), false) || IsS3APath(dir.c_str(), false));
+
+ // Look for the space from a previous created file.
+ if (!tmp_files_remote_.empty()) {
+ TmpFile* tmp_file_cur = tmp_files_remote_.back().get();
+ // If the file is blocklisted or is at capacity, we will create a new file instead.
+ if (!tmp_file_cur->is_blacklisted()) {
+ if (tmp_file_cur->AllocateSpace(num_bytes, file_offset)) {
+ *tmp_file = tmp_file_cur;
+ return Status::OK();
+ }
+ }
+ }
+
+ // Return an error if the new bytes is over the bytes limit of the query or the remote
+ // directory.
+ int64_t new_bytes =
+ current_bytes_allocated_.Load() + tmp_file_mgr_->GetRemoteTmpFileSize();
+ if (bytes_limit_ != -1 && new_bytes > bytes_limit_) {
+ return Status(TErrorCode::SCRATCH_LIMIT_EXCEEDED, bytes_limit_, GetBackendString());
+ }
+
+ int64_t remote_dir_bytes_limit = tmp_file_mgr_->tmp_dirs_remote_->bytes_limit;
+ if (remote_dir_bytes_limit != -1 && new_bytes > remote_dir_bytes_limit) {
+ return Status(
+ TErrorCode::SCRATCH_LIMIT_EXCEEDED, remote_dir_bytes_limit, GetBackendString());
+ }
+
+ // The device id of remote directory is defined as the max local device id
+ // plus the index of the remote dir. Since we only support one remote dir now,
+ // the id is the max local device id plus one.
+ DeviceId dev_id = tmp_file_mgr_->tmp_dirs_.size();
+ string unique_name = lexical_cast<string>(random_generator()());
+ stringstream file_name;
+ dir = dir + "/" + PrintId(ExecEnv::GetInstance()->backend_id(), "_") + "_"
+ + PrintId(unique_id(), "_");
+
+ string new_file_path = GenerateNewPath(dir, unique_name);
+ string local_buffer_dir = tmp_file_mgr_->local_buff_dir_->path;
+ string new_file_path_local = GenerateNewPath(local_buffer_dir, unique_name);
+
+ TmpFileRemote* tmp_file_r = new TmpFileRemote(
+ this, dev_id, new_file_path, new_file_path_local, false, dir.c_str());
+ if (tmp_file_r == nullptr) {
+ return Status("Failed to allocate temporary file object.");
+ }
+ if (tmp_file_r->hdfs_conn_ == nullptr) {
+ return Status(Substitute("Failed to connect to FS: $0.", dir));
+ }
+ shared_ptr<TmpFile> tmp_file_remote(move(tmp_file_r));
+ int64_t file_size = tmp_file_mgr_->GetRemoteTmpFileSize();
+ TmpFileMgr::TmpDir* tmp_dir_remote = tmp_file_remote->GetDir();
+ if (tmp_dir_remote->bytes_limit != -1
+ && tmp_dir_remote->bytes_used_metric->Increment(file_size)
+ > tmp_dir_remote->bytes_limit) {
+ tmp_dir_remote->bytes_used_metric->Increment(-file_size);
+ at_capacity_dirs->push_back(dev_id);
+ return Status(Substitute("Reach the size limit $0 of dir: $1",
+ tmp_dir_remote->bytes_limit, tmp_dir_remote->path));
+ }
+ UpdateScratchSpaceMetrics(file_size, true);
+
+ // It should be a successful return to allocate the first range from the new file.
+ DCHECK(tmp_file_remote->AllocateSpace(num_bytes, file_offset));
+
+ tmp_files_remote_.emplace_back(std::move(tmp_file_remote));
+ *tmp_file = tmp_files_remote_.back().get();
+ {
+ lock_guard<SpinLock> lock(tmp_files_remote_ptrs_lock_);
+ tmp_files_remote_ptrs_.emplace(*tmp_file, tmp_files_remote_.back());
+ }
+ *file_offset = 0;
+
+ // Try to reserve the space for local buffer with a quick return to avoid
+ // a long wait, if failed, caller should do the reservation for the buffer.
+ Status reserve_status = tmp_file_mgr_->ReserveLocalBufferSpace(true);
+ if (reserve_status.ok()) (*tmp_file)->GetWriteFile()->SetSpaceReserved();
+
+ return Status::OK();
+}
+
+Status TmpFileGroup::AllocateLocalSpace(int64_t num_bytes, TmpFile** tmp_file,
+ int64_t* file_offset, vector<int>* at_capacity_dirs, bool* alloc_full) {
int64_t scratch_range_bytes =
RoundUpToScratchRangeSize(tmp_file_mgr_->punch_holes(), num_bytes);
int free_ranges_idx = BitUtil::Log2Ceiling64(scratch_range_bytes);
@@ -524,18 +1080,13 @@ Status TmpFileGroup::AllocateSpace(
}
if (bytes_limit_ != -1
- && current_bytes_allocated_ + scratch_range_bytes > bytes_limit_) {
+ && current_bytes_allocated_.Load() + scratch_range_bytes > bytes_limit_) {
return Status(TErrorCode::SCRATCH_LIMIT_EXCEEDED, bytes_limit_, GetBackendString());
}
// Lazily create the files on the first write.
if (tmp_files_.empty()) RETURN_IF_ERROR(CreateFiles());
- // Track the indices of any directories where we failed due to capacity. This is
- // required for error reporting if we are totally out of capacity so that it's clear
- // that some disks were at capacity.
- vector<int> at_capacity_dirs;
-
// Find the next physical file in priority based round-robin order and allocate a range
// from it.
for (const auto& entry: tmp_files_index_range_) {
@@ -550,18 +1101,52 @@ Status TmpFileGroup::AllocateSpace(
next_allocation_index_[priority] = start + (idx - start + 1) % (end - start + 1);
*tmp_file = tmp_files_[idx].get();
if ((*tmp_file)->is_blacklisted()) continue;
-
// Check the per-directory limit.
if (!(*tmp_file)->AllocateSpace(scratch_range_bytes, file_offset)) {
- at_capacity_dirs.push_back(idx);
+ at_capacity_dirs->push_back(idx);
continue;
}
- scratch_space_bytes_used_counter_->Add(scratch_range_bytes);
- tmp_file_mgr_->scratch_bytes_used_metric_->Increment(scratch_range_bytes);
- current_bytes_allocated_ += scratch_range_bytes;
+ UpdateScratchSpaceMetrics(scratch_range_bytes);
return Status::OK();
}
}
+
+ // Using a bool to notify there is no more space left, could cost less overhead than
+ // using a Status, because we want the error reporting as fast as possible for the
+ // case of mixing use of remote and local scratch space, so that it can keep trying to
+ // allocate from the remote after this.
+ *alloc_full = true;
+ return Status::OK();
+}
+
+Status TmpFileGroup::AllocateSpace(
+ int64_t num_bytes, TmpFile** tmp_file, int64_t* file_offset) {
+ // Since in eviction, it probably waits for the async upload task if it
+ // reaches bytes limit, so it can be slow here.
+ lock_guard<SpinLock> lock(lock_);
+
+ // Track the indices of any directories where we failed due to capacity. This is
+ // required for error reporting if we are totally out of capacity so that it's clear
+ // that some disks were at capacity.
+ vector<int> at_capacity_dirs;
+
+ if (!tmp_file_mgr_->tmp_dirs_.empty()) {
+ // If alloc_full is set true, meaning all of the local directories are at capacity.
+ bool alloc_full = false;
+ Status status = AllocateLocalSpace(
+ num_bytes, tmp_file, file_offset, &at_capacity_dirs, &alloc_full);
+ // If the all of the dirs are at capacity, try remote scratch space.
+ // Otherwise, return the status (could be an okay or error).
+ if (!status.ok() || !alloc_full) return status;
+ }
+
+ // If can't find any space locally, allocate from remote scratch space.
+ if (tmp_file_mgr_->tmp_dirs_remote_ != nullptr) {
+ Status remote_status =
+ AllocateRemoteSpace(num_bytes, tmp_file, file_offset, &at_capacity_dirs);
+ if (remote_status.ok() || at_capacity_dirs.empty()) return remote_status;
+ }
+
return ScratchAllocationFailedStatus(at_capacity_dirs);
}
@@ -580,14 +1165,15 @@ void TmpFileGroup::RecycleFileRange(unique_ptr<TmpWriteHandle> handle) {
}
scratch_space_bytes_used_counter_->Add(-space_used_bytes);
tmp_file_mgr_->scratch_bytes_used_metric_->Increment(-space_used_bytes);
- {
+ current_bytes_allocated_.Add(-space_used_bytes);
+ } else {
+ // For the remote files, we don't recycle the file and range because the remote file
+ // is not allowed to in-place modification.
+ if (file->DiskFile()->disk_type() == io::DiskFileType::LOCAL) {
+ int free_ranges_idx = BitUtil::Log2Ceiling64(space_used_bytes);
lock_guard<SpinLock> lock(lock_);
- current_bytes_allocated_ -= space_used_bytes;
+ free_ranges_[free_ranges_idx].emplace_back(file, handle->write_range_->offset());
}
- } else {
- int free_ranges_idx = BitUtil::Log2Ceiling64(space_used_bytes);
- lock_guard<SpinLock> lock(lock_);
- free_ranges_[free_ranges_idx].emplace_back(file, handle->write_range_->offset());
}
}
@@ -639,11 +1225,27 @@ Status TmpFileGroup::ReadAsync(TmpWriteHandle* handle, MemRange buffer) {
// Don't grab handle->write_state_lock_, it is safe to touch all of handle's state
// since the write is not in flight.
handle->read_range_ = scan_range_pool_.Add(new ScanRange);
- handle->read_range_->Reset(nullptr, handle->write_range_->file(),
- handle->write_range_->len(), handle->write_range_->offset(),
- handle->write_range_->disk_id(), false, ScanRange::INVALID_MTIME,
- BufferOpts::ReadInto(
- read_buffer.data(), read_buffer.len(), BufferOpts::NO_CACHING));
+
+ if (handle->file_ != nullptr && !handle->file_->is_local()) {
+ TmpFileRemote* tmp_file = static_cast<TmpFileRemote*>(handle->file_);
+ DiskFile* disk_buffer_file = tmp_file->DiskBufferFile();
+ DiskFile* disk_file = tmp_file->DiskFile();
+ // Reset the read_range, use the remote filesystem's disk id.
+ handle->read_range_->Reset(tmp_file->hdfs_conn_, disk_file->path().c_str(),
+ handle->write_range_->len(), handle->write_range_->offset(), tmp_file->disk_id(),
+ false, tmp_file->mtime_,
+ BufferOpts::ReadInto(
+ read_buffer.data(), read_buffer.len(), BufferOpts::NO_CACHING),
+ nullptr, disk_file, disk_buffer_file);
+ } else {
+ // Read from local.
+ handle->read_range_->Reset(nullptr, handle->write_range_->file(),
+ handle->write_range_->len(), handle->write_range_->offset(),
+ handle->write_range_->disk_id(), false, ScanRange::INVALID_MTIME,
+ BufferOpts::ReadInto(
+ read_buffer.data(), read_buffer.len(), BufferOpts::NO_CACHING));
+ }
+
read_counter_->Add(1);
bytes_read_counter_->Add(read_buffer.len());
bool needs_buffers;
@@ -809,13 +1411,17 @@ Status TmpFileGroup::ScratchAllocationFailedStatus(
}
vector<string> at_capacity_dir_paths;
for (int dir_idx : at_capacity_dirs) {
- at_capacity_dir_paths.push_back(tmp_file_mgr_->tmp_dirs_[dir_idx].path);
+ if (dir_idx >= tmp_file_mgr_->tmp_dirs_.size()) {
+ at_capacity_dir_paths.push_back(tmp_file_mgr_->tmp_dirs_remote_->path);
+ } else {
+ at_capacity_dir_paths.push_back(tmp_file_mgr_->tmp_dirs_[dir_idx].path);
+ }
}
Status status(TErrorCode::SCRATCH_ALLOCATION_FAILED, join(tmp_dir_paths, ","),
GetBackendString(),
PrettyPrinter::PrintBytes(
- tmp_file_mgr_->scratch_bytes_used_metric_->current_value()->GetValue()),
- PrettyPrinter::PrintBytes(current_bytes_allocated_),
+ tmp_file_mgr_->scratch_bytes_used_metric_->current_value()->GetValue()),
+ PrettyPrinter::PrintBytes(current_bytes_allocated_.Load()),
join(at_capacity_dir_paths, ","));
// Include all previous errors that may have caused the failure.
for (Status& err : scratch_errors_) status.MergeStatus(err);
@@ -831,7 +1437,7 @@ string TmpFileGroup::DebugString() {
lock_guard<SpinLock> lock(lock_);
stringstream ss;
ss << "TmpFileGroup " << this << " bytes limit " << bytes_limit_
- << " current bytes allocated " << current_bytes_allocated_
+ << " current bytes allocated " << current_bytes_allocated_.Load()
<< " next allocation index [ ";
// Get priority based allocation index.
for (const auto& entry: next_allocation_index_) {
@@ -869,6 +1475,11 @@ string TmpWriteHandle::TmpFilePath() const {
return file_->path();
}
+string TmpWriteHandle::TmpFileBufferPath() const {
+ if (file_ == nullptr) return "";
+ return file_->LocalBuffPath();
+}
+
int64_t TmpWriteHandle::on_disk_len() const {
return write_range_->len();
}
@@ -890,6 +1501,10 @@ Status TmpWriteHandle::Write(RequestContext* io_ctx, MemRange buffer,
// Allocate space after doing compression, to avoid overallocating space.
TmpFile* tmp_file;
int64_t file_offset;
+ Status status = Status::OK();
+
+ // For the second unpin of a page, it will be written to a new file since the
+ // content should be changed
RETURN_IF_ERROR(parent_->AllocateSpace(buffer_to_write.len(), &tmp_file, &file_offset));
if (FLAGS_disk_spill_encryption) {
@@ -900,13 +1515,19 @@ Status TmpWriteHandle::Write(RequestContext* io_ctx, MemRange buffer,
// WriteComplete() may be called concurrently with the remainder of this function.
data_len_ = buffer.len();
file_ = tmp_file;
- write_range_.reset(new WriteRange(
- tmp_file->path(), file_offset, tmp_file->AssignDiskQueue(), callback));
+ write_range_.reset(new WriteRange(tmp_file->path(), file_offset,
+ tmp_file->AssignDiskQueue(!tmp_file->is_local()), callback));
write_range_->SetData(buffer_to_write.data(), buffer_to_write.len());
+ // For remote files, we write the range to the local buffer.
+ write_range_->SetDiskFile(tmp_file->GetWriteFile());
VLOG(3) << "Write " << tmp_file->path() << " " << file_offset << " "
<< buffer_to_write.len();
write_in_flight_ = true;
- Status status = io_ctx->AddWriteRange(write_range_.get());
+
+ write_range_->SetRequestContext(io_ctx);
+ // Add the write range asyncly to the DiskQueue for writing.
+ status = parent_->tmp_file_mgr()->AsyncWriteRange(write_range_.get(), tmp_file);
+
if (!status.ok()) {
// The write will not be in flight if we returned with an error.
write_in_flight_ = false;
@@ -963,6 +1584,7 @@ Status TmpWriteHandle::RetryWrite(RequestContext* io_ctx, TmpFile* file, int64_t
DCHECK(write_in_flight_);
file_ = file;
write_range_->SetRange(file->path(), offset, file->AssignDiskQueue());
+ write_range_->SetDiskFile(file->GetWriteFile());
Status status = io_ctx->AddWriteRange(write_range_.get());
if (!status.ok()) {
// The write will not be in flight if we returned with an error.
@@ -972,8 +1594,23 @@ Status TmpWriteHandle::RetryWrite(RequestContext* io_ctx, TmpFile* file, int64_t
return Status::OK();
}
+void TmpWriteHandle::UploadComplete(TmpFile* tmp_file, const Status& upload_status) {
+ if (upload_status.ok()) {
+ // If uploaded, the local buffer is available to be evicted, so enqueue it to the
+ // pool.
+ DCHECK(tmp_file != nullptr);
+ TmpFileGroup* file_group = tmp_file->file_group_;
+ file_group->tmp_file_mgr_->EnqueueTmpFilesPool(
+ file_group->FindTmpFileSharedPtr(tmp_file),
+ file_group->tmp_file_mgr_->GetRemoteTmpFileBufferPoolLifo());
+ } else {
+ LOG(WARNING) << "Upload temporary file: '" << tmp_file->path() << " failed";
+ }
+}
+
void TmpWriteHandle::WriteComplete(const Status& write_status) {
WriteDoneCallback cb;
+ Status status = write_status;
{
lock_guard<mutex> lock(write_state_lock_);
DCHECK(write_in_flight_);
@@ -987,13 +1624,30 @@ void TmpWriteHandle::WriteComplete(const Status& write_status) {
FreeCompressedBuffer();
}
+ if (status.ok() && !file_->expected_local_) {
+ // Do file upload if the local buffer file is finished.
+ if (write_range_->is_full()) {
+ TmpFileRemote* tmp_file = static_cast<TmpFileRemote*>(file_);
+ int disk_id = tmp_file->DiskFile()->disk_type() == io::DiskFileType::S3 ?
+ parent_->io_mgr_->RemoteS3DiskFileOperId() :
+ parent_->io_mgr_->RemoteDfsDiskFileOperId();
+ RemoteOperRange::RemoteOperDoneCallback u_callback =
+ [this, tmp_file](
+ const Status& upload_status) { UploadComplete(tmp_file, upload_status); };
+ tmp_file->upload_range_.reset(new RemoteOperRange(tmp_file->DiskBufferFile(),
+ tmp_file->DiskFile(), parent_->tmp_file_mgr()->GetRemoteTmpBlockSize(),
+ disk_id, RequestType::FILE_UPLOAD, parent_->io_mgr_, u_callback));
+ status = parent_->io_ctx_->AddRemoteOperRange(tmp_file->upload_range_.get());
+ }
+ }
+
// Notify before releasing the lock - after the lock is released 'this' may be
// destroyed.
write_complete_cv_.NotifyAll();
}
// Call 'cb' last - once 'cb' is called client code may call Read() or destroy this
// handle.
- cb(write_status);
+ cb(status);
}
void TmpWriteHandle::Cancel() {
@@ -1079,4 +1733,233 @@ string TmpWriteHandle::DebugString() {
}
return ss.str();
}
+
+TmpFileBufferPool::TmpFileBufferPool(TmpFileMgr* tmp_file_mgr)
+ : tmp_file_mgr_(tmp_file_mgr) {
+ tmp_file_dummy_.reset(new TmpFileDummy());
+}
+
+TmpFileBufferPool::~TmpFileBufferPool() {
+ {
+ unique_lock<mutex> l(lock_);
+ shut_down_ = true;
+ }
+ // Wake up the waiting thread.
+ work_available_.NotifyAll();
+}
+
+void TmpFileBufferPool::TmpFileSpaceReserveThreadLoop() {
+ while (true) {
+ {
+ unique_lock<mutex> l(lock_);
+ while (!shut_down_ && write_ranges_.empty()) {
+ // Wait if there are no ranges in the queue.
+ work_available_.Wait(l);
+ }
+ if (shut_down_) return;
+ DCHECK(!write_ranges_.empty());
+ cur_write_range_ = write_ranges_.front();
+ write_ranges_.pop_front();
+ DCHECK(cur_write_range_ != nullptr);
+
+ // Find out the TmpFile which the current range is associated with, and store the
+ // shared_ptr of the file to cur_tmp_file_ in case it is deconstructed while waiting
+ // for reservation.
+ auto it = write_ranges_iterator_.find(cur_write_range_);
+ DCHECK(it != write_ranges_iterator_.end());
+ TmpFile* tmp_file = it->second.second;
+ cur_tmp_file_ = tmp_file->FileGroup()->FindTmpFileSharedPtr(tmp_file);
+ DCHECK(cur_tmp_file_ != nullptr);
+ DCHECK_EQ(cur_write_range_->disk_file(), cur_tmp_file_->GetWriteFile());
+ write_ranges_iterator_.erase(it);
+ }
+
+ // Reserve space from the tmp_files_avail_pool_. The process could need a long wait.
+ Status status = tmp_file_mgr_->ReserveLocalBufferSpace(false);
+ vector<TmpFileMgr::WriteDoneCallback> write_callbacks;
+ {
+ unique_lock<mutex> lock(lock_);
+ if (status.ok()) {
+ DCHECK(cur_tmp_file_ != nullptr);
+ cur_tmp_file_->GetWriteFile()->SetSpaceReserved();
+ if (cur_write_range_ != nullptr) {
+ // Send all of the writes of the same disk file to the disk queue.
+ status = MoveWriteRangesHelper(
+ cur_write_range_->disk_file(), &write_callbacks, false);
+ } else {
+ // If the current range becomes a nullptr, it must be set by
+ // RemoveWriteRanges(). In this case, the io_ctx which the range belongs to is
+ // cancelled, and all the writes using that io_ctx are already cancelled. So, we
+ // are safe to return the TmpFile to the pool to recycle the buffer space.
+ EnqueueTmpFilesPool(cur_tmp_file_, true);
+ }
+ } else if (!status.ok() && cur_write_range_ != nullptr) {
+ // Cancel the spilling if fails to reserve the buffer.
+ RemoveWriteRangesInternal(cur_write_range_->io_ctx(), &write_callbacks);
+ status = Status::CancelledInternal(
+ Substitute("TmpFileBufferPool because: $0", status.GetDetail()).c_str());
+ }
+ cur_write_range_ = nullptr;
+ cur_tmp_file_.reset();
+ }
+ for (const TmpFileMgr::WriteDoneCallback& write_callback : write_callbacks) {
+ write_callback(status);
+ }
+ }
+}
+
+Status TmpFileBufferPool::MoveWriteRangesHelper(DiskFile* disk_file,
+ vector<TmpFileMgr::WriteDoneCallback>* write_callbacks, bool is_cancelled) {
+ Status status = Status::OK();
+ auto write_ranges_it = write_ranges_to_add_.find(disk_file);
+ if (write_ranges_it != write_ranges_to_add_.end()) {
+ auto write_range_it = write_ranges_it->second.begin();
+ while (write_range_it != write_ranges_it->second.end()) {
+ auto range = *write_range_it;
+ DCHECK(range != nullptr);
+ if (status.ok() && !is_cancelled) {
+ status = range->io_ctx()->AddWriteRange(range);
+ } else {
+ write_callbacks->push_back(range->callback());
+ if (is_cancelled && range->offset() == 0) {
+ // If is_cancelled is set, try to remove the range from the write_ranges list.
+ // If the range hasn't been popped, it must still be in the write_ranges list.
+ if (cur_write_range_ != range) {
+ auto key_range_it = write_ranges_iterator_.find(range);
+ DCHECK(key_range_it != write_ranges_iterator_.end());
+ DCHECK_EQ(*(key_range_it->second.first), range);
+ write_ranges_.erase(key_range_it->second.first);
+ write_ranges_iterator_.erase(key_range_it);
+ }
+ }
+ }
+ write_range_it = write_ranges_it->second.erase(write_range_it);
+ }
+ write_ranges_to_add_.erase(write_ranges_it);
+ }
+ return status;
+}
+
+Status TmpFileBufferPool::EnqueueWriteRange(io::WriteRange* range, TmpFile* tmp_file) {
+ Status status = Status::OK();
+ {
+ unique_lock<mutex> write_range_list_lock(lock_);
+ DCHECK(range != nullptr);
+ DCHECK(range->disk_file() != nullptr);
+ DCHECK(range->io_ctx() != nullptr);
+ if (range->disk_file()->IsSpaceReserved()) {
+ // If the space is reserved, send the range to the DiskQueue.
+ return range->io_ctx()->AddWriteRange(range);
+ } else if (range->io_ctx()->IsCancelled()) {
+ // If the io_ctx is cancelled, nofity the caller to cancel the query.
+ return TMP_FILE_BUFFER_POOL_CONTEXT_CANCELLED;
+ } else {
+ io_ctx_to_file_set_map_[range->io_ctx()].insert(range->disk_file());
+ write_ranges_to_add_[range->disk_file()].emplace_back(range);
+ }
+ // Put the first range of a file to the queue for waiting for the available space,
+ // the ranges in the queue would be popped one by one, when the space is reserved,
+ // all ranges of the file are added to the DiskQueue by io_ctx.
+ if (range->offset() == 0) {
+ write_ranges_.emplace_back(range);
+ DCHECK(tmp_file != nullptr);
+ write_ranges_iterator_[range] =
+ std::make_pair(prev(write_ranges_.cend()), tmp_file);
+ }
+ }
+ work_available_.NotifyAll();
+ return status;
+}
+
+void TmpFileBufferPool::RemoveWriteRangesInternal(
+ RequestContext* io_ctx, vector<TmpFileMgr::WriteDoneCallback>* write_callbacks) {
+ auto file_set_it = io_ctx_to_file_set_map_.find(io_ctx);
+ if (file_set_it != io_ctx_to_file_set_map_.end()) {
+ auto file_it = file_set_it->second.begin();
+ while (file_it != file_set_it->second.end()) {
+ DCHECK(*file_it != nullptr);
+ // Remove all the ranges belonging to the file, and fetch the callback
+ // functions of the ranges.
+ Status status = MoveWriteRangesHelper(*file_it, write_callbacks, true);
+ DCHECK_OK(status);
+ if (cur_write_range_ != nullptr && *file_it == cur_write_range_->disk_file()) {
+ // Set the current write range to nullptr if the TmpFileGroup is closing to
+ // notify the reservation thread (it is waiting for the reservation) that the
+ // space is no longer needed for the write range.
+ cur_write_range_ = nullptr;
+ }
+ file_it = file_set_it->second.erase(file_it);
+ }
+ io_ctx_to_file_set_map_.erase(file_set_it);
+ }
+}
+
+void TmpFileBufferPool::RemoveWriteRanges(RequestContext* io_ctx) {
+ DCHECK(io_ctx != nullptr);
+ vector<TmpFileMgr::WriteDoneCallback> write_callbacks;
+ {
+ unique_lock<mutex> lock(lock_);
+ RemoveWriteRangesInternal(io_ctx, &write_callbacks);
+ }
+ for (const TmpFileMgr::WriteDoneCallback& write_callback : write_callbacks) {
+ write_callback(TMP_FILE_BUFFER_POOL_CONTEXT_CANCELLED);
+ }
+}
+
+void TmpFileBufferPool::EnqueueTmpFilesPool(shared_ptr<TmpFile>& tmp_file, bool front) {
+ DCHECK(tmp_file != nullptr);
+ {
+ unique_lock<mutex> buffer_lock(tmp_files_avail_pool_lock_);
+ if (tmp_file->disk_type() != io::DiskFileType::DUMMY) {
+ TmpFileRemote* tmp_file_remote = static_cast<TmpFileRemote*>(tmp_file.get());
+ if (tmp_file_remote->is_enqueued()) return;
+ tmp_file_remote->SetEnqueued(true);
+ tmp_file_remote->SetBufferReturned();
+ }
+ if (front) {
+ tmp_files_avail_pool_.push_front(tmp_file);
+ } else {
+ tmp_files_avail_pool_.push_back(tmp_file);
+ }
+ tmp_file_mgr_->GetLocalBufferDir()->bytes_used_metric->Increment(
+ -1 * tmp_file_mgr_->GetRemoteTmpFileSize());
+ }
+ tmp_files_available_cv_.NotifyOne();
+}
+
+Status TmpFileBufferPool::DequeueTmpFilesPool(
+ shared_ptr<TmpFile>* tmp_file, bool quick_return) {
+ DCHECK(tmp_file != nullptr);
+ DCHECK(dequeue_timer_metric_ != nullptr);
+ ScopedHistogramTimer wait_timer(dequeue_timer_metric_);
+ unique_lock<mutex> buffer_lock(tmp_files_avail_pool_lock_);
+ // If quick return is set and no buffer is available, return immediately.
+ if (quick_return && tmp_files_avail_pool_.empty()) {
+ return TMP_FILE_MGR_NO_AVAILABLE_FILE_TO_EVICT;
+ }
+ while (tmp_files_avail_pool_.empty()) {
+ // Wait if there is no temporary file on the queue.
+ // If timeout, return immediately.
+ if (!tmp_files_available_cv_.WaitFor(
+ buffer_lock, tmp_file_mgr_->GetSpillBufferWaitTimeout())) {
+ return Status(Substitute("Timeout waiting for a local buffer in $0 seconds",
+ tmp_file_mgr_->GetSpillBufferWaitTimeout() / MICROS_PER_SEC));
+ };
+ }
+ DCHECK(!tmp_files_avail_pool_.empty());
+ *tmp_file = tmp_files_avail_pool_.front();
+ tmp_files_avail_pool_.pop_front();
+ DCHECK(*tmp_file != nullptr);
+ if ((*tmp_file)->disk_type() != io::DiskFileType::DUMMY) {
+ TmpFileRemote* tmp_file_remote = static_cast<TmpFileRemote*>(tmp_file->get());
+ // Assert the default size remains the same in case the object is corrupted.
+ DCHECK_EQ(tmp_file_remote->file_size_, tmp_file_mgr_->GetRemoteTmpFileSize());
+ DCHECK(tmp_file_remote->is_enqueued());
+ tmp_file_remote->SetEnqueued(false);
+ }
+ tmp_file_mgr_->GetLocalBufferDir()->bytes_used_metric->Increment(
+ tmp_file_mgr_->GetRemoteTmpFileSize());
+ return Status::OK();
+}
+
} // namespace impala
diff --git a/be/src/runtime/tmp-file-mgr.h b/be/src/runtime/tmp-file-mgr.h
index e088dd9..179db22 100644
--- a/be/src/runtime/tmp-file-mgr.h
+++ b/be/src/runtime/tmp-file-mgr.h
@@ -22,15 +22,19 @@
#include <memory>
#include <unordered_map>
#include <utility>
+#include <bits/stdc++.h>
#include <mutex>
#include <boost/scoped_ptr.hpp>
+#include <boost/thread/shared_mutex.hpp>
#include <gtest/gtest_prod.h>
+#include "common/atomic.h"
#include "common/object-pool.h"
#include "common/status.h"
#include "gen-cpp/CatalogObjects_types.h" // for THdfsCompression
#include "gen-cpp/Types_types.h" // for TUniqueId
+#include "hdfs-fs-cache.h"
#include "runtime/scoped-buffer.h"
#include "util/condition-variable.h"
#include "util/mem-range.h"
@@ -38,6 +42,7 @@
#include "util/openssl-util.h"
#include "util/runtime-profile.h"
#include "util/spinlock.h"
+#include "util/thread.h"
namespace impala {
namespace io {
@@ -45,10 +50,13 @@ namespace io {
class RequestContext;
class ScanRange;
class WriteRange;
+ class RemoteOperRange;
}
struct BufferPoolClientCounters;
class MemTracker;
class TmpFile;
+class TmpFileRemote;
+class TmpFileBufferPool;
class TmpFileGroup;
class TmpWriteHandle;
@@ -85,6 +93,32 @@ class TmpWriteHandle;
/// TmpFileMgr provides some basic support for managing local disk space consumption.
/// A TmpFileGroup can be created with a limit on the total number of bytes allocated
/// across all files. Writes that would exceed the limit fail with an error status.
+/// The TmpFileBufferPool provides the ability for write ranges to wait for local
+/// buffer space in a separate thread during spilling to remote file system, the purpose
+/// of the pool is to handle the competitions for buffer files allocation with limited
+/// local space.
+///
+/// Locking:
+/// During spilling, multiple locks could be acquired, to avoid deadlocks, the order
+/// of acquiring must be from lower-numbered locks to higher-numbered locks:
+/// 1. BufferPool::Client::lock_
+/// 2. BufferPool::Page::buffer_lock
+/// 3. TmpFileGroup::lock_
+/// 4. TmpFileBufferPool::lock_
+/// 5. TmpWriteHandle::write_state_lock_
+/// 6. RequestContext::lock_
+/// 7. ScanRange::lock_
+/// 8. DiskFile::physical_file_lock_
+/// Specially, it may need to acquire both locks of two types of DiskFile during spilling
+/// to remote filesystem, the order to acquire DiskFile's locks must be from local
+/// DiskFile to remote DiskFile.
+///
+/// Ohter than the locks above, terminal locks, meaning no other locks should be acquired
+/// while holding this one, could be used during spilling:
+/// TmpFileBufferPool::tmp_files_avail_pool_lock_
+/// TmpFileGroup::tmp_files_remote_ptrs_lock_
+/// DiskQueue::lock_
+/// DiskFile::status_lock_
///
/// TODO: IMPALA-4683: we could implement smarter handling of failures, e.g. to
/// temporarily blacklist devices that show I/O errors.
@@ -101,22 +135,62 @@ class TmpFileMgr {
/// A configured temporary directory that TmpFileMgr allocates files in.
struct TmpDir {
TmpDir(const std::string& path, int64_t bytes_limit, int priority,
- IntGauge* bytes_used_metric)
- : path(path), bytes_limit(bytes_limit), priority(priority),
- bytes_used_metric(bytes_used_metric) {}
+ IntGauge* bytes_used_metric, bool is_local_dir = true)
+ : path(path),
+ bytes_limit(bytes_limit),
+ priority(priority),
+ bytes_used_metric(bytes_used_metric),
+ is_local_dir(is_local_dir) {}
/// Path to the temporary directory.
- const std::string path;
+ std::string path;
/// Limit on bytes that should be written to this path. Set to maximum value
/// of int64_t if there is no limit.
- const int64_t bytes_limit;
+ int64_t bytes_limit;
/// Scratch directory priority.
- const int priority;
+ int priority;
/// The current bytes of scratch used for this temporary directory.
- IntGauge* const bytes_used_metric;
+ IntGauge* bytes_used_metric;
+
+ /// If the dir is expected in the local file system or in the remote.
+ bool is_local_dir;
+ };
+
+ /// A configuration for the control parameters of remote temporary directories.
+ /// The struct is used by TmpFileMgr and has the same lifecycle as TmpFileMgr.
+ struct TmpDirRemoteCtrl {
+ /// The high water mark metric for local buffer directory.
+ AtomicInt64 local_buff_dir_bytes_high_water_mark_{0};
+
+ /// The default size of a remote temporary file.
+ int64_t remote_tmp_file_size_;
+
+ /// The default block size of a remote temporary file. The block is used as a buffer
+ /// while doing upload and fetch a remote temporary file.
+ int64_t remote_tmp_block_size_;
+
+ /// Specify the mode to enqueue the tmp file to the pool.
+ /// If true, the file would be placed in the first to be poped out from the pool.
+ /// If false, the file would be placed in the last of the pool.
+ bool remote_tmp_files_avail_pool_lifo_;
+
+ /// The spill to hdfs is a test-only feature, and is not allowed unless the option of
+ /// allow_spill_to_hdfs is set true.
+ bool allow_spill_to_hdfs_;
+
+ /// Temporary file buffer pool managed by TmpFileMgr, is only activated when there is
+ /// a remote scratch space is registered. So, if TmpFileMgr::HasRemoteDir() is true,
+ /// the tmp_file_pool_ is non-null. Otherwise, it is null.
+ std::unique_ptr<TmpFileBufferPool> tmp_file_pool_;
+
+ /// Thread group containing threads created by the TmpFileMgr.
+ ThreadGroup tmp_file_mgr_thread_group_;
+
+ /// Timeout duration for waiting for the buffer (us).
+ int64_t wait_for_spill_buffer_timeout_us_;
};
TmpFileMgr();
@@ -139,9 +213,50 @@ class TmpFileMgr {
bool one_dir_per_device, const std::string& compression_codec, bool punch_holes,
MetricGroup* metrics) WARN_UNUSED_RESULT;
+ /// A helper function for InitCustom() to create a scratch directory.
+ Status CreateDirectory(const string& scratch_subdir_path, const string& tmp_path,
+ const std::unique_ptr<TmpDir>& tmp_dir, MetricGroup* metrics,
+ vector<bool>* is_tmp_dir_on_disk, bool* has_remote_dir,
+ int disk_id) WARN_UNUSED_RESULT;
+
+ // Create the TmpFile buffer pool thread for async buffer file reservation.
+ Status CreateTmpFileBufferPoolThread(MetricGroup* metrics) WARN_UNUSED_RESULT;
+
+ /// Try to reserve space for the buffer file from local buffer directory.
+ /// If quick_return is true, the function won't wait if there is no available space.
+ Status ReserveLocalBufferSpace(bool quick_return) WARN_UNUSED_RESULT;
+
/// Return the scratch directory path for the device.
std::string GetTmpDirPath(DeviceId device_id) const;
+ /// Return the remote temporary file size.
+ int64_t GetRemoteTmpFileSize() const {
+ return tmp_dirs_remote_ctrl_.remote_tmp_file_size_;
+ }
+
+ /// Return the remote temporary block size.
+ int64_t GetRemoteTmpBlockSize() const {
+ return tmp_dirs_remote_ctrl_.remote_tmp_block_size_;
+ }
+
+ /// Return the timeout duration of waiting for spill buffer.
+ int64_t GetSpillBufferWaitTimeout() const {
+ return tmp_dirs_remote_ctrl_.wait_for_spill_buffer_timeout_us_;
+ }
+
+ /// Return if return the remote temporary file to the pool by LIFO.
+ /// If false is returned, it is set by FIFO.
+ bool GetRemoteTmpFileBufferPoolLifo() {
+ return tmp_dirs_remote_ctrl_.remote_tmp_files_avail_pool_lifo_;
+ }
+
+ /// Return the local buffer dir for remote spilling.
+ TmpDir* GetLocalBufferDir() const;
+
+ /// Total number of devices with tmp directories that are active in local file system.
+ /// There is one tmp directory per device.
+ int NumActiveTmpDevicesLocal();
+
/// Total number of devices with tmp directories that are active. There is one tmp
/// directory per device.
int NumActiveTmpDevices();
@@ -150,6 +265,45 @@ class TmpFileMgr {
/// I.e. those that haven't been blacklisted.
std::vector<DeviceId> ActiveTmpDevices();
+ /// Add the write range to the TmpFileBufferPool for waiting the reservation of the
+ /// buffer of the TmpFile the range writes into. The pool would send the range to the
+ /// disk queue once reservation is done. If the space of the buffer is already reserved,
+ /// an error would return, and the caller should enqueue the range to the disk queue by
+ /// itself.
+ Status AsyncWriteRange(io::WriteRange* write_range, TmpFile* file);
+
+ /// A helper function to enqueue a dummy temporary file to the TmpFileBufferPool.
+ void EnqueueTmpFilesPoolDummyFile();
+
+ /// Enqueue the temporary files into the pool when the local buffer of the file is ready
+ /// to be deleted to release buffer space.
+ /// If front set to true, the file is enqueued to the front of the pool and is to be
+ /// popped first.
+ void EnqueueTmpFilesPool(std::shared_ptr<TmpFile>& tmp_file, bool front);
+
+ /// Dequeue the temporary files from the pool. The caller which successfully
+ /// dequeues a file from the pool has the right to delete the local buffer of the file
+ /// and create its buffer.
+ /// The function is called by TryEvictFile() to gain space from the pool if the
+ /// local buffer directory reaches byte limit. If there is no file in the pool, the
+ /// caller needs to wait until a file is enqueued (often after a file is uploaded).
+ /// It could be possible a long wait if no file is uploaded due a jammed disk queue
+ /// or a slow network, so it is not recommended to hold an exclusive lock while calling
+ /// the function.
+ /// If quick_return is true, one won't wait if no file is in the pool. Otherwise,
+ /// one would wait util a file is dequeued.
+ Status DequeueTmpFilesPool(std::shared_ptr<TmpFile>* tmp_file, bool quick_return);
+
+ /// Try to delete the buffer of a TmpFile to make some space for other buffers.
+ /// May return an error status if error happens during deletion of the buffer.
+ Status TryEvictFile(TmpFile* tmp_file);
+
+ /// If a remote scratch space is registered in the TmpFileMgr.
+ bool HasRemoteDir() { return tmp_dirs_remote_ != nullptr; }
+
+ /// Return default S3 options for spilling.
+ const vector<std::pair<string, string>>* s3a_options() { return &s3a_options_; }
+
MemTracker* compressed_buffer_tracker() const {
return compressed_buffer_tracker_.get();
}
@@ -169,9 +323,10 @@ class TmpFileMgr {
static constexpr int64_t HOLE_PUNCH_BLOCK_SIZE_BYTES = 4096;
private:
- friend class TmpFileMgrTest;
friend class TmpFile;
+ friend class TmpFileRemote;
friend class TmpFileGroup;
+ friend class TmpFileMgrTest;
/// Return a new TmpFile handle with a path based on file_group->unique_id. The file is
/// associated with the 'file_group' and the file path is within the (single) scratch
@@ -181,6 +336,9 @@ class TmpFileMgr {
void NewFile(TmpFileGroup* file_group, DeviceId device_id,
std::unique_ptr<TmpFile>* new_file);
+ /// Remove the remote directory which stores tmp files of the tmp file group.
+ void RemoveRemoteDir(TmpFileGroup* file_group, DeviceId device_id);
+
bool initialized_ = false;
/// The type of spill-to-disk compression in use for spilling. NONE means no
@@ -194,9 +352,25 @@ class TmpFileMgr {
/// Whether hole punching is enabled.
bool punch_holes_ = false;
- /// The paths of the created tmp directories.
+ /// The paths of the created tmp directories, which are used for spilling to local
+ /// filesystem.
std::vector<TmpDir> tmp_dirs_;
+ /// The paths of remote directories, which are used for spilling to remote filesystem.
+ std::unique_ptr<TmpDir> tmp_dirs_remote_;
+
+ /// The control parameters for remote temporary directories.
+ TmpDirRemoteCtrl tmp_dirs_remote_ctrl_;
+
+ /// The path of the directory to store local buffers for remote temporary files.
+ std::unique_ptr<TmpDir> local_buff_dir_;
+
+ /// Default S3 options for spilling to S3.
+ HdfsFsCache::HdfsConnOptions s3a_options_;
+
+ /// Local cache for HDFS connection handle.
+ HdfsFsCache::HdfsFsMap hdfs_conns_;
+
/// Memory tracker to track compressed buffers. Set up in InitCustom() if disk spill
/// compression is enabled
std::unique_ptr<MemTracker> compressed_buffer_tracker_;
@@ -236,6 +410,10 @@ class TmpFileGroup {
/// compression). The caller should not modify the data in 'buffer' until the write
/// completes or is cancelled, otherwise invalid data may be written to disk.
///
+ /// The write may take some time to complete. It may be queued behind other I/O
+ /// operations. If remote scratch is enabled, it may also need to wait for other queries
+ /// to make progress and release space in the local buffer directory.
+ ///
/// Returns an error if the scratch space cannot be allocated or the write cannot
/// be started. Otherwise 'handle' is set and 'cb' will be called asynchronously from
/// a different thread when the write completes successfully or unsuccessfully or is
@@ -281,6 +459,17 @@ class TmpFileGroup {
/// Calls Remove() on all the files in the group and deletes them.
void Close();
+ /// A function template to close files from local or remote scratch space.
+ template <typename T>
+ void CloseInternal(vector<T>& tmp_files);
+
+ /// Update the corresponding metrics of scratch space after new scratch space
+ /// is allocated.
+ void UpdateScratchSpaceMetrics(int64_t num_bytes, bool is_remote = false);
+
+ /// Assemble and return a new path.
+ std::string GenerateNewPath(string& dir, string& unique_name);
+
std::string DebugString();
const TUniqueId& unique_id() const { return unique_id_; }
@@ -292,10 +481,18 @@ class TmpFileGroup {
/// Return true if spill-to-disk failed due to local faulty disk.
bool IsSpillingDiskFaulty();
+ /// Return the shared_ptr of a remote TmpFile.
+ std::shared_ptr<TmpFile>& FindTmpFileSharedPtr(TmpFile* tmp_file);
+
private:
friend class TmpFile;
+ friend class TmpFileLocal;
+ friend class TmpFileRemote;
friend class TmpFileMgrTest;
friend class TmpWriteHandle;
+ friend class io::WriteRange;
+ friend class io::ScanRange;
+ friend class io::RemoteOperRange;
/// Initializes the file group with one temporary file per disk with a scratch
/// directory. Returns OK if at least one temporary file could be created.
@@ -309,6 +506,17 @@ class TmpFileGroup {
Status AllocateSpace(
int64_t num_bytes, TmpFile** tmp_file, int64_t* file_offset) WARN_UNUSED_RESULT;
+ /// Try to allocate 'num_bytes' bytes from local scratch space. Called by the
+ /// AllocateSpace().
+ /// The alloc_full returns true, if all of the directories are at capacity.
+ Status AllocateLocalSpace(int64_t num_bytes, TmpFile** tmp_file, int64_t* file_offset,
+ vector<int>* at_capacity_dirs, bool* alloc_full) WARN_UNUSED_RESULT;
+
+ /// Try to allocate 'num_bytes' bytes from remote scratch space when there is no
+ /// space left in the local scratch space. Called by the AllocateSpace().
+ Status AllocateRemoteSpace(int64_t num_bytes, TmpFile** tmp_file, int64_t* file_offset,
+ vector<int>* at_capacity_dirs) WARN_UNUSED_RESULT;
+
/// Recycle the range of bytes in a scratch file and destroy 'handle'. Called when the
/// range is no longer in use for 'handle'. The disk space associated with the file can
/// be reclaimed once this function, either by adding it to 'free_ranges_' for
@@ -389,6 +597,12 @@ class TmpFileGroup {
/// is not enabled.
RuntimeProfile::Counter* compression_timer_;
+ /// Protects tmp_files_remote_ptrs_lock_.
+ SpinLock tmp_files_remote_ptrs_lock_;
+
+ /// A map of raw pointer and its shared_ptr of remote TmpFiles.
+ std::unordered_map<TmpFile*, std::shared_ptr<TmpFile>> tmp_files_remote_ptrs_;
+
/// Protects below members.
SpinLock lock_;
@@ -404,6 +618,9 @@ class TmpFileGroup {
/// or getting disk error when reading temporary file back.
bool spilling_disk_faulty_;
+ /// List of remote files representing the TmpFileGroup.
+ std::vector<std::shared_ptr<TmpFile>> tmp_files_remote_;
+
/// Index Range in the 'tmp_files'. Used to keep track of index range
/// corresponding to a given priority.
struct TmpFileIndexRange {
@@ -419,7 +636,10 @@ class TmpFileGroup {
std::map<int, TmpFileIndexRange> tmp_files_index_range_;
/// Total space allocated in this group's files.
- int64_t current_bytes_allocated_;
+ AtomicInt64 current_bytes_allocated_;
+
+ /// Total space allocated remotely in this group's files.
+ AtomicInt64 current_bytes_allocated_remote_;
/// Index into 'tmp_files' denoting the file to which the next temporary file range
/// should be allocated from, for a given priority. Used to implement round-robin
@@ -464,6 +684,11 @@ class TmpWriteHandle {
/// Returns empty string if no backing file allocated.
std::string TmpFilePath() const;
+ /// Buffer path of temporary file backing the block. Intended for use in testing.
+ /// Returns empty string if no backing file allocated or the temporary file is not in
+ /// remote.
+ std::string TmpFileBufferPath() const;
+
/// The length of the in-memory data written to disk in bytes, before any compression.
int64_t data_len() const { return data_len_; }
@@ -510,6 +735,10 @@ class TmpWriteHandle {
/// then calls 'cb_'.
void WriteComplete(const Status& write_status);
+ /// Called when the upload has completed successfully or not.
+ /// Enqueue the file into the available pool if the upload succeeds.
+ void UploadComplete(TmpFile* file, const Status& write_status);
+
/// Cancels any in-flight writes or reads. Reads are cancelled synchronously and
/// writes are cancelled asynchronously. After Cancel() is called, writes are not
/// retried. The write callback may be called with a CANCELLED_INTERNALLY status
@@ -560,9 +789,8 @@ class TmpWriteHandle {
/// Protects all fields below while 'write_in_flight_' is true. At other times, it is
/// invalid to call WriteRange/TmpFileGroup methods concurrently from multiple
- /// threads,
- /// so no locking is required. This is a terminal lock and should not be held while
- /// acquiring other locks or invoking 'cb_'.
+ /// threads, so no locking is required.
+ /// The lock should not be held while invoking 'cb_' to avoid a deadlock.
std::mutex write_state_lock_;
/// True if the the write has been cancelled (but is not necessarily complete).
diff --git a/be/src/util/hdfs-util.cc b/be/src/util/hdfs-util.cc
index 5ba8872..bb4014a 100644
--- a/be/src/util/hdfs-util.cc
+++ b/be/src/util/hdfs-util.cc
@@ -29,6 +29,13 @@
namespace impala {
+const char* FILESYS_PREFIX_HDFS = "hdfs://";
+const char* FILESYS_PREFIX_S3 = "s3a://";
+const char* FILESYS_PREFIX_ABFS = "abfs://";
+const char* FILESYS_PREFIX_ABFS_SEC = "abfss://";
+const char* FILESYS_PREFIX_ADL = "adl://";
+const char* FILESYS_PREFIX_OZONE = "o3fs://";
+
string GetHdfsErrorMsg(const string& prefix, const string& file) {
string error_msg = GetStrErrMsg();
stringstream ss;
@@ -73,40 +80,36 @@ Status CopyHdfsFile(const hdfsFS& src_conn, const string& src_path,
return Status::OK();
}
-bool IsHdfsPath(const char* path) {
- if (strstr(path, ":/") == NULL) {
- return ExecEnv::GetInstance()->default_fs().compare(0, 7, "hdfs://") == 0;
+bool IsSpecificPath(
+ const char* path, const char* specific_prefix, bool check_default_fs) {
+ size_t prefix_len = strlen(specific_prefix);
+ if (check_default_fs && strstr(path, ":/") == NULL) {
+ return strncmp(
+ ExecEnv::GetInstance()->default_fs().c_str(), specific_prefix, prefix_len)
+ == 0;
}
- return strncmp(path, "hdfs://", 7) == 0;
+ return strncmp(path, specific_prefix, prefix_len) == 0;
}
-bool IsS3APath(const char* path) {
- if (strstr(path, ":/") == NULL) {
- return ExecEnv::GetInstance()->default_fs().compare(0, 6, "s3a://") == 0;
- }
- return strncmp(path, "s3a://", 6) == 0;
+bool IsHdfsPath(const char* path, bool check_default_fs) {
+ return IsSpecificPath(path, FILESYS_PREFIX_HDFS, check_default_fs);
}
-bool IsABFSPath(const char* path) {
- if (strstr(path, ":/") == NULL) {
- return ExecEnv::GetInstance()->default_fs().compare(0, 7, "abfs://") == 0 ||
- ExecEnv::GetInstance()->default_fs().compare(0, 8, "abfss://") == 0;
- }
- return strncmp(path, "abfs://", 7) == 0 || strncmp(path, "abfss://", 8) == 0;
+bool IsS3APath(const char* path, bool check_default_fs) {
+ return IsSpecificPath(path, FILESYS_PREFIX_S3, check_default_fs);
}
-bool IsADLSPath(const char* path) {
- if (strstr(path, ":/") == NULL) {
- return ExecEnv::GetInstance()->default_fs().compare(0, 6, "adl://") == 0;
- }
- return strncmp(path, "adl://", 6) == 0;
+bool IsABFSPath(const char* path, bool check_default_fs) {
+ return IsSpecificPath(path, FILESYS_PREFIX_ABFS, check_default_fs)
+ || IsSpecificPath(path, FILESYS_PREFIX_ABFS_SEC, check_default_fs);
}
-bool IsOzonePath(const char* path) {
- if (strstr(path, ":/") == NULL) {
- return ExecEnv::GetInstance()->default_fs().compare(0, 7, "o3fs://") == 0;
- }
- return strncmp(path, "o3fs://", 7) == 0;
+bool IsADLSPath(const char* path, bool check_default_fs) {
+ return IsSpecificPath(path, FILESYS_PREFIX_ADL, check_default_fs);
+}
+
+bool IsOzonePath(const char* path, bool check_default_fs) {
+ return IsSpecificPath(path, FILESYS_PREFIX_OZONE, check_default_fs);
}
// Returns the length of the filesystem name in 'path' which is the length of the
diff --git a/be/src/util/hdfs-util.h b/be/src/util/hdfs-util.h
index 5bb5f8f..85e780a 100644
--- a/be/src/util/hdfs-util.h
+++ b/be/src/util/hdfs-util.h
@@ -24,6 +24,14 @@
namespace impala {
+/// Define prefix of remote file systems
+extern const char* FILESYS_PREFIX_HDFS;
+extern const char* FILESYS_PREFIX_S3;
+extern const char* FILESYS_PREFIX_ABFS;
+extern const char* FILESYS_PREFIX_ABFS_SEC;
+extern const char* FILESYS_PREFIX_ADL;
+extern const char* FILESYS_PREFIX_OZONE;
+
/// Utility function to get error messages from HDFS. This function takes prefix/file and
/// appends errno to it. Note: any stdlib function can reset errno, this should be called
/// immediately following the failed call into libhdfs.
@@ -44,19 +52,22 @@ Status CopyHdfsFile(const hdfsFS& src_conn, const std::string& src_path,
const hdfsFS& dst_conn, const std::string& dst_path);
/// Returns true iff the path refers to a location on an HDFS filesystem.
-bool IsHdfsPath(const char* path);
+/// If check_default_fs is true, the function checks and returns true if the default
+/// filesystem is HDFS when the path doen't contain any prefix like 'hdfs://'.
+/// If check_default_fs is false, the fucntion checks the path only.
+bool IsHdfsPath(const char* path, bool check_default_fs = true);
/// Returns true iff the path refers to a location on an S3A filesystem.
-bool IsS3APath(const char* path);
+bool IsS3APath(const char* path, bool check_default_fs = true);
/// Returns true iff the path refers to a location on an ABFS filesystem.
-bool IsABFSPath(const char* path);
+bool IsABFSPath(const char* path, bool check_default_fs = true);
/// Returns true iff the path refers to a location on an ADL filesystem.
-bool IsADLSPath(const char* path);
+bool IsADLSPath(const char* path, bool check_default_fs = true);
/// Returns true iff the path refers to a location on an Ozone filesystem.
-bool IsOzonePath(const char* path);
+bool IsOzonePath(const char* path, bool check_default_fs = true);
/// Returns true iff 'pathA' and 'pathB' are on the same filesystem.
bool FilesystemsMatch(const char* pathA, const char* pathB);
diff --git a/common/thrift/metrics.json b/common/thrift/metrics.json
index 1e1b11f..3b6d3f1 100644
--- a/common/thrift/metrics.json
+++ b/common/thrift/metrics.json
@@ -2331,6 +2331,26 @@
"key": "tmp-file-mgr.scratch-space-bytes-used.dir-$0"
},
{
+ "description": "The current total spilled bytes for the local buffer directory.",
+ "contexts": [
+ "IMPALAD"
+ ],
+ "label": "Per-directory local buffer space bytes used",
+ "units": "BYTES",
+ "kind": "GAUGE",
+ "key": "tmp-file-mgr.local-buff-bytes-used.dir-$0"
+ },
+ {
+ "description": "The durations of dequeuing from the TmpFileBufferPool.",
+ "contexts": [
+ "IMPALAD"
+ ],
+ "label": "Dequeue durations for the TmpFileBufferPool",
+ "units": "TIME_NS",
+ "kind": "HISTOGRAM",
+ "key": "tmp-file-mgr.tmp-file-buff-pool-dequeue-durations"
+ },
+ {
"description": "Number of senders waiting for receiving fragment to initialize",
"contexts": [
"IMPALAD"
diff --git a/tests/custom_cluster/test_scratch_disk.py b/tests/custom_cluster/test_scratch_disk.py
index 805489e..23df692 100644
--- a/tests/custom_cluster/test_scratch_disk.py
+++ b/tests/custom_cluster/test_scratch_disk.py
@@ -25,6 +25,7 @@ import tempfile
from tests.common.custom_cluster_test_suite import CustomClusterTestSuite
from tests.verifiers.metric_verifier import MetricVerifier
+from tests.common.skip import SkipIf
class TestScratchDir(CustomClusterTestSuite):
@classmethod
@@ -169,7 +170,7 @@ class TestScratchDir(CustomClusterTestSuite):
def test_write_error_failover(self, vector):
""" Test that we can fail-over to writable directories if other directories
have permissions changed or are removed after impalad startup."""
- dirs = self.generate_dirs(3);
+ dirs = self.generate_dirs(3)
self._start_impala_cluster([
'--impalad_args=-logbuflevel=-1 -scratch_dirs={0}'.format(','.join(dirs)),
'--impalad_args=--allow_multiple_scratch_dirs_per_device=true',
@@ -214,8 +215,8 @@ class TestScratchDir(CustomClusterTestSuite):
metrics2 = self.get_metric('tmp-file-mgr.scratch-space-bytes-used.dir-2')
metrics3 = self.get_metric('tmp-file-mgr.scratch-space-bytes-used.dir-3')
metrics4 = self.get_metric('tmp-file-mgr.scratch-space-bytes-used.dir-4')
- assert (metrics0 > 0 and metrics1 > 0 and metrics2 > 0 and metrics3 > 0 and
- metrics4 > 0)
+ assert (metrics0 > 0 and metrics1 > 0 and metrics2 > 0 and metrics3 > 0
+ and metrics4 > 0)
results = client.fetch(self.spill_query, handle)
assert results.success
client.close_query(handle)
@@ -249,9 +250,186 @@ class TestScratchDir(CustomClusterTestSuite):
metrics3 = self.get_metric('tmp-file-mgr.scratch-space-bytes-used.dir-3')
metrics4 = self.get_metric('tmp-file-mgr.scratch-space-bytes-used.dir-4')
# dir1 and dir3 have highest priority and will be used as scratch disk.
- assert (metrics1 > 0 and metrics3 > 0 and metrics0 == 0 and metrics2 == 0 and
- metrics4 == 0)
+ assert (metrics1 > 0 and metrics3 > 0 and metrics0 == 0 and metrics2 == 0
+ and metrics4 == 0)
results = client.fetch(self.spill_query, handle)
assert results.success
client.close_query(handle)
client.close()
+
+ @pytest.mark.execute_serially
+ @SkipIf.not_hdfs
+ def test_scratch_dirs_remote_spill(self, vector):
+ # Test one remote directory with one its local buffer directory.
+ normal_dirs = self.generate_dirs(1)
+ # Use local hdfs for testing. Could be changed to S3.
+ normal_dirs.append('hdfs://localhost')
+ self._start_impala_cluster([
+ '--impalad_args=-logbuflevel=-1 -scratch_dirs={0}'.format(','.join(normal_dirs)),
+ '--impalad_args=--allow_multiple_scratch_dirs_per_device=true',
+ '--impalad_args=--allow_spill_to_hdfs=true'],
+ cluster_size=1,
+ expected_num_executors=1)
+ self.assert_impalad_log_contains("INFO", "Using scratch directory ",
+ expected_count=len(normal_dirs) - 1)
+ vector.get_value('exec_option')['buffer_pool_limit'] = self.buffer_pool_limit
+ impalad = self.cluster.impalads[0]
+ client = impalad.service.create_beeswax_client()
+ handle = self.execute_query_async_using_client(client, self.spill_query, vector)
+ verifier = MetricVerifier(impalad.service)
+ verifier.wait_for_metric("impala-server.num-fragments-in-flight", 2)
+ metrics0 = self.get_metric('tmp-file-mgr.scratch-space-bytes-used.dir-0')
+ # Dir0 is the remote directory.
+ assert (metrics0 > 0)
+ results = client.fetch(self.spill_query, handle)
+ assert results.success
+ client.close_query(handle)
+ client.close()
+
+ @pytest.mark.execute_serially
+ @SkipIf.not_hdfs
+ def test_scratch_dirs_mix_local_and_remote_dir_spill_local_only(self, vector):
+ '''Two local directories, the first one is always used as local buffer for
+ remote directories. Set the second directory big enough so that only spills
+ to local in the test'''
+ normal_dirs = self.generate_dirs(2)
+ normal_dirs[0] = '{0}::{1}'.format(normal_dirs[0], 1)
+ normal_dirs[1] = '{0}:100M:{1}'.format(normal_dirs[1], 0)
+ normal_dirs.append('hdfs://localhost')
+ self._start_impala_cluster([
+ '--impalad_args=-logbuflevel=-1 -scratch_dirs={0}'.format(','.join(normal_dirs)),
+ '--impalad_args=--allow_multiple_scratch_dirs_per_device=true',
+ '--impalad_args=--allow_spill_to_hdfs=true'],
+ cluster_size=1,
+ expected_num_executors=1)
+ self.assert_impalad_log_contains("INFO", "Using scratch directory ",
+ expected_count=len(normal_dirs) - 1)
+ vector.get_value('exec_option')['buffer_pool_limit'] = self.buffer_pool_limit
+ impalad = self.cluster.impalads[0]
+ client = impalad.service.create_beeswax_client()
+ handle = self.execute_query_async_using_client(client, self.spill_query, vector)
+ verifier = MetricVerifier(impalad.service)
+ verifier.wait_for_metric("impala-server.num-fragments-in-flight", 2)
+ metrics0 = self.get_metric('tmp-file-mgr.scratch-space-bytes-used.dir-0')
+ metrics1 = self.get_metric('tmp-file-mgr.scratch-space-bytes-used.dir-1')
+ # Local directory always ranks before the remote one, so dir0 is the local directory.
+ # Only spill to dir0 because it has enough space for the spilling.
+ assert (metrics0 > 0 and metrics1 == 0)
+ results = client.fetch(self.spill_query, handle)
+ assert results.success
+ client.close_query(handle)
+ client.close()
+
+ @pytest.mark.execute_serially
+ @SkipIf.not_hdfs
+ def test_scratch_dirs_mix_local_and_remote_dir_spill_both(self, vector):
+ '''Two local directories, the first one is always used as local buffer for
+ remote directories. Set the second directory small enough so that it spills
+ to both directories in the test'''
+ normal_dirs = self.generate_dirs(2)
+ normal_dirs[0] = '{0}:32MB:{1}'.format(normal_dirs[0], 0)
+ normal_dirs[1] = '{0}:4MB:{1}'.format(normal_dirs[1], 1)
+ normal_dirs.append('hdfs://localhost')
+ self._start_impala_cluster([
+ '--impalad_args=-logbuflevel=-1 -scratch_dirs={0}'.format(','.join(normal_dirs)),
+ '--impalad_args=--allow_multiple_scratch_dirs_per_device=true',
+ '--impalad_args=--allow_spill_to_hdfs=true'],
+ cluster_size=1,
+ expected_num_executors=1)
+ self.assert_impalad_log_contains("INFO", "Using scratch directory ",
+ expected_count=len(normal_dirs) - 1)
+ vector.get_value('exec_option')['buffer_pool_limit'] = self.buffer_pool_limit
+ impalad = self.cluster.impalads[0]
+ client = impalad.service.create_beeswax_client()
+ handle = self.execute_query_async_using_client(client, self.spill_query, vector)
+ verifier = MetricVerifier(impalad.service)
+ verifier.wait_for_metric("impala-server.num-fragments-in-flight", 2)
+ metrics0 = self.get_metric('tmp-file-mgr.scratch-space-bytes-used.dir-0')
+ metrics1 = self.get_metric('tmp-file-mgr.scratch-space-bytes-used.dir-1')
+ # Local directory always ranks before the remote one, so dir0 is the local directory.
+ # The query spills to both dir0 and dir1. By default the remote file is 16MB each,
+ # so the value of metrics1 should be at least one file size.
+ assert (metrics0 == 4 * 1024 * 1024 and metrics1 % (16 * 1024 * 1024) == 0)
+ results = client.fetch(self.spill_query, handle)
+ assert results.success
+ client.close_query(handle)
+ client.close()
+
+ @pytest.mark.execute_serially
+ @SkipIf.not_hdfs
+ def test_scratch_dirs_remote_spill_with_options(self, vector):
+ # One local buffer directory and one remote directory.
+ normal_dirs = self.generate_dirs(1)
+ normal_dirs[0] = '{0}:16MB:{1}'.format(normal_dirs[0], 0)
+ normal_dirs.append('hdfs://localhost')
+ self._start_impala_cluster([
+ '--impalad_args=-logbuflevel=-1 -scratch_dirs={0}'.format(','.join(normal_dirs)),
+ '--impalad_args=--allow_multiple_scratch_dirs_per_device=true',
+ '--impalad_args=--allow_spill_to_hdfs=true',
+ '--impalad_args=--remote_tmp_file_size=8M',
+ '--impalad_args=--remote_tmp_file_block_size=1m'],
+ cluster_size=1, expected_num_executors=1)
+ self.assert_impalad_log_contains("INFO", "Using scratch directory ",
+ expected_count=len(normal_dirs) - 1)
+ vector.get_value('exec_option')['buffer_pool_limit'] = self.buffer_pool_limit
+ impalad = self.cluster.impalads[0]
+ client = impalad.service.create_beeswax_client()
+ handle = self.execute_query_async_using_client(client, self.spill_query, vector)
+ verifier = MetricVerifier(impalad.service)
+ verifier.wait_for_metric("impala-server.num-fragments-in-flight", 2)
+ metrics0 = self.get_metric('tmp-file-mgr.scratch-space-bytes-used.dir-0')
+ # The query spills to the remote directories and creates remote files,
+ # so that the size is bigger than 0, and be integer times of remote file size.
+ assert (metrics0 > 0 and metrics0 % (8 * 1024 * 1024) == 0)
+ results = client.fetch(self.spill_query, handle)
+ assert results.success
+ client.close_query(handle)
+ client.close()
+
+ @pytest.mark.execute_serially
+ @SkipIf.not_hdfs
+ def test_scratch_dirs_remote_spill_concurrent(self, vector):
+ '''Concurrently execute multiple queries that trigger the spilling to the remote
+ directory to test if there is a deadlock issue.'''
+ normal_dirs = self.generate_dirs(1)
+ normal_dirs[0] = '{0}:16MB:{1}'.format(normal_dirs[0], 0)
+ normal_dirs.append('hdfs://localhost')
+ num = 5
+ self._start_impala_cluster([
+ '--impalad_args=-logbuflevel=-1 -scratch_dirs={0}'.format(','.join(normal_dirs)),
+ '--impalad_args=--allow_multiple_scratch_dirs_per_device=true',
+ '--impalad_args=--allow_spill_to_hdfs=true',
+ '--impalad_args=--remote_tmp_file_size=8M',
+ '--impalad_args=--remote_tmp_file_block_size=1m'],
+ cluster_size=num, num_coordinators=num, expected_num_executors=num)
+ self.assert_impalad_log_contains("INFO", "Using scratch directory ",
+ expected_count=len(normal_dirs) - 1)
+ vector.get_value('exec_option')['buffer_pool_limit'] = self.buffer_pool_limit
+ impalad_name = 'impalad'
+ client_name = 'client'
+ handle_name = 'handle'
+ for i in range(num):
+ impalad = self.cluster.impalads[i - 1]
+ locals()[client_name + str(i)] = impalad.service.create_beeswax_client()
+
+ for i in range(num):
+ client = locals()[client_name + str(i)]
+ locals()[handle_name + str(i)] = self.execute_query_async_using_client(client,
+ self.spill_query, vector)
+
+ total_size = 0
+ for i in range(num):
+ impalad = self.cluster.impalads[i - 1]
+ client = locals()[client_name + str(i)]
+ handle = locals()[handle_name + str(i)]
+ results = client.fetch(self.spill_query, handle)
+ assert results.success
+ metrics = impalad.service.get_metric_value(
+ 'tmp-file-mgr.scratch-space-bytes-used-high-water-mark')
+ total_size = total_size + metrics
+ client.close_query(handle)
+ client.close()
+
+ # assert that we did use the scratch space and should be integer times of the
+ # remote file size.
+ assert (total_size > 0 and total_size % (8 * 1024 * 1024) == 0)