{
  "cells": [
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "# Task Scheduling with Dask\n",
        "\n",
        "<!--\n",
        "Project: Dask Parallel Computing\n",
        "Author: Molla Samser\n",
        "Designer & Tester: Rima Khatun\n",
        "Website: https://rskworld.in\n",
        "Email: help@rskworld.in, support@rskworld.in\n",
        "Phone: +91 93305 39277\n",
        "-->\n",
        "\n",
        "This notebook demonstrates task scheduling and optimization in Dask.\n"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {},
      "outputs": [],
      "source": [
        "from dask import delayed, compute\n",
        "from dask.distributed import Client, LocalCluster\n",
        "import time\n",
        "import numpy as np\n",
        "\n",
        "print(\"Task Scheduling Demo\")\n",
        "print(\"=\" * 50)\n"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "## Setting Up Scheduler\n"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {},
      "outputs": [],
      "source": [
        "# Create a local cluster with scheduler\n",
        "cluster = LocalCluster(n_workers=2, threads_per_worker=2)\n",
        "client = Client(cluster)\n",
        "\n",
        "print(\"Scheduler Dashboard:\", client.dashboard_link)\n",
        "print(\"Scheduler ready\")\n"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "## Task Dependencies\n"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {},
      "outputs": [],
      "source": [
        "@delayed\n",
        "def load_data(source):\n",
        "    time.sleep(0.2)\n",
        "    return f\"Data from {source}\"\n",
        "\n",
        "@delayed\n",
        "def process_data(data):\n",
        "    time.sleep(0.3)\n",
        "    return f\"Processed: {data}\"\n",
        "\n",
        "@delayed\n",
        "def combine_data(data1, data2):\n",
        "    time.sleep(0.1)\n",
        "    return f\"Combined: {data1} + {data2}\"\n",
        "\n",
        "# Create task graph with dependencies\n",
        "data1 = load_data(\"source1\")\n",
        "data2 = load_data(\"source2\")\n",
        "\n",
        "processed1 = process_data(data1)\n",
        "processed2 = process_data(data2)\n",
        "\n",
        "final = combine_data(processed1, processed2)\n",
        "\n",
        "print(\"Task graph created with dependencies\")\n",
        "print(\"Executing...\")\n",
        "start_time = time.time()\n",
        "result = final.compute()\n",
        "end_time = time.time()\n",
        "\n",
        "print(f\"Result: {result}\")\n",
        "print(f\"Time: {end_time - start_time:.2f} seconds\")\n"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "## Task Prioritization\n"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {},
      "outputs": [],
      "source": [
        "@delayed\n",
        "def high_priority_task(n):\n",
        "    time.sleep(0.1)\n",
        "    return n * 2\n",
        "\n",
        "@delayed\n",
        "def low_priority_task(n):\n",
        "    time.sleep(0.1)\n",
        "    return n * 3\n",
        "\n",
        "# Create tasks with priorities\n",
        "high_priority = [high_priority_task(i) for i in range(10)]\n",
        "low_priority = [low_priority_task(i) for i in range(10)]\n",
        "\n",
        "print(\"Submitting tasks with different priorities...\")\n",
        "start_time = time.time()\n",
        "\n",
        "# Compute high priority first\n",
        "high_results = compute(*high_priority)\n",
        "low_results = compute(*low_priority)\n",
        "\n",
        "end_time = time.time()\n",
        "print(f\"All tasks completed in {end_time - start_time:.2f} seconds\")\n"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "## Monitoring Task Progress\n"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {},
      "outputs": [],
      "source": [
        "@delayed\n",
        "def long_running_task(n):\n",
        "    time.sleep(0.5)\n",
        "    return sum(range(n))\n",
        "\n",
        "# Create multiple tasks\n",
        "tasks = [long_running_task(i * 1000) for i in range(1, 21)]\n",
        "\n",
        "print(\"Submitting 20 tasks...\")\n",
        "futures = client.compute(tasks)\n",
        "\n",
        "# Monitor progress\n",
        "while not all(f.done() for f in futures):\n",
        "    done = sum(1 for f in futures if f.done())\n",
        "    print(f\"Progress: {done}/{len(futures)} tasks completed\", end=\"\\r\")\n",
        "    time.sleep(0.1)\n",
        "\n",
        "results = client.gather(futures)\n",
        "print(f\"\\nAll {len(results)} tasks completed!\")\n"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "## Cleanup\n"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {},
      "outputs": [],
      "source": [
        "client.close()\n",
        "cluster.close()\n",
        "print(\"Scheduler closed\")\n"
      ]
    }
  ],
  "metadata": {
    "language_info": {
      "name": "python"
    }
  },
  "nbformat": 4,
  "nbformat_minor": 2
}