From cb54f893a50d469bf98cb0a3c0f692157aa18ad2 Mon Sep 17 00:00:00 2001 From: xjules Date: Wed, 20 Nov 2019 17:22:22 +0100 Subject: [PATCH] Add ThreadPoolExecutor to workflowrunner - ThreadPoolExecutor replacing Thread in workflow runner - mock worflow.run to throw exception --- python/res/job_queue/workflow_runner.py | 29 +++-- .../res/job_queue/test_workflow_runner.py | 108 ++++++++++-------- 2 files changed, 85 insertions(+), 52 deletions(-) diff --git a/python/res/job_queue/workflow_runner.py b/python/res/job_queue/workflow_runner.py index bb1231d86a..76d0759220 100644 --- a/python/res/job_queue/workflow_runner.py +++ b/python/res/job_queue/workflow_runner.py @@ -1,7 +1,7 @@ -from threading import Thread from res.job_queue import Workflow from res.util.substitution_list import SubstitutionList - +from concurrent.futures import ThreadPoolExecutor +from concurrent import futures class WorkflowRunner(object): def __init__(self, workflow, ert=None, context=None): @@ -20,12 +20,20 @@ def __init__(self, workflow, ert=None, context=None): self.__context = context self.__workflow_result = None + self.__workflow_executor = None + self.__workflow_job = None + self.__workflow_exception = None + + def __enter__(self): + self.run() + return self + + def __exit__(self, type, value, traceback): + self.wait() def run(self): - workflow_thread = Thread(name="ert_gui_workflow_thread") - workflow_thread.setDaemon(True) - workflow_thread.run = self.__runWorkflow - workflow_thread.start() + self.__workflow_executor = ThreadPoolExecutor(max_workers=1) + self.__workflow_job = self.__workflow_executor.submit(self.__runWorkflow) def __runWorkflow(self): self.__workflow_result = self.__workflow.run(self.__ert, context=self.__context) @@ -41,9 +49,16 @@ def isCancelled(self): def cancel(self): if self.isRunning(): self.__workflow.cancel() + self.wait() + + def exception(self): + if self.__workflow_job is not None: + return self.__workflow_job._exception + return None def wait(self): - self.__workflow.wait() + # This returns a tuple (done, pending), since we run only one job we don't need to use it + _, _ = futures.wait([self.__workflow_job], timeout=None, return_when=futures.FIRST_EXCEPTION) def workflowResult(self): """ @rtype: bool or None """ diff --git a/python/tests/res/job_queue/test_workflow_runner.py b/python/tests/res/job_queue/test_workflow_runner.py index 7afc12b622..ea023493d5 100644 --- a/python/tests/res/job_queue/test_workflow_runner.py +++ b/python/tests/res/job_queue/test_workflow_runner.py @@ -5,7 +5,11 @@ from ecl.util.test import TestAreaContext from res.util.substitution_list import SubstitutionList from .workflow_common import WorkflowCommon - +import sys +if sys.version_info >= (3, 3): + from unittest.mock import patch +else: + from mock import patch class WorkflowRunnerTest(ResTest): @@ -26,32 +30,31 @@ def test_workflow_thread_cancel_ert_script(self): self.assertFalse(workflow_runner.isRunning()) - workflow_runner.run() - - self.assertIsNone(workflow_runner.workflowResult()) + with workflow_runner: + self.assertIsNone(workflow_runner.workflowResult()) - time.sleep(1) # wait for workflow to start - self.assertTrue(workflow_runner.isRunning()) - self.assertFileExists("wait_started_0") + time.sleep(1) # wait for workflow to start + self.assertTrue(workflow_runner.isRunning()) + self.assertFileExists("wait_started_0") - time.sleep(1) # wait for first job to finish + time.sleep(1) # wait for first job to finish - workflow_runner.cancel() - time.sleep(1) # wait for cancel to take effect - self.assertFileExists("wait_finished_0") + workflow_runner.cancel() + time.sleep(1) # wait for cancel to take effect + self.assertFileExists("wait_finished_0") - self.assertFileExists("wait_started_1") - self.assertFileExists("wait_cancelled_1") - self.assertFileDoesNotExist("wait_finished_1") + self.assertFileExists("wait_started_1") + self.assertFileExists("wait_cancelled_1") + self.assertFileDoesNotExist("wait_finished_1") - self.assertTrue(workflow_runner.isCancelled()) + self.assertTrue(workflow_runner.isCancelled()) - workflow_runner.wait() # wait for runner to complete + workflow_runner.wait() # wait for runner to complete - self.assertFileDoesNotExist("wait_started_2") - self.assertFileDoesNotExist("wait_cancelled_2") - self.assertFileDoesNotExist("wait_finished_2") + self.assertFileDoesNotExist("wait_started_2") + self.assertFileDoesNotExist("wait_cancelled_2") + self.assertFileDoesNotExist("wait_finished_2") @@ -73,29 +76,45 @@ def test_workflow_thread_cancel_external(self): self.assertFalse(workflow_runner.isRunning()) - workflow_runner.run() + with workflow_runner: + time.sleep(1) # wait for workflow to start + self.assertTrue(workflow_runner.isRunning()) + self.assertFileExists("wait_started_0") + + time.sleep(1) # wait for first job to finish + + workflow_runner.cancel() + time.sleep(1) # wait for cancel to take effect + self.assertFileExists("wait_finished_0") + + self.assertFileExists("wait_started_1") + self.assertFileDoesNotExist("wait_finished_1") - time.sleep(1) # wait for workflow to start - self.assertTrue(workflow_runner.isRunning()) - self.assertFileExists("wait_started_0") + self.assertTrue(workflow_runner.isCancelled()) - time.sleep(1) # wait for first job to finish + workflow_runner.wait() # wait for runner to complete - workflow_runner.cancel() - time.sleep(1) # wait for cancel to take effect - self.assertFileExists("wait_finished_0") + self.assertFileDoesNotExist("wait_started_2") + self.assertFileDoesNotExist("wait_cancelled_2") + self.assertFileDoesNotExist("wait_finished_2") - self.assertFileExists("wait_started_1") - self.assertFileDoesNotExist("wait_finished_1") - self.assertTrue(workflow_runner.isCancelled()) + def test_workflow_failed_job(self): + with TestAreaContext("python/job_queue/workflow_runner_fails") as work_area: + WorkflowCommon.createExternalDumpJob() - workflow_runner.wait() # wait for runner to complete + joblist = WorkflowJoblist() + self.assertTrue(joblist.addJobFromFile("DUMP", "dump_failing_job")) + workflow = Workflow("dump_workflow", joblist) + self.assertEqual(len(workflow), 2) - self.assertFileDoesNotExist("wait_started_2") - self.assertFileDoesNotExist("wait_cancelled_2") - self.assertFileDoesNotExist("wait_finished_2") + workflow_runner = WorkflowRunner(workflow, ert=None, context=SubstitutionList()) + self.assertFalse(workflow_runner.isRunning()) + with patch.object(Workflow, 'run', side_effect=Exception('mocked workflow error')), \ + workflow_runner: + workflow_runner.wait() + self.assertNotEqual(workflow_runner.exception(), None) def test_workflow_success(self): with TestAreaContext("python/job_queue/workflow_runner_fast") as work_area: @@ -113,17 +132,16 @@ def test_workflow_success(self): workflow_runner = WorkflowRunner(workflow, ert=None, context=SubstitutionList()) self.assertFalse(workflow_runner.isRunning()) + with workflow_runner: + time.sleep(1) # wait for workflow to start + workflow_runner.wait() - workflow_runner.run() - time.sleep(1) # wait for workflow to start - workflow_runner.wait() - - self.assertFileExists("wait_started_0") - self.assertFileDoesNotExist("wait_cancelled_0") - self.assertFileExists("wait_finished_0") + self.assertFileExists("wait_started_0") + self.assertFileDoesNotExist("wait_cancelled_0") + self.assertFileExists("wait_finished_0") - self.assertFileExists("wait_started_1") - self.assertFileDoesNotExist("wait_cancelled_1") - self.assertFileExists("wait_finished_1") + self.assertFileExists("wait_started_1") + self.assertFileDoesNotExist("wait_cancelled_1") + self.assertFileExists("wait_finished_1") - self.assertTrue(workflow_runner.workflowResult()) + self.assertTrue(workflow_runner.workflowResult())