Como devo registrar enquanto estiver usando o multiprocessamento no Python?

239

No momento, tenho um módulo central em uma estrutura que gera vários processos usando o multiprocessingmódulo Python 2.6 . Como ele usa multiprocessing, há um log com reconhecimento de multiprocessamento no nível do módulo LOG = multiprocessing.get_logger(). De acordo com os documentos , esse criador de logs possui bloqueios compartilhados por processos, para que você não detecte coisas sys.stderr(ou qualquer outra manipulação de arquivo) ao ter vários processos gravando nele simultaneamente.

O problema que tenho agora é que os outros módulos na estrutura não são compatíveis com multiprocessamento. Na minha opinião, preciso fazer com que todas as dependências desse módulo central usem o log com reconhecimento de multiprocessamento. Isso é irritante dentro da estrutura, muito menos para todos os clientes da estrutura. Existem alternativas em que não estou pensando?

cdleary
fonte
10
Os documentos aos quais você vincula indicam exatamente o oposto do que você diz, o criador de logs não possui bloqueios compartilhados por processo e as coisas se confundem - um problema que também tive.
Sebastian Blask
3
veja exemplos nos documentos stdlib: Registrando em um único arquivo a partir de vários processos . As receitas não exigem que outros módulos sejam compatíveis com multiprocessamento.
jfs
Então, qual é o caso de uso multiprocessing.get_logger()? Parece que, com base nessas outras formas de registro, a funcionalidade de registro tem multiprocessingpouco valor.
Tim Ludwinski
4
get_logger()é o logger usado pelo multiprocessingpróprio módulo. É útil se você deseja depurar um multiprocessingproblema.
JFS

Respostas:

69

A única maneira de lidar com isso de maneira não intrusiva é:

  1. Gera cada processo de trabalho de forma que seu log vá para um descritor de arquivo diferente (para disco ou canal). Idealmente, todas as entradas de log devem ter registro de data e hora.
  2. O processo do seu controlador pode fazer um dos seguintes:
    • Se estiver usando arquivos de disco: Agrale os arquivos de log no final da execução, classificados por carimbo de data / hora
    • Se estiver usando pipes (recomendado): agrupe as entradas de log on-the-fly de todos os pipes, em um arquivo de log central. (Por exemplo, periodicamente a selectpartir dos descritores de arquivo dos pipes, execute a classificação de mesclagem nas entradas de log disponíveis e limpe-as para o log centralizado. Repita.)
vladr
fonte
Bom, eram 35 anos antes de eu pensar nisso (pensei em usar atexit:-). O problema é que ele não fornecerá uma leitura em tempo real. Isso pode fazer parte do preço do multiprocessamento, em oposição ao multithreading.
cdleary
@cdleary, usando a abordagem encanada seria como quase em tempo real como se pode obter (especialmente se stderr não é tamponado nos processos gerados.)
vladr
1
Aliás, grande suposição aqui: não o Windows. Você está no Windows?
vladr
22
Por que não usar um multiprocessing.Queue e um thread de log no processo principal? Parece mais simples.
Brandon Rhodes
1
@BrandonRhodes - Como eu disse, de forma não intrusiva . Usando multiprocessing.Queuenão será mais simples se há um monte de código para religação de uso multiprocessing.Queue, e / ou se o desempenho é um problema
vladr
122

Acabei de escrever um manipulador de log que apenas envia tudo ao processo pai por meio de um pipe. Eu só o testei por dez minutos, mas parece funcionar muito bem.

( Nota: Este é codificado permanentemente RotatingFileHandler, que é meu próprio caso de uso.)


Atualização: o @javier agora mantém essa abordagem como um pacote disponível no Pypi - consulte registro de multiprocessamento no Pypi, github em https://github.com/jruere/multiprocessing-logging


Atualização: Implementação!

