9.3. Threading

thread
lock
daemon
worker
timer

9.3.1. Daemon

Some threads do background tasks, like sending keepalive packets, or performing periodic garbage collection, or whatever. These are only useful when the main program is running, and it's okay to kill them off once the other, non-daemon, threads have exited.

Without daemon threads, you'd have to keep track of them, and tell them to exit, before your program can completely quit. By setting them as daemon threads, you can let them run and forget about them, and when your program quits, any daemon threads are killed automatically.

9.3.2. Delay execution

  • dlaczego nie time.sleep()

  • rekurencyjny timer

Delay execution:

from threading import Timer


DELAY_SECONDS = 5.0

def hello():
    print('Hello world!')


t = Timer(DELAY_SECONDS, hello)
t.start()

print('Main Thread')

Recurrent timer:

from threading import Timer


DELAY_SECONDS = 5.0

def hello():
    print('Timer Thread')
    Timer(DELAY_SECONDS, hello).start()


t = Timer(DELAY_SECONDS, hello)
t.start()

print('Main Thread')

9.3.3. Creating Threads

from threading import Thread


class MyThread(Thread):
    def run(self):
        print('hello')


t = MyThread()
t.start()

9.3.4. Thread Synchronisation

from threading import Thread


class MyThread(Thread):
    def run(self):
        print('hello')


t1 = MyThread()
t1.start()

t2 = MyThread()
t2.start()

t1.join()
t2.join()

from threading import Thread

RUNNING = []


class MyThread(Thread):
    def run(self):
        print('hello')


t1 = MyThread()
t1.start()
RUNNING.append(t1)

t2 = MyThread()
t2.start()
RUNNING.append(t2)

for thread in RUNNING:
    thread.join()

from threading import Thread

RUNNING = []


class MyThread(Thread):
    def run(self):
        print('hello')


def spawn(cls, count=1):
    for i in range(count):
        t = cls()
        t.start()
        RUNNING.append(t)


spawn(MyThread, count=10)


for thread in RUNNING:
    thread.join()

9.3.5. Joining Threads

Joining Threads:

from queue import Queue
from threading import Thread, Lock
from time import sleep


EXIT = False
LOCK = Lock()
TODO = Queue()
RUNNING = []


class MyThread(Thread):
    def run(self):
        while not EXIT:
            # Remove and return an item from the queue.
            job = TODO.get()

            # Execute work
            print(f'Will do the work: {job}')

            # Indicate that a formerly enqueued task is complete.
            TODO.task_done()
            sleep(1)

        print(f'Exiting {self.name}')


# Create new threads
def spawn_worker(count=1):
    for i in range(count):
        thread = MyThread()
        thread.start()
        RUNNING.append(thread)


if __name__ == '__main__':
    spawn_worker(5)

    # Fill the queue
    with LOCK:
        for task in ['One', 'Two', 'Three', 'Four', 'Five']:
            TODO.put(task)

    # Wait for queue to empty
    while not TODO.empty():
        pass

    # Notify threads it's time to exit
    EXIT = True

    # Wait for all threads to complete
    for thread in RUNNING:
        thread.join()

    print(f'Exiting Main Thread')

9.3.6. Workers

Worker model:

from queue import Queue
from threading import Thread

TODO = Queue()


class Worker(Thread):
    def run(self):
        while True:
            # Remove and return an item from the queue.
            job = TODO.get()

            # Execute work
            print(f'Will do the work: {job}')

            # Indicate that a formerly enqueued task is complete.
            TODO.task_done()


def spawn_worker(count=1):
    for i in range(count):
        Worker().start()


if __name__ == '__main__':
    spawn_worker(3)

    TODO.put('ping')
    TODO.put('ls -la')
    TODO.put('echo "hello world"')
    TODO.put('cat /etc/passwd')

    # wait to complete all tasks
    TODO.join()

9.3.7. Assignments

Code 9.1. Solution
"""
* Assignment: Wielowątkowość
* Complexity: easy
* Lines of code: 20 lines
* Time: 21 min

English:
    TODO: English Translation

Polish:
    1. Stwórz kolejkę `queue` do której dodasz różne polecenia systemowe do wykonania, np.:
        a. Linux/macOS: `['/bin/ls /etc/', '/bin/echo "test"', '/bin/sleep 2']`,
        b. Windows: `['dir c:\\Users', 'echo "test"', 'type %HOMEPATH%\Desktop\README.txt']`.
    2. Następnie przygotuj trzy wątki workerów, które będą wykonywały polecenia z kolejki
    3. Wątki powinny być uruchamiane jako `subprocess.run()` w systemie operacyjnym z timeoutem równym `TIMEOUT = 2.0` sekundy
    4. Ilość poleceń może się zwiększać w miarę wykonywania zadania.
    5. Wątki mają być uruchomione w tle (ang. `daemon`)

:Extra task:
    1. Wątki powinny być uśpione za pomocą `Timer` przez `DELAY = 5.0` sekund, a następnie ruszyć do roboty
    2. Parametry rozbij za pomocą `shlex`
    3. Użyj logowania za pomocą biblioteki `logging` tak aby przy wyświetlaniu wyników widoczny był identyfikator procesu i wątku.

Hints:
    * Ustaw parametr `shell=True` dla `subprocess.run()`

Tests:
    TODO: Doctests
"""


# Given
TIMEOUT = 2.0
DELAY = 5.0
TODO = ['ping python.astrotech.io',
        'ls -la',
        'echo "hello world"',
        'cat /etc/passwd']

9.3.8. References

Hettinger2017

Hettinger, Raymond. Keynote on Concurrency. PyBay 2017. https://youtu.be/9zinZmE3Ogk?t=1243