You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@impala.apache.org by mi...@apache.org on 2018/09/21 16:03:14 UTC
[6/7] impala git commit: IMPALA-7335: Fix a race in HdfsScanNode
IMPALA-7335: Fix a race in HdfsScanNode
This change fixes the race between the done_ flag and the status_
variable in the HdfsScanNode. Previously, a scanner thread would
mark its scan range as complete even if it ran into an error.
Another scanner thread could notice that all scan ranges have
completed and set the done_ flag before the status_ variable is
updated with the non-ok status. This caused the main thread to
return an ok status despite the scanner thread running into an
error. This change ensures that when a scanner thread runs into
an error, it updates the status_ before marking its scan range as
complete.
Testing: Ran core tests.
Since there is no deterministic method to reproduce the
race, this change does not include any regression tests.
Additionally, this change also fixes IMPALA-7430 by removing the
logs added to debug this race.
Change-Id: Id470818846a5c69ad28a6ff695069702982aa793
Reviewed-on: http://gerrit.cloudera.org:8080/11337
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/ad3e1c94
Tree: http://git-wip-us.apache.org/repos/asf/impala/tree/ad3e1c94
Diff: http://git-wip-us.apache.org/repos/asf/impala/diff/ad3e1c94
Branch: refs/heads/master
Commit: ad3e1c94abc9a4c9bf37b5c42dafd149537375dc
Parents: a9d0305
Author: poojanilangekar <po...@cloudera.com>
Authored: Mon Aug 27 11:42:19 2018 -0700
Committer: Impala Public Jenkins <im...@cloudera.com>
Committed: Thu Sep 20 22:54:04 2018 +0000
----------------------------------------------------------------------
be/src/exec/hdfs-scan-node.cc | 70 ++++++++++++++++++--------------------
be/src/exec/hdfs-scan-node.h | 10 +++---
2 files changed, 40 insertions(+), 40 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/impala/blob/ad3e1c94/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 e406dba..6c01bf4 100644
--- a/be/src/exec/hdfs-scan-node.cc
+++ b/be/src/exec/hdfs-scan-node.cc
@@ -338,9 +338,7 @@ void HdfsScanNode::ThreadTokenAvailableCb(ThreadResourcePool* pool) {
// Abort the query. This is still holding the lock_, so done_ is known to be
// false and status_ must be ok.
- DCHECK(status_.ok());
- status_ = status;
- SetDoneInternal();
+ SetDoneInternal(status);
break;
}
// Thread successfully started
@@ -395,25 +393,12 @@ void HdfsScanNode::ScannerThread(bool first_thread, int64_t scanner_thread_reser
if (!status.ok()) {
unique_lock<mutex> l(lock_);
-#ifndef NDEBUG
- // TODO (IMPALA-7430): Remove the log message once the causes of IMPALA-7335 and
- // IMPALA-7418 are determined.
- VLOG_QUERY << "Non-ok status returned by ProcessSplit = " << status.msg().msg()
- << " for Scan node (id=" << id()
- << ", status_ = " << (status_.ok() ? "ok" : status_.msg().msg())
- << ", done_ = " << (done_ ? "1" : "0") << ")";
-#endif
// If there was already an error, the main thread will do the cleanup
if (!status_.ok()) break;
- if (status.IsCancelled() && done_) {
- // Scan node initiated scanner thread cancellation. No need to do anything.
- break;
- }
- // Set status_ before calling SetDone() (which shuts down the RowBatchQueue),
- // to ensure that GetNextInternal() notices the error status.
- status_ = status;
- SetDoneInternal();
+ // Invoke SetDoneInternal() with the error status (which shuts down the
+ // RowBatchQueue) to ensure that GetNextInternal() notices the error.
+ SetDoneInternal(status);
break;
}
@@ -505,23 +490,33 @@ Status HdfsScanNode::ProcessSplit(const vector<FilterContext>& filter_ctxs,
}
status = scanner->ProcessSplit();
- if (VLOG_QUERY_IS_ON && !status.ok() && !status.IsCancelled()) {
- // This thread hit an error, record it and bail
- stringstream ss;
- ss << "Scan node (id=" << id() << ") ran into a parse error for scan range "
- << scan_range->file() << "(" << scan_range->offset() << ":"
- << scan_range->len() << ").";
- // Parquet doesn't read the range end to end so the current offset isn't useful.
- // TODO: make sure the parquet reader is outputting as much diagnostic
- // information as possible.
- if (partition->file_format() != THdfsFileFormat::PARQUET) {
- ScannerContext::Stream* stream = context.GetStream();
- ss << " Processed " << stream->total_bytes_returned() << " bytes.";
+ if (!status.ok()) {
+ if (VLOG_QUERY_IS_ON && !status.IsCancelled()) {
+ // This thread hit an error, record it and bail
+ stringstream ss;
+ ss << "Scan node (id=" << id() << ") ran into a parse error for scan range "
+ << scan_range->file() << "(" << scan_range->offset() << ":" << scan_range->len()
+ << ").";
+ // Parquet doesn't read the range end to end so the current offset isn't useful.
+ // TODO: make sure the parquet reader is outputting as much diagnostic
+ // information as possible.
+ if (partition->file_format() != THdfsFileFormat::PARQUET) {
+ ScannerContext::Stream* stream = context.GetStream();
+ ss << " Processed " << stream->total_bytes_returned() << " bytes.";
+ }
+ VLOG_QUERY << ss.str();
}
- VLOG_QUERY << ss.str();
- }
- // Transfer remaining resources to a final batch and add it to the row batch queue.
+ // If status is the first non-ok status returned by any scanner thread, update
+ // HdfsScanNodeBase::status_ variable and notify other scanner threads. Ensure that
+ // status_ is updated before marking a range as complete (The scanner->Close() call
+ // marks a scan range as complete).
+ unique_lock<mutex> l(lock_);
+ // Update the status_ and set the done_ flag if this is the first non-ok status.
+ SetDoneInternal(status);
+ }
+ // Transfer remaining resources to a final batch and add it to the row batch queue and
+ // decrement progress_ to indicate that the scan range is complete.
scanner->Close();
// Reservation may have been increased by the scanner, e.g. Parquet may allocate
// additional reservation to scan columns.
@@ -529,14 +524,17 @@ Status HdfsScanNode::ProcessSplit(const vector<FilterContext>& filter_ctxs,
return status;
}
-void HdfsScanNode::SetDoneInternal() {
+void HdfsScanNode::SetDoneInternal(const Status& status) {
+ // If the scan node is already in the done state, do nothing.
if (done_) return;
+ DCHECK(status_.ok());
done_ = true;
+ if (!status.ok()) status_ = status;
if (reader_context_ != nullptr) reader_context_->Cancel();
thread_state_.Shutdown();
}
void HdfsScanNode::SetDone() {
unique_lock<mutex> l(lock_);
- SetDoneInternal();
+ SetDoneInternal(status_);
}
http://git-wip-us.apache.org/repos/asf/impala/blob/ad3e1c94/be/src/exec/hdfs-scan-node.h
----------------------------------------------------------------------
diff --git a/be/src/exec/hdfs-scan-node.h b/be/src/exec/hdfs-scan-node.h
index bc53fab..0e0f67c 100644
--- a/be/src/exec/hdfs-scan-node.h
+++ b/be/src/exec/hdfs-scan-node.h
@@ -177,11 +177,13 @@ class HdfsScanNode : public HdfsScanNodeBase {
Status GetNextInternal(RuntimeState* state, RowBatch* row_batch, bool* eos)
WARN_UNUSED_RESULT;
- /// Sets done_ to true and triggers threads to cleanup. Must be called with lock_
- /// taken. Calling it repeatedly ignores subsequent calls.
- void SetDoneInternal();
+ /// Sets done_ to true, updates status_ if there was an error and triggers threads to
+ /// cleanup. Must be called with lock_ taken. Calling it repeatedly ignores subsequent
+ /// calls.
+ void SetDoneInternal(const Status& status);
- /// Gets lock_ and calls SetDoneInternal()
+ /// Gets lock_ and calls SetDoneInternal(status_). Usually used after the scan node
+ /// completes execution successfully.
void SetDone();
};