1. Thread Basics.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "import time\n",
    "import random\n",
    "import threading\n",
    "from threading import Thread"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "In Python 3, `threading` is the module used to create and use threads. There's a low level module `_thread` but it's not recommended to use it directly. I'm mentioning it just as a warning, **don't use `_thread`!**.\n",
    "\n",
    "The most important class in the `threading` module is: `Thread` (doh!).\n",
    "\n",
    "Very simplified, this is how a thread is instantiated:\n",
    "\n",
    "```python\n",
    "class Thread:\n",
    "    def __init__(self, target, name=None, args=(), kwargs={}):\n",
    "        pass\n",
    "```\n",
    "(there's a `group` argument which should be always `None`, as it's reserved for future use)\n",
    "\n",
    "In this case, `target` is the function that will be executed in that particular thread.\n",
    "\n",
    "Once a thread has been _created_ (instantiated), we'll need to `start()` it in order for it to begin to process."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Basic example of a thread"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "def simple_worker():\n",
    "    print('hello', flush=True)\n",
    "    time.sleep(2)\n",
    "    print('world', flush=True)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [],
   "source": [
    "t1 = Thread(target=simple_worker)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "hello\n"
     ]
    }
   ],
   "source": [
    "t1.start()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [],
   "source": [
    "t1.join()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Running multiple threads in parallel"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [],
   "source": [
    "def simple_worker():\n",
    "    time.sleep(random.random() * 5)\n",
    "    value = random.randint(0, 99)\n",
    "    print(f'My value: {value}')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [],
   "source": [
    "threads = [Thread(target=simple_worker) for _ in range(5)]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[None, None, None, None, None]"
      ]
     },
     "execution_count": 12,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "[t.start() for t in threads]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "My value: 48\n",
      "My value: 94\n",
      "My value: 35\n",
      "My value: 16\n",
      "My value: 7\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "[None, None, None, None, None]"
      ]
     },
     "execution_count": 9,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "[t.join() for t in threads]"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Thread States\n",
    "\n",
    "A thread can be in multiple states, when a thread has just been created, its state is `\"ready\"`:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def simple_worker():\n",
    "\n",
    "    print('Thread running...')\n",
    "    time.sleep(5)\n",
    "    print('Thread finished...')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "t = Thread(target=simple_worker)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "t.is_alive()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "t.start()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "t.is_alive()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "A thread that has finished can't be started again, as shown in the following example:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "t.start()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**Important:** It's not possible(\\*) to manage thread states manually, for example, stopping a thread. A thread always has to run its natural cycle.\n",
    "\n",
    "(\\*) You might find hacks in the internet on how to stop threads, but **it's a bad practice**. We'll discuss more later."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Thread Identity\n",
    "\n",
    "The thread class has two attributes that lets us identify each thread. The human-ready `name`, which we can set when we construct the thread, and the machine-oriented `ident` one"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def simple_worker():\n",
    "    print('Thread running...')\n",
    "    time.sleep(5)\n",
    "    print('Thread exiting...')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "t = Thread(target=simple_worker)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "t.name"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "`ident` will be `None`until we run the thread."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "t.ident is None"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "t.start()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "t.name"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "t.ident"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We can create a thread and assign a custom name to it:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "t = Thread(target=simple_worker, name='PyCon 2020 Tutorial!')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "t.start()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "t.name"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "t.ident"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### A thread knows itself\n",
    "\n",
    "It's also possible to know the identity of the thread from within the thread itself. It might be counter intuitive as we don't have the reference to the created object, but the module function `threading.currentThread()` will provide access to it."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def simple_worker():\n",
    "    sleep_secs = random.randint(1, 5)\n",
    "    myself = threading.current_thread()\n",
    "    ident = threading.get_ident()\n",
    "    print(f\"I am thread {myself.name} (ID {ident}), and I'm sleeping for {sleep_secs}.\")\n",
    "    time.sleep(sleep_secs)\n",
    "    print(f'Thread {myself.name} exiting...')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "t1 = Thread(target=simple_worker, name='Bubbles')\n",
    "t2 = Thread(target=simple_worker, name='Blossom')\n",
    "t3 = Thread(target=simple_worker, name='Buttercup')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "t1.start()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "t2.start()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "t3.start()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "print('Waiting...')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Passing parameters to threads\n",
    "\n",
    "Passing parameters is simple with the thread constructor, just use the `args` argument:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def simple_worker(time_to_sleep):\n",
    "    myself = threading.current_thread()\n",
    "    ident = threading.get_ident()\n",
    "    print(f\"I am thread {myself.name} (ID {ident}), and I'm sleeping for {time_to_sleep}.\")\n",
    "    time.sleep(time_to_sleep)\n",
    "    print(f'Thread {myself.name} exiting...')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "t1 = Thread(target=simple_worker, name='Bubbles', args=(3, ))\n",
    "t2 = Thread(target=simple_worker, name='Blossom', args=(1.5, ))\n",
    "t3 = Thread(target=simple_worker, name='Buttercup', args=(2, ))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "t1.start()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "t2.start()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "t3.start()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Subclassing `Thread`\n",
    "\n",
    "So far, the way we've created threads is by passing a `target` function to be executed. There's an alternative, more OOP-way to do it, which is extending the Thread class:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "class MyThread(Thread):\n",
    "    def __init__(self, time_to_sleep, name=None):\n",
    "        super().__init__(name=name)\n",
    "        self.time_to_sleep = time_to_sleep\n",
    "        \n",
    "    def run(self):\n",
    "        ident = threading.get_ident()\n",
    "        print(f\"I am thread {self.name} (ID {ident}), and I'm sleeping for {self.time_to_sleep} secs.\")\n",
    "        time.sleep(self.time_to_sleep)\n",
    "        print(f'Thread {self.name} exiting...')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "t = MyThread(2)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "t.start()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Shared Data\n",
    "\n",
    "As we'll see, Threads can access shared data within the process they live in. Example:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "TIME_TO_SLEEP = 2"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def simple_worker():\n",
    "    myself = threading.current_thread()\n",
    "    print(f\"I am thread {myself.name}, and I'm sleeping for {TIME_TO_SLEEP}.\")\n",
    "    time.sleep(TIME_TO_SLEEP)\n",
    "    print(f'Thread {myself.name} exiting...')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "t1 = Thread(target=simple_worker, name='Bubbles')\n",
    "t2 = Thread(target=simple_worker, name='Blossom')\n",
    "t3 = Thread(target=simple_worker, name='Buttercup')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "t1.start()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "t2.start()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "t3.start()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "How is this possible?\n",
    "\n",
    "Remember, all threads live **within the same process**, and the variable `TIME_TO_SLEEP` is stored in the process. So all the threads created can access that variable."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<img src=\"img/thread_shared_data.png\" width=900 />"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## A real example\n",
    "\n",
    "In the `crypto-examples` directory, we've included a real example of a web server that contains prices of different cryptocurrencies. You can run it with `python flask_app.py --sleep [sleep in seconds]`. The server can be slowed down on purpose to simulate a real slow server.\n",
    "\n",
    "Let's check how to get one price as an example:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "BASE_URL = \"http://localhost:5000\""
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import requests"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "resp = requests.get(f\"{BASE_URL}/price/bitfinex/btc/2020-04-01\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "resp"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "resp.json()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now, let's suppose we want to get the price of Bitcoin from 3 different exchanges: `bitfinex`, `bitstamp` and `kraken`. The sequential requests would take us 6 seconds (with a sleep param in 2)."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "EXCHANGES = ['bitfinex', 'bitstamp', 'kraken']"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "start = time.time()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "for exchange in EXCHANGES:\n",
    "    resp = requests.get(f\"{BASE_URL}/price/{exchange}/btc/2020-04-01\")\n",
    "    print(f\"{exchange.title()}: ${resp.json()['close']}\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "time.time() - start"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Let's now move it to threads! For now, we'll just **print** the output, as we'll se data sharing in further lessons..."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def check_price(exchange, symbol, date, base_url=BASE_URL):\n",
    "    resp = requests.get(f\"{base_url}/price/{exchange}/{symbol}/{date}\")\n",
    "    print(f\"{exchange.title()}: ${resp.json()['close']}\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "check_price('bitfinex', 'btc', '2020-04-01')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "threads = [\n",
    "    Thread(target=check_price, args=(exchange, 'btc', '2020-04-01'))\n",
    "    for exchange in EXCHANGES\n",
    "]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "start = time.time()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "[t.start() for t in threads];"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "[t.join() for t in threads];"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "time.time() - start"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### How many threads can we start?\n",
    "\n",
    "Let's say we need to get prices for 10 exchanges, 3 symbols, for a total of 30 days. Those are a lot of requests:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "10 * 3 * 30"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Can we start 900 threads all at once? Sadly, we can't. Each threads consumes resources and too many threads are usually a problem."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "So, what can we do when we need to process too many concurrent jobs? We'll create workers and use a consumer-producer model. But first, we need to talk about shared data, race conditions and synchronization..."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Summary:\n",
    "\n",
    "* `threading` module ✅\n",
    "* `_thread`  module ❌\n",
    "\n",
    "A thread's life cycle is Instantiated > Started > Running > Finished."
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.8.0"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}