Как корректно работать с одним объектом при использовании Pool (multiprocessing)?

Question

[coderisimo]

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

Ниже синтетический упрощенный пример. Но он отражает в какой-то мере суть.

import random 
from multiprocessing import Pool

my_list = [1,2,3,4,5,6,7,8,9]
used = []

def test(i):
  indx =  random.randint(0,len(my_list)-1)
  while my_list[indx] in used: #ищу элемент, который ранее не использовался
    indx = random.randint(0,len(my_list)-1)
  used.append(my_list[indx]) #добавляю элемент в список используемых, чтобы избежать повторного использования
  print(my_list[indx]) #вывожу уникальный элемент на печать
  
with Pool(4) as p:
  p.map(test, [1,2,3,4,5,6,7])

например , я получаю :
9
7
6
6
3
1
4

две 6 подряд ((((( Многопоточность, черт бы ее побрал )))))
Спасибо!

Answer 1

[Ivan Yakushenko]

Потому что вы используете не потоки, а процессы. У каждого процесса своё собственное окружение, соответственно переменные из process_1 никак не пересекаются с переменными из process_2. Что-бы шарить данные между процессами нужно использовать Manager.

Вот пример простенького ротатора:

from multiprocessing import Pool, Manager


def rotator(data_list):
    data = data_list.pop(0)
    data_list.append(data)
    return data


def print_data(data_list):
    data = rotator(data_list)
    print(data)


if __name__ == "__main__":
    manager = Manager()
    data_list = manager.list()

    for x in range(5):
        data_list.append(x)
    
    with Pool(4) as pool:
        for _ in range(10):
            pool.apply_async(print_data, [data_list])
        pool.close()
        pool.join()

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

0
1
2
3
4
0
1
2
3
4

Только учтите, что если процессы, как и потоки, стартуют в определенном порядке еще не означает, что в таком-же порядке они и завершат работу. Т.е. вполне нормально что процесс, который запустился 4м отработал на N-мс быстрее и завершился первым, в итоге вывод может быть таким:

4
2
0
1

Comments

[coderisimo]

Спасибо за ответ.

А вообще, возможна несложная реализация подобного ? Есть my_list , из него процессы "одалживают" элементы. Если элемент уже одолжили, что для следующих процессов он недоступен? Может есть простые примеры? Я уже пару часов пытаюсь разобраться с этим и ..... В некоторых статьях пишут открытым текстом -
ребята , избегайте мультипроцессности и мультипоточности , везде, где только возможно! без поллитры не разберетесь!

[Ivan Yakushenko]

coderisimo, обновил ответ.

[Дмитрий Темников]

coderisimo, волков бояться в лес не ходить!!! официальная документация модуля multiprocessing написана вполне доступным языком, а если читать всякие такие статьи, то наверное стоит подумать, а надо ли оно Вам и Ваше ли это всё, может поллитра действително эффективнее :-) шутка, ответ ниже

[coderisimo]

Ivan Yakushenko, спасибо!

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

Буду изучать ваш и код

[Ivan Yakushenko]

coderisimo,

Когда задача завершается

Нет.
1. Удаляем элемент из списка и сохраняем во временную переменную функцией pop()
2. Сразу добавляем его в конец списка функцией append()
3. Отдаем элемент в вызывающую функцию.

Таким образом гарантируем, что процессы не будут использовать 1 и тот-же элемент, если конечно элементов больше, чем процессов, но даже так по крайней мере они не будут использоваться подряд.

[Ivan Yakushenko]

coderisimo, но это тонкости. Главное, что вы должны понимать - manager.list() создает самый обычный список, который вы можете шарить между процессами. Т.е. все, что вам нужно - это передать его в качестве аргумента нужной функции. Да, иногда возникает лапша с передачей подобных аргументов от функции к функции, что-бы достигнуть "цели", но это условность неуместного использования. Лучше делать подобные решения асинхронно, а если нужен именно процесс для обработки - запускать его в ProcessPoolExecutor.

[coderisimo]

Ivan Yakushenko, спасибо буду разбираться.

[Ivan Yakushenko]

