12_distributed_tensorflow.ipynb

{
  "nbformat": 4,
  "nbformat_minor": 0,
  "metadata": {
    "hide_input": false,
    "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.7.5"
    },
    "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
    },
    "varInspector": {
      "cols": {
        "lenName": 16,
        "lenType": 16,
        "lenVar": 40
      },
      "kernels_config": {
        "python": {
          "delete_cmd_postfix": "",
          "delete_cmd_prefix": "del ",
          "library": "var_list.py",
          "varRefreshCmd": "print(var_dic_list())"
        },
        "r": {
          "delete_cmd_postfix": ") ",
          "delete_cmd_prefix": "rm(",
          "library": "var_list.r",
          "varRefreshCmd": "cat(var_dic_list()) "
        }
      },
      "types_to_exclude": [
        "module",
        "function",
        "builtin_function_or_method",
        "instance",
        "_Feature"
      ],
      "window_display": false
    },
    "colab": {
      "name": "12_distributed_tensorflow.ipynb",
      "provenance": [],
      "collapsed_sections": []
    },
    "accelerator": "GPU"
  },
  "cells": [
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "kf-btQN3VCR3"
      },
      "source": [
        "# Distribuindo o TensorFlow por dispositivos e servidores"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "_Ii_-XOiVCSC"
      },
      "source": [
        "### Bibliotecas básicas"
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "ExecuteTime": {
          "end_time": "2021-01-20T09:57:18.275934Z",
          "start_time": "2021-01-20T09:56:36.091779Z"
        },
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "un4-TVdEVCSD",
        "outputId": "7a1777b6-35a1-4720-a087-35dfb940fe80"
      },
      "source": [
        "import numpy as np\n",
        "import pandas as pd\n",
        "from functools import partial\n",
        "import matplotlib.pyplot as plt\n",
        "%matplotlib inline\n",
        "\n",
        "import warnings\n",
        "warnings.filterwarnings('ignore')\n",
        "\n",
        "import tensorflow as tf\n",
        "print(tf.__version__)"
      ],
      "execution_count": 1,
      "outputs": [
        {
          "output_type": "stream",
          "text": [
            "2.4.0\n"
          ],
          "name": "stdout"
        }
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "FTyX5IwLWfuN"
      },
      "source": [
        "tf.compat.v1.disable_eager_execution()"
      ],
      "execution_count": 3,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "fYR1EP0eVCSH"
      },
      "source": [
        "### Servidor Local"
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "1y947Xz1WSrb",
        "outputId": "903a38ba-3556-4c70-9d6e-27883a3ccbbe"
      },
      "source": [
        "tf.compat.v1.reset_default_graph()\n",
        "\n",
        "c = tf.constant(\"Hello distributed TensorFlow!\")\n",
        "server = tf.compat.v1.train.Server.create_local_server()\n",
        "\n",
        "with tf.compat.v1.Session(server.target) as sess:\n",
        "    print(sess.run(c))"
      ],
      "execution_count": null,
      "outputs": [
        {
          "output_type": "stream",
          "text": [
            "b'Hello distributed TensorFlow!'\n"
          ],
          "name": "stdout"
        }
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "vzEBnB85b7s9"
      },
      "source": [
        "#### Posicionamento Simples"
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "5r53E62qWyvu",
        "outputId": "8c2a315a-0ed1-413c-e835-dd8ec995cd45"
      },
      "source": [
        "with tf.compat.v1.device('/cpu:0'):\n",
        "  a = tf.Variable(3.0)\n",
        "  b = tf.constant(4.0)\n",
        "c = a * b\n",
        "\n",
        "config = tf.compat.v1.ConfigProto()\n",
        "config.log_device_placement = True\n",
        "\n",
        "sess = tf.compat.v1.Session(config=config)\n",
        "a.initializer.run(session=sess)\n",
        "sess.run(c)"
      ],
      "execution_count": null,
      "outputs": [
        {
          "output_type": "stream",
          "text": [
            "Device mapping:\n",
            "/job:localhost/replica:0/task:0/device:GPU:0 -> device: 0, name: Tesla P100-PCIE-16GB, pci bus id: 0000:00:04.0, compute capability: 6.0\n",
            "\n"
          ],
          "name": "stdout"
        },
        {
          "output_type": "execute_result",
          "data": {
            "text/plain": [
              "12.0"
            ]
          },
          "metadata": {
            "tags": []
          },
          "execution_count": 4
        }
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "FIKwynxFcEwL"
      },
      "source": [
        "#### Posicionamento dinâmico"
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "2dSCmT2YbPyI",
        "outputId": "9dbfa86f-884a-4356-ee15-ccc77dc292bb"
      },
      "source": [
        "def variables_on_cpu(op):\n",
        "  if op.type == 'Variable':\n",
        "    return '/cpu:0'\n",
        "  else:\n",
        "    return '/gpu:0'\n",
        "\n",
        "with tf.compat.v1.device(variables_on_cpu):\n",
        "  a = tf.Variable(3.0)\n",
        "  b = tf.constant(4.0)\n",
        "c = a * b\n",
        "\n",
        "config = tf.compat.v1.ConfigProto()\n",
        "config.log_device_placement = True\n",
        "\n",
        "sess = tf.compat.v1.Session(config=config)\n",
        "a.initializer.run(session=sess)\n",
        "sess.run(c)"
      ],
      "execution_count": null,
      "outputs": [
        {
          "output_type": "stream",
          "text": [
            "Device mapping:\n",
            "/job:localhost/replica:0/task:0/device:GPU:0 -> device: 0, name: Tesla P100-PCIE-16GB, pci bus id: 0000:00:04.0, compute capability: 6.0\n",
            "\n"
          ],
          "name": "stdout"
        },
        {
          "output_type": "execute_result",
          "data": {
            "text/plain": [
              "12.0"
            ]
          },
          "metadata": {
            "tags": []
          },
          "execution_count": 5
        }
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "K3ec7gEacLW0"
      },
      "source": [
        "#### Posicionamento suave"
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "h7TSlB3mcN4l",
        "outputId": "6fbeabf5-d746-4868-ece5-267b265836ae"
      },
      "source": [
        "with tf.compat.v1.device('/gpu:0'):\n",
        "  a = tf.Variable(3)\n",
        "\n",
        "config = tf.compat.v1.ConfigProto()\n",
        "config.log_device_placement = True\n",
        "config.allow_soft_placement = True\n",
        "\n",
        "sess = tf.compat.v1.Session(config=config)\n",
        "sess.run(a.initializer)"
      ],
      "execution_count": null,
      "outputs": [
        {
          "output_type": "stream",
          "text": [
            "Device mapping:\n",
            "/job:localhost/replica:0/task:0/device:GPU:0 -> device: 0, name: Tesla P100-PCIE-16GB, pci bus id: 0000:00:04.0, compute capability: 6.0\n",
            "\n"
          ],
          "name": "stdout"
        }
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "HuCktBV5i998"
      },
      "source": [
        "#### Dependência de controle"
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "b-m9O24Ei9iM"
      },
      "source": [
        "a = tf.constant(1.0)\n",
        "b = a + 2.0\n",
        "\n",
        "with tf.compat.v1.control_dependencies([a, b]):\n",
        "  x = tf.constant(3.0)\n",
        "  y = tf.constant(4.0)\n",
        "z = x + y"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "kx55pQF9nkeZ"
      },
      "source": [
        "### Vários dispositivos em vários servidores"
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "4eYdIomOntcL"
      },
      "source": [
        "cluster_spec = tf.compat.v1.train.ClusterSpec({\n",
        "    'ps': [\n",
        "        '127.0.0.1:2221',\n",
        "        '127.0.0.1:2222'\n",
        "    ],\n",
        "    'worker': [\n",
        "        '127.0.0.1:2223',\n",
        "        '127.0.0.1:2224',\n",
        "        '127.0.0.1:2225'\n",
        "    ]})\n",
        "\n",
        "\n",
        "task_ps0 = tf.compat.v1.train.Server(cluster_spec, job_name='ps', task_index=0)\n",
        "task_ps1 = tf.compat.v1.train.Server(cluster_spec, job_name='ps', task_index=1)\n",
        "\n",
        "task_worker0 = tf.compat.v1.train.Server(cluster_spec, job_name='worker', task_index=0)\n",
        "task_worker1 = tf.compat.v1.train.Server(cluster_spec, job_name='worker', task_index=1)\n",
        "task_worker2 = tf.compat.v1.train.Server(cluster_spec, job_name='worker', task_index=2)"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "Oi62vYPHuzr8"
      },
      "source": [
        "#### Fixar operações em dispositivos e servidores"
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "cax8XwIgu1AQ",
        "outputId": "2ab08257-6b14-4aad-f8af-c73999635e9a"
      },
      "source": [
        "tf.compat.v1.reset_default_graph()\n",
        "\n",
        "with tf.compat.v1.device('/job:ps'):\n",
        "    a = tf.Variable(1.0, name='a')\n",
        "\n",
        "with tf.compat.v1.device('/job:worker'):\n",
        "    b = a + 2\n",
        "\n",
        "with tf.compat.v1.device('/job:worker/task:1'):\n",
        "    c = a + b\n",
        "\n",
        "with tf.compat.v1.Session('grpc://127.0.0.1:2221') as sess:\n",
        "    sess.run(a.initializer)\n",
        "    print(c.eval())"
      ],
      "execution_count": null,
      "outputs": [
        {
          "output_type": "stream",
          "text": [
            "4.0\n"
          ],
          "name": "stdout"
        }
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "16TG4oasw8tZ"
      },
      "source": [
        "#### Particionando as variáveis em múltiplos servidores"
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "0PgDIRwwxAc8"
      },
      "source": [
        "tf.compat.v1.reset_default_graph()\n",
        "\n",
        "with tf.compat.v1.device(tf.compat.v1.train.replica_device_setter(ps_tasks=2, ps_device='/job:ps', worker_device='/job:worker')):\n",
        "    v1 = tf.Variable(1.0, name='v1')\n",
        "    v2 = tf.Variable(2.0, name='v2')\n",
        "    v3 = tf.Variable(3.0, name='v3')\n",
        "    s = v1 + v2\n",
        "    with tf.compat.v1.device('/task:1'):\n",
        "        p1 = 2 * s\n",
        "        with tf.compat.v1.device('/cpu:0'):\n",
        "            p2 = 3 * s\n",
        "\n",
        "config = tf.compat.v1.ConfigProto()\n",
        "config.log_device_placement = True\n",
        "\n",
        "with tf.compat.v1.Session('grpc://127.0.0.1:2221', config=config) as sess:\n",
        "    v1.initializer.run()"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "_Zi8-tUEFdII"
      },
      "source": [
        "### Carregando dados diretamento do Grafo"
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "_RlUAh3cFg4f"
      },
      "source": [
        "tf.compat.v1.reset_default_graph()\n",
        "\n",
        "test_csv = open('my_test.csv', 'w')\n",
        "test_csv.write('x1, x2 , target\\n')\n",
        "test_csv.write('1., , 0\\n')\n",
        "test_csv.write('4., 5., 1\\n')\n",
        "test_csv.write('7., 8., 0\\n')\n",
        "test_csv.close()\n",
        "\n",
        "filename_queue = tf.compat.v1.FIFOQueue(capacity=10, dtypes=[tf.string], shapes=[()])\n",
        "filename = tf.compat.v1.placeholder(tf.string)\n",
        "enqueue_filename = filename_queue.enqueue([filename])\n",
        "close_filename_queue = filename_queue.close()\n",
        "\n",
        "reader = tf.compat.v1.TextLineReader(skip_header_lines=1)\n",
        "key, value = reader.read(filename_queue)\n",
        "\n",
        "x1, x2, target = tf.compat.v1.decode_csv(value, record_defaults=[[-1.], [-1.], [-1]])\n",
        "features = tf.compat.v1.stack([x1, x2])\n",
        "\n",
        "instance_queue = tf.compat.v1.RandomShuffleQueue(capacity=10, min_after_dequeue=2, dtypes=[tf.float32, tf.int32], shapes=[[2],[]], \n",
        "                                                 name='instance_q', shared_name='shared_instance_q')\n",
        "enqueue_instance = instance_queue.enqueue([features, target])\n",
        "close_instance_queue = instance_queue.close()\n",
        "\n",
        "minibatch_instances, minibatch_targets = instance_queue.dequeue_up_to(2)\n",
        "\n",
        "with tf.compat.v1.Session() as sess:\n",
        "    sess.run(enqueue_filename, feed_dict={filename: 'my_test.csv'})\n",
        "    sess.run(close_filename_queue)\n",
        "    try:\n",
        "        while True:\n",
        "            sess.run(enqueue_instance)\n",
        "    except tf.compat.v1.errors.OutOfRangeError as ex:\n",
        "      print('No more files to read')\n",
        "    sess.run(close_instance_queue)\n",
        "    try:\n",
        "        while True:\n",
        "            print(sess.run([minibatch_instances, minibatch_targets]))\n",
        "    except tf.compat.v1.errors.OutOfRangeError as ex:\n",
        "        print('No more training instances')"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "17SdVKfoOyLu"
      },
      "source": [
        "### Queue runners e coordinators"
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "y-OoTyYYO0Ht"
      },
      "source": [
        "tf.compat.v1.reset_default_graph()\n",
        "\n",
        "filename_queue = tf.compat.v1.FIFOQueue(capacity=10, dtypes=[compat.v1.string], shapes=[()])\n",
        "filename = tf.compat.v1.placeholder(tf.string)\n",
        "enqueue_filename = filename_queue.enqueue([filename])\n",
        "close_filename_queue = filename_queue.close()\n",
        "\n",
        "reader = tf.compat.v1.TextLineReader(skip_header_lines=1)\n",
        "key, value = reader.read(filename_queue)\n",
        "\n",
        "x1, x2, target = tf.comapt.v1.decode_csv(value, record_defaults=[[-1.], [-1.], [-1]])\n",
        "features = tf.compat.v1.stack([x1, x2])\n",
        "\n",
        "instance_queue = tf.compat.v1.RandomShuffleQueue( capacity=10, min_after_dequeue=2, dtypes=[tf.float32, tf.int32], shapes=[[2],[]], \n",
        "                                       name='instance_q', shared_name='shared_instance_q')\n",
        "enqueue_instance = instance_queue.enqueue([features, target])\n",
        "close_instance_queue = instance_queue.close()\n",
        "\n",
        "minibatch_instances, minibatch_targets = instance_queue.dequeue_up_to(2)\n",
        "\n",
        "n_threads = 5\n",
        "queue_runner = tf.compat.v1.train.QueueRunner(instance_queue, [enqueue_instance] * n_threads)\n",
        "coord = tf.compat.v1.train.Coordinator()\n",
        "\n",
        "with tf.compat.v1.Session() as sess:\n",
        "    sess.run(enqueue_filename, feed_dict={filename: 'my_test.csv'})\n",
        "    sess.run(close_filename_queue)\n",
        "    enqueue_threads = queue_runner.create_threads(sess, coord=coord, start=True)\n",
        "    try:\n",
        "        while True:\n",
        "            print(sess.run([minibatch_instances, minibatch_targets]))\n",
        "    except tf.errors.OutOfRangeError as ex:\n",
        "        print('No more training instances')"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "iEtFlCs6Pdyo"
      },
      "source": [
        "#### Com leitores multithreads"
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "YnYizdUuPd8S"
      },
      "source": [
        "tf.compat.v1.reset_default_graph()\n",
        "\n",
        "def read_and_push_instance(filename_queue, instance_queue):\n",
        "    reader = tf.compat.v1.TextLineReader(skip_header_lines=1)\n",
        "    key, value = reader.read(filename_queue)\n",
        "    x1, x2, target = tf.compat.v1.decode_csv(value, record_defaults=[[-1.], [-1.], [-1]])\n",
        "    features = tf.compat.v1.stack([x1, x2])\n",
        "    enqueue_instance = instance_queue.enqueue([features, target])\n",
        "    return enqueue_instance\n",
        "\n",
        "filename_queue = tf.compat.v1.FIFOQueue(capacity=10, dtypes=[tf.string], shapes=[()])\n",
        "filename = tf.compat.v1.placeholder(tf.string)\n",
        "enqueue_filename = filename_queue.enqueue([filename])\n",
        "close_filename_queue = filename_queue.close()\n",
        "\n",
        "instance_queue = tf.compat.v1.RandomShuffleQueue(capacity=10, min_after_dequeue=2, dtypes=[tf.float32, tf.int32], shapes=[[2],[]], \n",
        "                                       name='instance_q', shared_name='shared_instance_q')\n",
        "\n",
        "minibatch_instances, minibatch_targets = instance_queue.dequeue_up_to(2)\n",
        "\n",
        "read_and_enqueue_ops = [read_and_push_instance(filename_queue, instance_queue) for i in range(5)]\n",
        "queue_runner = tf.compat.v1.train.QueueRunner(instance_queue, read_and_enqueue_ops)\n",
        "\n",
        "with tf.compat.v1.Session() as sess:\n",
        "    sess.run(enqueue_filename, feed_dict={filename: 'my_test.csv'})\n",
        "    sess.run(close_filename_queue)\n",
        "    coord = tf.compat.v1.train.Coordinator()\n",
        "    enqueue_threads = queue_runner.create_threads(sess, coord=coord, start=True)\n",
        "    try:\n",
        "        while True:\n",
        "            print(sess.run([minibatch_instances, minibatch_targets]))\n",
        "    except tf.errors.OutOfRangeError as ex:\n",
        "        print('No more training instances')"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "3cGt6UDQAdbd"
      },
      "source": [
        "### Definindo um tempo limite"
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "QFXd6TuZActQ",
        "outputId": "66801bbb-06d9-4995-8da5-846b50e5039f"
      },
      "source": [
        "tf.compat.v1.reset_default_graph()\n",
        "\n",
        "q = tf.compat.v1.FIFOQueue(capacity=10, dtypes=[tf.float32], shapes=[()])\n",
        "v = tf.compat.v1.placeholder(tf.float32)\n",
        "enqueue = q.enqueue([v])\n",
        "dequeue = q.dequeue()\n",
        "output = dequeue + 1\n",
        "\n",
        "config = tf.compat.v1.ConfigProto()\n",
        "config.operation_timeout_in_ms = 1000\n",
        "\n",
        "with tf.compat.v1.Session(config=config) as sess:\n",
        "    sess.run(enqueue, feed_dict={v: 1.0})\n",
        "    sess.run(enqueue, feed_dict={v: 2.0})\n",
        "    sess.run(enqueue, feed_dict={v: 3.0})\n",
        "    print(sess.run(output))\n",
        "    print(sess.run(output, feed_dict={dequeue: 5}))\n",
        "    print(sess.run(output))\n",
        "    print(sess.run(output))\n",
        "    try:\n",
        "        print(sess.run(output))\n",
        "    except tf.errors.DeadlineExceededError as ex:\n",
        "        print('Timed out while dequeuing')"
      ],
      "execution_count": 4,
      "outputs": [
        {
          "output_type": "stream",
          "text": [
            "2.0\n",
            "6.0\n",
            "3.0\n",
            "4.0\n",
            "Timed out while dequeuing\n"
          ],
          "name": "stdout"
        }
      ]
    }
  ]
}