Distributed data parallel.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "id": "d4f0085d",
   "metadata": {},
   "source": [
    "\"\"\"\n",
    "As Data Parallel uses threading to achieve parallelism, it suffers from a major well-known issue that arise due to Global Interpreter Lock (GIL) in Python. The way Python interpreter is designed, it is not possible to achieve perfect parallelism in Python using threading. Let’s see what GIL is.\n",
    "\n",
    "Global Interpreter Lock (GIL)\n",
    "As I mentioned earlier, the way Python interpreter is implemented, it is very difficult to achieve perfect parallelism using threading. This is due to something called Global Interpreter Lock.\n",
    "\n",
    "GIL\n",
    "The Python Global Interpreter Lock or GIL, in simple words, is a mutex (or a lock) that allows only one thread to hold the control of the Python interpreter. Only one thread can be in a state of execution at any point in time.\n",
    "\n",
    "Mutex\n",
    "Mutex is a mutual exclusion object that synchronizes access to a resource. It is created with a unique name at the start of a program. The Mutex is a locking mechanism that makes sure only one thread can acquire the Mutex at a time and enter the critical section.\n",
    "\n",
    "This basically defeats the whole purpose of using threading in the first place. Which is why we have something in PyTorch that can be used to achieve perfect parallelism.\n",
    "\n",
    "\"\"\""
   ]
  },
  {
   "cell_type": "markdown",
   "id": "a4e2f77f",
   "metadata": {},
   "source": [
    "\"\"\"Distributed Data Parallel in PyTorch\n",
    "DDP in PyTorch does the same thing but in a much proficient way and also gives us better control while achieving perfect parallelism. DDP uses multiprocessing instead of threading and executes propagation through the model as a different process for each GPU. DDP duplicates the model across multiple GPUs, each of which is controlled by one process. A process here can be called a script that runs on your system. Usually we spawn processes such that there is a separate process for each GPU.\n",
    "\n",
    "Each of the process here does identical tasks but with different batch of data. Each process communicates with other processes to share gradients which needs to be all-reduced during the optimization step. At the end of an optimization step each process has averaged gradients, ensuring that the model weights stay synchronized.\n",
    "\"\"\""
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "16650e6c",
   "metadata": {},
   "outputs": [],
   "source": [
    "“node” is a system in your distributed architecture. In lay man’s terms, a single system that has multiple GPUs can be called as a node.\n",
    "\n",
    "“global rank” is a unique identification number for each node in our architecture.\n",
    "\n",
    "“local rank” is a unique identification number for processes in each node.\n",
    "\n",
    "“world” is a union of all of the above which can have multiple nodes where each node spawns multiple processes. (Ideally, one for each GPU)\n",
    "\n",
    "“world_size” is equal to number of nodes * number of gpus"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "id": "40b76317",
   "metadata": {
    "ExecuteTime": {
     "end_time": "2022-08-31T15:21:46.108910Z",
     "start_time": "2022-08-31T15:21:44.806859Z"
    }
   },
   "outputs": [],
   "source": [
    "import torch\n",
    "import torch.nn as nn\n",
    "import torch.distributed as dist\n",
    "import torchvision\n",
    "import torchvision.transforms as transforms\n",
    "\n",
    "import argparse\n",
    "from datetime import datetime\n",
    "import os\n",
    "import torch.multiprocessing as mp"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "id": "74d3abe4",
   "metadata": {
    "ExecuteTime": {
     "end_time": "2022-08-31T15:21:52.499470Z",
     "start_time": "2022-08-31T15:21:52.481851Z"
    }
   },
   "outputs": [],
   "source": [
    "class ConvNet(nn.Module):\n",
    "    def __init__(self, num_classes=10):\n",
    "        super(ConvNet, self).__init__()\n",
    "        self.layer1 = nn.Sequential(\n",
    "            nn.Conv2d(1, 16, kernel_size=5, stride=1, padding=2),\n",
    "            nn.BatchNorm2d(16),\n",
    "            nn.ReLU(),\n",
    "            nn.MaxPool2d(kernel_size=2, stride=2))\n",
    "        self.layer2 = nn.Sequential(\n",
    "            nn.Conv2d(16, 32, kernel_size=5, stride=1, padding=2),\n",
    "            nn.BatchNorm2d(32),\n",
    "            nn.ReLU(),\n",
    "            nn.MaxPool2d(kernel_size=2, stride=2))\n",
    "        self.fc = nn.Linear(7*7*32, num_classes)\n",
    "\n",
    "    def forward(self, x):\n",
    "        out = self.layer1(x)\n",
    "        out = self.layer2(out)\n",
    "        out = out.reshape(out.size(0), -1)\n",
    "        out = self.fc(out)\n",
    "        return out"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "0d798da9",
   "metadata": {},
   "outputs": [],
   "source": [
    "\n",
    "def initialize():\n",
    "    nodes = 1\n",
    "    nr = 0\n",
    "    gpus = torch.cuda.device_count()\n",
    "    epochs = 2\n",
    "    world_size = gpus * nodes\n",
    "#     parser = argparse.ArgumentParser()\n",
    "#     parser.add_argument('-n', '--nodes', default=1, type=int, metavar='N')\n",
    "#     parser.add_argument('-g', '--gpus', default=1, type=int,\n",
    "#                         help='number of gpus per node')\n",
    "#     parser.add_argument('-nr', '--nr', default=0, type=int,\n",
    "#                         help='ranking within the nodes')\n",
    "#     parser.add_argument('--epochs', default=2, type=int, metavar='N',\n",
    "#                         help='number of total epochs to run')\n",
    "#     args = parser.parse_args()\n",
    "\n",
    "#     args.world_size = args.gpus * args.nodes\n",
    "    os.environ['MASTER_ADDR'] = '192.168.1.3'\n",
    "    os.environ['MASTER_PORT'] = '8888'\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "d5dfde43",
   "metadata": {},
   "outputs": [],
   "source": [
    "def train(gpu, args):\n",
    "\n",
    "    rank = args.nr * args.gpus + gpu\n",
    "    dist.init_process_group(\n",
    "    \tbackend='nccl',\n",
    "        init_method='env://',\n",
    "    \tworld_size=args.world_size,\n",
    "    \trank=rank\n",
    "    )\n",
    "\n",
    "    torch.manual_seed(0)\n",
    "    model = ConvNet()\n",
    "    torch.cuda.set_device(gpu)\n",
    "    model.cuda(gpu)\n",
    "    batch_size = 100\n",
    "    # define loss function (criterion) and optimizer\n",
    "    criterion = nn.CrossEntropyLoss().cuda(gpu)\n",
    "    optimizer = torch.optim.SGD(model.parameters(), 1e-4)\n",
    "\n",
    "    # Wrapper around our model to handle parallel training\n",
    "    model = nn.parallel.DistributedDataParallel(model, device_ids=[gpu])\n",
    "\n",
    "    # Data loading code\n",
    "    train_dataset = torchvision.datasets.MNIST(root='./',\n",
    "                                               train=True,\n",
    "                                               transform=transforms.ToTensor(),\n",
    "                                               download=True)\n",
    "    \n",
    "    # Sampler that takes care of the distribution of the batches such that\n",
    "    # the data is not repeated in the iteration and sampled accordingly\n",
    "    train_sampler = torch.utils.data.distributed.DistributedSampler(\n",
    "    \ttrain_dataset,\n",
    "    \tnum_replicas=args.world_size,\n",
    "    \trank=rank\n",
    "    )\n",
    "    \n",
    "    # We pass in the train_sampler which can be used by the DataLoader\n",
    "    train_loader = torch.utils.data.DataLoader(dataset=train_dataset,\n",
    "                                               batch_size=batch_size,\n",
    "                                               shuffle=False,\n",
    "                                               num_workers=0,\n",
    "                                               pin_memory=True,\n",
    "                                               sampler=train_sampler)\n",
    "\n",
    "    start = datetime.now()\n",
    "    total_step = len(train_loader)\n",
    "    for epoch in range(args.epochs):\n",
    "        for i, (images, labels) in enumerate(train_loader):\n",
    "            images = images.cuda(non_blocking=True)\n",
    "            labels = labels.cuda(non_blocking=True)\n",
    "            # Forward pass\n",
    "            outputs = model(images)\n",
    "            loss = criterion(outputs, labels)\n",
    "\n",
    "            # Backward and optimize\n",
    "            optimizer.zero_grad()\n",
    "            loss.backward()\n",
    "            optimizer.step()\n",
    "            if (i + 1) % 100 == 0 and gpu == 0:\n",
    "                print('Epoch [{}/{}], Step [{}/{}], Loss: {:.4f}'.format(\n",
    "                    epoch + 1, \n",
    "                    args.epochs, \n",
    "                    i + 1, \n",
    "                    total_step,\n",
    "                    loss.item())\n",
    "                   )\n",
    "    if gpu == 0:\n",
    "        print(\"Training complete in: \" + str(datetime.now() - start))"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "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.6.9"
  },
  "toc": {
   "base_numbering": 1,
   "nav_menu": {},
   "number_sections": true,
   "sideBar": true,
   "skip_h1_title": false,
   "title_cell": "Table of Contents",
   "title_sidebar": "Contents",
   "toc_cell": false,
   "toc_position": {},
   "toc_section_display": true,
   "toc_window_display": false
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}