external
/
citus
mirror of https://github.com/citusdata/citus.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

refactor inner loop of background worker thing

background-job-details
Nils Dijk 2022-07-27 14:54:45 +02:00 committed by Jelte Fennema
parent 5a2ec73475
commit 80e2edf1a5
1 changed files with 41 additions and 38 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

41
src/backend/distributed/utils/maintenanced.c
View File

@ -912,28 +912,35 @@ RebalanceJobsBackgroundWorkerMain(Datum arg)
RebalanceJob *job = GetRunableRebalanceJob(); RebalanceJob *job = GetRunableRebalanceJob();
MemoryContextSwitchTo(oldContext); MemoryContextSwitchTo(oldContext);
if (job) if (!job)
{ {
ereport(LOG, (errmsg("found job with jobid: %ld", job->jobid)));
MemoryContext savedContext = CurrentMemoryContext;
UpdateJobStatus(job, REBALANCE_JOB_STATUS_RUNNING);
PopActiveSnapshot(); PopActiveSnapshot();
CommitTransactionCommand(); CommitTransactionCommand();
StartTransactionCommand(); hasJobs = false;
PushActiveSnapshot(GetTransactionSnapshot()); break;
}
BeginInternalSubTransaction(NULL); ereport(LOG, (errmsg("found job with jobid: %ld", job->jobid)));
/* Update job status to indicate it is running */
UpdateJobStatus(job, REBALANCE_JOB_STATUS_RUNNING);
PopActiveSnapshot();
CommitTransactionCommand();
MemoryContext savedContext = CurrentMemoryContext;
PG_TRY(); PG_TRY();
{ {
StartTransactionCommand();
PushActiveSnapshot(GetTransactionSnapshot());
if (ExecuteRebalanceJob(job)) if (ExecuteRebalanceJob(job))
{ {
UpdateJobStatus(job, REBALANCE_JOB_STATUS_DONE); UpdateJobStatus(job, REBALANCE_JOB_STATUS_DONE);
}
ReleaseCurrentSubTransaction(); PopActiveSnapshot();
CommitTransactionCommand();
}
} }
PG_CATCH(); PG_CATCH();
{ {
@ -942,10 +949,14 @@ RebalanceJobsBackgroundWorkerMain(Datum arg)
ErrorData *edata = CopyErrorData(); ErrorData *edata = CopyErrorData();
FlushErrorState(); FlushErrorState();
RollbackAndReleaseCurrentSubTransaction(); StartTransactionCommand();
PushActiveSnapshot(GetTransactionSnapshot());
UpdateJobError(job, edata); UpdateJobError(job, edata);
PopActiveSnapshot();
CommitTransactionCommand();
FreeErrorData(edata); FreeErrorData(edata);
edata = NULL; edata = NULL;
@ -953,16 +964,8 @@ RebalanceJobsBackgroundWorkerMain(Datum arg)
} }
PG_END_TRY(); PG_END_TRY();
} }
else
{
hasJobs = false;
}
} }
PopActiveSnapshot();
CommitTransactionCommand();
ProcessCompletedNotifies();
}
MemoryContextSwitchTo(oldContextPerJob); MemoryContextSwitchTo(oldContextPerJob);
MemoryContextDelete(perJobContext); MemoryContextDelete(perJobContext);
} }