Agora, isso usa uma fila para o tratamento correto da simultaneidade e também se recupera de erros corretamente. Agora estou usando isso em produção há vários meses, e a versão atual abaixo funciona sem problemas.

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)
zzzeek
fonte
4
O manipulador acima faz todo o arquivo gravado no processo pai e usa apenas um encadeamento para receber mensagens passadas dos processos filhos. Se você chamar o próprio manipulador a partir de um processo filho gerado, ele será usado incorretamente e você terá os mesmos problemas que o RotatingFileHandler. Eu usei o código acima por anos sem problemas.
Zzzeek
9
Infelizmente, essa abordagem não funciona no Windows. De docs.python.org/library/multiprocessing.html 16.6.2.12 "Observe que nos processos filhos do Windows herdarão apenas o nível do criador de logs do processo pai - qualquer outra personalização do criador de logs não será herdada." Os subprocessos não herdarão o manipulador, e você não pode transmiti-lo explicitamente porque não é selecionável.
Noé Yetter
2
É importante notar que multiprocessing.Queueutiliza um fio no put(). Portanto, não chame put(por exemplo, registre uma mensagem usando MultiProcessingLogmanipulador) antes de criar todos os subprocessos. Caso contrário, o encadeamento ficará inoperante no processo filho. Uma solução é chamar Queue._after_fork()no início de cada processo filho ou usá-lo multiprocessing.queues.SimpleQueue, o que não envolve encadeamento, mas está bloqueando.
Danqi Wang 11/10
5
Você poderia adicionar um exemplo simples que mostra a inicialização, bem como o uso de um processo filho hipotético? Não tenho muita certeza de como o processo filho deve ter acesso à fila sem instanciar outra instância da sua classe.
precisa saber é o seguinte
11
@zzzeek, ​​esta solução é boa, mas não consegui encontrar um pacote ou algo semelhante, então criei um chamado multiprocessing-logging.
Javier
30

QueueHandleré nativo no Python 3.2+ e faz exatamente isso. É facilmente replicado em versões anteriores.

Os documentos Python têm dois exemplos completos: Registrando em um único arquivo a partir de vários processos

Para aqueles que usam Python <3.2, copie QueueHandlerpara seu próprio código em: https://gist.github.com/vsajip/591589 ou importe alternativamente logutils .

Cada processo (incluindo o processo pai) coloca seu log no Queuee, em seguida, um listenerthread ou processo (um exemplo é fornecido para cada) os seleciona e os grava em um arquivo - sem risco de corrupção ou distorção.

fantabolous
fonte
21

Abaixo está outra solução com foco na simplicidade para qualquer pessoa (como eu) que chega aqui do Google. O registro deve ser fácil! Somente para 3,2 ou superior.

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()
user2133814
fonte
2
As classes QueueHandlere também QueueListenerpodem ser usadas no Python 2.7, disponíveis no logutilspacote.
Lev Levitsky
5
O criador de logs do processo principal também deve usar um QueueHandler. No seu código atual, o processo principal é ignorar a fila, para que possa haver condições de corrida entre o processo principal e os trabalhadores. Todos devem fazer logon na fila (por meio de um QueueHandler) e apenas o QueueListener deve ter permissão para fazer logon no StreamHandler.
Ismael EL ATIFI
Além disso, você não precisa inicializar o logger em cada filho. Apenas inicialize o criador de logs no processo pai e obtenha o criador de logs em cada processo filho.
okwap 14/01
20

Ainda outra alternativa pode ser os vários manipuladores de log não baseados em arquivo no loggingpacote :

  • SocketHandler
  • DatagramHandler
  • SyslogHandler

(e outros)

Dessa forma, você pode facilmente ter um daemon de log em algum lugar no qual possa gravar com segurança e manipular os resultados corretamente. (Por exemplo, um servidor de soquete simples que apenas retira a mensagem e a emite em seu próprio manipulador de arquivo rotativo.)

O SyslogHandlercuidaria disso também para você. Obviamente, você poderia usar sua própria instância syslog, não a do sistema.

Ali Afshar
fonte
13

Uma variante das outras que mantém o segmento de log e fila separados.

