Specify dask worker memory limits (#843)

* Enable mem specification

* Fixed docstring typo

* Updated changelog

* PEP8

Author: ddobie

CHANGELOG.md

@@ -8,7 +8,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 
 #### Added
 
-
+- V2: Enable specification of Dask worker memory limits [#843](https://github.com/askap-vast/vast-pipeline/pull/843)
 - V2: Migrate pipeline to used a Dask.distributed.LocalCluster throughout [#816](https://github.com/askap-vast/vast-pipeline/pull/816)
 - V2: Add Dask.distributed support
 - V2: Use `django-postgres-copy` for database uploads [#803](https://github.com/askap-vast/vast-pipeline/pull/803)
@@ -58,6 +58,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 
 #### List of PRs
 
+- [#843](https://github.com/askap-vast/vast-pipeline/pull/843): feat: V2: Enable specification of Dask worker memory limits
 - [#833](https://github.com/askap-vast/vast-pipeline/pull/833): feat: V2: Limit associations upload to using num_io_workers
 - [#829](https://github.com/askap-vast/vast-pipeline/pull/829): feat: V2: Allow user specification of dask dashboard paramters and add some further logging to dask setup
 - [#817](https://github.com/askap-vast/vast-pipeline/pull/817): fix: V2: Updates to pairs calculation to make it work with Dask `LocalCluster`.

vast_pipeline/daskmanager/manager.py

@@ -14,6 +14,7 @@ def _start_cluster():
     logger.info('Starting local Dask Cluster...')
     logger.info(f"n_workers: {s.DASK_NUM_WORKERS}")
     logger.info(f"threads_per_worker: {s.DASK_THREADS_PER_WORKER}")
+    logger.info(f"memory per worker: {s.DASK_MEM_PER_WORKER}")
     logger.info(f"scheduler_host: {s.DASK_SCHEDULER_HOST}")
     logger.info(f"scheduler_port: {s.DASK_SCHEDULER_PORT}")
     logger.info(f"dashboard_host: {s.DASK_DASHBOARD_HOST}")
@@ -24,6 +25,7 @@ def _start_cluster():
             threads_per_worker=s.DASK_THREADS_PER_WORKER,
             host=s.DASK_SCHEDULER_HOST,
             scheduler_port=int(s.DASK_SCHEDULER_PORT),
+            memory_limit=s.DASK_MEM_PER_WORKER,
             dashboard_address=f"{s.DASK_DASHBOARD_HOST}:{s.DASK_DASHBOARD_PORT}",
         )
     client = Client(cluster)

vast_pipeline/image/main.py

@@ -14,7 +14,7 @@
 from astropy.wcs.utils import proj_plane_pixel_scales
 from typing import Dict
 
-from .utils import calc_condon_flux_errors, open_fits
+from .utils import calc_condon_flux_errors
 
 from vast_pipeline import models
 from vast_pipeline.survey.translators import tr_selavy

vast_pipeline/management/commands/runlocalcluster.py

@@ -11,7 +11,7 @@
 class Command(BaseCommand):
     """
     This script will run a Dask LocalCluster on the IP and port
-    sepcified in the settings. Use --help for usage.
+    specified in the settings. Use --help for usage.
     """
     help = 'Run a Dask LocalCluster'
 

webinterface/.env.template

@@ -29,6 +29,7 @@ SOCIAL_AUTH_GITHUB_ADMIN_TEAM=fillMeUp
 # DASK_DASHBOARD_PORT=fillMeUp
 DASK_NUM_WORKERS=14
 DASK_THREADS_PER_WORKER=1
+DASK_MEM_PER_WORKER='1GB'
 
 # Pipeline
 PIPELINE_WORKING_DIR=pipeline-runs

webinterface/settings.py

@@ -253,6 +253,7 @@
 DASK_DASHBOARD_PORT = env('DASK_DASHBOARD_PORT', cast=str, default='8787')
 DASK_NUM_WORKERS = env('DASK_NUM_WORKERS', cast=int, default=14)
 DASK_THREADS_PER_WORKER =  env('DASK_THREADS_PER_WORKER', cast=int, default=1)
+DASK_MEM_PER_WORKER = env('DASK_MEM_PER_WORKER', cast=str, default='1GB')
 
 # Logging
 LOGGING = {