Wie soll ich mich bei der Verwendung von Multiprocessing in Python anmelden?

238

Im Moment habe ich ein zentrales Modul in einem Framework, das mit dem Python 2.6- multiprocessingModul mehrere Prozesse erzeugt . Da es verwendet wird multiprocessing, gibt es ein Multiprozessor-fähiges Protokoll auf Modulebene LOG = multiprocessing.get_logger(). Gemäß den Dokumenten verfügt dieser Logger über prozessfreigabene Sperren, damit Sie nicht die Dinge sys.stderr(oder was auch immer für ein Dateihandle) beschädigen, indem mehrere Prozesse gleichzeitig darauf schreiben.

Das Problem, das ich jetzt habe, ist, dass die anderen Module im Framework nicht multiprocessing-fähig sind. So wie ich es sehe, muss ich dafür sorgen, dass alle Abhängigkeiten von diesem zentralen Modul eine multiprozessorientierte Protokollierung verwenden. Das ist innerhalb des Frameworks ärgerlich , geschweige denn für alle Clients des Frameworks. Gibt es Alternativen, an die ich nicht denke?

cdleary
quelle
10
Die Dokumente, auf die Sie verlinken, geben genau das Gegenteil von dem an, was Sie sagen. Der Logger hat keine gemeinsamen Prozesssperren und die Dinge werden durcheinander gebracht - ein Problem, das ich auch hatte.
Sebastian Blask
3
Siehe Beispiele in den stdlib-Dokumenten: Protokollierung in einer einzelnen Datei aus mehreren Prozessen . Die Rezepte erfordern nicht, dass andere Module Multiprozessor-fähig sind.
JFS
Wofür ist der Anwendungsfall multiprocessing.get_logger()? Es scheint, dass auf diesen anderen Arten der Protokollierung die Protokollierungsfunktionalität multiprocessingvon geringem Wert ist.
Tim Ludwinski
4
get_logger()ist der vom multiprocessingModul selbst verwendete Logger . Dies ist nützlich, wenn Sie ein multiprocessingProblem beheben möchten .
JFS

Antworten:

69

Der einzige Weg, um nicht aufdringlich damit umzugehen, ist:

  1. Spawnen Sie jeden Worker-Prozess so, dass sein Protokoll in einen anderen Dateideskriptor (auf Festplatte oder Pipe) verschoben wird. Idealerweise sollten alle Protokolleinträge mit einem Zeitstempel versehen sein.
  2. Ihr Controller-Prozess kann dann einen der folgenden Schritte ausführen:
    • Bei Verwendung von Festplattendateien: Führen Sie die Protokolldateien am Ende des Laufs nach Zeitstempel sortiert zusammen
    • Bei Verwendung von Pipes (empfohlen): Zusammenführen von Protokolleinträgen aus allen Pipes im laufenden Betrieb in einer zentralen Protokolldatei. (Führen Sie beispielsweise in regelmäßigen Abständen selectanhand der Dateideskriptoren der Pipes eine Zusammenführungssortierung für die verfügbaren Protokolleinträge durch und leeren Sie sie in das zentralisierte Protokoll. Wiederholen Sie diesen Vorgang.)
vladr
quelle
Schön, das war 35s bevor ich daran dachte (dachte ich würde verwenden atexit:-). Das Problem ist, dass Sie keine Echtzeitanzeige erhalten. Dies kann Teil des Preises für Multiprocessing im Gegensatz zu Multithreading sein.
cdleary
@cdleary, mit dem Piped-Ansatz wäre es so nah wie möglich an der Echtzeit (insbesondere wenn stderr in den gespawnten Prozessen nicht gepuffert ist.)
vladr
1
Übrigens große Annahme hier: nicht Windows. Bist du unter Windows?
Vladr
22
Warum nicht stattdessen einfach eine Multiprocessing.Queue und einen Protokollierungsthread im Hauptprozess verwenden? Scheint einfacher.
Brandon Rhodes
1
@BrandonRhodes - Wie gesagt, nicht aufdringlich . Die Verwendung multiprocessing.Queuewird nicht einfacher, wenn viel Code neu verdrahtet werden multiprocessing.Queuemuss und / oder wenn die Leistung ein Problem darstellt
vladr
122

