Revert "Merge pull request #5553 from chrismeyersfsu/fix-waiting_blocked"

This reverts commit 9ba2122f4f85eecaeb6fa53ac92ea2811b05e83f, reversing
changes made to c3a5f2c96fd85dd1405a8f5c875ffc988dee16a4.
This commit is contained in:
Chris Meyers
2017-02-27 09:38:45 -05:00
parent ebd4064b82
commit dd513621f0
4 changed files with 9 additions and 9 deletions

View File

@@ -930,8 +930,7 @@ class UnifiedJob(PolymorphicModel, PasswordFieldsModel, CommonModelNameNotUnique
def start_celery_task(self, opts, error_callback, success_callback):
task_class = self._get_task_class()
async_result = task_class().apply_async((self.pk,), opts, link_error=error_callback, link=success_callback)
return async_result.id
task_class().apply_async((self.pk,), opts, link_error=error_callback, link=success_callback)
def start(self, error_callback, success_callback, **kwargs):
'''

View File

@@ -60,7 +60,8 @@ class TaskManager():
'''
Tasks that are running and SHOULD have a celery task.
'''
def get_running_tasks(self, status_list=('running',)):
def get_running_tasks(self):
status_list = ('running',)
jobs = JobDict.filter_partial(status=status_list)
inventory_updates = InventoryUpdateDict.filter_partial(status=status_list)
@@ -215,15 +216,15 @@ class TaskManager():
else:
if type(job_obj) is WorkflowJob:
job_obj.status = 'running'
else:
celery_task_id = job_obj.start_celery_task(opts, error_callback=error_handler, success_callback=success_handler)
job_obj.celery_task_id = celery_task_id
job_obj.save()
self.consume_capacity(task)
job_obj.save()
def post_commit():
job_obj.websocket_emit_status(job_obj.status)
if job_obj.status != 'failed':
job_obj.start_celery_task(opts, error_callback=error_handler, success_callback=success_handler)
connection.on_commit(post_commit)

View File

@@ -52,7 +52,7 @@ def run_fail_inconsistent_running_jobs():
# TODO: Failed to contact celery. We should surface this.
return None
all_running_sorted_tasks = scheduler.get_running_tasks(status_list=('running', 'waiting',))
all_running_sorted_tasks = scheduler.get_running_tasks()
scheduler.process_celery_tasks(active_tasks, all_running_sorted_tasks)
except DatabaseError:
return

View File

@@ -665,7 +665,7 @@ class BaseTask(Task):
'''
Run the job/task and capture its output.
'''
instance = self.update_model(pk, status='running')
instance = self.update_model(pk, status='running', celery_task_id='' if self.request.id is None else self.request.id)
instance.websocket_emit_status("running")
status, rc, tb = 'error', None, ''