coderisimo, если вам нужно просто достать элемент из списка, что-то с ним сделать, а затем вернуть - вы все так же можете воспользоваться pop(). Если элемента не будет в списке, то и взять его процесс не сможет =)
Только учтите, что если в какой-то момент список окажется пустым, а какой-то процесс попытается взять из него элемент с pop(), то будет ошибка IndexError.

[coderisimo]

Ivan Yakushenko,

как будет время, можете пояснить, что происходит?

from multiprocessing import Pool, Manager
import random, time

my_list = [1,2,3,4,6,7,8,9,10,11]

def rotator(used):
    index = random.randint(0,len(my_list)-1)
    while my_list[index] in used: #данный лист один для всех, там актуальна инфа
      index = random.randint(0,len(my_list)-1)
    used.append(my_list[index])   
    return my_list[index]


def print_data(used):
    data = rotator(used)
    print(data)


if __name__ == "__main__":
    manager = Manager()
    used = manager.list() # создаем лист used, который будет доступен всем процессам
    
    with Pool(4) as pool:
        for _ in range(10):
            pool.apply_async(print_data, [used]) # передаем лист used  в ф-цию для каждого процесса
        pool.close()
        pool.join()

однако , значения двоятся. Возможно , просто использовать общий лист недостаточно, а нужно как-то гарантировать, что когда его изменяет ОДИН процесс, все другие не имеют к нему доступа.

[Ivan Yakushenko]

coderisimo, у вас алгоритм недееспособный. Посмотрите на вот этот кусок кода:

def rotator(used):
    index = random.randint(0,len(my_list)-1)
    while my_list[index] in used: 
      index = random.randint(0,len(my_list)-1)

И хорошенько подумайте что тут происходит. Помогу вам: у вас запускается 4 процесса, в теории все эти 4 процесса могут выбрать одно и тоже значение и оно будет валидным, т.к. на данном этапе в used оно отсутствует, и когда все эти 4 процесса уже взяли какое-то значение только после этого оно добавляется в used. Я показал вам пример как производить ротацию, что бы гарантировать то, что 2 процесса не будут обращаться к одному и тому-же объекту, нужно что бы этого объекта просто не было в списке. Вы можете играться с Lock, но тогда конкретно в вашем случае поток выполнения будет блокироваться пока 1 процесс не закончит операцию и только затем начнет обрабатывать следующий процесс. Абсурд, т.к. по сути вы превратите многопроцессорный-параллельный скрипт в многопроцессорный-последовательный.

[coderisimo]

Ivan Yakushenko, кажется я понял.

[Ivan Yakushenko]

coderisimo, да, все так. Как я и сказал в теории все запущенные процессы в один момент могут получить одинаковые значения и чем мощнее компьютер, тем эта вероятность больше. Т.е. если все процессы запустятся ровно в одном и тоже время вплоть до миллисекунды, то они 100% получат одинаковое значение.

[coderisimo]

Ivan Yakushenko,

мне кажется, я написал штуку , которая делает что нужно - одалживает элемент из набора друзей и возвращает , когда задача завершена .

import rotator, time, random
from multiprocessing import Pool, Manager

friends = ['Vasya', 'Petya', 'Kolya', 'Sasha', 'Ivan', 'Alex', 'Semen', 'Goga Aka Gosha']


def drinking_with_friend(friends_list):
    data = friends_list.pop(0)
    interval = random.random() * 10
    print('Start Drinking with {:>20}   time for drinking (duration {} )'.format(data, interval))
    time.sleep(interval)
    friends_list.append(data)
    print('Finish Drinking with {:>19}   time for drinking (duration {} )'.format(data, interval))


if __name__ == "__main__":
    manager = Manager()
    friends_list = manager.list(friends)

    with Pool(4) as pool:
        for _ in range(10):
            pool.apply_async(drinking_with_friend, [friends_list])
        pool.close()
        pool.join()

exit()

каждый друг ангажируется на выпивку. когда он ангажирован - его уже нельзя пригласить, ибо занят. когда освобождается, его опять можно пригласить. time.sleep(interval) - создает вариации во времени выполнения. В результате видно что каждый процесс идет своим ходом. Например выпивка с Иваном затянулась duration 9.88406377301575 - его не ангажировали повторно, ибо был недоступен. А вот Петя пил быстро , соответственно смог принять участие в двух пьянках ))

