Skip to content

Commit

Permalink
Merge pull request #1604 from Avaiga/fix/#1602-wait-with-no-timeout-s…
Browse files Browse the repository at this point in the history
…hould-wait-indefinitely

Fix/#1602 - Wait with no timeout should wait indefinitely
  • Loading branch information
trgiangdo authored Jul 31, 2024
2 parents eee7269 + 3a18d68 commit d087955
Show file tree
Hide file tree
Showing 7 changed files with 20 additions and 12 deletions.
14 changes: 9 additions & 5 deletions taipy/core/_orchestrator/_orchestrator.py
Original file line number Diff line number Diff line change
Expand Up @@ -65,7 +65,8 @@ def submit(
wait (bool): Wait for the orchestrated jobs created from the scenario or sequence submission to be
finished in asynchronous mode.
timeout (Union[float, int]): The optional maximum number of seconds to wait for the jobs to be finished
before returning.
before returning.<br/>
If not provided and *wait* is True, the function waits indefinitely.
**properties (dict[str, any]): A key worded variable length list of user additional arguments
that will be stored within the `Submission^`. It can be accessed via `Submission.properties^`.
Returns:
Expand Down Expand Up @@ -97,7 +98,7 @@ def submit(
if Config.job_config.is_development:
cls._check_and_execute_jobs_if_development_mode()
elif wait:
cls._wait_until_job_finished(jobs, timeout=timeout or 0)
cls._wait_until_job_finished(jobs, timeout)
return submission

@classmethod
Expand All @@ -119,7 +120,8 @@ def submit_task(
wait (bool): Wait for the orchestrated job created from the task submission to be finished
in asynchronous mode.
timeout (Union[float, int]): The optional maximum number of seconds to wait for the job
to be finished before returning.
to be finished before returning.<br/>
If not provided and *wait* is True, the function waits indefinitely.
**properties (dict[str, any]): A key worded variable length list of user additional arguments
that will be stored within the `Submission^`. It can be accessed via `Submission.properties^`.
Returns:
Expand All @@ -145,7 +147,7 @@ def submit_task(
cls._check_and_execute_jobs_if_development_mode()
else:
if wait:
cls._wait_until_job_finished(job, timeout=timeout or 0)
cls._wait_until_job_finished(job, timeout)
return submission

@classmethod
Expand Down Expand Up @@ -199,9 +201,11 @@ def _orchestrate_job_to_run_or_block(cls, jobs: List[Job]) -> None:
cls.jobs_to_run.put(job)

@classmethod
def _wait_until_job_finished(cls, jobs: Union[List[Job], Job], timeout: float = 0) -> None:
def _wait_until_job_finished(cls, jobs: Union[List[Job], Job], timeout: Optional[Union[float, int]] = None) -> None:
# Note: this method should be prefixed by two underscores, but it has only one, so it can be mocked in tests.
def __check_if_timeout(st, to):
if to is None:
return True
return (datetime.now() - st).seconds < to

start = datetime.now()
Expand Down
3 changes: 2 additions & 1 deletion taipy/core/scenario/scenario.py
Original file line number Diff line number Diff line change
Expand Up @@ -608,7 +608,8 @@ def submit(
wait (bool): Wait for the orchestrated jobs created from the scenario submission to be finished in
asynchronous mode.
timeout (Union[float, int]): The optional maximum number of seconds to wait for the jobs to be finished
before returning.
before returning.<br/>
If not provided and *wait* is True, the function waits indefinitely.
**properties (dict[str, any]): A keyworded variable length list of additional arguments.
Returns:
A `Submission^` containing the information of the submission.
Expand Down
3 changes: 2 additions & 1 deletion taipy/core/sequence/sequence.py
Original file line number Diff line number Diff line change
Expand Up @@ -311,7 +311,8 @@ def submit(
wait (bool): Wait for the orchestrated jobs created from the sequence submission to be finished
in asynchronous mode.
timeout (Union[float, int]): The maximum number of seconds to wait for the jobs to be finished before
returning.
returning.<br/>
If not provided and *wait* is True, the function waits indefinitely.
**properties (dict[str, any]): A keyworded variable length list of additional arguments.
Returns:
A `Submission^` containing the information of the submission.
Expand Down
3 changes: 2 additions & 1 deletion taipy/core/taipy.py
Original file line number Diff line number Diff line change
Expand Up @@ -240,7 +240,8 @@ def submit(
wait (bool): Wait for the orchestrated jobs created from the submission to be finished
in asynchronous mode.
timeout (Union[float, int]): The optional maximum number of seconds to wait
for the jobs to be finished before returning.
for the jobs to be finished before returning.<br/>
If not provided and *wait* is True, the function waits indefinitely.
**properties (dict[str, any]): A key-worded variable length list of user additional arguments
that will be stored within the `Submission^`. It can be accessed via `Submission.properties^`.
Expand Down
3 changes: 2 additions & 1 deletion taipy/core/task/task.py
Original file line number Diff line number Diff line change
Expand Up @@ -231,7 +231,8 @@ def submit(
wait (bool): Wait for the orchestrated job created from the task submission to be finished in asynchronous
mode.
timeout (Union[float, int]): The maximum number of seconds to wait for the job to be finished before
returning.
returning.<br/>
If not provided and *wait* is True, the function waits indefinitely.
**properties (dict[str, any]): A keyworded variable length list of additional arguments.
Returns:
A `Submission^` containing the information of the submission.
Expand Down
4 changes: 2 additions & 2 deletions tests/core/_orchestrator/test_orchestrator__submit.py
Original file line number Diff line number Diff line change
Expand Up @@ -320,7 +320,7 @@ def test_submit_scenario_with_callbacks_and_force_and_wait():
assert jobs[2]._subscribers[0].__code__ == nothing.__code__
assert jobs[2]._subscribers[1].__code__ == _Orchestrator._update_submission_status.__code__
assert jobs[2]._subscribers[2].__code__ == _Orchestrator._on_status_change.__code__
mck.assert_called_once_with(jobs, timeout=5)
mck.assert_called_once_with(jobs, 5)


def test_submit_sequence_development_mode():
Expand Down Expand Up @@ -475,7 +475,7 @@ def test_submit_sequence_with_callbacks_and_force_and_wait():

with mock.patch("taipy.core._orchestrator._orchestrator._Orchestrator._wait_until_job_finished") as mck:
jobs = orchestrator.submit(scenario, callbacks=[nothing], force=True, wait=True, timeout=5).jobs
mck.assert_called_once_with(jobs, timeout=5)
mck.assert_called_once_with(jobs, 5)

# jobs are created in a specific order and are correct
assert len(jobs) == 4
Expand Down
2 changes: 1 addition & 1 deletion tests/core/_orchestrator/test_orchestrator__submit_task.py
Original file line number Diff line number Diff line change
Expand Up @@ -236,4 +236,4 @@ def test_submit_task_with_callbacks_and_force_and_wait():
assert job._subscribers[1].__code__ == _Orchestrator._update_submission_status.__code__
assert job._subscribers[2].__code__ == _Orchestrator._on_status_change.__code__

mck.assert_called_once_with(job, timeout=2)
mck.assert_called_once_with(job, 2)

0 comments on commit d087955

Please sign in to comment.