056-k8s-reduce.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "# Multinode Reduction Operations"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "slideshow": {
     "slide_type": "skip"
    }
   },
   "outputs": [],
   "source": [
    "maybe() {\n",
    "    \"$@\" > .last_maybe 2>&1 || true\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "# Multinode Reduction Operations\n",
    "\n",
    "The core of distributed SGD:\n",
    "\n",
    "- start off with the same weight vector\n",
    "- forward pass on all nodes (using different batches)\n",
    "- backward pass to compute all gradients\n",
    "- _sum all gradients across all nodes_\n",
    "- add total gradient to all weights\n",
    "\n",
    "How does the summation work on Kubernetes?"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "slideshow": {
     "slide_type": "skip"
    }
   },
   "outputs": [],
   "source": [
    "cat > kubetpl.yaml <<'EOF'\n",
    "image: gcr.io/research-191823/bigdata19\n",
    "memory: 4G\n",
    "cpu: 1\n",
    "app: bigdata19\n",
    "subdomain: bigdata19\n",
    "port:\n",
    "  - 7880\n",
    "config_map: files\n",
    "env:\n",
    "  - MASTER_ADDR=master.bigdata19\n",
    "  - MASTER_PORT=7880\n",
    "EOF"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {
    "slideshow": {
     "slide_type": "skip"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "service \"bigdata19\" deleted\n",
      "service/bigdata19 created\n"
     ]
    }
   ],
   "source": [
    "maybe kubectl delete service/bigdata19\n",
    "kubetpl service | kubectl apply -f -"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "# Python Script\n",
    "\n",
    "- sum reduction across a number of machines\n",
    "- core operation of synchronous distributed SGD"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [],
   "source": [
    "cat > reduce.py <<'EOF'\n",
    "\n",
    "import os, torch\n",
    "import torch.distributed as dist\n",
    "\n",
    "rank, world = int(os.environ[\"rank\"]), int(os.environ[\"world\"])\n",
    "\n",
    "dist.init_process_group(\"gloo\", rank=rank, world_size=world)\n",
    "\n",
    "deltas = torch.ones(1)*(rank+1)\n",
    "\n",
    "print(f\"before {rank} data {deltas[0]}\")\n",
    "\n",
    "all_nodes = dist.new_group(list(range(world)))\n",
    "dist.all_reduce(deltas, op=dist.ReduceOp.SUM, group=all_nodes)\n",
    "\n",
    "print(f\"after {rank} data {deltas[0]}\")\n",
    "\n",
    "EOF"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "# Upload to Shared File System"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "-- --from-file=reduce.py=reduce.py\n",
      "configmap \"files\" deleted\n",
      "configmap/files created\n"
     ]
    }
   ],
   "source": [
    "kubefcm files reduce.py"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "# Create Master Node"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "pod \"master\" deleted\n",
      "pod \"node1\" deleted\n",
      "pod \"node2\" deleted\n",
      "pod \"node3\" deleted\n",
      "pod/master created\n"
     ]
    }
   ],
   "source": [
    "maybe kubectl delete pods --all\n",
    "\n",
    "kubetpl pod -n master -c 'rank=0 world=4 python3 /files/reduce.py' | kubectl apply -f -"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {
    "slideshow": {
     "slide_type": "skip"
    }
   },
   "outputs": [],
   "source": [
    "sleep 15"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {
    "slideshow": {
     "slide_type": "skip"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "NAME     READY   STATUS    RESTARTS   AGE\n",
      "master   1/1     Running   0          16s\n"
     ]
    }
   ],
   "source": [
    "kubectl get pods"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "# Create Additional Compute Nodes"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "pod/node1 created\n",
      "pod/node2 created\n",
      "pod/node3 created\n"
     ]
    }
   ],
   "source": [
    "for i in {1..3}; do\n",
    "    kubetpl pod -n node$i -c \"rank=$i world=4 python3 /files/reduce.py\" | kubectl apply -f -\n",
    "done"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {
    "slideshow": {
     "slide_type": "skip"
    }
   },
   "outputs": [],
   "source": [
    "sleep 15"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "# Result of Computation\n",
    "\n",
    "- all values added and distributed to all nodes"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "before 0 data 1.0\n",
      "after 0 data 10.0\n"
     ]
    }
   ],
   "source": [
    "kubectl logs master"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "before 3 data 4.0\n",
      "after 3 data 10.0\n"
     ]
    }
   ],
   "source": [
    "kubectl logs node3"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "slideshow": {
     "slide_type": "skip"
    }
   },
   "outputs": [],
   "source": [
    "kubectl delete jobs --all\n",
    "kubectl delete pods --all"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "# Global Reduction Operations\n",
    "\n",
    "That's the core of synchronous distributed SGD."
   ]
  }
 ],
 "metadata": {
  "celltoolbar": "Slideshow",
  "kernelspec": {
   "display_name": "Bash",
   "language": "bash",
   "name": "bash"
  },
  "language_info": {
   "codemirror_mode": "shell",
   "file_extension": ".sh",
   "mimetype": "text/x-sh",
   "name": "bash"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}