반응형
파이썬에서 멀티 프로세싱을 사용하는 동안 어떻게 로그인해야합니까?
현재 Python 2.6 multiprocessing모듈을 사용하여 여러 프로세스를 생성하는 프레임 워크에 중앙 모듈이 있습니다. 를 사용하기 때문에 multiprocessing모듈 수준의 다중 처리 인식 로그가 LOG = multiprocessing.get_logger()있습니다. 당 워드 프로세서 ,이 로거는 제대로 해석 것들에하지 않도록 프로세스 공유 잠금을 가지고 sys.stderr동시에 그것을 쓰기 여러 프로세스를함으로써 (또는 무엇이든 핸들).
내가 가진 문제는 프레임 워크의 다른 모듈이 다중 처리를 인식하지 못한다는 것입니다. 내가 보는 방식으로,이 중앙 모듈에 대한 모든 종속성이 다중 처리 인식 로깅을 사용하도록해야합니다. 그것은 프레임 워크의 모든 클라이언트 에게는 물론 프레임 워크 내 에서 성가신 일입니다. 내가 생각하지 않는 대안이 있습니까?
방해하지 않고 이것을 처리하는 유일한 방법은 다음과 같습니다.
- 각 작업자 프로세스를 생성하여 로그가 다른 파일 디스크립터 (디스크 또는 파이프) 로 이동하도록 하는 것이 가장 좋습니다. 모든 로그 항목의 타임 스탬프를 지정해야합니다.
- 그런 다음 컨트롤러 프로세스는 다음 중 하나 를 수행 할 수 있습니다 .
- 디스크 파일을 사용하는 경우 : 실행이 끝날 때 로그 파일을 타임 스탬프별로 정렬하여 병합 합니다.
- 파이프를 사용하는 경우 (권장) : 모든 파이프에서 중앙 로그 파일로 즉시 로그 항목을 병합합니다. (예 :
select파이프의 파일 디스크립터에서 주기적 으로 사용 가능한 로그 항목에 대해 병합 정렬을 수행하고 중앙 로그로 플러시하십시오. 반복하십시오.)
방금 파이프를 통해 모든 것을 부모 프로세스에 공급하는 로그 핸들러를 작성했습니다. 나는 그것을 10 분 동안 만 테스트했지만 꽤 잘 작동하는 것 같습니다.
( 참고 : 이 하드 코딩되어 RotatingFileHandler내 유스 케이스입니다.)
업데이트 : @javier는 이제 Pypi에서 사용 가능한 패키지 로이 접근법을 유지합니다-https://github.com/jruere/multiprocessing-logging에서 Pypi의 다중 처리 로깅 , github 참조
업데이트 : 구현!
이제는 동시성을 올바르게 처리하기 위해 큐를 사용하고 오류를 올바르게 복구합니다. 나는 이것을 몇 달 동안 프로덕션 환경에서 사용 해 왔으며 아래의 현재 버전은 문제없이 작동합니다.
from logging.handlers import RotatingFileHandler
import multiprocessing, threading, logging, sys, traceback
class MultiProcessingLog(logging.Handler):
def __init__(self, name, mode, maxsize, rotate):
logging.Handler.__init__(self)
self._handler = RotatingFileHandler(name, mode, maxsize, rotate)
self.queue = multiprocessing.Queue(-1)
t = threading.Thread(target=self.receive)
t.daemon = True
t.start()
def setFormatter(self, fmt):
logging.Handler.setFormatter(self, fmt)
self._handler.setFormatter(fmt)
def receive(self):
while True:
try:
record = self.queue.get()
self._handler.emit(record)
except (KeyboardInterrupt, SystemExit):
raise
except EOFError:
break
except:
traceback.print_exc(file=sys.stderr)
def send(self, s):
self.queue.put_nowait(s)
def _format_record(self, record):
# ensure that exc_info and args
# have been stringified. Removes any chance of
# unpickleable things inside and possibly reduces
# message size sent over the pipe
if record.args:
record.msg = record.msg % record.args
record.args = None
if record.exc_info:
dummy = self.format(record)
record.exc_info = None
return record
def emit(self, record):
try:
s = self._format_record(record)
self.send(s)
except (KeyboardInterrupt, SystemExit):
raise
except:
self.handleError(record)
def close(self):
self._handler.close()
logging.Handler.close(self)
또 다른 대안은 logging패키지의 파일 기반이 아닌 다양한 로깅 핸들러 일 수 있습니다 .
SocketHandlerDatagramHandlerSyslogHandler
(다른 사람)
이렇게하면 안전하게 쓸 수 있고 결과를 올바르게 처리 할 수있는 로깅 데몬을 쉽게 가질 수 있습니다. (예를 들어, 메시지를 피클 링하고 자체 회전 파일 핸들러로 내보내는 간단한 소켓 서버)
은 SyslogHandler너무, 당신이 알아서 것입니다. 물론 syslog시스템 인스턴스가 아닌 자체 인스턴스를 사용할 수 있습니다 .
python logging cookbook에는 두 가지 완전한 예제가 있습니다. https://docs.python.org/3/howto/logging-cookbook.html#logging-to-a-single-file-from-multiple-processes
그것은 사용 QueueHandler에서 (내가 파이썬 2.7에서 자신처럼) 파이썬 3.2의 새로운하지만 자신의 코드에 복사하기 쉬운 인 : https://gist.github.com/vsajip/591589
각 프로세스는에 로깅을 Queue기록한 다음 listener스레드 또는 프로세스 (예 : 각각에 대해 하나의 예제가 제공됨)에서 해당 프로세스를 선택하여 모두 파일에 기록합니다. 손상이나 손상의 위험이 없습니다.
다음은 Google에서 온 다른 사람 (예 : 나와 같은)의 단순성에 중점을 둔 또 다른 솔루션입니다. 로깅이 쉬워야합니다! 3.2 이상 만.
import multiprocessing
import logging
from logging.handlers import QueueHandler, QueueListener
import time
import random
def f(i):
time.sleep(random.uniform(.01, .05))
logging.info('function called with {} in worker thread.'.format(i))
time.sleep(random.uniform(.01, .05))
return i
def worker_init(q):
# all records from worker processes go to qh and then into q
qh = QueueHandler(q)
logger = logging.getLogger()
logger.setLevel(logging.DEBUG)
logger.addHandler(qh)
def logger_init():
q = multiprocessing.Queue()
# this is the handler for all log records
handler = logging.StreamHandler()
handler.setFormatter(logging.Formatter("%(levelname)s: %(asctime)s - %(process)s - %(message)s"))
# ql gets records from the queue and sends them to the handler
ql = QueueListener(q, handler)
ql.start()
logger = logging.getLogger()
logger.setLevel(logging.DEBUG)
# add the handler to the logger so records from this process are handled
logger.addHandler(handler)
return ql, q
def main():
q_listener, q = logger_init()
logging.info('hello from main thread')
pool = multiprocessing.Pool(4, worker_init, [q])
for result in pool.map(f, range(10)):
pass
pool.close()
pool.join()
q_listener.stop()
if __name__ == '__main__':
main()
로깅 및 큐 스레드를 별도로 유지하는 다른 변형.
"""sample code for logging in subprocesses using multiprocessing
* Little handler magic - The main process uses loggers and handlers as normal.
* Only a simple handler is needed in the subprocess that feeds the queue.
* Original logger name from subprocess is preserved when logged in main
process.
* As in the other implementations, a thread reads the queue and calls the
handlers. Except in this implementation, the thread is defined outside of a
handler, which makes the logger definitions simpler.
* Works with multiple handlers. If the logger in the main process defines
multiple handlers, they will all be fed records generated by the
subprocesses loggers.
tested with Python 2.5 and 2.6 on Linux and Windows
"""
import os
import sys
import time
import traceback
import multiprocessing, threading, logging, sys
DEFAULT_LEVEL = logging.DEBUG
formatter = logging.Formatter("%(levelname)s: %(asctime)s - %(name)s - %(process)s - %(message)s")
class SubProcessLogHandler(logging.Handler):
"""handler used by subprocesses
It simply puts items on a Queue for the main process to log.
"""
def __init__(self, queue):
logging.Handler.__init__(self)
self.queue = queue
def emit(self, record):
self.queue.put(record)
class LogQueueReader(threading.Thread):
"""thread to write subprocesses log records to main process log
This thread reads the records written by subprocesses and writes them to
the handlers defined in the main process's handlers.
"""
def __init__(self, queue):
threading.Thread.__init__(self)
self.queue = queue
self.daemon = True
def run(self):
"""read from the queue and write to the log handlers
The logging documentation says logging is thread safe, so there
shouldn't be contention between normal logging (from the main
process) and this thread.
Note that we're using the name of the original logger.
"""
# Thanks Mike for the error checking code.
while True:
try:
record = self.queue.get()
# get the logger for this record
logger = logging.getLogger(record.name)
logger.callHandlers(record)
except (KeyboardInterrupt, SystemExit):
raise
except EOFError:
break
except:
traceback.print_exc(file=sys.stderr)
class LoggingProcess(multiprocessing.Process):
def __init__(self, queue):
multiprocessing.Process.__init__(self)
self.queue = queue
def _setupLogger(self):
# create the logger to use.
logger = logging.getLogger('test.subprocess')
# The only handler desired is the SubProcessLogHandler. If any others
# exist, remove them. In this case, on Unix and Linux the StreamHandler
# will be inherited.
for handler in logger.handlers:
# just a check for my sanity
assert not isinstance(handler, SubProcessLogHandler)
logger.removeHandler(handler)
# add the handler
handler = SubProcessLogHandler(self.queue)
handler.setFormatter(formatter)
logger.addHandler(handler)
# On Windows, the level will not be inherited. Also, we could just
# set the level to log everything here and filter it in the main
# process handlers. For now, just set it from the global default.
logger.setLevel(DEFAULT_LEVEL)
self.logger = logger
def run(self):
self._setupLogger()
logger = self.logger
# and here goes the logging
p = multiprocessing.current_process()
logger.info('hello from process %s with pid %s' % (p.name, p.pid))
if __name__ == '__main__':
# queue used by the subprocess loggers
queue = multiprocessing.Queue()
# Just a normal logger
logger = logging.getLogger('test')
handler = logging.StreamHandler()
handler.setFormatter(formatter)
logger.addHandler(handler)
logger.setLevel(DEFAULT_LEVEL)
logger.info('hello from the main process')
# This thread will read from the subprocesses and write to the main log's
# handlers.
log_queue_reader = LogQueueReader(queue)
log_queue_reader.start()
# create the processes.
for i in range(10):
p = LoggingProcess(queue)
p.start()
# The way I read the multiprocessing warning about Queue, joining a
# process before it has finished feeding the Queue can cause a deadlock.
# Also, Queue.empty() is not realiable, so just make sure all processes
# are finished.
# active_children joins subprocesses when they're finished.
while multiprocessing.active_children():
time.sleep(.1)
모든 현재 솔루션이 핸들러를 사용하여 로깅 구성에 너무 연결되어 있습니다. 내 솔루션에는 다음과 같은 아키텍처와 기능이 있습니다.
- 당신이 사용할 수 있는 당신이 원하는 로깅 구성
- 데몬 스레드에서 로깅이 수행됩니다.
- 컨텍스트 관리자를 사용하여 데몬을 안전하게 종료
- 로깅 스레드와의 통신은 다음과 같이 수행됩니다.
multiprocessing.Queue - 서브 프로세스
logging.Logger(및 이미 정의 된 인스턴스)에서 모든 레코드를 큐에 보내도록 패치 됩니다. - 새로운 : 형식 추적 및 메시지 산세 오류를 방지하기 위해 큐에 보내기 전에
사용 예제 및 출력 코드는 다음 요점에서 찾을 수 있습니다. https://gist.github.com/schlamar/7003737
다중 프로세스 로깅을 다수의 게시자와 하나의 구독자 (리스너)로 나타낼 수 있으므로 ZeroMQ 를 사용하여 PUB-SUB 메시징을 구현하는 것이 실제로 선택 사항입니다.
또한, PyZMQ의 모듈, ZMQ에 대한 파이썬 바인딩 구현 PUBHandler zmq.PUB의 소켓을 통해 로깅 메시지를 게시하기위한 목적이다.
있다 웹 솔루션을 쉽게 여러 출판 프로세스와 로컬 작업에 채택 될 수 PyZMQ 및 PUBHandler를 사용하여 분산 응용 프로그램에서 중앙 집중식 로깅은.
formatters = {
logging.DEBUG: logging.Formatter("[%(name)s] %(message)s"),
logging.INFO: logging.Formatter("[%(name)s] %(message)s"),
logging.WARN: logging.Formatter("[%(name)s] %(message)s"),
logging.ERROR: logging.Formatter("[%(name)s] %(message)s"),
logging.CRITICAL: logging.Formatter("[%(name)s] %(message)s")
}
# This one will be used by publishing processes
class PUBLogger:
def __init__(self, host, port=config.PUBSUB_LOGGER_PORT):
self._logger = logging.getLogger(__name__)
self._logger.setLevel(logging.DEBUG)
self.ctx = zmq.Context()
self.pub = self.ctx.socket(zmq.PUB)
self.pub.connect('tcp://{0}:{1}'.format(socket.gethostbyname(host), port))
self._handler = PUBHandler(self.pub)
self._handler.formatters = formatters
self._logger.addHandler(self._handler)
@property
def logger(self):
return self._logger
# This one will be used by listener process
class SUBLogger:
def __init__(self, ip, output_dir="", port=config.PUBSUB_LOGGER_PORT):
self.output_dir = output_dir
self._logger = logging.getLogger()
self._logger.setLevel(logging.DEBUG)
self.ctx = zmq.Context()
self._sub = self.ctx.socket(zmq.SUB)
self._sub.bind('tcp://*:{1}'.format(ip, port))
self._sub.setsockopt(zmq.SUBSCRIBE, "")
handler = handlers.RotatingFileHandler(os.path.join(output_dir, "client_debug.log"), "w", 100 * 1024 * 1024, 10)
handler.setLevel(logging.DEBUG)
formatter = logging.Formatter("%(asctime)s;%(levelname)s - %(message)s")
handler.setFormatter(formatter)
self._logger.addHandler(handler)
@property
def sub(self):
return self._sub
@property
def logger(self):
return self._logger
# And that's the way we actually run things:
# Listener process will forever listen on SUB socket for incoming messages
def run_sub_logger(ip, event):
sub_logger = SUBLogger(ip)
while not event.is_set():
try:
topic, message = sub_logger.sub.recv_multipart(flags=zmq.NOBLOCK)
log_msg = getattr(logging, topic.lower())
log_msg(message)
except zmq.ZMQError as zmq_error:
if zmq_error.errno == zmq.EAGAIN:
pass
# Publisher processes loggers should be initialized as follows:
class Publisher:
def __init__(self, stop_event, proc_id):
self.stop_event = stop_event
self.proc_id = proc_id
self._logger = pub_logger.PUBLogger('127.0.0.1').logger
def run(self):
self._logger.info("{0} - Sending message".format(proc_id))
def run_worker(event, proc_id):
worker = Publisher(event, proc_id)
worker.run()
# Starting subscriber process so we won't loose publisher's messages
sub_logger_process = Process(target=run_sub_logger,
args=('127.0.0.1'), stop_event,))
sub_logger_process.start()
#Starting publisher processes
for i in range(MAX_WORKERS_PER_CLIENT):
processes.append(Process(target=run_worker,
args=(stop_event, i,)))
for p in processes:
p.start()
나는 또한 zzzeek의 답변을 좋아하지만 Andre는 방해를 피하기 위해 대기열이 필요하다는 것이 맞습니다. 나는 파이프와 약간의 운이 있었지만, 어느 정도 예상되는 garbling을 보았습니다. 전역 변수 및 물건에 대한 추가 제한이있는 Windows에서 실행하기 때문에 생각보다 어렵게 구현했습니다 ( Python Multiprocessing은 Windows에서 어떻게 구현됩니까? )
그러나 마침내 작동하게되었습니다. 이 예제는 아마도 완벽하지 않으므로 의견과 제안을 환영합니다. 또한 포매터 또는 루트 로거 이외의 설정을 지원하지 않습니다. 기본적으로 큐를 사용하여 각 풀 프로세스에서 로거를 다시 초기화하고 로거에서 다른 속성을 설정해야합니다.
코드를 개선하는 방법에 대한 제안은 언제든지 환영합니다. 나는 아직 모든 파이썬 트릭을 모른다 :-)
import multiprocessing, logging, sys, re, os, StringIO, threading, time, Queue
class MultiProcessingLogHandler(logging.Handler):
def __init__(self, handler, queue, child=False):
logging.Handler.__init__(self)
self._handler = handler
self.queue = queue
# we only want one of the loggers to be pulling from the queue.
# If there is a way to do this without needing to be passed this
# information, that would be great!
if child == False:
self.shutdown = False
self.polltime = 1
t = threading.Thread(target=self.receive)
t.daemon = True
t.start()
def setFormatter(self, fmt):
logging.Handler.setFormatter(self, fmt)
self._handler.setFormatter(fmt)
def receive(self):
#print "receive on"
while (self.shutdown == False) or (self.queue.empty() == False):
# so we block for a short period of time so that we can
# check for the shutdown cases.
try:
record = self.queue.get(True, self.polltime)
self._handler.emit(record)
except Queue.Empty, e:
pass
def send(self, s):
# send just puts it in the queue for the server to retrieve
self.queue.put(s)
def _format_record(self, record):
ei = record.exc_info
if ei:
dummy = self.format(record) # just to get traceback text into record.exc_text
record.exc_info = None # to avoid Unpickleable error
return record
def emit(self, record):
try:
s = self._format_record(record)
self.send(s)
except (KeyboardInterrupt, SystemExit):
raise
except:
self.handleError(record)
def close(self):
time.sleep(self.polltime+1) # give some time for messages to enter the queue.
self.shutdown = True
time.sleep(self.polltime+1) # give some time for the server to time out and see the shutdown
def __del__(self):
self.close() # hopefully this aids in orderly shutdown when things are going poorly.
def f(x):
# just a logging command...
logging.critical('function number: ' + str(x))
# to make some calls take longer than others, so the output is "jumbled" as real MP programs are.
time.sleep(x % 3)
def initPool(queue, level):
"""
This causes the logging module to be initialized with the necessary info
in pool threads to work correctly.
"""
logging.getLogger('').addHandler(MultiProcessingLogHandler(logging.StreamHandler(), queue, child=True))
logging.getLogger('').setLevel(level)
if __name__ == '__main__':
stream = StringIO.StringIO()
logQueue = multiprocessing.Queue(100)
handler= MultiProcessingLogHandler(logging.StreamHandler(stream), logQueue)
logging.getLogger('').addHandler(handler)
logging.getLogger('').setLevel(logging.DEBUG)
logging.debug('starting main')
# when bulding the pool on a Windows machine we also have to init the logger in all the instances with the queue and the level of logging.
pool = multiprocessing.Pool(processes=10, initializer=initPool, initargs=[logQueue, logging.getLogger('').getEffectiveLevel()] ) # start worker processes
pool.map(f, range(0,50))
pool.close()
logging.debug('done')
logging.shutdown()
print "stream output is:"
print stream.getvalue()
로거 인스턴스 어딘가에 게시하십시오. 이렇게하면 다른 모듈과 클라이언트가 API를 사용하여 로거를 가져올 필요가 없습니다 import multiprocessing.
zzzeek의 답변이 마음에 들었습니다. 여러 스레드 / 프로세스가 동일한 파이프 끝을 사용하여 로그 메시지를 생성하면 오류가 발생하기 때문에 파이프를 큐로 대체합니다.
큐에서 모든 로그 항목을 읽는 다른 프로세스로 모든 로깅을 위임하는 것은 어떻습니까?
LOG_QUEUE = multiprocessing.JoinableQueue()
class CentralLogger(multiprocessing.Process):
def __init__(self, queue):
multiprocessing.Process.__init__(self)
self.queue = queue
self.log = logger.getLogger('some_config')
self.log.info("Started Central Logging process")
def run(self):
while True:
log_level, message = self.queue.get()
if log_level is None:
self.log.info("Shutting down Central Logging process")
break
else:
self.log.log(log_level, message)
central_logger_process = CentralLogger(LOG_QUEUE)
central_logger_process.start()
멀티 프로세스 메커니즘이나 상속을 통해 LOG_QUEUE를 공유하기 만하면됩니다.
내 코드 중 일부에서 logging.exception을 사용한다는 점을 제외하고는 IronHacker와 비슷한 솔루션이 있으며 트레이스 백을 피클 할 수 없으므로 큐를 통해 예외를 다시 전달하기 전에 예외를 포맷해야한다는 것을 알았습니다.
class QueueHandler(logging.Handler):
def __init__(self, queue):
logging.Handler.__init__(self)
self.queue = queue
def emit(self, record):
if record.exc_info:
# can't pass exc_info across processes so just format now
record.exc_text = self.formatException(record.exc_info)
record.exc_info = None
self.queue.put(record)
def formatException(self, ei):
sio = cStringIO.StringIO()
traceback.print_exception(ei[0], ei[1], ei[2], None, sio)
s = sio.getvalue()
sio.close()
if s[-1] == "\n":
s = s[:-1]
return s
아래는 Windows 환경에서 사용할 수있는 클래스이며 ActivePython이 필요합니다. 다른 로깅 핸들러 (StreamHandler 등)에 대해서도 상속 할 수 있습니다.
class SyncronizedFileHandler(logging.FileHandler):
MUTEX_NAME = 'logging_mutex'
def __init__(self , *args , **kwargs):
self.mutex = win32event.CreateMutex(None , False , self.MUTEX_NAME)
return super(SyncronizedFileHandler , self ).__init__(*args , **kwargs)
def emit(self, *args , **kwargs):
try:
win32event.WaitForSingleObject(self.mutex , win32event.INFINITE)
ret = super(SyncronizedFileHandler , self ).emit(*args , **kwargs)
finally:
win32event.ReleaseMutex(self.mutex)
return ret
다음은 사용법을 보여주는 예입니다.
import logging
import random , time , os , sys , datetime
from string import letters
import win32api , win32event
from multiprocessing import Pool
def f(i):
time.sleep(random.randint(0,10) * 0.1)
ch = random.choice(letters)
logging.info( ch * 30)
def init_logging():
'''
initilize the loggers
'''
formatter = logging.Formatter("%(levelname)s - %(process)d - %(asctime)s - %(filename)s - %(lineno)d - %(message)s")
logger = logging.getLogger()
logger.setLevel(logging.INFO)
file_handler = SyncronizedFileHandler(sys.argv[1])
file_handler.setLevel(logging.INFO)
file_handler.setFormatter(formatter)
logger.addHandler(file_handler)
#must be called in the parent and in every worker process
init_logging()
if __name__ == '__main__':
#multiprocessing stuff
pool = Pool(processes=10)
imap_result = pool.imap(f , range(30))
for i , _ in enumerate(imap_result):
pass
여기 내 간단한 해킹 / 해결 방법이 있습니다 ... 가장 포괄적이지는 않지만 쉽게 수정하고 쉽게 읽고 이해할 수 있습니다.이 글을 작성하기 전에 찾은 다른 답변보다 생각합니다.
import logging
import multiprocessing
class FakeLogger(object):
def __init__(self, q):
self.q = q
def info(self, item):
self.q.put('INFO - {}'.format(item))
def debug(self, item):
self.q.put('DEBUG - {}'.format(item))
def critical(self, item):
self.q.put('CRITICAL - {}'.format(item))
def warning(self, item):
self.q.put('WARNING - {}'.format(item))
def some_other_func_that_gets_logger_and_logs(num):
# notice the name get's discarded
# of course you can easily add this to your FakeLogger class
local_logger = logging.getLogger('local')
local_logger.info('Hey I am logging this: {} and working on it to make this {}!'.format(num, num*2))
local_logger.debug('hmm, something may need debugging here')
return num*2
def func_to_parallelize(data_chunk):
# unpack our args
the_num, logger_q = data_chunk
# since we're now in a new process, let's monkeypatch the logging module
logging.getLogger = lambda name=None: FakeLogger(logger_q)
# now do the actual work that happens to log stuff too
new_num = some_other_func_that_gets_logger_and_logs(the_num)
return (the_num, new_num)
if __name__ == '__main__':
multiprocessing.freeze_support()
m = multiprocessing.Manager()
logger_q = m.Queue()
# we have to pass our data to be parallel-processed
# we also need to pass the Queue object so we can retrieve the logs
parallelable_data = [(1, logger_q), (2, logger_q)]
# set up a pool of processes so we can take advantage of multiple CPU cores
pool_size = multiprocessing.cpu_count() * 2
pool = multiprocessing.Pool(processes=pool_size, maxtasksperchild=4)
worker_output = pool.map(func_to_parallelize, parallelable_data)
pool.close() # no more tasks
pool.join() # wrap up current tasks
# get the contents of our FakeLogger object
while not logger_q.empty():
print logger_q.get()
print 'worker output contained: {}'.format(worker_output)
대안 중 하나는 다중 처리 로깅을 알려진 파일에 기록하고 atexit처리기에 등록 할 핸들러를 등록하여 stderr에서 다시 읽는 것입니다. 그러나 그런 식으로 stderr의 출력 메시지로 실시간 흐름을 얻지 못합니다.
logging모듈 에서 잠금, 스레드 및 포크의 조합으로 교착 상태가 발생하면 버그 보고서 6721 에보고됩니다 ( 관련 SO 질문 참조 ).
여기에 작은 해결책이 게시되어 있습니다 .
그러나 이는 잠재적 교착 상태를 해결합니다 logging. 그래도 문제가 해결되지는 않습니다. 여기에 제시된 다른 답변을 참조하십시오.
이 위대한 패키지가 있습니다
패키지 : https://pypi.python.org/pypi/multiprocessing-logging/
코드 : https://github.com/jruere/multiprocessing-logging
설치:
pip install multiprocessing-logging
그런 다음 다음을 추가하십시오.
import multiprocessing_logging
# This enables logs inside process
multiprocessing_logging.install_mp_handler()
언급 된 가장 간단한 아이디어 :
- 현재 프로세스의 파일 이름과 프로세스 ID를 가져옵니다.
- 을 설정하십시오
[WatchedFileHandler][1]. 이 핸들러에 대한 이유는 여기 에서 자세히 설명 하지만 다른 로깅 핸들러와의 경쟁 조건은 더 나쁩니다. 경쟁 조건에 가장 짧은 창이 있습니다.- / var / log / ...와 같이 로그를 저장할 경로를 선택하십시오.
수십 년 동안 같은 문제를 겪고이 사이트에서이 질문을 발견 한 내 아이들에게 나는이 대답을 남깁니다.
단순성 대 복잡성. 다른 도구를 사용하십시오. 파이썬은 훌륭하지만 어떤 일을하도록 설계되지 않았습니다.
logrotate 데몬에 대한 다음 스 니펫은 저에게 효과적이며 지나치게 복잡하지 않습니다. 매시간 실행되도록 예약하고
/var/log/mylogfile.log {
size 1
copytruncate
create
rotate 10
missingok
postrotate
timeext=`date -d '1 hour ago' "+%Y-%m-%d_%H"`
mv /var/log/mylogfile.log.1 /var/log/mylogfile-$timeext.log
endscript
}
이것은 내가 설치하는 방법입니다 (symlink는 logrotate에서 작동하지 않습니다).
sudo cp /directpath/config/logrotate/myconfigname /etc/logrotate.d/myconfigname
sudo cp /etc/cron.daily/logrotate /etc/cron.hourly/logrotate
참고 URL : https://stackoverflow.com/questions/641420/how-should-i-log-while-using-multiprocessing-in-python
반응형
'IT story' 카테고리의 다른 글
| '? :'의 반환 유형 (삼항 조건 연산자) (0) | 2020.05.05 |
|---|---|
| 새로운 PostgreSQL JSON 데이터 유형의 필드를 사용하여 쿼리하려면 어떻게합니까? (0) | 2020.05.05 |
| ASP.NET MVC에서 요청 조절을 구현하는 가장 좋은 방법은 무엇입니까? (0) | 2020.05.05 |
| 속성 탐색기처럼 작동하는 GUI 기반 또는 웹 기반 JSON 편집기 (0) | 2020.05.05 |
| 부모 구성 요소의 CSS 파일에서 자식 구성 요소의 스타일을 지정하는 방법은 무엇입니까? (0) | 2020.05.05 |