Ich habe gerade einen eigenen Log-Handler geschrieben, der einfach alles über eine Pipe an den übergeordneten Prozess weiterleitet. Ich habe es nur zehn Minuten lang getestet, aber es scheint ziemlich gut zu funktionieren.

( Hinweis: Dies ist fest codiert RotatingFileHandler, was mein eigener Anwendungsfall ist.)


Update: @javier behält diesen Ansatz jetzt als Paket bei, das auf Pypi verfügbar ist - siehe Multiprocessing-Logging auf Pypi, github unter https://github.com/jruere/multiprocessing-logging


Update: Implementierung!

Dies verwendet jetzt eine Warteschlange für die korrekte Behandlung der Parallelität und stellt auch Fehler korrekt wieder her. Ich benutze dies jetzt seit mehreren Monaten in der Produktion und die aktuelle Version unten funktioniert ohne Probleme.

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
quelle
4
Der obige Handler führt das gesamte Schreiben von Dateien aus dem übergeordneten Prozess aus und verwendet nur einen Thread, um Nachrichten zu empfangen, die von untergeordneten Prozessen übergeben wurden. Wenn Sie den Handler selbst aus einem untergeordneten Prozess aufrufen, wird er falsch verwendet, und es treten dieselben Probleme wie bei RotatingFileHandler auf. Ich habe den obigen Code jahrelang ohne Probleme verwendet.
Zzzeek
9
Leider funktioniert dieser Ansatz unter Windows nicht. Von docs.python.org/library/multiprocessing.html 16.6.2.12 "Beachten Sie, dass untergeordnete Windows-Prozesse nur die Ebene des Loggers des übergeordneten Prozesses erben - andere Anpassungen des Loggers werden nicht vererbt." Unterprozesse erben den Handler nicht und Sie können ihn nicht explizit übergeben, da er nicht auswählbar ist.
Noah Yetter
2
Es ist erwähnenswert, dass multiprocessing.Queueein Thread verwendet wird, um in put(). Rufen Sie also nicht auf put(dh protokollieren Sie eine Nachricht mit dem MultiProcessingLogHandler), bevor Sie alle Unterprozesse erstellt haben. Andernfalls ist der Thread im untergeordneten Prozess tot. Eine Lösung besteht darin, Queue._after_fork()zu Beginn jedes untergeordneten Prozesses aufzurufen oder multiprocessing.queues.SimpleQueuestattdessen zu verwenden , was keinen Thread beinhaltet, sondern blockiert.
Danqi Wang
5
Könnten Sie ein einfaches Beispiel hinzufügen, das die Initialisierung sowie die Verwendung eines hypothetischen untergeordneten Prozesses zeigt? Ich bin nicht ganz sicher, wie der untergeordnete Prozess Zugriff auf die Warteschlange erhalten soll, ohne eine andere Instanz Ihrer Klasse zu instanziieren.
JesseBuesking
11
@zzzeek, ​​diese Lösung ist gut, aber ich konnte kein Paket damit oder ähnliches finden, also habe ich ein Paket namens erstellt multiprocessing-logging.
Javier
30

QueueHandlerist in Python 3.2+ nativ und macht genau das. Es ist in früheren Versionen leicht zu replizieren.

Python-Dokumente enthalten zwei vollständige Beispiele: Protokollierung in einer einzelnen Datei aus mehreren Prozessen

Wenn Sie Python <3.2 verwenden, kopieren QueueHandlerSie einfach Ihren eigenen Code von: https://gist.github.com/vsajip/591589 oder importieren Sie alternativ logutils .

Jeder Prozess (einschließlich des übergeordneten Prozesses) legt seine Protokollierung auf dem ab Queue, und dann nimmt ein listenerThread oder Prozess (für jeden wird ein Beispiel bereitgestellt) diese auf und schreibt sie alle in eine Datei - ohne das Risiko von Beschädigung oder Verstümmelung.

