7.分布式训练.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "initial_id",
   "metadata": {
    "collapsed": true,
    "jupyter": {
     "outputs_hidden": true
    }
   },
   "outputs": [],
   "source": [
    "# 单机多卡：NCCL通信协议，torch.nn.parallel.DistributedDataParallel\n",
    "# 多机多卡：先考虑NCCL通信协议，不行再考虑Gloo通信协议; 类似与map reduce过程\n",
    "\n",
    "# 同一机器GPU之间的通信带宽 > CPU-GPU带宽 > 机器之间网络\n",
    "\n",
    "# nccl: GPU之间的通信（英伟达）\n",
    "# mpi: CPU和GPU之间的通信"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "id": "95e3da4bc685bc40",
   "metadata": {
    "ExecuteTime": {
     "end_time": "2023-08-23T03:37:11.902987900Z",
     "start_time": "2023-08-23T03:37:11.869653200Z"
    },
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "1 0.3859357237815857\n",
      "2 0.39890220761299133\n",
      "3 0.2811512351036072\n",
      "4 0.873574435710907\n",
      "5 0.42149585485458374\n",
      "6 0.4353399872779846\n",
      "7 0.421011745929718\n",
      "8 0.44859200716018677\n",
      "9 0.6005834937095642\n",
      "10 0.2511233687400818\n"
     ]
    }
   ],
   "source": [
    "# 1.数据并行DataParallel：模型和优化器需要进一步包装。\n",
    "# model = DataParallel(model, device_ids=[0,1], output_device=0) \n",
    "# optimizer = DataParallel(optimizer, device_ids=[0,1], output_device=0)\n",
    "# 单机多卡：多gpu并行计算，不支持多机器。好处是改动代码很少。\n",
    "# 前向过程：batch数据会送到指定的GPU上（所以要注意batch_size大于能整除GPU个数）\n",
    "# 反向传播：每个副本梯度累加到指定GPU上汇总（下面是0，默认也是0，所以第一块显卡的占用内存会多一些，类似于参数服务器）\n",
    "\n",
    "# 许多低效率之处：\n",
    "# 1.冗余数据副本：数据从主机复制到主GPU，然后将子微型批分散在其他GPU上\n",
    "# 2.在前向传播之前跨GPU进行模型复制：由于模型参数是在主GPU上更新的，因此模型必须在每次正向传递的开始时重新同步\n",
    "# 3.每批的线程创建/销毁开销：并行转发是在多个线程中实现的（这可能只是PyTorch问题）\n",
    "# 4.梯度减少流水线机会未开发：在Pytorch 1.0数据并行实现中，梯度下降发生在反向传播的末尾。\n",
    "# 5.在主GPU上不必要地收集模型输出output\n",
    "# 6.GPU利用率不均：在主GPU上执行损失loss计算\n",
    "# 7.梯度下降，在主GPU上更新参数\n",
    "\n",
    "# 单进程多线程性能被python的GIL锁约束。\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "id": "2194235e9db04eb",
   "metadata": {
    "ExecuteTime": {
     "end_time": "2023-08-23T12:47:04.143872Z",
     "start_time": "2023-08-23T12:46:59.813740Z"
    },
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "30\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "30it [00:06,  4.64it/s]\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Finish  1  Loss: 5.958403, Acc: 0.332633\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "30it [00:06,  4.71it/s]\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Finish  2  Loss: 0.374562, Acc: 0.885100\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "30it [00:06,  4.72it/s]\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Finish  3  Loss: 0.138205, Acc: 0.957133\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "30it [00:06,  4.73it/s]\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Finish  4  Loss: 0.089404, Acc: 0.972033\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "30it [00:06,  4.76it/s]\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Finish  5  Loss: 0.070034, Acc: 0.978483\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "30it [00:06,  4.74it/s]\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Finish  6  Loss: 0.058024, Acc: 0.981717\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "30it [00:06,  4.66it/s]\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Finish  7  Loss: 0.049302, Acc: 0.984400\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "30it [00:06,  4.73it/s]\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Finish  8  Loss: 0.040589, Acc: 0.987650\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "30it [00:06,  4.72it/s]\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Finish  9  Loss: 0.032878, Acc: 0.989967\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "30it [00:06,  4.65it/s]"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Finish  10  Loss: 0.030534, Acc: 0.990367\n",
      "with only one GPU, it cost time: 63.787768602371216 (s)!\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "\n"
     ]
    }
   ],
   "source": [
    "# 1.1 标准单机单卡版本：cnn on minist\n",
    "import torch, time, os\n",
    "from torch import nn, optim\n",
    "from torchvision import datasets, transforms\n",
    "from torch.autograd import Variable\n",
    "from torch.utils.data import DataLoader\n",
    "from tqdm import tqdm\n",
    "\n",
    "os.environ[\"CUDA_VISIBLE_DEVICES\"] = \"0\"\n",
    "\n",
    "class SimpleCNN(nn.Module):\n",
    "    def __init__(self):\n",
    "        super(SimpleCNN, self).__init__()\n",
    "        self.layer1 = nn.Sequential(nn.Conv2d(1,16,kernel_size=3), nn.BatchNorm2d(16), nn.ReLU(inplace=True))\n",
    "        self.layer2 = nn.Sequential(nn.Conv2d(16,32,kernel_size=3), nn.BatchNorm2d(32), nn.ReLU(inplace=True), nn.MaxPool2d(kernel_size=2 , stride=2))\n",
    "        self.layer3 = nn.Sequential(nn.Conv2d(32,64,kernel_size=3), nn.BatchNorm2d(64), nn.ReLU(inplace=True))\n",
    "        self.layer4 = nn.Sequential(nn.Conv2d(64,128,kernel_size=3), nn.BatchNorm2d(128), nn.ReLU(inplace=True), nn.MaxPool2d(kernel_size=2 , stride=2))\n",
    "        self.fc = nn.Sequential(nn.Linear(128*4*4,1024), nn.ReLU(inplace=True), nn.Linear(1024,128), nn.ReLU(inplace=True), nn.Linear(128,10))\n",
    "    def forward( self , x):\n",
    "        x = self.layer4(self.layer3(self.layer2(self.layer1(x))))\n",
    "        x = x.reshape(x.size(0) , -1)\n",
    "        fc_out = self.fc(x)\n",
    "        return fc_out\n",
    "\n",
    "# 定义超参数\n",
    "learning_rate = 1e-2      # 学习率\n",
    "batch_size    = 2000       # 批的大小\n",
    "epoches_num   = 10        # 遍历训练集的次数\n",
    "# 下载训练集 MNIST 手写数字训练集\n",
    "train_dataset = datasets.MNIST( root='./data', train=True, transform=transforms.ToTensor(), download=True)\n",
    "train_loader  = DataLoader( train_dataset, batch_size=batch_size, shuffle=True)\n",
    "\n",
    "print(len(train_loader))\n",
    "\n",
    "# 定义model 、loss 、optimizer\n",
    "model = SimpleCNN().to('cuda:0')\n",
    "criterion = nn.CrossEntropyLoss().to('cuda:0')\n",
    "optimizer = optim.Adam(model.parameters(), lr=learning_rate )\n",
    "# 开始训练\n",
    "model.train()\n",
    "starttime = time.time()\n",
    "for epoch in range(epoches_num):\n",
    "    train_loss = 0.0\n",
    "    train_acc  = 0.0\n",
    "    # 训练\n",
    "    for i, data in tqdm(enumerate(train_loader, 1)):\n",
    "        img, label = data\n",
    "        img   = img.to('cuda:0')\n",
    "        label = label.to('cuda:0')\n",
    "        # 前向传播\n",
    "        optimizer.zero_grad()\n",
    "        out  = model(img)\n",
    "        loss = criterion(out, label)\n",
    "        # 反向传播\n",
    "        loss.backward()\n",
    "        optimizer.step()\n",
    "        # 损失/准确率计算\n",
    "        train_loss += loss.item() * label.size(0)\n",
    "        _ , pred    = out.max(1)\n",
    "        num_correct = pred.eq(label).sum()\n",
    "        accuracy    = pred.eq(label).float().mean()\n",
    "        train_acc  += num_correct.item()\n",
    "    print('Finish  {}  Loss: {:.6f}, Acc: {:.6f}'.format( epoch+1 , train_loss / len(train_dataset), train_acc / len(train_dataset )))\n",
    "endtime = time.time()\n",
    "print('with only one GPU, it cost time:', endtime - starttime, '(s)!')\n",
    "# 保存模型\n",
    "torch.save(model, './model/cnn_one_gpu.pt')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "id": "afff19ba637e71e5",
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "15\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "15it [00:09,  1.62it/s]\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Finish  1  Loss: 8.045089, Acc: 0.147517\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "15it [00:05,  2.58it/s]\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Finish  2  Loss: 1.237980, Acc: 0.569300\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "15it [00:05,  2.57it/s]\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Finish  3  Loss: 0.538698, Acc: 0.814600\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "15it [00:05,  2.57it/s]\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Finish  4  Loss: 0.281035, Acc: 0.907167\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "15it [00:05,  2.57it/s]\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Finish  5  Loss: 0.148703, Acc: 0.953367\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "15it [00:05,  2.59it/s]\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Finish  6  Loss: 0.097726, Acc: 0.969883\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "15it [00:05,  2.60it/s]\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Finish  7  Loss: 0.074189, Acc: 0.977117\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "15it [00:05,  2.58it/s]\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Finish  8  Loss: 0.060719, Acc: 0.981200\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "15it [00:05,  2.58it/s]\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Finish  9  Loss: 0.051184, Acc: 0.984583\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "15it [00:05,  2.57it/s]"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Finish  10  Loss: 0.044030, Acc: 0.986583\n",
      "with only two GPUs, it cost time: 61.6509530544281 (s)!\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "\n"
     ]
    }
   ],
   "source": [
    "# 1.2 单机多卡版本：cnn on minist 采用DataParallel来实现（多进程）\n",
    "\n",
    "import torch, time, os\n",
    "from torch import nn, optim\n",
    "from torchvision import datasets, transforms\n",
    "from torch.autograd import Variable\n",
    "from torch.utils.data import DataLoader\n",
    "from tqdm import tqdm\n",
    "from torch.nn.parallel import DataParallel\n",
    "\n",
    "os.environ[\"CUDA_DEVICE_ORDER\"] = \"PCI_BUS_ID\"\n",
    "os.environ[\"CUDA_VISIBLE_DEVICES\"] = \"0, 1\"\n",
    "\n",
    "device_ids = [0,1]\n",
    "main_device = 0\n",
    "\n",
    "class SimpleCNN(nn.Module):\n",
    "    def __init__(self):\n",
    "        super(SimpleCNN, self).__init__()\n",
    "        self.layer1 = nn.Sequential(nn.Conv2d(1,16,kernel_size=3), nn.BatchNorm2d(16), nn.ReLU(inplace=True))\n",
    "        self.layer2 = nn.Sequential(nn.Conv2d(16,32,kernel_size=3), nn.BatchNorm2d(32), nn.ReLU(inplace=True), nn.MaxPool2d(kernel_size=2 , stride=2))\n",
    "        self.layer3 = nn.Sequential(nn.Conv2d(32,64,kernel_size=3), nn.BatchNorm2d(64), nn.ReLU(inplace=True))\n",
    "        self.layer4 = nn.Sequential(nn.Conv2d(64,128,kernel_size=3), nn.BatchNorm2d(128), nn.ReLU(inplace=True), nn.MaxPool2d(kernel_size=2 , stride=2))\n",
    "        self.fc = nn.Sequential(nn.Linear(128*4*4,1024), nn.ReLU(inplace=True), nn.Linear(1024,128), nn.ReLU(inplace=True), nn.Linear(128,10))\n",
    "    def forward( self , x):\n",
    "        x = self.layer4(self.layer3(self.layer2(self.layer1(x))))\n",
    "        x = x.reshape(x.size(0) , -1)\n",
    "        fc_out = self.fc(x)\n",
    "        return fc_out\n",
    "\n",
    "# 定义超参数\n",
    "learning_rate = 1e-2      # 学习率\n",
    "batch_size    = 4000       # 批的大小\n",
    "epoches_num   = 10        # 遍历训练集的次数\n",
    "# 下载训练集 MNIST 手写数字训练集\n",
    "train_dataset = datasets.MNIST(root='./data', train=True, transform=transforms.ToTensor(), download=True )\n",
    "train_loader  = DataLoader( train_dataset, batch_size=batch_size, shuffle=True)\n",
    "\n",
    "print(len(train_loader))\n",
    "\n",
    "# 定义model 、loss 、optimizer\n",
    "model = DataParallel(module=SimpleCNN(), device_ids = device_ids, output_device=main_device).cuda()\n",
    "criterion = nn.CrossEntropyLoss().cuda()\n",
    "optimizer = optim.Adam(model.parameters(), lr=learning_rate)\n",
    "# 开始训练\n",
    "model.train()\n",
    "starttime = time.time()\n",
    "for epoch in range(epoches_num):\n",
    "    train_loss = 0.0\n",
    "    train_acc  = 0.0\n",
    "    # 训练\n",
    "    for i, data in tqdm(enumerate(train_loader, 1)):\n",
    "        img, label = data\n",
    "        img   = img.cuda(non_blocking=True)\n",
    "        label = label.cuda(non_blocking=True)\n",
    "        # 前向传播\n",
    "        optimizer.zero_grad()\n",
    "        out  = model(img)\n",
    "        loss = criterion(out, label)\n",
    "        # 反向传播\n",
    "        loss.backward()\n",
    "        optimizer.step()\n",
    "        # 损失/准确率计算\n",
    "        train_loss += loss.item() * label.size(0)\n",
    "        _ , pred    = out.max(1)\n",
    "        num_correct = pred.eq(label).sum()\n",
    "        accuracy    = pred.eq(label).float().mean()\n",
    "        train_acc  += num_correct.item()\n",
    "    print('Finish  {}  Loss: {:.6f}, Acc: {:.6f}'.format( epoch+1 , train_loss / len(train_dataset), train_acc / len(train_dataset )))\n",
    "endtime = time.time()\n",
    "print('with only two GPUs, it cost time:', endtime - starttime, '(s)!')\n",
    "# 保存模型\n",
    "torch.save(model, './model/cnn_two_gpus_dp.pt')\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "outputs": [],
   "source": [
    "# 2.torch.nn.parallel.DistributedDataParallel(DDP)分布式训练\n",
    "# 每个gpu对应一个进程\n",
    "# ring all-reduce算法（百度提出）：保证GPU之间model和optimizer同步一致。所有的GPU连成一个ring，不需要等待所有卡计算完一轮。分布式同步。\n",
    "# 算法本质是充分利用带宽，每部分都要走一圈ring，错开走。ring上相邻两个GPU之间每步只传一部分。\n",
    "# 参数：world_size, ranks值范围为[0,world_size-1], local_rank\n",
    "# 需要定义DistributedDataParallel和DistributedSampler\n"
   ],
   "metadata": {
    "collapsed": false
   },
   "id": "ebfca1b4df7f1d65"
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "outputs": [
    {
     "ename": "RuntimeError",
     "evalue": "Distributed package doesn't have NCCL built in",
     "output_type": "error",
     "traceback": [
      "\u001B[0;31m---------------------------------------------------------------------------\u001B[0m",
      "\u001B[0;31mRuntimeError\u001B[0m                              Traceback (most recent call last)",
      "Input \u001B[0;32mIn [4]\u001B[0m, in \u001B[0;36m<cell line: 58>\u001B[0;34m()\u001B[0m\n\u001B[1;32m     59\u001B[0m local_rank \u001B[38;5;241m=\u001B[39m \u001B[38;5;241m0\u001B[39m\n\u001B[1;32m     60\u001B[0m gpu_ids \u001B[38;5;241m=\u001B[39m [\u001B[38;5;241m0\u001B[39m,\u001B[38;5;241m1\u001B[39m,\u001B[38;5;241m2\u001B[39m]\n\u001B[0;32m---> 61\u001B[0m \u001B[43mddp_setup\u001B[49m\u001B[43m(\u001B[49m\u001B[43mlocal_rank\u001B[49m\u001B[43m,\u001B[49m\u001B[43m \u001B[49m\u001B[43mworld_size\u001B[49m\u001B[43m)\u001B[49m\n\u001B[1;32m     62\u001B[0m sampler \u001B[38;5;241m=\u001B[39m DistributedSampler(dataset, shuffle\u001B[38;5;241m=\u001B[39m\u001B[38;5;28;01mTrue\u001B[39;00m)\n\u001B[1;32m     63\u001B[0m dataloader \u001B[38;5;241m=\u001B[39m DataLoader(dataset\u001B[38;5;241m=\u001B[39mTensorDataset(X,Y), batch_size\u001B[38;5;241m=\u001B[39mbatch_size, shuffle\u001B[38;5;241m=\u001B[39m\u001B[38;5;28;01mFalse\u001B[39;00m, sampler\u001B[38;5;241m=\u001B[39msampler) \u001B[38;5;66;03m# 注意这里\u001B[39;00m\n",
      "Input \u001B[0;32mIn [4]\u001B[0m, in \u001B[0;36mddp_setup\u001B[0;34m(local_rank, world_size)\u001B[0m\n\u001B[1;32m     34\u001B[0m os\u001B[38;5;241m.\u001B[39menviron[\u001B[38;5;124m'\u001B[39m\u001B[38;5;124mMASTER_PORT\u001B[39m\u001B[38;5;124m'\u001B[39m] \u001B[38;5;241m=\u001B[39m \u001B[38;5;124m'\u001B[39m\u001B[38;5;124m12355\u001B[39m\u001B[38;5;124m'\u001B[39m\n\u001B[1;32m     35\u001B[0m \u001B[38;5;66;03m# world_size：总的进程数\u001B[39;00m\n\u001B[1;32m     36\u001B[0m \u001B[38;5;66;03m# rank: 进程号，代表GPU号\u001B[39;00m\n\u001B[1;32m     37\u001B[0m \u001B[38;5;66;03m# gpu间通信协议：NCCL\u001B[39;00m\n\u001B[1;32m     38\u001B[0m \u001B[38;5;66;03m# init_process_group(backend='gloo', rank=local_rank, world_size=world_size)\u001B[39;00m\n\u001B[0;32m---> 39\u001B[0m \u001B[43minit_process_group\u001B[49m\u001B[43m(\u001B[49m\u001B[43mbackend\u001B[49m\u001B[38;5;241;43m=\u001B[39;49m\u001B[38;5;124;43m'\u001B[39;49m\u001B[38;5;124;43mnccl\u001B[39;49m\u001B[38;5;124;43m'\u001B[39;49m\u001B[43m,\u001B[49m\u001B[43m \u001B[49m\u001B[43mrank\u001B[49m\u001B[38;5;241;43m=\u001B[39;49m\u001B[43mlocal_rank\u001B[49m\u001B[43m,\u001B[49m\u001B[43m \u001B[49m\u001B[43mworld_size\u001B[49m\u001B[38;5;241;43m=\u001B[39;49m\u001B[43mworld_size\u001B[49m\u001B[43m)\u001B[49m\n\u001B[1;32m     40\u001B[0m torch\u001B[38;5;241m.\u001B[39mcuda\u001B[38;5;241m.\u001B[39mset_device(local_rank)\n",
      "File \u001B[0;32m~/opt/anaconda3/lib/python3.9/site-packages/torch/distributed/distributed_c10d.py:907\u001B[0m, in \u001B[0;36minit_process_group\u001B[0;34m(backend, init_method, timeout, world_size, rank, store, group_name, pg_options)\u001B[0m\n\u001B[1;32m    903\u001B[0m         \u001B[38;5;66;03m# Use a PrefixStore to avoid accidental overrides of keys used by\u001B[39;00m\n\u001B[1;32m    904\u001B[0m         \u001B[38;5;66;03m# different systems (e.g. RPC) in case the store is multi-tenant.\u001B[39;00m\n\u001B[1;32m    905\u001B[0m         store \u001B[38;5;241m=\u001B[39m PrefixStore(\u001B[38;5;124m\"\u001B[39m\u001B[38;5;124mdefault_pg\u001B[39m\u001B[38;5;124m\"\u001B[39m, store)\n\u001B[0;32m--> 907\u001B[0m     default_pg \u001B[38;5;241m=\u001B[39m \u001B[43m_new_process_group_helper\u001B[49m\u001B[43m(\u001B[49m\n\u001B[1;32m    908\u001B[0m \u001B[43m        \u001B[49m\u001B[43mworld_size\u001B[49m\u001B[43m,\u001B[49m\n\u001B[1;32m    909\u001B[0m \u001B[43m        \u001B[49m\u001B[43mrank\u001B[49m\u001B[43m,\u001B[49m\n\u001B[1;32m    910\u001B[0m \u001B[43m        \u001B[49m\u001B[43m[\u001B[49m\u001B[43m]\u001B[49m\u001B[43m,\u001B[49m\n\u001B[1;32m    911\u001B[0m \u001B[43m        \u001B[49m\u001B[43mbackend\u001B[49m\u001B[43m,\u001B[49m\n\u001B[1;32m    912\u001B[0m \u001B[43m        \u001B[49m\u001B[43mstore\u001B[49m\u001B[43m,\u001B[49m\n\u001B[1;32m    913\u001B[0m \u001B[43m        \u001B[49m\u001B[43mpg_options\u001B[49m\u001B[38;5;241;43m=\u001B[39;49m\u001B[43mpg_options\u001B[49m\u001B[43m,\u001B[49m\n\u001B[1;32m    914\u001B[0m \u001B[43m        \u001B[49m\u001B[43mgroup_name\u001B[49m\u001B[38;5;241;43m=\u001B[39;49m\u001B[43mgroup_name\u001B[49m\u001B[43m,\u001B[49m\n\u001B[1;32m    915\u001B[0m \u001B[43m        \u001B[49m\u001B[43mtimeout\u001B[49m\u001B[38;5;241;43m=\u001B[39;49m\u001B[43mtimeout\u001B[49m\u001B[43m,\u001B[49m\n\u001B[1;32m    916\u001B[0m \u001B[43m    \u001B[49m\u001B[43m)\u001B[49m\n\u001B[1;32m    917\u001B[0m     _update_default_pg(default_pg)\n\u001B[1;32m    919\u001B[0m _world\u001B[38;5;241m.\u001B[39mpg_group_ranks[GroupMember\u001B[38;5;241m.\u001B[39mWORLD] \u001B[38;5;241m=\u001B[39m {i: i \u001B[38;5;28;01mfor\u001B[39;00m i \u001B[38;5;129;01min\u001B[39;00m \u001B[38;5;28mrange\u001B[39m(GroupMember\u001B[38;5;241m.\u001B[39mWORLD\u001B[38;5;241m.\u001B[39msize())}  \u001B[38;5;66;03m# type: ignore[attr-defined, index]\u001B[39;00m\n",
      "File \u001B[0;32m~/opt/anaconda3/lib/python3.9/site-packages/torch/distributed/distributed_c10d.py:1013\u001B[0m, in \u001B[0;36m_new_process_group_helper\u001B[0;34m(group_size, group_rank, global_ranks_in_group, backend, store, pg_options, group_name, timeout)\u001B[0m\n\u001B[1;32m   1011\u001B[0m \u001B[38;5;28;01melif\u001B[39;00m backend_str \u001B[38;5;241m==\u001B[39m Backend\u001B[38;5;241m.\u001B[39mNCCL:\n\u001B[1;32m   1012\u001B[0m     \u001B[38;5;28;01mif\u001B[39;00m \u001B[38;5;129;01mnot\u001B[39;00m is_nccl_available():\n\u001B[0;32m-> 1013\u001B[0m         \u001B[38;5;28;01mraise\u001B[39;00m \u001B[38;5;167;01mRuntimeError\u001B[39;00m(\u001B[38;5;124m\"\u001B[39m\u001B[38;5;124mDistributed package doesn\u001B[39m\u001B[38;5;124m'\u001B[39m\u001B[38;5;124mt have NCCL \u001B[39m\u001B[38;5;124m\"\u001B[39m \u001B[38;5;124m\"\u001B[39m\u001B[38;5;124mbuilt in\u001B[39m\u001B[38;5;124m\"\u001B[39m)\n\u001B[1;32m   1014\u001B[0m     \u001B[38;5;28;01mif\u001B[39;00m pg_options \u001B[38;5;129;01mis\u001B[39;00m \u001B[38;5;129;01mnot\u001B[39;00m \u001B[38;5;28;01mNone\u001B[39;00m:\n\u001B[1;32m   1015\u001B[0m         \u001B[38;5;28;01massert\u001B[39;00m \u001B[38;5;28misinstance\u001B[39m(\n\u001B[1;32m   1016\u001B[0m             pg_options, ProcessGroupNCCL\u001B[38;5;241m.\u001B[39mOptions\n\u001B[1;32m   1017\u001B[0m         ), \u001B[38;5;124m\"\u001B[39m\u001B[38;5;124mExpected pg_options argument to be of type ProcessGroupNCCL.Options\u001B[39m\u001B[38;5;124m\"\u001B[39m\n",
      "\u001B[0;31mRuntimeError\u001B[0m: Distributed package doesn't have NCCL built in"
     ]
    }
   ],
   "source": [
    "# 2.1. 单机多卡简单案例实现（网上例子）（成功执行）\n",
    "\n",
    "import torch\n",
    "import torch.nn as nn\n",
    "from torch.autograd import Variable\n",
    "from torch.utils.data import Dataset, DataLoader\n",
    "import os\n",
    "from torch.utils.data.distributed import DistributedSampler\n",
    "# 1) 初始化\n",
    "torch.distributed.init_process_group(backend=\"nccl\")\n",
    "\n",
    "input_size = 5\n",
    "output_size = 2\n",
    "batch_size = 30\n",
    "data_size = 90\n",
    "epoches = 10\n",
    "\n",
    "# 2） 配置每个进程的gpu\n",
    "local_rank = torch.distributed.get_rank()\n",
    "torch.cuda.set_device(local_rank)\n",
    "device = torch.device(\"cuda\", local_rank)\n",
    "\n",
    "class RandomDataset(Dataset):\n",
    "    def __init__(self, size, length):\n",
    "        self.len = length\n",
    "        self.data = torch.randn(length, size).to('cuda')\n",
    "\n",
    "    def __getitem__(self, index):\n",
    "        return self.data[index]\n",
    "\n",
    "    def __len__(self):\n",
    "        return self.len\n",
    "dataset = RandomDataset(input_size, data_size)\n",
    "# 3）使用DistributedSampler\n",
    "rand_loader = DataLoader(dataset=dataset, batch_size=batch_size, sampler=DistributedSampler(dataset))\n",
    "\n",
    "class Model(nn.Module):\n",
    "    def __init__(self, input_size, output_size):\n",
    "        super(Model, self).__init__()\n",
    "        self.fc = nn.Linear(input_size, output_size)\n",
    "    def forward(self, input):\n",
    "        output = self.fc(input)\n",
    "        print(\"  In Model: input size\", input.size(),\n",
    "              \"output size\", output.size())\n",
    "        return output\n",
    "\n",
    "model = Model(input_size, output_size)\n",
    "\n",
    "# 4) 封装之前要把模型移到对应的gpu\n",
    "model.to(device)\n",
    "if torch.cuda.device_count() > 1:\n",
    "    print(\"Let's use\", torch.cuda.device_count(), \"GPUs!\")\n",
    "    # 5) 封装\n",
    "    model = torch.nn.parallel.DistributedDataParallel(model, device_ids=[local_rank], output_device=local_rank)\n",
    "for epoch in range(epoches):\n",
    "    for data in rand_loader:\n",
    "        if torch.cuda.is_available():\n",
    "            input_var = data\n",
    "        else:\n",
    "            input_var = data\n",
    "        output = model(input_var)\n",
    "        print(\"Outside: input size\", input_var.size(), \"output_size\", output.size())\n",
    "\n",
    "# 单机多卡1：命令，成功执行\n",
    "# CUDA_VISIBLE_DEVICES=0,1 python -m torch.distributed.launch --nproc_per_node=2 ddp.py"
   ],
   "metadata": {
    "collapsed": false,
    "ExecuteTime": {
     "start_time": "2023-08-23T12:00:14.023009Z"
    }
   },
   "id": "786caf1a7593a96b"
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "outputs": [],
   "source": [
    "# 2.2 单机多卡简单版本 cnn on minist（成功执行）\n",
    "\n",
    "import torch\n",
    "import torch.nn as nn\n",
    "from torch.autograd import Variable\n",
    "from torch.utils.data import Dataset, DataLoader\n",
    "from torchvision import datasets, transforms\n",
    "import os\n",
    "from torch.utils.data.distributed import DistributedSampler\n",
    "# 1) 初始化\n",
    "# torch.distributed.init_process_group(backend=\"nccl\") # 推荐这个，两个gpu占用率才7%左右\n",
    "# torch.distributed.init_process_group(backend=\"nccl\", init_method='env://') # 报错ERROR:torch.distributed.elastic.multiprocessing.api:failed (exitcode: -6) local_rank: 0 (pid: 3596) of binary: /root/miniconda3/bin/python\n",
    "torch.distributed.init_process_group(backend=\"nccl\", world_size=2) # 推荐这个,gpu:1占用率会偶尔飙升一下\n",
    "# torch.distributed.init_process_group(backend=\"nccl\", rank=0, world_size=2)  # 这个有问题，只用了gpu:0\n",
    "\n",
    "# 定义超参数\n",
    "learning_rate = 1e-2      # 学习率\n",
    "batch_size    = 1000       # 批的大小\n",
    "epoches_num   = 10        # 遍历训练集的次数\n",
    "\n",
    "# 2） 配置每个进程的gpu\n",
    "local_rank = torch.distributed.get_rank()\n",
    "torch.cuda.set_device(local_rank)\n",
    "device = torch.device(\"cuda\", local_rank)\n",
    "\n",
    "# 3）使用DistributedSampler\n",
    "dataset = datasets.MNIST(root='./data', train=True, transform=transforms.ToTensor(), download=True )\n",
    "sampler=DistributedSampler(dataset)\n",
    "rand_loader = DataLoader(dataset=dataset, batch_size=batch_size, sampler=sampler)\n",
    "\n",
    "class SimpleCNN(nn.Module):\n",
    "    def __init__(self):\n",
    "        super(SimpleCNN, self).__init__()\n",
    "        self.layer1 = nn.Sequential(nn.Conv2d(1,16,kernel_size=3), nn.BatchNorm2d(16), nn.ReLU(inplace=True))\n",
    "        self.layer2 = nn.Sequential(nn.Conv2d(16,32,kernel_size=3), nn.BatchNorm2d(32), nn.ReLU(inplace=True), nn.MaxPool2d(kernel_size=2 , stride=2))\n",
    "        self.layer3 = nn.Sequential(nn.Conv2d(32,64,kernel_size=3), nn.BatchNorm2d(64), nn.ReLU(inplace=True))\n",
    "        self.layer4 = nn.Sequential(nn.Conv2d(64,128,kernel_size=3), nn.BatchNorm2d(128), nn.ReLU(inplace=True), nn.MaxPool2d(kernel_size=2 , stride=2))\n",
    "        self.fc = nn.Sequential(nn.Linear(128*4*4,1024), nn.ReLU(inplace=True), nn.Linear(1024,128), nn.ReLU(inplace=True), nn.Linear(128,10))\n",
    "    def forward( self , x):\n",
    "        x = self.layer4(self.layer3(self.layer2(self.layer1(x))))\n",
    "        x = x.reshape(x.size(0) , -1)\n",
    "        fc_out = self.fc(x)\n",
    "        return fc_out\n",
    "model = SimpleCNN()\n",
    "\n",
    "# 4) 封装之前要把模型移到对应的gpu\n",
    "model.to(device)\n",
    "# 尽管会降低GPU利用率，但可以提高模型在多卡场景下的表现\n",
    "# model = torch.nn.SyncBatchNorm.convert_sync_batchnorm(model)  # BN层同步\n",
    "# 同步容易报错，内存越界 RuntimeError: CUDA error: an illegal memory access was encountered\n",
    "if torch.cuda.device_count() > 1:\n",
    "    print(\"Let's use\", torch.cuda.device_count(), \"GPUs!\")\n",
    "    # 5) 封装\n",
    "    model = torch.nn.parallel.DistributedDataParallel(model, device_ids=[local_rank], output_device=local_rank)\n",
    "for epoch in range(epoches_num):\n",
    "    sampler.set_epoch(epoch)\n",
    "    for batch_input, batch_label in rand_loader:\n",
    "        output = model(batch_input)\n",
    "        print(\"Outside: input size\", batch_input.size(), \"output_size\", output.size())\n",
    "\n",
    "# 单机多卡2：命令，成功执行\n",
    "# CUDA_VISIBLE_DEVICES=0,1 python -m torch.distributed.launch --nproc_per_node=2 ddp.py"
   ],
   "metadata": {
    "collapsed": false
   },
   "id": "1bfc9d158b73d7e0"
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "outputs": [],
   "source": [
    "# 2.3 单机多卡复杂版本 cnn on minist（成功执行）\n",
    "\n",
    "import torch\n",
    "import torch.nn as nn\n",
    "from torch.autograd import Variable\n",
    "from torch.utils.data import Dataset, DataLoader\n",
    "from torchvision import datasets, transforms\n",
    "import os\n",
    "from torch.utils.data.distributed import DistributedSampler\n",
    "\n",
    "def reduce_loss(tensor, rank, world_size):\n",
    "    with torch.no_grad():\n",
    "        dist.reduce(tensor, dst=0)\n",
    "        if rank == 0:\n",
    "            tensor /= world_size\n",
    "\n",
    "# 1) 初始化\n",
    "# torch.distributed.init_process_group(backend=\"nccl\") # 推荐这个，两个gpu占用率才7%左右\n",
    "# torch.distributed.init_process_group(backend=\"nccl\", init_method='env://') # 报错ERROR:torch.distributed.elastic.multiprocessing.api:failed (exitcode: -6) local_rank: 0 (pid: 3596) of binary: /root/miniconda3/bin/python\n",
    "torch.distributed.init_process_group(backend=\"nccl\", world_size=2) # 推荐这个,gpu:1占用率会偶尔飙升一下\n",
    "# torch.distributed.init_process_group(backend=\"nccl\", rank=0, world_size=2)  # 这个有问题，只用了gpu:0\n",
    "\n",
    "# 定义超参数\n",
    "learning_rate = 1e-2      # 学习率\n",
    "batch_size    = 1000       # 批的大小\n",
    "epoches_num   = 10        # 遍历训练集的次数\n",
    "\n",
    "# 2） 配置每个进程的gpu\n",
    "local_rank = torch.distributed.get_rank()\n",
    "torch.cuda.set_device(local_rank)\n",
    "device = torch.device(\"cuda\", local_rank)\n",
    "\n",
    "# 3）使用DistributedSampler\n",
    "dataset = datasets.MNIST(root='./data', train=True, transform=transforms.ToTensor(), download=True )\n",
    "sampler=DistributedSampler(dataset)\n",
    "rand_loader = DataLoader(dataset=dataset, batch_size=batch_size, sampler=sampler)\n",
    "\n",
    "class SimpleCNN(nn.Module):\n",
    "    def __init__(self):\n",
    "        super(SimpleCNN, self).__init__()\n",
    "        self.layer1 = nn.Sequential(nn.Conv2d(1,16,kernel_size=3), nn.BatchNorm2d(16), nn.ReLU(inplace=True))\n",
    "        self.layer2 = nn.Sequential(nn.Conv2d(16,32,kernel_size=3), nn.BatchNorm2d(32), nn.ReLU(inplace=True), nn.MaxPool2d(kernel_size=2 , stride=2))\n",
    "        self.layer3 = nn.Sequential(nn.Conv2d(32,64,kernel_size=3), nn.BatchNorm2d(64), nn.ReLU(inplace=True))\n",
    "        self.layer4 = nn.Sequential(nn.Conv2d(64,128,kernel_size=3), nn.BatchNorm2d(128), nn.ReLU(inplace=True), nn.MaxPool2d(kernel_size=2 , stride=2))\n",
    "        self.fc = nn.Sequential(nn.Linear(128*4*4,1024), nn.ReLU(inplace=True), nn.Linear(1024,128), nn.ReLU(inplace=True), nn.Linear(128,10))\n",
    "    def forward( self , x):\n",
    "        x = self.layer4(self.layer3(self.layer2(self.layer1(x))))\n",
    "        x = x.reshape(x.size(0) , -1)\n",
    "        fc_out = self.fc(x)\n",
    "        return fc_out\n",
    "model = SimpleCNN()\n",
    "\n",
    "# 4) 封装之前要把模型移到对应的gpu\n",
    "model.to(device)\n",
    "# 尽管会降低GPU利用率，但可以提高模型在多卡场景下的表现\n",
    "# model = torch.nn.SyncBatchNorm.convert_sync_batchnorm(model)  # BN层同步\n",
    "# 同步容易报错，内存越界 RuntimeError: CUDA error: an illegal memory access was encountered\n",
    "if torch.cuda.device_count() > 1:\n",
    "    print(\"Let's use\", torch.cuda.device_count(), \"GPUs!\")\n",
    "    # 5) 封装\n",
    "    model = torch.nn.parallel.DistributedDataParallel(model, device_ids=[local_rank], output_device=local_rank)\n",
    "for epoch in range(epoches_num):\n",
    "    sampler.set_epoch(epoch)\n",
    "    for batch_input, batch_label in rand_loader:\n",
    "        output = model(batch_input)\n",
    "        print(\"Outside: input size\", batch_input.size(), \"output_size\", output.size())\n",
    "\n",
    "# 单机多卡2：命令，成功执行\n",
    "# CUDA_VISIBLE_DEVICES=0,1 python -m torch.distributed.launch --nproc_per_node=2 ddp.py"
   ],
   "metadata": {
    "collapsed": false
   },
   "id": "5a7e289136fe199e"
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "outputs": [],
   "source": [
    "\n",
    "# 2.3.单机多卡复杂版本 cnn on minist（成功执行）\n",
    "\n",
    "import torch, time, os\n",
    "from torch import nn, optim\n",
    "from torchvision import datasets, transforms\n",
    "from torch.autograd import Variable\n",
    "from torch.utils.data import DataLoader, DistributedSampler\n",
    "from tqdm import tqdm\n",
    "from torch.nn.parallel import DistributedDataParallel\n",
    "from torch.distributed import init_process_group, destroy_process_group, get_rank\n",
    "\n",
    "device_ids = [0,1]\n",
    "main_device = 0\n",
    "world_size = len(device_ids)\n",
    "\n",
    "def reduce_loss(tensor, rank, world_size):\n",
    "    with torch.no_grad():\n",
    "        dist.reduce(tensor, dst=0)\n",
    "        if rank == 0:\n",
    "            tensor /= world_size\n",
    "\n",
    "os.environ['MASTER_ADDR'] = 'localhost'  # 0号机器的IP\n",
    "os.environ['MASTER_PORT'] = '19198'  # 0号机器的可用端口\n",
    "os.environ['WORLD_SIZE'] = str(world_size)\n",
    "os.environ['RANK'] = str(main_device)\n",
    "os.environ['LOCAL_RANK'] = str(main_device)\n",
    "init_process_group(backend='gloo', init_method='env://')\n",
    "torch.cuda.set_device(main_device)\n",
    "global_rank = get_rank()\n",
    "print(global_rank)\n",
    "\n",
    "class SimpleCNN(nn.Module):\n",
    "    def __init__(self):\n",
    "        super(SimpleCNN, self).__init__()\n",
    "        self.layer1 = nn.Sequential(nn.Conv2d(1,16,kernel_size=3), nn.BatchNorm2d(16), nn.ReLU(inplace=True))\n",
    "        self.layer2 = nn.Sequential(nn.Conv2d(16,32,kernel_size=3), nn.BatchNorm2d(32), nn.ReLU(inplace=True), nn.MaxPool2d(kernel_size=2 , stride=2))\n",
    "        self.layer3 = nn.Sequential(nn.Conv2d(32,64,kernel_size=3), nn.BatchNorm2d(64), nn.ReLU(inplace=True))\n",
    "        self.layer4 = nn.Sequential(nn.Conv2d(64,128,kernel_size=3), nn.BatchNorm2d(128), nn.ReLU(inplace=True), nn.MaxPool2d(kernel_size=2 , stride=2))\n",
    "        self.fc = nn.Sequential(nn.Linear(128*4*4,1024), nn.ReLU(inplace=True), nn.Linear(1024,128), nn.ReLU(inplace=True), nn.Linear(128,10))\n",
    "    def forward( self , x):\n",
    "        x = self.layer4(self.layer3(self.layer2(self.layer1(x))))\n",
    "        x = x.reshape(x.size(0) , -1)\n",
    "        fc_out = self.fc(x)\n",
    "        return fc_out\n",
    "model = DistributedDataParallel(SimpleCNN().cuda(), device_ids=device_ids, output_device=main_device)\n",
    "\n",
    "# 定义超参数\n",
    "learning_rate = 1e-2      # 学习率\n",
    "batch_size    = 2000       # 批的大小\n",
    "epoches_num   = 10        # 遍历训练集的次数\n",
    "# 下载训练集 MNIST 手写数字训练集\n",
    "train_dataset = datasets.MNIST(root='./data', train=True, transform=transforms.ToTensor(), download=True)\n",
    "sampler = DistributedSampler(train_dataset, shuffle=True)\n",
    "train_loader  = DataLoader(train_dataset, batch_size=batch_size, shuffle=True, pin_memory=True, sampler=sampler) # 注意这里\n",
    "criterion = nn.CrossEntropyLoss()\n",
    "\n",
    "model.train()\n",
    "for epoch in range(epoches):\n",
    "    # 让每张卡在每轮迭代中数据顺序随机\n",
    "    sampler.set_epoch(epoch)\n",
    "    for batch_input, batch_label in tqdm(dataloader):\n",
    "        batch_input, batch_label = batch_input.cuda(), batch_label.cuda()\n",
    "        batch_output = model(batch_input)\n",
    "        batch_loss = criterion(batch_output, batch_label)\n",
    "        \n",
    "        optimizer.zero_grad()\n",
    "        batch_loss.backward()\n",
    "        optimizer.step()\n",
    "        reduce_loss(batch_loss, global_rank, world_size)\n",
    "    # 损失/准确率计算\n",
    "    train_loss += loss.item() * label.size(0)\n",
    "    _ , pred    = out.max(1)\n",
    "    num_correct = pred.eq(label).sum()\n",
    "    accuracy    = pred.eq(label).float().mean()\n",
    "    train_acc  += num_correct.item()\n",
    "    if global_rank == 0:\n",
    "        print('Finish  {}  Loss: {:.6f}, Acc: {:.6f}'.format( epoch+1 , train_loss / len(train_dataset), train_acc / len(train_dataset )))\n",
    "# destroy_process_group()\n",
    "\n",
    "# 单机多卡：命令\n",
    "# python -m torch.distributed.launch --nproc_per_node=n_gpus DDP.py\n",
    "\n",
    "# 多机多卡：命令（晚上跑一跑）\n",
    "# python -m torch.distributed.launch --nproc_per_node=n_gpus --nodes=2 --node_rank=0 --master_addr=\"主节点ip\" --master_port=\"主节点端口\" DDP.py\n",
    "# python -m torch.distributed.launch --nproc_per_node=n_gpus --nodes=2 --node_rank=1 --master_addr=\"主节点ip\" --master_port=\"主节点端口\" DDP.py\n",
    "# python -m torch.distributed.launch --nproc_per_node=n_gpus --nodes=2 --node_rank=2 --master_addr=\"主节点ip\" --master_port=\"主节点端口\" DDP.py"
   ],
   "metadata": {
    "collapsed": false
   },
   "id": "81991e7ddc1ee323"
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "outputs": [],
   "source": [
    "\n",
    "# 2.1.单机多卡实现\n",
    "\n",
    "import torch, time, os\n",
    "from torch import nn, optim\n",
    "from torchvision import datasets, transforms\n",
    "from torch.autograd import Variable\n",
    "from torch.utils.data import DataLoader, DistributedSampler\n",
    "from tqdm import tqdm\n",
    "from torch.nn.parallel import DistributedDataParallel\n",
    "from torch.distributed import init_process_group, destroy_process_group, get_rank, reduce\n",
    "from multiprocessing import Process\n",
    "\n",
    "device_ids = [0,1]\n",
    "main_device = 0\n",
    "world_size = len(device_ids)\n",
    "\n",
    "def reduce_loss(tensor, rank, world_size):\n",
    "    with torch.no_grad():\n",
    "        reduce(tensor, dst=0)\n",
    "        if rank == 0:\n",
    "            tensor /= world_size\n",
    "def setup(rank, world_size):\n",
    "    \"Sets up the process group and configuration for PyTorch Distributed Data Parallelism\"\n",
    "    os.environ[\"MASTER_ADDR\"] = 'localhost'\n",
    "    os.environ[\"MASTER_PORT\"] = \"12355\"\n",
    "    # Initialize the process group\n",
    "    init_process_group(\"gloo\", rank=rank, world_size=world_size)\n",
    "def cleanup():\n",
    "    \"Cleans up the distributed environment\"\n",
    "    destroy_process_group()\n",
    "\n",
    "class SimpleCNN(nn.Module):\n",
    "    def __init__(self):\n",
    "        super(SimpleCNN, self).__init__()\n",
    "        self.layer1 = nn.Sequential(nn.Conv2d(1,16,kernel_size=3), nn.BatchNorm2d(16), nn.ReLU(inplace=True))\n",
    "        self.layer2 = nn.Sequential(nn.Conv2d(16,32,kernel_size=3), nn.BatchNorm2d(32), nn.ReLU(inplace=True), nn.MaxPool2d(kernel_size=2 , stride=2))\n",
    "        self.layer3 = nn.Sequential(nn.Conv2d(32,64,kernel_size=3), nn.BatchNorm2d(64), nn.ReLU(inplace=True))\n",
    "        self.layer4 = nn.Sequential(nn.Conv2d(64,128,kernel_size=3), nn.BatchNorm2d(128), nn.ReLU(inplace=True), nn.MaxPool2d(kernel_size=2 , stride=2))\n",
    "        self.fc = nn.Sequential(nn.Linear(128*4*4,1024), nn.ReLU(inplace=True), nn.Linear(1024,128), nn.ReLU(inplace=True), nn.Linear(128,10))\n",
    "    def forward( self , x):\n",
    "        x = self.layer4(self.layer3(self.layer2(self.layer1(x))))\n",
    "        x = x.reshape(x.size(0) , -1)\n",
    "        fc_out = self.fc(x)\n",
    "        return fc_out\n",
    "model = SimpleCNN()\n",
    "\n",
    "# 定义超参数\n",
    "learning_rate = 1e-2      # 学习率\n",
    "batch_size    = 2000       # 批的大小\n",
    "epoches_num   = 10        # 遍历训练集的次数\n",
    "# 下载训练集 MNIST 手写数字训练集\n",
    "train_dataset = datasets.MNIST(root='./data', train=True, transform=transforms.ToTensor(), download=True)\n",
    "criterion = nn.CrossEntropyLoss()\n",
    "\n",
    "def train(rank, world_size):\n",
    "    setup(rank, world_size)\n",
    "    sampler = DistributedSampler(train_dataset, shuffle=True)\n",
    "    train_loader  = DataLoader(train_dataset, batch_size=batch_size, shuffle=False, pin_memory=True, sampler=sampler) # 注意这里\n",
    "    ddp_model = DistributedDataParallel(model.to(rank), device_ids=[rank])\n",
    "    optimizer = optim.AdamW(ddp_model.parameters(), lr=1e-3)\n",
    "    # Train for one epoch\n",
    "    model.train()\n",
    "    for batch_idx, (data, target) in enumerate(train_loader):\n",
    "        data, target = data.to(device), target.to(device)\n",
    "        output = model(data)\n",
    "        loss = criterion(output, target)\n",
    "        loss.backward()\n",
    "        optimizer.step()\n",
    "        optimizer.zero_grad()\n",
    "        if get_rank() == 0:\n",
    "            print(loss.item())\n",
    "    cleanup()\n",
    "\n",
    "def main():\n",
    "    size = 2\n",
    "    processes=[]\n",
    "    for i in range(size):\n",
    "        p = Process(target=train, args=(i, size))\n",
    "        p.start()\n",
    "        processes.append(p)\n",
    "        # init_processes(i, size)\n",
    "    for p in processes:\n",
    "        p.join() \n",
    "if __name__=='__main__':\n",
    "    main()\n",
    "\n",
    "# 单机多卡：命令\n",
    "# python -m torch.distributed.launch --nproc_per_node=n_gpus DDP.py\n",
    "\n",
    "# 多机多卡：命令（晚上跑一跑）\n",
    "# python -m torch.distributed.launch --nproc_per_node=n_gpus --nodes=2 --node_rank=0 --master_addr=\"主节点ip\" --master_port=\"主节点端口\" DDP.py\n",
    "# python -m torch.distributed.launch --nproc_per_node=n_gpus --nodes=2 --node_rank=1 --master_addr=\"主节点ip\" --master_port=\"主节点端口\" DDP.py\n",
    "# python -m torch.distributed.launch --nproc_per_node=n_gpus --nodes=2 --node_rank=2 --master_addr=\"主节点ip\" --master_port=\"主节点端口\" DDP.py"
   ],
   "metadata": {
    "collapsed": false
   },
   "id": "6d2f4fe386c0865"
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "outputs": [],
   "source": [
    "\n",
    "# 2.1.单机多卡实现\n",
    "\n",
    "import torch, time, os\n",
    "from torch import nn, optim\n",
    "from torchvision import datasets, transforms\n",
    "from torch.autograd import Variable\n",
    "from torch.utils.data import DataLoader, DistributedSampler\n",
    "from tqdm import tqdm\n",
    "from torch.nn.parallel import DistributedDataParallel\n",
    "from torch.distributed import init_process_group, destroy_process_group, get_rank\n",
    "\n",
    "device_ids = [0,1]\n",
    "main_device = 0\n",
    "world_size = len(device_ids)\n",
    "\n",
    "def reduce_loss(tensor, rank, world_size):\n",
    "    with torch.no_grad():\n",
    "        dist.reduce(tensor, dst=0)\n",
    "        if rank == 0:\n",
    "            tensor /= world_size\n",
    "\n",
    "os.environ['MASTER_ADDR'] = 'localhost'  # 0号机器的IP\n",
    "os.environ['MASTER_PORT'] = '19198'  # 0号机器的可用端口\n",
    "os.environ['WORLD_SIZE'] = str(world_size)\n",
    "os.environ['RANK'] = str(main_device)\n",
    "os.environ['LOCAL_RANK'] = str(main_device)\n",
    "init_process_group(backend='nccl', init_method='env://')\n",
    "torch.cuda.set_device(main_device)\n",
    "global_rank = get_rank()\n",
    "\n",
    "class SimpleCNN(nn.Module):\n",
    "    def __init__(self):\n",
    "        super(SimpleCNN, self).__init__()\n",
    "        self.layer1 = nn.Sequential(nn.Conv2d(1,16,kernel_size=3), nn.BatchNorm2d(16), nn.ReLU(inplace=True))\n",
    "        self.layer2 = nn.Sequential(nn.Conv2d(16,32,kernel_size=3), nn.BatchNorm2d(32), nn.ReLU(inplace=True), nn.MaxPool2d(kernel_size=2 , stride=2))\n",
    "        self.layer3 = nn.Sequential(nn.Conv2d(32,64,kernel_size=3), nn.BatchNorm2d(64), nn.ReLU(inplace=True))\n",
    "        self.layer4 = nn.Sequential(nn.Conv2d(64,128,kernel_size=3), nn.BatchNorm2d(128), nn.ReLU(inplace=True), nn.MaxPool2d(kernel_size=2 , stride=2))\n",
    "        self.fc = nn.Sequential(nn.Linear(128*4*4,1024), nn.ReLU(inplace=True), nn.Linear(1024,128), nn.ReLU(inplace=True), nn.Linear(128,10))\n",
    "    def forward( self , x):\n",
    "        x = self.layer4(self.layer3(self.layer2(self.layer1(x))))\n",
    "        x = x.reshape(x.size(0) , -1)\n",
    "        fc_out = self.fc(x)\n",
    "        return fc_out\n",
    "model = DistributedDataParallel(SimpleCNN().cuda(), device_ids=device_ids, output_device=main_device)\n",
    "\n",
    "# 定义超参数\n",
    "learning_rate = 1e-2      # 学习率\n",
    "batch_size    = 4000       # 批的大小\n",
    "epoches_num   = 10        # 遍历训练集的次数\n",
    "# 下载训练集 MNIST 手写数字训练集\n",
    "train_dataset = datasets.MNIST(root='./data', train=True, transform=transforms.ToTensor(), download=True )\n",
    "sampler = DistributedSampler(train_dataset, shuffle=True)\n",
    "train_loader  = DataLoader(train_dataset, batch_size=batch_size, shuffle=True, pin_memory=True, sampler=sampler) # 注意这里\n",
    "criterion = nn.CrossEntropyLoss()\n",
    "\n",
    "model.train()\n",
    "for epoch in range(epoches):\n",
    "    # 让每张卡在每轮迭代中数据顺序随机\n",
    "    sampler.set_epoch(epoch)\n",
    "    for batch_input, batch_label in tqdm(dataloader):\n",
    "        batch_input, batch_label = batch_input.cuda(), batch_label.cuda()\n",
    "        batch_output = model(batch_input)\n",
    "        batch_loss = criterion(batch_output, batch_label)\n",
    "        \n",
    "        optimizer.zero_grad()\n",
    "        batch_loss.backward()\n",
    "        optimizer.step()\n",
    "        reduce_loss(batch_loss, global_rank, world_size)\n",
    "    # 损失/准确率计算\n",
    "    train_loss += loss.item() * label.size(0)\n",
    "    _ , pred    = out.max(1)\n",
    "    num_correct = pred.eq(label).sum()\n",
    "    accuracy    = pred.eq(label).float().mean()\n",
    "    train_acc  += num_correct.item()\n",
    "    if global_rank == 0:\n",
    "        print('Finish  {}  Loss: {:.6f}, Acc: {:.6f}'.format( epoch+1 , train_loss / len(train_dataset), train_acc / len(train_dataset )))\n",
    "# destroy_process_group()\n",
    "\n",
    "# 单机多卡：命令\n",
    "# python -m torch.distributed.launch --nproc_per_node=n_gpus DDP.py\n",
    "\n",
    "# 多机多卡：命令（晚上跑一跑）\n",
    "# python -m torch.distributed.launch --nproc_per_node=n_gpus --nodes=2 --node_rank=0 --master_addr=\"主节点ip\" --master_port=\"主节点端口\" DDP.py\n",
    "# python -m torch.distributed.launch --nproc_per_node=n_gpus --nodes=2 --node_rank=1 --master_addr=\"主节点ip\" --master_port=\"主节点端口\" DDP.py\n",
    "# python -m torch.distributed.launch --nproc_per_node=n_gpus --nodes=2 --node_rank=2 --master_addr=\"主节点ip\" --master_port=\"主节点端口\" DDP.py"
   ],
   "metadata": {
    "collapsed": false
   },
   "id": "d65f93df089435ec"
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "outputs": [
    {
     "ename": "ValueError",
     "evalue": "Error initializing torch.distributed using env:// rendezvous: environment variable RANK expected, but not set",
     "output_type": "error",
     "traceback": [
      "\u001B[0;31m---------------------------------------------------------------------------\u001B[0m",
      "\u001B[0;31mValueError\u001B[0m                                Traceback (most recent call last)",
      "Input \u001B[0;32mIn [4]\u001B[0m, in \u001B[0;36m<cell line: 8>\u001B[0;34m()\u001B[0m\n\u001B[1;32m      6\u001B[0m \u001B[38;5;28;01mfrom\u001B[39;00m \u001B[38;5;21;01mtorch\u001B[39;00m\u001B[38;5;21;01m.\u001B[39;00m\u001B[38;5;21;01mutils\u001B[39;00m\u001B[38;5;21;01m.\u001B[39;00m\u001B[38;5;21;01mdata\u001B[39;00m\u001B[38;5;21;01m.\u001B[39;00m\u001B[38;5;21;01mdistributed\u001B[39;00m \u001B[38;5;28;01mimport\u001B[39;00m DistributedSampler\n\u001B[1;32m      7\u001B[0m \u001B[38;5;66;03m# 1) 初始化\u001B[39;00m\n\u001B[0;32m----> 8\u001B[0m \u001B[43mtorch\u001B[49m\u001B[38;5;241;43m.\u001B[39;49m\u001B[43mdistributed\u001B[49m\u001B[38;5;241;43m.\u001B[39;49m\u001B[43minit_process_group\u001B[49m\u001B[43m(\u001B[49m\u001B[43mbackend\u001B[49m\u001B[38;5;241;43m=\u001B[39;49m\u001B[38;5;124;43m\"\u001B[39;49m\u001B[38;5;124;43mnccl\u001B[39;49m\u001B[38;5;124;43m\"\u001B[39;49m\u001B[43m)\u001B[49m\n\u001B[1;32m     10\u001B[0m input_size \u001B[38;5;241m=\u001B[39m \u001B[38;5;241m5\u001B[39m\n\u001B[1;32m     11\u001B[0m output_size \u001B[38;5;241m=\u001B[39m \u001B[38;5;241m2\u001B[39m\n",
      "File \u001B[0;32m~/opt/anaconda3/lib/python3.9/site-packages/torch/distributed/distributed_c10d.py:900\u001B[0m, in \u001B[0;36minit_process_group\u001B[0;34m(backend, init_method, timeout, world_size, rank, store, group_name, pg_options)\u001B[0m\n\u001B[1;32m    896\u001B[0m \u001B[38;5;28;01mif\u001B[39;00m store \u001B[38;5;129;01mis\u001B[39;00m \u001B[38;5;28;01mNone\u001B[39;00m:\n\u001B[1;32m    897\u001B[0m     rendezvous_iterator \u001B[38;5;241m=\u001B[39m rendezvous(\n\u001B[1;32m    898\u001B[0m         init_method, rank, world_size, timeout\u001B[38;5;241m=\u001B[39mtimeout\n\u001B[1;32m    899\u001B[0m     )\n\u001B[0;32m--> 900\u001B[0m     store, rank, world_size \u001B[38;5;241m=\u001B[39m \u001B[38;5;28;43mnext\u001B[39;49m\u001B[43m(\u001B[49m\u001B[43mrendezvous_iterator\u001B[49m\u001B[43m)\u001B[49m\n\u001B[1;32m    901\u001B[0m     store\u001B[38;5;241m.\u001B[39mset_timeout(timeout)\n\u001B[1;32m    903\u001B[0m     \u001B[38;5;66;03m# Use a PrefixStore to avoid accidental overrides of keys used by\u001B[39;00m\n\u001B[1;32m    904\u001B[0m     \u001B[38;5;66;03m# different systems (e.g. RPC) in case the store is multi-tenant.\u001B[39;00m\n",
      "File \u001B[0;32m~/opt/anaconda3/lib/python3.9/site-packages/torch/distributed/rendezvous.py:235\u001B[0m, in \u001B[0;36m_env_rendezvous_handler\u001B[0;34m(url, timeout, **kwargs)\u001B[0m\n\u001B[1;32m    233\u001B[0m     rank \u001B[38;5;241m=\u001B[39m \u001B[38;5;28mint\u001B[39m(query_dict[\u001B[38;5;124m\"\u001B[39m\u001B[38;5;124mrank\u001B[39m\u001B[38;5;124m\"\u001B[39m])\n\u001B[1;32m    234\u001B[0m \u001B[38;5;28;01melse\u001B[39;00m:\n\u001B[0;32m--> 235\u001B[0m     rank \u001B[38;5;241m=\u001B[39m \u001B[38;5;28mint\u001B[39m(\u001B[43m_get_env_or_raise\u001B[49m\u001B[43m(\u001B[49m\u001B[38;5;124;43m\"\u001B[39;49m\u001B[38;5;124;43mRANK\u001B[39;49m\u001B[38;5;124;43m\"\u001B[39;49m\u001B[43m)\u001B[49m)\n\u001B[1;32m    237\u001B[0m \u001B[38;5;28;01mif\u001B[39;00m \u001B[38;5;124m\"\u001B[39m\u001B[38;5;124mworld_size\u001B[39m\u001B[38;5;124m\"\u001B[39m \u001B[38;5;129;01min\u001B[39;00m query_dict:\n\u001B[1;32m    238\u001B[0m     world_size \u001B[38;5;241m=\u001B[39m \u001B[38;5;28mint\u001B[39m(query_dict[\u001B[38;5;124m\"\u001B[39m\u001B[38;5;124mworld_size\u001B[39m\u001B[38;5;124m\"\u001B[39m])\n",
      "File \u001B[0;32m~/opt/anaconda3/lib/python3.9/site-packages/torch/distributed/rendezvous.py:220\u001B[0m, in \u001B[0;36m_env_rendezvous_handler.<locals>._get_env_or_raise\u001B[0;34m(env_var)\u001B[0m\n\u001B[1;32m    218\u001B[0m env_val \u001B[38;5;241m=\u001B[39m os\u001B[38;5;241m.\u001B[39menviron\u001B[38;5;241m.\u001B[39mget(env_var, \u001B[38;5;28;01mNone\u001B[39;00m)\n\u001B[1;32m    219\u001B[0m \u001B[38;5;28;01mif\u001B[39;00m \u001B[38;5;129;01mnot\u001B[39;00m env_val:\n\u001B[0;32m--> 220\u001B[0m     \u001B[38;5;28;01mraise\u001B[39;00m _env_error(env_var)\n\u001B[1;32m    221\u001B[0m \u001B[38;5;28;01melse\u001B[39;00m:\n\u001B[1;32m    222\u001B[0m     \u001B[38;5;28;01mreturn\u001B[39;00m env_val\n",
      "\u001B[0;31mValueError\u001B[0m: Error initializing torch.distributed using env:// rendezvous: environment variable RANK expected, but not set"
     ]
    }
   ],
   "source": [
    "import torch\n",
    "import torch.nn as nn\n",
    "from torch.autograd import Variable\n",
    "from torch.utils.data import Dataset, DataLoader\n",
    "import os\n",
    "from torch.utils.data.distributed import DistributedSampler\n",
    "# 1) 初始化\n",
    "torch.distributed.init_process_group(backend=\"nccl\")\n",
    "\n",
    "input_size = 5\n",
    "output_size = 2\n",
    "batch_size = 30\n",
    "data_size = 90\n",
    "\n",
    "# 2） 配置每个进程的gpu\n",
    "local_rank = torch.distributed.get_rank()\n",
    "torch.cuda.set_device(local_rank)\n",
    "device = torch.device(\"cuda\", local_rank)\n",
    "\n",
    "class RandomDataset(Dataset):\n",
    "    def __init__(self, size, length):\n",
    "        self.len = length\n",
    "        self.data = torch.randn(length, size).to('cuda')\n",
    "\n",
    "    def __getitem__(self, index):\n",
    "        return self.data[index]\n",
    "\n",
    "    def __len__(self):\n",
    "        return self.len\n",
    "\n",
    "dataset = RandomDataset(input_size, data_size)\n",
    "# 3）使用DistributedSampler\n",
    "rand_loader = DataLoader(dataset=dataset,\n",
    "                         batch_size=batch_size,\n",
    "                         sampler=DistributedSampler(dataset))\n",
    "\n",
    "class Model(nn.Module):\n",
    "    def __init__(self, input_size, output_size):\n",
    "        super(Model, self).__init__()\n",
    "        self.fc = nn.Linear(input_size, output_size)\n",
    "\n",
    "    def forward(self, input):\n",
    "        output = self.fc(input)\n",
    "        print(\"  In Model: input size\", input.size(),\n",
    "              \"output size\", output.size())\n",
    "        return output\n",
    "\n",
    "model = Model(input_size, output_size)\n",
    "\n",
    "# 4) 封装之前要把模型移到对应的gpu\n",
    "model.to(device)\n",
    "\n",
    "if torch.cuda.device_count() > 1:\n",
    "    print(\"Let's use\", torch.cuda.device_count(), \"GPUs!\")\n",
    "    # 5) 封装\n",
    "    model = torch.nn.parallel.DistributedDataParallel(model,\n",
    "                                                      device_ids=[local_rank],\n",
    "                                                      output_device=local_rank)\n",
    "\n",
    "for data in rand_loader:\n",
    "    if torch.cuda.is_available():\n",
    "        input_var = data\n",
    "    else:\n",
    "        input_var = data\n",
    "\n",
    "    output = model(input_var)\n",
    "    print(\"Outside: input size\", input_var.size(), \"output_size\", output.size())"
   ],
   "metadata": {
    "collapsed": false,
    "ExecuteTime": {
     "end_time": "2023-08-23T22:32:56.544656Z",
     "start_time": "2023-08-23T22:32:56.141091Z"
    }
   },
   "id": "ccf1bdb4a9ff2b52"
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "outputs": [
    {
     "ename": "RuntimeError",
     "evalue": "trying to initialize the default process group twice!",
     "output_type": "error",
     "traceback": [
      "\u001B[0;31m---------------------------------------------------------------------------\u001B[0m",
      "\u001B[0;31mRuntimeError\u001B[0m                              Traceback (most recent call last)",
      "Input \u001B[0;32mIn [11]\u001B[0m, in \u001B[0;36m<cell line: 10>\u001B[0;34m()\u001B[0m\n\u001B[1;32m      8\u001B[0m os\u001B[38;5;241m.\u001B[39menviron[\u001B[38;5;124m'\u001B[39m\u001B[38;5;124mMASTER_ADDR\u001B[39m\u001B[38;5;124m'\u001B[39m] \u001B[38;5;241m=\u001B[39m \u001B[38;5;124m'\u001B[39m\u001B[38;5;124mlocalhost\u001B[39m\u001B[38;5;124m'\u001B[39m\n\u001B[1;32m      9\u001B[0m os\u001B[38;5;241m.\u001B[39menviron[\u001B[38;5;124m'\u001B[39m\u001B[38;5;124mMASTER_PORT\u001B[39m\u001B[38;5;124m'\u001B[39m] \u001B[38;5;241m=\u001B[39m \u001B[38;5;124m'\u001B[39m\u001B[38;5;124m12345\u001B[39m\u001B[38;5;124m'\u001B[39m\n\u001B[0;32m---> 10\u001B[0m \u001B[43mtorch\u001B[49m\u001B[38;5;241;43m.\u001B[39;49m\u001B[43mdistributed\u001B[49m\u001B[38;5;241;43m.\u001B[39;49m\u001B[43minit_process_group\u001B[49m\u001B[43m(\u001B[49m\u001B[43mbackend\u001B[49m\u001B[38;5;241;43m=\u001B[39;49m\u001B[38;5;124;43m\"\u001B[39;49m\u001B[38;5;124;43mnccl\u001B[39;49m\u001B[38;5;124;43m\"\u001B[39;49m\u001B[43m,\u001B[49m\u001B[43m \u001B[49m\u001B[43minit_method\u001B[49m\u001B[38;5;241;43m=\u001B[39;49m\u001B[38;5;124;43m'\u001B[39;49m\u001B[38;5;124;43menv://\u001B[39;49m\u001B[38;5;124;43m'\u001B[39;49m\u001B[43m,\u001B[49m\u001B[43m \u001B[49m\u001B[43mrank\u001B[49m\u001B[43m \u001B[49m\u001B[38;5;241;43m=\u001B[39;49m\u001B[43m \u001B[49m\u001B[38;5;241;43m0\u001B[39;49m\u001B[43m,\u001B[49m\u001B[43m \u001B[49m\u001B[43mworld_size\u001B[49m\u001B[43m \u001B[49m\u001B[38;5;241;43m=\u001B[39;49m\u001B[43m \u001B[49m\u001B[38;5;241;43m2\u001B[39;49m\u001B[43m)\u001B[49m\n\u001B[1;32m     12\u001B[0m input_size \u001B[38;5;241m=\u001B[39m \u001B[38;5;241m5\u001B[39m\n\u001B[1;32m     13\u001B[0m output_size \u001B[38;5;241m=\u001B[39m \u001B[38;5;241m2\u001B[39m\n",
      "File \u001B[0;32m~/opt/anaconda3/lib/python3.9/site-packages/torch/distributed/distributed_c10d.py:865\u001B[0m, in \u001B[0;36minit_process_group\u001B[0;34m(backend, init_method, timeout, world_size, rank, store, group_name, pg_options)\u001B[0m\n\u001B[1;32m    860\u001B[0m     \u001B[38;5;28;01mraise\u001B[39;00m \u001B[38;5;167;01mRuntimeError\u001B[39;00m(\n\u001B[1;32m    861\u001B[0m         \u001B[38;5;124m\"\u001B[39m\u001B[38;5;124mExpected timeout argument to be of type\u001B[39m\u001B[38;5;124m\"\u001B[39m \u001B[38;5;124m\"\u001B[39m\u001B[38;5;124mdatetime.timedelta\u001B[39m\u001B[38;5;124m\"\u001B[39m\n\u001B[1;32m    862\u001B[0m     )\n\u001B[1;32m    864\u001B[0m \u001B[38;5;28;01mif\u001B[39;00m GroupMember\u001B[38;5;241m.\u001B[39mWORLD \u001B[38;5;129;01mis\u001B[39;00m \u001B[38;5;129;01mnot\u001B[39;00m \u001B[38;5;28;01mNone\u001B[39;00m:\n\u001B[0;32m--> 865\u001B[0m     \u001B[38;5;28;01mraise\u001B[39;00m \u001B[38;5;167;01mRuntimeError\u001B[39;00m(\u001B[38;5;124m\"\u001B[39m\u001B[38;5;124mtrying to initialize the default process group \u001B[39m\u001B[38;5;124m\"\u001B[39m \u001B[38;5;124m\"\u001B[39m\u001B[38;5;124mtwice!\u001B[39m\u001B[38;5;124m\"\u001B[39m)\n\u001B[1;32m    867\u001B[0m \u001B[38;5;28;01massert\u001B[39;00m (store \u001B[38;5;129;01mis\u001B[39;00m \u001B[38;5;28;01mNone\u001B[39;00m) \u001B[38;5;129;01mor\u001B[39;00m (\n\u001B[1;32m    868\u001B[0m     init_method \u001B[38;5;129;01mis\u001B[39;00m \u001B[38;5;28;01mNone\u001B[39;00m\n\u001B[1;32m    869\u001B[0m ), \u001B[38;5;124m\"\u001B[39m\u001B[38;5;124mCannot specify both init_method and store.\u001B[39m\u001B[38;5;124m\"\u001B[39m\n\u001B[1;32m    871\u001B[0m \u001B[38;5;28;01mif\u001B[39;00m store \u001B[38;5;129;01mis\u001B[39;00m \u001B[38;5;129;01mnot\u001B[39;00m \u001B[38;5;28;01mNone\u001B[39;00m:\n",
      "\u001B[0;31mRuntimeError\u001B[0m: trying to initialize the default process group twice!"
     ]
    }
   ],
   "source": [
    "import torch\n",
    "import torch.nn as nn\n",
    "from torch.autograd import Variable\n",
    "from torch.utils.data import Dataset, DataLoader\n",
    "import os\n",
    "from torch.utils.data.distributed import DistributedSampler\n",
    "# 1) 初始化\n",
    "os.environ['MASTER_ADDR'] = 'localhost'\n",
    "os.environ['MASTER_PORT'] = '12345'\n",
    "torch.distributed.init_process_group(backend=\"nccl\", init_method='env://', rank = 0, world_size = 2)\n",
    "\n",
    "input_size = 5\n",
    "output_size = 2\n",
    "batch_size = 30\n",
    "data_size = 90\n",
    "\n",
    "# 2） 配置每个进程的gpu\n",
    "local_rank = torch.distributed.get_rank()\n",
    "torch.cuda.set_device(local_rank)\n",
    "device = torch.device(\"cuda\", local_rank)\n",
    "\n",
    "class RandomDataset(Dataset):\n",
    "    def __init__(self, size, length):\n",
    "        self.len = length\n",
    "        self.data = torch.randn(length, size).to('cuda')\n",
    "\n",
    "    def __getitem__(self, index):\n",
    "        return self.data[index]\n",
    "\n",
    "    def __len__(self):\n",
    "        return self.len\n",
    "\n",
    "dataset = RandomDataset(input_size, data_size)\n",
    "# 3）使用DistributedSampler\n",
    "rand_loader = DataLoader(dataset=dataset,\n",
    "                         batch_size=batch_size,\n",
    "                         sampler=DistributedSampler(dataset))\n",
    "\n",
    "class Model(nn.Module):\n",
    "    def __init__(self, input_size, output_size):\n",
    "        super(Model, self).__init__()\n",
    "        self.fc = nn.Linear(input_size, output_size)\n",
    "\n",
    "    def forward(self, input):\n",
    "        output = self.fc(input)\n",
    "        print(\"  In Model: input size\", input.size(),\n",
    "              \"output size\", output.size())\n",
    "        return output\n",
    "\n",
    "model = Model(input_size, output_size)\n",
    "\n",
    "# 4) 封装之前要把模型移到对应的gpu\n",
    "model.to(device)\n",
    "\n",
    "if torch.cuda.device_count() > 1:\n",
    "    print(\"Let's use\", torch.cuda.device_count(), \"GPUs!\")\n",
    "    # 5) 封装\n",
    "    model = torch.nn.parallel.DistributedDataParallel(model,\n",
    "                                                      device_ids=[local_rank],\n",
    "                                                      output_device=local_rank)\n",
    "\n",
    "for data in rand_loader:\n",
    "    if torch.cuda.is_available():\n",
    "        input_var = data\n",
    "    else:\n",
    "        input_var = data\n",
    "\n",
    "    output = model(input_var)\n",
    "    print(\"Outside: input size\", input_var.size(), \"output_size\", output.size())"
   ],
   "metadata": {
    "collapsed": false,
    "ExecuteTime": {
     "end_time": "2023-08-23T22:38:25.837098Z",
     "start_time": "2023-08-23T22:38:25.810443Z"
    }
   },
   "id": "452bd76942320ba"
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "outputs": [],
   "source": [
    "#!/usr/bin/env python\n",
    "import os\n",
    "import torch\n",
    "import torch.distributed as dist\n",
    "from torch.multiprocessing import Process\n",
    "import torch.backends.cudnn as cudnn\n",
    "from torch.autograd import Variable\n",
    "import time\n",
    "\n",
    "def allreduce(send, recv):\n",
    "    \"\"\" Implementation of a ring-reduce. \"\"\"\n",
    "    rank = dist.get_rank()\n",
    "    size = dist.get_world_size()\n",
    "    send_buff = torch.zeros(send.size())\n",
    "    recv_buff = torch.zeros(send.size())\n",
    "    accum = torch.zeros(send.size())\n",
    "    accum[:] = send[:]\n",
    "    # th.cuda.synchronize()\n",
    "\n",
    "    left = ((rank - 1) + size) % size\n",
    "    right = (rank + 1) % size\n",
    "\n",
    "    for i in range(size - 1):\n",
    "        if i % 2 == 0:\n",
    "            # Send send_buff\n",
    "            send_req = dist.isend(send_buff, right)\n",
    "            dist.recv(recv_buff, left)\n",
    "            accum[:] += recv[:]\n",
    "        else:\n",
    "            # Send recv_buff\n",
    "            send_req = dist.isend(recv_buff, right)\n",
    "            dist.recv(send_buff, left)\n",
    "            accum[:] += send[:]\n",
    "        send_req.wait()\n",
    "    # th.cuda.synchronize()\n",
    "    recv[:] = accum[:]\n",
    "\n",
    "\n",
    "def run(rank, size):\n",
    "    \"\"\" Distributed function to be implemented later. \"\"\"\n",
    "    model = Model()\n",
    "    model = torch.nn.parallel.DistributedDataParallel(model.cuda())\n",
    "    criterion = torch.nn.MSELoss(size_average=False).cuda()\n",
    "    optimizer = torch.optim.SGD(model.parameters(), lr=0.01)\n",
    "\n",
    "    cudnn.benchmark = True\n",
    "    x_data = Variable(torch.Tensor([[1.0], [2.0], [3.0]]))\n",
    "    y_data = Variable(torch.Tensor([[2.0], [4.0], [6.0]]))\n",
    "\n",
    "    for epoch in range(500000):\n",
    "        y_pred = model(x_data.cuda())\n",
    "        # Compute loss\n",
    "        loss = criterion(y_pred.cuda(), y_data.cuda())\n",
    "        print(epoch, loss.data[0])\n",
    "        # Zero gradients\n",
    "        optimizer.zero_grad()\n",
    "        # perform backward pass\n",
    "        loss.backward()\n",
    "        # update weights\n",
    "        optimizer.step()\n",
    "\n",
    "    hour_var = Variable(torch.Tensor([[7.0]]))\n",
    "    print(\"predict (after training)\", 7, model.forward(hour_var).data[0][0])\n",
    "\n",
    "\n",
    "def init_processes(rank, size, backend='gloo'):\n",
    "    \"\"\" Initialize the distributed environment. \"\"\"\n",
    "    os.environ['MASTER_ADDR'] = '192.168.0.12'\n",
    "    os.environ['MASTER_PORT'] = '29555'\n",
    "    dist.init_process_group(backend, rank=rank, world_size=size)\n",
    "    # print(\"MM\")\n",
    "    ## 实现代码\n",
    "    model = Model()\n",
    "    model = torch.nn.parallel.DistributedDataParallel(model.cuda())\n",
    "    criterion = torch.nn.MSELoss(size_average=False).cuda()\n",
    "    optimizer = torch.optim.SGD(model.parameters(), lr=0.01)\n",
    "\n",
    "    cudnn.benchmark = True\n",
    "    x_data = Variable(torch.Tensor([[1.0], [2.0], [3.0]]))\n",
    "    y_data = Variable(torch.Tensor([[2.0], [4.0], [6.0]]))\n",
    "\n",
    "    for epoch in range(500000):\n",
    "        y_pred = model(x_data.cuda())\n",
    "        # Compute loss\n",
    "        loss = criterion(y_pred.cuda(), y_data.cuda())\n",
    "        print(epoch, loss.data[0])\n",
    "        # Zero gradients\n",
    "        optimizer.zero_grad()\n",
    "        # perform backward pass\n",
    "        loss.backward()\n",
    "        # update weights\n",
    "        optimizer.step()\n",
    "    hour_var = Variable(torch.Tensor([[7.0]]))\n",
    "    print(\"predict (after training)\", 7, model.forward(hour_var).data[0][0])\n",
    "\n",
    "def main():\n",
    "    size = 2\n",
    "    processes=[]\n",
    "    for i in range(size):\n",
    "        p = Process(target=init_processes, args=(i, size))\n",
    "        p.start()\n",
    "        processes.append(p)\n",
    "        # init_processes(i, size)\n",
    "    for p in processes:\n",
    "        p.join()\n",
    "\n",
    "if __name__ == \"__main__\":\n",
    "    start_time = time.time()\n",
    "    main()\n",
    "    end_time = time.time()\n",
    "    print(\"耗时：\", end_time-start_time)"
   ],
   "metadata": {
    "collapsed": false
   },
   "id": "91f633ac49763ecc"
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "a75f5c8edb6500a6",
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# 另一个案例：单机多卡\n",
    "\n",
    "import os\n",
    "import torch\n",
    "import torch.cuda\n",
    "import torch.distributed as dist # 多卡通信\n",
    "import torch.multiprocessing as mp # dpp启动\n",
    "from torch.cuda.amp import GradScaler # 混合精度训练用\n",
    "from torch.utils.data.distributed import DistributedSampler # 分布式采样\n",
    "from torch.nn.parallel import DistributedDataParallel as DDP # 模型传递\n",
    "\n",
    "# 启动1:\n",
    "# mp.spawn()，下面代码就是这个模式\n",
    "\n",
    "# 启动2:采用torchrun 命令启动所有机器上的gpu\n",
    "# torchrun --standalone --nproc_per_node=2 ddp_main.py --gpu 0,1\n",
    "\n",
    "# 进程初始化\n",
    "def init_ddp(local_rank): # 当前cuda编号\n",
    "    torch.cuda.set_device(local_rank)\n",
    "    os.environ['RANK'] = str(local_rank)\n",
    "    # 初始化进程组\n",
    "    dist.init_process_group(backend='nccl', init_method='env://')\n",
    "\n",
    "def get_ddp_generator(seed=3407):\n",
    "    # 对每个进程使用不同的随机种子，增强训练的随机性\n",
    "    local_rank = dist.get_rank()\n",
    "    g = torch.Generator()\n",
    "    g.manual_seed(seed + local_rank)\n",
    "    return g\n",
    "# 对多个进程对计算结果进行汇总\n",
    "def reduce_tensor(tensor: torch.Tensor):\n",
    "    rt = tensor.clone()\n",
    "    dist.all_reduce(rt, op=dist.reduce_op.SUM)\n",
    "    rt /= dist.get_world_size()\n",
    "    return rt\n",
    "# 数据采样：训练和测试\n",
    "\n",
    "def get_dataloader(path, args, tokenizer, train:bool): \n",
    "    '''\n",
    "    根据给定的路径获取数据，并将数据和训练标志传递给数据加载器，这样可以方便地从给定路径加载数据并生成数据加载器，以供后续的模型训练和评估使用。\n",
    "    path：数据存放路径\n",
    "    tokenizer：分词器\n",
    "    train：是否是训练阶段\n",
    "    '''\n",
    "    texts, labels = load_dataset(path, args['num_labels'])\n",
    "    texts = tokenizer(texts, padding='max_length', truncation=True, return_tensors='pt', max_length=args['max_length']) \n",
    "    data = TensorDataset(texts['input_ids'], texts['attention_mask'], torch.tensor(labels)) \n",
    "    \n",
    "    if train:\n",
    "        train_sampler = DistributedSampler(data, shuffle=True)  # 创建一个分布式随机采样器。\n",
    "        g = get_ddp_generator()\n",
    "        dataloader = DataLoader(dataset=data,\n",
    "                                batch_size=args['batch_size'],\n",
    "                                num_workers=args['num_workers'],\n",
    "                                pin_memory=True,\n",
    "                                shuffle=False, # 已经指定了采样器\n",
    "                                sampler=train_sampler, #采用随机采样器。\n",
    "                                generator=g) \n",
    "        \n",
    "    else:\n",
    "        test_sampler = DistributedSampler(data, shuffle=False) # 创建一个顺序采样器。\n",
    "        dataloader = DataLoader(dataset=data,\n",
    "                                batch_size=args['batch_size'],\n",
    "                                num_workers=args['num_workers'],\n",
    "                                pin_memory=True,\n",
    "                                shuffle=False,\n",
    "                                sampler=test_sampler #采用顺序采样器。\n",
    "                                )\n",
    "    return dataloader\n",
    "# 训练函数\n",
    "\n",
    "def train(model, train_dataloader, optimizer, scheduler, criterion, actual_epoch, scaler, args):\n",
    "    model.train()\n",
    "    \n",
    "    tr_loss = 0\n",
    "    num_train_samples = 0\n",
    "    \n",
    "    for step, batch in enumerate(train_dataloader):\n",
    "        batch = tuple(t.cuda(non_blocking=True) for t in batch)\n",
    "        b_input_ids, b_input_mask, b_labels = batch\n",
    "        with torch.cuda.amp.autocast(): # amp 自动混合精度训练\n",
    "\t        output = model(b_input_ids, attention_mask=b_input_mask, labels=b_labels) # 运行到这一行会增加一下显存\n",
    "\t        loss = criterion(output.logits.view(-1,args['num_labels']), b_labels.type_as(output.logits).view(-1,args['num_labels']))\n",
    "        reduced_loss = reduce_tensor(loss.data)  # 对并行进程计算的多个 loss 取平均\n",
    "        if dist.get_rank() == 0:  # 防止重复输出\n",
    "            print(\"\\nOutput Loss: \", reduced_loss.item())\n",
    "        tr_loss += reduced_loss.item()\n",
    "        # 并行状态下的更新，不同进程分别根据自己计算的 loss 更新数据\n",
    "        optimizer.zero_grad()\n",
    "        scaler.scale(loss).backward() # 混合精度\n",
    "        scaler.step(optimizer)  #  运行到这一行会增加一下显存\n",
    "        # 下面四行，多个进程只执行一次\n",
    "        scheduler.step() # 梯度schedule\n",
    "        scaler.update() # 更新梯度\n",
    "        num_train_samples += b_labels.size(0) #将批次中的样本数量添加到 num_train_samples 中。\n",
    "        torch.cuda.empty_cache()  # 释放GPU reserved memory显存\n",
    "    \n",
    "    epoch_train_loss = tr_loss / num_train_samples  # num_train_samples 代表每个进程承接的样本数量，由于上面已经有对loss取平均的操作，这里分母无需再乘以进程数\n",
    "\n",
    "    if dist.get_rank() == 0:\n",
    "        print(\"\\nTrain loss after Epoch {} : {}\".format(actual_epoch, epoch_train_loss))\n",
    "# validate\n",
    "\n",
    "@torch.no_grad()\n",
    "def validate(model, valid_dataloader, criterion, epoch, args, threshold=0.5):\n",
    "\n",
    "    model.eval()\n",
    "\n",
    "    eval_loss = 0.0\n",
    "    num_eval_samples = 0\n",
    "    \n",
    "    pred_labels = []\n",
    "    true_labels = []\n",
    "    \n",
    "    for step, batch in enumerate(valid_dataloader):\n",
    "        \n",
    "        batch = tuple(t.cuda(non_blocking=True) for t in batch)\n",
    "        b_input_ids, b_input_mask, b_labels = batch\n",
    "        \n",
    "        with torch.no_grad():\n",
    "            with torch.cuda.amp.autocast(): # 混合精度\n",
    "\t            output = model(b_input_ids, attention_mask=b_input_mask)\n",
    "\t            logits = output.logits\n",
    "                        \n",
    "            loss = criterion(logits.view(-1,args['num_labels']), b_labels.type_as(logits).view(-1,args['num_labels']))\n",
    "            \n",
    "            reduced_loss = reduce_tensor(loss.data)\n",
    "            eval_loss += reduced_loss.item()\n",
    "            \n",
    "            pred_label = torch.sigmoid(logits)\n",
    "            pred_label = pred_label.to('cpu').numpy()\n",
    "            b_labels = b_labels.to('cpu').numpy()\n",
    "            \n",
    "        pred_labels.append(pred_label)\n",
    "        true_labels.append(b_labels)\n",
    "\n",
    "        num_eval_samples += b_labels.shape[0]  # 这里是针对单个 进程 的 计算样本数\n",
    "    \n",
    "    epoch_eval_loss = eval_loss/num_eval_samples  \n",
    "    \n",
    "    if dist.get_rank() == 0:\n",
    "        print(\"Validation loss after Epoch {} : {}\".format(epoch, epoch_eval_loss))\n",
    "# 进程主函数：单机多卡模拟\n",
    "\n",
    "\n",
    "def initialise_model(*args):\n",
    "    pass\n",
    "\n",
    "def main(local_rank, args):  # 参数列表更新\n",
    "    init_ddp(local_rank)  ### 进程初始化\n",
    "\n",
    "    best_macro = 0\n",
    "\n",
    "    model, tokenizer = initialise_model(args['modelname'], args['num_labels'])\n",
    "    model.cuda()\n",
    "    # 尽管会降低GPU利用率，但可以提高模型在多卡场景下的表现\n",
    "    model = torch.nn.SyncBatchNorm.convert_sync_batchnorm(model)  # BN层同步\n",
    "    \n",
    "    num_gpus = torch.cuda.device_count()\n",
    "    if num_gpus > 1:\n",
    "        print('use {} gpus!'.format(num_gpus))\n",
    "        # 数据并行：单机多gpu，对模型封装\n",
    "        model = torch.nn.parallel.DistributedDataParallel(model, device_ids=[local_rank], output_device=local_rank)  ### 套 DDP\n",
    "    \n",
    "    num_training_steps = args['num_epochs'] * (args['num_samples'] // args['batch_size']) #总的训练步数\n",
    "    \n",
    "    if args['requires_grad']:  # 权重衰减\n",
    "        param_optimizer = list(model.named_parameters())\n",
    "        no_decay = ['bias', 'LayerNorm.bias', 'LayerNorm.weight']\n",
    "        # 设置模型参数的权重衰减\n",
    "        optimizer_grouped_parameters = [\n",
    "            {'params': [p for n, p in param_optimizer if not any(nd in n for nd in no_decay)],\n",
    "             'weight_decay': 0.01},\n",
    "            {'params': [p for n, p in param_optimizer if any(nd in n for nd in no_decay)], 'weight_decay': 0.0}\n",
    "        ]\n",
    "        optimizer = AdamW(optimizer_grouped_parameters, lr=float(args['learning_rate'])) # 部分参数更新\n",
    "    else:\n",
    "        optimizer = AdamW(model.parameters(), lr=float(args['learning_rate'])) # 部分参数更新\n",
    "    \n",
    "    scheduler = get_linear_schedule_with_warmup(optimizer,  num_warmup_steps=100, num_training_steps=num_training_steps) #创建学习率调度器。\n",
    "    \n",
    "    scaler = GradScaler()  ###  用于混合精度训练,FP32和FP16\n",
    "    criterion = BCEWithLogitsLoss().cuda() #定义损失函数。\n",
    "\n",
    "    train_dataloader = get_dataloader(args['traincsvpath'], args, tokenizer, train=True)\n",
    "    valid_dataloader = get_dataloader(args['valcsvpath'], args, tokenizer, train=False)\n",
    "\n",
    "    for actual_epoch in trange(args['num_epochs'], desc=\"Epoch\"):\n",
    "        if local_rank == 0:  ### 防止每个进程都输出一次，只允许0号进程打印中间结果\n",
    "            print(\"begin training of epoch %d / %d\" % (actual_epoch + 1, args['num_epochs']))\n",
    "        \n",
    "        train_dataloader.sampler.set_epoch(actual_epoch)  # 训练时每次的 sampling 顺序不同\n",
    "        train(model, train_dataloader, optimizer, scheduler, criterion, actual_epoch, scaler, args)   \n",
    "        \n",
    "        if local_rank == 0:\n",
    "            print(f'begin validating')  \n",
    "        macro = validate(model, valid_dataloader, criterion, actual_epoch, args) #在验证集上评估模型。\n",
    "     \n",
    "        if macro > best_macro:\n",
    "            best_macro = macro\n",
    "            if local_rank == 0:  # 防止每个进程都保存一次\n",
    "                save_model(actual_epoch, model, scaler, args['model_save_dir'] + '/best_macro_model_DDP_direct.pt')\n",
    "    \n",
    "    dist.destroy_process_group()  # 消除进程组，和 init_process_group 相对\n",
    "# main函数\n",
    "\n",
    "import argparse, time\n",
    "\n",
    "def main():\n",
    "    pass\n",
    "def test():\n",
    "    pass\n",
    "\n",
    "if __name__ == '__main__':\n",
    "    parser = argparse.ArgumentParser()\n",
    "    parser.add_argument('-args', help=\"priority\", type=bool, required=False, default=True)\n",
    "    parser.add_argument('-gpu', default='0,1', type=str, help='gpu device ids for CUDA_VISIBLE_DEVICES')\n",
    "    parser.add_argument('-mode', help=\"train&test\", type=str, required=False, default='train')\n",
    "    parser.add_argument('-requires_grad', help=\"whether to weight_decay\", type= bool, required=False, default=True)\n",
    "    args = parser.parse_args()\n",
    "    os.environ['MASTER_ADDR'] = 'localhost'  # 0号机器的IP\n",
    "    os.environ['MASTER_PORT'] = '19198'  # 0号机器的可用端口\n",
    "    os.environ['CUDA_VISIBLE_DEVICES'] = args['gpu']  # 使用哪些GPU\n",
    "    world_size = torch.cuda.device_count()\n",
    "    os.environ['WORLD_SIZE'] = str(world_size)\n",
    "    os.environ[\"PYTORCH_CUDA_ALLOC_CONF\"] = \"max_split_size_mb:128\"\n",
    "    os.environ[\"TOKENIZERS_PARALLELISM\"] = \"false\"  # 指定程序在分词时不并行执行\n",
    "    \n",
    "    # train模式\n",
    "    if args['mode'] == 'train':\n",
    "        time_start = time.time()\n",
    "        mp.spawn(fn=main, args=(args, ), nprocs=world_size) # 启动进程\n",
    "        time_elapsed = time.time() - time_start\n",
    "        print(f'\\ntime elapsed: {time_elapsed:.2f} seconds.')\n",
    "    # test模式\n",
    "    elif args['mode'] == 'test':  \n",
    "        time_start = time.time()\n",
    "        mp.spawn(fn=test, args=(args, ), nprocs=world_size)\n",
    "        time_elapsed = time.time() - time_start\n",
    "        print(f'\\ntime elapsed: {time_elapsed:.2f} seconds.')\n"
   ]
  }
 ],
 "metadata": {
  "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.8.10"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}