{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Dask deployment on Grid'5000\n",
    "\n",
    "This notebook will deploy a Dask cluster on Grid'5000 and launch a simpe computation.\n",
    "\n",
    "Requirements: \n",
    "  - A conda[[1]] environment setup on the Grid'5000 frontend with dask installed and EnOSlib.\n",
    "  - The same environment can be use to run this notebook from your local machine. \n",
    "\n",
    "[1]: https://docs.conda.io/en/latest/miniconda.html#linux-installers\n",
    "\n",
    "\n",
    "## Initial impors\n",
    "    "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from enoslib import *\n",
    "import logging\n",
    "\n",
    "# get some logs\n",
    "logging.basicConfig(level=logging.INFO)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Get some resources on Grid'5000\n",
    "\n",
    "This will reserve two nodes, where the Dask cluster will be deployed later."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "prod = G5kNetworkConf(id=\"prod\", roles=[\"network\"], type=\"prod\", site=\"rennes\")\n",
    "conf = (\n",
    "    G5kConf.from_settings(job_name=\"dask\", job_type=[])\n",
    "    .add_machine(roles=[\"scheduler\"], cluster=\"parapide\", nodes=1, primary_network=prod)\n",
    "    .add_machine(roles=[\"worker\"], cluster=\"parapide\", nodes=1, primary_network=prod)\n",
    "    .add_network_conf(prod)\n",
    ").finalize()\n",
    "provider = G5k(conf)\n",
    "roles, _ = provider.init()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Deploy Dask on the nodes\n",
    "This assumes that the conda environment (dask-base) is configured in your home directory in `/home/<user>/miniconda3`.\n",
    "\n",
    "If the installation path differs, you can specify it using the `conda_prefix` parameter. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "username = g5k_api_utils.get_api_username()\n",
    "dask = Dask(\"dask-base\", scheduler=roles[\"scheduler\"][0], workers=roles[\"worker\"], run_as=username)\n",
    "dask.deploy()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Using Dask\n",
    "\n",
    "Here we go with a simple computation (3 tasks, 2 dependent-ones). \n",
    "The below code will create all the tunnels needed to access the Dask dashboard and the scheduler."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from dask import delayed\n",
    "import time\n",
    "\n",
    "def inc(x):\n",
    "    time.sleep(5)\n",
    "    return x + 1\n",
    "\n",
    "def dec(x):\n",
    "    time.sleep(3)\n",
    "    return x - 1\n",
    "\n",
    "def add(x, y):\n",
    "    time.sleep(7)\n",
    "    return x + y\n",
    "\n",
    "x = delayed(inc)(1)\n",
    "y = delayed(dec)(2)\n",
    "total = delayed(add)(x, y)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Launch the computation\n",
    "\n",
    "In the mean time you can check the web dashboard. The connection URL will be displayed."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from dask.distributed import Client\n",
    "# Tunnel to the dashboard\n",
    "addr, port, tunnel = G5kTunnel(dask.scheduler.address, 8787).start()\n",
    "print(f\"dashboard: http://{addr}:{port}\")\n",
    "with G5kTunnel(dask.scheduler.address, 8786) as (addr, port, _):\n",
    "    print(f\"Scheduler address: {addr}:{port}\")\n",
    "    client = Client(f\"tcp://{addr}:{port}\")\n",
    "    # launch a computation\n",
    "    print(f\"result={total.compute()}\")\n",
    "\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# will stop the tunnel to the dashboard and the Dask cluster.\n",
    "if tunnel is not None:\n",
    "    tunnel.stop(force=True)\n",
    "dask.destroy()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3.8.5 64-bit ('andromak': conda)",
   "metadata": {
    "interpreter": {
     "hash": "044ddde3072526b14262047f16da4fe10c08ccab664ff325d5dd19288f92d50e"
    }
   },
   "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.8.5-final"
  },
  "orig_nbformat": 2
 },
 "nbformat": 4,
 "nbformat_minor": 2
}