Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Tickets/dm 37334 #759

Open
wants to merge 4 commits into
base: main
Choose a base branch
from
Open

Tickets/dm 37334 #759

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
92dabd2
Temporary logging changes.
jgates108 Dec 16, 2022
0b04b38
Added log messages for bug hunting.
jgates108 Dec 20, 2022
767f11f
Logging changes.
jgates108 Jan 10, 2023
66878f4
Added temporary code to track status of requests.
jgates108 Jan 13, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
15 changes: 13 additions & 2 deletions src/ccontrol/MergingHandler.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -95,9 +95,11 @@ bool MergingHandler::flush(int bLen, BufPtr const& bufPtr, bool& last, bool& lar
if (bLen < 0) {
throw util::Bug(ERR_LOC, "MergingHandler invalid blen=" + to_string(bLen) + " from " + _wName);
}
util::InstanceCount ica(_tableName + "_Merge_flush_LDB_a");

switch (_state) {
case MsgState::HEADER_WAIT:
case MsgState::HEADER_WAIT: {
//&&&util::InstanceCount icl(_tableName + "_Merge_flush_LDB_l_header_wait");
_response->headerSize = static_cast<unsigned char>((*bufPtr)[0]);
if (!proto::ProtoHeaderWrap::unwrap(_response, *bufPtr)) {
std::string sErr =
Expand Down Expand Up @@ -140,8 +142,10 @@ bool MergingHandler::flush(int bLen, BufPtr const& bufPtr, bool& last, bool& lar
_state = MsgState::RESULT_RECV;
}
}
}
return true;
case MsgState::RESULT_WAIT: {
//&&&util::InstanceCount icp(_tableName + "_Merge_flush_LDB_p_result_wait");
nextBufSize = proto::ProtoHeaderWrap::getProtoHeaderSize();
auto jobQuery = getJobQuery().lock();
if (!_verifyResult(bufPtr, bLen)) {
Expand All @@ -160,7 +164,9 @@ bool MergingHandler::flush(int bLen, BufPtr const& bufPtr, bool& last, bool& lar
LOGS(_log, LOG_LVL_DEBUG, "Flushed last=" << last << " for tableName=" << _tableName);

auto success = _merge();
//&&&util::InstanceCount icpx(_tableName + "_Merge_flush_LDB_px");
_response.reset(new WorkerResponse());
//&&&util::InstanceCount icpz(_tableName + "_Merge_flush_LDB_pz");
return success;
}
case MsgState::RESULT_RECV:
Expand All @@ -169,6 +175,7 @@ bool MergingHandler::flush(int bLen, BufPtr const& bufPtr, bool& last, bool& lar
case MsgState::HEADER_ERR:
[[fallthrough]];
case MsgState::RESULT_ERR: {
util::InstanceCount iczerror(_tableName + "_Merge_flush_LDB_zerror");
std::ostringstream eos;
eos << "Unexpected message From:" << _wName << " flush state=" << getStateStr(_state)
<< " last=" << last;
Expand Down Expand Up @@ -224,22 +231,26 @@ void MergingHandler::_initState() {
}

bool MergingHandler::_merge() {
util::InstanceCount ica(_tableName + "_Merge_merge_LDB_a");
if (auto job = getJobQuery().lock()) {
if (_flushed) {
throw util::Bug(ERR_LOC, "MergingRequester::_merge : already flushed");
}
//&&&util::InstanceCount icb(_tableName + "_Merge_merge_LDB_b");
bool success = _infileMerger->merge(_response);
//&&&util::InstanceCount icc(_tableName + "_Merge_merge_LDB_c");
if (!success) {
LOGS(_log, LOG_LVL_WARN, "_merge() failed");
rproc::InfileMergerError const& err = _infileMerger->getError();
_setError(ccontrol::MSG_RESULT_ERROR, err.getMsg());
_state = MsgState::RESULT_ERR;
}
_response.reset();

//&&&util::InstanceCount icx(_tableName + "_Merge_merge_LDB_x");
return success;
}
LOGS(_log, LOG_LVL_ERROR, "MergingHandler::_merge() failed, jobQuery was NULL");
//&&&util::InstanceCount icz(_tableName + "_Merge_merge_LDB_z");
return false;
}

Expand Down
Loading