Skip to content

Commit cae38b9

Browse files
iagaponenkoiagaponenko
committed
TEMPORARY: commented out log statement in Czar query dispatching and result processing
1 parent ad33f42 commit cae38b9

27 files changed

+353
-349
lines changed

‎src/ccontrol/MergingHandler.cc

+28-26
Original file line numberDiff line numberDiff line change
@@ -118,7 +118,7 @@ bool readXrootFileResourceAndMerge(string const& xrootUrl,
118118
function<bool(char const*, uint32_t, bool&)> const& messageIsReady) {
119119
string const context = "MergingHandler::" + string(__func__) + " ";
120120

121-
LOGS(_log, LOG_LVL_DEBUG, context << "xrootUrl=" << xrootUrl);
121+
// LOGS(_log, LOG_LVL_DEBUG, context << "xrootUrl=" << xrootUrl);
122122

123123
// Track the file while the control flow is staying within the function.
124124
ResultFileTracker const resultFileTracker;
@@ -129,8 +129,8 @@ bool readXrootFileResourceAndMerge(string const& xrootUrl,
129129
XrdCl::XRootDStatus status;
130130
status = file.Open(xrootUrl, XrdCl::OpenFlags::Read);
131131
if (!status.IsOK()) {
132-
LOGS(_log, LOG_LVL_ERROR,
133-
context << "failed to open " << xrootUrl << ", " << xrootdStatus2str(status));
132+
// LOGS(_log, LOG_LVL_ERROR,
133+
// context << "failed to open " << xrootUrl << ", " << xrootdStatus2str(status));
134134
return false;
135135
}
136136

@@ -215,22 +215,22 @@ bool readXrootFileResourceAndMerge(string const& xrootUrl,
215215
if (!success) break;
216216
}
217217
} catch (exception const& ex) {
218-
LOGS(_log, LOG_LVL_ERROR, ex.what());
218+
// LOGS(_log, LOG_LVL_ERROR, ex.what());
219219
success = false;
220220
}
221221
status = file.Close();
222222
if (!status.IsOK()) {
223-
LOGS(_log, LOG_LVL_WARN,
224-
context << "failed to close " << xrootUrl << ", " << xrootdStatus2str(status));
223+
// LOGS(_log, LOG_LVL_WARN,
224+
// context << "failed to close " << xrootUrl << ", " << xrootdStatus2str(status));
225225
}
226226

227227
// Remove the file from the worker if it still exists. Report and ignore errors.
228228
// The files will be garbage-collected by workers.
229229
XrdCl::FileSystem fileSystem(xrootUrl);
230230
status = fileSystem.Rm(xrootUrl2path(xrootUrl));
231231
if (!status.IsOK()) {
232-
LOGS(_log, LOG_LVL_WARN,
233-
context << "failed to remove " << xrootUrl << ", " << xrootdStatus2str(status));
232+
// LOGS(_log, LOG_LVL_WARN,
233+
// context << "failed to remove " << xrootUrl << ", " << xrootdStatus2str(status));
234234
}
235235
return success;
236236
}
@@ -240,7 +240,7 @@ bool readHttpFileAndMerge(string const& httpUrl,
240240
shared_ptr<http::ClientConnPool> const& httpConnPool) {
241241
string const context = "MergingHandler::" + string(__func__) + " ";
242242

243-
LOGS(_log, LOG_LVL_DEBUG, context << "httpUrl=" << httpUrl);
243+
// LOGS(_log, LOG_LVL_DEBUG, context << "httpUrl=" << httpUrl);
244244

245245
// Track the file while the control flow is staying within the function.
246246
ResultFileTracker const resultFileTracker;
@@ -360,7 +360,7 @@ bool readHttpFileAndMerge(string const& httpUrl,
360360
to_string(offset - msgSizeBytes) + ", file: " + httpUrl);
361361
}
362362
} catch (exception const& ex) {
363-
LOGS(_log, LOG_LVL_ERROR, ex.what());
363+
// LOGS(_log, LOG_LVL_ERROR, ex.what());
364364
success = false;
365365
}
366366

