Тайм-аут многопроцессорности Python

Question

У меня есть программа, которая непрерывно получает данные из базы данных, затем обрабатывает данные и затем загружает их на сервер через FTP. Он выглядит так:

Проблема заключается в том, что загрузка ftp может зависеть вечно, что приведет к зависанию программы. Я хочу реализовать способ сделать загрузку в подпроцессах, установить тайм-аут для каждого из них, и, если какой-то один раз, просто убей его. Мне не нужно получать какие-либо данные из подпроцессов, и мне не нужно повторять попытку, если загрузка не удалась.

Но я не хочу делать .join(time), потому что основной процесс должен прекратить там ждать. Я просто хочу, чтобы основной процесс непрерывно извлекал данные, а затем выполнял подпроцесс fork для загрузки, он превысил тайм-аут, просто убил его, чтобы избежать зомби. Это все. Может ли кто-нибудь предложить архитектуру? Благодарю.

Answer 1

Для этого можно использовать Pebble.

from pebble.process import concurrent 

TIMEOUT_IN_SECONDS = 5 

feed = connect2database() 
for data in feed.changes(): 
    data = process(data) 
    concurrent(target=ftpUpload, args=(data,), timeout=TIMEOUT_IN_SECONDS) 

Answer 2

Одна хорошая основа, чтобы использовать для задачи, как это Luigi. Он разработан для зависимости данных, но вы можете легко установить тайм-аут для каждой задачи, и вы можете установить его, чтобы он не выполнял неудачные задачи. Here's оригинальный пример использования Луиджи.

Чтобы установить таймаут для каждой задачи, установите worker-timeout на количество секунд в секундах. http://luigi.readthedocs.org/en/stable/configuration.html

Чтобы установить Luigi, поэтому он не будет повторять задачи, установите disable_failures 0. http://luigi.readthedocs.org/en/stable/_modules/luigi/scheduler.html

Это займет хорошее количество работы, чтобы узнать, как луиджи работает, хотя и документация скудной, чем хотелось бы. Luigi Google Group - это то, где много вопросов/ответов luigi.