Как реализовать индивидуальный многопроцессорный непрерывный (асинхронный) контроль в python?

Question

У меня есть два типа браузера: firefox (40 экземпляров) и хром (40 экземпляров) на моей сетке селена. Кроме того, у меня есть куча тестов, некоторые из них должны быть выполнены на firefox, некоторые из них под хром. Некоторым из них все равно.

Первое решение, которое было рекомендовано @skrrgwasme, состояло в том, чтобы разделить тесты, которые не нуждаются в конкретных браузерах в двух группах, чтобы иметь, наконец, две очереди (одна - для запуска на firefox, вторая для выполнения на хроме): How to implement custom control over python multiprocessing.Pool?

Это решение является приятным, но его можно улучшить: поскольку браузер обрабатывает свои запросы с разной скоростью, хром будет быстрее заканчивать свою очередь, а очередь firefox будет работать намного дольше. Это можно решить с помощью пользовательского непрерывного управления, когда я решаю, какой браузер использовать не до создания пулов, но когда этот пул уже запущен.

Итак, у нас должен быть один бассейн, где каждый процесс икры контролируется нами.

@skrrgwasme рекомендовал использовать для этого apply_async. Но я просто не могу понять, как это может быть достигнуто в python, так как это не async, как node.js.

Не могли бы вы привести несколько примеров? У меня есть очень крошечный опыт работы с Python, и, кажется, полностью застрял с этим :(

Answer 1

Я думаю, что проще всего каждый рабочий процесс потреблять из двух объектов Queue. Один из них - это браузер Queue, а другой общий «общий» Queue. Таким образом, вы можете использовать 40 процессов из Chrome Queue, а затем переключиться на общий Queue после его удаления, а также выполнить 40 процессов из Firefox Queue, а затем переключиться на общий Queue после его слива.Вот пример с использованием 8-процессов вместо 80:

from multiprocessing import Pool, Manager 
from Queue import Empty 
import time 

ff_tests = [1,2,3,4,5] 
chrome_tests = [10, 11, 12, 13, 14, 15] 
general_tests = [20, 21,22, 23,24,25] 

def process_func(spec_queue, general_queue, browser): 
    while True: 
     try: 
      test = spec_queue.get_nowait() 
      print("Processing {} in {} process".format(test, browser)) 
      time.sleep(2) 
     except Empty: 
      break 

    while True: 
     try: 
      test = general_queue.get_nowait() 
      print("Processing {} in {} process".format(test, browser)) 
      time.sleep(2) 
     except Empty: 
      break 


if __name__ == "__main__": 
    m = Manager() 
    ff_queue = m.Queue() 
    chrome_queue = m.Queue() 
    general_queue = m.Queue() 

    for queue, tests in [(ff_queue, ff_tests), (chrome_queue, chrome_tests), 
         (general_queue, general_tests)]: 
     for test in tests: 
      queue.put(test) 


    pool = Pool(8) 
    for _ in range(4): 
     pool.apply_async(process_func, args=(ff_queue, general_queue, "firefox")) 
     pool.apply_async(process_func, args=(chrome_queue, general_queue, "chrome")) 
    pool.close() 
    pool.join() 

Выход:

Processing 1 in firefox process 
Processing 10 in chrome process 
Processing 2 in firefox process 
Processing 11 in chrome process 
Processing 3 in firefox process 
Processing 12 in chrome process 
Processing 4 in firefox process 
Processing 13 in chrome process 
Processing 5 in firefox process 
Processing 14 in chrome process 
Processing 20 in firefox process 
Processing 15 in chrome process 
Processing 21 in firefox process 
Processing 22 in chrome process 
Processing 23 in firefox process 
Processing 24 in chrome process 
Processing 25 in chrome process 

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

Answer 2

Использование pool.apply_async можно рассматривать очень как вручную настройки каждого из вызовов, сделанных под капотом на map в вашем предыдущем вопросе. Вы просто добавляете все задачи в пул, и они взрываются через них, когда становятся доступными новые рабочие процессы. Вы можете использовать один и тот же подход на основе одной функции для браузера, который skrrgwasme suggested. Нижеприведенный код сильно зависит от его ответа на предыдущий вопрос :

from multiprocessing import Pool 

params = [1,2,3,4,5 ... ] 

def ff_func(param): 
    # Do FireFox stuff 

def ch_func(param): 
    # Do Chrome stuff 

pool = Pool(80) 

# For each parameter, add two tasks to the pool--one FF, one Chrome. 
for param in params: 
    pool.apply_async(ff_func, param) 
    pool.apply_async(ch_func, param) 

pool.close() 
pool.join() 

---

Что здесь происходит заключается в том, что вы создаете большую асинхронную очередь задач для пула для обработки. Затем пул обрабатывает все определенные задачи в любом порядке, который он считает нужным.

Обратите внимание: в отличие от предыдущего ответа это фактически не гарантирует максимальный размер пула 40 для каждого браузера, потому что вы просили лучше использовать наши ресурсы. Лучший способ использовать максимум 80 процессов состоит в том, чтобы они все работали все время, насколько это возможно.

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