@@ -370,7 +370,7 @@ bool readHttpFileAndMerge(string const& httpUrl,
370370
http::Client remover(http::Method::DELETE, httpUrl);
371371
remover.read([](char const* inBuf, size_t inBufSize) {});
372372
} catch (exception const& ex) {
373-
LOGS(_log, LOG_LVL_WARN, context << "failed to remove " << httpUrl << ", ex: " << ex.what());
373+
// LOGS(_log, LOG_LVL_WARN, context << "failed to remove " << httpUrl << ", ex: " << ex.what());
374374
}
375375
return success;
376376
}
@@ -396,7 +396,9 @@ MergingHandler::MergingHandler(std::shared_ptr<rproc::InfileMerger> merger, std:
396396
_initState();
397397
}
398398

399-
MergingHandler::~MergingHandler() { LOGS(_log, LOG_LVL_DEBUG, __func__); }
399+
MergingHandler::~MergingHandler() {
400+
// LOGS(_log, LOG_LVL_DEBUG, __func__);
401+
}
400402

401403
bool MergingHandler::flush(proto::ResponseSummary const& responseSummary, uint32_t& resultRows) {
402404
_wName = responseSummary.wname();
@@ -405,23 +407,23 @@ bool MergingHandler::flush(proto::ResponseSummary const& responseSummary, uint32
405407
// of the operation to prevent inconsistency witin the application.
406408
auto const jobQuery = getJobQuery().lock();
407409
if (jobQuery == nullptr) {
408-
LOGS(_log, LOG_LVL_ERROR, __func__ << " failed, jobQuery was NULL");
410+
// LOGS(_log, LOG_LVL_ERROR, __func__ << " failed, jobQuery was NULL");
409411
return false;
410412
}
411-
LOGS(_log, LOG_LVL_TRACE,
412-
"MergingHandler::" << __func__ << " jobid=" << responseSummary.jobid()
413-
<< " transmitsize=" << responseSummary.transmitsize()
414-
<< " rowcount=" << responseSummary.rowcount() << " rowSize="
415-
<< " attemptcount=" << responseSummary.attemptcount() << " errorcode="
416-
<< responseSummary.errorcode() << " errormsg=" << responseSummary.errormsg());
413+
// LOGS(_log, LOG_LVL_TRACE,
414+
// "MergingHandler::" << __func__ << " jobid=" << responseSummary.jobid()
415+
// << " transmitsize=" << responseSummary.transmitsize()
416+
// << " rowcount=" << responseSummary.rowcount() << " rowSize="
417+
// << " attemptcount=" << responseSummary.attemptcount() << " errorcode="
418+
// << responseSummary.errorcode() << " errormsg=" << responseSummary.errormsg());
417419

418420
if (responseSummary.errorcode() != 0 || !responseSummary.errormsg().empty()) {
419421
_error = util::Error(responseSummary.errorcode(), responseSummary.errormsg(),
420422
util::ErrorCode::MYSQLEXEC);
421423
_setError(ccontrol::MSG_RESULT_ERROR, _error.getMsg());
422-
LOGS(_log, LOG_LVL_ERROR,
423-
"MergingHandler::" << __func__ << " error from worker:" << responseSummary.wname()
424-
<< " error: " << _error);
424+
// LOGS(_log, LOG_LVL_ERROR,
425+
// "MergingHandler::" << __func__ << " error from worker:" << responseSummary.wname()
426+
// << " error: " << _error);
425427
return false;
426428
}
427429

@@ -450,7 +452,7 @@ bool MergingHandler::flush(proto::ResponseSummary const& responseSummary, uint32
450452
MergingHandler::_getHttpConnPool());
451453
} else {
452454
string const err = "Unexpected result delivery protocol";
453-
LOGS(_log, LOG_LVL_ERROR, __func__ << " " << err);
455+
// LOGS(_log, LOG_LVL_ERROR, __func__ << " " << err);
454456
throw util::Bug(ERR_LOC, err);
455457
}
456458
if (success) {
@@ -463,7 +465,7 @@ void MergingHandler::errorFlush(std::string const& msg, int code) {
463465
_setError(code, msg);
464466
// Might want more info from result service.
465467
// Do something about the error. FIXME.
466-
LOGS(_log, LOG_LVL_ERROR, "Error receiving result.");
468+
// LOGS(_log, LOG_LVL_ERROR, "Error receiving result.");
467469
}
468470

469471
bool MergingHandler::finished() const { return _flushed; }
@@ -501,15 +503,15 @@ bool MergingHandler::_merge(proto::ResponseSummary const& responseSummary,
501503
}
502504
bool success = _infileMerger->merge(responseSummary, responseData, jobQuery);
503505
if (!success) {
504-
LOGS(_log, LOG_LVL_WARN, __func__ << " failed");
506+
// LOGS(_log, LOG_LVL_WARN, __func__ << " failed");
505507
util::Error const& err = _infileMerger->getError();
506508
_setError(ccontrol::MSG_RESULT_ERROR, err.getMsg());
507509
}
508510
return success;
509511
}
510512

511513
void MergingHandler::_setError(int code, std::string const& msg) {
512-
LOGS(_log, LOG_LVL_DEBUG, "_setErr: code: " << code << ", message: " << msg);
514+
// LOGS(_log, LOG_LVL_DEBUG, "_setErr: code: " << code << ", message: " << msg);
513515
std::lock_guard<std::mutex> lock(_errorMutex);
514516
_error = Error(code, msg);
515517
}

‎src/ccontrol/UserQuerySelect.cc

+31-31
Original file line numberDiff line numberDiff line change
@@ -157,7 +157,7 @@ std::string UserQuerySelect::getError() const {
157157

158158
/// Attempt to kill in progress.
159159
void UserQuerySelect::kill() {
160-
LOGS(_log, LOG_LVL_DEBUG, "UserQuerySelect kill");
160+
// LOGS(_log, LOG_LVL_DEBUG, "UserQuerySelect kill");
161161
std::lock_guard<std::mutex> lock(_killMutex);
162162
if (!_killed) {
163163
_killed = true;
@@ -228,7 +228,7 @@ std::string UserQuerySelect::getResultQuery() const {
228228
if (not orderBy.empty()) {
229229
resultQuery += " " + orderBy;
230230
}
231-
LOGS(_log, LOG_LVL_DEBUG, "made result query:" << resultQuery);
231+
// LOGS(_log, LOG_LVL_DEBUG, "made result query:" << resultQuery);
232232
return resultQuery;
233233
}
234234

@@ -238,7 +238,7 @@ void UserQuerySelect::submit() {
238238

239239
// Using the QuerySession, generate query specs (text, db, chunkId) and then
240240
// create query messages and send them to the async query manager.
241-
LOGS(_log, LOG_LVL_DEBUG, "UserQuerySelect beginning submission");
241+
// LOGS(_log, LOG_LVL_DEBUG, "UserQuerySelect beginning submission");
242242
assert(_infileMerger);
243243

244244
auto taskMsgFactory = std::make_shared<qproc::TaskMsgFactory>();
@@ -249,9 +249,9 @@ void UserQuerySelect::submit() {
249249

250250
auto queryTemplates = _qSession->makeQueryTemplates();
251251

252-
LOGS(_log, LOG_LVL_DEBUG,
253-
"first query template:" << (queryTemplates.size() > 0 ? queryTemplates[0].sqlFragment()
254-
: "none produced."));
252+
// LOGS(_log, LOG_LVL_DEBUG,
253+
// "first query template:" << (queryTemplates.size() > 0 ? queryTemplates[0].sqlFragment()
254+
// : "none produced."));
255255

256256
// Writing query for each chunk, stop if query is cancelled.
257257
// attempt to change priority, requires root
@@ -266,7 +266,7 @@ void UserQuerySelect::submit() {
266266
try {
267267
_queryStatsData->queryStatsTmpRegister(_qMetaQueryId, _qSession->getChunksSize());
268268
} catch (qmeta::SqlError const& e) {
269-
LOGS(_log, LOG_LVL_WARN, "Failed queryStatsTmpRegister " << e.what());
269+
// LOGS(_log, LOG_LVL_WARN, "Failed queryStatsTmpRegister " << e.what());
270270
}
271271

272272
_executive->setScanInteractive(_qSession->getScanInteractive());
@@ -278,7 +278,7 @@ void UserQuerySelect::submit() {
278278
std::function<void(util::CmdData*)> funcBuildJob = [this, sequence, // sequence must be a copy
279279
&chunkSpec, &queryTemplates, &chunks, &chunksMtx,
280280
&ttn, &taskMsgFactory](util::CmdData*) {
281-
QSERV_LOGCONTEXT_QUERY(_qMetaQueryId);
281+
// QSERV_LOGCONTEXT_QUERY(_qMetaQueryId);
282282

283283
qproc::ChunkQuerySpec::Ptr cs;
284284
{
@@ -308,7 +308,7 @@ void UserQuerySelect::submit() {
308308
threadPriority.restoreOriginalValues();
309309
}
310310

311-
LOGS(_log, LOG_LVL_DEBUG, "total jobs in query=" << sequence);
311+
// LOGS(_log, LOG_LVL_DEBUG, "total jobs in query=" << sequence);
312312
_executive->waitForAllJobsToStart();
313313

314314
// we only care about per-chunk info for ASYNC queries
@@ -327,7 +327,7 @@ QueryState UserQuerySelect::join() {
327327
int64_t finalRows = 0;
328328
if (!_infileMerger->finalize(collectedBytes, finalRows)) {
329329
successful = false;
330-
LOGS(_log, LOG_LVL_ERROR, "InfileMerger::finalize failed");
330+
// LOGS(_log, LOG_LVL_ERROR, "InfileMerger::finalize failed");
331331
// Error: 1105 SQLSTATE: HY000 (ER_UNKNOWN_ERROR) Message: Unknown error
332332
_messageStore->addMessage(-1, "MERGE", 1105, "Failure while merging result",
333333
MessageSeverity::MSG_ERROR);
@@ -339,7 +339,7 @@ QueryState UserQuerySelect::join() {
339339
} catch (std::exception const& exc) {
340340
// exception here means error in qserv logic, we do not want to leak
341341
// it or expose it to user, just dump it to log
342-
LOGS(_log, LOG_LVL_ERROR, "exception from _discardMerger: " << exc.what());
342+
// LOGS(_log, LOG_LVL_ERROR, "exception from _discardMerger: " << exc.what());
343343
}
344344

345345
// Update the permanent message table.
@@ -355,15 +355,15 @@ QueryState UserQuerySelect::join() {
355355
QueryState state = SUCCESS;
356356
if (successful) {
357357
_qMetaUpdateStatus(qmeta::QInfo::COMPLETED, collectedRows, collectedBytes, finalRows);
358-
LOGS(_log, LOG_LVL_INFO, "Joined everything (success)");
358+
// LOGS(_log, LOG_LVL_INFO, "Joined everything (success)");
359359
} else if (_killed) {
360360
// status is already set to ABORTED
361-
LOGS(_log, LOG_LVL_ERROR, "Joined everything (killed)");
361+
// LOGS(_log, LOG_LVL_ERROR, "Joined everything (killed)");
362362
operation = proto::QueryManagement::CANCEL;
363363
state = ERROR;
364364
} else {
365365
_qMetaUpdateStatus(qmeta::QInfo::FAILED, collectedRows, collectedBytes, finalRows);
366-
LOGS(_log, LOG_LVL_ERROR, "Joined everything (failure!)");
366+
// LOGS(_log, LOG_LVL_ERROR, "Joined everything (failure!)");
367367
operation = proto::QueryManagement::CANCEL;
368368
state = ERROR;
369369
}
@@ -373,7 +373,7 @@ QueryState UserQuerySelect::join() {
373373
xrdreq::QueryManagementAction::notifyAllWorkers(czarConfig->getXrootdFrontendUrl(), operation,
374374
_qMetaCzarId, _qMetaQueryId);
375375
} catch (std::exception const& ex) {
376-
LOGS(_log, LOG_LVL_WARN, ex.what());
376+
// LOGS(_log, LOG_LVL_WARN, ex.what());
377377
}
378378
}
379379
return state;
@@ -409,18 +409,18 @@ void UserQuerySelect::discard() {
409409
// Silence merger discarding errors, because this object is being released.
410410
// client no longer cares about merger errors.
411411
}
412-
LOGS(_log, LOG_LVL_INFO, "Discarded UserQuerySelect");
412+
// LOGS(_log, LOG_LVL_INFO, "Discarded UserQuerySelect");
413413
}
414414

415415
/// Setup merger (for results handling and aggregation)
416416
void UserQuerySelect::setupMerger() {
417-
LOGS(_log, LOG_LVL_TRACE, "Setup merger");
417+
// LOGS(_log, LOG_LVL_TRACE, "Setup merger");
418418
_infileMergerConfig->targetTable = _resultTable;
419419
_infileMergerConfig->mergeStmt = _qSession->getMergeStmt();
420-
LOGS(_log, LOG_LVL_DEBUG,
421-
"setting mergeStmt:" << (_infileMergerConfig->mergeStmt != nullptr
422-
? _infileMergerConfig->mergeStmt->getQueryTemplate().sqlFragment()
423-
: "nullptr"));
420+
// LOGS(_log, LOG_LVL_DEBUG,
421+
// "setting mergeStmt:" << (_infileMergerConfig->mergeStmt != nullptr
422+
// ? _infileMergerConfig->mergeStmt->getQueryTemplate().sqlFragment()
423+
// : "nullptr"));
424424
_infileMerger =
425425
std::make_shared<rproc::InfileMerger>(*_infileMergerConfig, _databaseModels, _semaMgrConn);
426426

@@ -474,7 +474,7 @@ void UserQuerySelect::_expandSelectStarInMergeStatment(std::shared_ptr<query::Se
474474
void UserQuerySelect::saveResultQuery() { _queryMetadata->saveResultQuery(_qMetaQueryId, getResultQuery()); }
475475

476476
void UserQuerySelect::_setupChunking() {
477-
LOGS(_log, LOG_LVL_TRACE, "Setup chunking");
477+
// LOGS(_log, LOG_LVL_TRACE, "Setup chunking");
478478
// Do not throw exceptions here, set _errorExtra .
479479
std::shared_ptr<qproc::IndexMap> im;
480480
std::string dominantDb = _qSession->getDominantDb();
@@ -488,11 +488,11 @@ void UserQuerySelect::_setupChunking() {
488488
eSet = _qSession->getEmptyChunks();
489489
if (!eSet) {
490490
eSet = std::make_shared<IntSet>();
491-
LOGS(_log, LOG_LVL_WARN, "Missing empty chunks info for " << dominantDb);
491+
// LOGS(_log, LOG_LVL_WARN, "Missing empty chunks info for " << dominantDb);
492492
}
493493
}
494494
// FIXME add operator<< for QuerySession
495-
LOGS(_log, LOG_LVL_TRACE, "_qSession: " << _qSession);
495+
// LOGS(_log, LOG_LVL_TRACE, "_qSession: " << _qSession);
496496
if (_qSession->hasChunks()) {
497497
auto areaRestrictors = _qSession->getAreaRestrictors();
498498
auto secIdxRestrictors = _qSession->getSecIdxRestrictors();
@@ -506,15 +506,15 @@ void UserQuerySelect::_setupChunking() {
506506
csv = im->getAllChunks();
507507
}
508508

509-
LOGS(_log, LOG_LVL_TRACE, "Chunk specs: " << util::printable(csv));
509+
// LOGS(_log, LOG_LVL_TRACE, "Chunk specs: " << util::printable(csv));
510510
// Filter out empty chunks
511511
for (qproc::ChunkSpecVector::const_iterator i = csv.begin(), e = csv.end(); i != e; ++i) {
512512
if (eSet->count(i->chunkId) == 0) { // chunk not in empty?
513513
_qSession->addChunk(*i);
514514
}
515515
}
516516
} else {
517-
LOGS(_log, LOG_LVL_TRACE, "No chunks added, QuerySession will add dummy chunk");
517+
// LOGS(_log, LOG_LVL_TRACE, "No chunks added, QuerySession will add dummy chunk");
518518
}
519519
_qSession->setScanInteractive();
520520
}
@@ -578,8 +578,8 @@ void UserQuerySelect::qMetaRegister(std::string const& resultLocation, std::stri
578578
_qMetaQueryId = _queryMetadata->registerQuery(qInfo, tableNames);
579579
_queryIdStr = QueryIdHelper::makeIdStr(_qMetaQueryId);
580580
// Add logging context with query ID
581-
QSERV_LOGCONTEXT_QUERY(_qMetaQueryId);
582-
LOGS(_log, LOG_LVL_DEBUG, "UserQuery registered " << _qSession->getOriginal());
581+
// QSERV_LOGCONTEXT_QUERY(_qMetaQueryId);
582+
// LOGS(_log, LOG_LVL_DEBUG, "UserQuery registered " << _qSession->getOriginal());
583583

584584
// update #QID# with actual query ID
585585
boost::replace_all(_resultLoc, "#QID#", std::to_string(_qMetaQueryId));
@@ -597,7 +597,7 @@ void UserQuerySelect::qMetaRegister(std::string const& resultLocation, std::stri
597597
if (_executive != nullptr) {
598598
_executive->setQueryId(_qMetaQueryId);
599599
} else {
600-
LOGS(_log, LOG_LVL_WARN, "No Executive, assuming invalid query");
600+
// LOGS(_log, LOG_LVL_WARN, "No Executive, assuming invalid query");
601601
}
602602

603603
// Note that ordering is important here, this check must happen after
@@ -626,7 +626,7 @@ void UserQuerySelect::_qMetaUpdateStatus(qmeta::QInfo::QStatus qStatus, size_t r
626626
try {
627627
_queryStatsData->queryStatsTmpRemove(_qMetaQueryId);
628628
} catch (qmeta::SqlError const&) {
629-
LOGS(_log, LOG_LVL_WARN, "queryStatsTmp remove failed " << _queryIdStr);
629+
// LOGS(_log, LOG_LVL_WARN, "queryStatsTmp remove failed " << _queryIdStr);
630630
}
631631
}
632632

@@ -637,7 +637,7 @@ void UserQuerySelect::_qMetaUpdateMessages() {
637637
try {
638638
_queryMetadata->addQueryMessages(_qMetaQueryId, msgStore);
639639
} catch (qmeta::SqlError const& ex) {
640-
LOGS(_log, LOG_LVL_WARN, "UserQuerySelect::_qMetaUpdateMessages failed " << ex.what());
640+
// LOGS(_log, LOG_LVL_WARN, "UserQuerySelect::_qMetaUpdateMessages failed " << ex.what());
641641
}
642642
}
643643

0 commit comments

Comments
 (0)