Start Drinking with                Vasya   time for drinking (duration 5.154979043445119 )
Start Drinking with                Petya   time for drinking (duration 0.0003130800248030141 )
Start Drinking with                Kolya   time for drinking (duration 1.833550237898618 )
Finish Drinking with               Petya   time for drinking (duration 0.0003130800248030141 )
Start Drinking with                Sasha   time for drinking (duration 6.215763259101572 )
Start Drinking with                 Ivan   time for drinking (duration 9.88406377301575 )
Finish Drinking with               Kolya   time for drinking (duration 1.833550237898618 )
Start Drinking with                 Alex   time for drinking (duration 3.848374725013499 )
Finish Drinking with               Vasya   time for drinking (duration 5.154979043445119 )
Start Drinking with                Semen   time for drinking (duration 9.993493984622905 )
Finish Drinking with                Alex   time for drinking (duration 3.848374725013499 )
Start Drinking with       Goga Aka Gosha   time for drinking (duration 1.5210977154160177 )
Finish Drinking with               Sasha   time for drinking (duration 6.215763259101572 )
Start Drinking with                Petya   time for drinking (duration 0.33348320324053105 )
Finish Drinking with               Petya   time for drinking (duration 0.33348320324053105 )
Start Drinking with                Kolya   time for drinking (duration 2.7868309945262726 )
Finish Drinking with      Goga Aka Gosha   time for drinking (duration 1.5210977154160177 )
Finish Drinking with               Kolya   time for drinking (duration 2.7868309945262726 )
Finish Drinking with                Ivan   time for drinking (duration 9.88406377301575 )
Finish Drinking with               Semen   time for drinking (duration 9.993493984622905 )

[Ivan Yakushenko]

coderisimo, ну да, всё правильно, как я изначально и писал. Если вы не хотите что-бы один и тот-же элемент вообще когда-либо повторялся, то можете обратно в общий список не добавлять. Тогда в какой-то момент список останется пустым, а оставшиеся процессы при попытке выполнить pop() получат IndexError и умрут, ну лучше так не делает и обрабатывать исключение и нормально завершать процесс.

[Ivan Yakushenko]

coderisimo, если что, то примерно так:

def drinking_with_friend(friends_list):
    try:
        data = friends_list.pop(0)
    except IndexError:
        print('There are no more sober friends')
        return False
    interval = random.random() * 10
    print('Start Drinking with {:>20}   time for drinking (duration {} )'.format(data, interval))
    time.sleep(interval)
    print('Finish Drinking with {:>19}   time for drinking (duration {} )'.format(data, interval))

И вывод:

Start Drinking with                Vasya   time for drinking (duration 0.525207635674314 )
Start Drinking with                Petya   time for drinking (duration 8.101208126205927 )
Start Drinking with                Kolya   time for drinking (duration 5.656888553205503 )
Start Drinking with                Sasha   time for drinking (duration 9.985692547228792 )
Finish Drinking with               Vasya   time for drinking (duration 0.525207635674314 )
Start Drinking with                 Ivan   time for drinking (duration 6.944745503696382 )
Finish Drinking with               Kolya   time for drinking (duration 5.656888553205503 )
Start Drinking with                 Alex   time for drinking (duration 8.269980798554679 )
Finish Drinking with                Ivan   time for drinking (duration 6.944745503696382 )
Start Drinking with                Semen   time for drinking (duration 7.068649522212147 )
Finish Drinking with               Petya   time for drinking (duration 8.101208126205927 )
Start Drinking with       Goga Aka Gosha   time for drinking (duration 5.178367620321175 )
Finish Drinking with               Sasha   time for drinking (duration 9.985692547228792 )
There are no more sober friends
There are no more sober friends
Finish Drinking with      Goga Aka Gosha   time for drinking (duration 5.178367620321175 )
Finish Drinking with                Alex   time for drinking (duration 8.269980798554679 )
Finish Drinking with               Semen   time for drinking (duration 7.068649522212147 )

[coderisimo]

Ivan Yakushenko, здесь речь идет о прокси. т.е они берутся из списка и после выполнения запроса туда же и возвращаются. единственное что требуется - не использовать ОДИН и тот же прокси сразу для двух запросов.
Так что очищать список не требуется.

