Запись в 1 переменную с двух потоков. Как правильно обработать очередь в python?

Question

[Кирилл Петров]

К примеру есть такая упрощённая конструкция с запуском 2х потоков и одновременной перезаписью одной переменной, в итоге образуется рандом в последней строчке вывода.

import threading as th

lenAllData = 1000000
queueData = []

def threadAddQueue():
    global queueData
    # Симулируем заполнение очереди
    for i in range(lenAllData):
        queueData.append(i)

def threadProcQueue():
    global queueData
    queueProceed = []
    # симулируем разбор очереди
    while True:
        if len(queueData) > 0:
            queueProceed = queueProceed + queueData
            queueData = []
            print(f'Length {len(queueProceed)}, last value {queueProceed[-1]}')

# Запускаем и получаем рандом в выводе
if __name__ == '__main__':
    th1 = th.Thread(target=threadAddQueue)
    th1.start()

    th2 = th.Thread(target=threadProcQueue)
    th2.start()

Пример вывода

Length 33021, last value 33020
Length 52147, last value 52147
Length 71771, last value 71772
Length 106174, last value 106175
Length 143665, last value 143667
Length 184920, last value 184922
Length 223584, last value 223587
Length 242001, last value 242004
Length 277876, last value 277879
Length 313757, last value 313761
Length 331471, last value 351468
Length 374527, last value 394524
Length 393301, last value 413299
Length 408631, last value 428629
Length 425010, last value 464092
Length 441459, last value 502028
Length 479035, last value 557692
Length 498091, last value 597529
Length 536072, last value 652986
Length 572724, last value 709981
Length 609640, last value 764660
Length 646480, last value 824379
Length 682897, last value 881066
Length 724371, last value 943361
Length 761276, last value 997796

И при этом нельзя ставить в паузу первый поток на долго, так как из за этого зависнет другая часть программы.
Как правильно обработать очередь не приостанавливая первый поток и не потерять данные? При этом мне нужен паттерн похожий на глобальные переменные и потоки (не asyncio)

Answer 1

[Кирилл Петров]

В общем решил так:

import threading as th
from collections import deque

lenAllData = 1000000

class ClassForQueue:
    queueData = deque()

def threadAddQueue():
    # Симулируем заполнение очереди
    print('Start added to Queue')
    for i in range(lenAllData):
        ClassForQueue.queueData.append(i)
    print('All added to Queue')


def threadProcQueue():
    # симулируем разбор очереди
    result = deque()
    while True:
        if len(ClassForQueue.queueData):
            result.append(ClassForQueue.queueData.popleft())
        if result[-1] == lenAllData - 1:
            print(f'final! Length {len(result)}, last value {result[-1]}')
            break

# Запускаем и получаем адекватные данные в выводе
if __name__ == '__main__':
    th1 = th.Thread(target=threadAddQueue)
    th1.start()

    th2 = th.Thread(target=threadProcQueue)
    th2.start()

Как выяснилось, очереди сильно тормозили потоки, так как мне не требуется шарить данные больше чем в 1 поток, то решил опробовать deque, стало работать ещё быстрее чем в исходном примере

Answer 2

[bbkmzzzz]

На помощь придет Queue

Comments

[Кирилл Петров]

Да, знаю про эту чтуковину, но как ей воспользоваться в данном случае? И я всё таки бы хотел получить паттерн схожий с глобальными переменными...

[Кирилл Петров]

Возможно можно как то использовать contextvars, но тоже пока не хватает ума... Разбираюсь

[bbkmzzzz]

Как пользоваться очередью?) Один поток добавляет, второй забирает, все просто)

import threading as th
from queue import Queue

lenAllData = 1000000


class ClassForQueue:
    queueData = Queue()


def threadAddQueue():
    # Симулируем заполнение очереди
    for i in range(lenAllData):
        ClassForQueue.queueData.put(i)


def threadProcQueue():
    # симулируем разбор очереди
    while True:
        data_from_queue = ClassForQueue.queueData.get()
        print(f'Length {ClassForQueue.queueData.qsize()}, last value {data_from_queue}')


# Запускаем и получаем рандом в выводе
if __name__ == '__main__':
    th1 = th.Thread(target=threadAddQueue)
    th1.start()

    th2 = th.Thread(target=threadProcQueue)
    th2.start()

P.S. global - нафик

[Кирилл Петров]

bbkmzzzz, вот, я написал похожий код как у вас:

Код с очередями и глобальными переменными XD

import threading as th
from queue import Queue

lenAllData = 1000000


class ClassForQueue:
    queueData = Queue()


def threadAddQueue():
    # Симулируем заполнение очереди
    for i in range(lenAllData):
        ClassForQueue.queueData.put(i)


def threadProcQueue():
    # симулируем разбор очереди
    while True:
        data_from_queue = ClassForQueue.queueData.get()
        print(f'Length {ClassForQueue.queueData.qsize()}, last value {data_from_queue}')


# Запускаем и получаем рандом в выводе
if __name__ == '__main__':
    th1 = th.Thread(target=threadAddQueue)
    th1.start()

    th2 = th.Thread(target=threadProcQueue)
    th2.start()

Только что в вашем коде, что в моём производительность оставляет желать лучшего... Как ускорить, что бы работало со скоростью первого кода в вопросе?

[bbkmzzzz]

Это плата за синхронизацию потоков + если убрать принты будет быстрее. У меня без принтов время работы уменьшилось с 12 секунд до 4.
Первый вариант работает быстрее, так как нет синхронизации, просто пишется в память. Работает быстро, но неправильно, или работает медленнее, но коррерктно.

[bbkmzzzz]

Только почитайте про queue.Queue.get(), там есть таймаут, выкидывающий исключение, если по его истечению очередь пуста

[Кирилл Петров]

bbkmzzzz, а у меня на ноуте где то полторы минуты без принтов... По сравнению с 1 сек в первом примере. Это очень критично... И как я понял это из за того что в очередь не могут поместиться новые данные, пока не обработается то что в ней есть. И из за этого висит первый поток, так как его очередь ждёт... И я вернулся к тому что было... В общем это не то :D

[Кирилл Петров]

bbkmzzzz, как вам такая производительность и все данные целы :D Можно добавить ещё 1 нолик)
Добавил ответ отдельно

[bbkmzzzz]

Кирилл Петров, Нет, это от того, что очередь блокируется для добавления в нее данных и для считывания. Пока один поток имеет доступ к ней, все остальные ждут.
Если такой вариант не устраивает, можно использовать примитивы синхронизации, они есть в модуле threading и нагородить свою высокопроизводительную очередь с преферансом и куртизанками. С другой стороны, зачем очередь и потоки, если все данные обрабатываются за 1 сек? Если получение данных из очереди идет как долгая по времени операция, и быстрее без нее, зачем вообще многопоточность? При каких-то мало-мальски значимых расчетах, время доступа к очереди будет ничтожным на фоне остального.

[Кирилл Петров]

bbkmzzzz, у меня в проекте бывают нагрузки до 32к запросов за 0.2 сек, система держит 10к запросов в сек. И если будут тормоза, будет висеть всё. Поэтому данные собираем сейчас, а обрабатываем по очереди потом.

Answer 3

[Владимир Коротенко]

Стандартное решение во всех языках это либо лок после получения данных, либо lockfree примитивы. В последнем проекте я использовал тупой поток для получения данных и опросники которые просто возвращают текущее значение. Это позволило поднять rps c жалких 5 до 400

Comments

[Кирилл Петров]

Тут я кладу данные вконце, забираю вначале. Поэтому блокировки вообще не понадобились. Наверное deque, это как вы сказали, lockfree примитив)