Ошибка завершения многопроцессорной обработки Python

Question

Я вижу это, когда нажимаю Ctrl-C для выхода из приложения

Error in atexit._run_exitfuncs:
Traceback (most recent call last):
  File "/usr/lib/python2.6/atexit.py", line 24, in _run_exitfuncs
    func(*targs, **kargs)
  File "/usr/lib/python2.6/multiprocessing/util.py", line 269, in _exit_function
    p.join()
  File "/usr/lib/python2.6/multiprocessing/process.py", line 119, in join
    res = self._popen.wait(timeout)
  File "/usr/lib/python2.6/multiprocessing/forking.py", line 117, in wait
    return self.poll(0)
  File "/usr/lib/python2.6/multiprocessing/forking.py", line 106, in poll
    pid, sts = os.waitpid(self.pid, flag)
OSError: [Errno 4] Interrupted system call
Error in sys.exitfunc:
Traceback (most recent call last):
  File "/usr/lib/python2.6/atexit.py", line 24, in _run_exitfuncs
    func(*targs, **kargs)
  File "/usr/lib/python2.6/multiprocessing/util.py", line 269, in _exit_function
    p.join()
  File "/usr/lib/python2.6/multiprocessing/process.py", line 119, in join
    res = self._popen.wait(timeout)
  File "/usr/lib/python2.6/multiprocessing/forking.py", line 117, in wait
    return self.poll(0)
  File "/usr/lib/python2.6/multiprocessing/forking.py", line 106, in poll
    pid, sts = os.waitpid(self.pid, flag)
OSError: [Errno 4] Interrupted system call

Я использую витую поверх моих собственных вещей,

Я зарегистрировал сигнал Ctrl-C со следующим кодом

    def sigHandler(self, arg1, arg2):
        if not self.backuped:
            self.stopAll()
        else:
            out('central', 'backuped ALREADY, now FORCE exiting')
            exit()


    def stopAll(self):
        self.parserM.shutdown()
        for each in self.crawlM:
            each.shutdown()
        self.backup()
        reactor.stop()

и когда они подают сигнал другим, чтобы они отключились, он пытается сказать им, чтобы они хорошо закрылись до

exit = multiprocessing.Event()
def shutdown(self):
    self.exit.set()

где все мои процессы в той или иной форме,

def run(self):
    while not self.exit.is_set():
        do something
    out('crawler', 'crawler exited sucessfully')

Есть идеи, что это за ошибка? Я получаю его только тогда, когда у меня есть несколько экземпляров определенного потока.

Answer 1

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

Проблема заключается в том, что многопроцессорная обработка блокирует ожидание до тех пор, пока дочерний процесс, которого он ожидает, не завершится. Однако, поскольку вы установили обработчик сигнала для SIGINT, и ваша программа получает этот сигнал, он прерывает системный вызов для выполнения вашего обработчика сигнала и завершает работу waitpid, указывая, что его прервал сигнал. Python обрабатывает этот случай исключениями.

В качестве обходного пути вы можете заключить поврежденные разделы в цикл while и попробовать / перехватить блоки вроде этого, где бы вы ни ожидали завершения потоков, или создать подкласс мультипроцессинга.

import errno
from multiprocessing import Process

p = Process( target=func, args=stuff )
p.start()
notintr = False
while not notintr:
  try:
    p.join() # "Offending code"
    notintr = True
  except OSError, ose:
    if ose.errno != errno.EINTR:
      raise ose

Чтобы разобраться с multiprocessing.forking.Popen, вам нужно сделать что-то вроде этого:

import errno
from multiprocessing import Process
from multiprocessing.forking import Popen
import os

# see /path/to/python/libs/multiprocessing/forking.py
class MyPopen(Popen):
  def poll(self, flag=os.WNOHANG): # from forking.py
    if self.returncode is None: # from forking.py
      notintr = False
      while not notintr:
        try:
          pid, sts = os.waitpid(self.pid, flag) # from forking.py
          notintr = True
        except OSError, ose:
          if ose.errno != errno.EINTR:
            raise ose
      # Rest of Popen.poll from forking.py goes here

p = Process( target=func args=stuff )
p._Popen = p
p.start()
p.join()

Answer 2

Я видел это, но оно ушло, когда я перегрузил обработчики сигналов своими собственными. Использовать реактор.run (installSignalHandlers = False) и определите свои собственные функции для SIGINT, SIGTERM и т. Д.