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

create a utility method to mark tasks as failed (#3150)

pull/3185/head
SaitTalhaNisanci 2019-11-19 16:35:56 +03:00 committed by GitHub
parent 306d159072
commit 2cb82ae9bd
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
1 changed files with 18 additions and 8 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

26
src/backend/distributed/worker/task_tracker.c
View File

@ -92,6 +92,7 @@ static void ManageWorkerTasksHash(HTAB *WorkerTasksHash);
static void ManageWorkerTask(WorkerTask *workerTask, HTAB *WorkerTasksHash); static void ManageWorkerTask(WorkerTask *workerTask, HTAB *WorkerTasksHash);
static void RemoveWorkerTask(WorkerTask *workerTask, HTAB *WorkerTasksHash); static void RemoveWorkerTask(WorkerTask *workerTask, HTAB *WorkerTasksHash);
static void CreateJobDirectoryIfNotExists(uint64 jobId); static void CreateJobDirectoryIfNotExists(uint64 jobId);
static void MarkWorkerTaskAsFailed(WorkerTask *workerTask);
static int32 ConnectToLocalBackend(const char *databaseName, const char *userName); static int32 ConnectToLocalBackend(const char *databaseName, const char *userName);
@ -931,8 +932,7 @@ ManageWorkerTask(WorkerTask *workerTask, HTAB *WorkerTasksHash)
} }
else else
{ {
workerTask->taskStatus = TASK_FAILED; MarkWorkerTaskAsFailed(workerTask);
workerTask->failureCount++;
MultiClientDisconnect(workerTask->connectionId); MultiClientDisconnect(workerTask->connectionId);
workerTask->connectionId = INVALID_CONNECTION_ID; workerTask->connectionId = INVALID_CONNECTION_ID;
@ -940,8 +940,7 @@ ManageWorkerTask(WorkerTask *workerTask, HTAB *WorkerTasksHash)
} }
else else
{ {
workerTask->taskStatus = TASK_FAILED; MarkWorkerTaskAsFailed(workerTask);
workerTask->failureCount++;
} }
break; break;
@ -962,8 +961,7 @@ ManageWorkerTask(WorkerTask *workerTask, HTAB *WorkerTasksHash)
} }
else if (queryStatus == CLIENT_QUERY_FAILED) else if (queryStatus == CLIENT_QUERY_FAILED)
{ {
workerTask->taskStatus = TASK_FAILED; MarkWorkerTaskAsFailed(workerTask);
workerTask->failureCount++;
} }
else else
{ {
@ -976,8 +974,7 @@ ManageWorkerTask(WorkerTask *workerTask, HTAB *WorkerTasksHash)
} }
else if (resultStatus == CLIENT_RESULT_UNAVAILABLE) else if (resultStatus == CLIENT_RESULT_UNAVAILABLE)
{ {
workerTask->taskStatus = TASK_FAILED; MarkWorkerTaskAsFailed(workerTask);
workerTask->failureCount++;
} }
/* clean up the connection if we are done with the task */ /* clean up the connection if we are done with the task */
@ -1065,6 +1062,19 @@ ManageWorkerTask(WorkerTask *workerTask, HTAB *WorkerTasksHash)
} }
/*
* MarkWorkerTaskAsFailed marks the given worker task as failed
* and increases the failure count. Failure count is used to
* determine if the task should be marked as permanently failed.
*/
static void
MarkWorkerTaskAsFailed(WorkerTask *workerTask)
{
workerTask->taskStatus = TASK_FAILED;
workerTask->failureCount++;
}
/* Wrapper function to remove the worker task from the shared hash. */ /* Wrapper function to remove the worker task from the shared hash. */
static void static void
RemoveWorkerTask(WorkerTask *workerTask, HTAB *WorkerTasksHash) RemoveWorkerTask(WorkerTask *workerTask, HTAB *WorkerTasksHash)