"""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)
ironhacker
fonte
Eu gosto de uma idéia de buscar o nome do logger no registro da fila. Isso permite o uso convencional fileConfig()no MainProcess e um criador de logs mal configurado no PoolWorkers (apenas setLevel(logging.NOTSET)). Como mencionei em outro comentário, estou usando Pool, então tive que obter minha Fila (proxy) do Manager em vez de multiprocessamento, para que pudesse ser decapada. Isso me permite passar a fila para um trabalhador dentro de um dicionário (a maioria dos quais é derivada do objeto argsparse usando vars()). Sinto que, no final, essa é a melhor abordagem para o MS Windows que não possui o fork () e quebra a solução @zzzeak.
Mlt #
@mlt Acho que você também pode colocar uma fila de multiprocessamento no init em vez de usar um gerente (consulte a resposta para stackoverflow.com/questions/25557686/… - trata-se de bloqueios, mas acredito que funciona para filas também)
fantabolous
@fantabolous Isso não funcionará no MS Windows ou em qualquer outra plataforma que não possua fork. Dessa forma, cada processo terá sua própria fila inútil e independente. A segunda abordagem na Q / A vinculada não funcionará nessas plataformas. É uma maneira de código não portátil.
Mlt
@mlt Interessante. Estou usando o Windows e parece funcionar bem para mim - pouco depois do último comentário, configurei um conjunto de processos compartilhando um multiprocessing.Queuecom o processo principal e o uso constantemente desde então. Não pretendo entender por que funciona.
fantabolous
10

Todas as soluções atuais estão muito acopladas à configuração de log usando um manipulador. Minha solução possui a seguinte arquitetura e recursos:

  • Você pode usar qualquer configuração de log que desejar
  • O registro é feito em um encadeamento daemon
  • Encerramento seguro do daemon usando um gerenciador de contexto
  • A comunicação com o encadeamento de log é feita por multiprocessing.Queue
  • Nos subprocessos, logging.Logger(e instâncias já definidas) são corrigidos para enviar todos os registros para a fila
  • Novo : formate o retorno e a mensagem antes de enviar para a fila para evitar erros de decapagem

O código com exemplo de uso e saída pode ser encontrado na seguinte lista: https://gist.github.com/schlamar/7003737

Schlamar
fonte
A menos que eu estou faltando alguma coisa, isso não é realmente um fio daemon, já que você nunca se punha daemon_thread.daemona True. Eu precisava fazer isso para que meu programa Python saia corretamente quando ocorrer uma exceção no gerenciador de contexto.
blah238
Eu também precisava de captura, de log e engolir exceções lançadas pelo alvo funcem logged_call, caso contrário, a exceção seria se ilegível com outra saída registrada. Aqui está minha versão modificada: gist.github.com/blah238/8ab79c4fe9cdb254f5c37abfc5dc85bf
blah238
8

Como podemos representar o log de multiprocessos como muitos editores e um assinante (ouvinte), usar o ZeroMQ para implementar as mensagens do PUB-SUB é realmente uma opção.

Além disso, o módulo PyZMQ , as ligações Python para o ZMQ, implementa o PUBHandler , que é objeto para publicar mensagens de log em um soquete zmq.PUB.

Existe uma solução na Web , para registro centralizado de aplicativos distribuídos usando PyZMQ e PUBHandler, que pode ser facilmente adotado para trabalhar localmente com vários processos de publicação.

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()
Samuel
fonte
6

Eu também gosto da resposta de zzzeek, ​​mas Andre está certo de que uma fila é necessária para evitar erros. Tive alguma sorte com o cachimbo, mas vi gargalhar o que é algo esperado. A implementação acabou sendo mais difícil do que eu pensava, principalmente devido à execução no Windows, onde existem algumas restrições adicionais sobre variáveis ​​globais e outras coisas (consulte: Como o Multiprocessamento Python é implementado no Windows? )

Mas finalmente consegui funcionar. Esse exemplo provavelmente não é perfeito, portanto, comentários e sugestões são bem-vindos. Ele também não suporta a configuração do formatador ou qualquer outra coisa que não seja o registrador raiz. Basicamente, você precisa reiniciar o criador de logs em cada um dos processos do pool com a fila e configurar os outros atributos no criador de logs.

Novamente, todas as sugestões sobre como melhorar o código são bem-vindas. Eu certamente ainda não conheço todos os truques do Python :-)

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()
Mike Miller
fonte
1
Gostaria de saber se if 'MainProcess' == multiprocessing.current_process().name:pode ser usado no lugar da passagem child?
Mlt 02/10
Caso outra pessoa esteja tentando usar o pool de processos em vez de objetos de processo separados no Windows, vale ressaltar que o Manager deve ser usado para passar a fila para subprocessos, pois não pode ser selecionado diretamente.
Mlt 2/13
Essa implementação funcionou bem para mim. Eu o modifiquei para trabalhar com um número arbitrário de manipuladores. Dessa forma, você pode configurar seu manipulador raiz de maneira que não seja de multiprocessamento e, em seguida, onde é seguro criar a fila, passar os manipuladores raiz para isso, excluí-los e torná-lo o único manipulador.
Jaxor24
3

basta publicar em algum lugar sua instância do logger. Dessa forma, os outros módulos e clientes podem usar sua API para obter o criador de logs sem a necessidade import multiprocessing.

Javier
fonte
1
O problema é que os registradores de multiprocessamento aparecem sem nome, para que você não consiga decifrar o fluxo de mensagens facilmente. Talvez fosse possível nomeá-los após a criação, o que tornaria mais razoável olhar.
cdleary
publique um criador de logs para cada módulo, ou melhor, exporte diferentes fechamentos que usem o criador de logs com o nome do módulo. o ponto é deixar que outros módulos usar seu API
Javier
Definitivamente razoável (e +1 de mim!), Mas sentiria falta de ser capaz import logging; logging.basicConfig(level=logging.DEBUG); logging.debug('spam!')de qualquer lugar e fazê-lo funcionar corretamente.
cdleary
3
É um fenômeno interessante que eu vejo quando uso o Python, que nos acostumamos a fazer o que queremos em 1 ou 2 linhas simples que a abordagem simples e lógica em outras linguagens (por exemplo, para publicar o logger ou empacotamento de multiprocessamento em um acessador) ainda parece um fardo. :)
Kylotan 13/03/09
3

Gostei da resposta do zzzeek. Gostaria apenas de substituir o Pipe por uma fila, pois, se vários threads / processos usarem o mesmo fim de pipe para gerar mensagens de log, eles ficarão ilegíveis.

André Cruz
fonte
Eu estava tendo alguns problemas com o manipulador, embora não houvesse mensagens ilegíveis, a coisa toda iria parar de funcionar. Mudei o Pipe para ser Fila, pois é mais apropriado. No entanto, os erros que eu estava recebendo não foram resolvidos com isso - finalmente adicionei uma tentativa / exceto ao método receive () - muito raramente, uma tentativa de registrar exceções falhará e acabará sendo capturada lá. Depois de adicionar o try / except, ele é executado por semanas sem nenhum problema, e um arquivo padrãoerr captura duas exceções erradas por semana.
zzzeek
2

Que tal delegar todo o log para outro processo que lê todas as entradas de log de uma fila?

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()

Simplesmente compartilhe LOG_QUEUE através de qualquer um dos mecanismos de multiprocessos ou até mesmo herança e tudo funciona bem!

Sawan
fonte
1

Eu tenho uma solução semelhante à da ironhacker, exceto pelo uso do logging.exception em alguns dos meus códigos e descobri que precisava formatar a exceção antes de transmiti-la pela Fila, pois os rastreamentos não são selecionáveis:

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
Richard Jones
fonte
Encontrei um exemplo completo nesse sentido aqui .
Aryeh Leib Taurog
1

Abaixo está uma classe que pode ser usada no ambiente Windows, requer ActivePython. Você também pode herdar para outros manipuladores de log (StreamHandler etc.)

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

E aqui está um exemplo que demonstra o uso:

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
user6336812
fonte
Provavelmente o uso em multiprocessing.Lock()vez do Windows Mutex tornaria a solução portátil.
precisa saber é o seguinte
1

Aqui está o meu truque / solução alternativa simples ... não o mais abrangente, mas facilmente modificável e mais simples de ler e entender, penso que qualquer outra resposta que encontrei antes de escrever isso:

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)
nmz787
fonte
1

Existe esse ótimo pacote

Pacote: https://pypi.python.org/pypi/multiprocessing-logging/

código: https://github.com/jruere/multiprocessing-logging

Instalar:

pip install multiprocessing-logging

Adicione então:

import multiprocessing_logging

# This enables logs inside process
multiprocessing_logging.install_mp_handler()
Juan Isaza
fonte
3
Esta biblioteca é literalmente baseada em outro comentário na publicação atual do SO: stackoverflow.com/a/894284/1698058 .
Chris Hunt
Origens: stackoverflow.com/a/894284/1663382 Agradeço o exemplo de uso do módulo, além da documentação na página inicial.
Liquidgenius
0

Uma das alternativas é gravar o log do mutliprocessing em um arquivo conhecido e registrar um atexitmanipulador para ingressar nesses processos; leia-o novamente no stderr; no entanto, você não receberá um fluxo em tempo real para as mensagens de saída no stderr dessa maneira.

cdleary
fonte
é a abordagem que você está propondo abaixo idêntico ao do seu comentário aqui stackoverflow.com/questions/641420/...
Iruvar
0

Se houver deadlocks em uma combinação de bloqueios, threads e garfos no loggingmódulo, isso é relatado no relatório de erros 6721 (consulte também a questão SO relacionada ).

Há uma pequena solução de correção postada aqui .

No entanto, isso apenas corrigirá possíveis impasses logging. Isso não vai resolver que as coisas talvez estejam distorcidas. Veja as outras respostas apresentadas aqui.

Albert
fonte
0

Idéia mais simples, como mencionado:

  • Pegue o nome do arquivo e a identificação do processo atual.
  • Configurar a [WatchedFileHandler][1]. As razões para esse manipulador são discutidas em detalhes aqui , mas, em suma, existem certas condições de corrida piores com os outros manipuladores de registro. Este tem a janela mais curta para a condição de corrida.
    • Escolha um caminho para salvar os logs, como / var / log / ...
user1460675
fonte
0

Para quem precisar, escrevi um decorador para o pacote multiprocessing_logging que adiciona o nome do processo atual aos logs, para ficar claro quem registra o que.

Também executa install_mp_handler (), tornando-se inútil executá-lo antes de criar um pool.

Isso me permite ver qual trabalhador cria quais mensagens de log.

Aqui está o plano com um exemplo:

import sys
import logging
from functools import wraps
import multiprocessing
import multiprocessing_logging

# Setup basic console logger as 'logger'
logger = logging.getLogger()
console_handler = logging.StreamHandler(sys.stdout)
console_handler.setFormatter(logging.Formatter(u'%(asctime)s :: %(levelname)s :: %(message)s'))
logger.setLevel(logging.DEBUG)
logger.addHandler(console_handler)


# Create a decorator for functions that are called via multiprocessing pools
def logs_mp_process_names(fn):
    class MultiProcessLogFilter(logging.Filter):
        def filter(self, record):
            try:
                process_name = multiprocessing.current_process().name
            except BaseException:
                process_name = __name__
            record.msg = f'{process_name} :: {record.msg}'
            return True

    multiprocessing_logging.install_mp_handler()
    f = MultiProcessLogFilter()

    # Wraps is needed here so apply / apply_async know the function name
    @wraps(fn)
    def wrapper(*args, **kwargs):
        logger.removeFilter(f)
        logger.addFilter(f)
        return fn(*args, **kwargs)

    return wrapper


# Create a test function and decorate it
@logs_mp_process_names
def test(argument):
    logger.info(f'test function called via: {argument}')


# You can also redefine undecored functions
def undecorated_function():
    logger.info('I am not decorated')


@logs_mp_process_names
def redecorated(*args, **kwargs):
    return undecorated_function(*args, **kwargs)


# Enjoy
if __name__ == '__main__':
    with multiprocessing.Pool() as mp_pool:
        # Also works with apply_async
        mp_pool.apply(test, ('mp pool',))
        mp_pool.apply(redecorated)
        logger.info('some main logs')
        test('main program')
Orsiris de Jong
fonte
-5

Aos meus filhos que encontram o mesmo problema em décadas e encontraram esta pergunta neste site, deixo esta resposta.

Simplicidade vs supercomplicar. Basta usar outras ferramentas. Python é incrível, mas não foi projetado para fazer algumas coisas.

O seguinte trecho para o daemon logrotate funciona para mim e não complica demais as coisas. Programe-o para executar a cada hora e

/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
}

É assim que eu o instalo (os links simbólicos não funcionam para o logrotate):

sudo cp /directpath/config/logrotate/myconfigname /etc/logrotate.d/myconfigname
sudo cp /etc/cron.daily/logrotate /etc/cron.hourly/logrotate
careca
fonte