Seminars/15-efficiency/efficiency.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "# `Практикум по программированию на языке Python`\n",
    "<br>\n",
    "\n",
    "## `Занятие 15: Методы повышения эффективности кода`\n",
    "<br><br>\n",
    "\n",
    "### `Мурат Апишев (mel-lain@yandex.ru)`\n",
    "**Источник:** https://github.com/MelLain/mipt-python\n",
    "[Запись 2022](https://www.youtube.com/watch?v=IzejCnDXrxg&t=2334s&ab_channel=%D0%9A%D0%B0%D1%84%D0%B5%D0%B4%D1%80%D0%B0%D0%9C%D0%9C%D0%9F)\n",
    "\n",
    "#### `Москва, 2023`"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "slideshow": {
     "slide_type": "skip"
    }
   },
   "outputs": [],
   "source": [
    "import warnings\n",
    "warnings.filterwarnings('ignore')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "### `Производительность Python`\n",
    "\n",
    "- Быстродействие кода на Python существенно уступает коду, написанному на компилируемых языках<br><br>\n",
    "- Но Python очень удобен и популярен, поэтому сделано много попыток его ускорения<br><br>\n",
    "- Популярен подход с написанием основного кода на C/C++/CUDA и предоставлением пользователю удобного интерфейса к нему в Python (numpy, pycuda, библиотеки для глубинного обучения)<br><br>\n",
    "- Однако часто хочется получать эффективный код прямо в Python"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "### `Возможные методы повышения производительности`\n",
    "\n",
    "- Многопоточность и многопроцессорность\n",
    "- Альтернативные реализации Python: Pypy, Jython, Iron Python\n",
    "- Библиотека Numba\n",
    "- Расширение Cython\n",
    "- Фреймворки Hadoop/Spark"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "### `Процесс и поток`\n",
    "\n",
    "- __Процесс__ - это единица приложения, работающая в операционной системе, имеет собственную изолированную память\n",
    "- __Поток__ - вычислительная единица внутри процесса, работает в памяти процесса, имеет собственный код для выполнения<br><br>\n",
    "- В процессе может быть >= 1 потока\n",
    "- Одно ядро процессора в моменте времени может выполнять код одного потока\n",
    "- Если потоков много, а ядро одно, то оно будет переключаться между ними, создавая иллюзию параллельности для пользователя\n",
    "- Если и потоков, и ядер много, то потоки могут действительно выполняться параллельно\n",
    "- Аналогичным образом работает параллелизм между процессами"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### `Мотивация` \n",
    "\n",
    "Рассмотрим простенький пример с обкачкой статей Хабра, однако не будем это делать в лоб"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "import requests\n",
    "import time \n",
    "\n",
    "from bs4 import BeautifulSoup # удобная библиотека для парсинга html\n",
    "from multiprocessing.dummy import Pool as ThreadPool"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [],
   "source": [
    "base_url = \"https://habr.com/ru/articles/\"\n",
    "\n",
    "def get_page(url, n_attempts=5, t_sleep=1, headers={}, **kwargs):\n",
    "    for i in range(n_attempts):\n",
    "        page = requests.get(url, headers=headers)\n",
    "        if page.status_code == 200:\n",
    "            return page\n",
    "        time.sleep(t_sleep)\n",
    "    print(\"DEBUG: url doesn't load\", url)\n",
    "    return None\n",
    "\n",
    "def get_habr_artical(artical_number):\n",
    "    \n",
    "    page = get_page(f\"{base_url}{artical_number}\")\n",
    "    if page:\n",
    "        return page.content\n",
    "    return None"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "CPU times: user 18.4 ms, sys: 14.8 ms, total: 33.2 ms\n",
      "Wall time: 440 ms\n"
     ]
    }
   ],
   "source": [
    "%%time\n",
    "\n",
    "page = get_page(f\"{base_url}{1}\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "DEBUG: url doesn't load https://habr.com/ru/articles/3\n",
      "DEBUG: url doesn't load https://habr.com/ru/articles/5\n",
      "DEBUG: url doesn't load https://habr.com/ru/articles/15\n",
      "DEBUG: url doesn't load https://habr.com/ru/articles/16\n",
      "CPU times: user 883 ms, sys: 25.8 ms, total: 909 ms\n",
      "Wall time: 38.9 s\n"
     ]
    }
   ],
   "source": [
    "%%time\n",
    "N = 20\n",
    "\n",
    "for i in range(1, N):\n",
    "    get_habr_artical(i)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "DEBUG: url doesn't load https://habr.com/ru/articles/15\n",
      "DEBUG: url doesn't load https://habr.com/ru/articles/16\n",
      "DEBUG: url doesn't load https://habr.com/ru/articles/3\n",
      "DEBUG: url doesn't load https://habr.com/ru/articles/5\n",
      "CPU times: user 682 ms, sys: 83.2 ms, total: 765 ms\n",
      "Wall time: 8.02 s\n"
     ]
    }
   ],
   "source": [
    "%%time\n",
    "N = 20\n",
    "\n",
    "with ThreadPool(10) as pool:\n",
    "    pages = pool.map(get_habr_artical, range(1, N))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "(15,\n",
       " ['Wiki-FAQ для Хабрахабра / Хабр',\n",
       "  'Мы знаем много недоделок на сайте… но! / Хабр',\n",
       "  'Маслов, Сокур и партнеры пиарят Google в России / Хабр',\n",
       "  'Подкасты на Хабрахабре / Хабр',\n",
       "  'Самопроизвольное разлогинивание / Хабр',\n",
       "  'Неожиданная популярность ресурсов среди региональных аудиторий / Хабр',\n",
       "  'А нужно ли удаление? / Хабр',\n",
       "  'Пометки у новых комментариев / Хабр',\n",
       "  'Stand-along cообщества против сообществ в рамках социальных сетей / Хабр',\n",
       "  'Поиск по сайту / Хабр',\n",
       "  'Рекомендации в профиле – что это такое? / Хабр',\n",
       "  'Миграция чистых социальных сетей в направлении stand-along сообществ / Хабр',\n",
       "  'Все домены Гугла / Хабр',\n",
       "  'Теги — тэги — таги / Хабр',\n",
       "  'RSS комментариев / Хабр'])"
      ]
     },
     "execution_count": 9,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "soups = [BeautifulSoup(page, 'lxml') for page in pages if page]\n",
    "titles = [soup.title.text for soup in soups]\n",
    "\n",
    "len(titles), titles"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "### `Создание потока в Python`\n",
    "\n",
    "- Для работы с потоками в Python есть стандартный модуль threading\n",
    "- Основным классом является `Thread` (поток)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "import threading"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "-"
    }
   },
   "source": [
    "Функциональное создание потока:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "1000\n"
     ]
    }
   ],
   "source": [
    "def thread_function(x):\n",
    "    print(x ** 3)\n",
    "\n",
    "thread = threading.Thread(target=thread_function, args=(10, ))\n",
    "thread.start()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- Потоку передается функция, которую он должен выполнить, и аргументы\n",
    "- После запуска `start` он будет выполняться до выхода из функции или ошибки\n",
    "- Запустить его второй раз не получится (RuntimeError)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "### `Создание потока в Python`\n",
    "\n",
    "- Поток также можно описать в виде класс-наследника `Thread`\n",
    "- Класс должен определять метод `run`"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "1000\n"
     ]
    }
   ],
   "source": [
    "class MyThread(threading.Thread):\n",
    "    def __init__(self, x, power):\n",
    "        threading.Thread.__init__(self)\n",
    "        self.x = x\n",
    "        self.power = power\n",
    "\n",
    "    def thread_function(self):\n",
    "        print(self.x ** self.power)\n",
    "\n",
    "    def run(self):\n",
    "        self.thread_function()\n",
    "\n",
    "print_thread = MyThread(x=10, power=3)\n",
    "print_thread.start()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "### `Потоки-демоны`\n",
    "\n",
    "- Потоки могут выполняться в обычном и фоновом (daemon) режимах\n",
    "- Потоки-демоны используются для выполнения фоновых задач (отправка ping-ов, очистка мусора), они имеют смысл только в процессе существования основного потока\n",
    "- При своем завершении процесс будет дожидаться окончания работы обычных потоков\n",
    "- Потоки-демоны будут автоматически уничтожены при завершении процесса\n",
    "- Такие потоки можно создавать и дальше за ними не следить, обычные же потоки нужно явно завершать<br><br>\n",
    "\n",
    "Создание потока в режиме демона:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [],
   "source": [
    "thread = threading.Thread(target=thread_function, args=(10, ), daemon=True)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "### `Основные методы и атрибуты класса Thread`\n",
    "\n",
    "Методы:\n",
    "\n",
    "- `start` - запуск потока\n",
    "- `run` - выполняет код потока\n",
    "- `join` - блокирует вызывающий поток до завершения потока, у которого вызван `join`\n",
    "- `is_alive` - возвращает флаг незавершенности потока\n",
    "\n",
    "Атрибуты:\n",
    "- `name` - имя потока\n",
    "- `daemon` - флаг потоком фоновый или обычный\n",
    "- `ident` - идентификатор потока, пока не вызван `start` он `None`."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### `Пример запуска потока`"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "wait started\n",
      "Is alive:  True\n",
      "Is daemon:  False\n",
      "Thread name:  Wait Thread\n",
      "Thread identificator:  139682461509376\n",
      "wait finished\n",
      "Is alive:  False\n"
     ]
    }
   ],
   "source": [
    "import time\n",
    "\n",
    "def wait(s):\n",
    "    print('wait started')\n",
    "    time.sleep(s)\n",
    "    print('wait finished')\n",
    "\n",
    "thread = threading.Thread(target=wait, args=(10, ), name='Wait Thread')\n",
    "thread.start()\n",
    "\n",
    "print('Is alive: ', thread.is_alive())\n",
    "print('Is daemon: ', thread.daemon)\n",
    "print('Thread name: ', thread.name)\n",
    "print('Thread identificator: ', thread.ident)\n",
    "thread.join()\n",
    "print('Is alive: ', thread.is_alive())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "### `Доступ к разделяемым ресурсам`\n",
    "\n",
    "- Потоки не имеют своей памяти и все работают с одной памятью родительского процесса\n",
    "- Порядок выполнения потоков недетерминирован, в многопоточной среде несколько потоков могут одновременно получить доступ к одному ресурсу\n",
    "- Без надлежащего контроля это может привести к ошибкам и падению процесса\n",
    "- Потоки нельзя просто так останавливать извне, поэтому ограничения нужно наложить заранее, до запуска<br><br>\n",
    "\n",
    "- Для контроля за доступом к ресурсам есть несколько инструментов:\n",
    "    - Мьютекс (Mutex, Lock)\n",
    "    - Рекурсивный мьютекс (RLock)\n",
    "    - Семафор (Semaphore)\n",
    "    - Событие (Event)\n",
    "    - Условная переменная (Condition variable)\n",
    "    - Барьер (Barrier)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "### `Простая блокировка Lock`\n",
    "\n",
    "- Mutex - это объект, который может в каждый момент времени быть занят не более чем одним потоком\n",
    "- Все потоки проверяют его на занятость и, если он занят, ожидают освобождения"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [],
   "source": [
    "mutex = threading.Lock()\n",
    "\n",
    "def thread_safe_function():\n",
    "    mutex.acquire()\n",
    "    # Critical section\n",
    "    mutex.release()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Для избежания взаимных блокировок (deadlocks) освобождение mutex нужно делать всегда, в т.ч. при ошибках и исключениях"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "### `Пример использования Mutex`\n",
    "\n",
    "Сравним потокобезопасный и непотокобезопасный коды в задаче инкремента разделяемого ресурса ([источник](https://geekbrains.ru/posts/python_multithreading_pt2))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [],
   "source": [
    "protected_resource, unprotected_resource = 0, 0\n",
    "mutex = threading.Lock()\n",
    "NUM = 100000"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [],
   "source": [
    "def safe_plus():\n",
    "    global protected_resource\n",
    "    for i in range(NUM):\n",
    "        with mutex:\n",
    "            protected_resource += 1\n",
    "\n",
    "def risky_plus():\n",
    "    global unprotected_resource\n",
    "    for i in range(NUM):\n",
    "        unprotected_resource += 1\n",
    "\n",
    "def safe_minus():\n",
    "    global protected_resource\n",
    "    for i in range(NUM):\n",
    "        with mutex:\n",
    "            protected_resource -= 1\n",
    "\n",
    "def risky_minus():\n",
    "    global unprotected_resource\n",
    "    for i in range(NUM):\n",
    "        unprotected_resource -= 1"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "### `Пример использования Mutex`"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Threadsafe result: 0\n",
      "Non-threadsafe result: 0\n"
     ]
    }
   ],
   "source": [
    "thread1 = threading.Thread(target=safe_plus)\n",
    "thread2 = threading.Thread(target=safe_minus)\n",
    "thread3 = threading.Thread(target=risky_plus)\n",
    "thread4 = threading.Thread(target=risky_minus)\n",
    "\n",
    "thread1.start(), thread2.start(), thread3.start(), thread4.start()\n",
    "thread1.join(), thread2.join(), thread3.join(), thread4.join()\n",
    "\n",
    "print (f'Threadsafe result: {protected_resource}')\n",
    "print (f'Non-threadsafe result: {unprotected_resource}')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "### `Прочие инструменты контроля доступа`\n",
    "\n",
    "- Рекурсивный мьютекс - такой же, как и обычный, но один поток может многократно входить в критическую секцию<br><br>\n",
    "- Семафор - дает доступ к ресурсу заданному числу потоков<br><br>\n",
    "- Событие - блокирует ресурс до выполнения некоторого условия, о котором уведовляет один из потоков<br><br>\n",
    "- Условная переменная - обертка типа Event вокруг других блокировок для блокирования по определенным событиям<br><br>\n",
    "- Барьер - создает точку синхронизации для заданного числа потоков, пока они все не дойдут до него, никто дальше не выполняется"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "### `Проблемы с многопоточностью в Python`\n",
    "\n",
    "- Существует несколько реализаций Python, наиболее популярная (стандартная) - CPython\n",
    "- Исторически CPython был реализован с глобальной блокировкой потоков Global Interpreter Lock (GIL)\n",
    "\n",
    "Что такое GIL:\n",
    "- GIL - это блокировка, которая не позволяет в каждый момент времени выполняться не более чем одному потоку\n",
    "- Для этого через равные промежутки времени (100 тиков) интерпретатор усыпляет работающий поток и дает возможность поработать один промежуток какому-то иному потоку\n",
    "- Остановка происходит даже в том случае, если поток всего один\n",
    "\n",
    "Зачем нужен GIL:\n",
    "- Реализация менеджмента памяти и сборки мусора в CPython не является потокобезопасной\n",
    "- Без него C-расширения языка, которые активно используются, не смогут нормально работать\n",
    "- За счет отсутствия необходимости учета разных ссылок на объекты из разных потоков однопоточные приложения выполняются с GIL быстрее\n",
    "\n",
    "Где еще есть GIL:\n",
    "- В Jython и IronPython GIL нет\n",
    "- В PyPy есть\n",
    "- В Cython есть, но может выключаться на секцию кода"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "### `К чему приводит наличие GIL`\n",
    "\n",
    "- Многопоточные приложения, ориентированные на использование вычислений на CPU (не I/O операций) не могут работать параллельно<br><br>\n",
    "- Вычисления не только не становятся быстрее, но и сильно замедляются при увеличении числа потоков, поскольку каждое переключение требует большого числа системных вызовов<br><br>\n",
    "- Замедление на многоядерном процессоре будет еще больше, чем на одноядерном, поскольку интерпретатор не синхронизирован с ОС (он хочет разместить все на одном ядре, а ОС пытается разложить потоки на разные ядра)<br><br>\n",
    "- В интерпретаторе нет своего расписания очереди потоков, используется расписание ОС, это приводит к проблемам при обработке сигналов"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "### `Реакция на сигналы ОС`\n",
    "\n",
    "- В Python сигналы могут обрабатываться только в главном потоке<br><br>\n",
    "- При этом интерпретатор не может сам потребовать, чтобы на следующем промежутке включился главный поток<br><br>\n",
    "- Он пытается переключать потоки с GIL не раз в 100 тиков, а каждый тик<br><br>\n",
    "- В результате выполнение кода может замедлиться, и до перехода в главный поток Python не реагирует на сигналы<br><br>\n",
    "- Помимо этого, внутри одного тика интерпретатор невозможно прервать<br><br>\n",
    "- Тик длится не фиксированное время, а некоторый набор инструкций (по-умолчанию - 100 байткодов), в которые интерпретатор транслировал исходный код<br><br>"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "### `Пример проблем с GIL`\n",
    "\n",
    "Однопоточный код:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Elapsed time: 2.62\n"
     ]
    }
   ],
   "source": [
    "def func(i):\n",
    "    while i < 5e+7: i += 1\n",
    "\n",
    "ts = time.time()\n",
    "func(0)\n",
    "print(f'Elapsed time: {round(time.time() - ts, 2)}')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Многопоточный код:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Elapsed time: 5.27\n"
     ]
    }
   ],
   "source": [
    "ts = time.time()\n",
    "threads = []\n",
    "for _ in range(2):\n",
    "    thread = threading.Thread(target=func, args=(0, ))\n",
    "    thread.start()\n",
    "    threads.append(thread)\n",
    "\n",
    "for thread in threads:\n",
    "    thread.join()\n",
    "\n",
    "print(f'Elapsed time: {round(time.time() - ts, 2)}')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "### `Когда многопоточный код полезен`\n",
    "\n",
    "- Независимые и не интенсивные I/O-операции с файлами, запросами или БД, логгирование<br><br>\n",
    "    - время ожидание завершения работы системных вызовов, связанных с I/O-операциями относительно велико\n",
    "    - поток, ожидающий этого, не меняет ничего в состоянии интерпретатора до выполнения следующего фрагмента кода\n",
    "    - значит на это время он может освободить GIL и дать другому потоку продолжить выполнение<br><br>\n",
    "\n",
    "- Отрисовка GUI\n",
    "\n",
    "Пример с логгированием:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "I print this while func works\n",
      "I print this while func works\n",
      "I print this while func works\n",
      "Finished\n"
     ]
    }
   ],
   "source": [
    "flag = True\n",
    "def print_message():\n",
    "    while flag:\n",
    "        time.sleep(1)\n",
    "        print('I print this while func works')\n",
    "    \n",
    "log_thread = threading.Thread(target=print_message)\n",
    "log_thread.start()\n",
    "func(-5e+7)\n",
    "print('Finished')\n",
    "flag = False"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "### `Будущее GIL`\n",
    "\n",
    "- Проблема с GIL действительно есть, однако её важность для большинства пользователей преувеличена\n",
    "- Попытки избавиться от GIL предпринимались многократно, но часто проваливались из-за требования неухудшения производительности для однопоточных приложений\n",
    "- Но есть и успешные перспективные примеры (https://github.com/colesbury/nogil)\n",
    "- Одна из возможных альтернатив - механизм подинтерпретаторов (subinterpreters, версия Python 3.9+)\n",
    "- Пробуются разные идеи, основанные на изменении методов подсчёта ссылок, аллокации памяти и принципов работы виртуальной машины"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "### `Многопроцессорный параллелизм`\n",
    "\n",
    "- Процессы работают в собственных адресных пространствах и имеют каждый свои данные, в т.ч. и GIL в случае Python\n",
    "- Взаимодействие между ними можно организовать также, как и для любых других процессов: через файлы (в т.ч. вирутуальные) или обмен сообщениями (multiprocessing, MPI)<br><br>\n",
    "\n",
    "- У каждого процесса есть `pid`, который можно получить вызовом `os.getpid()`\n",
    "- Воспользуемся для примера модулем multiprocessing, все похоже на потоки:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "func was called by process id 110787 with number 5\n",
      "func was called by process id 110790 with number 10\n",
      "func was called by process id 110795 with number 15\n",
      "func was called by process id 110800 with number 20\n",
      "func was called by process id 110805 with number 25\n"
     ]
    }
   ],
   "source": [
    "import os\n",
    "from multiprocessing import Process\n",
    "\n",
    "def func(number):\n",
    "    result = number ** 2\n",
    "    print(f'func was called by process id {os.getpid()} with number {number}')\n",
    "\n",
    "procs, numbers = [], [5, 10, 15, 20, 25]\n",
    "    \n",
    "for index, number in enumerate(numbers):\n",
    "    proc = Process(target=func, args=(number,))\n",
    "    procs.append(proc)\n",
    "    proc.start()\n",
    "    \n",
    "for proc in procs:\n",
    "    proc.join()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "### `Блокировки для процессов`\n",
    "\n",
    "- `multiprocessing.Lock` реализован как объект семафора в ОС вне памяти процесса, и каждый процесс может получить к нему доступ и изменять его"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "0\n",
      "\n",
      "12\n",
      "3\n",
      "4\n",
      "5\n",
      "6\n",
      "7\n",
      "8\n",
      "9\n"
     ]
    }
   ],
   "source": [
    "from multiprocessing import Process, Lock\n",
    "\n",
    "def func(lock, num):\n",
    "    with lock:\n",
    "        print(num)\n",
    "\n",
    "lock = Lock()\n",
    "for num in range(10):\n",
    "    Process(target=func, args=(lock, num)).start()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "### `Совместный доступ к объектам`\n",
    "\n",
    "- Передавать параметры по ссылке бессмысленно - они копируются и сериализуются для передачи в подпроцесс\n",
    "- Тем не менее, можно пользоваться блоками разделяемой памяти в ОЗУ - `multiprocessing.shared_memory`\n",
    "- Это относительно низкоуровневый подход, на его основе реализованы более удобные `Value` и `Array`:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "10.0\n",
      "[0, -1, -2, -3, -4, -5, -6, -7, -8, -9]\n",
      "I print this while func works\n"
     ]
    }
   ],
   "source": [
    "from multiprocessing import Value, Array\n",
    "\n",
    "def func(number, array):\n",
    "    number.value = 10\n",
    "    for i in range(len(array)):\n",
    "        array[i] = -array[i]\n",
    "\n",
    "number = Value('d', 0.0)\n",
    "array = Array('i', range(10))\n",
    "\n",
    "p = Process(target=func, args=(number, array))\n",
    "p.start()\n",
    "p.join()\n",
    "\n",
    "print(number.value)\n",
    "print(array[:])"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "### `Pypy Python`\n",
    "\n",
    "- Реализация интерпретатора Python, написанная на фреймворке RPython (синтаксис аналогичен Python)<br><br>\n",
    "- Использует just-in-time (JIT) компиляцию:\n",
    "    - при запуске код анализируется, наиболее часто используемые фрагменты (как циклы) переводятся в машинный код\n",
    "    - в таком виде они оптимизируются\n",
    "    - при выполнении исходные фрагменты заменяются на оптимизированные<br><br>\n",
    "- В ряде случаев JIT позволяет в разы ускорить выполнение кода на Python, как правило, код должен выполняться длительное время<br><br>\n",
    "- Цена использования JIT - более высокое потребление памяти в сложных и долго работающих процессах<br><br>\n",
    "- PyPy поддерживает все конструкции соответствующей версии языка Python, но работает только с частью библиотек и C-расширений<br><br>\n",
    "- JIT - не компиляция, т.е. получить на выходе исполняемый бинарный файл не получится"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "### `Библиотека Numba`\n",
    "\n",
    "- Инструмент для компиляции части кода в LLVM, переводит часть кода на Python и NumPy в машинный код<br><br>\n",
    "- Далее эти фрагменты кода выполняются на CPU (или GPU) минуя интерпретатор<br><br>\n",
    "- Охватывает не весь язык, но может помочь в каких-то высоконагруженных секциях кода, скорость получается сопоставимой с реализацией на C/C++<br><br>\n",
    "- Numba использует JIT для ускорения отдельных функций в коде<br><br>\n",
    "- Библитека представлена набором декораторов и позволяет\n",
    "    - оставаться в рамках синтаксиса Python\n",
    "    - работать с его стандартным интерпретатором CPython<br><br>\n",
    "- Все, что нужно - поставить библиотеку и правильно расставить декораторы"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "### `Поддерживаемые возможности Python и NumPy`\n",
    "\n",
    "- Ссылки на актуальные версии для [Python](https://numba.pydata.org/numba-doc/dev/reference/pysupported.html) и [NumPy](https://numba.pydata.org/numba-doc/dev/reference/numpysupported.html)\n",
    "\n",
    "Основные поддерживаемые или частично поддерживаемые Numba возможности Python:\n",
    "\n",
    "- условные операторы\n",
    "- операторы циклов\n",
    "- генераторы\n",
    "- исключения\n",
    "- менеджеры контекстов\n",
    "- функции (\\*\\*kwargs не поддерживается)\n",
    "- передача функций в качестве аргументов\n",
    "\n",
    "Большинство стандартных типов поддерживается, строки тоже, хотя часть их методов реализуется не очень эффективно"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "### `Использование Numba`\n",
    "\n",
    "__Декораторы функций для их ускорения__:\n",
    "- `@jit` - Numba ускоряет все, что поддерживается до первой неподдерживаемой операции, после весь код будет выполняться интерпретатором (даже поддерживаемый)\n",
    "- `@njit`(или `@jit(nopython=True)`) - Numba будет сообщать об ошибке при работе с неподдерживаемым кодом (по возможности его стоит переписать)<br><br>\n",
    "- Из ускоренных функций допускается вызов только ускоренных функций\n",
    "- Ускоренные функции можно вызывать из любых функций"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {
    "slideshow": {
     "slide_type": "skip"
    }
   },
   "outputs": [],
   "source": [
    "def timed(method):\n",
    "    import time\n",
    "    def __timed(*args, **kw):\n",
    "        time_start = time.time()\n",
    "        result = method(*args, **kw)\n",
    "        time_end = time.time()\n",
    "        \n",
    "        print('{}  {:.3f} ms'.format(method.__name__, (time_end - time_start) * 1000))\n",
    "        return result\n",
    "\n",
    "    return __timed"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "### `Пример использования Numba`"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {},
   "outputs": [],
   "source": [
    "import numba\n",
    "import numpy as np\n",
    "from numba import jit\n",
    "\n",
    "@timed\n",
    "def dot_product(x, y):\n",
    "    np.dot(x, y)\n",
    "\n",
    "@timed\n",
    "@jit(cache=True, nopython=True)\n",
    "def dot_product_numba(x, y):\n",
    "    np.dot(x, y)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Запуск до и после кэширования кода:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "dot_product  37.443 ms\n",
      "dot_product_numba  821.502 ms\n",
      "dot_product_numba  9.248 ms\n"
     ]
    }
   ],
   "source": [
    "N = 1000\n",
    "x, y = np.random.random((N, N)), np.random.random((N, N))\n",
    "\n",
    "dot_product(x, y)\n",
    "dot_product_numba(x, y)\n",
    "dot_product_numba(x, y)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Оптимизация не всегда оказывается действенной, в каждом случае нужно проверять"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "### `Параллелельные вычисления в Numba`\n",
    "\n",
    "- Параметр `parallel=True` декоратора приводит поиску в коде функции участки кода, которые можно выполнить параллельно, и запускает их на нескольких ядрах ([ссылка на документацию](https://numba.pydata.org/numba-doc/dev/user/parallel.html))\n",
    "- Проще всего рассматривать на примере цикла"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Обычный код с циклом:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "two_d_array_reduction_prod  4066.887 ms\n",
      "two_d_array_reduction_prod  4101.387 ms\n"
     ]
    }
   ],
   "source": [
    "@timed\n",
    "def two_d_array_reduction_prod(n):\n",
    "    result = 2 * np.ones((13, 17), np.int_)\n",
    "    tmp = 2 * np.ones_like(result)\n",
    "\n",
    "    for i in range(n):\n",
    "        result *= tmp\n",
    "\n",
    "two_d_array_reduction_prod(10000000)\n",
    "two_d_array_reduction_prod(10000000)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "### `Параллелельные вычисления в Numba`\n",
    "\n",
    "Обычный код с numba и код с numba и параллелизацией цикла (`prange`):"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "two_d_array_reduction_prod_numba  2092.364 ms\n",
      "two_d_array_reduction_prod_numba  1530.556 ms\n"
     ]
    }
   ],
   "source": [
    "@timed\n",
    "@jit(cache=True, nopython=True)\n",
    "def two_d_array_reduction_prod_numba(n):\n",
    "    result = 2 * np.ones((13, 17), np.int_)\n",
    "    tmp = 2 * np.ones_like(result)\n",
    "\n",
    "    for i in range(n):\n",
    "        result *= tmp\n",
    "\n",
    "two_d_array_reduction_prod_numba(10000000)\n",
    "two_d_array_reduction_prod_numba(10000000)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "two_d_array_reduction_prod_numba_parallel  2741.759 ms\n",
      "two_d_array_reduction_prod_numba_parallel  376.208 ms\n"
     ]
    }
   ],
   "source": [
    "@timed\n",
    "@jit(cache=True, nopython=True, parallel=True)\n",
    "def two_d_array_reduction_prod_numba_parallel(n):\n",
    "    result = 2 * np.ones((13, 17), np.int_)\n",
    "    tmp = 2 * np.ones_like(result)\n",
    "\n",
    "    for i in numba.prange(n):\n",
    "        result *= tmp\n",
    "\n",
    "two_d_array_reduction_prod_numba_parallel(10000000)\n",
    "two_d_array_reduction_prod_numba_parallel(10000000)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "### `Cython`\n",
    "\n",
    "- [Cython](https://cython.org/) - это расширение языка Python + статический компилятор<br><br>\n",
    "- Он сильно упрощает использование C-расширений в Python<br><br>\n",
    "- Библиотека позволяет:\n",
    "    - вызывать C/C++ код из любой точки кода на Python\n",
    "    - модифицировав синтаксис кода на Python транслировать его в код на C\n",
    "    - выполнять код параллельно<br><br>\n",
    "\n",
    "- Для использования в Jupyter-ноутбуке нужно выполнить `%load_ext cython`<br><br>\n",
    "- Для использования Cython в блоке кода достаточно добавить перед ним `%%cython`<br><br>"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 32,
   "metadata": {},
   "outputs": [],
   "source": [
    "# !pip install Cython"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 33,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "The Cython extension is already loaded. To reload it, use:\n",
      "  %reload_ext Cython\n"
     ]
    }
   ],
   "source": [
    "%load_ext Cython"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "### `Примеры объявления переменных`"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 34,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "1 <class 'int'>\n",
      "0.5 <class 'float'>\n",
      "63.4 <class 'float'>\n",
      "b'str' <class 'bytes'>\n",
      "[1, 'fdfs', 'i', 0.4] <class 'list'>\n"
     ]
    }
   ],
   "source": [
    "%%cython\n",
    "cdef int a = 1\n",
    "cdef float x = 0.5\n",
    "cdef double y = 63.4\n",
    "cdef char* s = 'str'\n",
    "cdef list names = [1, 'fdfs', 'i', 0.4]\n",
    "\n",
    "print(a, type(a))\n",
    "print(x, type(x))\n",
    "print(y, type(y))\n",
    "print(s, type(s))\n",
    "print(names, type(names))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "### `Cython в Jupyter Notebook`\n",
    "\n",
    "- Вызов ячейки в первый раз приведет к компиляции, выполнению кода и загрузке результат\n",
    "- Второй вызов без изменения кода [ничего не даст](https://stackoverflow.com/questions/60802385/why-jupyter-notebook-only-prints-the-cython-result-once) - Cython не хочет повторять уже сделанную работу\n",
    "- Повторное выполнение можно потребовать флагом `--force`:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 35,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "0\n"
     ]
    }
   ],
   "source": [
    "%%cython --force\n",
    "cdef int a = 0\n",
    "print(a)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "- Правильный способ - разделить компиляцию и вызов, вынеся код в функцию:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 36,
   "metadata": {},
   "outputs": [],
   "source": [
    "%%cython\n",
    "def cfunc():\n",
    "    cdef int a = 0\n",
    "    print(a)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 37,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "0\n"
     ]
    }
   ],
   "source": [
    "cfunc()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "### `Пример использования Cython`"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 38,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "1000000000\n",
      "func  44372.208 ms\n"
     ]
    }
   ],
   "source": [
    "@timed\n",
    "def func(n):\n",
    "    res = 0\n",
    "    for _ in range(n):\n",
    "        res += 1\n",
    "    print(res)\n",
    "\n",
    "func(1000000000)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 39,
   "metadata": {},
   "outputs": [],
   "source": [
    "%%cython\n",
    "def func_cython(n):\n",
    "    cdef long res = 0\n",
    "    for _ in range(n):\n",
    "        res += 1\n",
    "    print(res)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 40,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "1000000000\n",
      "wrapper  12280.263 ms\n"
     ]
    }
   ],
   "source": [
    "@timed\n",
    "def wrapper(n):\n",
    "    func_cython(n)\n",
    "\n",
    "wrapper(1000000000)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "### `Работа с функциями`\n",
    "\n",
    "- В коде на Cython можно объявлять функции с помощью оператора `def`, которые можно вызывать из Python\n",
    "- Также есть возможность объявления с помощью `cdef`, такие функции допустимо использовать только в коде CPython\n",
    "- Наконец, можно использовать `cpdef` - такие функции могут вызываться отовсюду, в т.ч. из кода на C"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 41,
   "metadata": {},
   "outputs": [],
   "source": [
    "%%cython\n",
    "\n",
    "cdef double cdef_func(double x, int y):\n",
    "    return x + y\n",
    "\n",
    "cpdef double cpdef_func(double x, int y):\n",
    "    return x + y\n",
    "\n",
    "def def_func(x, y):\n",
    "    return cdef_func(x, y)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 42,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "13.0\n",
      "13.0\n"
     ]
    }
   ],
   "source": [
    "print(def_func(10, 3))\n",
    "print(cpdef_func(10, 3))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "### `Код из C в Cython`"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 43,
   "metadata": {},
   "outputs": [],
   "source": [
    "%%cython\n",
    "\n",
    "cdef extern from \"math.h\":\n",
    "    cpdef double sin(double x)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 44,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "-0.5440211108893698"
      ]
     },
     "execution_count": 44,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "sin(10)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 45,
   "metadata": {},
   "outputs": [],
   "source": [
    "%%cython\n",
    "from libc.stdlib cimport atoi\n",
    "\n",
    "cpdef parse_charptr_to_py_int(char* s):\n",
    "    return atoi(s)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 46,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "10"
      ]
     },
     "execution_count": 46,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "parse_charptr_to_py_int(b'10')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "### `Фреймворки Hadoop и Spark`\n",
    "\n",
    "- Hadoop - это фреймворк для хранения, обработки и анализа больших массивов данных<br><br>\n",
    "- Может быть развернут и на ноутбуке, но по факту рассчитан на кластеры из большого числа машин<br><br>\n",
    "- В основе Hadoop лежит HDFS - распределенная файловая система с репликацией данных<br><br>\n",
    "- Допускается использование различных движков для обработки данных (MapReduce, Spark)<br><br>\n",
    "    - MapReduce как движок используется для работы с большими статическими данными на диске\n",
    "    - Spark работает с данными в оперативной памяти (MapReduce как парадигма вычислений тоже может использоваться)<br><br>\n",
    "- В экосистеме Hadoop есть SQL (Hive) и NoSQL (HBase) базы данных, системы обмена сообщениями (Kafka)<br><br>\n",
    "- Часто Spark и Hadoop используются вместе, но можно использовать и по-отдельности: Spark работает с другими системами управления (Mesos) и распределенными ФС (Cassandra, Amazon S3)<br><br>\n",
    "- С Hadoop и Spark можно работать из Python"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "### `Парадигма MapReduce`\n",
    "\n",
    "- Программа в парадигме MapReduce состоит из наборов процедур Map и Reduce и применяется к массиву данных<br><br>\n",
    "- Шаг Map вычисляется для каждого элемента данных независимо от прочих и возвращается с определенным ключом<br><br>\n",
    "- Шаг Reduce применяется к выходным элементам с определенными ключами и агрегирует их в итоговый элемент или набор элементов\n",
    "\n",
    "Простой пример - поиск максимального элемента:\n",
    "\n",
    "```\n",
    "data = [1, 8, 7, 3, 2, 0]\n",
    "parts = [1, 8, 7], [3, 2, 0]\n",
    "map(parts) = (8, 'key'), (3, 'key')\n",
    "reduce(map(parts)) = 8\n",
    "```\n",
    "\n",
    "- Некоторая аналогия есть с функциями `map` и `functools.reduce`"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "### `Пример задачи на MapReduce`\n",
    "\n",
    "- Обычно задача на MapReduce сводится к описанию функций, выполняющих эти шаги, с определенными сигнатурами\n",
    "- Функция Map должна принимать на вход значение или набор значений, а возвращать значение или набор значений с ключами\n",
    "- Функция Reduce должна принимать на вход набор значений с заданным ключом, агрегировать их и выдавать одно или несколько результирующих значений\n",
    "\n",
    "__Пример__ - функция, делающая join двух списков по ключу:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 47,
   "metadata": {},
   "outputs": [],
   "source": [
    "a = [('1', 'A'), ('2', 'B'), ('3', 'C'), ('4', 'D')]\n",
    "b = [('I', '1'), ('II', '2'), ('III', '3'), ('IV', '4')]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 48,
   "metadata": {},
   "outputs": [],
   "source": [
    "def mapper(x, key_index):\n",
    "    for e in x:\n",
    "        yield (e[key_index], e)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 49,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "('1', ('1', 'A'))\t('2', ('2', 'B'))\t('3', ('3', 'C'))\t('4', ('4', 'D'))\t\n",
      "('1', ('I', '1'))\t('2', ('II', '2'))\t('3', ('III', '3'))\t('4', ('IV', '4'))\t"
     ]
    }
   ],
   "source": [
    "for e in mapper(a, key_index=0):\n",
    "    print(e, end='\\t')\n",
    "\n",
    "print()\n",
    "for e in mapper(b, key_index=1):\n",
    "    print(e, end='\\t')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "### `Пример задачи на MapReduce`\n",
    "\n",
    "__Пример__ - функция, делающая join двух списков по ключу:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 50,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[('1', ('1', 'A')), ('1', ('I', '1'))]\n"
     ]
    }
   ],
   "source": [
    "# Map step\n",
    "elements = list(sorted([e for e in mapper(a, 0)] + [e for e in mapper(b, 1)]))\n",
    "\n",
    "# Intermediate steps\n",
    "outputs = [elements[i: i + 2] for i in range(0, len(elements), 2)]\n",
    "print(outputs[0])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 51,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[('1', 'A', 'I', '1'),\n",
       " ('2', 'B', 'II', '2'),\n",
       " ('3', 'C', 'III', '3'),\n",
       " ('4', 'D', 'IV', '4')]"
      ]
     },
     "execution_count": 51,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "def reducer(x):\n",
    "    return tuple([k for e in x for k in e[1]])\n",
    "\n",
    "# Reduce step\n",
    "[reducer(e) for e in outputs]"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "### `Какие еще есть шаги MapReduce`\n",
    "\n",
    "На самом деле, помимо основных шагов есть неявные, но тоже важные:\n",
    "- Sorting - сортировки выходов Map по ключам<br><br>\n",
    "- Partitioning - комбинирование выходов Map по наборам ключей для сокращения числа Reduce-ов (по одному не на ключ, а на надор ключей)<br><br>\n",
    "- Shuffling - направляет выходы Map к соответствующим Reduce-ам<br><br>\n",
    "- Combining - делает предварительный Reduce локальных выходов Map (на этом вычислительном узле)<br><br>\n",
    "- Merging - сливает все выходы Map, пришедшие к данному Reducer-у с сохранением порядка сортировки"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "### `Профилирование кода на Python`\n",
    "\n",
    "- Профилирование - важный этап анализа производительности кода, который должен работать быстро\n",
    "- Существует два основных подхода к соданию программ-профилировщиков:\n",
    "    - __deterministic__\n",
    "        - запускается совместно с отлаживаемой программой и даёт точную информацию о числе вызовов и их относительной длительности\n",
    "        - существенно замедляет выполнение кода\n",
    "        - хорошо подходит для первичной отладки\n",
    "        - пример инструмента для Python - cProfile (стандартная библиотека)\n",
    "    - __sampling__\n",
    "        - запускается отдельно, периодически сэмплируя стек вызовов кода, и пытается эвристически определить число и длительность вызовов\n",
    "        - результат получается менее точным, зато на абсолютные значения таймингов можно полагаться\n",
    "        - подходит для профилировки работающих процессов\n",
    "        - пример инструмента для Python - py-spy (внешний модуль)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "### `Подготовка примеров`"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 52,
   "metadata": {},
   "outputs": [],
   "source": [
    "def add(x, k):\n",
    "    if k % 2 == 0:\n",
    "        return x + k\n",
    "    return x + 2 * k\n",
    "\n",
    "def some_func(n):\n",
    "    n, m = 10, n\n",
    "    return m\n",
    "\n",
    "def func(n):\n",
    "    i = 0\n",
    "    while i < n:\n",
    "        i = add(i, 1)\n",
    "        if i % 2 == 0:\n",
    "            i = add(i, 2)\n",
    "        n = some_func(n)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 53,
   "metadata": {},
   "outputs": [],
   "source": [
    "import inspect\n",
    "with open('temp.py', 'w') as fout:\n",
    "    fout.write(''.join(inspect.findsource(func)[0]))\n",
    "    fout.write('func(100000000 * 10)')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "### `Пример использования cProfile`\n",
    "\n",
    "Основной вызов - `cProfile.run`, который принимает на всё, что можно подать в вызов `exec`:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 54,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "         75000004 function calls in 16.160 seconds\n",
      "\n",
      "   Ordered by: standard name\n",
      "\n",
      "   ncalls  tottime  percall  cumtime  percall filename:lineno(function)\n",
      " 50000000    4.904    0.000    4.904    0.000 2864233318.py:1(add)\n",
      "        1    9.810    9.810   16.160   16.160 2864233318.py:10(func)\n",
      " 25000000    1.446    0.000    1.446    0.000 2864233318.py:6(some_func)\n",
      "        1    0.000    0.000   16.160   16.160 <string>:1(<module>)\n",
      "        1    0.000    0.000   16.160   16.160 {built-in method builtins.exec}\n",
      "        1    0.000    0.000    0.000    0.000 {method 'disable' of '_lsprof.Profiler' objects}\n",
      "\n",
      "\n",
      "16.16090154647827 sec.\n"
     ]
    }
   ],
   "source": [
    "import cProfile, time\n",
    "\n",
    "ts = time.time()\n",
    "cProfile.run('func(100000000)')\n",
    "print(f'{time.time() - ts} sec.')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 55,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "8.401335954666138 sec.\n"
     ]
    }
   ],
   "source": [
    "ts = time.time()\n",
    "func(100000000)\n",
    "print(f'{time.time() - ts} sec.')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "### `Пример использования cProfile`\n",
    "\n",
    "Выход профилировщика можно отсортировать по ключам:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 56,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "         75000004 function calls in 16.405 seconds\n",
      "\n",
      "   Ordered by: internal time\n",
      "\n",
      "   ncalls  tottime  percall  cumtime  percall filename:lineno(function)\n",
      "        1    9.943    9.943   16.404   16.404 2864233318.py:10(func)\n",
      " 50000000    5.033    0.000    5.033    0.000 2864233318.py:1(add)\n",
      " 25000000    1.428    0.000    1.428    0.000 2864233318.py:6(some_func)\n",
      "        1    0.000    0.000   16.405   16.405 {built-in method builtins.exec}\n",
      "        1    0.000    0.000   16.404   16.404 <string>:1(<module>)\n",
      "        1    0.000    0.000    0.000    0.000 {method 'disable' of '_lsprof.Profiler' objects}\n",
      "\n",
      "\n"
     ]
    }
   ],
   "source": [
    "from pstats import SortKey\n",
    "cProfile.run('func(100000000)', sort=SortKey.TIME)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Начиная с версии 3.8 языка код можно профилировать прямо в менеджере контекста:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 57,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "         75000003 function calls in 15.778 seconds\n",
      "\n",
      "   Ordered by: standard name\n",
      "\n",
      "   ncalls  tottime  percall  cumtime  percall filename:lineno(function)\n",
      " 50000000    4.764    0.000    4.764    0.000 2864233318.py:1(add)\n",
      "        1    9.589    9.589   15.778   15.778 2864233318.py:10(func)\n",
      " 25000000    1.426    0.000    1.426    0.000 2864233318.py:6(some_func)\n",
      "        1    0.000    0.000    0.000    0.000 cProfile.py:117(__exit__)\n",
      "        1    0.000    0.000    0.000    0.000 {method 'disable' of '_lsprof.Profiler' objects}\n",
      "\n",
      "\n"
     ]
    }
   ],
   "source": [
    "with cProfile.Profile() as pf:\n",
    "    func(100000000)\n",
    "pf.print_stats()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "### `Пример использования py-spy`\n",
    "\n",
    "Сперва запустим скрипт (можно запустить в терминале):"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 63,
   "metadata": {},
   "outputs": [],
   "source": [
    "# !pip install py-spy"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 69,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "112333"
      ]
     },
     "execution_count": 69,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "import subprocess\n",
    "\n",
    "pid = subprocess.Popen(['python', 'temp.py'], shell=False).pid\n",
    "pid"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 67,
   "metadata": {},
   "outputs": [],
   "source": [
    "# sudo py-spy top --pid $pid"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Теперь запустим в терминале команду\n",
    "\n",
    "`sudo /home/mel-lain/.pyenv/shims/py-spy top --pid 20147`\n",
    "\n",
    "и получим интерактивный монитор (нагрузка в скрипте в файле в 10 раз выше):"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "```\n",
    "%Own   %Total  OwnTime  TotalTime  Function (filename:line)\n",
    "\n",
    " 17.00%  17.00%   15.84s    15.84s   add (temp.py:2)\n",
    " 24.00%  37.00%   14.18s    27.24s   func (temp.py:15)\n",
    " 15.00%  32.00%   13.83s    29.73s   func (temp.py:13)\n",
    " 13.00%  20.00%   10.15s    15.30s   func (temp.py:16)\n",
    "  7.00%   7.00%    7.38s     7.38s   func (temp.py:14)\n",
    "  8.00%   8.00%    7.24s     7.24s   add (temp.py:4)\n",
    "  5.00%   5.00%    4.12s     4.12s   add (temp.py:3)\n",
    "  2.00%   2.00%    3.60s     3.60s   some_func (temp.py:7)\n",
    "  4.00%   4.00%    3.35s     3.35s   func (temp.py:12)\n",
    "  0.00%   0.00%    1.76s     1.76s   add (temp.py:1)\n",
    "  1.00%   1.00%   0.810s    0.810s   some_func (temp.py:8)\n",
    "  4.00%   4.00%   0.740s    0.740s   some_func (temp.py:6)\n",
    "  0.00% 100.00%   0.000s    83.00s   <module> (temp.py:17)\n",
    "```"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "### `Пример использования py-spy`\n",
    "\n",
    "Результат можно записать и визуализировать:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 59,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "111152"
      ]
     },
     "execution_count": 59,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "subprocess.Popen(['python', 'temp.py'], shell=False).pid"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Запустим в терминале команду\n",
    "\n",
    "`sudo /home/mel-lain/.pyenv/shims/py-spy record -p 19409 --output result.svg`\n",
    "\n",
    "и отобразим полученную картинку:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 70,
   "metadata": {},
   "outputs": [],
   "source": [
    "# from IPython.display import SVG, display\n",
    "# display(SVG(filename='result.svg'))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "## `Спасибо за внимание!`"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "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.9.16"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}