diff --git a/CHANGELOG.md b/CHANGELOG.md index 6abcf4a4b..541e13fb4 100644 --- a/CHANGELOG.md +++ b/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`. diff --git a/vast_pipeline/daskmanager/manager.py b/vast_pipeline/daskmanager/manager.py index 8db23e323..0d11c1701 100644 --- a/vast_pipeline/daskmanager/manager.py +++ b/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) diff --git a/vast_pipeline/image/main.py b/vast_pipeline/image/main.py index 0a014636c..1a0f7aab5 100644 --- a/vast_pipeline/image/main.py +++ b/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 diff --git a/vast_pipeline/management/commands/runlocalcluster.py b/vast_pipeline/management/commands/runlocalcluster.py index d3d5e4f09..98e9fe686 100644 --- a/vast_pipeline/management/commands/runlocalcluster.py +++ b/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' diff --git a/webinterface/.env.template b/webinterface/.env.template index 595e75266..63ef1fff2 100644 --- a/webinterface/.env.template +++ b/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 diff --git a/webinterface/settings.py b/webinterface/settings.py index 0583c120f..135165fc9 100644 --- a/webinterface/settings.py +++ b/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 = {