[Ivan Yakushenko]

coderisimo, я просто на всякий случай варианты применения. Но лучше не делайте ротацию прокси на процессах - ресурсы в никуда.

[coderisimo]

Ivan Yakushenko, лучше в потоках ?

[Ivan Yakushenko]

coderisimo, трудно сказать не зная что именно вы делаете. Если вы делаете просто ротатор, что-бы был, то да лучше в потоках, если работаете с сетью - асинхронно, но тогда принцип немного другой - используйте Queue. Но только ни в коем случае не мешайте всё в одно. Т.е. если у вас есть функция, которая по сети получает информацию - выполняйте её асинхронно и если затем вы хотите эту информацию как-то обработать - саму обработку лучше делать в отдельном процессе.

[coderisimo]

Ivan Yakushenko,
Вот я и хотел создать некий класс. Его экземпляр получает список прокси, сколько процессов запускать ,список урлов и коллбэк для обработки полученного.
Далее он создает пулл запросов. У каждого свой прокси (он берется из списка , и используется только одним запросом в единицу времени). Каждый запрос получает респонс, передает данные в коллбэк (тот, например ,парсит и сохраняет нужную информацию в БД). После завершения задачи, старый прокси возвращается "на склад" , берется новый , берется новый урл и все повторяется. Все это будет делаться в ф-ции аналоге drinking_with_friend . Выглядит не слишком сложно. Хотя , как всегда бывает - это лишь иллюзия )))

[Ivan Yakushenko]

coderisimo, в общем ваша задача - написать парсер? Т.е. заходите на сайт и ищите там нужную информацию? В идеале это должно быть так:
1. Асинхронно загружаем URL'ы. Результат загрузки отдаем в очередь (Queue) на обработку.
2. Как только в очереди есть новая загрузка - запускаем процесс, которые будет обрабатывать загруженные данные. Дальше результат обработки помещаем в другую очередь на запись.
3. Как только в очередь на запись поступает новый элемент - записываем его асинхронно.

Раз в идеале загрузка должна быть асинхронной, то и ротатор тоже. Но это все в идеале, на деле прийти к подобному решению самому и сразу - практически невозможно, я так думаю. Возможно позже (не сегодня) я на примере какого-то условного сайта покажу пример подобного парсера. Если конечно я понял верно постановку задачи.

[coderisimo]

Ivan Yakushenko, На данном этапе я попробую поместить всю логику по обработки одного урла в один процесс.
Их не требуется много (не 100 и не более). В базу же можем писать не заботясь о доступе? Какие профиты усложнения архитектуры? Более четко разбить блоки кода по выполняемой работе ? Если выигрыш в производительности незначительный , то в данным момент наверное не стоит свеч. 1 час парсить или 1:20 не принципиально.

[Ivan Yakushenko]

coderisimo, сколько процессов собираетесь запускать? Какой стек (бибилотеки/фреймворки) собираетесь использовать? Тут ситуация такая, что нужно знать тонкости. В 1 случае из 10 вы ничего не измените, ничего не получите, в другом случае из 10 вы все только усложните и вообще упретесь в стенку, но в 8 случаев из 10 правильная архитектура - это минимальное потребление ресурсов, максимальная скорость работы скрипта в целом и возможность масштабирования любого этапа работы скрипта.

[coderisimo]

Ivan Yakushenko, процессов 10, requests + суп (BeautifulSoup4) . Больше ничего. Например мне нужно парсить ежедневно 10.000 товаров на амазоне. Это не слишком монструозно ))). В прошлом , для более примитивных задач, я просто запускал несколько экземпляров Selenium и парсил часа по 4 ))). Сейчас хочу немного улучшить скиллы в данной области ))

[Ivan Yakushenko]

coderisimo, нууууууууу с натяжечкой окей, допустим. Но только если в какой то момент у вас встанет задача парсить не 10к товаров, а 50-100к и желательно не за день, а за несколько часов (хотя бы), то перед вами встанет дилемма - или платить сотни долларов в месяц за инстанс в условном google cloud, который будет вытягивать сотни параллельных процессов, или делать нормальную архитектуру.
В итоге для именно этой задачи решение в качестве ознакомления допустимо, но зазора на будущее нет.