9.27. Threading Timer¶
9.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')
9.27.2. Assignments¶
"""
* 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():
...
"""
* 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')
"""
* 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']