Skip to content

Commit c2263f7

Browse files
jgates108jgates108
committed
Minor changes for integration tests.
1 parent 56677ed commit c2263f7

File tree

5 files changed

+63
-22
lines changed

5 files changed

+63
-22
lines changed

‎core/modules/ccontrol/UserQuerySelect.cc

+28-19
Original file line numberDiff line numberDiff line change
@@ -368,31 +368,24 @@ void UserQuerySelect::submit() {
368368

369369
if (!uberJobsEnabled) {
370370
std::function<void(util::CmdData*)> funcBuildJob =
371-
[this, sequence, job{move(job)}](util::CmdData*) { // references in captures cause races
371+
//&&&[this, sequence, job{move(job)}](util::CmdData*) { // references in captures cause races
372+
[this, job{move(job)}](util::CmdData*) { // references in captures cause races
372373
QSERV_LOGCONTEXT_QUERY(_qMetaQueryId);
373374
job->runJob();
374375
};
375376
auto cmd = std::make_shared<qdisp::PriorityCommand>(funcBuildJob);
376377
_executive->queueJobStart(cmd);
377-
378378
}
379379
++sequence;
380380
}
381381

382382
if (uberJobsEnabled) {
383383
vector<qdisp::UberJob::Ptr> uberJobs;
384-
/* &&&
385-
vector<czar::WorkerResource> workers; // &&& delete and replace with a real list of workers
386-
throw Bug("&&&NEED_CODE to find all workers"); // workers = all workers found in database
387-
for (auto&& worker:workers) {
388-
worker.fillChunkIdSet();
389-
}
390-
*/
391384

392385
czar::WorkerResources workerResources;
393-
workerResources.setMonoNodeTest(); //&&& TODO:UJ only good for mono-node test.
386+
workerResources.setMonoNodeTest(); //&&& TODO:UJ only good for mono-node test. Need a real list of workers and their chunks. ******
394387

395-
// &&& make a map of all jobs in the executive.
388+
// Make a map of all jobs in the executive.
396389
// &&& TODO:UJ for now, just using ints. At some point, need to check that ResourceUnit databases can be found for all databases in the query
397390
qdisp::Executive::ChunkIdJobMapType chunksInQuery = _executive->getChunkJobMapAndInvalidate();
398391

@@ -405,13 +398,6 @@ void UserQuerySelect::submit() {
405398
/// make a map<worker, deque<chunkId> that will be destroyed as chunks are checked/used
406399
map<string, deque<int>> tmpWorkerList = workerResources.getDequesFor(dbName);
407400

408-
/* &&&
409-
list<std::reference_wrapper<czar::WorkerResource>> tmpWorkerList;
410-
for(auto&& worker:workers) {
411-
tmpWorkerList.push_back(worker);
412-
}
413-
*/
414-
415401
// TODO:UJ So UberJobIds don't conflict with chunk numbers or jobIds, start at a large number.
416402
// This could use some refinement.
417403
int uberJobId = qdisp::UberJob::getFirstIdNumber();
@@ -461,26 +447,49 @@ void UserQuerySelect::submit() {
461447
workerIter = tmpWorkerList.begin();
462448
}
463449
}
450+
LOGS(_log, LOG_LVL_INFO, "&&& submit m");
464451
_executive->addUberJobs(uberJobs);
452+
LOGS(_log, LOG_LVL_INFO, "&&& submit n");
465453
for (auto&& uJob:uberJobs) {
454+
LOGS(_log, LOG_LVL_INFO, "&&& submit o");
466455
uJob->runUberJob();
456+
LOGS(_log, LOG_LVL_INFO, "&&& submit p");
467457
}
468-
_executive->startRemainingJobs();
458+
LOGS(_log, LOG_LVL_INFO, "&&& submit q");
459+
// If any chunks in the query were not found on a worker's list, run them individually.
460+
//&&&_executive->startRemainingJobs(chunksInQuery); //&&& delete func in Executive.
461+
for (auto& ciq:chunksInQuery) {
462+
qdisp::JobQuery* jqRaw = ciq.second;
463+
qdisp::JobQuery::Ptr job = _executive->getSharedPtrForRawJobPtr(jqRaw);
464+
std::function<void(util::CmdData*)> funcBuildJob =
465+
[this, job{move(job)}](util::CmdData*) { // references in captures cause races
466+
QSERV_LOGCONTEXT_QUERY(_qMetaQueryId);
467+
job->runJob();
468+
};
469+
auto cmd = std::make_shared<qdisp::PriorityCommand>(funcBuildJob);
470+
_executive->queueJobStart(cmd);
471+
}
472+
473+
LOGS(_log, LOG_LVL_INFO, "&&& submit r");
469474
}
470475

471476
// attempt to restore original thread priority, requires root
472477
if (increaseThreadPriority) {
473478
threadPriority.restoreOriginalValues();
474479
}
480+
LOGS(_log, LOG_LVL_INFO, "&&& submit s");
475481

476482
LOGS(_log, LOG_LVL_DEBUG, "total jobs in query=" << sequence);
477483
_executive->waitForAllJobsToStart();
484+
LOGS(_log, LOG_LVL_INFO, "&&& submit t");
478485

479486
// we only care about per-chunk info for ASYNC queries
480487
if (_async) {
488+
LOGS(_log, LOG_LVL_INFO, "&&& submit u");
481489
std::lock_guard<std::mutex> lock(chunksMtx);
482490
_qMetaAddChunks(chunks);
483491
}
492+
LOGS(_log, LOG_LVL_INFO, "&&& submit v");
484493
}
485494

