Как передавать данные между потоками на Python?

Question

[Николай]

Искал в интернете решения, с использованием Value, Arrays. Не сильно разобрался, по этому прошу помощи.
Идея такая. Подбрасываем монетку много раз и смотрим, сколько раз выпал орёл и сколько раз - решка. Сначала проблема была в том, что если бросать монетку 10млн раз, и для статистики повторить это 10 раз, то программа выполнялась долго. Почитал информацию про потоки, написал код, где монетка подбрасывается 10-ю потоками в каждом по 10млн раз. Было быстрее раза в 3.5. Но, предположим, мне надо подбросить монетку не 10млн раз, а 100млн раз. В один поток это выполняется долго(около 25 минут). В 4 потока по-идее около 7 минут. Но в 4 потока я получаю данные монетки, подброшенной по 25млн. раз и так 4 раза. Как можно передавать информацию между потоками, пробовал через текстовый файл. Но работает не всегда корректно. Иногда не отрабатывает в полном объёме. Т.е. кол-во бросков меняется (25млн/50млн/75млн/100млн), я догадываюсь, почему так происходит, но из-за того, что плохо разбираюсь в теме программирования, не уверен.
Начал учиться Python недавно, для общего развития, так что, вероятно, что код ужасен. Прошу поправлять, дабы я мог получить больше знаний на эту тему.

Сам код:

import random
import os
import time
from multiprocessing import Process

def rnd(kolvo):
    start_time = time.time() # Сохраняем время
    o = 0 # Количество "орлов"
    r = 0 # Количество "решек"
    kol = 0 # Всего бросков
    proc = os.getpid() # Получаем номер процесса, для отладки
    while kol != kolvo:
        # Кидаем монетку
        tmp = random.randint(0, 1)
        if tmp == 0:
            o = int(o) + 1
        if tmp == 1:
            r = int(r) + 1
        kol = kol + 1
    # Читаем из файла информацию о бросках и монетках
    file = open('test.txt', 'r', encoding='utf-8')
    read = file.readlines()
    file.close()
    # Перезаписываем файл с учетом полученной информации
    file = open('test.txt', 'w', encoding='utf-8')
    file.write(str(int(read[0]) + kol) + '\n' + str(int(read[1]) + o) + '\n' + str(int(read[2]) + r))
    file.close()
    # Для отладки
    print('Номер процесса - ' + str(proc))
    print("--- %s секунд ---" % (time.time() - start_time))

if __name__ == '__main__': # Не понял зачем эта строка, но так работает
    start_time = time.time() # Сохраняем время
    # Создаём файл со значениями 0\n0\n0 что бы при считывании получить list
    file = open('test.txt', 'w', encoding='utf-8')
    file.write('0\n0\n0\n')
    file.close()
    kolvo = 2500000 # Количество бросков для каждого процесса

    # Инициализируем
    proc1 = Process(target=rnd, args=(kolvo,))
    proc2 = Process(target=rnd, args=(kolvo,))
    proc3 = Process(target=rnd, args=(kolvo,))
    proc4 = Process(target=rnd, args=(kolvo,))

    # Стартууууууем
    proc1.start()
    proc2.start()
    proc3.start()
    proc4.start()
    # Для отладки
    #print("--- %s секунд --- Процессы стартанули!" % (time.time() - start_time))

    # Заканчиваем
    proc1.join()
    proc2.join()
    proc3.join()
    proc4.join()

    # Считываем данные из файла и выводим их на экран
    file = open('test.txt', 'r', encoding='utf-8')
    read = file.readlines()
    file.close()
    print('% выпадения решки ' + str(int(read[2])*100/int(read[0])))
    print('% выпадения орла ' + str(int(read[1])*100/int(read[0])))
    print('Всего бросков ' + str(int(read[0])))

    print("--- %s секунд --- ВЕСЬ КОД!" % (time.time() - start_time)) # Время выполнения всего кода

Answer 1

[bbkmzzzz]

if __name__ == '__main__': # Не понял зачем эта строка, но так работает

При запуске файла на исполнение интерпретатор создает переменные окружения, для того файла, который запустили переменная __name__ == '__main__'. Если файл импортировали, в __name__ будет относительный путь к модулю, и в блок кода после if управление не попадет. (при импорте файл исполняется)

multiprocessing.Queue. Это синхронизированная очередь.
Создайте экземпляр и отдайте его каждому потоку. Поток пусть добавляет каждую итерацию расчета в список, а список потом отдает в очередь. Как только все потоки завершены, пробегаете по очереди, обрабатываете списки, делаете с ними что хотите.

Comments

[Николай]

Спасибо, а можете привести какой-нибудь коротенький пример по multiprocessing.Queue
Т.к. мне кажется, что у меня некорректно установился модуль multiprocess, и queue некорректно работает. При попытке использовать kolvo.put(kol) мне пишется, что AttributeError: 'int' object has no attribute 'put'.

[bbkmzzzz]

В комментариях к коду написал

