Python Multiprocessing Async не может завершить процесс

Question

У меня бесконечный цикл, выполняющий async, но я не могу его прекратить. Ниже приведена аналогичная версия моего кода:

from multiprocessing import Pool 
test_pool = Pool(processes=1) 
self.button1.clicked.connect(self.starter) 
self.button2.clicked.connect(self.stopper) 

    def starter(self): 
     global test_pool 
     test_pool.apply_async(self.automatizer) 

    def automatizer(self): 
     i = 0 
     while i != 0 : 
      self.job1() 
      # safe stop point 
      self.job2() 
      # safe stop point 
      self.job3() 
      # safe stop point 

    def job1(self): 
     # doing some stuff 


    def job2(self): 
     # doing some stuff 


    def job3(self): 
     # doing some stuff 


    def stopper(self): 
     global test_pool 
     test_pool.terminate() 

Проблема с концом() внутри функции остановки не работает. Я попытался поместить terminate() внутри job1, job2, job3 функции все еще не работают, попытался положить конец цикла в стартовой функции, снова не работая. Как я могу остановить этот асинхронный процесс?

Хотя остановка процесса в любое время является достаточно хорошей, можно ли остановить ее в точках, которые я хочу? Я имею в виду, что если команда завершения (не уверенная о том, какая команда) предназначена для обработки, я хочу, чтобы она выполнила шаги к маркеру «# безопасной остановки», а затем завершала процесс.

Answer 1

Вы действительно должны избегать использования terminate() при нормальной работе. Его следует использовать только в необычных случаях, таких как висячие или невосприимчивые процессы. Обычный способ завершения пула процессов - позвонить pool.close(), а затем pool.join().

Эти методы требуют функции, которую ваш пул выполняет для возврата, а ваш вызов pool.join() блокирует ваш основной процесс до тех пор, пока он этого не сделает. Я хотел бы предложить вам добавить multiprocess.Queue, чтобы дать себе возможность рассказать свою подпроцесс для выхода:

# this import is NOT the same as multiprocessing.Queue - this is here for the 
# queue.Empty exception 
import Queue 

queue = multiprocessing.Queue() # not the same as a Queue.Queue() 

def stopper(self): 
    # don't need "global" keyword to call a global object's method 
    # it's only necessary if we want to modify a global 
    queue.put("Stop") 
    test_pool.close() 
    test_pool.join() 

def automatizer(self): 
    while True: # cleaner infinite loop - yours was never executing 
     for func in [self.job1, self.job2, self.job3]: # iterate over methods 
      func() # call each one 

      # between each function call, check the queue for "poison pill" 
      try: 
       if queue.get(block=False) == "Stop": 
        return 
      except Queue.Empty: 
       pass 

Поскольку вы не предоставили более полный пример кода, вы должны выяснить, где на самом деле создать экземпляр multiprocessing.Queue и как пройти мимо. Кроме того, комментарий от Janne Karila был правильным. Вы должны переключить свой код на использование одного Process вместо пула, если в любом случае вы используете только один процесс. Класс Process также использует метод блокировки join(), чтобы сообщить ему об окончании после его возврата. Единственный безопасный способ прекращения процессов в «известных безопасных точках» - это реализация какого-то межпроцессного взаимодействия, как я здесь делал. Трубы тоже работали бы.