fantastisch
quelle
21

Im Folgenden finden Sie eine weitere Lösung mit dem Schwerpunkt auf Einfachheit für alle anderen (wie mich), die von Google hierher kommen. Die Protokollierung sollte einfach sein! Nur für 3.2 oder höher.

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
quelle
2
Die Klassen QueueHandlerund QueueListenerkönnen auch in Python 2.7 verwendet werden, das im logutilsPaket verfügbar ist .
Lev Levitsky
5
Der Logger des Hauptprozesses sollte auch einen QueueHandler verwenden. In Ihrem aktuellen Code umgeht der Hauptprozess die Warteschlange, sodass zwischen dem Hauptprozess und den Arbeitern Race-Bedingungen bestehen können. Jeder sollte sich in der Warteschlange anmelden (über einen QueueHandler) und nur der QueueListener sollte sich beim StreamHandler anmelden dürfen.
Ismael EL ATIFI
Außerdem müssen Sie den Logger nicht in jedem Kind initialisieren. Initialisieren Sie einfach den Logger im übergeordneten Prozess und rufen Sie den Logger in jedem untergeordneten Prozess ab.
Okwap
20

Eine weitere Alternative könnten die verschiedenen nicht dateibasierten Protokollierungshandler im loggingPaket sein :

  • SocketHandler
  • DatagramHandler
  • SyslogHandler

(und andere)

Auf diese Weise könnten Sie leicht irgendwo einen Protokollierungsdämon haben, in den Sie sicher schreiben und die Ergebnisse korrekt verarbeiten können. (ZB ein einfacher Socket-Server, der die Nachricht einfach abhebt und an seinen eigenen rotierenden Datei-Handler ausgibt.)

Das SyslogHandlerwürde sich auch für Sie erledigen. Natürlich können Sie auch Ihre eigene Instanz verwenden syslog, nicht die des Systems.

Ali Afshar
quelle
13

Eine Variante der anderen, die den Protokollierungs- und Warteschlangenthread getrennt hält.

