Python - Advanced Workshop.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "id": "c7595815",
   "metadata": {},
   "source": [
    "# Python Advanced Workshop\n",
    "\n",
    "#### Einleitung\n",
    "Dieses Notebook enthält das gesamte Material des Workshops und dient als interaktive Resource.  \n",
    "Code-Zellen können mit Strg+Enter oder Shift+Enter ausgeführt werden."
   ]
  },
  {
   "cell_type": "markdown",
   "id": "67728b9b",
   "metadata": {},
   "source": [
    "## Themen\n",
    "\n",
    "Es gibt viele Konzepte und Themen die für einen Fortgeschrittenen-Workshop infrage kommen.  \n",
    "Hier sollen drei Konzepte vorgestellt und mit Beispielen aus der Praxis besprochen werden:\n",
    "* Decorator\n",
    "* Contextmanager\n",
    "* Concurrency"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "4e44f27f",
   "metadata": {},
   "source": [
    "## Decorator\n",
    "Im Basic Workshop haben wir besprochen, dass Funktionen in Python ebenfalls First Class Objekte  \n",
    "sind und an andere Funktionen übergeben werden können."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "fc466b16",
   "metadata": {},
   "outputs": [],
   "source": [
    "def my_add(a,b):\n",
    "    return a+b\n",
    "\n",
    "def hashbox_print(fcn, *args,**kwargs):          # Funktionen sind 1. class Objekte und können ganz normal übergeben werden\n",
    "    result_string = str( fcn(*args,**kwargs) )   # *args und **kwargs werden weitergereicht\n",
    "    print(\"#\"*(len(result_string)+4))\n",
    "    print(\"# \"+result_string+\" #\")\n",
    "    print(\"#\"*(len(result_string)+4))\n",
    "\n",
    "hashbox_print(my_add,2,3)                        # my_add() ruft die Funktion auf, my_add ist das Funktionsobjekt"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "96912311",
   "metadata": {},
   "source": [
    "Auf diesem Prinzip basierend können wir Funktionen bauen, die andere Funktionen erweitern oder deren Rückgabewerte abfangen und verändern."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "bb2a206d",
   "metadata": {},
   "outputs": [],
   "source": [
    "def hashbox_dec(fcn):                          # Decorator Funktion, bekommt eine Funktion als Parameter\n",
    "    def wrapper(*args,**kwargs):               # Wrapped die Originalfunktion\n",
    "        value = fcn(*args,**kwargs)            # Aufruf der Originalfunktion, *args und **kwargs werden weitergereicht\n",
    "        result_string = str(value)\n",
    "        print(\"#\"*(len(result_string)+4))\n",
    "        print(\"# \"+result_string+\" #\")         # Zusätzliche Funktionalität\n",
    "        print(\"#\"*(len(result_string)+4))\n",
    "        return value                           # Rückgabe des ursprünglichen Funktionswertes\n",
    "    return wrapper                             # Rückgabe der neudefinierten wrapper Funktion\n",
    "\n",
    "decorated_my_add = hashbox_dec(my_add)         # Dekorieren einer Funktion, umständliche Syntax\n",
    "\n",
    "decorated_my_add(2,3)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "eb371065",
   "metadata": {},
   "source": [
    "Python bringt syntaktischen Zucker mit um das Dekorieren von Funktionen zu vereinfachen."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "3087ac62",
   "metadata": {},
   "outputs": [],
   "source": [
    "@hashbox_dec            # Decorator Syntax\n",
    "def my_add(a,b):\n",
    "    return a+b\n",
    "\n",
    "# Gleichbedeutend mit:\n",
    "# my_add = hashbox_dec(my_add)\n",
    "\n",
    "my_add(4,5)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "543de7ea",
   "metadata": {},
   "source": [
    "Alternativ können auch Klassen als Decorator verwendet werden."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "98c76d11",
   "metadata": {},
   "outputs": [],
   "source": [
    "class OneMore:                                # Decorator Klasse\n",
    "    def __init__(self, fcn):                  # Init nimmt die zu dekorierende Funktion entgegen\n",
    "        self.fcn = fcn                        # und speichert diese\n",
    "        \n",
    "    def __call__(self,*args,**kwargs):        # Anstelle einer inneren Funktion definieren wir die call Methode\n",
    "        original = self.fcn(*args,**kwargs)   # Aufruf der originalen Funktion\n",
    "        return original+1                     # Rückgabe eines veränderten Wertes\n",
    "    \n",
    "@OneMore\n",
    "def my_sub(a,b):\n",
    "    return a-b\n",
    "\n",
    "my_sub(5,2)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "43a63825",
   "metadata": {},
   "source": [
    "Dieser sehr simple Weg hat einige Nachteile, darunter der Verlust von Funktionssignatur und Docstrings."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "3102883d",
   "metadata": {},
   "outputs": [],
   "source": [
    "def hashbox_dec(fcn):\n",
    "    \"\"\"\n",
    "    Decorate a function with nice hashbox printing.\n",
    "    \n",
    "    Parameters\n",
    "    ----------\n",
    "    fcn : callable\n",
    "        Function to decorate.\n",
    "    \n",
    "    Returns\n",
    "    -------\n",
    "    callable\n",
    "        Decorated function\n",
    "    \"\"\"\n",
    "    def wrapper(*args,**kwargs):\n",
    "        \"\"\"\n",
    "        Print the result of the original function in a hashbox before returning the value.\n",
    "        \n",
    "        Parameters\n",
    "        ----------\n",
    "        *args : tuple\n",
    "            Positional arguments to original function.\n",
    "        **kwargs : dict\n",
    "            Keyword arguments to original function.\n",
    "        \n",
    "        Returns\n",
    "        -------\n",
    "        any\n",
    "            Result of original function.\n",
    "        \"\"\"\n",
    "        value = fcn(*args,**kwargs)            \n",
    "        result_string = str(value)\n",
    "        print(\"#\"*(len(result_string)+4))\n",
    "        print(\"# \"+result_string+\" #\")         \n",
    "        print(\"#\"*(len(result_string)+4))\n",
    "        return value                           \n",
    "    return wrapper    \n",
    "\n",
    "@hashbox_dec\n",
    "def my_mult(a,b):\n",
    "    \"\"\"\n",
    "    Multiply two numbers.\n",
    "    \n",
    "    Parameters\n",
    "    ----------\n",
    "    a : int or float\n",
    "        First factor.\n",
    "    b : int or float\n",
    "        Second factor.\n",
    "    \n",
    "    Returns\n",
    "    -------\n",
    "    int or float\n",
    "        Product of `a` and `b`.\n",
    "    \"\"\"\n",
    "    return a*b"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "a49ab109",
   "metadata": {},
   "outputs": [],
   "source": [
    "my_mult"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "54595a28",
   "metadata": {},
   "outputs": [],
   "source": [
    "help(my_mult)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "5c9908b4",
   "metadata": {},
   "source": [
    "Dieses Problem lässt sich einfach lösen, nämlich mit einem Decorator!"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "321f9acb",
   "metadata": {},
   "outputs": [],
   "source": [
    "import functools\n",
    "\n",
    "def hashbox_dec(fcn):\n",
    "    \"\"\"\n",
    "    Print fcn result in nice hashbox before returning the value.\n",
    "    \n",
    "    Parameters\n",
    "    ----------\n",
    "    fcn : callable\n",
    "        Function to decorate.\n",
    "    \n",
    "    Returns\n",
    "    -------\n",
    "    callable\n",
    "        Decorated function\n",
    "    \"\"\"\n",
    "    @functools.wraps(fcn)                   # <---- Magic happens here!\n",
    "    def wrapper(*args,**kwargs):\n",
    "        value = fcn(*args,**kwargs)            \n",
    "        result_string = str(value)\n",
    "        print(\"#\"*(len(result_string)+4))\n",
    "        print(\"# \"+result_string+\" #\")         \n",
    "        print(\"#\"*(len(result_string)+4))\n",
    "        return value                           \n",
    "    return wrapper    \n",
    "\n",
    "@hashbox_dec\n",
    "def my_mult(a,b):\n",
    "    \"\"\"\n",
    "    Multiply two numbers.\n",
    "    \n",
    "    Parameters\n",
    "    ----------\n",
    "    a : int or float\n",
    "        First factor.\n",
    "    b : int or float\n",
    "        Second factor.\n",
    "    \n",
    "    Returns\n",
    "    -------\n",
    "    int or float\n",
    "        Product of `a` and `b`.\n",
    "    \"\"\"\n",
    "    return a*b"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "c668be50",
   "metadata": {},
   "outputs": [],
   "source": [
    "my_mult"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "f88aa94a",
   "metadata": {},
   "outputs": [],
   "source": [
    "help(my_mult)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "69a39efc",
   "metadata": {},
   "source": [
    "### Best practise: Nutzt immer @functools.wraps wenn ihr Decorator schreibt!\n",
    "\n",
    "Bei genauem Hinsehen sieht der Code oben komisch aus!   \n",
    "\n",
    "@functools.wraps(fcn)  \n",
    "def wrapper(\\*args,\\*\\*kwargs)\n",
    "\n",
    "Hier wird ein Parameter (fcn) an functools.wraps übergeben und das ganze als decorator verwendet.\n",
    "Zum Verständnis können wir uns das ganze ohne den Zucker ansehen.  "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "ff71c487",
   "metadata": {},
   "outputs": [],
   "source": [
    "def wrapper(*args,**kwargs):\n",
    "    pass\n",
    "\n",
    "def fcn():\n",
    "    pass\n",
    "\n",
    "fcn = functools.wraps(fcn)(wrapper)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "ca3f92e7",
   "metadata": {},
   "source": [
    "functools.wraps(fcn) muss also ein callable Objekt zurückgeben, das eine Funktion als Parameter nimmt.  \n",
    "Oder andersgesagt functools.wraps() ist kein Decorator sondern eine Decorator-Factory!  \n",
    "\n",
    "Im Folgenden wird zur Klarheit zwischen Decorator und Factory unterschieden, oft wird das aber nicht klar getrennt und beides wird einfach als Decorator bezeichnet."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "92916bbd",
   "metadata": {},
   "outputs": [],
   "source": [
    "def create_count_dec(counter_start):          # Factory, gibt einen Decorator zurück\n",
    "    def count_dec(fcn):                       # Der eigentliche Decorator\n",
    "        @functools.wraps(fcn)\n",
    "        def wrapper(*args,**kwargs):          # Der Wrapper\n",
    "            wrapper.counter +=1\n",
    "            return fcn(*args,**kwargs)\n",
    "        wrapper.counter=counter_start         # Initialisierung des states\n",
    "        return wrapper\n",
    "    return count_dec\n",
    "\n",
    "@create_count_dec(0)\n",
    "def my_div(a,b):\n",
    "    return a/b\n",
    "\n",
    "print(my_div(6,2))\n",
    "print(my_div(9,3))\n",
    "print(my_div(100,2))\n",
    "\n",
    "print(f\"my_div was called {my_div.counter} times!\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "2d20eaff",
   "metadata": {},
   "outputs": [],
   "source": [
    "import functools\n",
    "class CounterDecFactory:                            # Für Klassen gibt es verschiedene Ansätze\n",
    "    def __init__(self,counter_start):               # Die init Methode bekommt jetzt die Parameter\n",
    "        self.counter = counter_start                \n",
    "        self.mode = \"decorating\"                    # Hilfsvariable um Verhalten von call zu steuern\n",
    "        \n",
    "    def __call__(self,*args,**kwargs):              \n",
    "        if self.mode == \"decorating\":\n",
    "            self.fcn = args[0]                      # Erster Aufruf, erstes Argument ist zu dekorierende Funktion\n",
    "            self.mode = \"calling\"\n",
    "            functools.update_wrapper(self,self.fcn) # in etwa äquivalent zu functools.wraps\n",
    "            return self                             # Rückgabe der Dekorierten Funktion\n",
    "         \n",
    "        self.counter +=1                            # weitere Aufrufe\n",
    "        return self.fcn(*args, **kwargs)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "c81a821e",
   "metadata": {},
   "outputs": [],
   "source": [
    "@CounterDecFactory(5)\n",
    "def my_div(a,b):\n",
    "    \"\"\"\n",
    "    Divide a by b.\n",
    "    \n",
    "    Parameters\n",
    "    ----------\n",
    "    a : number\n",
    "        The dividend.\n",
    "    b : number\n",
    "        The divisor.\n",
    "        \n",
    "    Returns\n",
    "    -------\n",
    "    number\n",
    "        The quotient a/b.\n",
    "    \"\"\"\n",
    "    return a/b\n",
    "\n",
    "print(my_div(6,2))\n",
    "print(my_div(9,3))\n",
    "print(my_div(100,2))\n",
    "\n",
    "print(f\"my_div was called {my_div.counter} times!\")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "b9a2f2e4",
   "metadata": {},
   "source": [
    "Häufige Anwendungsfälle für Decorator sind Logging, Authentifizierungschecks, Caching und Metalogik da Informationen über die dekorierte Funktion verfügbar sind.  \n",
    "Auch beim Definieren von Klassen sind Decorator wie @property, @staticmethod, @classmethod oder @abstractmethod nützlich und häufig anzutreffen.  \n",
    "Funktionen als Decorator haben den Vorteil, dass die Signatur und Docstrings sowie der Type besser erhalten bleiben.  \n",
    "Klassen sind besser geeignet wenn ein (komplexer) State erhalten werden soll."
   ]
  },
  {
   "cell_type": "markdown",
   "id": "9cef1681",
   "metadata": {},
   "source": [
    "## Contextmanager\n",
    "Contextmanager definieren einen Block bei dessen Ein- und Austritt spezieller Code ausgeführt wird.  \n",
    "Sie vereinfachen das Exceptionhandling indem sie sicherstellen, dass egal an welcher Stelle im Inneren eine Exception geworfen wird, der Austrittscode IMMER ausgeführt wird. Dabei sind sie simpler und schlanker zu nutzen als try / except / finally Blöcke.\n",
    "\n",
    "Häufige Anwendungsfälle sind Resourcenmanagement (file handles, database connections, sockets, locks & semaphores)\n",
    "und das Standardbeispiel ist das  \n",
    "with open() as f:  \n",
    "Statement.\n",
    "\n",
    "Hier soll es darum gehen wie wir unsere eigenen Contextmanager erstellen können und wie Exceptions verarbeitet werden."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "8582ff04",
   "metadata": {},
   "outputs": [],
   "source": [
    "class file:                                            # Contextmanager als Klasse \n",
    "                                                       # muss enter und exit Methoden haben    \n",
    "    def __init__(self, file_name, method):             # Initialisierung\n",
    "        print(\"Init file\")\n",
    "        self.file_name = file_name\n",
    "        self.method = method\n",
    "        \n",
    "    def __enter__(self):                               # Eintritt\n",
    "        print(\"Entering the context...\")\n",
    "        self.file_obj = open(self.file_name, self.method)\n",
    "        return self.file_obj\n",
    "    \n",
    "    def __exit__(self, exctype, excinst, traceback):   # Austritt\n",
    "        print(\"...leaving the context.\")\n",
    "        self.file_obj.close()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "d25fc97d",
   "metadata": {},
   "outputs": [],
   "source": [
    "with file(\"foo.txt\",\"w\") as ofile:\n",
    "    print(\"inside the context\")\n",
    "    ofile.write(\"bar\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "064f2e89",
   "metadata": {},
   "outputs": [],
   "source": [
    "with file(\"foo.txt\",\"w\") as ofile:\n",
    "    print(\"inside the context\")\n",
    "    raise Exception(\"My custom exception text.\")     # Exception innerhalb des with Blocks\n",
    "    ofile.write(\"bar\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "524a4150",
   "metadata": {},
   "outputs": [],
   "source": [
    "class file:\n",
    "    \n",
    "    def __init__(self, file_name, method):            # Initialisierung\n",
    "        print(\"Init file\")\n",
    "        self.file_name = file_name\n",
    "        self.method = method\n",
    "        \n",
    "    def __enter__(self):                              # Eintritt\n",
    "        print(\"Entering the context...\")\n",
    "        self.file_obj = open(self.file_name, self.method)\n",
    "        return self.file_obj\n",
    "    \n",
    "    def __exit__(self, exctype, excinst, traceback):  # Austritt\n",
    "        print(\"...leaving the context.\")\n",
    "        self.file_obj.close()\n",
    "        if exctype is not None:                       # Zusätzliche Argumente werden bei Exceptions genutzt\n",
    "            print(exctype)\n",
    "            print(excinst)\n",
    "            print(traceback)\n",
    "            print(\"Exception has been handled.\")\n",
    "            return True                               # Exception abgefangen, andere return Werte re-raisen"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "3d5b6b0e",
   "metadata": {},
   "outputs": [],
   "source": [
    "with file(\"foo.txt\",\"w\") as ofile:\n",
    "    print(\"inside the context\")\n",
    "    raise Exception(\"My custom exception text.\")     # Exception innerhalb des with Blocks\n",
    "    ofile.write(\"bar\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "34ab6cdd",
   "metadata": {},
   "outputs": [],
   "source": [
    "class suppress:                              # Implementierung von contextlib.suppress\n",
    "    def __init__(self, *exceptions):\n",
    "        print(\"Init suppress\")\n",
    "        self._exceptions = exceptions\n",
    "\n",
    "    def __enter__(self):\n",
    "        print(\"Entering suppress\")\n",
    "        pass\n",
    "\n",
    "    def __exit__(self, exctype, excinst, exctb):\n",
    "        print(\"Exiting suppress\")\n",
    "        return exctype is not None and issubclass(exctype, self._exceptions)\n",
    "\n",
    "with suppress(FileNotFoundError):\n",
    "    print(\"inside with\")\n",
    "    open(\"non_existing_file.txt\",\"r\")\n",
    "    print(\"after raise\")\n",
    "print(\"continuing after suppress context\")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "7a9ffe5f",
   "metadata": {},
   "source": [
    "Oft können Contextmanager sehr einfach gehalten werden und die Klassendefinition ist unnötiger Overhead.  \n",
    "Das Stdlib Modul contextlib bringt einige Hilfstools zur Vereinfachung mit."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "2d2d7d2f",
   "metadata": {},
   "outputs": [],
   "source": [
    "from contextlib import closing     # Contextmanager closing(resource) der in __exit__ resource.close() aufruft\n",
    "from urllib.request import urlopen\n",
    "\n",
    "with closing(urlopen('https://www.python.org')) as page:\n",
    "    for line in page:\n",
    "        print(line)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "f794133c",
   "metadata": {},
   "outputs": [],
   "source": [
    "import contextlib\n",
    "\n",
    "@contextlib.contextmanager            # Decorator!\n",
    "def open_file(name):                  # Generator Funktion\n",
    "    print(\"Init from function\")       # Initialisierung\n",
    "    f = open(name, 'w')\n",
    "    try:                              # wie __enter__\n",
    "        print(\"Entering\")\n",
    "        yield f                       # wird an das \"as\" zurückgegeben. falls Exceptions vorkommen werden sie hier reraised\n",
    "    finally:                          # wie __exit__\n",
    "        print(\"Leaving\")\n",
    "        f.close()\n",
    "\n",
    "with open_file(\"foo.txt\") as ofile:\n",
    "    print(\"inside with block\")\n",
    "    ofile.write(\"bar\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "0a8777c9",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Contextmanager die von contextlib.ContextDecorator erben oder solche die \n",
    "# von contextlib.contextmanager erzeugt werden können auch als Decorator genutzt werden...\n",
    "import time\n",
    "\n",
    "@contextlib.contextmanager\n",
    "def log_time(name):\n",
    "    start_time = time.time()\n",
    "    try:\n",
    "        yield\n",
    "    finally:\n",
    "        print(f\"{name} ran {time.time()-start_time:.2f} seconds\")\n",
    "\n",
    "@log_time(\"output\")\n",
    "def delayed_out(val, delay=1):\n",
    "    time.sleep(delay)\n",
    "    print(val)  \n",
    "\n",
    "# Syntaktischer zucker für\n",
    "# def delayed_out(val, delay=1):\n",
    "#     with log_time(output):\n",
    "#         ....\n",
    "\n",
    "delayed_out(\"spam\")\n",
    "\n",
    "with log_time(\"calculation\"):\n",
    "    print(2*5)\n",
    "    time.sleep(0.5)\n"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "0b37ca60",
   "metadata": {},
   "source": [
    "## Ein reales Beispiel: Registrierung und Checkpointing von PySpark Transformationen\n",
    "Siehe Examples"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "90ee0d48",
   "metadata": {},
   "source": [
    "## Concurrency & Parallelism\n",
    "In Python gibt es drei Arten von Concurrency:  \n",
    "* Multiprocessing\n",
    "* Threading\n",
    "* Asyncio  \n",
    "\n",
    "Jedes dieser Konzepte kommt mit Vor- und Nachteilen und es gilt das Richtige für den konkreten Anwendungsfall auszuwählen.  \n",
    "\n",
    "Dieses Kapitel folgt Vorträgen von Raymond Hettinger und David Beazley:  \n",
    "[Raymond Hettinger - Keynote on Concurrency, PyBay 2017](https://www.youtube.com/watch?v=9zinZmE3Ogk) und  \n",
    "[David Beazley - Python Concurrency From the Ground Up: LIVE!, PyCon 2015](https://www.youtube.com/watch?v=MCs5OvhV9S4)  \n",
    "Von beiden gibt es eine Menge an Vorträgen auf Youtube die sich lohnen!\n",
    "\n",
    "### Multiprocessing\n",
    "Multiprocessing nutzt mehrere CPUs eines Systems parallel. Dabei werden mehrere Python Interpreter als individuelle Prozesse ohne shared-memory gestartet, die weitesgehend unabhängig voneinander laufen. Multiprocessing in Python unterscheidet sich damit kaum von Multiprocessing in anderen Programmiersprachen.  \n",
    "\n",
    "#### Vorteile\n",
    "* Nutzung mehrerer CPUs (cores)\n",
    "* Getrennte Speicherbereiche (keine race conditions)\n",
    "\n",
    "#### Nachteile\n",
    "* Getrennte Speicherbereiche (langsame Kommunikation)\n",
    "* Overhead\n",
    "* Limitierung durch Hardware und Betriebssystem (100e Prozesse sind keine gute Idee)\n",
    "\n",
    "#### Wann\n",
    "* CPU limitierter Workload in purem Python"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "0ba5e550",
   "metadata": {},
   "source": [
    "### Multithreading\n",
    "Python hat einen **Global Interpreter Lock** (GIL) welcher den gleichzeitigen Zugriff auf Python Objekte verhindert da Pythons Memorymanagement nicht thread-safe ist!  \n",
    "Das bedeutet, das standard Python Code in Threads **nicht** parallel ausgeführt wird, sondern der Interpreter zwischen den Threads hin und her springt.  \n",
    "Einige potenziell blockende oder lang laufenden Berechnungen wie z.B. I/O oder NumPy Berechnungen geben den GIL frei und können parallel laufen.\n",
    "\n",
    "#### Vorteile\n",
    "* Shared Memory\n",
    "* Funktioniert mit bestehendem Code\n",
    "* Wenig zusätzliche Tools nötig (Locks, Queues, Pools)\n",
    "\n",
    "#### Nachteile\n",
    "* Shared Memory (race conditions)\n",
    "* Overhead (weniger als bei Multiprocessing)\n",
    "* Limitierung durch Betriebssystem (1000e Threads sind immernoch keine gute Idee)\n",
    "* Threadwechsel wird durch Interpreter gesteuert --> Overhead und Locks nötig\n",
    "\n",
    "#### Wann\n",
    "* I/O limitierter Workload\n",
    "* Workload der den GIL freigibt"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "c8322cfe",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Ausgangspunkt: keine concurrency\n",
    "\n",
    "counter = 0\n",
    "def worker():\n",
    "    \"\"\"Increment and print the current counter\"\"\"\n",
    "    global counter\n",
    "\n",
    "    counter += 1\n",
    "    print(f\"The count is {counter}\")\n",
    "    print(\"---------------\")\n",
    "\n",
    "print(\"Starting up\")\n",
    "for i in range(10):\n",
    "    worker()\n",
    "print(\"Finishing up\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "0e4e4ed4",
   "metadata": {},
   "outputs": [],
   "source": [
    "import threading\n",
    "\n",
    "counter = 0\n",
    "\n",
    "def worker():\n",
    "    \"\"\"Increment and print the current counter\"\"\"\n",
    "    global counter\n",
    "\n",
    "    counter += 1\n",
    "    print(f\"The count is {counter}\")\n",
    "    print(\"---------------\")\n",
    "\n",
    "print(\"Starting up\")\n",
    "for i in range(10):\n",
    "    threading.Thread(target=worker).start()  # sehr simples threaden des bestehenden codes\n",
    "print(\"Finishing up\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "43b363b0",
   "metadata": {},
   "outputs": [],
   "source": [
    "import threading, time, random\n",
    "\n",
    "##########################################################################################\n",
    "# Fuzzing is a technique for amplifying race condition errors to make them more visible\n",
    "\n",
    "FUZZ = True\n",
    "\n",
    "def fuzz():\n",
    "    if FUZZ:\n",
    "        time.sleep(random.random())\n",
    "\n",
    "###########################################################################################\n",
    "\n",
    "counter = 0\n",
    "\n",
    "def worker():\n",
    "    \"\"\"Increment and print the current counter\"\"\"\n",
    "    global counter\n",
    "\n",
    "    fuzz()\n",
    "    oldcnt = counter\n",
    "    fuzz()\n",
    "    counter = oldcnt + 1\n",
    "    fuzz()\n",
    "    print(f\"The count is {counter}\",end=\"\")\n",
    "    fuzz()\n",
    "    print()\n",
    "    fuzz()\n",
    "    print(\"---------------\",end=\"\")\n",
    "    fuzz()\n",
    "    print()\n",
    "    fuzz()\n",
    "\n",
    "print(\"Starting up\")\n",
    "fuzz()\n",
    "for i in range(10):\n",
    "    threading.Thread(target=worker).start()\n",
    "    fuzz()\n",
    "print(\"Finishing up\")\n",
    "fuzz()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "f1ec6f4b",
   "metadata": {},
   "outputs": [],
   "source": [
    "import threading, time, random\n",
    "##########################################################################################\n",
    "# Fuzzing is a technique for amplifying race condition errors to make them more visible\n",
    "\n",
    "FUZZ = True\n",
    "\n",
    "def fuzz():\n",
    "    if FUZZ:\n",
    "        time.sleep(random.random())\n",
    "\n",
    "###########################################################################################\n",
    "\n",
    "counter_lock = threading.Lock()  # to guard critical sections\n",
    "printer_lock = threading.Lock()\n",
    "\n",
    "counter = 0\n",
    "\n",
    "def worker():\n",
    "    \"\"\"Increment and print the current counter\"\"\"\n",
    "    global counter\n",
    "    \n",
    "    with counter_lock: # Critical section! Counter is accessed!\n",
    "        fuzz()\n",
    "        oldcnt = counter\n",
    "        fuzz()\n",
    "        counter = oldcnt + 1\n",
    "        fuzz()\n",
    "        with printer_lock: # Critical section! Print is used\n",
    "            print(f\"The count is {counter}\",end=\"\")\n",
    "            fuzz()\n",
    "            print()\n",
    "            fuzz()\n",
    "            print(\"---------------\",end=\"\")\n",
    "            fuzz()\n",
    "            print()\n",
    "            fuzz()\n",
    "\n",
    "with printer_lock:\n",
    "    print(\"Starting up\",end=\"\")\n",
    "    fuzz()\n",
    "    print()\n",
    "\n",
    "worker_threads = []\n",
    "for i in range(10):\n",
    "    t = threading.Thread(target=worker)\n",
    "    worker_threads.append(t)\n",
    "    t.start()\n",
    "    fuzz()\n",
    "for t in worker_threads:\n",
    "    t.join()\n",
    "    fuzz()\n",
    "\n",
    "with printer_lock:\n",
    "    print('Finishing up', end='')\n",
    "    fuzz()\n",
    "    print()\n",
    "\n",
    "fuzz()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "5c35f889",
   "metadata": {},
   "source": [
    "### Performance von Multithreading\n",
    "Siehe Beispiel"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "5861ffd1",
   "metadata": {},
   "source": [
    "### Asyncio\n",
    "Python bringt ein weiteres Konzept der Concurrency mit: Coroutines  \n",
    "Coroutinen baseieren auf und verhalten sich sehr ähnlich zu Generator-Funktionen.  \n",
    "Das heißt, ihre Ausführung kann unterbrochen und fortgesetzt werden und das mit sehr wenig CPU Zeit!  \n",
    "\n",
    "Auch asyncio ist kein Weg um den GIL! Es wird weiterhin nur ein CPU-Kern genutzt.\n",
    "\n",
    "#### Vorteile\n",
    "* Sehr wenig Overhead (1000e oder sogar 10000e Concurrent Coroutines sind kein Problem)\n",
    "* Kooperatives Wechseln zwischen Tasks --> keine Locks nötig! Sehr viel einfacher bei komplexem Code\n",
    "\n",
    "#### Nachteile\n",
    "* Funktioniert nicht mit bestehendem Code, eigene Funktionen nötig\n",
    "* Alles muss non-blocking sein!\n",
    "* Mehr zusätzliche Tools nötig (futures, event loops, \n",
    "* Ungewohnter als Threading\n",
    "\n",
    "#### Wann\n",
    "* I/O limitierter Workload\n",
    "* Sehr große Menge an Tasks"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "563e3683",
   "metadata": {},
   "source": [
    "### Ein Beispiel: Azure File Upload mit Python SDK -- Threading vs Asyncio"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "487bb01d",
   "metadata": {},
   "outputs": [],
   "source": [
    "# uncomment and run this if you don't have the packages installed\n",
    "#!pip install azure-storage-blob aiohttp"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "f27b3173",
   "metadata": {},
   "outputs": [],
   "source": [
    "adls_uri = <your_adls>\n",
    "container_name = <your_container>\n",
    "sas_token = <your_token>"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "d2d85b9c",
   "metadata": {},
   "outputs": [],
   "source": [
    "import hashlib\n",
    "import time\n",
    "from concurrent.futures import ThreadPoolExecutor\n",
    "from pathlib import Path\n",
    "\n",
    "from azure.storage.blob import ContainerClient\n",
    "\n",
    "\n",
    "def calc_md5_sum(file):\n",
    "    file = Path(file)\n",
    "    with file.open(\"rb\") as data:\n",
    "        md5 = hashlib.md5()\n",
    "        while chunk := data.read(4096):\n",
    "            md5.update(chunk)\n",
    "    return md5.digest()\n",
    "\n",
    "con_client = ContainerClient(adls_uri,container_name,credential=sas_token)\n",
    "source_path = Path(\"files\")\n",
    "\n",
    "def work(file):\n",
    "    blob_name = f\"PythonWorkshop/{file.name}\"\n",
    "    blob_client = con_client.get_blob_client(blob_name)\n",
    "    change = True\n",
    "    if blob_client.exists():\n",
    "        remote_md5 = blob_client.get_blob_properties()[\"content_settings\"][\"content_md5\"]\n",
    "        local_md5 = calc_md5_sum(file)\n",
    "        change = not(local_md5 == remote_md5)\n",
    "    if change:\n",
    "        with file.open(\"rb\") as src_file:\n",
    "            blob_client.upload_blob(src_file,overwrite=True)\n",
    "    \n",
    "\n",
    "start = time.time()\n",
    "with con_client:\n",
    "    with ThreadPoolExecutor() as executor:    # Tooling Teil 1\n",
    "        for file in source_path.iterdir():\n",
    "            executor.submit(work,file)        # Tooling Teil 2\n",
    "end = time.time()\n",
    "print(f\"upload took {end-start:.2f}s\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "6e04affa",
   "metadata": {},
   "outputs": [],
   "source": [
    "import asyncio\n",
    "import hashlib\n",
    "import time\n",
    "from pathlib import Path\n",
    "\n",
    "from azure.storage.blob.aio import ContainerClient as ContainerClientAIO\n",
    "\n",
    "\n",
    "con_client = ContainerClientAIO(adls_uri,container_name,credential=sas_token)\n",
    "source_path = Path(\"files\")\n",
    "\n",
    "background_tasks = set()                                          # Tooling\n",
    "\n",
    "def calc_md5_sum(file):   # Blocking! Kein async keyword\n",
    "    file = Path(file)\n",
    "    with file.open(\"rb\") as data:\n",
    "        md5 = hashlib.md5()\n",
    "        while chunk := data.read(4096):\n",
    "            md5.update(chunk)\n",
    "    return md5.digest()\n",
    "\n",
    "async def getFiles():  # Tooling\n",
    "    for f in source_path.iterdir():\n",
    "        yield f\n",
    "\n",
    "async def work(file):                                              # Tooling\n",
    "    blob_name = f\"PythonWorkshop/{file.name}\"\n",
    "    blob_client = con_client.get_blob_client(blob_name)\n",
    "    change = True\n",
    "    if await blob_client.exists():                                 # Tooling\n",
    "        blob_prop = await blob_client.get_blob_properties()        # Tooling\n",
    "        remote_md5 = blob_prop[\"content_settings\"][\"content_md5\"]\n",
    "        local_md5 = await asyncio.to_thread(calc_md5_sum,file)     # Tooling² + Threading for blocking function!\n",
    "        change = not(local_md5 == remote_md5)\n",
    "    if change:\n",
    "        with file.open(\"rb\") as src_file:\n",
    "            await blob_client.upload_blob(src_file,overwrite=True) # Tooling\n",
    "\n",
    "async def main():                                                  # Tooling\n",
    "    async with con_client:                                         # Tooling\n",
    "        async for file in getFiles():                              # Tooling\n",
    "            task = asyncio.create_task(work(file))                 # Tooling\n",
    "            background_tasks.add(task)                             # Tooling\n",
    "            task.add_done_callback(background_tasks.discard)       # Tooling\n",
    "        await asyncio.gather(*background_tasks)                    # Tooling"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "4f0202d5",
   "metadata": {},
   "outputs": [],
   "source": [
    "start = time.time()\n",
    "await main()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "e40e625a",
   "metadata": {},
   "outputs": [],
   "source": [
    "end = time.time()\n",
    "print(f\"upload took {end-start:.2f}s\")"
   ]
  }
 ],
 "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.10.7"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}