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

Jump to bottom

Add a retry logic with backoff and jitter #448

Merged
merged 9 commits into from
Sep 28, 2023
Merged

Add a retry logic with backoff and jitter #448

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
9 commits
Select commit Hold shift + click to select a range
4749415
Add a retry logic with backoff and jitter
karan6181 Sep 26, 2023
701b5a7
Merge remote-tracking branch 'upstream/main' into retry
karan6181 Sep 27, 2023
b6a2c64
Moved everything under util file for backward compatibility
karan6181 Sep 27, 2023
4f27cd8
Fixed a bug where if one thread crashes, it should force other thread…
karan6181 Sep 28, 2023
8169783
Merge remote-tracking branch 'upstream/main' into retry
karan6181 Sep 28, 2023
d065c07
Add retry logic to upload function
karan6181 Sep 28, 2023
cfb9151
Upload index.json at the end once all the shard files are uploaded su…
karan6181 Sep 28, 2023
1738284
Terminate the main process if one of the upload thread receives excep…
karan6181 Sep 28, 2023
657848d
Fixed test
karan6181 Sep 28, 2023
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
11 changes: 9 additions & 2 deletions streaming/base/format/base/writer.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -60,6 +60,8 @@ class Writer(ABC):
max_workers (int): Maximum number of threads used to upload output dataset files in
parallel to a remote location. One thread is responsible for uploading one shard
file to a remote location. Default to ``min(32, (os.cpu_count() or 1) + 4)``.
retry (int): Number of times to retry uploading a file to a remote location.
Default to ``2``.
"""

format: str = '' # Name of the format (like "mds", "csv", "json", etc).
Expand Down Expand Up @@ -96,7 +98,7 @@ def __init__(self,

# Validate keyword arguments
invalid_kwargs = [
arg for arg in kwargs.keys() if arg not in ('progress_bar', 'max_workers')
arg for arg in kwargs.keys() if arg not in ('progress_bar', 'max_workers', 'retry')
]
if invalid_kwargs:
raise ValueError(f'Invalid Writer argument(s): {invalid_kwargs} ')
Expand All @@ -112,7 +114,8 @@ def __init__(self,

self.shards = []

self.cloud_writer = CloudUploader.get(out, keep_local, kwargs.get('progress_bar', False))
self.cloud_writer = CloudUploader.get(out, keep_local, kwargs.get('progress_bar', False),
kwargs.get('retry', 2))
self.local = self.cloud_writer.local
self.remote = self.cloud_writer.remote
# `max_workers`: The maximum number of threads that can be executed in parallel.
Expand Down Expand Up @@ -340,6 +343,8 @@ class JointWriter(Writer):
max_workers (int): Maximum number of threads used to upload output dataset files in
parallel to a remote location. One thread is responsible for uploading one shard
file to a remote location. Default to ``min(32, (os.cpu_count() or 1) + 4)``.
retry (int): Number of times to retry uploading a file to a remote location.
Default to ``2``.
"""

def __init__(self,
Expand Down Expand Up @@ -418,6 +423,8 @@ class SplitWriter(Writer):
max_workers (int): Maximum number of threads used to upload output dataset files in
parallel to a remote location. One thread is responsible for uploading one shard
file to a remote location. Default to ``min(32, (os.cpu_count() or 1) + 4)``.
retry (int): Number of times to retry uploading a file to a remote location.
Default to ``2``.
"""

extra_bytes_per_shard = 0
Expand Down
Loading
Loading