Use ray instead of luigi#625
Conversation
Introduces a features/ directory with 14 .feature files covering the full observable behavior of ITDE: health checks, environment spawning, parameter validation, reuse, cleanup, SSL certificates, port allocation, test container build/push, data population, CLI/API consistency, Docker runtime selection, Luigi base task lifecycle, Docker image analysis and build, and dependency logging. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
# Conflicts: # exasol_integration_test_docker_environment/test/test_base_task.py # exasol_integration_test_docker_environment/test/test_base_task_cleanup.py # exasol_integration_test_docker_environment/test/test_certificate_docker_build.py # exasol_integration_test_docker_environment/test/test_common_run_task.py # exasol_integration_test_docker_environment/test/test_docker_access_method.py # exasol_integration_test_docker_environment/test/test_docker_build_base.py # exasol_integration_test_docker_environment/test/test_generate_graph_plot.py # exasol_integration_test_docker_environment/test/test_populate_data.py # exasol_integration_test_docker_environment/test/test_test_container_reuse.py # exasol_integration_test_docker_environment/test/test_test_env_reuse.py
2. Updated formatting
| """ | ||
| task_id = task.task_id | ||
|
|
||
| if task.complete(): |
There was a problem hiding this comment.
This might be problematic , because it could be a local copy of the object and might not have attributes populated. task.complete checks the completion target, if I remember correctly. I think, we need to check the stored refs. And, and maybe tasks can't use local attributes anymore, if any worker could rely on them.
| """Wait for all static deps, then drive the task generator.""" | ||
| ray.get(dep_refs) | ||
| if task.complete(): | ||
| return |
There was a problem hiding this comment.
This is only correct, if jumped before out of TaskRegistry.schedule
| _drive(task, registry) | ||
|
|
||
|
|
||
| def _drive(task: Task, registry: ray.actor.ActorHandle) -> None: |
There was a problem hiding this comment.
We need to return the task after we processed run, because the attributes of the Task might have changed and the tasks that depend on this task need get the updated values.
| futures: list[ray.ObjectRef] = [ | ||
| f for t in flatten(yielded) if (f := _schedule(t, registry)) is not None | ||
| ] | ||
| ray.get(futures) |
There was a problem hiding this comment.
We actually need to fetch the task instances the futures point to and update the run futures of the current task with them.
| registry: ray.actor.ActorHandle, | ||
| ) -> None: | ||
| """Wait for all static deps, then drive the task generator.""" | ||
| ray.get(dep_refs) |
There was a problem hiding this comment.
| f for t in tasks if (f := _schedule(t, registry)) is not None | ||
| ] | ||
| try: | ||
| ray.get(futures) |
There was a problem hiding this comment.
2 participants
All Submissions: