15_wątki_i_procesy.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "# Język Python\n",
    "## Wątki i procesy"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "https://docs.python.org/3/library/threading.html"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "fragment"
    }
   },
   "source": [
    "A programmer had a problem.\n",
    "\n",
    "\"I know! I will solve it with threads!\" - he thought to himself.\n",
    "\n",
    "has Now problems. two he"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "### Tworzenie i uruchamianie wątku"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import threading\n",
    "\n",
    "class Thread1(threading.Thread):\n",
    "    def run(self):\n",
    "        for i in range(int(1e6)):\n",
    "            if i % 1e4 == 0:\n",
    "                print(i)\n",
    "                \n",
    "t1 = Thread1()\n",
    "t2 = Thread1()\n",
    "\n",
    "t1.start()\n",
    "t2.start()\n",
    "print(\"Finished\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "outputs": [],
   "source": [
    "p = threading.Thread(target=print, args=('bob', 'john'), kwargs={'sep': '\\t'})\n",
    "p.start()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "A gdyby tak:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "@threaded\n",
    "def f():\n",
    "    pass\n",
    "\n",
    "f()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "### Synchronizacja wątków"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "fragment"
    }
   },
   "source": [
    "#### Lock"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import threading\n",
    "import time\n",
    "\n",
    "queue = list(range(10))\n",
    "lock = threading.Lock()\n",
    "\n",
    "class Consumer(threading.Thread):\n",
    "    def run(self):\n",
    "        running = True\n",
    "        while running:\n",
    "            lock.acquire()\n",
    "            if not queue:\n",
    "                running = False\n",
    "            else:\n",
    "                elem = queue.pop()\n",
    "            lock.release()\n",
    "            time.sleep(2) # do something with elem\n",
    "            lock.acquire()\n",
    "            print(self.name, elem)\n",
    "            lock.release()\n",
    "            \n",
    "t1 = Consumer()\n",
    "t2 = Consumer()\n",
    "t1.start()\n",
    "t2.start()\n",
    "t1.join()\n",
    "t2.join()\n",
    "print(\"Finished\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "outputs": [],
   "source": [
    "import threading\n",
    "import time\n",
    "\n",
    "queue = list(range(10))\n",
    "lock = threading.Lock()\n",
    "\n",
    "class Consumer(threading.Thread):\n",
    "    def run(self):\n",
    "        while True:\n",
    "            with lock:\n",
    "                if not queue:\n",
    "                    break\n",
    "                else:\n",
    "                    elem = queue.pop()\n",
    "            time.sleep(3) # do something with elem\n",
    "            with lock:\n",
    "                print(self.name, elem)\n",
    "            \n",
    "t1 = Consumer()\n",
    "t2 = Consumer()\n",
    "t1.start()\n",
    "t2.start()\n",
    "t1.join()\n",
    "t2.join()\n",
    "print(\"Finished\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "`Lock`:\n",
    "* może zostać zwolniony przez **dowolny wątek**\n",
    "* próba ponownego zajęcia przez ten sam wątek blokuje go *ad infinitum*\n",
    "* nie można zwolnić, jeżeli nie został zajęty\n",
    "* można spróbować zająć w trybie nieblokującym\n",
    "* nie jest automatycznie zwalniany\n",
    "\n",
    "Polecam `RLock`:\n",
    "* może zostać zwolniony tylko przez wątek, który go posiada\n",
    "* można go zajmować wielokrotnie, ale należy zwolnić tyle samo razy\n",
    "* jest zwalniany automatycznie po zakończeniu wątku"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "A gdyby tak:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "@synchronized\n",
    "def f():\n",
    "    pass"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "outputs": [],
   "source": [
    "type(threading.Lock())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "outputs": [],
   "source": [
    "import threading\n",
    "import time\n",
    "\n",
    "lock = threading.RLock()\n",
    "\n",
    "class Thread1(threading.Thread):\n",
    "    def run(self, *args):\n",
    "        print(lock.acquire(False))\n",
    "#         print(lock.acquire(timeout=3))  # alt\n",
    "        time.sleep(1)\n",
    "        lock.release()\n",
    "        \n",
    "                \n",
    "Thread1().start()\n",
    "# time.sleep(2)\n",
    "Thread1().start()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "#### Condition"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import threading\n",
    "import time\n",
    "\n",
    "cv = threading.Condition()\n",
    "\n",
    "l = []\n",
    "\n",
    "class Consumer(threading.Thread):\n",
    "    def run(self):\n",
    "        '''Consume one item'''\n",
    "        with cv:\n",
    "            while True:\n",
    "                while not l:\n",
    "                    cv.wait()\n",
    "                print(l.pop(0))\n",
    "\n",
    "class Producer(threading.Thread):\n",
    "    def run(self):\n",
    "        '''Produce one item'''\n",
    "        global l\n",
    "        for i in range(10):\n",
    "            with cv:\n",
    "                l += [i]\n",
    "                cv.notify()\n",
    "            time.sleep(1)\n",
    "\n",
    "Consumer().start()\n",
    "time.sleep(2)\n",
    "Producer().start()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "outputs": [],
   "source": [
    "print(l)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "#### Inne"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "* `Semaphore`\n",
    "* `BoundedSemaphore`\n",
    "* `Event`\n",
    "* `Timer`\n",
    "* `Barrier`"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "### Wydajność"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "slideshow": {
     "slide_type": "-"
    }
   },
   "outputs": [],
   "source": [
    "import threading\n",
    "import time\n",
    "\n",
    "class Thread1(threading.Thread):\n",
    "    def run(self):\n",
    "        for i in range(int(1e6)):\n",
    "            a = i**2\n",
    "            b = i**3\n",
    "            c = i**4\n",
    "                \n",
    "t1 = Thread1()\n",
    "t2 = Thread1()\n",
    "\n",
    "start = time.time()\n",
    "\n",
    "t1.start()\n",
    "t2.start()\n",
    "print(\"Finished\")\n",
    "\n",
    "t1.join()\n",
    "t2.join()\n",
    "\n",
    "finish = time.time()\n",
    "print(finish - start)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "outputs": [],
   "source": [
    "import threading\n",
    "import time\n",
    "\n",
    "class Thread1(threading.Thread):\n",
    "    def run(self):\n",
    "        for i in range(int(1e6)):\n",
    "            a = i**2\n",
    "            b = i**3\n",
    "            c = i**4\n",
    "                \n",
    "t1 = Thread1()\n",
    "t2 = Thread1()\n",
    "\n",
    "start = time.time()\n",
    "\n",
    "t1.run()\n",
    "t2.run()\n",
    "print(\"Finished\")\n",
    "\n",
    "finish = time.time()\n",
    "print(finish - start)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "## Procesy"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "https://docs.python.org/3/library/multiprocessing.html"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "### Wydajność"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import multiprocessing\n",
    "import time\n",
    "\n",
    "class Thread1(multiprocessing.Process):\n",
    "    def run(self):\n",
    "        for i in range(int(1e6)):\n",
    "            a = i**2\n",
    "            b = i**3\n",
    "            c = i**4\n",
    "                \n",
    "t1 = Thread1()\n",
    "t2 = Thread1()\n",
    "\n",
    "start = time.time()\n",
    "\n",
    "t1.run()\n",
    "t2.run()\n",
    "print(\"Finished\")\n",
    "\n",
    "finish = time.time()\n",
    "print(finish - start)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "outputs": [],
   "source": [
    "import multiprocessing\n",
    "import time\n",
    "\n",
    "class Thread1(multiprocessing.Process):\n",
    "    def run(self):\n",
    "        for i in range(int(1e6)):\n",
    "            a = i**2\n",
    "            b = i**3\n",
    "            c = i**4\n",
    "                \n",
    "t1 = Thread1()\n",
    "t2 = Thread1()\n",
    "\n",
    "start = time.time()\n",
    "\n",
    "t1.start()\n",
    "t2.start()\n",
    "print(\"Finished\")\n",
    "\n",
    "t1.join()\n",
    "t2.join()\n",
    "finish = time.time()\n",
    "print(finish - start)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "Procesy są cięższe i mniej wygodne w użyciu, ale omijają GIL."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "outputs": [],
   "source": [
    "import multiprocessing\n",
    "import time\n",
    "\n",
    "queue = list(range(10))\n",
    "lock = multiprocessing.Lock()\n",
    "\n",
    "class Consumer(multiprocessing.Process):\n",
    "    def run(self):\n",
    "        while True:\n",
    "            with lock:\n",
    "                if not queue:\n",
    "                    break\n",
    "                else:\n",
    "                    elem = queue.pop()\n",
    "            time.sleep(2) # do something with elem\n",
    "            with lock:\n",
    "                print(self.pid, elem)\n",
    "                        \n",
    "t1 = Consumer()\n",
    "t2 = Consumer()\n",
    "t1.start()\n",
    "t2.start()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "Przydatne klasy:\n",
    "- Lock\n",
    "- Queue\n",
    "- JoinableQueue\n",
    "- Pipe\n",
    "- Value\n",
    "- Array"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "## Wątki w telegraficznym skrócie\n",
    "\n",
    "- moduł `threading`\n",
    "- dziedziczymy z `Thread` i nadpisujemy `run(self)`\n",
    "- albo instancjonujemy `Thread` i przekazujemy funkcję do konstruktora\n",
    "- uruchamiamy metodą `start()`\n",
    "- Global Interpreter Lock zabija wydajność... chyba że C\n",
    "- do synchronizacji: `RLock`, `Condition`, `Semaphore`, `BoundedSemaphore`, `Event`, `Timer`, `Barrier`"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "## Procesy w telegraficznym skrócie\n",
    "\n",
    "- moduł `multiprocessing`\n",
    "- dziedziczymy z `Process` i nadpisujemy `run(self)`\n",
    "- albo instancjonujemy `Process` i przekazujemy funkcję do konstruktora\n",
    "- uruchamiamy metodą `start()`\n",
    "- nie dotyczy ich GIL... ale działają w osobnej przestrzeni, więc komunikacja musi być jawna\n",
    "- przydatne klasy: `Lock`, `Queue`, `JoinableQueue`, `Pipe`, `Value`, `Array`"
   ]
  }
 ],
 "metadata": {
  "celltoolbar": "Slideshow",
  "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.10.12"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}