8.27. Threading Timer

8.27.1. 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')

8.27.2. Assignments

Code 8.37. Solution
"""
* Assignment: Concurrency Threading Timer
* Complexity: easy
* Lines of code: 4 lines
* Time: 8 min

English:
    1. Define function `ping()`, with optional parameter
       `n: int`, which defaults to 1
    2. Function `ping()` should append value of `n` to `result`
    3. Function should be called every `INTERVAL`
    4. Function should be called maximum `MAX` times
    5. Use `Timer` from `threading` module
    6. Run doctests - all must succeed

Polish:
    1. Zdefiniuj funkcję `ping(n: int)` z opcjonalnym parametrem
       `n: int`, który domyślnie jest 1
    2. Funkcja `ping()` powinna dopisywać wartość `n` do `result`
    3. Funkcja powinna być wywoływana co `INTERVAL`
    4. Funkcja powinna być wywołana maksymalnie `MAX` razy
    5. Użyj `Timer` z modułu `threading`
    6. Uruchom doctesty - wszystkie muszą się powieść

Tests:
    >>> import sys; sys.tracebacklimit = 0

    >>> def check(result):
    ...     assert result == [1, 2, 3], f'Result is {result}'

    >>> Timer(INTERVAL, ping).start()
    >>> Timer(INTERVAL*MAX+1, check, [result]).start()
"""
from threading import Timer


INTERVAL = 0.1
MAX = 3
result = []


# type: Callable[[int], None]
def ping():
    ...


Code 8.38. Solution
"""
* Assignment: Threading Timer File
* Complexity: medium
* Lines of code: 13 lines
* Time: 13 min

English:
    1. Modify class `File`
    2. Add class configuration attribute `AUTOSAVE_SECONDS: float = 1.0`
    3. Save buffer content to file every `AUTOSAVE_SECONDS` seconds
    4. Writing and reading takes time, how to make buffer save data in the background, but it could be still used?
    5. Run doctests - all must succeed

Polish:
    1. Zmodyfikuj klasę `File`
    2. Dodaj klasowy atrybut konfiguracyjny `AUTOSAVE_SECONDS: float = 1.0`
    3. Zapisuj zawartość bufora do pliku co `AUTOSAVE_SECONDS` sekund
    4. Operacje zapisu i odczytu trwają, jak zrobić, aby do bufora podczas zapisu na dysk, nadal można było pisać?
    5. Uruchom doctesty - wszystkie muszą się powieść

Hint:
    * `from threading import Timer`
    * `timer = Timer(interval, function)`
    * `timer.start()`
    * `timer.cancel()`
    * `ctrl+c` or stop button kills infinite loop

Tests:
    >>> import sys; sys.tracebacklimit = 0
    >>> from inspect import isclass, ismethod
    >>> from os import remove

    >>> assert isclass(File)
    >>> assert hasattr(File, 'append')
    >>> assert hasattr(File, 'AUTOSAVE_SECONDS')
    >>> assert hasattr(File, '__enter__')
    >>> assert hasattr(File, '__exit__')
    >>> assert ismethod(File(None).append)
    >>> assert ismethod(File(None).__enter__)
    >>> assert ismethod(File(None).__exit__)
    >>> assert File.AUTOSAVE_SECONDS == 1.0

    >>> with File('_temporary.txt') as file:
    ...    file.append('One')
    ...    file.append('Two')
    ...    file.append('Three')
    ...    file.append('Four')
    ...    file.append('Five')
    ...    file.append('Six')

    >>> open('_temporary.txt').read()
    'One\\nTwo\\nThree\\nFour\\nFive\\nSix\\n'

    >>> remove('_temporary.txt')
"""

from threading import Timer


# type: Type
class File:
    filename: str
    _content: list[str]

    def __init__(self, filename):
        self.filename = filename
        self._content = list()

    def __enter__(self):
        return self

    def __exit__(self, *args):
        with open(self.filename, mode='w') as file:
            file.writelines(self._content)

    def append(self, line):
        self._content.append(line + '\n')


Code 8.39. Solution
"""
* Assignment: Concurrency Threading Subprocess
* Complexity: easy
* Lines of code: 20 lines
* Time: 21 min

English:
    TODO: English Translation
    X. Run doctests - all must succeed

Polish:
    1. Stwórz kolejkę `queue` do której dodasz różne polecenia systemowe do wykonania, np.:
        a. Linux/macOS: `['ls /tmp/', 'echo "test"', 'sleep 2']`,
        b. Windows: `['dir c:\\Windows', '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 = 1.0` sekundy
    4. Ilość poleceń może się zwiększać w miarę wykonywania zadania.
    5. Wątki mają być uruchomione w tle (ang. `daemon`)
    6. Uruchom doctesty - wszystkie muszą się powieść

:Extra task:
    1. Wątki powinny być uśpione za pomocą `Timer` przez `DELAY = 1.0` sekund, a następnie ruszyć do roboty
    2. 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
"""
import logging


TIMEOUT = 1.0
DELAY = 1.0
TODO = ['ls /tmp/',
        'echo "test"',
        'sleep 2']