{ "cells": [ { "cell_type": "markdown", "metadata": {}, "source": [ "# Resilience against hardware failures\n", "\n", "Scenario: We have a cluster that partially consists of preemptible ressources. That is, we'll have to deal with workers suddenly being shut down during computation. While demonstrated here with a `LocalCluster`, Dask's resilience against preempted ressources is most useful with, e.g., [Dask Kubernetes](https://kubernetes.dask.org/) or [Dask Jobqueue](https://jobqueue.dask.org).\n", "\n", "Relevant docs: " ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Increase resilience\n", "\n", "Whenever a worker shuts down, the scheduler will increment the suspiciousness counter of _all_ tasks that were assigned (not necessarily computing) to the worker in question. Whenever the suspiciousness of a task exceeds a certain threshold (3 by default), the task will be considered broken. We want to compute many tasks on only a few workers with workers shutting down randomly. So we expect the suspiciousness of all tasks to grow rapidly. Let's increase the threshold:" ] }, { "cell_type": "code", "execution_count": 1, "metadata": { "execution": { "iopub.execute_input": "2022-05-16T13:50:13.052206Z", "iopub.status.busy": "2022-05-16T13:50:13.051794Z", "iopub.status.idle": "2022-05-16T13:50:13.181857Z", "shell.execute_reply": "2022-05-16T13:50:13.181209Z" } }, "outputs": [], "source": [ "import dask\n", "\n", "dask.config.set({'distributed.scheduler.allowed-failures': 100});" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## All other imports" ] }, { "cell_type": "code", "execution_count": 2, "metadata": { "execution": { "iopub.execute_input": "2022-05-16T13:50:13.185172Z", "iopub.status.busy": "2022-05-16T13:50:13.184970Z", "iopub.status.idle": "2022-05-16T13:50:13.465132Z", "shell.execute_reply": "2022-05-16T13:50:13.464573Z" } }, "outputs": [], "source": [ "from dask.distributed import Client, LocalCluster\n", "from dask import bag as db\n", "import os\n", "import random\n", "from time import sleep" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## A cluster" ] }, { "cell_type": "code", "execution_count": 3, "metadata": { "execution": { "iopub.execute_input": "2022-05-16T13:50:13.468438Z", "iopub.status.busy": "2022-05-16T13:50:13.468041Z", "iopub.status.idle": "2022-05-16T13:50:16.250103Z", "shell.execute_reply": "2022-05-16T13:50:16.249462Z" } }, "outputs": [ { "data": { "text/html": [ "
\n", "
\n", "
\n", "

Client

\n", "

Client-1db99cb4-d51f-11ec-9c24-000d3aeabb7a

\n", " \n", "\n", " \n", " \n", " \n", " \n", " \n", " \n", "\n", " \n", " \n", " \n", " \n", " \n", " \n", "\n", "
Connection method: Cluster objectCluster type: distributed.LocalCluster
\n", " Dashboard: http://127.0.0.1:8787/status\n", "
\n", "\n", " \n", "
\n", "

Cluster Info

\n", "
\n", "
\n", "
\n", "
\n", "

LocalCluster

\n", "

2d66534e

\n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", "\n", "\n", " \n", "
\n", " Dashboard: http://127.0.0.1:8787/status\n", " \n", " Workers: 4\n", "
\n", " Total threads: 4\n", " \n", " Total memory: 1.49 GiB\n", "
Status: runningUsing processes: True
\n", "\n", "
\n", " \n", "

Scheduler Info

\n", " \n", "\n", "
\n", "
\n", "
\n", "
\n", "

Scheduler

\n", "

Scheduler-d27ee06a-15bc-4952-8968-601c83dfd3b8

\n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", "
\n", " Comm: tcp://127.0.0.1:43055\n", " \n", " Workers: 4\n", "
\n", " Dashboard: http://127.0.0.1:8787/status\n", " \n", " Total threads: 4\n", "
\n", " Started: Just now\n", " \n", " Total memory: 1.49 GiB\n", "
\n", "
\n", "
\n", "\n", "
\n", " \n", "

Workers

\n", " \n", "\n", " \n", "
\n", "
\n", "
\n", "
\n", " \n", "

Worker: 0

\n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", "\n", " \n", "\n", " \n", "\n", "
\n", " Comm: tcp://127.0.0.1:34073\n", " \n", " Total threads: 1\n", "
\n", " Dashboard: http://127.0.0.1:33771/status\n", " \n", " Memory: 381.47 MiB\n", "
\n", " Nanny: tcp://127.0.0.1:45095\n", "
\n", " Local directory: /home/runner/work/dask-examples/dask-examples/dask-worker-space/worker-qbhiti_i\n", "
\n", "
\n", "
\n", "
\n", " \n", "
\n", "
\n", "
\n", "
\n", " \n", "

Worker: 1

\n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", "\n", " \n", "\n", " \n", "\n", "
\n", " Comm: tcp://127.0.0.1:37871\n", " \n", " Total threads: 1\n", "
\n", " Dashboard: http://127.0.0.1:33373/status\n", " \n", " Memory: 381.47 MiB\n", "
\n", " Nanny: tcp://127.0.0.1:37151\n", "
\n", " Local directory: /home/runner/work/dask-examples/dask-examples/dask-worker-space/worker-jk1j64xa\n", "
\n", "
\n", "
\n", "
\n", " \n", "
\n", "
\n", "
\n", "
\n", " \n", "

Worker: 2

\n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", "\n", " \n", "\n", " \n", "\n", "
\n", " Comm: tcp://127.0.0.1:38635\n", " \n", " Total threads: 1\n", "
\n", " Dashboard: http://127.0.0.1:35205/status\n", " \n", " Memory: 381.47 MiB\n", "
\n", " Nanny: tcp://127.0.0.1:36509\n", "
\n", " Local directory: /home/runner/work/dask-examples/dask-examples/dask-worker-space/worker-zd_d79az\n", "
\n", "
\n", "
\n", "
\n", " \n", "
\n", "
\n", "
\n", "
\n", " \n", "

Worker: 3

\n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", "\n", " \n", "\n", " \n", "\n", "
\n", " Comm: tcp://127.0.0.1:35215\n", " \n", " Total threads: 1\n", "
\n", " Dashboard: http://127.0.0.1:34825/status\n", " \n", " Memory: 381.47 MiB\n", "
\n", " Nanny: tcp://127.0.0.1:44715\n", "
\n", " Local directory: /home/runner/work/dask-examples/dask-examples/dask-worker-space/worker-gotg3az2\n", "
\n", "
\n", "
\n", "
\n", " \n", "\n", "
\n", "
\n", "\n", "
\n", "
\n", "
\n", "
\n", " \n", "\n", "
\n", "
" ], "text/plain": [ "" ] }, "execution_count": 3, "metadata": {}, "output_type": "execute_result" } ], "source": [ "cluster = LocalCluster(threads_per_worker=1, n_workers=4, memory_limit=400e6)\n", "client = Client(cluster)\n", "client" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## A simple workload\n", "\n", "We'll multiply a range of numbers by two, add some sleep to simulate some real work, and then reduce the whole sequence of doubled numbers by summing them." ] }, { "cell_type": "code", "execution_count": 4, "metadata": { "execution": { "iopub.execute_input": "2022-05-16T13:50:16.254025Z", "iopub.status.busy": "2022-05-16T13:50:16.253504Z", "iopub.status.idle": "2022-05-16T13:50:16.256900Z", "shell.execute_reply": "2022-05-16T13:50:16.256317Z" } }, "outputs": [], "source": [ "def multiply_by_two(x):\n", " sleep(0.02)\n", " return 2 * x" ] }, { "cell_type": "code", "execution_count": 5, "metadata": { "execution": { "iopub.execute_input": "2022-05-16T13:50:16.259603Z", "iopub.status.busy": "2022-05-16T13:50:16.259140Z", "iopub.status.idle": "2022-05-16T13:50:16.264203Z", "shell.execute_reply": "2022-05-16T13:50:16.263791Z" } }, "outputs": [], "source": [ "N = 400\n", "\n", "x = db.from_sequence(range(N), npartitions=N // 2)\n", "\n", "mults = x.map(multiply_by_two)\n", "\n", "summed = mults.sum()" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Suddenly shutting down workers" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "Let's mark two worker process id's as non-preemptible." ] }, { "cell_type": "code", "execution_count": 6, "metadata": { "execution": { "iopub.execute_input": "2022-05-16T13:50:16.267038Z", "iopub.status.busy": "2022-05-16T13:50:16.266650Z", "iopub.status.idle": "2022-05-16T13:50:16.270006Z", "shell.execute_reply": "2022-05-16T13:50:16.269477Z" } }, "outputs": [], "source": [ "all_current_workers = [w.pid for w in cluster.scheduler.workers.values()]\n", "non_preemptible_workers = all_current_workers[:2]" ] }, { "cell_type": "code", "execution_count": 7, "metadata": { "execution": { "iopub.execute_input": "2022-05-16T13:50:16.273192Z", "iopub.status.busy": "2022-05-16T13:50:16.272738Z", "iopub.status.idle": "2022-05-16T13:50:16.276709Z", "shell.execute_reply": "2022-05-16T13:50:16.275965Z" } }, "outputs": [], "source": [ "def kill_a_worker():\n", " preemptible_workers = [\n", " w.pid for w in cluster.scheduler.workers.values()\n", " if w.pid not in non_preemptible_workers]\n", " if preemptible_workers:\n", " os.kill(random.choice(preemptible_workers), 15)" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Start the computation and keep shutting down workers while it's running" ] }, { "cell_type": "code", "execution_count": 8, "metadata": { "execution": { "iopub.execute_input": "2022-05-16T13:50:16.279475Z", "iopub.status.busy": "2022-05-16T13:50:16.279178Z", "iopub.status.idle": "2022-05-16T13:50:19.444847Z", "shell.execute_reply": "2022-05-16T13:50:19.443272Z" } }, "outputs": [ { "name": "stderr", "output_type": "stream", "text": [ "2022-05-16 13:50:16,526 - distributed.nanny - WARNING - Restarting worker\n" ] } ], "source": [ "summed = client.compute(summed)\n", "\n", "while not summed.done():\n", " kill_a_worker()\n", " sleep(3.0)" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Check if results match" ] }, { "cell_type": "code", "execution_count": 9, "metadata": { "execution": { "iopub.execute_input": "2022-05-16T13:50:19.448776Z", "iopub.status.busy": "2022-05-16T13:50:19.448363Z", "iopub.status.idle": "2022-05-16T13:50:19.460303Z", "shell.execute_reply": "2022-05-16T13:50:19.459589Z" } }, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "`sum(range(400))` on cluster: 159600\t(should be 159600)\n" ] } ], "source": [ "print(f\"`sum(range({N}))` on cluster: {summed.result()}\\t(should be {N * (N-1)})\")" ] } ], "metadata": { "anaconda-cloud": {}, "kernelspec": { "display_name": "Python 3 (ipykernel)", "language": "python", "name": "python3" }, "language_info": { "codemirror_mode": { "name": "ipython", "version": 3 }, "file_extension": ".py", "mimetype": "text/x-python", "name": "python", "nbconvert_exporter": "python", "pygments_lexer": "ipython3", "version": "3.9.12" } }, "nbformat": 4, "nbformat_minor": 4 }