Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Issue633 threaded download #708

Draft
wants to merge 3 commits into
base: master
Choose a base branch
from
+21 −5
Draft

Issue633 threaded download #708

Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
f4c5dad
threading of download
HansVRP Jan 14, 2025
789b83c
refactoring
HansVRP Jan 15, 2025
1842808
refactoring
HansVRP Jan 15, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
26 changes: 21 additions & 5 deletions openeo/extra/job_management/__init__.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -606,27 +606,43 @@ def _launch_job(self, start_job, df, i, backend_name, stats: Optional[dict] = No
df.loc[i, "status"] = "skipped"
stats["start_job skipped"] += 1


def on_job_done(self, job: BatchJob, row):
"""
Handles jobs that have finished. Can be overridden to provide custom behaviour.

Default implementation downloads the results into a folder containing the title.
Default implementation runs the download in a separate thread.
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think threading should be opt-in and classic serial downloading should be the default (at least for now).
It's easy to get in trouble with threading (and other parallelism paradigms like async as used in notebooks).


:param job: The job that has finished.
:param row: DataFrame row containing the job's metadata.
"""
# TODO: param `row` is never accessed in this method. Remove it? Is this intended for future use?
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I'd keep this TODO comment, it's not resolved yet as far as I know

_log.info(f"Job {job.job_id} completed. Preparing to handle completion.")

job_metadata = job.describe()
job_dir = self.get_job_dir(job.job_id)
metadata_path = self.get_job_metadata_path(job.job_id)

self.ensure_job_dir_exists(job.job_id)
job.get_results().download_files(target=job_dir)

# Save metadata
_log.info(f"Saving metadata for job {job.job_id} to {metadata_path}")
with metadata_path.open("w", encoding="utf-8") as f:
json.dump(job_metadata, f, ensure_ascii=False)

# Define download logic inline
def download_task():
try:
_log.info(f"Starting download for job {job.job_id} to directory {job_dir}")
job.get_results().download_files(target=job_dir)
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

you use job here as non-local variable from the on_job_done closure. I wonder how thread-safe this is.

A batch job object on it's own is not that complex except for the connection inside it: that's quite a complex object with a lot of state and risk of being thread-unsafe. For robust threaded downloading I would try to minimize the data structures you are sharing from the main thread to the download thread. E.g. just the backend URL (a string), the job id (a string) and a valid access token (a string) might be enough. Or even better: just pass the download URLs if they are signed (and don't require a working access token)

_log.info(f"Successfully downloaded job {job.job_id} results to {job_dir}")
except Exception as e:
_log.error(f"Error downloading job {job.job_id}: {e}")

# Start the download in a separate thread
_log.info(f"Starting download thread for job {job.job_id}")
downloader = Thread(target=download_task, daemon=True)
downloader.start()
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

you should keep track of the Thread objects, to properly .join() them when they are done later


def on_job_error(self, job: BatchJob, row):
"""
Handles jobs that stopped with errors. Can be overridden to provide custom behaviour.
Expand Down Expand Up @@ -722,8 +738,8 @@ def _track_statuses(self, job_db: JobDatabaseInterface, stats: Optional[dict] =

if new_status == "finished":
stats["job finished"] += 1
self.on_job_done(the_job, active.loc[i])

self.on_job_done(the_job, active.loc[i])
if previous_status != "error" and new_status == "error":
stats["job failed"] += 1
self.on_job_error(the_job, active.loc[i])
Expand Down
Loading