Cap max number of workers in weather-dl. (#170)

* Capped max number of workers in weather-dl tool.

* Corrected doc string of class PipelineArgs weather-dl.

* updated weather-dl version to 0.1.10

weather_dl/download_pipeline/pipeline.py

@@ -24,6 +24,7 @@
 from apache_beam.options.pipeline_options import (
     PipelineOptions,
     StandardOptions,
+    WorkerOptions,
 )
 
 from .clients import CLIENTS
@@ -69,7 +70,7 @@ class PipelineArgs:
         client_name: The type of download client (e.g. Copernicus, Mars, or a fake).
         store: A Store, which is responsible for where downloads end up.
         manifest: A Manifest, which records download progress.
-        subsections: A
+        num_requesters_per_key: Number of requests per subsection (license).
     """
     known_args: argparse.Namespace
     pipeline_options: PipelineOptions
@@ -88,6 +89,11 @@ def pipeline(args: PipelineArgs) -> None:
 
     subsections = get_subsections(args.config)
 
+    # capping the max number of workers to N i.e. possible simultaneous requests + fudge factor
+    max_num_workers = len(subsections) * args.num_requesters_per_key + 10
+    args.pipeline_options.view_as(WorkerOptions).max_num_workers = max_num_workers
+    logger.info(f"Capped the max number of workers to '{max_num_workers}'.")
+
     request_idxs = {name: itertools.cycle(range(args.num_requesters_per_key)) for name, _ in subsections}
 
     def subsection_and_request(it: Config) -> t.Tuple[str, int]:

weather_dl/setup.py

@@ -32,7 +32,7 @@
 setup(
     name='download_pipeline',
     packages=find_packages(),
-    version='0.1.9',
+    version='0.1.10',
     author='Anthromets',
     author_email='anthromets-ecmwf@google.com',
     url='https://weather-tools.readthedocs.io/en/latest/weather_dl/',