threads.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Parallelism and concurrency in Python"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Table of Contents\n",
    "  - [Parallelism and concurrency in Python](#Parallelism-and-concurrency-in-Python)\n",
    "    - [References](#References)\n",
    "    - [Introduction](#Introduction)\n",
    "    - [Parallelism vs. concurrency](#Parallelism-vs.-concurrency)\n",
    "      - [Parallelism](#Parallelism)\n",
    "      - [Concurrency](#Concurrency)\n",
    "      - [Quiz: parallel or not](#Quiz:-parallel-or-not)\n",
    "    - [Parallelism in python: pre-emptive multitasking](#Parallelism-in-python:-pre-emptive-multitasking)\n",
    "      - [High-level interface: Process pools](#High-level-interface:-Process-pools)\n",
    "      - [Low-level interface: Process, run, join and deadlocks](#Low-level-interface:-Process,-run,-join-and-deadlocks)\n",
    "      - [Higher level interface: concurrent.futures](#Higher-level-interface:-concurrent.futures)\n",
    "      - [Inter-process communication and data dependencies](#Inter-process-communication-and-data-dependencies)\n",
    "      - [Synchronization using `lock`](#Synchronization-using-lock)\n",
    "      - [Threads, GIL and the illusion of concurrency](#Threads,-GIL-and-the-illusion-of-concurrency)\n",
    "        - [Threads vs processes](#Threads-vs-processes)\n",
    "        - [When to use threads](#When-to-use-threads)\n",
    "    - [Asynchronous programming and coroutines: cooperative multitasking](#Asynchronous-programming-and-coroutines:-cooperative-multitasking)\n",
    "    - [Exercises](#Exercises)\n",
    "      - [Exercise 1: Counting words in a file🌶️🌶️](#Exercise-1:-Counting-words-in-a-file🌶️🌶️)\n",
    "      - [Exercise 2: Find super secret server key🌶️🌶️🌶️](#Exercise-2:-Find-super-secret-server-key🌶️🌶️🌶️)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "\n",
    "\n",
    "## References\n",
    "- [How async/await is implemented in Python](https://tenthousandmeters.com/blog/python-behind-the-scenes-12-how-asyncawait-works-in-python/)\n",
    "- [Coroutines from the Python standard library](https://docs.python.org/3/library/asyncio-task.html#coroutine)\n",
    "- [What color is your function?](https://journal.stuffwithstuff.com/2015/02/01/what-color-is-your-function/)\n",
    "\n",
    "## Introduction\n",
    "There are many cases where we could execute multiple tasks in parallel or switch between tasks while we wait for some time-consuming task to be completed.\n",
    "You can just think of examples from daily life to see why: \n",
    "\n",
    "- While your soup cooks on the stove, you start washing your dishes\n",
    "- You check the news while drinking your morning coffee\n",
    "- You work on your python course exercises while attending a video call\n",
    "\n",
    "Indeed, the language of computing is so engrained in many of us that today we refer to these sort of behaviors as *multitasking*, an expression borrowed from computer science.\n",
    "\n",
    "Because in computer science we like to be precise, let us define these terms better.\n",
    "\n",
    "## Parallelism vs. concurrency\n",
    "\n",
    "### Parallelism\n",
    "When we have two or more tasks *running and progressing simultaneously*, we can talk about **parallelism**. \n",
    "Think for example of the situation of paying at the supermarket where there are multiple lines: more than one customer can pay their purchases at the same time.\n",
    "\n",
    "### Concurrency\n",
    "When two or more tasks run in overlapping time periods (but **not necessarily simultaneously**) instead of sequentially, we say that their execution is **concurrent**.\n",
    "This is the typical human multitasking, where we work on multiple tasks in a time period, but we must switch between them to be able to perform them correctly.\n",
    "For example, we sit in a meeting, listen passively while working on our python program and stop working on our code to answer a question directed to us.\n",
    "\n",
    "\n",
    "\n",
    "The image below can help you understanding the difference between concurrent and parallel work.\n",
    "<figure>\n",
    "  <img\n",
    "  src=\"../../images/concurrency_vs_parallelism.jpg\"\n",
    "  height=\"400px\"\n",
    "  alt=\"A schema of parallelism and concurrency\">\n",
    "  <figcaption>A simple time diagram illustrating the difference between parallelism and concurrency (<a href=\"https://openclassrooms.com/en/courses/5684021-scale-up-your-code-with-java-concurrency/5684028-identify-the-advantages-of-concurrency-and-parallelism\">Source</a>)</figcaption>\n",
    "</figure>\n",
    "\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "\n",
    "### Quiz: parallel or not\n",
    "\n",
    "<div class=\"alert alert-block alert-danger\">\n",
    "    <h4><b>Heads-up</b></h4>\n",
    "    Please, don't forget to evaluate the cell below ⬇️\n",
    "</div>\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "For each of the examples below, decide whether the situation represents **parallel** actions or not."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from tutorial.threads import Threads\n",
    "\n",
    "Threads()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Parallelism in python: pre-emptive multitasking\n",
    "\n",
    "By default, in Python tasks do not run in parallel.\n",
    "Consider this example:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from datetime import datetime as dt\n",
    "from time import sleep\n",
    "\n",
    "def task(name: str):\n",
    "    \"\"\"\n",
    "    This function defines a fictional task that takes one second\n",
    "    to complete and prints when it started and finished.\n",
    "    \"\"\"\n",
    "    print(f\"{name} started at {dt.now()}\")\n",
    "    sleep(1)\n",
    "    print(f\"{name} finished at {dt.now()}\")\n",
    "\n",
    "\n",
    "\n",
    "def two_tasks():\n",
    "    task(\"First task\")\n",
    "    task(\"Second task\")\n",
    "\n",
    "\n",
    "two_tasks()\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<div class=\"alert alert-block alert-warning\">\n",
    "    <h4><b>Warning</b></h4> In a standard Python distribution we would use the <a href=https://docs.python.org/3/library/multiprocessing.html>multiprocessing</a> module part of the standard library. \n",
    "    To be able to run the examples in Ipython notebooks, we used the <b>multiprocess</b> module which is a drop-in replacement for <b>multiprocessing</b>\n",
    "</div>"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We see that the first task to be started (`First task`) finished before the second one could start. \n",
    "This is the sequential computational model we are used to when we first learn programming. \n",
    "However, in Python we can introduce **parallelism** by using the [multiprocess](https://github.com/uqfoundation/multiprocess) module. \n",
    "\n",
    "Using this module, we can execute code in different operating system **processes**. \n",
    "A process is a representation of a task, with all the code, memory and resources (files, network connections, etc.) needed to run it. \n",
    "In most cases, processes are managed by the operating system (OS).\n",
    "The OS schedules what processes to run when. \n",
    "Moreover, it ensures  that no process runs forever and that they regularly yield computing resources to other processes.\n",
    "This approach is called **pre-emptive multitasking** and is the standard way of running multiple processes in modern desktop and server operating systems. \n",
    "\n",
    "When working on a multi-core or multi-CPU system, it is possible to leverage multitasking to run your computations in parallel. \n",
    "We will see how in the following sections.\n",
    "\n",
    "### High-level interface: Process pools\n",
    "\n",
    "Let's rewrite our example from before using `multiprocess.Pool`, which executes jobs on a pool of shared processes.  \n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "from datetime import datetime as dt\n",
    "from time import sleep\n",
    "from multiprocess import Pool\n",
    "\n",
    "def task(name: str):\n",
    "    \"\"\"\n",
    "    This function defines a fictional task that takes one second\n",
    "    to complete and prints when it started and finished.\n",
    "    \"\"\"\n",
    "    print(f\"{name} started at {dt.now()}\")\n",
    "    sleep(1)\n",
    "    print(f\"{name} finished at {dt.now()}\")\n",
    "\n",
    "\n",
    "\n",
    "def two_tasks():\n",
    "    with Pool(3) as p:\n",
    "        p.map(task, [\"First task\", \"Second task\"])\n",
    "\n",
    "\n",
    "for i in range(10):\n",
    "    two_tasks()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We use `Pool` as a [context manager](https://book.pythontips.com/en/latest/context_managers.html) and use the `map` method of the pool object to call the function `task` with a list of arguments. \n",
    "Internally, this will create and run a separate process for each value in the list.\n",
    "\n",
    "As you can see from the console output, the two tasks not only run simultaneously (**concurrently**) but also in parallel. \n",
    "This output highlights quite well one problem with concurrent computations: the order of completion is **non-deterministic**. \n",
    "We cannot know a priori which process will be started first and which process will complete first. \n",
    "If the order of the results is important, you need to make sure to send and return some sort of identifier with each job, so that you can reconstruct the right order.\n",
    "\n",
    "However, if we use `map`, it takes care of managing the order of tasks automatically:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "from datetime import datetime as dt\n",
    "from time import sleep\n",
    "from multiprocess import Pool\n",
    "\n",
    "def increment(number: int) -> int:\n",
    "    \"\"\"\n",
    "    This function increments the number by 1.\n",
    "    \"\"\"\n",
    "    name = \"Process \" + str(number)\n",
    "    print(f\"{name} started at {dt.now()}\")\n",
    "    result = number + 1\n",
    "    print(f\"{name} finished at {dt.now()}\")\n",
    "    return result\n",
    "\n",
    "\n",
    "\n",
    "def two_tasks():\n",
    "    with Pool(3) as p:\n",
    "        res = p.map(increment, range(10))\n",
    "    print(res)\n",
    "\n",
    "\n",
    "\n",
    "two_tasks()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Low-level interface: Process, run, join and deadlocks\n",
    "\n",
    "In some situations, we want more control over the execution of multiple processes. \n",
    "In that case, you can directly create processes using the [`Process`](https://docs.python.org/3/library/multiprocessing.html#multiprocessing.Process) object. \n",
    "This offers several methods, primarily:\n",
    "- `run()`: by default, it runs the callable object with the argument passed at the process creation time. \n",
    "    <div class=\"alert alert-block alert-warning\">\n",
    "        <h4><b>Warning</b></h4> The <code>run</code> method is <b>blocking</b> and will just execute the function in the current python process, blocking it until the execution finishes.\n",
    "    </div>\n",
    "- `start()`: it will start the computation defined by `run()` in a separate process.\n",
    "- `join()`: this methods blocks the python interpreter process until the task defined by the owning `Process` finishes. \n",
    "     A process cannot join itself because this would cause a **deadlock**. This is a situation where there is a cycling dependency between some waiting resources. \n",
    "    In real life, imagine the situation of two friends waiting for each other to call before going out. `join` is useful when we want to make sure a given process finishes its job before continuing.\n",
    "\n",
    "let's see an example on how to use this:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "from multiprocess import Process\n",
    "from time import sleep\n",
    "from datetime import datetime as dt\n",
    "\n",
    "def log(message: str):\n",
    "    \"\"\"\n",
    "    This function just prints a message.\n",
    "    \"\"\"\n",
    "    print(f\"{dt.now()}: {message}\")\n",
    "\n",
    "def friend(n: int, sleep_time: int = 1):\n",
    "    \"\"\"\n",
    "    This function prints a friendly message.\n",
    "    \"\"\"\n",
    "    log(f\"Hello from process {n}\")\n",
    "    sleep(sleep_time)\n",
    "    log(f\"After sleeping, process {n} is done\")\n",
    "\n",
    "def waiting_friend(n: int, wait_for_friend: bool = True, sleep_time: int = 1):\n",
    "    \"\"\"\n",
    "    This function waits for n friend processes to finish.\n",
    "    \"\"\"\n",
    "    f = [Process(target=friend, args=(i, sleep_time)) for i in range(n)]\n",
    "    for p in f:\n",
    "        p.start()\n",
    "        if wait_for_friend:\n",
    "            p.join()\n",
    "    log(\"Finished\")\n",
    "\n",
    "\n",
    "\n",
    "#Start without waiting for friend\n",
    "waiting_friend(3, True, 10)\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "You can see the effect of `join` in the output of `waiting_friend`. \n",
    "If we do not `join` on the child processes, they can finish in any order; we don't have any guarantee that they will finish at all. \n",
    "Indeed, you can see that `waiting_friends` prints \"finished\" before any of the child processes wakes up from sleep.  \n",
    "\n",
    "On the other hand, if we join on them in the for loop by setting the second argument of `waiting_friend` to `True`, the function `waiting_friend` will wait for each friend (1, 2, 3) to finish before starting the next process.\n",
    "\n",
    "### Higher level interface: concurrent.futures\n",
    "\n",
    "Because `join` **blocks** the main process until the child process finished, this is rarely the solution we need for scientific computing, where we want to split a large unit of work into smaller blocks and have multiple processes handle each of the blocks independently.\n",
    "For this reason, in most cases we advise starting your parallel processing adventure using higher-level solutions that take care of this low-level synchronization for you. \n",
    "A first good starting point is the [`concurrent.futures`](https://docs.python.org/3/library/concurrent.futures.html) module of the python standard library. \n",
    "This module wraps the methods of `multiprocess` (and `multithreading`) modules and offers more convenient ways to launch parallel computations.\n",
    "\n",
    "The way it works is roughly as follows:\n",
    "\n",
    "- It creates a *pool* of `n` worker processes\n",
    "- It sends the functions (and the data) to execute to the processes in batches of size `n`\n",
    "- It waits (joins) to the processes until they finish.\n",
    "- It repeats the steps above until it exhausts the data to process\n",
    "\n",
    "Let's see how this works in practice with an artificial example: we call a function `work` that receives a number `i`, then waits 0.1 seconds and then return the number `i`. \n",
    "We want to call this function `n` times and compute the sum of the results.\n",
    "\n",
    "\n",
    "To see the definition of `parallel_work` and `sequential_work` functions, please have a look [here](tutorial/threads.py).\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "from concurrent.futures import ProcessPoolExecutor\n",
    "from time import sleep\n",
    "from multiprocess import cpu_count\n",
    "from functools import partial\n",
    "from tutorial.threads import parallel_work, sequential_work\n",
    "\n",
    "#Here we run the parallel and sequential functions\n",
    "n = 10\n",
    "with ProcessPoolExecutor(max_workers=cpu_count()) as executor:\n",
    "    res = parallel_work(executor, n)\n",
    "    res1 = sequential_work(n)\n",
    "print(f\"The parallel sum is {res}, the regular is {res1}\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Success! We can see that the sum computed in parallel using multiple processes equals the sum computed *locally* in one python process. Now it is interesting to try and see how much speedup we gain from this trick. To do so, we use the `timeit` module part of the python standard library:"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<div class=\"alert alert-block alert-warning\">\n",
    "    <h4><b>Warning</b></h4> The cell below might take a very long time to run. Do not run it unless you are fine with waiting a couple of minutes.\n",
    "</div>"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "import timeit\n",
    "from multiprocess import cpu_count\n",
    "from concurrent.futures import ProcessPoolExecutor\n",
    "\n",
    "def time_for(fun: callable, args: list[any]):\n",
    "    return [{arg: timeit.repeat(lambda : fun(arg), number=1, repeat=3)[1]} for arg in args]\n",
    "\n",
    "\n",
    "sizes = range(1, 20, 1)\n",
    "\n",
    "with ProcessPoolExecutor(max_workers=cpu_count()) as executor:\n",
    "    res = time_for(lambda i : parallel_work(executor, i, ), sizes)\n",
    "    res1 = time_for(sequential_work, sizes)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "import matplotlib.pyplot as plt\n",
    "from matplotlib.pyplot import Axes\n",
    "\n",
    "def plot_times(ax: Axes, times: list[dict[str, float]], title: str):\n",
    "    return ax.scatter(sizes, [list(t.values())[0] for t in times], label=title)\n",
    "\n",
    "f, ax = plt.subplots(1, 1)\n",
    "plot_times(ax, res, \"Parallel\")\n",
    "plot_times(ax, res1, \"Sequential\")\n",
    "h1 = ax.set_xlabel(\"Size\")\n",
    "h2 = ax.set_ylabel(\"Time\")\n",
    "f.legend()\n",
    "f.show()\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "\n",
    "<figure>\n",
    "  <img\n",
    "  src=\"../../images/process_performance.png\"\n",
    "  height=\"400px\"\n",
    "  alt=\"Graph of performance.\">\n",
    "  <figcaption>Comparison of the runtime (in seconds) of the <i>parallel_work</i> and <i>sequential_work</i> functions as a function of the input size. Here we show until n=100, but the code above will only run 20 to same time.</figcaption>\n",
    "</figure>\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "As you can see from the graph, there is a clear difference in execution time between `sequential_work` and `parallel_work` as the number `n` increases. \n",
    "This is due to the fact that if we execute the function `work` sequentially for `n` inputs, we need to wait **at least** `n` times the duration of execution of `work`. \n",
    "On the other hand, if we process in parallel, we can cut down the time by maximally `n_proc` where `n_proc` is the number of CPUs our system offers. \n",
    "In reality, the speed up will be a bit smaller because of the overhead of starting different processes and the synchronization effort.\n",
    "\n",
    "\n",
    "<div class=\"alert alert-block alert-warning\">\n",
    "    <h4><b>Warning</b></h4> Multiprocessing/parallelism should be the last resort to improve the performance of your Python code. \n",
    "    You first should try improving your code by using the appropriate algorithms or by adopting numerical libraries like NumPy that offers vectorized operations. \n",
    "    If this fails, you could try just-in-time compilers like <a href=\"https://numba.pydata.org\">Numba</a>, which only requires you to decorate existing functions. Only if all these steps yield no improvement, should you attempt to use multiprocessing explicitly. \n",
    "    If not used carefully, it can even harm the performance of your code.\n",
    "</div>\n",
    "\n",
    "\n",
    "### Inter-process communication and data dependencies\n",
    "\n",
    "The pattern of using `ProcessPoolExecutor` works well for a variety of tasks where we can split the  problem into **independent** unit of works that can be performed at the same time; sometimes these problems are called *embarrassingly parallel*.  \n",
    "However, many problems in scientific computing contain **data dependencies** where computations depend on the result of other computations. In this case, we cannot simply split our input data across processes as we did before. In these situations, we have a few possible solutions:\n",
    "\n",
    "1. Split the problem across a dimension where there are no dependencies. \n",
    "1. Add communication between processes, so that individual processes can access the result computed by other processes.\n",
    "\n",
    "In the rest of this section, we will address the second solution, as python offers methods to communicate between processes. The first solution cannot be easily addressed in this course because it requires domain knowledge and we cannot provide a simple recipe that will work in all cases.\n",
    "\n",
    "We now consider an example of a problem that can be parallelized but where there are dependencies between the processes: the **producer-consumer** problem. \n",
    "Consider the following situation: we have two **producers**, each of which produces a sequence of numbers. \n",
    "Additionally, we have a **consumer**; this is a function that wants to get the sequence of numbers of each of the producers and compute the sum of **all the numbers** produced by all producers.\n",
    "Because a sum can be computed in any order, we can have both producers work at the same time and get all the numbers and sum then in any order. \n",
    "To exchange data between the producers and the consumers we use a `Queue`. \n",
    "This is a list-like data structure where we put elements into it from one side and we extract them from the other side; we call this *first in, first out* (FIFO). \n",
    "The `multiprocess` module offers an implementation of a queue that works with multiple processing, it is called Queue.\n",
    "\n",
    "Let's see how we do use a Queue with multiprocessing to solve the consumer-producer problem: "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "from multiprocess import Process, Queue, Event\n",
    "from dataclasses import dataclass\n",
    "import time\n",
    "\n",
    "@dataclass\n",
    "class QueueItem:\n",
    "    \"\"\"\n",
    "    This class represents an item in the queue.\n",
    "    It contains the item and the producer.\n",
    "    \"\"\"\n",
    "    item: int\n",
    "    producer: str\n",
    "\n",
    "@dataclass\n",
    "class Stop:\n",
    "    \"\"\"\n",
    "    This class represents the stop signal.\n",
    "    \"\"\"\n",
    "    producer: str\n",
    "\n",
    "# This is a type alias for the queue elements: either we put an item or a stop signal\n",
    "QueueElement = QueueItem | Stop\n",
    "\n",
    "def generate_items(start: int, max: int):\n",
    "    \"\"\"\n",
    "    This function generates the items to be consumed.\n",
    "    \"\"\"\n",
    "    for i in range(start, max):\n",
    "        yield i\n",
    "\n",
    "def producer(start: int, max: int, name: str, queue: Queue):\n",
    "    \"\"\"\n",
    "    This function produces the items to be consumed and puts them in the queue.\n",
    "    \"\"\"\n",
    "    for i in generate_items(start, max):\n",
    "        print(f\"Putting item in queue from {name}\")\n",
    "        queue.put(QueueItem(i, f\"Producer {name}\"))\n",
    "        #We sleep a bit to simulate a long task, so that\n",
    "        #other producers can put items in the queue\n",
    "    queue.put(Stop(name))\n",
    "\n",
    "def consumer(producers: list[str], queue: Queue):\n",
    "    \"\"\"\n",
    "    This function consumes the items in the queue. \n",
    "    We pass it a list of producers so that it knows when to stop.\n",
    "    \"\"\"\n",
    "    sum = 0\n",
    "    stopped = set()\n",
    "    while True:\n",
    "        if stopped == set(producers):\n",
    "            print(\"All producers have stopped\")\n",
    "            #If the producers have all stopped, we can stop the consumer\n",
    "            break\n",
    "        item = queue.get()\n",
    "        match item:\n",
    "            case Stop(producer):\n",
    "                print(\"We got the stop signal\")\n",
    "                print(f\"The final sum is {sum}\")\n",
    "                stopped.add(producer)\n",
    "                continue\n",
    "            case QueueItem(item, producer):\n",
    "                sum += item\n",
    "                print(f\"We got {item} from {producer}\")\n",
    "                print(f\"The current sum is {sum}\")\n",
    "\n",
    "\n",
    "\n",
    "def producer_consumer():\n",
    "    \"\"\"\n",
    "    This function starts the producers and the consumer.\n",
    "    \"\"\"\n",
    "    queue = Queue(maxsize=1)\n",
    "    producer1 = Process(target=producer, args=(1, 10, \"First producer\", queue))\n",
    "    producer2 = Process(target=producer, args=(10, 20, \"Second producer\", queue,))\n",
    "    #We pass the list of producers to the consumer so it knows when to stop\n",
    "    consumer1 = Process(target=consumer, args=([\"First producer\", \"Second producer\"] ,queue,))\n",
    "    producer1.start()\n",
    "    producer2.start()\n",
    "    consumer1.start()\n",
    "\n",
    "    producer1.join()\n",
    "    producer2.join()\n",
    "    consumer1.join()\n",
    "    \n",
    "\n",
    "producer_consumer()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Let's now understand how this code works:\n",
    "\n",
    "To begin, we define two different `dataclasses` to represent the objects we will put in the queue. \n",
    "The class `QueueItem` will represent a number produced by one producer, while `Stop` represents the case where a producer is finished with its work.\n",
    "The function `producer` takes the initial number of the sequence and the maximum, as well as `Queue` where its data should be put in.\n",
    "The function iterates over the `generate_items` generator, for each number it produces, it puts it on the queue.\n",
    "When no items are available anymore, it puts a `Stop` signal instead.\n",
    "The message send with `Stop` also contains the name of who puts the message; this is used later to know when to stop.\n",
    "Now we can analyze `consumer`; this function takes a queue as well as the names of the producers. \n",
    "The function runs an infinite loop and keeps receiving messages from the queue; when the message matches with a `QueueItem`, it extracts the number, prints a message and increments the sum.\n",
    "When the function receives  a `Stop`, it extracts the name of the sender of the stop message and adds it to the set of senders. \n",
    "When it receives the stop signal from all the senders, it exits the loop and prints the final sum to the screen.\n",
    "\n",
    "If we now look at the function `producer_consumer`, we see that we initialize a queue as well as two `Process` for the producers, to which we pass the initial value of the sequence and the queue we just created.\n",
    "We also create a new `consumer` and we pass the same queue to it, for it to be able to communicate with the producers.\n",
    "Finally, we start all the processes and call their `join` method in order to wait for them to finish.\n",
    "\n",
    "Notice that the producers alternate in sending messages; this is thank to the line `queue = Queue(maxsize=1)`. \n",
    "This forces the queue to have a maximum size of one element; that means that we have the following sequence:\n",
    "1. `Producer1` puts an element on the queue\n",
    "2. `Producer1` moves to the next iteration \n",
    "3. Because the queue is already filled, `put` blocks\n",
    "4. `consumer` gets the first element from `Producer1` from the queue\n",
    "5. Now one of the two producers can again put an element on the queue. We can't guarantee which producer, but there's a good chance any of them at random gets to put an element.\n",
    "6. The cycle repeats.\n",
    "\n",
    "### Synchronization using `lock`\n",
    "Notice that because of the operating system scheduling of processes, there's a certain *non-determinism* regarding the order where tasks run. \n",
    "This means that sometimes the `put-consume` loop will briefly favor one `Producer` over another. \n",
    "If we want both tasks to strictly alternate, we can synchronize them using a `Lock`. \n",
    "This is an object that offers two methods `acquire()` and `release()`. \n",
    "When we `acquire` a lock, any other function that accesses the **same** lock and tries to call `acquire()` will be blocked until the other function calls `release()`\n",
    "\n",
    "Let's see an example how to use this:\n",
    "\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "from multiprocess import Process, Queue, Event, Lock\n",
    "from multiprocess.synchronize import Lock as LockBase\n",
    "import time\n",
    "def process(name: str, lock: LockBase):\n",
    "    \"\"\"\n",
    "    This function prints a message and then waits for the lock.\n",
    "    \"\"\"\n",
    "    print(f\"{name} started\")\n",
    "    for i in range(4):\n",
    "        #This is a context manager, so it will release the lock when the block ends\n",
    "        with lock:\n",
    "            #Here `lock.acquire()` is called implicitly\n",
    "            print(f\"{name} got the lock\")\n",
    "            time.sleep(1)\n",
    "            #Here `lock.release()` is called implicitly\n",
    "        print(f\"{name} released the lock\")\n",
    "\n",
    "def lock_example(): \n",
    "    print(\"Starting\")\n",
    "    l = Lock()\n",
    "    p1 = Process(target=process, args=(\"First process\", l))\n",
    "    p2 = Process(target=process, args=(\"Second process\", l))\n",
    "    p1.start()\n",
    "    p2.start()\n",
    "    p1.join()\n",
    "    p2.join()\n",
    "\n",
    "lock_example()\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "As you see in the console output, the processes now work in the right order; the second process only can work when the first one releases the lock and the other way around.\n",
    "It is important to think about your problem well: do you really need locking in order to obtain your result? \n",
    "In the case before, it was not necessary: the sum does not depend on the order in which the producer worked."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Threads, GIL and the illusion of concurrency\n",
    "\n",
    "Thus far, we only discussed one approach to concurrency (and parallelism) in python:  the use of operating system **processes** leveraging the ability of the OS to schedule and coordinate multiple processes across multiple CPUs. \n",
    "\n",
    "#### Threads vs processes\n",
    "Other than multiprocessing, there's a second, very similar approach  called **threading**. \n",
    "Like a **process**, a **thread** is a representation of a task including all needed resources. \n",
    "In fact, a process usually consists of multiple threads all sharing a common block of memory. \n",
    "Because of shared memory, a thread is usually lighter in its resource usage than a process, meaning that we can start more threads.\n",
    "Moreover the shared memory means that multiple threads can communicate using shared variables. \n",
    "Even if this is possible, this style of concurrency comes with severe performance and safety pitfalls and it is discouraged in most cases. \n",
    "Considering all this, in most programming languages we try to achieve concurrency by starting multiple threads in a process.  \n",
    "\n",
    "However, due to the way the python interpreter is written (the infamous [global interpreter lock](https://wiki.python.org/moin/GlobalInterpreterLock)), we cannot get any performance benefits from running python code across multiple threads: only one thread can perform CPU operations at a time. \n",
    "\n",
    "#### When to use threads\n",
    "However, there are some cases where we could benefit from this style: when we have multiple threads sitting and waiting for input from the network or another process, but that don't perform any heavy computation.\n",
    "This is a case commonly encountered in user interface programming, where we don't want the main user interface to block waiting for the program to fetch data from the network.\n",
    "In that situation, we can run the GUI in a main thread and have a second thread being responsible for network interaction. \n",
    "Here we won't have any speedup but we will give the user the **illusion** of concurrency because the GUI won't freeze while the network thread fetches the data.\n",
    "\n",
    "\n",
    "Let's see this with an example:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from IPython.display import display\n",
    "import ipywidgets as widgets\n",
    "from threading import Thread\n",
    "from queue import Queue\n",
    "import time\n",
    "import sys\n",
    "from logging import getLogger, StreamHandler, Formatter, DEBUG, INFO\n",
    "import logging\n",
    "\n",
    "def make_logger():\n",
    "    logging.basicConfig()\n",
    "    logger = logging.getLogger('Button')\n",
    "    if not logger.handlers:\n",
    "        handler = StreamHandler(sys.stdout)\n",
    "        handler.setFormatter(Formatter('%(asctime)s - Thread: %(thread)d - %(message)s'))\n",
    "        logger.addHandler(handler)  \n",
    "    logger.setLevel(INFO)\n",
    "    return logger\n",
    "\n",
    "\n",
    "def make_button():\n",
    "    logger = make_logger()\n",
    "    button = widgets.Button(description=\"Click Me!\")\n",
    "    output = widgets.Output()\n",
    "\n",
    "    task_status = Queue()\n",
    "    task_status.put('not_started')\n",
    "\n",
    "    def background_task():\n",
    "        task_status.put('running')\n",
    "        logger.info(\"Background task started.\")\n",
    "        time.sleep(60)\n",
    "        logger.info(\"Background task finished.\")\n",
    "        task_status.put('finished')\n",
    "\n",
    "    def on_button_clicked(b):\n",
    "        status = task_status.get()\n",
    "        if status == 'not_started' or status == 'finished':\n",
    "            logger.info(\"Button clicked while task not started, a new task will be started.\")\n",
    "            Thread(target=background_task).start()\n",
    "        elif status == 'running':\n",
    "            task_status.put('running')\n",
    "            logger.info(\"Button clicked while task runs.\")\n",
    "\n",
    "    button.on_click(on_button_clicked)\n",
    "    return button, output\n",
    "\n",
    "button, output = make_button()\n",
    "display(button, output)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "As you can see in the output, pressing the button starts a task. \n",
    "However, while the task runs, you can still interact with the button. \n",
    "This is possible because the task started by the button runs on a separated thread. \n",
    "\n",
    "To communicate between the main thread and the second thread, we use a *Queue*. \n",
    "This is a list-like data structure where one side can put data in with the `put` function and the other side can get the data from it using `get`.\n",
    "`get` is a **blocking** operation: this function will block the execution until data is available in the queue. \n",
    "This is how we use it in this script:\n",
    "\n",
    "1. When the function `make_button` is first started, we create a `Queue` and put the initial status `not_started` on it.\n",
    "2. The function will return a button and output field. The other functions created in its **closure** can access the queue.\n",
    "3. We add the output field and the button returned by `make_button` to the current notebook cell using `display`\n",
    "4. The button is connected to the function `on_button_clicked`.\n",
    "5. When the button is clicked, `on_button_clicked` tries to get the current status of the background task from the queue. Initially the queue contains the status `not_started`, therefore a new background task is started using `Thread`\n",
    "6. The background task adds a `running` status to the queue and sleeps for 60 seconds. When it finishes sleeping, it adds `finished` to the queue.\n",
    "7. The next time the button is called, we try again to get the status of the task from the queue. If we click while the background task is running, we get `running` and the display shows the message \"Button clicked while task runs.\"\n",
    "\n",
    "In this way, we can have a background task running while the user can interact with the button in the main task."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Asynchronous programming and coroutines: cooperative multitasking"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "So far, the style of programming we saw relies on *threads* or *processes* to split and coordinate works between multiple tasks that might be executed simultaneously on multiple CPUs. \n",
    "Even with a single CPU, we obtain the **illusion** of parallel work thanks to the operating system switching work between multiple tasks and making sure they all get a more or less regular portion of the CPU time. \n",
    "This style is commonly known as (**preemptive multitasking**)[https://en.wikipedia.org/wiki/Preemption_(computing)] and is the most common form on multitasking in modern server and desktop operating systems.\n",
    "\n",
    "In contrast to this style, we can also employ **cooperative multitasking**. \n",
    "This is the style used in python **coroutines** and often known as **asynchronous programming** or **async/await** in other languages. \n",
    "In cooperative multitasking, the tasks are responsible themselves for *yielding* control and resources back to other tasks. \n",
    "This has a few benefits over operating system threads or processes:\n",
    "\n",
    "- They require much less resources than threads and processes, which means potentially a very large number of tasks can run concurrently. \n",
    "- The programmer is responsible for passing control between the tasks. Because the points of switching are explicit, it is easier to reason about the behavior of the program. \n",
    "- Because the tasks are handled by the programming language, it is much easier to cancel their execution.\n",
    "\n",
    "\n",
    "Because the operating system is not involved in the running of asynchronous tasks (or coroutines), the programming language is responsible for this task and needs to provide an **executor** or **event loop**. At the core, this is nothing but a function that handles the execution of the tasks in the right order, stopping them when they need to yield control, resuming them etc. \n",
    "\n",
    "After this abstract discussion, let's have a look at how cooperative multitasking is used in python. To do so, we revisit our very first example and rewrite it in the async style:\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "import asyncio\n",
    "from datetime import datetime as dt\n",
    "async def task(name: str):\n",
    "    \"\"\"\n",
    "    This function defines a fictional task that takes one second\n",
    "    to complete and prints when it started and finished.\n",
    "    \"\"\"\n",
    "    print(f\"{name} started at {dt.now()}\")\n",
    "    await asyncio.sleep(1)\n",
    "    print(f\"{name} finished at {dt.now()}\")\n",
    "\n",
    "async def main():\n",
    "    await asyncio.gather(task(\"one\"), task(\"two\"))    \n",
    "\n",
    "\n",
    "await main()\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Note that we prepend the `async` keyword before the function definition. \n",
    "This defines a **coroutine**, a function whose execution can be suspended and restarted at a later time. \n",
    "Be aware that calling a coroutine using the usual function invocation syntax merely creates a task to be scheduled; to run the function you need to **await** for its result using the `await` keyword. \n",
    "Whenever we `await` on another coroutine, we effectively pass (yield) the control to that particular coroutine. \n",
    "Once that task finishes, the control returns to the coroutine awaiting it.\n",
    "\n",
    "We see an example of this in the `task` function, where we `await` `asyncio.sleep(1)`. \n",
    "This means \"create a task that waits for a second and wait for it to complete\".\n",
    "\n",
    "\n",
    "Finally, you can see that we used the `asyncio.gather` function. This is an utility function from the [asyncio](https://docs.python.org/3/library/asyncio.html) module of the standard library that schedules a number of tasks to run. \n",
    "Under the scenes, it creates a new coroutine (technically a `Future`) that internally awaits for all the tasks to complete. That's the reason why in `main` we need to `await` it. \n",
    "\n",
    "<div class=\"alert alert-block alert-warning\">\n",
    "    <h4><b>Warning</b></h4> In the example above, we simply awaited <code>main</code> in the top level script. \n",
    "    This is possible because IPython code already runs in an event loop which is responsible for running asynchronous tasks. \n",
    "    In a normal python application, the entrypoint to asynchronous programming should be the function <code>asyncio.run()</code>:</br>\n",
    "    <code>asyncio.run(main())</code>\n",
    "</div>\n",
    "\n",
    "\n",
    "As the main module (asyncio) could suggest, the main place where this style of programming is beneficial is in I/O-limited programs, for example in web or database programming, where most of the application time is spent waiting for pages to load or files to open.\n",
    "Unlike multiprocessing, coroutine cannot run on more than one CPU at the same time, so there is no speedup for computation-intensive programs. \n",
    "\n",
    "Let's see an example where using asyncio could benefit the performance."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "import random\n",
    "from dataclasses import dataclass\n",
    "\n",
    "\n",
    "def random_sentence():\n",
    "    words = [\"apple\", \"banana\", \"cherry\", \"date\", \"elderberry\"]\n",
    "    sentence_length = random.randint(5, 10)  # Choose a random sentence length\n",
    "    sentence = \" \".join(random.choice(words) for _ in range(sentence_length))\n",
    "    return sentence.capitalize() + \".\"\n",
    "\n",
    "@dataclass\n",
    "class Response:\n",
    "    sentence: str\n",
    "    url: str\n",
    "    success: bool\n",
    "\n",
    "async def respond(url: str) -> Response:\n",
    "    await asyncio.sleep(random.random())\n",
    "    return {\"sentence\": random_sentence(), \"url\": url, \"success\": True}\n",
    "\n",
    "\n",
    "async def get_url(url: str):\n",
    "    print(f\"Getting response from server: {url}\")\n",
    "    res = await respond(url)\n",
    "    print(f\"Response from server {url}: {res}\")\n",
    "    await process_url(res)\n",
    "    return res\n",
    "\n",
    "async def process_url(response: Response) -> str:\n",
    "    print(f\"Processing {response}\")\n",
    "    return response\n",
    "\n",
    "async def main():\n",
    "    res = await asyncio.gather(*[get_url(f) for f in [\"url1\", \"url2\", \"url3\"]])\n",
    "    print(res)\n",
    "\n",
    "\n",
    "await main()\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "In this artificial example, we ask three fictional web servers with urls \"url1\", \"url2\" and \"url3\" to respond. \n",
    "Their response is simulated by `respond`, which waits a random amount of time before answering. \n",
    "Because of this delay, we don't know in advance which server will be faster in processing our request.\n",
    "When we receive a response, we process it using `process_url`. \n",
    "\n",
    "As you can see from the console output, we launch three coroutines in sequence, but the order of the responses is random and determined by the time each \"server\" takes to reply. \n",
    "The advantage of this approach is that we can process a server's response while we wait for the other servers to respond. \n",
    "If we did this sequentially, everything would be blocked until the server would respond. \n",
    "Instead, using couroutines we can already process the responses from the faster servers while we wait for the results from the slower server.\n",
    "\n",
    "<div class=\"alert alert-block alert-warning\">\n",
    "    <h4><b>Warning</b></h4>If you simply call a coroutine like a function without <code>await</code>, nothing will happen. The call will return a `Future` object that \n",
    "    can be awaited on, but the task will not run until you <b>explicitly</b> use <code>await</code> on it.\n",
    "</div>\n",
    "\n",
    "\n",
    "\n",
    "<div class=\"alert alert-block alert-warning\">\n",
    "    <h4><b>Warning</b></h4>Coroutines are <b>infectious</b>. Any function that calls a coroutine becomes a couroutine itself (unless we synchronously wait on the coroutine using <code>asyncio.run()</code>). This is sometimes called the <b>function coloring problem</b>. For a good explanation of this issue, see <a href=https://journal.stuffwithstuff.com/2015/02/01/what-color-is-your-function/>this article</a>\n",
    "</div>\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Exercises"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "\n",
    "\n",
    "### Exercise 1: Counting words in a file🌶️🌶️\n",
    "\n",
    "Write a **parallel** function `letter_statistics` that returns the statistics of letter counts in the large file `input_file`.\n",
    "This means that the function should return a  `dict[str, int]` containing the counts for each letter in sorted order.\n",
    "\n",
    "<div class=\"alert alert-block alert-info\">\n",
    "    <h4><b>Hints</b></h4>\n",
    "    <ul>\n",
    "        <li>\n",
    "            You can open the file <b>read-only</b> multiple times. \n",
    "        </li>\n",
    "        <li>\n",
    "           To facilitate your work, we pass the size of the file (in number of characters) using the <code>size</code> argument.\n",
    "        </li>\n",
    "        <li>\n",
    "            The input <code>n_processes</code> determines how many processes your solution should use.\n",
    "        </li>\n",
    "        <li>\n",
    "            Using <code>seek</code> you can specify a line offset from the start of the file. Using <code>read(size)</code> you can read <code>size</code> characters only. \n",
    "        </li>\n",
    "        <li>\n",
    "            Write your function in the cell below inside of the <code>solution_exercise1</code> function. The function receives a <code>Path</code> object <code>input_file</code> as an input and should return a single <code>dict[str, int]</code> dictionary.\n",
    "        </li>\n",
    "        <li>\n",
    "            Consider using the <code>collections.Counter</code> class to count the number of letters in a string.\n",
    "        </li>\n",
    "        <li>\n",
    "            In case the test fails with a <code>BrokenProcessPool</code> error, consider moving the definition of your solution in a separate file, importing it in the notebook and calling it from <code>solution_exercise1</code>. \n",
    "        </li>\n",
    "    </ul>\n",
    "<div>\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "%reload_ext tutorial.tests.testsuite\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "%%ipytest \n",
    "import pathlib\n",
    "\n",
    "def solution_exercise1(input_file: pathlib.Path, size: int, n_processes: int) -> dict[str, int]:\n",
    "    \"\"\"Write your solution here\"\"\"\n",
    "    return dict()\n",
    "\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Exercise 2: Find super secret server key🌶️🌶️🌶️\n",
    "We have a super secret server that was started with a super secret key. \n",
    "Our goal is to retrieve the key. \n",
    "The server behaves like this:\n",
    "1. every time we **await** the `get_value` coroutine method,  it returns a a character. If you run:\n",
    "    ```python\n",
    "    import asyncio\n",
    "    server = SecretServer()\n",
    "    async def my_fun():\n",
    "        await server.get_value()\n",
    "    value = asyincio.run(my_fun())\n",
    "    ```\n",
    "    your `value` will contain the current character.\n",
    "\n",
    "1. If we can get all the characters from the server, they form a sequence that is the secret password to our server:\n",
    "   1. The beginning and end of the sequence are separated by the character `/`.\n",
    "2. **But be careful!**: the server has a special security mechanism: \n",
    "   1. after the first time you call it, you only have a limited number of times to try and call `get_value`. \n",
    "   2. If you wait too long to get the next character, the sequence will reset. \n",
    "   3. In this case too the server will return `/` and then restart.\n",
    "\n",
    "Your task is to write a **coroutine** that gathers all characters from the server that form a sentence delimited by `/` and use them as a password to get the secret message.\n",
    "\n",
    "In summary:\n",
    "\n",
    "- Complete the function `solution_exercise2`. The function should return the string containing the secret message. Write all the asynchronous code inside of `get_secret`\n",
    "- The function receives an object of type `SecretServer`. \n",
    "- Using the async method `get_value` you can get the next value of the sequence\n",
    "- Whenever the sequence stops or the time expires, the function returns the character `/`\n",
    "- You can check if your sequence is correct using the method `check_key`\n",
    "\n",
    "<div class=\"alert alert-block alert-info\">\n",
    "    <h4><b>Hints</b></h4>\n",
    "    <ul>\n",
    "        <li>\n",
    "            Use <code>asyncio.gather</code> to await on multiple coroutines concurrently.\n",
    "        </li>\n",
    "        <li>\n",
    "           To facilitate your work, assume that the key  isn't longer than 50 characters.\n",
    "        </li>\n",
    "        <li>\n",
    "            You can check if your sequence is correct using the method `check_key`\n",
    "        </li>\n",
    "    </ul>\n",
    "<div>\n",
    "\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%%ipytest debug async\n",
    "import asyncio\n",
    "from tutorial.tests.test_threads import SecretServer\n",
    "\n",
    "async def solution_exercise2(server: SecretServer) -> str:\n",
    "    \"\"\"Write your solution here\"\"\"\n",
    "    return await server.get_value()\n",
    "    \n"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3 (ipykernel)",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.11.1"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}