import random
import os
import time
from multiprocessing import Process, Queue


def rnd(kolvo, resultQ):
    start_time = time.time() # Сохраняем время
    o = 0 # Количество "орлов"
    r = 0 # Количество "решек"
    kol = 0 # Всего бросков
    proc = os.getpid() # Получаем номер процесса, для отладки

    while kol != kolvo:
        # Кидаем монетку
        tmp = random.randint(0, 1) # возвращает всегда int
        if tmp == 0: # у нас всего два возможных варианта. (можно еще сократить, написав if tmp:, при 0 будет False)
            o += 1 # короткая форма записи, аналог x = x + 1
        else:
            r += 1
        kol += 1

    resultQ.put((o, r)) # Кладем в очередь объект (кортеж)
    # Для отладки
    print('Номер процесса - ' + str(proc))
    print("--- %s секунд ---" % (time.time() - start_time))


if __name__ == '__main__': # Не понял зачем эта строка, но так работает
    start_time = time.time() # Сохраняем время
    resultQueue = Queue()  # создаем очередь
    kolvo = 2500000  # Количество бросков для каждого процесса
    procNum = 5  # количество воркеров
    processes = []  # список с процессами, дабы иметь к ним доступ, и что бы их не удалил сборщик мусора

    # Инициализируем (и сразу стартуем)
    for _ in range(procNum): #  переменная _ используется как и обычные, но по соглашению именно _ означает, что она
                             #  использоваться не будет
        proc = Process(target=rnd, args=(kolvo, resultQueue))
        processes.append(proc)
        proc.start()

    # Для отладки
    #print("--- %s секунд --- Процессы стартанули!" % (time.time() - start_time))

    #  Заканчиваем (не заканчиваем, а ожидаем завершения) пока не завешится процесс к которого м ждем, выполнение останавливается
    for proc in processes:
        proc.join()

    o = 0
    r = 0
    for _ in range(resultQueue.qsize()):
        queueItem = resultQueue.get() # читаем из очереди
        o += queueItem[0]
        r += queueItem[1]



    # Считываем данные из файла и выводим их на экран
    # file = open('test.txt', 'r', encoding='utf-8')
    # read = file.readlines()
    # file.close()
    print('%% выпадения решки = %s' % (r / (kolvo * procNum) * 100))
    print('%% выпадения орла {0}'.format(o / (kolvo * procNum) * 100))
    print('Всего бросков %s' % (kolvo * procNum))

    print("--- %s секунд --- ВЕСЬ КОД!" % (time.time() - start_time)) # Время выполнения всего кода

[bbkmzzzz]

По очередям:
есть еще модуль queue, и там тоже есть Queue. Но multiprocessing содержит очередь, адаптированную под процессы.
тыц

Queue.put() выкинет исключение, если очередь полная и вышел таймаут (по умолчанию None, то есть ждет все время, пока там что-то не появится).
Queue.put_nowait() выкинет исключение сразу, если очередь полная, таймаут не используется

Queue.get() выкинет исключение, если очередь пустая и вышел таймаут
Queue.get_nowait() аналогично put_nowait

[Николай]

bbkmzzzz, Спасибо, вроде, более-менее понял.

Answer 2

[Сергей Тихонов]

* Во-первых, нормально работать в несколько потоков Вам помешает GIL
* Во-вторых, синхронизировать все результаты не обязательно, вам в конце ведь только счетчики нужны
* Смотрите модуль multiprocessing, он позволяет выполнять методы в отдельных процессах
* В каждом процессе считаете свою статистику, потом получаете результаты в главном процессе и суммируете.

Comments

[Николай]

1. По этому я использую модуль multiprocessing, как я понял, он спасает от GIL
2. Я и не писал, что хочу их синхронизировать. Да, счётчики меня устроят, только они не работают между процессами
3. Да, я могу выполнить методы между процессами. Т.е. в 4-ёх процессах одновременно выполнять разные задачи, но данные внутри процесса и остаются внутри процесса, а мне надо данные как-то вытащить.
4. По моей логике так и должно быть, только я не понимаю, каким образом я могу из процесса передать данные в главный процесс.

[bbkmzzzz]

процессы != потоки.
multiprocessing создает экземпляр интерпретатора со своим GIL

[Николай]

bbkmzzzz, Так, т.е., как я понял, процесс - это отдельный кусок памяти, который выделяется под конкретную функцию.
А поток - это просто один из способов выполнения функции. Несколько потоков будут действовать на одном и том же процессе.

[bbkmzzzz]

процесс - это отдельный процесс операционной системы. Как если вы запустите несколько калькуляторов, так тут запускаются несколько копий python. Запустите свой скрипт, и посмотрите в системном диспетчере задач. Увидите python в количестве основной + кол-во воркеров.
Потоки существуют внутри процесса, и интерпретатор со своим процессом у них один. При завершении процесса, завершаются все его потоки. Потоки имеют общую с родительским процессом память и контекст.