486495

‎core/modules/qdisp/Executive.cc

+15-1
Original file line numberDiff line numberDiff line change
@@ -246,6 +246,20 @@ bool Executive::_addJobToMap(JobQuery::Ptr const& job) {
246246
return res;
247247
}
248248

249+
250+
JobQuery::Ptr Executive::getSharedPtrForRawJobPtr(JobQuery* jqRaw) {
251+
assert(jqRaw != nullptr);
252+
int jobId = jqRaw->getIdInt();
253+
lock_guard<recursive_mutex> lockJobMap(_jobMapMtx);
254+
auto iter = _jobMap.find(jobId);
255+
if (iter == _jobMap.end()) {
256+
throw Bug("Could not find the entry for jobId=" + to_string(jobId));
257+
}
258+
JobQuery::Ptr jq = iter->second;
259+
return jq;
260+
}
261+
262+
249263
bool Executive::join() {
250264
// To join, we make sure that all of the chunks added so far are complete.
251265
// Check to see if _requesters is empty, if not, then sleep on a condition.
@@ -616,7 +630,7 @@ bool Executive::startUberJob(UberJob::Ptr const& uJob) {
616630
}
617631

618632

619-
void Executive::startRemainingJobs() {
633+
void Executive::startRemainingJobs(ChunkIdJobMapType& remainingChunks) {
620634
throw Bug("&&&NEED_CODE executive start remaining jobs");
621635
}
622636

‎core/modules/qdisp/Executive.h

+2-1
Original file line numberDiff line numberDiff line change
@@ -141,12 +141,13 @@ class Executive : public std::enable_shared_from_this<Executive> {
141141
bool startQuery(std::shared_ptr<JobQuery> const& jobQuery);
142142

143143
/// Start any jobs that were not started as part of UberJobs.
144-
void startRemainingJobs();
144+
void startRemainingJobs(ChunkIdJobMapType& remainingJobs); // &&& delete
145145

146146
///&&& TODO:UJ UberJob
147147
void addUberJobs(std::vector<std::shared_ptr<UberJob>> const& jobsToAdd);
148148
ChunkIdJobMapType& getChunkJobMapAndInvalidate();
149149
bool startUberJob(std::shared_ptr<UberJob> const& uJob);
150+
std::shared_ptr<JobQuery> getSharedPtrForRawJobPtr(JobQuery* jqRaw);
150151

151152
private:
152153
Executive(ExecutiveConfig const& c, std::shared_ptr<MessageStore> const& ms,

‎core/modules/qdisp/UberJob.cc

+17-1
Original file line numberDiff line numberDiff line change
@@ -60,6 +60,7 @@ UberJob::UberJob(Executive::Ptr const& executive,
6060
: JobBase(), _executive(executive), _respHandler(respHandler), _queryId(queryId), _uberJobId(uberJobId),
6161
_czarId(czarId), _idStr("QID=" + to_string(_queryId) + ":uber=" + to_string(uberJobId)) {
6262
_qdispPool = executive->getQdispPool();
63+
_jobStatus = make_shared<JobStatus>();
6364
}
6465

6566

@@ -77,6 +78,7 @@ bool UberJob::addJob(JobQuery* job) {
7778

7879
bool UberJob::runUberJob() {
7980
QSERV_LOGCONTEXT_QUERY_JOB(getQueryId(), getIdInt());
81+
LOGS(_log, LOG_LVL_INFO, "&&& runUberJob a");
8082
// Build the uberjob payload.
8183
// TODO:UJ For simplicity in the first pass, just make a TaskMsg for each Job and append it to the UberJobMsg.
8284
// This is terribly inefficient and should be replaced by using a template and list of chunks that the
@@ -86,22 +88,32 @@ bool UberJob::runUberJob() {
8688
proto::UberJobMsg* ujMsg = google::protobuf::Arena::CreateMessage<proto::UberJobMsg>(&arena);
8789
ujMsg->set_queryid(getQueryId());
8890
ujMsg->set_czarid(_czarId);
91+
LOGS(_log, LOG_LVL_INFO, "&&& runUberJob b");
8992
for (auto&& job:_jobs) {
93+
LOGS(_log, LOG_LVL_INFO, "&&& runUberJob b1");
9094
proto::TaskMsg* tMsg = ujMsg->add_taskmsgs();
95+
LOGS(_log, LOG_LVL_INFO, "&&& runUberJob b2");
9196
job->getDescription()->fillTaskMsg(tMsg);
97+
LOGS(_log, LOG_LVL_INFO, "&&& runUberJob b3");
9298
}
99+
LOGS(_log, LOG_LVL_INFO, "&&& runUberJob c");
93100
ujMsg->SerializeToString(&_payload);
101+
LOGS(_log, LOG_LVL_INFO, "&&& runUberJob d");
94102
}
95103

104+
LOGS(_log, LOG_LVL_INFO, "&&& runUberJob e");
96105
auto executive = _executive.lock();
97106
if (executive == nullptr) {
98107
LOGS(_log, LOG_LVL_ERROR, "runUberJob failed executive==nullptr");
99108
return false;
100109
}
110+
LOGS(_log, LOG_LVL_INFO, "&&& runUberJob f");
101111
bool cancelled = executive->getCancelled();
102112
bool handlerReset = _respHandler->reset();
103113
bool started = _started.exchange(true);
114+
LOGS(_log, LOG_LVL_INFO, "&&& runUberJob g");
104115
if (!cancelled && handlerReset && !started) {
116+
LOGS(_log, LOG_LVL_INFO, "&&& runUberJob h");
105117
auto criticalErr = [this, &executive](std::string const& msg) {
106118
LOGS(_log, LOG_LVL_ERROR, msg << " "
107119
<< *this << " Canceling user query!");
@@ -114,16 +126,20 @@ bool UberJob::runUberJob() {
114126
return false;
115127
}
116128

129+
LOGS(_log, LOG_LVL_INFO, "&&& runUberJob i");
117130
// At this point we are all set to actually run the queries. We create a
118131
// a shared pointer to this object to prevent it from escaping while we
119132
// are trying to start this whole process. We also make sure we record
120133
// whether or not we are in SSI as cancellation handling differs.
121134
//
122-
LOGS(_log, LOG_LVL_TRACE, "runJob calls StartQuery()");
135+
LOGS(_log, LOG_LVL_TRACE, "runUberJob calls StartQuery()");
123136
std::shared_ptr<UberJob> uJob(dynamic_pointer_cast<UberJob>(shared_from_this()));
137+
LOGS(_log, LOG_LVL_INFO, "&&& runUberJob j");
124138
_inSsi = true;
125139
if (executive->startUberJob(uJob)) {
140+
LOGS(_log, LOG_LVL_INFO, "&&& runUberJob k");
126141
_jobStatus->updateInfo(_idStr, JobStatus::REQUEST);
142+
LOGS(_log, LOG_LVL_INFO, "&&& runUberJob l");
127143
return true;
128144
}
129145
_inSsi = false;

‎core/modules/qproc/TaskMsgFactory.cc

+1
Original file line numberDiff line numberDiff line change
@@ -114,6 +114,7 @@ bool TaskMsgFactory::fillTaskMsg(proto::TaskMsg* taskMsg, ChunkQuerySpec const&
114114
_addFragment(*taskMsg, resultTable, chunkQuerySpec.subChunkTables,
115115
chunkQuerySpec.subChunkIds, chunkQuerySpec.queries);
116116
}
117+
LOGS(_log, LOG_LVL_WARN, "&&& _makeMsg end chunkId=" << chunkQuerySpec.chunkId);
117118
return true;
118119
}
119120

0 commit comments

Comments
 (0)