You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@impala.apache.org by ta...@apache.org on 2018/06/10 19:04:16 UTC
[6/6] impala git commit: IMPALA-6953: part 2: clean up DiskIoMgr
IMPALA-6953: part 2: clean up DiskIoMgr
Follow-on changes.
Made further mechanical changes that involved a lot of code motion:
* Move code that manipulates RequestContext internal state to
RequestContext
* Don't use protected friend pattern that doesn't work as I thought
* Remove some unused member variables and arguments
* Move PerDiskState definition into .cc
Change-Id: Ie82849652266b6660dd479689e5134273b172eb5
Reviewed-on: http://gerrit.cloudera.org:8080/10479
Reviewed-by: Impala Public Jenkins <im...@cloudera.com>
Tested-by: Impala Public Jenkins <im...@cloudera.com>
Project: http://git-wip-us.apache.org/repos/asf/impala/repo
Commit: http://git-wip-us.apache.org/repos/asf/impala/commit/b0feb2c1
Tree: http://git-wip-us.apache.org/repos/asf/impala/tree/b0feb2c1
Diff: http://git-wip-us.apache.org/repos/asf/impala/diff/b0feb2c1
Branch: refs/heads/2.x
Commit: b0feb2c1e6445b2ab100a5dcd794f909b9f89a72
Parents: 908865a
Author: Tim Armstrong <ta...@cloudera.com>
Authored: Tue May 22 12:33:40 2018 -0700
Committer: Impala Public Jenkins <im...@gerrit.cloudera.org>
Committed: Sun Jun 10 06:42:03 2018 +0000
----------------------------------------------------------------------
be/src/exec/hdfs-orc-scanner.cc | 5 +-
be/src/exec/hdfs-parquet-scanner.cc | 6 +-
be/src/exec/hdfs-parquet-scanner.h | 4 +-
be/src/exec/hdfs-scan-node-base.cc | 7 +-
be/src/exec/hdfs-scan-node.cc | 3 +-
be/src/exec/parquet-column-readers.cc | 8 +-
be/src/exec/scanner-context.cc | 8 +-
be/src/runtime/io/disk-io-mgr-stress.cc | 7 +-
be/src/runtime/io/disk-io-mgr-test.cc | 72 +++---
be/src/runtime/io/disk-io-mgr.cc | 161 +------------
be/src/runtime/io/disk-io-mgr.h | 131 +++-------
be/src/runtime/io/request-context.cc | 346 ++++++++++++++++++++++++++-
be/src/runtime/io/request-context.h | 245 +++++--------------
be/src/runtime/io/request-ranges.h | 40 ++--
be/src/runtime/io/scan-range.cc | 26 +-
be/src/runtime/tmp-file-mgr.cc | 15 +-
be/src/runtime/tmp-file-mgr.h | 4 +-
17 files changed, 525 insertions(+), 563 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/impala/blob/b0feb2c1/be/src/exec/hdfs-orc-scanner.cc
----------------------------------------------------------------------
diff --git a/be/src/exec/hdfs-orc-scanner.cc b/be/src/exec/hdfs-orc-scanner.cc
index b755e7a..a98e16a 100644
--- a/be/src/exec/hdfs-orc-scanner.cc
+++ b/be/src/exec/hdfs-orc-scanner.cc
@@ -22,6 +22,7 @@
#include "exec/scanner-context.inline.h"
#include "exprs/expr.h"
#include "runtime/exec-env.h"
+#include "runtime/io/request-context.h"
#include "runtime/runtime-filter.inline.h"
#include "runtime/tuple-row.h"
#include "util/decompress.h"
@@ -115,8 +116,8 @@ void HdfsOrcScanner::ScanRangeInputStream::read(void* buf, uint64_t length,
{
SCOPED_TIMER(scanner_->state_->total_storage_wait_timer());
bool needs_buffers;
- status = ExecEnv::GetInstance()->disk_io_mgr()->StartScanRange(
- scanner_->scan_node_->reader_context(), range, &needs_buffers);
+ status =
+ scanner_->scan_node_->reader_context()->StartScanRange(range, &needs_buffers);
DCHECK(!status.ok() || !needs_buffers) << "Already provided a buffer";
if (status.ok()) status = range->GetNext(&io_buffer);
}
http://git-wip-us.apache.org/repos/asf/impala/blob/b0feb2c1/be/src/exec/hdfs-parquet-scanner.cc
----------------------------------------------------------------------
diff --git a/be/src/exec/hdfs-parquet-scanner.cc b/be/src/exec/hdfs-parquet-scanner.cc
index 8cec3d2..ef8ce52 100644
--- a/be/src/exec/hdfs-parquet-scanner.cc
+++ b/be/src/exec/hdfs-parquet-scanner.cc
@@ -30,6 +30,7 @@
#include "runtime/collection-value-builder.h"
#include "runtime/exec-env.h"
#include "runtime/io/disk-io-mgr.h"
+#include "runtime/io/request-context.h"
#include "runtime/runtime-state.h"
#include "runtime/runtime-filter.inline.h"
#include "rpc/thrift-util.h"
@@ -1235,7 +1236,6 @@ Status HdfsParquetScanner::ProcessFooter() {
return scan_node_->mem_tracker()->MemLimitExceeded(state_, details, metadata_size);
}
metadata_ptr = metadata_buffer.buffer();
- DiskIoMgr* io_mgr = scan_node_->runtime_state()->io_mgr();
// Read the header into the metadata buffer.
ScanRange* metadata_range = scan_node_->AllocateScanRange(
@@ -1245,8 +1245,8 @@ Status HdfsParquetScanner::ProcessFooter() {
unique_ptr<BufferDescriptor> io_buffer;
bool needs_buffers;
- RETURN_IF_ERROR(io_mgr->StartScanRange(
- scan_node_->reader_context(), metadata_range, &needs_buffers));
+ RETURN_IF_ERROR(
+ scan_node_->reader_context()->StartScanRange(metadata_range, &needs_buffers));
DCHECK(!needs_buffers) << "Already provided a buffer";
RETURN_IF_ERROR(metadata_range->GetNext(&io_buffer));
DCHECK_EQ(io_buffer->buffer(), metadata_buffer.buffer());
http://git-wip-us.apache.org/repos/asf/impala/blob/b0feb2c1/be/src/exec/hdfs-parquet-scanner.h
----------------------------------------------------------------------
diff --git a/be/src/exec/hdfs-parquet-scanner.h b/be/src/exec/hdfs-parquet-scanner.h
index 82e761a..7a88831 100644
--- a/be/src/exec/hdfs-parquet-scanner.h
+++ b/be/src/exec/hdfs-parquet-scanner.h
@@ -68,8 +68,8 @@ class BoolColumnReader;
/// the split size, the mid point guarantees that we have at least 50% of the row group in
/// the current split. ProcessSplit() then computes the column ranges for these row groups
/// and submits them to the IoMgr for immediate scheduling (so they don't surface in
-/// DiskIoMgr::GetNextUnstartedRange()). Scheduling them immediately also guarantees they
-/// are all read at once.
+/// RequestContext::GetNextUnstartedRange()). Scheduling them immediately also guarantees
+/// they are all read at once.
///
/// Like the other scanners, each parquet scanner object is one to one with a
/// ScannerContext. Unlike the other scanners though, the context will have multiple
http://git-wip-us.apache.org/repos/asf/impala/blob/b0feb2c1/be/src/exec/hdfs-scan-node-base.cc
----------------------------------------------------------------------
diff --git a/be/src/exec/hdfs-scan-node-base.cc b/be/src/exec/hdfs-scan-node-base.cc
index 25be748..6719d7b 100644
--- a/be/src/exec/hdfs-scan-node-base.cc
+++ b/be/src/exec/hdfs-scan-node-base.cc
@@ -515,8 +515,7 @@ Status HdfsScanNodeBase::StartNextScanRange(int64_t* reservation,
ScanRange** scan_range) {
DiskIoMgr* io_mgr = ExecEnv::GetInstance()->disk_io_mgr();
bool needs_buffers;
- RETURN_IF_ERROR(io_mgr->GetNextUnstartedRange(
- reader_context_.get(), scan_range, &needs_buffers));
+ RETURN_IF_ERROR(reader_context_->GetNextUnstartedRange(scan_range, &needs_buffers));
if (*scan_range == nullptr) return Status::OK();
if (needs_buffers) {
// Check if we should increase our reservation to read this range more efficiently.
@@ -529,7 +528,7 @@ Status HdfsScanNodeBase::StartNextScanRange(int64_t* reservation,
initial_range_ideal_reservation_stats_->UpdateCounter(ideal_scan_range_reservation);
initial_range_actual_reservation_stats_->UpdateCounter(*reservation);
RETURN_IF_ERROR(io_mgr->AllocateBuffersForRange(
- reader_context_.get(), &buffer_pool_client_, *scan_range, *reservation));
+ &buffer_pool_client_, *scan_range, *reservation));
}
return Status::OK();
}
@@ -596,7 +595,7 @@ Status HdfsScanNodeBase::AddDiskIoRanges(
const vector<ScanRange*>& ranges, int num_files_queued) {
DCHECK(!progress_.done()) << "Don't call AddScanRanges() after all ranges finished.";
DCHECK_GT(ranges.size(), 0);
- RETURN_IF_ERROR(runtime_state_->io_mgr()->AddScanRanges(reader_context_.get(), ranges));
+ RETURN_IF_ERROR(reader_context_->AddScanRanges(ranges));
num_unqueued_files_.Add(-num_files_queued);
DCHECK_GE(num_unqueued_files_.Load(), 0);
return Status::OK();
http://git-wip-us.apache.org/repos/asf/impala/blob/b0feb2c1/be/src/exec/hdfs-scan-node.cc
----------------------------------------------------------------------
diff --git a/be/src/exec/hdfs-scan-node.cc b/be/src/exec/hdfs-scan-node.cc
index 711c486..7b911de 100644
--- a/be/src/exec/hdfs-scan-node.cc
+++ b/be/src/exec/hdfs-scan-node.cc
@@ -244,8 +244,7 @@ void HdfsScanNode::AddMaterializedRowBatch(unique_ptr<RowBatch> row_batch) {
Status HdfsScanNode::AddDiskIoRanges(const vector<ScanRange*>& ranges,
int num_files_queued) {
- RETURN_IF_ERROR(
- runtime_state_->io_mgr()->AddScanRanges(reader_context_.get(), ranges));
+ RETURN_IF_ERROR(reader_context_->AddScanRanges(ranges));
num_unqueued_files_.Add(-num_files_queued);
DCHECK_GE(num_unqueued_files_.Load(), 0);
if (!ranges.empty()) ThreadTokenAvailableCb(runtime_state_->resource_pool());
http://git-wip-us.apache.org/repos/asf/impala/blob/b0feb2c1/be/src/exec/parquet-column-readers.cc
----------------------------------------------------------------------
diff --git a/be/src/exec/parquet-column-readers.cc b/be/src/exec/parquet-column-readers.cc
index f9e67dd..7d624ed 100644
--- a/be/src/exec/parquet-column-readers.cc
+++ b/be/src/exec/parquet-column-readers.cc
@@ -32,6 +32,7 @@
#include "runtime/collection-value-builder.h"
#include "runtime/exec-env.h"
#include "runtime/io/disk-io-mgr.h"
+#include "runtime/io/request-context.h"
#include "runtime/tuple-row.h"
#include "runtime/tuple.h"
#include "runtime/runtime-state.h"
@@ -1003,12 +1004,11 @@ Status BaseScalarColumnReader::StartScan() {
ScannerContext* context = parent_->context_;
DCHECK_GT(io_reservation_, 0);
bool needs_buffers;
- RETURN_IF_ERROR(io_mgr->StartScanRange(
- parent_->scan_node_->reader_context(), scan_range_, &needs_buffers));
+ RETURN_IF_ERROR(parent_->scan_node_->reader_context()->StartScanRange(
+ scan_range_, &needs_buffers));
if (needs_buffers) {
RETURN_IF_ERROR(io_mgr->AllocateBuffersForRange(
- parent_->scan_node_->reader_context(), context->bp_client(),
- scan_range_, io_reservation_));
+ context->bp_client(), scan_range_, io_reservation_));
}
stream_ = parent_->context_->AddStream(scan_range_, io_reservation_);
DCHECK(stream_ != nullptr);
http://git-wip-us.apache.org/repos/asf/impala/blob/b0feb2c1/be/src/exec/scanner-context.cc
----------------------------------------------------------------------
diff --git a/be/src/exec/scanner-context.cc b/be/src/exec/scanner-context.cc
index 280eac9..75aacee 100644
--- a/be/src/exec/scanner-context.cc
+++ b/be/src/exec/scanner-context.cc
@@ -22,6 +22,7 @@
#include "exec/hdfs-scan-node-base.h"
#include "exec/hdfs-scan-node.h"
#include "runtime/io/disk-io-mgr.h"
+#include "runtime/io/request-context.h"
#include "runtime/exec-env.h"
#include "runtime/mem-pool.h"
#include "runtime/row-batch.h"
@@ -155,8 +156,8 @@ Status ScannerContext::Stream::GetNextBuffer(int64_t read_past_size) {
scan_range_->fs(), filename(), read_past_buffer_size, offset, partition_id,
scan_range_->disk_id(), false, BufferOpts::Uncached());
bool needs_buffers;
- RETURN_IF_ERROR(io_mgr->StartScanRange(
- parent_->scan_node_->reader_context(), range, &needs_buffers));
+ RETURN_IF_ERROR(
+ parent_->scan_node_->reader_context()->StartScanRange(range, &needs_buffers));
if (needs_buffers) {
// Allocate fresh buffers. The buffers for 'scan_range_' should be released now
// since we hit EOS.
@@ -166,8 +167,7 @@ Status ScannerContext::Stream::GetNextBuffer(int64_t read_past_size) {
reservation_, io_mgr->min_buffer_size()));
}
RETURN_IF_ERROR(io_mgr->AllocateBuffersForRange(
- parent_->scan_node_->reader_context(), parent_->bp_client_, range,
- reservation_));
+ parent_->bp_client_, range, reservation_));
}
RETURN_IF_ERROR(range->GetNext(&io_buffer_));
DCHECK(io_buffer_->eosr());
http://git-wip-us.apache.org/repos/asf/impala/blob/b0feb2c1/be/src/runtime/io/disk-io-mgr-stress.cc
----------------------------------------------------------------------
diff --git a/be/src/runtime/io/disk-io-mgr-stress.cc b/be/src/runtime/io/disk-io-mgr-stress.cc
index fd360c4..c9b6870 100644
--- a/be/src/runtime/io/disk-io-mgr-stress.cc
+++ b/be/src/runtime/io/disk-io-mgr-stress.cc
@@ -118,12 +118,11 @@ void DiskIoMgrStress::ClientThread(int client_id) {
while (!eos) {
ScanRange* range;
bool needs_buffers;
- Status status =
- io_mgr_->GetNextUnstartedRange(client->reader.get(), &range, &needs_buffers);
+ Status status = client->reader->GetNextUnstartedRange(&range, &needs_buffers);
CHECK(status.ok() || status.IsCancelled());
if (range == NULL) break;
if (needs_buffers) {
- status = io_mgr_->AllocateBuffersForRange(client->reader.get(),
+ status = io_mgr_->AllocateBuffersForRange(
&buffer_pool_clients_[client_id], range, MAX_BUFFER_BYTES_PER_SCAN_RANGE);
CHECK(status.ok()) << status.GetDetail();
}
@@ -281,6 +280,6 @@ void DiskIoMgrStress::NewClient(int i) {
<< exec_env->buffer_reservation()->DebugString();
client.reader = io_mgr_->RegisterContext();
- status = io_mgr_->AddScanRanges(client.reader.get(), client.scan_ranges);
+ status = client.reader->AddScanRanges(client.scan_ranges);
CHECK(status.ok());
}
http://git-wip-us.apache.org/repos/asf/impala/blob/b0feb2c1/be/src/runtime/io/disk-io-mgr-test.cc
----------------------------------------------------------------------
diff --git a/be/src/runtime/io/disk-io-mgr-test.cc b/be/src/runtime/io/disk-io-mgr-test.cc
index 3e8a9ae..9239658 100644
--- a/be/src/runtime/io/disk-io-mgr-test.cc
+++ b/be/src/runtime/io/disk-io-mgr-test.cc
@@ -171,10 +171,10 @@ class DiskIoMgrTest : public testing::Test {
int expected_len = -1) {
unique_ptr<BufferDescriptor> buffer;
bool needs_buffers;
- ASSERT_OK(io_mgr->StartScanRange(reader, range, &needs_buffers));
+ ASSERT_OK(reader->StartScanRange(range, &needs_buffers));
if (needs_buffers) {
ASSERT_OK(io_mgr->AllocateBuffersForRange(
- reader, client, range, io_mgr->max_buffer_size()));
+ client, range, io_mgr->max_buffer_size()));
}
ASSERT_OK(range->GetNext(&buffer));
ASSERT_TRUE(buffer != nullptr);
@@ -215,12 +215,12 @@ class DiskIoMgrTest : public testing::Test {
while (max_ranges == 0 || num_ranges < max_ranges) {
ScanRange* range;
bool needs_buffers;
- Status status = io_mgr->GetNextUnstartedRange(reader, &range, &needs_buffers);
+ Status status = reader->GetNextUnstartedRange(&range, &needs_buffers);
ASSERT_TRUE(status.ok() || status.code() == expected_status.code());
if (range == nullptr) break;
if (needs_buffers) {
ASSERT_OK(io_mgr->AllocateBuffersForRange(
- reader, client, range, io_mgr->max_buffer_size() * 3));
+ client, range, io_mgr->max_buffer_size() * 3));
}
ValidateScanRange(io_mgr, range, expected_result, expected_len, expected_status);
num_ranges_processed->Add(1);
@@ -246,8 +246,8 @@ class DiskIoMgrTest : public testing::Test {
const string& function_name, int err_no, const string& expected_error);
// Auxiliary function to ExecuteWriteFailureTest(). Handles the
- // DiskIoMgr::AddWriteRange() call.
- void AddWriteRange(DiskIoMgr* io_mgr, int num_of_writes, int32_t* data,
+ // AddWriteRange() call.
+ void AddWriteRange(int num_of_writes, int32_t* data,
const string& tmp_file, int offset, RequestContext* writer,
const string& expected_output);
@@ -309,7 +309,7 @@ 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));
- EXPECT_OK(io_mgr.AddWriteRange(writer.get(), *new_range));
+ EXPECT_OK(writer->AddWriteRange(*new_range));
cur_offset += sizeof(int32_t);
}
@@ -349,7 +349,7 @@ 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));
- EXPECT_OK(io_mgr.AddWriteRange(writer.get(), *new_range));
+ EXPECT_OK(writer->AddWriteRange(*new_range));
// Write to a bad location in a file that exists.
tmp_file = "/tmp/disk_io_mgr_test.txt";
@@ -367,7 +367,7 @@ TEST_F(DiskIoMgrTest, InvalidWrite) {
*new_range = pool_.Add(new WriteRange(tmp_file, -1, 0, callback));
(*new_range)->SetData(reinterpret_cast<uint8_t*>(data), sizeof(int32_t));
- EXPECT_OK(io_mgr.AddWriteRange(writer.get(), *new_range));
+ EXPECT_OK(writer->AddWriteRange(*new_range));
{
unique_lock<mutex> lock(written_mutex_);
@@ -434,8 +434,7 @@ void DiskIoMgrTest::ExecuteWriteFailureTest(DiskIoMgr* io_mgr, const string& fil
fs->SetWriteFaultInjection(function_name, err_no);
// DiskIoMgr takes responsibility of fs from this point.
io_mgr->SetLocalFileSystem(std::move(fs));
- AddWriteRange(io_mgr, num_of_writes, &data, file_name, 0, writer.get(),
- expected_error);
+ AddWriteRange(num_of_writes, &data, file_name, 0, writer.get(), expected_error);
{
unique_lock<mutex> lock(written_mutex_);
@@ -445,7 +444,7 @@ void DiskIoMgrTest::ExecuteWriteFailureTest(DiskIoMgr* io_mgr, const string& fil
io_mgr->UnregisterContext(writer.get());
}
-void DiskIoMgrTest::AddWriteRange(DiskIoMgr* io_mgr, int num_of_writes, int32_t* data,
+void DiskIoMgrTest::AddWriteRange(int num_of_writes, int32_t* data,
const string& file_name, int offset, RequestContext* writer,
const string& expected_output) {
WriteRange::WriteDoneCallback callback =
@@ -454,7 +453,7 @@ void DiskIoMgrTest::AddWriteRange(DiskIoMgr* io_mgr, int num_of_writes, int32_t*
Status(TErrorCode::DISK_IO_ERROR, 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));
- EXPECT_OK(io_mgr->AddWriteRange(writer, write_range));
+ EXPECT_OK(writer->AddWriteRange(write_range));
}
// Issue a number of writes, cancel the writer context and issue more writes.
@@ -505,7 +504,7 @@ TEST_F(DiskIoMgrTest, SingleWriterCancel) {
new WriteRange(tmp_file, cur_offset, num_ranges % num_disks, callback));
(*new_range)->SetData(reinterpret_cast<uint8_t*>(data), sizeof(int32_t));
cur_offset += sizeof(int32_t);
- Status add_status = io_mgr.AddWriteRange(writer.get(), *new_range);
+ Status add_status = writer->AddWriteRange(*new_range);
EXPECT_TRUE(add_status.code() == validate_status.code());
}
@@ -558,7 +557,7 @@ TEST_F(DiskIoMgrTest, SingleReader) {
int disk_id = i % num_disks;
ranges.push_back(InitRange(&tmp_pool, tmp_file, 0, len, disk_id, stat_val.st_mtime));
}
- ASSERT_OK(io_mgr.AddScanRanges(reader.get(), ranges));
+ ASSERT_OK(reader->AddScanRanges(ranges));
AtomicInt32 num_ranges_processed;
thread_group threads;
@@ -622,14 +621,14 @@ TEST_F(DiskIoMgrTest, AddScanRangeTest) {
AtomicInt32 num_ranges_processed;
// Issue first half the scan ranges.
- ASSERT_OK(io_mgr.AddScanRanges(reader.get(), ranges_first_half));
+ ASSERT_OK(reader->AddScanRanges(ranges_first_half));
// Read a couple of them
ScanRangeThread(&io_mgr, reader.get(), &read_client, data, strlen(data),
Status::OK(), 2, &num_ranges_processed);
// Issue second half
- ASSERT_OK(io_mgr.AddScanRanges(reader.get(), ranges_second_half));
+ ASSERT_OK(reader->AddScanRanges(ranges_second_half));
// Start up some threads and then cancel
thread_group threads;
@@ -693,7 +692,7 @@ TEST_F(DiskIoMgrTest, SyncReadTest) {
ranges.push_back(
InitRange(&tmp_pool, tmp_file, 0, len, disk_id, stat_val.st_mtime));
}
- ASSERT_OK(io_mgr.AddScanRanges(reader.get(), ranges));
+ ASSERT_OK(reader->AddScanRanges(ranges));
AtomicInt32 num_ranges_processed;
thread_group threads;
@@ -757,7 +756,7 @@ TEST_F(DiskIoMgrTest, SingleReaderCancel) {
ranges.push_back(
InitRange(&tmp_pool, tmp_file, 0, len, disk_id, stat_val.st_mtime));
}
- ASSERT_OK(io_mgr.AddScanRanges(reader.get(), ranges));
+ ASSERT_OK(reader->AddScanRanges(ranges));
AtomicInt32 num_ranges_processed;
int num_succesful_ranges = ranges.size() / 2;
@@ -825,20 +824,20 @@ TEST_F(DiskIoMgrTest, MemScarcity) {
for (int i = 0; i < num_ranges; ++i) {
ranges.push_back(InitRange(&pool_, tmp_file, 0, DATA_BYTES, 0, stat_val.st_mtime));
}
- ASSERT_OK(io_mgr.AddScanRanges(reader.get(), ranges));
+ ASSERT_OK(reader->AddScanRanges(ranges));
// Keep starting new ranges without returning buffers until we run out of
// reservation.
while (read_client.GetUnusedReservation() >= MIN_BUFFER_SIZE) {
ScanRange* range = nullptr;
bool needs_buffers;
- ASSERT_OK(io_mgr.GetNextUnstartedRange(reader.get(), &range, &needs_buffers));
+ ASSERT_OK(reader->GetNextUnstartedRange(&range, &needs_buffers));
if (range == nullptr) break;
ASSERT_TRUE(needs_buffers);
// Pick a random amount of memory to reserve.
int64_t max_bytes_to_alloc = uniform_int_distribution<int64_t>(MIN_BUFFER_SIZE,
min<int64_t>(read_client.GetUnusedReservation(), MAX_BUFFER_SIZE * 3))(rng_);
ASSERT_OK(io_mgr.AllocateBuffersForRange(
- reader.get(), &read_client, range, max_bytes_to_alloc));
+ &read_client, range, max_bytes_to_alloc));
// Start a thread fetching from the range. The thread will either finish the
// range or be cancelled.
threads.add_thread(new thread([&data, DATA_BYTES, range] {
@@ -910,7 +909,7 @@ TEST_F(DiskIoMgrTest, CachedReads) {
ranges.push_back(
InitRange(&pool_, tmp_file, 0, len, disk_id, stat_val.st_mtime, nullptr, true));
}
- ASSERT_OK(io_mgr.AddScanRanges(reader.get(), ranges));
+ ASSERT_OK(reader->AddScanRanges(ranges));
AtomicInt32 num_ranges_processed;
thread_group threads;
@@ -1007,7 +1006,7 @@ TEST_F(DiskIoMgrTest, MultipleReaderWriter) {
new_range->SetData(
reinterpret_cast<const uint8_t*>(data + (write_offset % strlen(data))),
1);
- status = io_mgr.AddWriteRange(contexts[context_index].get(), new_range);
+ status = contexts[context_index]->AddWriteRange(new_range);
++write_offset;
}
@@ -1093,7 +1092,7 @@ TEST_F(DiskIoMgrTest, MultipleReader) {
int disk_id = j % num_disks;
ranges.push_back(InitRange(&tmp_pool, file_names[i].c_str(), j, 1, disk_id, mtimes[i]));
}
- ASSERT_OK(io_mgr.AddScanRanges(readers[i].get(), ranges));
+ ASSERT_OK(readers[i]->AddScanRanges(ranges));
}
AtomicInt32 num_ranges_processed;
@@ -1159,10 +1158,10 @@ TEST_F(DiskIoMgrTest, PartialRead) {
ScanRange* range = InitRange(&pool_, tmp_file, 0, read_len, 0, stat_val.st_mtime);
unique_ptr<BufferDescriptor> buffer;
bool needs_buffers;
- ASSERT_OK(io_mgr.StartScanRange(reader.get(), range, &needs_buffers));
+ ASSERT_OK(reader->StartScanRange(range, &needs_buffers));
if (needs_buffers) {
ASSERT_OK(io_mgr.AllocateBuffersForRange(
- reader.get(), &read_client, range, 3 * max_buffer_size));
+ &read_client, range, 3 * max_buffer_size));
}
int64_t bytes_read = 0;
@@ -1212,10 +1211,10 @@ TEST_F(DiskIoMgrTest, ZeroLengthScanRange) {
ScanRange* range = InitRange(&pool_, tmp_file, 0, 0, 0, stat_val.st_mtime);
bool needs_buffers;
- Status status = io_mgr.StartScanRange(reader.get(), range, &needs_buffers);
+ Status status = reader->StartScanRange(range, &needs_buffers);
ASSERT_EQ(TErrorCode::DISK_IO_ERROR, status.code());
- status = io_mgr.AddScanRanges(reader.get(), vector<ScanRange*>({range}));
+ status = reader->AddScanRanges(vector<ScanRange*>({range}));
ASSERT_EQ(TErrorCode::DISK_IO_ERROR, status.code());
io_mgr.UnregisterContext(reader.get());
@@ -1248,14 +1247,13 @@ TEST_F(DiskIoMgrTest, SkipAllocateBuffers) {
}
bool needs_buffers;
// Test StartScanRange().
- ASSERT_OK(io_mgr.StartScanRange(reader.get(), ranges[0], &needs_buffers));
+ ASSERT_OK(reader->StartScanRange(ranges[0], &needs_buffers));
EXPECT_TRUE(needs_buffers);
- ASSERT_OK(io_mgr.StartScanRange(reader.get(), ranges[1], &needs_buffers));
+ ASSERT_OK(reader->StartScanRange(ranges[1], &needs_buffers));
EXPECT_TRUE(needs_buffers);
// Test AddScanRanges()/GetNextUnstartedRange().
- ASSERT_OK(
- io_mgr.AddScanRanges(reader.get(), vector<ScanRange*>({ranges[2], ranges[3]})));
+ ASSERT_OK(reader->AddScanRanges(vector<ScanRange*>({ranges[2], ranges[3]})));
// Cancel two directly, cancel the other two indirectly via the context.
ranges[0]->Cancel(Status::CANCELLED);
@@ -1295,9 +1293,9 @@ TEST_F(DiskIoMgrTest, CancelReleasesResources) {
for (int i = 0; i < 10; ++i) {
ScanRange* range = InitRange(&pool_, tmp_file, 0, len, 0, stat_val.st_mtime);
bool needs_buffers;
- ASSERT_OK(io_mgr.StartScanRange(reader.get(), range, &needs_buffers));
+ ASSERT_OK(reader->StartScanRange(range, &needs_buffers));
EXPECT_TRUE(needs_buffers);
- ASSERT_OK(io_mgr.AllocateBuffersForRange(reader.get(), &read_client, range, MAX_BUFFER_SIZE));
+ ASSERT_OK(io_mgr.AllocateBuffersForRange(&read_client, range, MAX_BUFFER_SIZE));
// Give disk I/O thread a chance to start read.
SleepForMs(1);
@@ -1331,7 +1329,7 @@ TEST_F(DiskIoMgrTest, ReadIntoClientBuffer) {
range->Reset(nullptr, tmp_file, scan_len, 0, 0, true,
BufferOpts::ReadInto(client_buffer.data(), buffer_len));
bool needs_buffers;
- ASSERT_OK(io_mgr->StartScanRange(reader.get(), range, &needs_buffers));
+ ASSERT_OK(reader->StartScanRange(range, &needs_buffers));
ASSERT_FALSE(needs_buffers);
unique_ptr<BufferDescriptor> io_buffer;
@@ -1370,7 +1368,7 @@ TEST_F(DiskIoMgrTest, ReadIntoClientBufferError) {
range->Reset(nullptr, tmp_file, SCAN_LEN, 0, 0, true,
BufferOpts::ReadInto(client_buffer.data(), SCAN_LEN));
bool needs_buffers;
- ASSERT_OK(io_mgr->StartScanRange(reader.get(), range, &needs_buffers));
+ ASSERT_OK(reader->StartScanRange(range, &needs_buffers));
ASSERT_FALSE(needs_buffers);
/// Also test the cancellation path. Run multiple iterations since it is racy whether
http://git-wip-us.apache.org/repos/asf/impala/blob/b0feb2c1/be/src/runtime/io/disk-io-mgr.cc
----------------------------------------------------------------------
diff --git a/be/src/runtime/io/disk-io-mgr.cc b/be/src/runtime/io/disk-io-mgr.cc
index ff4c8f8..e7d68c0 100644
--- a/be/src/runtime/io/disk-io-mgr.cc
+++ b/be/src/runtime/io/disk-io-mgr.cc
@@ -180,8 +180,6 @@ DiskIoMgr::DiskIoMgr() :
THREADS_PER_SOLID_STATE_DISK)),
max_buffer_size_(BitUtil::RoundUpToPowerOfTwo(FLAGS_read_size)),
min_buffer_size_(BitUtil::RoundDownToPowerOfTwo(FLAGS_min_buffer_size)),
- total_bytes_read_counter_(TUnit::BYTES),
- read_timer_(TUnit::TIME_NS),
file_handle_cache_(min(FLAGS_max_cached_file_handles,
FileSystemUtil::MaxNumFileHandles()),
FLAGS_num_file_handle_cache_partitions,
@@ -206,8 +204,6 @@ DiskIoMgr::DiskIoMgr(int num_local_disks, int threads_per_rotational_disk,
num_io_threads_per_solid_state_disk_(threads_per_solid_state_disk),
max_buffer_size_(BitUtil::RoundUpToPowerOfTwo(max_buffer_size)),
min_buffer_size_(BitUtil::RoundDownToPowerOfTwo(min_buffer_size)),
- total_bytes_read_counter_(TUnit::BYTES),
- read_timer_(TUnit::TIME_NS),
file_handle_cache_(min(FLAGS_max_cached_file_handles,
FileSystemUtil::MaxNumFileHandles()),
FLAGS_num_file_handle_cache_partitions,
@@ -275,7 +271,7 @@ Status DiskIoMgr::Init() {
}
unique_ptr<RequestContext> DiskIoMgr::RegisterContext() {
- return unique_ptr<RequestContext>(new RequestContext(this, num_total_disks()));
+ return unique_ptr<RequestContext>(new RequestContext(this, disk_queues_));
}
void DiskIoMgr::UnregisterContext(RequestContext* reader) {
@@ -299,129 +295,7 @@ Status DiskIoMgr::ValidateScanRange(ScanRange* range) {
return Status::OK();
}
-Status DiskIoMgr::AddScanRanges(
- RequestContext* reader, const vector<ScanRange*>& ranges) {
- DCHECK_GT(ranges.size(), 0);
- // Validate and initialize all ranges
- for (int i = 0; i < ranges.size(); ++i) {
- RETURN_IF_ERROR(ValidateScanRange(ranges[i]));
- ranges[i]->InitInternal(this, reader);
- }
-
- unique_lock<mutex> reader_lock(reader->lock_);
- DCHECK(reader->Validate()) << endl << reader->DebugString();
-
- if (reader->state_ == RequestContext::Cancelled) return Status::CANCELLED;
-
- // Add each range to the queue of the disk the range is on
- for (ScanRange* range : ranges) {
- // Don't add empty ranges.
- DCHECK_NE(range->len(), 0);
- reader->AddActiveScanRangeLocked(reader_lock, range);
- if (range->try_cache()) {
- reader->cached_ranges_.Enqueue(range);
- } else {
- reader->AddRangeToDisk(reader_lock, range, ScheduleMode::UPON_GETNEXT);
- }
- }
- DCHECK(reader->Validate()) << endl << reader->DebugString();
- return Status::OK();
-}
-
-Status DiskIoMgr::StartScanRange(RequestContext* reader, ScanRange* range,
- bool* needs_buffers) {
- RETURN_IF_ERROR(ValidateScanRange(range));
- range->InitInternal(this, reader);
-
- unique_lock<mutex> reader_lock(reader->lock_);
- DCHECK(reader->Validate()) << endl << reader->DebugString();
- if (reader->state_ == RequestContext::Cancelled) return Status::CANCELLED;
-
- DCHECK_NE(range->len(), 0);
- if (range->try_cache()) {
- bool cached_read_succeeded;
- RETURN_IF_ERROR(range->ReadFromCache(reader_lock, &cached_read_succeeded));
- if (cached_read_succeeded) {
- DCHECK(reader->Validate()) << endl << reader->DebugString();
- *needs_buffers = false;
- return Status::OK();
- }
- // Cached read failed, fall back to normal read path.
- }
- // If we don't have a buffer yet, the caller must allocate buffers for the range.
- *needs_buffers =
- range->external_buffer_tag() == ScanRange::ExternalBufferTag::NO_BUFFER;
- if (*needs_buffers) range->SetBlockedOnBuffer();
- reader->AddActiveScanRangeLocked(reader_lock, range);
- reader->AddRangeToDisk(reader_lock, range,
- *needs_buffers ? ScheduleMode::BY_CALLER : ScheduleMode::IMMEDIATELY);
- DCHECK(reader->Validate()) << endl << reader->DebugString();
- return Status::OK();
-}
-
-// This function returns the next scan range the reader should work on, checking
-// for eos and error cases. If there isn't already a cached scan range or a scan
-// range prepared by the disk threads, the caller waits on the disk threads.
-Status DiskIoMgr::GetNextUnstartedRange(RequestContext* reader, ScanRange** range,
- bool* needs_buffers) {
- DCHECK(reader != nullptr);
- DCHECK(range != nullptr);
- *range = nullptr;
- *needs_buffers = false;
-
- unique_lock<mutex> reader_lock(reader->lock_);
- DCHECK(reader->Validate()) << endl << reader->DebugString();
- while (true) {
- if (reader->state_ == RequestContext::Cancelled) return Status::CANCELLED;
-
- if (reader->num_unstarted_scan_ranges_.Load() == 0 &&
- reader->ready_to_start_ranges_.empty() && reader->cached_ranges_.empty()) {
- // All ranges are done, just return.
- return Status::OK();
- }
-
- if (!reader->cached_ranges_.empty()) {
- // We have a cached range.
- *range = reader->cached_ranges_.Dequeue();
- DCHECK((*range)->try_cache());
- bool cached_read_succeeded;
- RETURN_IF_ERROR((*range)->ReadFromCache(reader_lock, &cached_read_succeeded));
- if (cached_read_succeeded) return Status::OK();
-
- // This range ended up not being cached. Loop again and pick up a new range.
- reader->AddRangeToDisk(reader_lock, *range, ScheduleMode::UPON_GETNEXT);
- DCHECK(reader->Validate()) << endl << reader->DebugString();
- *range = nullptr;
- continue;
- }
-
- if (reader->ready_to_start_ranges_.empty()) {
- reader->ready_to_start_ranges_cv_.Wait(reader_lock);
- } else {
- *range = reader->ready_to_start_ranges_.Dequeue();
- DCHECK(*range != nullptr);
- int disk_id = (*range)->disk_id();
- DCHECK_EQ(*range, reader->disk_states_[disk_id].next_scan_range_to_start());
- // Set this to nullptr, the next time this disk runs for this reader, it will
- // get another range ready.
- reader->disk_states_[disk_id].set_next_scan_range_to_start(nullptr);
- ScanRange::ExternalBufferTag buffer_tag = (*range)->external_buffer_tag();
- if (buffer_tag == ScanRange::ExternalBufferTag::NO_BUFFER) {
- // We can't schedule this range until the client gives us buffers. The context
- // must be rescheduled regardless to ensure that 'next_scan_range_to_start' is
- // refilled.
- reader->disk_states_[disk_id].ScheduleContext(reader_lock, reader, disk_id);
- (*range)->SetBlockedOnBuffer();
- *needs_buffers = true;
- } else {
- reader->ScheduleScanRange(reader_lock, *range);
- }
- return Status::OK();
- }
- }
-}
-
-Status DiskIoMgr::AllocateBuffersForRange(RequestContext* reader,
+Status DiskIoMgr::AllocateBuffersForRange(
BufferPool::ClientHandle* bp_client, ScanRange* range, int64_t max_bytes) {
DCHECK_GE(max_bytes, min_buffer_size_);
DCHECK(range->external_buffer_tag() == ScanRange::ExternalBufferTag::NO_BUFFER)
@@ -434,8 +308,7 @@ Status DiskIoMgr::AllocateBuffersForRange(RequestContext* reader,
BufferPool::BufferHandle handle;
status = bp->AllocateBuffer(bp_client, buffer_size, &handle);
if (!status.ok()) goto error;
- buffers.emplace_back(new BufferDescriptor(
- this, reader, range, bp_client, move(handle)));
+ buffers.emplace_back(new BufferDescriptor(range, bp_client, move(handle)));
}
range->AddUnusedBuffers(move(buffers), false);
return Status::OK();
@@ -524,25 +397,6 @@ RequestRange* DiskQueue::GetNextRequestRange(RequestContext** request_context) {
return nullptr;
}
-void DiskIoMgr::HandleWriteFinished(
- RequestContext* writer, 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();
-
- // 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);
- {
- unique_lock<mutex> writer_lock(writer->lock_);
- DCHECK(writer->Validate()) << endl << writer->DebugString();
- RequestContext::PerDiskState& state = writer->disk_states_[disk_id];
- state.DecrementDiskThread(writer_lock, writer);
- --state.num_remaining_ranges();
- }
-}
-
void DiskQueue::DiskThreadLoop(DiskIoMgr* io_mgr) {
// The thread waits until there is work or the queue is shut down. If there is work,
// performs the read or write requested. Locks are not taken when reading from or
@@ -579,7 +433,7 @@ void DiskIoMgr::Write(RequestContext* writer_context, WriteRange* write_range) {
if (ret_status.ok() && !close_status.ok()) ret_status = close_status;
end:
- HandleWriteFinished(writer_context, write_range, ret_status);
+ writer_context->WriteDone(write_range, ret_status);
}
Status DiskIoMgr::WriteRangeHelper(FILE* file_handle, WriteRange* write_range) {
@@ -598,13 +452,6 @@ Status DiskIoMgr::WriteRangeHelper(FILE* file_handle, WriteRange* write_range) {
return Status::OK();
}
-Status DiskIoMgr::AddWriteRange(RequestContext* writer, WriteRange* write_range) {
- unique_lock<mutex> writer_lock(writer->lock_);
- if (writer->state_ == RequestContext::Cancelled) return Status::CANCELLED;
- writer->AddRangeToDisk(writer_lock, write_range, ScheduleMode::IMMEDIATELY);
- return Status::OK();
-}
-
int DiskIoMgr::AssignQueue(const char* file, int disk_id, bool expected_local) {
// If it's a remote range, check for an appropriate remote disk queue.
if (!expected_local) {
http://git-wip-us.apache.org/repos/asf/impala/blob/b0feb2c1/be/src/runtime/io/disk-io-mgr.h
----------------------------------------------------------------------
diff --git a/be/src/runtime/io/disk-io-mgr.h b/be/src/runtime/io/disk-io-mgr.h
index 586b339..35d11b9 100644
--- a/be/src/runtime/io/disk-io-mgr.h
+++ b/be/src/runtime/io/disk-io-mgr.h
@@ -63,18 +63,18 @@ class DiskQueue;
/// to read based on disk activity and begins reading and queuing buffers for that range.
///
/// For Writers:
-/// Data is written via AddWriteRange(). This is non-blocking and adds a WriteRange to a
-/// per-disk queue. After the write is complete, a callback in WriteRange is invoked.
-/// No memory is allocated within IoMgr for writes and no copies are made. It is the
-/// responsibility of the client to ensure that the data to be written is valid and that
-/// the file to be written to exists until the callback is invoked.
+/// Data is written via RequestContext::AddWriteRange(). This is non-blocking and adds
+/// a WriteRange to a per-disk queue. After the write is complete, a callback in
+/// WriteRange is invoked. No memory is allocated within IoMgr for writes and no copies
+/// 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.
///
/// There are several key methods for scanning data with the IoMgr.
-/// 1. StartScanRange(): adds range to the IoMgr to start immediately.
-/// 2. AddScanRanges(): adds ranges to the IoMgr that the reader wants to scan, but does
-/// not start them until GetNextUnstartedRange() is called.
-/// 3. GetNextUnstartedRange(): returns to the caller the next scan range it should
-/// process.
+/// 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.
+/// 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
/// data is available.
///
@@ -102,14 +102,14 @@ class DiskQueue;
/// ranges of different clients.
///
/// IoMgr clients may want to work on multiple scan ranges at a time to maximize CPU and
-/// I/O utilization. Clients can call GetNextUnstartedRange() to start as many concurrent
-/// scan ranges as required, e.g. from each parallel scanner thread. Once a scan range has
-/// been returned via GetNextUnstartedRange(), the caller must allocate any memory needed
-/// for buffering reads, after which the IoMgr wil start to fill the buffers with data
-/// while the caller concurrently consumes and processes the data. For example, the logic
-/// in a scanner thread might look like:
+/// I/O utilization. Clients can call RequestContext::GetNextUnstartedRange() to start as
+/// many concurrent scan ranges as required, e.g. from each parallel scanner thread. Once
+/// a scan range has been returned via GetNextUnstartedRange(), the caller must allocate
+/// any memory needed for buffering reads, after which the IoMgr wil start to fill the
+/// buffers with data while the caller concurrently consumes and processes the data. For
+/// example, the logic in a scanner thread might look like:
/// while (more_ranges)
-/// range = GetNextUnstartedRange()
+/// range = context->GetNextUnstartedRange()
/// while (!range.eosr)
/// buffer = range.GetNext()
///
@@ -247,49 +247,9 @@ class DiskIoMgr : public CacheLineAligned {
/// up.
void UnregisterContext(RequestContext* context);
- /// Adds the scan ranges to reader's queues, but does not start scheduling it. The range
- /// can be scheduled by a thread calling GetNextUnstartedRange(). This call is
- /// non-blocking. The caller must not deallocate the scan range pointers before
- /// UnregisterContext(). 'ranges' must not be empty.
- Status AddScanRanges(
- RequestContext* reader, const std::vector<ScanRange*>& ranges) WARN_UNUSED_RESULT;
-
- /// Adds the scan range to the queues, as with AddScanRanges(), but immediately
- /// start scheduling the scan range. This can be used to do synchronous reads as well
- /// as schedule dependent ranges, e.g. for columnar formats. This call is non-blocking.
- /// The caller must not deallocate the scan range pointers before UnregisterContext().
- ///
- /// If this returns true in '*needs_buffers', the caller must then call
- /// AllocateBuffersForRange() to add buffers for the data to be read into before the
- /// range can be scheduled. Otherwise, the range is scheduled and the IoMgr will
- /// asynchronously read the data for the range and the caller can call
- /// ScanRange::GetNext() to read the data.
- Status StartScanRange(
- RequestContext* reader, ScanRange* range, bool* needs_buffers) WARN_UNUSED_RESULT;
-
- /// Add a WriteRange for the writer. This is non-blocking and schedules the context
- /// on the IoMgr disk queue. Does not create any files.
- Status AddWriteRange(
- RequestContext* writer, WriteRange* write_range) WARN_UNUSED_RESULT;
-
- /// Tries to get an unstarted scan range that was added to 'reader' with
- /// AddScanRanges(). On success, returns OK and returns the range in '*range'.
- /// If 'reader' was cancelled, returns CANCELLED. If another error is encountered,
- /// an error status is returned. Otherwise, if error or cancellation wasn't encountered
- /// and there are no unstarted ranges for 'reader', returns OK and sets '*range' to
- /// nullptr.
- ///
- /// If '*needs_buffers' is returned as true, the caller must call
- /// AllocateBuffersForRange() to add buffers for the data to be read into before the
- /// range can be scheduled. Otherwise, the range is scheduled and the IoMgr will
- /// asynchronously read the data for the range and the caller can call
- /// ScanRange::GetNext() to read the data.
- Status GetNextUnstartedRange(RequestContext* reader, ScanRange** range,
- bool* needs_buffers) WARN_UNUSED_RESULT;
-
/// Allocates up to 'max_bytes' buffers to read the data from 'range' into and schedules
- /// the range. Called after StartScanRange() or GetNextUnstartedRange() returns
- /// *needs_buffers=true.
+ /// the range. Called after StartScanRange() or reader->GetNextUnstartedRange()
+ /// returns *needs_buffers=true.
///
/// The buffer sizes are chosen based on range->len(). 'max_bytes' must be >=
/// min_read_buffer_size() so that at least one buffer can be allocated. The caller
@@ -299,7 +259,7 @@ class DiskIoMgr : public CacheLineAligned {
/// Setting 'max_bytes' to IDEAL_MAX_SIZED_BUFFERS_PER_SCAN_RANGE * max_buffer_size()
/// will typically maximize I/O throughput. See the "Buffer Management" section of
/// the class comment for explanation.
- Status AllocateBuffersForRange(RequestContext* reader,
+ Status AllocateBuffersForRange(
BufferPool::ClientHandle* bp_client, ScanRange* range, int64_t max_bytes);
/// Determine which disk queue this file should be assigned to. Returns an index into
@@ -386,33 +346,29 @@ class DiskIoMgr : public CacheLineAligned {
/// See "Buffer Management" in the class comment for explanation.
static const int64_t IDEAL_MAX_SIZED_BUFFERS_PER_SCAN_RANGE = 3;
- protected:
- // Protected methods are used by other classes in io::.
+ /// Validates that range is correctly initialized. Return an error status if there
+ /// is something invalid about the scan range.
+ Status ValidateScanRange(ScanRange* range) WARN_UNUSED_RESULT;
+
+ private:
+ DISALLOW_COPY_AND_ASSIGN(DiskIoMgr);
+ friend class DiskIoMgrTest_Buffers_Test;
+ friend class DiskIoMgrTest_BufferSizeSelection_Test;
+ friend class DiskIoMgrTest_VerifyNumThreadsParameter_Test;
+
+ /////////////////////////////////////////
+ /// BEGIN: private members that are accessed by other io:: classes
friend class DiskQueue;
friend class ScanRange;
- friend class RequestContext;
- /// Write the specified range to disk and calls HandleWriteFinished() when done.
+ /// Write the specified range to disk and calls writer_context->WriteDone() when done.
/// Responsible for opening and closing the file that is written.
void Write(RequestContext* writer_context, WriteRange* write_range);
- DiskQueue* GetDiskQueue(int disk_id) {
- DCHECK_GE(disk_id, 0);
- DCHECK_LT(disk_id, disk_queues_.size());
- return disk_queues_[disk_id];
- }
-
- RuntimeProfile::Counter* total_bytes_read_counter() {
- return &total_bytes_read_counter_;
- }
- RuntimeProfile::Counter* read_timer() { return &read_timer_; }
struct hadoopRzOptions* cached_read_options() { return cached_read_options_; }
- private:
- DISALLOW_COPY_AND_ASSIGN(DiskIoMgr);
- friend class DiskIoMgrTest_Buffers_Test;
- friend class DiskIoMgrTest_BufferSizeSelection_Test;
- friend class DiskIoMgrTest_VerifyNumThreadsParameter_Test;
+ /// END: private members that are accessed by other io:: classes
+ /////////////////////////////////////////
// Handles the low level I/O functionality.
std::unique_ptr<LocalFileSystem> local_file_system_;
@@ -437,12 +393,6 @@ class DiskIoMgr : public CacheLineAligned {
/// Options object for cached hdfs reads. Set on startup and never modified.
struct hadoopRzOptions* cached_read_options_ = nullptr;
- /// Total bytes read by the IoMgr.
- RuntimeProfile::Counter total_bytes_read_counter_;
-
- /// Total time spent in hdfs reading
- RuntimeProfile::Counter read_timer_;
-
/// Per disk queues. This is static and created once at Init() time. One queue is
/// allocated for each local disk on the system and for each remote filesystem type.
/// It is indexed by disk id.
@@ -458,17 +408,6 @@ class DiskIoMgr : public CacheLineAligned {
// handles are closed.
FileHandleCache file_handle_cache_;
- /// 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.
- /// The write_status does not affect the writer->status_. That is, an write error does
- /// not cancel the writer context - that decision is left to the callback handler.
- void HandleWriteFinished(
- RequestContext* writer, WriteRange* write_range, const Status& write_status);
-
- /// Validates that range is correctly initialized
- Status ValidateScanRange(ScanRange* range) WARN_UNUSED_RESULT;
-
/// Helper method to write a range using the specified FILE handle. Returns Status:OK
/// if the write succeeded, or a RUNTIME_ERROR with an appropriate message otherwise.
/// Does not open or close the file that is written.
http://git-wip-us.apache.org/repos/asf/impala/blob/b0feb2c1/be/src/runtime/io/request-context.cc
----------------------------------------------------------------------
diff --git a/be/src/runtime/io/request-context.cc b/be/src/runtime/io/request-context.cc
index 6c56b8c..8c2b740 100644
--- a/be/src/runtime/io/request-context.cc
+++ b/be/src/runtime/io/request-context.cc
@@ -24,6 +24,167 @@
using namespace impala;
using namespace impala::io;
+/// Struct containing state per disk. See comments in the disk read loop on how
+/// they are used.
+class RequestContext::PerDiskState {
+ public:
+ /// Set reference to global disk queue corresponding to this state. Must be called
+ /// immediately after construction.
+ void set_disk_queue(DiskQueue* queue) { disk_queue_ = queue; }
+
+ bool done() const { return done_; }
+ void set_done(bool b) { done_ = b; }
+
+ int num_remaining_ranges() const { return num_remaining_ranges_; }
+ int& num_remaining_ranges() { return num_remaining_ranges_; }
+
+ ScanRange* next_scan_range_to_start() { return next_scan_range_to_start_; }
+ void set_next_scan_range_to_start(ScanRange* range) {
+ next_scan_range_to_start_ = range;
+ }
+
+ bool is_on_queue() const { return is_on_queue_.Load() != 0; }
+
+ int num_threads_in_op() const { return num_threads_in_op_.Load(); }
+
+ const InternalQueue<ScanRange>* unstarted_scan_ranges() const {
+ return &unstarted_scan_ranges_;
+ }
+ const InternalQueue<WriteRange>* unstarted_write_ranges() const {
+ return &unstarted_write_ranges_;
+ }
+ const InternalQueue<RequestRange>* in_flight_ranges() const {
+ return &in_flight_ranges_;
+ }
+
+ InternalQueue<ScanRange>* unstarted_scan_ranges() { return &unstarted_scan_ranges_; }
+ InternalQueue<WriteRange>* unstarted_write_ranges() {
+ return &unstarted_write_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.
+ /// context->lock_ must be held by the caller via 'context_lock'.
+ void ScheduleContext(const unique_lock<mutex>& context_lock,
+ RequestContext* context, int disk_id);
+
+ /// See RequestContext::IncrementDiskThreadAfterDequeue() comment for usage.
+ ///
+ /// The caller does not need to hold 'lock_', so this can execute concurrently with
+ /// itself and DecrementDiskThread().
+ void IncrementDiskThreadAfterDequeue() {
+ /// Incrementing 'num_threads_in_op_' first so that there is no window when other
+ /// threads see 'is_on_queue_ == num_threads_in_op_ == 0' and think there are no
+ /// references left to this context.
+ num_threads_in_op_.Add(1);
+ is_on_queue_.Store(0);
+ }
+
+ /// Decrement the count of disks threads with a reference to this context. Does final
+ /// cleanup if the context is cancelled and this is the last thread for the disk.
+ /// context->lock_ must be held by the caller via 'context_lock'.
+ void DecrementDiskThread(const unique_lock<mutex>& context_lock,
+ RequestContext* context) {
+ DCHECK(context_lock.mutex() == &context->lock_ && context_lock.owns_lock());
+ num_threads_in_op_.Add(-1);
+
+ if (context->state_ != Cancelled) {
+ DCHECK_EQ(context->state_, Active);
+ return;
+ }
+ // The state is cancelled, check to see if we're the last thread to touch the
+ // context on this disk. We need to load 'is_on_queue_' and 'num_threads_in_op_'
+ // in this order to avoid a race with IncrementDiskThreadAfterDequeue().
+ if (is_on_queue_.Load() == 0 && num_threads_in_op_.Load() == 0 && !done_) {
+ context->DecrementDiskRefCount(context_lock);
+ done_ = true;
+ }
+ }
+
+ private:
+ /// The IoMgr disk queue corresponding to this context. Assigned during construction
+ /// of the RequestContext and not modified after.
+ DiskQueue* disk_queue_ = nullptr;
+
+ /// If true, this disk is all done for this request context, including any cleanup.
+ /// If done is true, it means that this request must not be on this disk's queue
+ /// *AND* there are no threads currently working on this context. To satisfy
+ /// this, only the last thread (per disk) can set this to true.
+ bool done_ = true;
+
+ /// For each disk, keeps track if the context is on this disk's queue, indicating
+ /// the disk must do some work for this context. 1 means that the context is on the
+ /// disk queue, 0 means that it's not on the queue (either because it has on ranges
+ /// active for the disk or because a disk thread dequeued the context and is
+ /// currently processing a request).
+ ///
+ /// The disk needs to do work in 4 cases:
+ /// 1) in_flight_ranges is not empty, the disk needs to read for this reader.
+ /// 2) next_range_to_start is NULL, the disk needs to prepare a scan range to be
+ /// read next.
+ /// 3) the reader has been cancelled and this disk needs to participate in the
+ /// cleanup.
+ /// 4) A write range is added to queue.
+ /// In general, we only want to put a context on the disk queue if there is something
+ /// useful that can be done. If there's nothing useful, the disk queue will wake up
+ /// and then remove the reader from the queue. Doing this causes thrashing of the
+ /// threads.
+ ///
+ /// This variable is important during context cancellation because it indicates
+ /// whether a queue has a reference to the context that must be released before
+ /// the context is considered unregistered. Atomically set to false after
+ /// incrementing 'num_threads_in_op_' when dequeueing so that there is no window
+ /// when other threads see 'is_on_queue_ == num_threads_in_op_ == 0' and think there
+ /// are no references left to this context.
+ /// TODO: this could be combined with 'num_threads_in_op_' to be a single refcount.
+ AtomicInt32 is_on_queue_{0};
+
+ /// For each disks, the number of request ranges that have not been fully read.
+ /// In the non-cancellation path, this will hit 0, and done will be set to true
+ /// by the disk thread. This is undefined in the cancellation path (the various
+ /// threads notice by looking at the RequestContext's state_).
+ int num_remaining_ranges_ = 0;
+
+ /// Queue of ranges that have not started being read. This list is exclusive
+ /// with in_flight_ranges.
+ InternalQueue<ScanRange> unstarted_scan_ranges_;
+
+ /// Queue of pending IO requests for this disk in the order that they will be
+ /// processed. A ScanRange is added to this queue when it is returned in
+ /// GetNextUnstartedRange(), or when it is added with schedule_mode == IMMEDIATELY.
+ /// A WriteRange is added to this queue from unstarted_write_ranges_ for each
+ /// invocation of GetNextRequestRange() in WorkLoop().
+ /// The size of this queue is always less than or equal to num_remaining_ranges.
+ InternalQueue<RequestRange> in_flight_ranges_;
+
+ /// The next range to start for this reader on this disk. Each disk (for each reader)
+ /// picks the next range to start. The range is set here and also added to the
+ /// ready_to_start_ranges_ queue. The reader pulls from the queue in FIFO order,
+ /// so the ranges from different disks are round-robined. When the range is pulled
+ /// off the ready_to_start_ranges_ queue, it sets this variable to NULL, so the disk
+ /// knows to populate it again and add it to ready_to_start_ranges_ i.e. it is used
+ /// as a flag by DiskIoMgr::GetNextScanRange to determine if it needs to add another
+ /// range to ready_to_start_ranges_.
+ ScanRange* next_scan_range_to_start_ = nullptr;
+
+ /// For each disk, the number of disk threads issuing the underlying read/write on
+ /// behalf of this context. There are a few places where we release the context lock,
+ /// do some work, and then grab the lock again. Because we don't hold the lock for
+ /// the entire operation, we need this ref count to keep track of which thread should
+ /// do final resource cleanup during cancellation.
+ /// Only the thread that sees the count at 0 should do the final cleanup.
+ AtomicInt32 num_threads_in_op_{0};
+
+ /// Queue of write ranges to process for this disk. A write range is always added
+ /// to in_flight_ranges_ in GetNextRequestRange(). There is a separate
+ /// unstarted_read_ranges_ and unstarted_write_ranges_ to alternate between reads
+ /// and writes. (Otherwise, since next_scan_range_to_start is set
+ /// in GetNextRequestRange() whenever it is null, repeated calls to
+ /// GetNextRequestRange() and GetNextUnstartedRange() may result in only reads being
+ /// processed)
+ InternalQueue<WriteRange> unstarted_write_ranges_;
+};
+
void RequestContext::ReadDone(int disk_id, ReadOutcome outcome, ScanRange* range) {
// TODO: IMPALA-4249: it is safe to touch 'range' until DecrementDiskThread() is
// called because all clients of DiskIoMgr keep ScanRange objects alive until they
@@ -52,6 +213,24 @@ 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();
+
+ // 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);
+ {
+ unique_lock<mutex> lock(lock_);
+ DCHECK(Validate()) << endl << DebugString();
+ RequestContext::PerDiskState& state = disk_states_[disk_id];
+ state.DecrementDiskThread(lock, this);
+ --state.num_remaining_ranges();
+ }
+}
+
// Cancellation of a RequestContext requires coordination from multiple threads that may
// hold references to the context:
// 1. Disk threads that are currently processing a range for this context.
@@ -133,7 +312,7 @@ void RequestContext::Cancel() {
void RequestContext::CancelAndMarkInactive() {
Cancel();
- boost::unique_lock<boost::mutex> l(lock_);
+ unique_lock<mutex> l(lock_);
DCHECK_NE(state_, Inactive);
DCHECK(Validate()) << endl << DebugString();
@@ -190,6 +369,131 @@ void RequestContext::AddRangeToDisk(const unique_lock<mutex>& lock,
++disk_state->num_remaining_ranges();
}
+Status RequestContext::AddScanRanges(const vector<ScanRange*>& ranges) {
+ DCHECK_GT(ranges.size(), 0);
+ // Validate and initialize all ranges
+ for (int i = 0; i < ranges.size(); ++i) {
+ RETURN_IF_ERROR(parent_->ValidateScanRange(ranges[i]));
+ ranges[i]->InitInternal(parent_, this);
+ }
+
+ unique_lock<mutex> lock(lock_);
+ DCHECK(Validate()) << endl << DebugString();
+
+ if (state_ == RequestContext::Cancelled) return Status::CANCELLED;
+
+ // Add each range to the queue of the disk the range is on
+ for (ScanRange* range : ranges) {
+ // Don't add empty ranges.
+ DCHECK_NE(range->len(), 0);
+ AddActiveScanRangeLocked(lock, range);
+ if (range->try_cache()) {
+ cached_ranges_.Enqueue(range);
+ } else {
+ AddRangeToDisk(lock, range, ScheduleMode::UPON_GETNEXT);
+ }
+ }
+ DCHECK(Validate()) << endl << DebugString();
+ return Status::OK();
+}
+
+// This function returns the next scan range the reader should work on, checking
+// for eos and error cases. If there isn't already a cached scan range or a scan
+// range prepared by the disk threads, the caller waits on the disk threads.
+Status RequestContext::GetNextUnstartedRange(ScanRange** range, bool* needs_buffers) {
+ DCHECK(range != nullptr);
+ *range = nullptr;
+ *needs_buffers = false;
+
+ unique_lock<mutex> lock(lock_);
+ DCHECK(Validate()) << endl << DebugString();
+ while (true) {
+ if (state_ == RequestContext::Cancelled) return Status::CANCELLED;
+
+ if (num_unstarted_scan_ranges_.Load() == 0 && ready_to_start_ranges_.empty()
+ && cached_ranges_.empty()) {
+ // All ranges are done, just return.
+ return Status::OK();
+ }
+
+ if (!cached_ranges_.empty()) {
+ // We have a cached range.
+ *range = cached_ranges_.Dequeue();
+ DCHECK((*range)->try_cache());
+ bool cached_read_succeeded;
+ RETURN_IF_ERROR((*range)->ReadFromCache(lock, &cached_read_succeeded));
+ if (cached_read_succeeded) return Status::OK();
+
+ // This range ended up not being cached. Loop again and pick up a new range.
+ AddRangeToDisk(lock, *range, ScheduleMode::UPON_GETNEXT);
+ DCHECK(Validate()) << endl << DebugString();
+ *range = nullptr;
+ continue;
+ }
+
+ if (ready_to_start_ranges_.empty()) {
+ ready_to_start_ranges_cv_.Wait(lock);
+ } else {
+ *range = ready_to_start_ranges_.Dequeue();
+ DCHECK(*range != nullptr);
+ int disk_id = (*range)->disk_id();
+ DCHECK_EQ(*range, disk_states_[disk_id].next_scan_range_to_start());
+ // Set this to nullptr, the next time this disk runs for this reader, it will
+ // get another range ready.
+ disk_states_[disk_id].set_next_scan_range_to_start(nullptr);
+ ScanRange::ExternalBufferTag buffer_tag = (*range)->external_buffer_tag();
+ if (buffer_tag == ScanRange::ExternalBufferTag::NO_BUFFER) {
+ // We can't schedule this range until the client gives us buffers. The context
+ // must be rescheduled regardless to ensure that 'next_scan_range_to_start' is
+ // refilled.
+ disk_states_[disk_id].ScheduleContext(lock, this, disk_id);
+ (*range)->SetBlockedOnBuffer();
+ *needs_buffers = true;
+ } else {
+ ScheduleScanRange(lock, *range);
+ }
+ return Status::OK();
+ }
+ }
+}
+
+Status RequestContext::StartScanRange(ScanRange* range, bool* needs_buffers) {
+ RETURN_IF_ERROR(parent_->ValidateScanRange(range));
+ range->InitInternal(parent_, this);
+
+ unique_lock<mutex> lock(lock_);
+ DCHECK(Validate()) << endl << DebugString();
+ if (state_ == RequestContext::Cancelled) return Status::CANCELLED;
+
+ DCHECK_NE(range->len(), 0);
+ if (range->try_cache()) {
+ bool cached_read_succeeded;
+ RETURN_IF_ERROR(range->ReadFromCache(lock, &cached_read_succeeded));
+ if (cached_read_succeeded) {
+ DCHECK(Validate()) << endl << DebugString();
+ *needs_buffers = false;
+ return Status::OK();
+ }
+ // Cached read failed, fall back to normal read path.
+ }
+ // If we don't have a buffer yet, the caller must allocate buffers for the range.
+ *needs_buffers =
+ range->external_buffer_tag() == ScanRange::ExternalBufferTag::NO_BUFFER;
+ if (*needs_buffers) range->SetBlockedOnBuffer();
+ AddActiveScanRangeLocked(lock, range);
+ AddRangeToDisk(lock, range,
+ *needs_buffers ? ScheduleMode::BY_CALLER : ScheduleMode::IMMEDIATELY);
+ DCHECK(Validate()) << endl << DebugString();
+ return Status::OK();
+}
+
+Status RequestContext::AddWriteRange(WriteRange* write_range) {
+ unique_lock<mutex> lock(lock_);
+ if (state_ == RequestContext::Cancelled) return Status::CANCELLED;
+ AddRangeToDisk(lock, write_range, ScheduleMode::IMMEDIATELY);
+ return Status::OK();
+}
+
void RequestContext::AddActiveScanRangeLocked(
const unique_lock<mutex>& lock, ScanRange* range) {
DCHECK(lock.mutex() == &lock_ && lock.owns_lock());
@@ -278,8 +582,18 @@ RequestRange* RequestContext::GetNextRequestRange(int disk_id) {
return range;
}
-RequestContext::RequestContext(DiskIoMgr* parent, int num_disks)
- : parent_(parent), disk_states_(num_disks) {}
+RequestContext::RequestContext(
+ DiskIoMgr* parent, const std::vector<DiskQueue*>& disk_queues)
+ : parent_(parent), disk_states_(disk_queues.size()) {
+ // PerDiskState is not movable, so we need to initialize the vector in this awkward way.
+ for (int i = 0; i < disk_queues.size(); ++i) {
+ disk_states_[i].set_disk_queue(disk_queues[i]);
+ }
+}
+
+RequestContext::~RequestContext() {
+ DCHECK_EQ(state_, Inactive) << "Must be unregistered. " << DebugString();
+}
// Dumps out request context information. Lock should be taken by caller
string RequestContext::DebugString() const {
@@ -412,6 +726,30 @@ void RequestContext::PerDiskState::ScheduleContext(const unique_lock<mutex>& con
DCHECK(context_lock.mutex() == &context->lock_ && context_lock.owns_lock());
if (is_on_queue_.Load() == 0 && !done_) {
is_on_queue_.Store(1);
- context->parent_->GetDiskQueue(disk_id)->EnqueueContext(context);
+ disk_queue_->EnqueueContext(context);
+ }
+}
+
+void RequestContext::IncrementDiskThreadAfterDequeue(int disk_id) {
+ disk_states_[disk_id].IncrementDiskThreadAfterDequeue();
+}
+
+void RequestContext::DecrementDiskRefCount(
+ const boost::unique_lock<boost::mutex>& lock) {
+ DCHECK(lock.mutex() == &lock_ && lock.owns_lock());
+ DCHECK_GT(num_disks_with_ranges_, 0);
+ if (--num_disks_with_ranges_ == 0) {
+ disks_complete_cond_var_.NotifyAll();
}
+ DCHECK(Validate()) << std::endl << DebugString();
+}
+
+void RequestContext::ScheduleScanRange(
+ const boost::unique_lock<boost::mutex>& lock, ScanRange* range) {
+ DCHECK(lock.mutex() == &lock_ && lock.owns_lock());
+ DCHECK_EQ(state_, Active);
+ DCHECK(range != nullptr);
+ RequestContext::PerDiskState& state = disk_states_[range->disk_id()];
+ state.in_flight_ranges()->Enqueue(range);
+ state.ScheduleContext(lock, this, range->disk_id());
}
http://git-wip-us.apache.org/repos/asf/impala/blob/b0feb2c1/be/src/runtime/io/request-context.h
----------------------------------------------------------------------
diff --git a/be/src/runtime/io/request-context.h b/be/src/runtime/io/request-context.h
index 2687ee1..0013cc8 100644
--- a/be/src/runtime/io/request-context.h
+++ b/be/src/runtime/io/request-context.h
@@ -94,9 +94,43 @@ enum class ScheduleMode {
/// be queued up behind at most one write range.
class RequestContext {
public:
- ~RequestContext() {
- DCHECK_EQ(state_, Inactive) << "Must be unregistered. " << DebugString();
- }
+ ~RequestContext();
+
+ /// Adds the scan ranges to this context's queues, but does not start scheduling it.
+ /// The range can be scheduled by a thread calling GetNextUnstartedRange(). This call
+ /// is non-blocking. The caller must not deallocate the scan range pointers before
+ /// UnregisterContext(). 'ranges' must not be empty.
+ Status AddScanRanges(const std::vector<ScanRange*>& ranges) WARN_UNUSED_RESULT;
+
+ /// Tries to get an unstarted scan range that was added to this context with
+ /// AddScanRanges(). On success, returns OK and returns the range in '*range'.
+ /// If this context was cancelled, returns CANCELLED. If another error is encountered,
+ /// an error status is returned. Otherwise, if error or cancellation wasn't encountered
+ /// and there are no unstarted ranges for this context, returns OK and sets '*range' to
+ /// nullptr.
+ ///
+ /// If '*needs_buffers' is returned as true, the caller must call
+ /// AllocateBuffersForRange() to add buffers for the data to be read into before the
+ /// range can be scheduled. Otherwise, the range is scheduled and the IoMgr will
+ /// asynchronously read the data for the range and the caller can call
+ /// ScanRange::GetNext() to read the data.
+ Status GetNextUnstartedRange(ScanRange** range, bool* needs_buffers) WARN_UNUSED_RESULT;
+
+ /// Adds the scan range to the queues, as with AddScanRanges(), but immediately
+ /// start scheduling the scan range. This can be used to do synchronous reads as well
+ /// as schedule dependent ranges, e.g. for columnar formats. This call is non-blocking.
+ /// The caller must not deallocate the scan range pointers before UnregisterContext().
+ ///
+ /// If this returns true in '*needs_buffers', the caller must then call
+ /// AllocateBuffersForRange() to add buffers for the data to be read into before the
+ /// range can be scheduled. Otherwise, the range is scheduled and the IoMgr will
+ /// asynchronously read the data for the range and the caller can call
+ /// ScanRange::GetNext() to read the data.
+ Status StartScanRange(ScanRange* range, bool* needs_buffers) WARN_UNUSED_RESULT;
+
+ /// Add a WriteRange for the writer. This is non-blocking and schedules the context
+ /// on the IoMgr disk queue.
+ Status AddWriteRange(WriteRange* 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.
@@ -143,8 +177,11 @@ class RequestContext {
disks_accessed_bitmap_ = disks_accessed_bitmap;
}
- protected:
- // Protected methods are accessed by other classes in io:: but not external classes.
+ private:
+ DISALLOW_COPY_AND_ASSIGN(RequestContext);
+ class PerDiskState;
+ /////////////////////////////////////////
+ /// BEGIN: private members that are accessed by other io:: classes
friend class DiskQueue;
friend class DiskIoMgr;
friend class ScanRange;
@@ -163,7 +200,7 @@ class RequestContext {
Inactive,
};
- RequestContext(DiskIoMgr* parent, int num_disks);
+ RequestContext(DiskIoMgr* parent, const std::vector<DiskQueue*>& disk_queues);
/// Called when dequeueing this RequestContext from the disk queue to increment the
/// count of disk threads with a reference to this context for 'disk_id. These threads
@@ -204,6 +241,12 @@ 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);
+
/// 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.
void CancelAndMarkInactive();
@@ -220,7 +263,7 @@ class RequestContext {
/// immediately added to the 'in_flight_ranges_' queue where it will be processed
/// asynchronously by disk threads. If UPON_GETNEXT, the range is added to the
/// 'unstarted_ranges_' queue, from which it can be returned to a client by
- /// DiskIoMgr::GetNextUnstartedRange(). If BY_CALLER, the scan range is not added to
+ /// GetNextUnstartedRange(). If BY_CALLER, the scan range is not added to
/// any queues. The range will be scheduled later as a separate step, e.g. when it is
/// unblocked by adding buffers to it. Caller must hold 'lock_' via 'lock'.
void AddRangeToDisk(const boost::unique_lock<boost::mutex>& lock, RequestRange* range,
@@ -282,10 +325,9 @@ class RequestContext {
/// Total number of file handle opens where the file handle was not in the cache
AtomicInt32 cached_file_handles_miss_count_{0};
- private:
- DISALLOW_COPY_AND_ASSIGN(RequestContext);
+ /// END: private members that are accessed by other io:: classes
+ /////////////////////////////////////////
- /// Validates invariants of reader. 'lock_' must be held by caller.
bool Validate() const;
/// Dumps out reader information. 'lock_' must be held by caller.
@@ -325,7 +367,7 @@ class RequestContext {
/// A list of ranges that should be returned in subsequent calls to
/// GetNextUnstartedRange().
/// There is a trade-off with when to populate this list. Populating it on
- /// demand means consumers need to wait (happens in DiskIoMgr::GetNextUnstartedRange()).
+ /// demand means consumers need to wait (happens in GetNextUnstartedRange()).
/// Populating it preemptively means we make worse scheduling decisions.
/// We currently populate one range per disk.
/// TODO: think about this some more.
@@ -335,189 +377,10 @@ class RequestContext {
/// Condition variable for UnregisterContext() to wait for all disks to complete
ConditionVariable disks_complete_cond_var_;
- private:
/// Per disk states to synchronize multiple disk threads accessing the same request
- /// context.
- class PerDiskState;
+ /// context. One state per IoMgr disk queue.
std::vector<PerDiskState> disk_states_;
};
-
-/// Struct containing state per disk. See comments in the disk read loop on how
-/// they are used.
-class RequestContext::PerDiskState {
- public:
- bool done() const { return done_; }
- void set_done(bool b) { done_ = b; }
-
- int num_remaining_ranges() const { return num_remaining_ranges_; }
- int& num_remaining_ranges() { return num_remaining_ranges_; }
-
- ScanRange* next_scan_range_to_start() { return next_scan_range_to_start_; }
- void set_next_scan_range_to_start(ScanRange* range) {
- next_scan_range_to_start_ = range;
- }
-
- bool is_on_queue() const { return is_on_queue_.Load() != 0; }
-
- int num_threads_in_op() const { return num_threads_in_op_.Load(); }
-
- const InternalQueue<ScanRange>* unstarted_scan_ranges() const {
- return &unstarted_scan_ranges_;
- }
- const InternalQueue<WriteRange>* unstarted_write_ranges() const {
- return &unstarted_write_ranges_;
- }
- const InternalQueue<RequestRange>* in_flight_ranges() const {
- return &in_flight_ranges_;
- }
-
- InternalQueue<ScanRange>* unstarted_scan_ranges() { return &unstarted_scan_ranges_; }
- InternalQueue<WriteRange>* unstarted_write_ranges() {
- return &unstarted_write_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.
- /// context->lock_ must be held by the caller via 'context_lock'.
- void ScheduleContext(const boost::unique_lock<boost::mutex>& context_lock,
- RequestContext* context, int disk_id);
-
- /// See RequestContext::IncrementDiskThreadAfterDequeue() comment for usage.
- ///
- /// The caller does not need to hold 'lock_', so this can execute concurrently with
- /// itself and DecrementDiskThread().
- void IncrementDiskThreadAfterDequeue() {
- /// Incrementing 'num_threads_in_op_' first so that there is no window when other
- /// threads see 'is_on_queue_ == num_threads_in_op_ == 0' and think there are no
- /// references left to this context.
- num_threads_in_op_.Add(1);
- is_on_queue_.Store(0);
- }
-
- /// Decrement the count of disks threads with a reference to this context. Does final
- /// cleanup if the context is cancelled and this is the last thread for the disk.
- /// context->lock_ must be held by the caller via 'context_lock'.
- void DecrementDiskThread(const boost::unique_lock<boost::mutex>& context_lock,
- RequestContext* context) {
- DCHECK(context_lock.mutex() == &context->lock_ && context_lock.owns_lock());
- num_threads_in_op_.Add(-1);
-
- if (context->state_ != Cancelled) {
- DCHECK_EQ(context->state_, Active);
- return;
- }
- // The state is cancelled, check to see if we're the last thread to touch the
- // context on this disk. We need to load 'is_on_queue_' and 'num_threads_in_op_'
- // in this order to avoid a race with IncrementDiskThreadAfterDequeue().
- if (is_on_queue_.Load() == 0 && num_threads_in_op_.Load() == 0 && !done_) {
- context->DecrementDiskRefCount(context_lock);
- done_ = true;
- }
- }
-
- private:
- /// If true, this disk is all done for this request context, including any cleanup.
- /// If done is true, it means that this request must not be on this disk's queue
- /// *AND* there are no threads currently working on this context. To satisfy
- /// this, only the last thread (per disk) can set this to true.
- bool done_ = true;
-
- /// For each disk, keeps track if the context is on this disk's queue, indicating
- /// the disk must do some work for this context. 1 means that the context is on the
- /// disk queue, 0 means that it's not on the queue (either because it has on ranges
- /// active for the disk or because a disk thread dequeued the context and is
- /// currently processing a request).
- ///
- /// The disk needs to do work in 4 cases:
- /// 1) in_flight_ranges is not empty, the disk needs to read for this reader.
- /// 2) next_range_to_start is NULL, the disk needs to prepare a scan range to be
- /// read next.
- /// 3) the reader has been cancelled and this disk needs to participate in the
- /// cleanup.
- /// 4) A write range is added to queue.
- /// In general, we only want to put a context on the disk queue if there is something
- /// useful that can be done. If there's nothing useful, the disk queue will wake up
- /// and then remove the reader from the queue. Doing this causes thrashing of the
- /// threads.
- ///
- /// This variable is important during context cancellation because it indicates
- /// whether a queue has a reference to the context that must be released before
- /// the context is considered unregistered. Atomically set to false after
- /// incrementing 'num_threads_in_op_' when dequeueing so that there is no window
- /// when other threads see 'is_on_queue_ == num_threads_in_op_ == 0' and think there
- /// are no references left to this context.
- /// TODO: this could be combined with 'num_threads_in_op_' to be a single refcount.
- AtomicInt32 is_on_queue_{0};
-
- /// For each disks, the number of request ranges that have not been fully read.
- /// In the non-cancellation path, this will hit 0, and done will be set to true
- /// by the disk thread. This is undefined in the cancellation path (the various
- /// threads notice by looking at the RequestContext's state_).
- int num_remaining_ranges_ = 0;
-
- /// Queue of ranges that have not started being read. This list is exclusive
- /// with in_flight_ranges.
- InternalQueue<ScanRange> unstarted_scan_ranges_;
-
- /// Queue of pending IO requests for this disk in the order that they will be
- /// processed. A ScanRange is added to this queue when it is returned in
- /// GetNextUnstartedRange(), or when it is added with schedule_mode == IMMEDIATELY.
- /// A WriteRange is added to this queue from unstarted_write_ranges_ for each
- /// invocation of GetNextRequestRange() in WorkLoop().
- /// The size of this queue is always less than or equal to num_remaining_ranges.
- InternalQueue<RequestRange> in_flight_ranges_;
-
- /// The next range to start for this reader on this disk. Each disk (for each reader)
- /// picks the next range to start. The range is set here and also added to the
- /// ready_to_start_ranges_ queue. The reader pulls from the queue in FIFO order,
- /// so the ranges from different disks are round-robined. When the range is pulled
- /// off the ready_to_start_ranges_ queue, it sets this variable to NULL, so the disk
- /// knows to populate it again and add it to ready_to_start_ranges_ i.e. it is used
- /// as a flag by DiskIoMgr::GetNextScanRange to determine if it needs to add another
- /// range to ready_to_start_ranges_.
- ScanRange* next_scan_range_to_start_ = nullptr;
-
- /// For each disk, the number of disk threads issuing the underlying read/write on
- /// behalf of this context. There are a few places where we release the context lock,
- /// do some work, and then grab the lock again. Because we don't hold the lock for
- /// the entire operation, we need this ref count to keep track of which thread should
- /// do final resource cleanup during cancellation.
- /// Only the thread that sees the count at 0 should do the final cleanup.
- AtomicInt32 num_threads_in_op_{0};
-
- /// Queue of write ranges to process for this disk. A write range is always added
- /// to in_flight_ranges_ in GetNextRequestRange(). There is a separate
- /// unstarted_read_ranges_ and unstarted_write_ranges_ to alternate between reads
- /// and writes. (Otherwise, since next_scan_range_to_start is set
- /// in GetNextRequestRange() whenever it is null, repeated calls to
- /// GetNextRequestRange() and GetNextUnstartedRange() may result in only reads being
- /// processed)
- InternalQueue<WriteRange> unstarted_write_ranges_;
-};
-
-inline void RequestContext::IncrementDiskThreadAfterDequeue(int disk_id) {
- disk_states_[disk_id].IncrementDiskThreadAfterDequeue();
-}
-
-inline void RequestContext::DecrementDiskRefCount(
- const boost::unique_lock<boost::mutex>& lock) {
- DCHECK(lock.mutex() == &lock_ && lock.owns_lock());
- DCHECK_GT(num_disks_with_ranges_, 0);
- if (--num_disks_with_ranges_ == 0) {
- disks_complete_cond_var_.NotifyAll();
- }
- DCHECK(Validate()) << std::endl << DebugString();
-}
-
-inline void RequestContext::ScheduleScanRange(
- const boost::unique_lock<boost::mutex>& lock, ScanRange* range) {
- DCHECK(lock.mutex() == &lock_ && lock.owns_lock());
- DCHECK_EQ(state_, Active);
- DCHECK(range != nullptr);
- RequestContext::PerDiskState& state = disk_states_[range->disk_id()];
- state.in_flight_ranges()->Enqueue(range);
- state.ScheduleContext(lock, this, range->disk_id());
-}
}
}
http://git-wip-us.apache.org/repos/asf/impala/blob/b0feb2c1/be/src/runtime/io/request-ranges.h
----------------------------------------------------------------------
diff --git a/be/src/runtime/io/request-ranges.h b/be/src/runtime/io/request-ranges.h
index 0d2ba85..22c6c36 100644
--- a/be/src/runtime/io/request-ranges.h
+++ b/be/src/runtime/io/request-ranges.h
@@ -44,6 +44,11 @@ class ScanRange;
/// time.
class BufferDescriptor {
public:
+ /// Create a buffer descriptor allocated from the buffer pool. Public to
+ /// allow access by DiskIoMgr.
+ BufferDescriptor(ScanRange* scan_range, BufferPool::ClientHandle* bp_client,
+ BufferPool::BufferHandle handle);
+
~BufferDescriptor() {
DCHECK(buffer_ == nullptr); // Check we didn't leak a buffer.
}
@@ -57,25 +62,14 @@ class BufferDescriptor {
/// Returns the offset within the scan range that this buffer starts at
int64_t scan_range_offset() const { return scan_range_offset_; }
- protected:
- // Protected methods are called from other classes in io::.
- friend class DiskIoMgr;
-
- /// Create a buffer descriptor for a new reader, range and data buffer.
- BufferDescriptor(DiskIoMgr* io_mgr, RequestContext* reader,
- ScanRange* scan_range, uint8_t* buffer, int64_t buffer_len);
-
- /// Create a buffer descriptor allocated from the buffer pool.
- BufferDescriptor(DiskIoMgr* io_mgr, RequestContext* reader,
- ScanRange* scan_range, BufferPool::ClientHandle* bp_client,
- BufferPool::BufferHandle handle);
-
private:
DISALLOW_COPY_AND_ASSIGN(BufferDescriptor);
-
/// This class is tightly coupled with ScanRange. Making them friends is easiest.
friend class ScanRange;
+ /// Create a buffer descriptor for a range and data buffer.
+ BufferDescriptor(ScanRange* scan_range, uint8_t* buffer, int64_t buffer_len);
+
/// Return true if this is a cached buffer owned by HDFS.
bool is_cached() const;
@@ -89,11 +83,6 @@ class BufferDescriptor {
/// is called, buffer() is NULL. Does not acquire 'lock_'.
void Free();
- DiskIoMgr* const io_mgr_;
-
- /// Reader that this buffer is for.
- RequestContext* const reader_;
-
/// Scan range that this buffer is for. Non-NULL when initialised.
ScanRange* const scan_range_;
@@ -275,11 +264,14 @@ class ScanRange : public RequestRange {
int64_t mtime() const { return mtime_; }
- protected:
- // Protected methods are called from other classes in io::.
+ private:
+ DISALLOW_COPY_AND_ASSIGN(ScanRange);
+
+ /////////////////////////////////////////
+ /// BEGIN: private members that are accessed by other io:: classes
friend class BufferDescriptor;
- friend class DiskIoMgr;
friend class DiskQueue;
+ friend class DiskIoMgr;
friend class RequestContext;
// Tag for the buffer associated with range. See external_buffer_tag_ for details.
@@ -332,8 +324,8 @@ class ScanRange : public RequestRange {
ExternalBufferTag external_buffer_tag() const { return external_buffer_tag_; }
- private:
- DISALLOW_COPY_AND_ASSIGN(ScanRange);
+ /// END: private members that are accessed by other io:: classes
+ /////////////////////////////////////////
/// Enqueues a ready buffer with valid data for this range. This does not block.
/// The caller passes ownership of buffer to the scan range and it is not
http://git-wip-us.apache.org/repos/asf/impala/blob/b0feb2c1/be/src/runtime/io/scan-range.cc
----------------------------------------------------------------------
diff --git a/be/src/runtime/io/scan-range.cc b/be/src/runtime/io/scan-range.cc
index 163da28..6028fb4 100644
--- a/be/src/runtime/io/scan-range.cc
+++ b/be/src/runtime/io/scan-range.cc
@@ -179,7 +179,7 @@ ReadOutcome ScanRange::DoRead(int disk_id) {
if (external_buffer_tag_ == ScanRange::ExternalBufferTag::CLIENT_BUFFER) {
buffer_desc = unique_ptr<BufferDescriptor>(new BufferDescriptor(
- io_mgr_, reader_, this, client_buffer_.data, client_buffer_.len));
+ this, client_buffer_.data, client_buffer_.len));
} else {
DCHECK(external_buffer_tag_ == ScanRange::ExternalBufferTag::NO_BUFFER)
<< "This code path does not handle other buffer types, i.e. HDFS cache"
@@ -207,7 +207,6 @@ ReadOutcome ScanRange::DoRead(int disk_id) {
buffer_desc->scan_range_offset_ = bytes_read_ - buffer_desc->len_;
COUNTER_ADD_IF_NOT_NULL(reader_->bytes_read_counter_, buffer_desc->len_);
- COUNTER_ADD(io_mgr_->total_bytes_read_counter(), buffer_desc->len_);
COUNTER_ADD_IF_NOT_NULL(reader_->active_read_thread_counter_, -1L);
}
@@ -594,7 +593,6 @@ Status ScanRange::Read(
int64_t max_chunk_size = MaxReadChunkSize();
Status status = Status::OK();
{
- ScopedTimer<MonotonicStopWatch> io_mgr_read_timer(io_mgr_->read_timer());
ScopedTimer<MonotonicStopWatch> req_context_read_timer(reader_->read_timer_);
while (*bytes_read < bytes_to_read) {
int chunk_size = min(bytes_to_read - *bytes_read, max_chunk_size);
@@ -650,13 +648,11 @@ Status ScanRange::Read(
}
// The error may be due to a bad file handle. Reopen the file handle and retry.
// Exclude this time from the read timers.
- io_mgr_read_timer.Stop();
req_context_read_timer.Stop();
++num_retries;
RETURN_IF_ERROR(io_mgr_->ReopenCachedHdfsFileHandle(fs_, file_string(),
mtime(), reader_, &borrowed_hdfs_fh));
hdfs_file = borrowed_hdfs_fh->file();
- io_mgr_read_timer.Start();
req_context_read_timer.Start();
}
if (!status.ok()) break;
@@ -749,7 +745,7 @@ Status ScanRange::ReadFromCache(
// Create a single buffer desc for the entire scan range and enqueue that.
// The memory is owned by the HDFS java client, not the Impala backend.
unique_ptr<BufferDescriptor> desc = unique_ptr<BufferDescriptor>(new BufferDescriptor(
- io_mgr_, reader_, this, reinterpret_cast<uint8_t*>(buffer), 0));
+ this, reinterpret_cast<uint8_t*>(buffer), 0));
desc->len_ = bytes_read;
desc->scan_range_offset_ = 0;
desc->eosr_ = true;
@@ -777,31 +773,23 @@ void ScanRange::GetHdfsStatistics(hdfsFile hdfs_file) {
}
}
-BufferDescriptor::BufferDescriptor(DiskIoMgr* io_mgr,
- RequestContext* reader, ScanRange* scan_range, uint8_t* buffer,
- int64_t buffer_len)
- : io_mgr_(io_mgr),
- reader_(reader),
- scan_range_(scan_range),
+BufferDescriptor::BufferDescriptor(ScanRange* scan_range,
+ uint8_t* buffer, int64_t buffer_len)
+ : scan_range_(scan_range),
buffer_(buffer),
buffer_len_(buffer_len) {
- DCHECK(io_mgr != nullptr);
DCHECK(scan_range != nullptr);
DCHECK(buffer != nullptr);
DCHECK_GE(buffer_len, 0);
}
-BufferDescriptor::BufferDescriptor(DiskIoMgr* io_mgr, RequestContext* reader,
- ScanRange* scan_range, BufferPool::ClientHandle* bp_client,
- BufferPool::BufferHandle handle) :
- io_mgr_(io_mgr),
- reader_(reader),
+BufferDescriptor::BufferDescriptor(ScanRange* scan_range,
+ BufferPool::ClientHandle* bp_client, BufferPool::BufferHandle handle) :
scan_range_(scan_range),
buffer_(handle.data()),
buffer_len_(handle.len()),
bp_client_(bp_client),
handle_(move(handle)) {
- DCHECK(io_mgr != nullptr);
DCHECK(scan_range != nullptr);
DCHECK(bp_client_->is_registered());
DCHECK(handle_.is_open());