Nonblocking file logging handler using ZeroMQ.

nbfilehandler.py

#!/usr/bin/env python

"""
Nonblocking file logging handler using ZeroMQ.
"""

import atexit
import collections
import logging
import multiprocessing as mp
import threading as th
import time

import gevent
import gevent.pool
import gevent.monkey
gevent.monkey.patch_all()
import zmq.green as zmq
from zmq.utils.strtypes import cast_bytes

class RecordSaver(mp.Process):
    def __init__(self, filename, mode, int_data, int_sync, deque_thresh,
                 *args, **kwargs):
        self.filename = filename
        self.mode = mode
        self.int_data = int_data
        self.int_sync = int_sync
        self.deque_thresh = deque_thresh
        self.data = collections.deque()
        self.fh = open(filename, mode)

        super(RecordSaver, self).__init__(*args, **kwargs)

    def run(self):

        self.ctx = zmq.Context()
        self.sock_data = self.ctx.socket(zmq.SUB)
        self.sock_data.setsockopt(zmq.SUBSCRIBE, '')
        self.sock_data.connect(self.int_data)

        # Synchronize to avoid dropping published messages:
        self.sock_sync = self.ctx.socket(zmq.REQ)
        self.sock_sync.connect(self.int_sync)
        self.sock_sync.send('sync')
        self.sock_sync.recv()

        while True:

            # Store received records in a deque:
            msg = self.sock_data.recv()
            if msg == 'quit':
                break
            self.data.append(msg)

            # When the queue length exceeds the threshold, dump its
            # contents to disk:
            if len(self.data) > self.deque_thresh:
                self.fh.writelines(self.data)
                self.data.clear()

        # When logging is over, flush the deque and close the file:
        if len(self.data):
            self.fh.writelines(self.data)
        self.fh.close()

class NBFileHandler(logging.Handler):
    """
    A nonblocking handler class which writes formatted logging records to disk files.
    """

    def __init__(self, filename, mode='a', deque_thresh=1000):
        self.int_data = 'ipc://data'
        self.int_sync = 'ipc://sync'

        # Set up data connection:
        logging.Handler.__init__(self)
        self.ctx = zmq.Context()
        self.socket = self.ctx.socket(zmq.PUB)
        self.socket.bind(self.int_data)

        # Create additional process for queuing emitted records:
        self.saver = RecordSaver(filename, mode,
                                 self.int_data, self.int_sync, deque_thresh)
        self.saver.start()

        # Synchronize to avoid dropping published messages:
        sock_sync = self.ctx.socket(zmq.REP)
        sock_sync.bind(self.int_sync)
        sock_sync.recv()
        sock_sync.send('ack')

        # Not sure why this is necessary to get the saver process to be cleaned
        # up:
        atexit.register(self.close)

        # Use a greenlet group to enable forcing all greenlets to finish by
        # shutdown:
        self.group = gevent.pool.Group()

    def emit(self, record):
        self.socket.send(cast_bytes(self.format(record)+'\n'))

    def handle(self, record):
        self.group.spawn(super(NBFileHandler, self).handle, record)

    def close(self):

        # Wait for greenlets to finish:
        self.group.join()

        # Tell record saving process to stop:
        self.socket.send('quit')

        # Wait for record saving process to finish:
        self.saver.join()
        super(NBFileHandler, self).close()

if __name__ == '__main__':

    N = 20000
    filename = 'log'
    mode = 'w'
    fmt = logging.Formatter('%(asctime)s %(name)s %(levelname)s %(message)s')

    logger = logging.getLogger('log_nb')
    logger.setLevel(logging.DEBUG)
    h = NBFileHandler(filename+'_nb.log', mode)
    h.setFormatter(fmt)
    logger.addHandler(h)
    start = time.time()
    for i in xrange(N):
        logger.info('test %i' % i)
    nb_time = time.time()-start
    print 'nonblocking: ', nb_time

    logger = logging.getLogger('log_bl')
    logger.setLevel(logging.DEBUG)
    h = logging.FileHandler(filename+'_bl.log', mode)
    h.setFormatter(fmt)
    logger.addHandler(h)
    start = time.time()
    for i in xrange(N):
        logger.info('test %i' % i)
    bl_time = time.time()-start
    print 'blocking:    ', bl_time

    print 'nb/bl:       ', nb_time/bl_time