Чтение стандартного вывода с подчиненных узлов с помощью ipcluster

Question

Я настроил кластер с помощью

ipcluster start --n=8

затем доступ к нему с помощью

from IPython.parallel import Client
c=Client()
dview=c[:]
e=[i for i in c]

Я запускаю процессы на подчиненных узлах (e[0]-e[7]), которые занимают много времени, и я бы хотел, чтобы они отправляли отчеты о ходе выполнения мастеру, чтобы я мог следить за тем, как далеко они проходят являются.

Есть два способа сделать это, но до сих пор я не смог реализовать ни один из них, несмотря на то, что многие часы пролистывали страницы с вопросами.

Либо я хочу, чтобы узлы отправляли некоторые данные обратно на мастер без запроса. т. е. в длинном процессе, который выполняется на узлах, я реализую функцию, которая через регулярные промежутки времени передает свой ход мастеру.

Или я мог бы перенаправить стандартный вывод узлов на мастер, а затем просто следить за ходом печати. Это то, над чем я работаю до сих пор. У каждого узла есть свой стандартный вывод, поэтому print ничего не делает, если работает удаленно. Я пытался протолкнуть sys.stdout к узлам, но это просто закрывает его.

Я не могу поверить, что я единственный человек, который хочет это сделать, поэтому, может быть, я упускаю что-то очень простое. Как я могу отслеживать длинные процессы, происходящие удаленно с помощью ipython?

Answer 1

stdout уже захвачен, зарегистрирован и отслежен, и поступает к Клиентам по мере поступления до того, как результат будет завершен.

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

В самом Клиенте вы можете проверить диктат метаданных для stdout / err (Client.metadata[msg_id].stdout) прежде чем результаты будут сделаны. использование Client.spin() сбрасывать любые входящие сообщения из сокетов zeromq, чтобы обеспечить актуальность этих данных.

Если вы хотите, чтобы стандартный вывод обновлялся часто, обязательно позвоните sys.stdout.flush() чтобы гарантировать, что поток действительно публикуется в этой точке, а не полагаться на неявные сбросы, которые могут не произойти, пока работа не будет завершена.