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

Fix multi_task_string_size sometimes leaking intermediate files

pull/3453/head
Philip Dubé 2020-01-28 00:07:28 +00:00
parent 2e096d4eb9
commit 025cb94159
5 changed files with 14 additions and 10 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
src/backend/distributed/executor/intermediate_results.c
View File

@ -558,11 +558,10 @@ char *
CreateIntermediateResultsDirectory(void)
{
char *resultDirectory = IntermediateResultsDirectory();
int makeOK = 0;
if (!CreatedResultsDirectory)
{
makeOK = mkdir(resultDirectory, S_IRWXU);
int makeOK = mkdir(resultDirectory, S_IRWXU);
if (makeOK != 0)
{
if (errno == EEXIST)

8
src/backend/distributed/executor/multi_task_tracker_executor.c
View File

@ -2844,8 +2844,8 @@ TrackerHashCleanupJob(HTAB *taskTrackerHash, Task *jobCleanupTask)
else if (timedOut)
{
ereport(WARNING, (errmsg("could not receive response for cleanup "
"query status for job " UINT64_FORMAT " "
"on node \"%s:%u\" with status %d",
"query status for job " UINT64_FORMAT
" on node \"%s:%u\" with status %d",
jobId,
nodeName, nodePort, (int) queryStatus),
errhint("Manually clean job resources on node "
@ -2863,8 +2863,8 @@ TrackerHashCleanupJob(HTAB *taskTrackerHash, Task *jobCleanupTask)
{
/* CLIENT_RESULT_UNAVAILABLE is returned if the connection failed somehow */
ereport(WARNING, (errmsg("could not receive response for cleanup query "
"result for job " UINT64_FORMAT " on node "
"\"%s:%u\" with status %d",
"result for job " UINT64_FORMAT
" on node \"%s:%u\" with status %d",
jobId, nodeName,
nodePort, (int) resultStatus),
errhint("Manually clean job resources on node "

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

@ -1021,6 +1021,12 @@ ManageWorkerTask(WorkerTask *workerTask, HTAB *WorkerTasksHash)
workerTask->connectionId = INVALID_CONNECTION_ID;
}
if (workerTask->taskId == JOB_CLEANUP_TASK_ID)
{
StringInfo jobDirectoryName = JobDirectoryName(workerTask->jobId);
CitusRemoveDirectory(jobDirectoryName->data);
}
workerTask->taskStatus = TASK_TO_REMOVE;
break;
}

4
src/backend/distributed/worker/task_tracker_protocol.c
View File

@ -188,6 +188,7 @@ task_tracker_cleanup_job(PG_FUNCTION_ARGS)
CheckCitusVersion(ERROR);
StringInfo jobSchemaName = JobSchemaName(jobId);
StringInfo jobDirectoryName = JobDirectoryName(jobId);
/*
* We'll keep this lock for a while, but that's ok because nothing
@ -230,7 +231,6 @@ task_tracker_cleanup_job(PG_FUNCTION_ARGS)
* schema drop call can block if another process is creating the schema or
* writing to a table within the schema.
*/
StringInfo jobDirectoryName = JobDirectoryName(jobId);
CitusRemoveDirectory(jobDirectoryName->data);
RemoveJobSchema(jobSchemaName);
@ -450,7 +450,7 @@ CleanupTask(WorkerTask *workerTask)
return;
}
/* remove the task from the shared hash */
/* remove task from the shared hash */
WorkerTask *taskRemoved = hash_search(TaskTrackerTaskHash, hashKey, HASH_REMOVE,
NULL);
if (taskRemoved == NULL)

3
src/test/regress/multi_schedule
View File

@ -326,6 +326,5 @@ test: multi_deparse_function multi_deparse_procedure
# ---------
# test that no tests leaked intermediate results. This should always be last
# Causes random test failures so commented out for now
# ---------
# test: ensure_no_intermediate_data_leak
test: ensure_no_intermediate_data_leak