Введение
Знаете, почему я решил написать эту статью? Я писал программу, где использовал потоки. Во время работы с ними в Python всё больше убеждаешь себя, что тут с ними всё плохо. Нет, не то, чтобы они плохо работали. Просто использовать их, мягко говоря, неудобно. Я решил написать простую, но более удобную библиотеку, и здесь поделюсь процессом.
P.S.: В конце оставлю ссылку на GitHub
Первые шаги
Первое, с чего я решил начать - это удобное создание потоков из функций. Для этого я написал простенький декоратор, выглядит он так:
# thr.py
from threading import Thread as thrd
__all__ = ['Thr']
# Класс для всего содержимого либы
class Thr:
# Собственно, декоратор
def thread(fn):
def thr(*args, **kwargs):
thrd(target = fn, args = (*args,), kwargs={**kwargs,}).start()
pass
return thr
pass
# конец файлаВот и всё. Теперь можно очень удобно создавать потоки:
from thr import Thr
# пример использования
@Thr.thread
def func1(a, b, c):
if a+b > c:
if a+c > b:
if b+c > a:
print("треугольник существует")
pass
pass
pass
print("треугольник не существует")
pass # возвращение значений пока не предусмотрено
for a, b, c in zip(range(1, 10), range(6, 15), range(11, 20)):
func1(a, b, c)
passУдобнее чем было, так ведь? Но мне стало мало.
Среды для потоков
Мне стало не удобно обеспечивать верное взаимодействий потоков. Например, мне надо чтобы 2 цикла решали последовательности чисел для 3n+1. Тогда, мне приходилось мучить Python глобальными переменными. И я нашел выход!
P.S.: Возможно, это и есть ThreadPoolExecutor, которым я никогда не пользовался. Если это так, то, возможно, просто кому-то мой метод покажется удобнее.
Давайте объясню. Вы создаёте среду для потоков(можно несколько) и немного адаптируете потоковые функции под себя. код библиотеки:
# thr.py
from curses.ascii import isalnum
from threading import Thread as thrd
from random import randrange
from sys import exit as exitall
__all__ = ['Thr']
# f-str не позволяет использовать "\" напрямую,
# пришлось выкручиваться =)
nl = "\n"
bs = "\b"
tb = "\t"
rt = "\r"
# просто полезная функция
def strcleanup(s: str = ""):
while s[0] == ' ': s = s[1:]
while s[-1] == ' ': s = s[:-1]
if not isalnum(s[0]): s = '_' + s
s = s.replace(' ', '_')
for i in range(len(s)):
if not isalnum(s[i]):
s = s.replace(s[i], '_')
pass
pass
s += f"{randrange(100, 999)}"
return s
# Класс для всего содержимого либы
class Thr:
# класс для сред потоков
class Env(object):
# поля
# потоки
thrs: list = None
# возвращаемые значения
rets: dict = None
# название среды
name: str = None
# методы
# инициализация
def __init__(self, name):
self.thrs = []
self.rets = {}
self.__name__ = self.name = name
# self.name на всякий случай.
# __name__ - магическая переменная, вдруг поменяется.
pass
# в строку
__str__ = lambda self:\
f"""ThreadSpace "{self.name}": {len(self.thrs)} threads"""
# тоже в строку, но скорее для дебага, чем для печати юзеру
__repr__ = lambda self:\
f"""ThreadSpace "{self.name}"
threads:
{(nl+" ").join(self.thrs)}
total: {len(self.thrs)}
"""
def __add__(self, other):
self.thrs = {**self.thrs, **other.thrs}
pass
# Декоратор/метод для добавления в список потоков.
def append(self, fn):
# функции нужен docstring
ID = strcleanup(fn.__doc__.casefold())
self.thrs += [ID]
self.rets[ID] = None
#
class Thrd(object):
ID = None
space = None
fn = None
thr = None
runned = None
ret = None
def __init__(slf, ID, self, fn):
slf.ID = ID
slf.space = self
slf.fn = fn
slf.thr = None
slf.runned = False
slf.ret = False
pass
def run(slf, *args):
if slf.runned:
print(f"Exception: Thread \"{slf.ID[:-3]}\" of threadspace \"{slf.space.name}\" already started")
exitall(1)
pass
slf.thr = thrd(target = slf.fn, args = (slf, slf.space, slf.ID, *args,))
slf.thr.start()
slf.runned = True
pass
def join(slf):
if not slf.runned:
print(f"Exception: Thread \"{slf.ID[:-3]}\" of threadspace \"{slf.space.name}\" not started yet")
exitall(1)
pass
slf.thr.join()
slf.runned = False
pass
def get(slf):
if not slf.ret:
print(f"Exception: Thread \"{slf.ID[:-3]}\" of threadspace \"{slf.space.name}\" didn`t return anything yet")
exitall(1)
pass
slf.runned = False
return slf.space.rets[slf.ID]
def getrun(slf, *args):
slf.run(*args)
slf.join()
return slf.get()
pass
return Thrd(ID, self, fn)
pass
# Декоратор для "голого" потока
def thread(fn):
def thr(*args, **kwargs):
thrd(target = fn, args = (*args,), kwargs={**kwargs,}).start()
pass
return thr
pass
# конец файлаПример вывода об ошибке:
Traceback (most recent call last):
File "test.py", line 37, in <module>
loop.run()
File "thr.py", line 93, in run
raise Exception(...)
Exception: Thread "3n_1_mainloop" of threadspace "3n+1" already startedКак быстро вырос объём кода по сравнению с предыдущим вариантом! Итак, пример использования:
from random import randint
from thr import Thr
Space = Thr.Env("3n+1")
@Space.append
def hdl(t, spc, ID, num):
"""3n+1_handle"""
if num % 2 == 0:
# значения возвращать так
t.ret = True
spc.rets[ID] = num/2
return
# значения возвращать так
t.ret = True
spc.rets[ID] = 3*num+1
return
@Space.append
def loop(t, spc, ID, num):
"""3n+1_mainloop"""
steps = 0
while num not in (4, 2, 1):
num = hdl.getrun(num)
steps += 1
pass
# значения возвращать так
t.ret = True
spc.rets[ID] = steps
return
print()
print(Space)
print()
print(repr(Space))
ticks = 0
num = randint(5, 100)
loop.run(num)
while not loop.ret:
ticks += 1
pass
print(f"loop reached 4 -> 2 -> 1 trap,\n"
f"time has passed (loop ticks):\n"
f"{ticks}, steps has passed: {loop.get()}, start number: {num}")Вывод:
ThreadSpace "3n+1": 2 threads
ThreadSpace "3n+1"
threads:
3n_1_handle840
3n_1_mainloop515
total: 2
loop reached 4 -> 2 -> 1 trap,
time has passed (loop ticks):
13606839, steps has passed: 33, start number: 78Итак, на этом и завершу эту статью. Спасибо за внимание!
