Add information on how to run synchronous Dask scheduler (d0d97669) · Commits · Services / Code Samples

computing/dask/dask_htcondor.ipynb

+36 −1

Original line number	Diff line number	Diff line
		%% Cell type:markdown id:bd440c63 tags:

		# Dask + HTCondor

		%% Cell type:markdown id:c471c42a tags:

		# Creating a cluster

		%% Cell type:markdown id:86cf720e tags:

		## using dask-labextension

		%% Cell type:markdown id:99343293 tags:

		To create a Dask cluster from the extension. Go to the dask-labextension tab and click "+ NEW"

		%% Cell type:markdown id:f32e926e tags:

		![kk](static/dask_labextension_1.png)

		%% Cell type:markdown id:5cdebde9 tags:

		You can see the cluster information and have some shortcuts
		* "<>" insert a cell with the necessary lines to create a Dask client. Some environment variables have been set in order to be able to use a "regular" Dask Client
		* "SCALE" increase/decrease the number of workers
		* "SHUTDOWN" shutdown the cluster

		%% Cell type:markdown id:7c2597ed tags:

		![kk](static/dask_add_cluster_cell.png)

		%% Cell type:code id:61fb5ed7 tags:

		``` python
		from dask.distributed import Client

		client = Client("tcp://192.168.101.82:34781'")
		client
		```

		%% Cell type:code id:5aa4058d tags:

		``` python
		client.cluster
		```

		%% Cell type:markdown id:e2e751cf tags:

		The encryption of the traffic between the different Dask components is enforced through environment variables

		%% Cell type:code id:909b23d9 tags:

		``` python
		[f'{k}={v}' for k,v in os.environ.items() if 'DASK' in k]
		```

		%% Cell type:markdown id:b762046d tags:

		![kk](static/dask_labextension_2.png)

		%% Cell type:markdown id:86bfa20e tags:

		## From python

		### Using the default environment

		This could be a notebook or a script submitted to HTCondor

		%% Cell type:code id:24e002d8 tags:

		``` python
		from pic_jupyterhub.dask_condor import SecureHTCondor
		from dask.distributed import Client
		```

		%% Cell type:code id:cbcfe3a9 tags:

		``` python
		cluster = SecureHTCondor() #scheduler_options={'dashboard_address':":8788"})
		```

		%% Cell type:markdown id:43c320bb tags:

		The cluster is initially created without workers. We need to scale it.

		%% Cell type:code id:e5d95606 tags:

		``` python
		cluster.scale(2)
		```

		%% Cell type:code id:ca7188ff tags:

		``` python
		client = Client(cluster)
		```

		%% Cell type:code id:b5b4d24f tags:

		``` python
		cluster.close()
		```

		%% Cell type:markdown id:90b5c961 tags:

		### Using a custom environment

		The environment needs to have:

		* `dask-jobqueue` to be able to start Dask clusters with HTCondor
		* `ipykernel` to use it as a notebook kernel
		* `numpy` and `pandas` to be able to create Dask arrays and dataframes
		* `bokeh` for the Dask dashboard

		`mamba create (-n {env name} \| -p {env Path}) dask-jobqueue ipykernel numpy pandas bokeh`

		%% Cell type:code id:1ba4781a tags:

		``` python
		from dask_jobqueue import HTCondorCluster
		from dask.distributed import Client
		```

		%% Cell type:code id:7dd3c523 tags:

		``` python
		cluster = HTCondorCluster(cores=1, memory='2GB', disk='10 GB',
		job_extra_directives={'getenv': 'True'}) # needed to propagate the security
		```

		%% Output

		/data/jupyter/software/envs/jupyter_8440a3ce4a306902/lib/python3.12/site-packages/distributed/node.py:182: UserWarning: Port 8787 is already in use.
		Perhaps you already have a cluster running?
		Hosting the HTTP server on port 45723 instead
		warnings.warn(
		WARNING:bokeh.server.util:Host wildcard '*' will allow connections originating from multiple (or possibly all) hostnames or IPs. Use non-wildcard values to restrict access explicitly

		%% Cell type:code id:e3a6efad tags:

		``` python
		cluster
		```

		%% Cell type:code id:e0c0f659 tags:

		``` python
		cluster.scale(2)
		```

		%% Cell type:code id:66cf3e55 tags:

		``` python
		c = Client(cluster)
		```

		%% Cell type:markdown id:a101a9d8 tags:

		Security is inherited from environment variables

		%% Cell type:code id:8b92c694 tags:

		``` python
		cluster.security
		```

		%% Cell type:markdown id:b0ffb263 tags:

		## Connect to existing cluster

		%% Cell type:markdown id:3570b4ab tags:

		Meaning a cluster that was launched from outside you jupyterlab instance, e.g. an independent HTCondor job or somebody else's cluster.

		Check IP address of running job
		```
		nslookup $(condor_q $JOB_ID -af RemoteHost \| cut -d "@" -f 2)
		```

		%% Cell type:code id:7c0b19a6 tags:

		``` python
		from dask.distributed import Client

		client = Client("tls://192.168.100.5:42166")
		client
		```

		%% Cell type:markdown id:c35159c9-9eeb-4170-b802-283eea920958 tags:

		## Workers with GPUs

		If you want your workers to have a GPU assigned, you need to add the `+RequestGpus=1` argument to the worker creation. You can achieve this by adding it to the `job_extra_directives` attribute when instantiating the `HTCondorCluster` class
		To assign a GPU to your workers, include the `+RequestGpus=1` argument in the worker setup. This can be done by adding it to the `job_extra_directives` attribute when you instantiate the `HTCondorCluster` class.

		%% Cell type:code id:f21e666b-39f6-4f6c-91b9-0540eb17d5f6 tags:

		``` python
		cluster = HTCondorCluster(
		cores=1, memory='2GB', disk='10 GB',
		job_extra_directives={
		'getenv': 'True',
		'+RequestGpus': 1
		}
		)
		```

		%% Cell type:markdown id:9a9fe241-16c1-4eb6-84d2-20aee52fef68 tags:

		or through the [configuration of the jupyterlab extension](#Configuration)

		```
		labextension:
		factory:
		module: 'dask_jobqueue'
		class: 'HTCondorCluster'
		args: []
		kwargs: {"job_extra_directives": {"getenv": "True", "+RequestGpus": 1}}
		```

		%% Cell type:markdown id:0f349885-8d33-44e8-8c5c-fadade3c4071 tags:

		# Security

		It is of uttermost importance to enable security when lauching a Dask cluster, otherwise a malicious user can impersonate you by connecting to your cluster and submitting jobs to it. The attacker would then have access to your personal (e.g. SSH keys and private files in your home) and shared data.

		## in jupyter.pic.es

		In order to enforce security in the communication with Dask clusters, the notebooks started through jupyter.pic.es are populated with the environment variables:

		```
		DASK_DISTRIBUTED__COMM__REQUIRE_ENCRYPTION=true
		DASK_DISTRIBUTED__COMM__TLS__SCHEDULER__CERT=${HOME}/.config/dask/security/cert.pem
		DASK_DISTRIBUTED__COMM__TLS__CA_FILE=${HOME}/.config/dask/security/ca_file.pem
		DASK_DISTRIBUTED__COMM__TLS__CLIENT__KEY=${HOME}/.config/dask/security/key.pem
		DASK_DISTRIBUTED__COMM__TLS__WORKER__CERT=${HOME}/.config/dask/security/cert.pem
		DASK_DISTRIBUTED__COMM__TLS__SCHEDULER__KEY=${HOME}/.config/dask/security/key.pem
		DASK_DISTRIBUTED__COMM__TLS__WORKER__KEY=${HOME}/.config/dask/security/key.pem
		DASK_DISTRIBUTED__COMM__TLS__CLIENT__CERT=${HOME}/.config/dask/security/cert.pem
		```

		See the [official documentation](https://distributed.dask.org/en/stable/tls.html#tls-ssl) to understand the function of each file.

		When starting a Dask cluster through the jupyterlab extension or using the `pic_jupyterhub.dask_condor.SecureHTCondor` module in the standard environment these files will be created at runtime.

		There might be issues when trying to communicate with a running Dask cluster if the security is not properly configured. See the [Troubleshooting section](#DaskSecurityTroubleshooting) for details

		## in a custom cluster

		When launching a cluster from a cell or a job, there are two way of providing encryption.

		### Temporary security

		Using temporary information stored in memory, the cluster won't be accessible from outside the process that spawned it. For this solution, the `cryptography` packages is needed. You can install it with `conda install cryptography`.

		%% Cell type:code id:c726c306-8276-4939-937a-7ef9b77e109b tags:

		``` python
		from dask_jobqueue import HTCondorCluster
		from dask.distributed import Client
		from distributed.security import Security

		temp_sec = Security.temporary()

		temp_sec_cluster = HTCondorCluster(
		cores=1, memory='2GB', disk='10 GB',
		job_extra_directives={'getenv': 'True'},
		security=temp_sec)

		temp_sec_cluster.scale(2)
		```

		%% Output

		/data/jupyter/software/envs/jupyter_8440a3ce4a306902/lib/python3.12/site-packages/dask_jobqueue/core.py:745: UserWarning: Using a temporary security object without explicitly setting a shared_temp_directory: writing temp files to current working directory (/nfs/pic.es/user/t/torradeflot/services-code-samples/computing/dask) instead. You can set this value by using dask for e.g. `dask.config.set({'jobqueue.pbs.shared_temp_directory': '~'})`or by setting this value in the config file found in `~/.config/dask/jobqueue.yaml`
		warnings.warn(
		/data/jupyter/software/envs/jupyter_8440a3ce4a306902/lib/python3.12/site-packages/distributed/node.py:182: UserWarning: Port 8787 is already in use.
		Perhaps you already have a cluster running?
		Hosting the HTTP server on port 46143 instead
		warnings.warn(

		%% Cell type:code id:60d9cd01-9c75-4d0d-a5bf-6c50ca72fb23 tags:

		``` python
		temp_sec_client = Client(temp_sec_cluster, security=temp_sec)
		```

		%% Cell type:markdown id:d7857eca-b75c-4edc-843d-12b1c3c42e58 tags:

		### Encryption using preexisting files

		If you already have a set of CA, key, certificate files, you can use them to encrypt the traffic with and within the Dask Cluster.

		%% Cell type:code id:7af5c771-0278-4a81-a87b-76d54499e327 tags:

		``` python
		from dask_jobqueue import HTCondorCluster
		from dask.distributed import Client
		from distributed.security import Security

		# Point this variables to the files in the file system
		ca_file = '/path/to/ca/file'
		cert_file = '/path/to/certificate/file'
		key_file = '/path/to/key/file'

		sec = Security(
		require_encryption=True,
		tls_ca_file=ca_file,
		tls_client_cert=cert_file,
		tls_client_key=key_file,
		tls_worker_cert=cert_file,
		tls_worker_key=key_file,
		tls_scheduler_cert=cert_file,
		tls_scheduler_key=key_file
		)

		secure_cluster = HTCondorCluster(
		cores=1, memory='2GB', disk='10 GB',
		job_extra_directives={'getenv': 'True'},
		security=sec)

		secure_cluster.scale(1)
		```

		%% Output

		/data/jupyter/software/envs/jupyter_8440a3ce4a306902/lib/python3.12/site-packages/distributed/node.py:182: UserWarning: Port 8787 is already in use.
		Perhaps you already have a cluster running?
		Hosting the HTTP server on port 39937 instead
		warnings.warn(

		%% Cell type:markdown id:c74519fc-f224-4a84-a493-535adfc24c29 tags:

		Then the cluster can be accessed from an independent process

		%% Cell type:code id:9019bd0d-725c-4f6e-98df-c8f0e903a0e7 tags:

		``` python
		client = Client(secure_cluster.scheduler_address, security=sec)
		```

		%% Cell type:markdown id:3c2c9fbb-27f8-405f-9b59-e99bc4b70dc2 tags:

		### Encryption using files generated with SecureHTCondor

		In a similar fashion as in the previous section, if you ever started a cluster using the `SecureHTCondor` module availalbe in jupyter's base environment ([how to](#Creating-a-cluster)), a set of files that can be used for SSL encryption will already be available in the folder `${HOME}/.config/dask/security` and can be used as follows.

		%% Cell type:code id:ec8eedc8-7fbc-4b2a-9171-27614199d815 tags:

		``` python
		import pathlib
		from dask_jobqueue import HTCondorCluster
		from dask.distributed import Client
		from distributed.security import Security

		user_home = pathlib.Path.home()
		security_folder = user_home / '.config' / 'dask' / 'security'
		ca_file = str(security_folder / 'ca_file.pem')
		cert_file = str(security_folder / 'cert.pem')
		key_file = str(security_folder / 'key.pem')

		sec = Security(
		require_encryption=True,
		tls_ca_file=ca_file,
		tls_client_cert=cert_file,
		tls_client_key=key_file,
		tls_worker_cert=cert_file,
		tls_worker_key=key_file,
		tls_scheduler_cert=cert_file,
		tls_scheduler_key=key_file
		)

		secure_cluster = HTCondorCluster(
		cores=1, memory='2GB', disk='10 GB',
		job_extra_directives={'getenv': 'True'},
		security=sec)

		secure_cluster.scale(1)
		```

		%% Output

		/data/jupyter/software/envs/jupyter_8440a3ce4a306902/lib/python3.12/site-packages/distributed/node.py:182: UserWarning: Port 8787 is already in use.
		Perhaps you already have a cluster running?
		Hosting the HTTP server on port 39937 instead
		warnings.warn(

		%% Cell type:markdown id:cbf71bd3-4ed8-486d-8fcc-f41b104daa41 tags:

		Using the scheduler address and the same security configuration you would be able to connect a client from a completely independent process.

		%% Cell type:code id:3f367ad5-7f4f-4bcc-bf66-c2d987080c7e tags:

		``` python
		client = Client(secure_cluster.scheduler_address, security=sec)
		```

		%% Cell type:markdown id:98630f16-a2d3-493b-b2df-4adc7fc7248a tags:

		# Configuration

		There are different ways to configure Dask and the related libraries, check the [official documentation](https://docs.dask.org/en/stable/configuration.html)

		## Configuration files

		You can add YAML files to `~/.config/dask/` to customize Dask's configuration. Specifics for each of the libraries in the Dask stack:

		* [distributed](https://distributed.dask.org/en/stable/): framework for distributed computing
		* [jobqueue](https://jobqueue.dask.org/en/latest/clusters-configuration.html): library to scale up Dask clusters to a batch system (e.g. HTCondor)
		* [dask-labextension](https://github.com/dask/dask-labextension): jupyterlab extension to be able to manage/monitor Dask Clusters through a GUI in jupyterlab

		## Programatic access to configuration

		Configuration can be managed through the `dask.config` module. In particular, you can:

		* `dask.config.get` or `dask.config.set` specific parameters
		* list current configuration: `dask.config.config`




		%% Cell type:markdown id:3809e604 tags:

		# Troubleshooting

		## Compatibility issues

		If you try to connect a notebook to a Dask cluster, and the notebook's environment is different from the one used to launch de cluster, you may encounter compatibility issues.

		You will tipycally receive a "Mismatched versions found" warning like this:

		```
		/data/astro/scratch2/torradeflot/envs/dask/lib/python3.11/site-packages/distributed/client.py:1388: VersionMismatchWarning: Mismatched versions found

		+---------+----------------+----------------+----------------+
		\| Package \| Client \| Scheduler \| Workers \|
		+---------+----------------+----------------+----------------+
		\| lz4 \| 4.3.3 \| 4.3.2 \| 4.3.2 \|
		\| msgpack \| 1.0.7 \| 1.0.5 \| 1.0.5 \|
		\| numpy \| 1.26.3 \| 1.24.3 \| 1.24.3 \|
		\| pandas \| 2.2.0 \| 2.0.2 \| 2.0.2 \|
		\| python \| 3.11.7.final.0 \| 3.11.5.final.0 \| 3.11.5.final.0 \|
		\| toolz \| 0.12.1 \| 0.12.0 \| 0.12.0 \|
		+---------+----------------+----------------+----------------+
		warnings.warn(version_module.VersionMismatchWarning(msg[0]["warning"]))
		```

		Some mismatches might be blocking, it is recommended to match the major and minor versions. A mismatch in the patch version shouldn't be a problem.

		%% Cell type:markdown id:8b2a984a tags:

		<a name="DaskSecurityTroubleshooting"></a>
		## Security

		If you are trying to connect to a running Dask cluster and security is no properly configured you may encounter an error such as:

		```
		Cluster Start Error
		Cluster
		failed to start: TLS certificate does not match. Check your security
		settings. More info at https://distributed.dask.org/en/latest/tls.html
		```

		There are different reasons why you can see this type of errors

		### SSL files don't exist

		If you launch a Dask cluster from a notebook in `jupyter.pic.es` but you have never launched a cluster from the dask-labextension or using the `pic_jupyterhub` module, you may encounter a problem because encryption is enforced but the certificates do not exist.

		If this is the case, you can launch a cluster using one of these options as shown above. This will generate the certificate files and the subsequent creation of a Dask cluster from a notebook should succeed.

		### SSL files got corrupted or expired

		In some cases, the files used for the encryption can get corrupted. These files can be found in `${HOME}/.config/dask/security`. If you think this is the case, remove all the files in this folder and start a new Dask cluster from the jupyterlab extension or the `pic_jupyterhub` module in the main environment so that they can be regenerated.

		### SSL configuration mismatch between client and cluster

		Make sure that the encripytion configuration is consistent. Check Dask environment variables, configuration files and direct arguments used on both ends.

		%% Cell type:markdown id:d556cd0b-1d7c-4943-93c0-1abb9f55d1fd tags:

		## Starting Dask Cluster through the extension fails

		### Got multiple values for keyword argument asynchronous

		There's currently a bug in the extension and the second time you try to start a cluster from it you will see this error. You can still spawn newe clusters from a notebook, but not from the extension.

		The only workaraound as of now is to close the jupyterlab server and request a new one.

		%% Cell type:markdown id:1d589c7c-aa6b-4932-8400-934647b5d324 tags:

		## Debugging

		### Redirect the HTCondor worker jobs logs to files

		Adding this section in the Dask configuration:

		```
		jobqueue:
		htcondor:
		log-directory: /path/to/the/folder/where/you/want/the/logs
		```

		will create `worker-{Id}.{out\|err\|log}` files in the chosen folder.

		### Change log level

		```
		logging:
		version: 1
		formatters:
		default:
		format: '%(asctime)s %(levelname)-8s %(name)-15s %(message)s'
		datefmt: '%Y-%m-%d %H:%M:%S'
		handlers:
		console:
		class: logging.StreamHandler
		level: DEBUG
		formatter: default
		loggers:
		distributed.worker:
		level: DEBUG
		handlers:
		- console
		```

		### Single thread synchronous scheduler

		It can be difficult to debug a problem when using the parallell computing paradigm. For debugging purposes the scheduler can be configured to be single-threaded and synchronous ([doc](https://docs.dask.org/en/stable/scheduling.html#single-thread) ). With this configuration the processing will run serially in the same thread as the notebook, so the log/print messages will be displayed in the notebook.

		WARNING Using this approach will dramaticallly reduce speed

		%% Cell type:code id:f4dbab01-4ae4-4ecc-b27b-5f1dfd9df1c6 tags:

		``` python
		import dask.array as da
		import dask

		N = 10

		def verbose_sq(chunk):
		print(f'chunk: {chunk}')
		return chunk**2

		with dask.config.set(scheduler='synchronous'):
		x = da.arange(N, chunks=(2,))
		x2 = x.map_blocks(verbose_sq).compute()
		print(f'x^2 = {x2}')
		```

		%% Output

		chunk: []
		chunk: [1]
		chunk: [8 9]
		chunk: [6 7]
		chunk: [4 5]
		chunk: [2 3]
		chunk: [0 1]
		x^2 = [ 0 1 4 9 16 25 36 49 64 81]

		%% Cell type:markdown id:f10f22ba-223f-4d33-af00-5f8315526e74 tags:

		## Dask task is stuck

		### Retire a worker

		First you need to get the worker name, e.g. `SecureHTCondor-1`, then you can retire it

		%% Cell type:code id:97fd2c39-7fbe-4d24-9e6b-338f1785c011 tags:

		``` python
		client.retire_workers(names=['SecureHTCondor-1'], close_workers=True)
		```

		%% Cell type:markdown id:cfc82432-da32-4adf-af70-c5ef10190565 tags:

		This will kill the worker job and the task will be reallocated to another worker

		%% Cell type:markdown id:a28e457b tags:

		# Examples
		## Dask example 1

		picked from https://docs.dask.org/en/stable/10-minutes-to-dask.html

		%% Cell type:code id:5353d8fe tags:

		``` python
		import numpy as np
		import pandas as pd

		import dask.dataframe as dd
		import dask.array as da
		import dask.bag as db
		```

		%% Cell type:markdown id:03a01ccf tags:

		### DataFrame

		%% Cell type:code id:c7a4f7bd tags:

		``` python
		index = pd.date_range("2021-09-01", periods=2400, freq="1H")
		df = pd.DataFrame({"a": np.arange(2400), "b": list("abcaddbe" * 300)}, index=index)
		ddf = dd.from_pandas(df, npartitions=10)
		ddf
		```

		%% Cell type:code id:4102f382 tags:

		``` python
		ddf.divisions
		```

		%% Cell type:code id:0f0b932d tags:

		``` python
		ddf.partitions[1]
		```

		%% Cell type:code id:0c79af96 tags:

		``` python
		ddf["2000-10-01": "2021-10-09 5:00"].compute()
		```

		%% Cell type:markdown id:7277eca8 tags:

		### Array

		%% Cell type:code id:87156672 tags:

		``` python
		import numpy as np
		import dask.array as da

		data = np.arange(100_000).reshape(200, 500)
		a = da.from_array(data, chunks=(100, 100))
		a
		```

		%% Cell type:code id:eded606c tags:

		``` python
		a.chunks
		```

		%% Cell type:code id:9b36946c tags:

		``` python
		a.blocks[1, 3]
		```

		%% Cell type:code id:253153f3 tags:

		``` python
		a[:50, 200]
		```

		%% Cell type:code id:82717424 tags:

		``` python
		a[:50, 200].compute()
		```

		%% Cell type:code id:c0222fbb tags:

		``` python
		a.mean()
		a.mean().compute()
		np.sin(a)
		np.sin(a).compute()
		a.T
		a.T.compute()
		```

		%% Cell type:code id:73831cd0 tags:

		``` python
		b = a.max(axis=1)[::-1] + 10
		```

		%% Cell type:code id:6af83c0b tags:

		``` python
		b[:10].compute()
		```

		%% Cell type:code id:81b4ad35 tags:

		``` python
		b.dask
		```

		%% Cell type:markdown id:cc48cd7c tags:

		## Dask example 2

		%% Cell type:code id:0f9c1ff3 tags:

		``` python
		import dask.array as da
		x = da.random.random((30_000, 30_000), chunks=(1000, 1000))
		x
		```

		%% Cell type:code id:35ec13c3 tags:

		``` python
		y = x + x.T
		```

		%% Cell type:code id:4ae20942 tags:

		``` python
		y.sum().compute()
		```

		%% Cell type:code id:11550bf3 tags:

		``` python
		y[:, :10].compute()
		```

		%% Cell type:code id:90dbcbc0 tags:

		``` python
		```

computing/dask/dask_htcondor.py

+22 −1

Original line number	Diff line number	Diff line
		@@ -110,7 +110,7 @@ client

		# ## Workers with GPUs
		#
		# If you want your workers to have a GPU assigned, you need to add the `+RequestGpus=1` argument to the worker creation. You can achieve this by adding it to the `job_extra_directives` attribute when instantiating the `HTCondorCluster` class
		# To assign a GPU to your workers, include the `+RequestGpus=1` argument in the worker setup. This can be done by adding it to the `job_extra_directives` attribute when you instantiate the `HTCondorCluster` class.

		cluster = HTCondorCluster(
		cores=1, memory='2GB', disk='10 GB',
		@@ -377,6 +377,27 @@ client = Client(secure_cluster.scheduler_address, security=sec)
		# - console
		# ```
		#
		# ### Single thread synchronous scheduler
		#
		# It can be difficult to debug a problem when using the parallell computing paradigm. For debugging purposes the scheduler can be configured to be single-threaded and synchronous ([doc](https://docs.dask.org/en/stable/scheduling.html#single-thread) ). With this configuration the processing will run serially in the same thread as the notebook, so the log/print messages will be displayed in the notebook.
		#
		# WARNING Using this approach will dramaticallly reduce speed

		# +
		import dask.array as da
		import dask

		N = 10

		def verbose_sq(chunk):
		print(f'chunk: {chunk}')
		return chunk**2

		with dask.config.set(scheduler='synchronous'):
		x = da.arange(N, chunks=(2,))
		x2 = x.map_blocks(verbose_sq).compute()
		print(f'x^2 = {x2}')
		# -

		# ## Dask task is stuck
		#