В настоящее время я играю с многопроцессорностью и очередями. Я написал фрагмент кода для экспорта данных из mongoDB, сопоставил его в реляционную (плоскую) структуру, преобразовал все значения в строку и вставлял их в mysql.
Каждый из этих шагов представляется в виде процесса и задает очереди импорта/экспорта, безопасные для экспорта mongoDB, которые обрабатываются в родительском.
Как вы увидите ниже, я использую очереди, а дочерние процессы завершаются, когда они читают "Нет" из очереди. Проблема, которую я имею в настоящее время, заключается в том, что если дочерний процесс запускается в необработанное Exception, это не распознается родителем, а остальные просто продолжают работать. То, что я хочу, это то, что весь shebang завершает работу и, в лучшем случае, ререйзирует ошибку ребенка.
У меня есть два вопроса:
- Как определить дочернюю ошибку в родительском?
- Как убить мои дочерние процессы после обнаружения ошибки (наилучшей практики)? Я понимаю, что поставить "Нет" в очередь, чтобы убить ребенка, довольно грязно.
Я использую python 2.7.
Вот основные части моего кода:
# Establish communication queues
mongo_input_result_q = multiprocessing.Queue()
mapper_result_q = multiprocessing.Queue()
converter_result_q = multiprocessing.Queue()
[...]
# create child processes
# all processes generated here are subclasses of "multiprocessing.Process"
# create mapper
mappers = [mongo_relational_mapper.MongoRelationalMapper(mongo_input_result_q, mapper_result_q, columns, 1000)
for i in range(10)]
# create datatype converter, converts everything to str
converters = [datatype_converter.DatatypeConverter(mapper_result_q, converter_result_q, 'str', 1000)
for i in range(10)]
# create mysql writer
# I create a list of writers. currently only one,
# but I have the option to parallellize it further
writers = [mysql_inserter.MySqlWriter(mysql_host, mysql_user, mysql_passwd, mysql_schema, converter_result_q
, columns, 'w_'+mysql_table, 1000) for i in range(1)]
# starting mapper
for mapper in mappers:
mapper.start()
time.sleep(1)
# starting converter
for converter in converters:
converter.start()
# starting writer
for writer in writers:
writer.start()
[... инициализация соединения mongo db...]
# put each dataset read to queue for the mapper
for row in mongo_collection.find({inc_column: {"$gte": start}}):
mongo_input_result_q.put(row)
count += 1
if count % log_counter == 0:
print 'Mongo Reader' + " " + str(count)
print "MongoReader done"
# Processes are terminated when they read "None" object from queue
# now that reading is finished, put None for each mapper in the queue so they terminate themselves
# the same for all followup processes
for mapper in mappers:
mongo_input_result_q.put(None)
for mapper in mappers:
mapper.join()
for converter in converters:
mapper_result_q.put(None)
for converter in converters:
converter.join()
for writer in writers:
converter_result_q.put(None)
for writer in writers:
writer.join()