"""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
quelle
Ich mag die Idee, den Loggernamen aus dem Warteschlangendatensatz abzurufen. Dies ermöglicht die Verwendung von konventionell fileConfig()in MainProcess und eines kaum konfigurierten Loggers in PoolWorkers (nur mit setLevel(logging.NOTSET)). Wie ich in einem anderen Kommentar erwähnt habe, verwende ich Pool, sodass ich meine Warteschlange (Proxy) vom Manager beziehen musste, anstatt sie zu verarbeiten, damit sie eingelegt werden kann. Auf diese Weise kann ich die Warteschlange an einen Mitarbeiter innerhalb eines Wörterbuchs übergeben (von dem der größte Teil von argsparse object using abgeleitet ist vars()). Ich denke, am Ende ist dies der beste Ansatz für MS Windows, dem fork () fehlt und die @ zzzeak-Lösung kaputt geht.
Mlt
@mlt Ich denke, Sie könnten auch eine Multiprozessor-Warteschlange in den Init einfügen, anstatt einen Manager zu verwenden (siehe Antwort auf stackoverflow.com/questions/25557686/… - es geht um Sperren, aber ich glaube, es funktioniert auch für Warteschlangen)
fantastisch
@fantabolous Das funktioniert nicht unter MS Windows oder einer anderen Plattform, die fehlt fork. Auf diese Weise hat jeder Prozess seine eigene unabhängige nutzlose Warteschlange. Der zweite Ansatz in der verknüpften Frage / Antwort funktioniert auf solchen Plattformen nicht. Dies ist ein Weg zu nicht portierbarem Code.
Mlt
@mlt Interessant. Ich verwende Windows und es scheint für mich in Ordnung zu sein - nicht lange nach meinem letzten Kommentar habe ich einen Pool von Prozessen eingerichtet, die a multiprocessing.Queuemit dem Hauptprozess teilen , und seitdem benutze ich ihn ständig. Ich werde nicht behaupten zu verstehen, warum es funktioniert.
fantastisch
10

Alle aktuellen Lösungen sind mithilfe eines Handlers zu stark an die Protokollierungskonfiguration gekoppelt. Meine Lösung hat die folgende Architektur und Funktionen:

  • Sie können jede verwenden gewünschte Protokollierungskonfiguration verwenden
  • Die Protokollierung erfolgt in einem Daemon-Thread
  • Sicheres Herunterfahren des Dämons mithilfe eines Kontextmanagers
  • Die Kommunikation zum Protokollierungsthread erfolgt über multiprocessing.Queue
  • In Unterprozessen werden logging.Logger(und bereits definierte Instanzen) gepatcht, um alle Datensätze an die Warteschlange zu senden
  • Neu : Formatieren Sie Traceback und Nachricht vor dem Senden an die Warteschlange, um Beizfehler zu vermeiden

Code mit Verwendungsbeispiel und Ausgabe finden Sie in der folgenden Liste: https://gist.github.com/schlamar/7003737

schlamar
quelle
Es sei denn , ich bin etwas fehlt, dann ist dies nicht wirklich ein Dämon - Thread, da man nie eingestellt daemon_thread.daemonzu True. Ich musste dies tun, damit mein Python-Programm ordnungsgemäß beendet wird, wenn im Kontextmanager eine Ausnahme auftritt.
blah238
Ich musste auch zu fangen, log und schlucken durch das Ziel geworfen Ausnahmen funcin logged_call, andernfalls wird die Ausnahme wäre mit anderen protokollierten Ausgabe verstümmelt bekommen. Hier ist meine modifizierte Version davon: gist.github.com/blah238/8ab79c4fe9cdb254f5c37abfc5dc85bf
blah238
8

Da wir die Multiprozessprotokollierung für so viele Publisher und einen Abonnenten (Listener) darstellen können, ist die Verwendung von ZeroMQ zur Implementierung von PUB-SUB-Messaging in der Tat eine Option.

Darüber hinaus implementiert das PyZMQ- Modul, die Python-Bindungen für ZMQ, PUBHandler , ein Objekt zum Veröffentlichen von Protokollierungsnachrichten über einen zmq.PUB-Socket.

Im Web gibt es eine Lösung für die zentralisierte Protokollierung von verteilten Anwendungen mit PyZMQ und PUBHandler, die problemlos für die lokale Arbeit mit mehreren Veröffentlichungsprozessen verwendet werden kann.

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
quelle
6

Ich mag auch die Antwort von zzzeek, ​​aber Andre hat Recht, dass eine Warteschlange erforderlich ist, um ein Verstümmeln zu verhindern. Ich hatte etwas Glück mit der Pfeife, sah aber ein Gurgeln, was etwas zu erwarten ist. Die Implementierung erwies sich als schwieriger als ich dachte, insbesondere aufgrund der Ausführung unter Windows, wo es einige zusätzliche Einschränkungen für globale Variablen und andere Dinge gibt (siehe: Wie wird Python Multiprocessing unter Windows implementiert? ).

Aber ich habe es endlich geschafft. Dieses Beispiel ist wahrscheinlich nicht perfekt, daher sind Kommentare und Vorschläge willkommen. Es wird auch nicht unterstützt, den Formatierer oder etwas anderes als den Root-Logger festzulegen. Grundsätzlich müssen Sie den Logger in jedem der Poolprozesse mit der Warteschlange neu starten und die anderen Attribute im Logger einrichten.

Auch hier sind Vorschläge zur Verbesserung des Codes willkommen. Ich kenne noch nicht alle Python-Tricks :-)

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
quelle
1
Ich frage mich, ob if 'MainProcess' == multiprocessing.current_process().name:anstelle des Passierens verwendet werden kann child?
Mlt
Wenn jemand anderes versucht, einen Prozesspool anstelle separater Prozessobjekte unter Windows zu verwenden, sollte erwähnt werden, dass der Manager zum Übergeben der Warteschlange an Unterprozesse verwendet werden soll, da er nicht direkt ausgewählt werden kann.
Mlt
Diese Implementierung hat bei mir gut funktioniert. Ich habe es so geändert, dass es mit einer beliebigen Anzahl von Handlern funktioniert. Auf diese Weise können Sie Ihren Root-Handler nicht multiprozessorisch konfigurieren und dann, wo es sicher ist, die Warteschlange zu erstellen, die Root-Handler an diesen übergeben, sie löschen und diesen zum einzigen Handler machen.
Jaxor24
3

Veröffentlichen Sie einfach irgendwo Ihre Instanz des Loggers. Auf diese Weise können die anderen Module und Clients Ihre API verwenden, um den Logger abzurufen, ohne dies zu müssen import multiprocessing.

Javier
quelle
1
Das Problem dabei ist, dass die Multiprozessor-Logger unbenannt erscheinen, sodass Sie den Nachrichtenstrom nicht einfach entschlüsseln können. Vielleicht wäre es möglich, sie nach der Erstellung zu benennen, was das Betrachten vernünftiger machen würde.
cdleary
Veröffentlichen Sie einen Logger für jedes Modul oder exportieren Sie verschiedene Abschlüsse, die den Logger mit dem Modulnamen verwenden. Der Punkt ist, andere Module Ihre API verwenden zu lassen
Javier
Auf jeden Fall vernünftig (und +1 von mir!), Aber ich würde es vermissen, einfach import logging; logging.basicConfig(level=logging.DEBUG); logging.debug('spam!')von überall aus in der Lage zu sein und es richtig funktionieren zu lassen.
cdleary
3
Es ist ein interessantes Phänomen, das ich sehe, wenn ich Python verwende, dass wir uns so daran gewöhnen, in 1 oder 2 einfachen Zeilen das zu tun, was wir wollen, dass der einfache und logische Ansatz in anderen Sprachen (z. B. das Veröffentlichen des Multiprocessing-Loggers oder des Wraps) es fühlt sich immer noch wie eine Bürde an. :)
Kylotan
3

Ich mochte die Antwort von zzzeek. Ich würde nur die Pipe durch eine Warteschlange ersetzen, da mehrere Threads / Prozesse, die dasselbe Pipe-Ende zum Generieren von Protokollnachrichten verwenden, verstümmelt werden.

André Cruz
quelle
Ich hatte einige Probleme mit dem Handler, obwohl es nicht so war, dass Nachrichten verstümmelt waren, es würde nur das Ganze aufhören zu funktionieren. Ich habe Pipe in Queue geändert, da dies angemessener ist. Die Fehler, die ich bekam, wurden dadurch jedoch nicht behoben - letztendlich habe ich der Methode receive () einen Versuch / außer hinzugefügt - sehr selten schlägt ein Versuch, Ausnahmen zu protokollieren, fehl und wird dort abgefangen. Sobald ich try / Except hinzugefügt habe, läuft es wochenlang ohne Probleme, und eine Standarddatei ruft ungefähr zwei fehlerhafte Ausnahmen pro Woche ab.
Zzzeek
2

Wie wäre es, wenn Sie die gesamte Protokollierung an einen anderen Prozess delegieren, der alle Protokolleinträge aus einer Warteschlange liest?

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

Teilen Sie einfach LOG_QUEUE über einen der Multiprozessmechanismen oder sogar über die Vererbung, und alles funktioniert einwandfrei!

Sawan
quelle
1

Ich habe eine ähnliche Lösung wie Ironhacker, außer dass ich logging.exception in einem Teil meines Codes verwende und festgestellt habe, dass ich die Ausnahme formatieren musste, bevor ich sie über die Warteschlange zurückgeben konnte, da Tracebacks nicht pickle'able sind:

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
quelle
Ich habe hier ein vollständiges Beispiel in dieser Richtung gefunden .
Aryeh Leib Taurog
1

Unten finden Sie eine Klasse, die in einer Windows-Umgebung verwendet werden kann und ActivePython erfordert. Sie können auch für andere Protokollierungshandler (StreamHandler usw.) erben.

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

Und hier ist ein Beispiel, das die Verwendung demonstriert:

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
quelle
Wahrscheinlich würde die Verwendung von multiprocessing.Lock()anstelle von Windows Mutex die Lösung portabel machen.
Xmedeko
1

Hier ist mein einfacher Hack / Workaround ... nicht der umfassendste, aber leicht zu ändernde und einfacher zu lesen und zu verstehen, denke ich als alle anderen Antworten, die ich vor dem Schreiben gefunden habe:

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
quelle
1

Es gibt dieses tolle Paket

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

Code: https://github.com/jruere/multiprocessing-logging

Installieren:

pip install multiprocessing-logging

Dann füge hinzu:

import multiprocessing_logging

# This enables logs inside process
multiprocessing_logging.install_mp_handler()
Juan Isaza
quelle
3
Diese Bibliothek basiert buchstäblich auf einem anderen Kommentar zum aktuellen SO-Beitrag: stackoverflow.com/a/894284/1698058 .
Chris Hunt
Ursprung: stackoverflow.com/a/894284/1663382 Ich schätze die beispielhafte Verwendung des Moduls zusätzlich zur Dokumentation auf der Homepage.
Liquidgenius
0

Eine der Alternativen besteht darin, die Mehrfachverarbeitungsprotokollierung in eine bekannte Datei zu schreiben und einen atexitHandler zu registrieren , der an diesen Prozessen teilnimmt. Lesen Sie ihn auf stderr zurück. Auf diese Weise erhalten Sie jedoch keinen Echtzeitfluss zu den Ausgabemeldungen auf stderr.

cdleary
quelle
ist der Ansatz, den Sie unten vorschlagen, identisch mit dem aus Ihrem Kommentar hier stackoverflow.com/questions/641420/…
iruvar
0

Wenn in einer Kombination von Sperren, Threads und Gabeln im loggingModul Deadlocks auftreten , wird dies im Fehlerbericht 6721 gemeldet (siehe auch verwandte SO-Frage ).

Es gibt eine kleine Fixup Lösung gepostet hier .

Dadurch werden jedoch nur mögliche Deadlocks behoben logging. Das wird nicht beheben, dass die Dinge vielleicht verstümmelt sind. Siehe die anderen Antworten hier.

Albert
quelle
0

Einfachste Idee wie erwähnt:

  • Holen Sie sich den Dateinamen und die Prozess-ID des aktuellen Prozesses.
  • Richten Sie ein [WatchedFileHandler][1]. Die Gründe für diesen Handler werden hier ausführlich besprochen , aber kurz gesagt, es gibt bestimmte schlechtere Rennbedingungen mit den anderen Logging-Handlern. Dieser hat das kürzeste Fenster für die Rennbedingung.
    • Wählen Sie einen Pfad zum Speichern der Protokolle aus, z. B. / var / log / ...
user1460675
quelle
0

Für alle, die dies benötigen, habe ich einen Decorator für das Paket multiprocessing_logging geschrieben, der den aktuellen Prozessnamen zu den Protokollen hinzufügt, damit klar wird, wer was protokolliert.

Außerdem wird install_mp_handler () ausgeführt, sodass es nicht mehr sinnvoll ist, es vor dem Erstellen eines Pools auszuführen.

Auf diese Weise kann ich sehen, welcher Mitarbeiter welche Protokollnachrichten erstellt.

Hier ist die Blaupause mit einem Beispiel:

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
quelle
-5

Ich überlasse diese Antwort meinen Kindern, die seit Jahrzehnten das gleiche Problem haben und diese Frage auf dieser Website gefunden haben.

Einfachheit gegen Überkomplikation. Verwenden Sie einfach andere Werkzeuge. Python ist fantastisch, aber es wurde nicht entwickelt, um einige Dinge zu tun.

Das folgende Snippet für den logrotate Daemon funktioniert für mich und macht die Dinge nicht zu kompliziert . Planen Sie es stündlich und

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

So installiere ich es (Symlinks funktionieren nicht für logrotate):

sudo cp /directpath/config/logrotate/myconfigname /etc/logrotate.d/myconfigname
sudo cp /etc/cron.daily/logrotate /etc/cron.hourly/logrotate
baldr
quelle