logging.SocketHandler connections

[oj]

On Nov 16, 2:31 pm, Vinay Sajip <vinay_sa... at yahoo.co.uk> wrote:
> On Nov 15, 3:23 pm, oj <ojee... at gmail.com> wrote:
>
>
>
> > However, initially, I had tried it with a server that closed the
> > connection after receiving each record, and the SocketHandler doesn't
> > seem to behave as advertised.
>
> > My test script was simply this:
>
> [snip]
> > The SocketHandler documentation says that it will re-establish the
> > connection if it has been closed. After a bit of digging, I found a
> > patch had been submitted and accepted that made it back off
> > exponentially. However, this should be time based. Even if I make my
> > sleep here 30 seconds, my server still only receives messages 0, 3, 6
> > and 9.
>
> > I'm concerned that if a connection is lost at some point, I will
> > always lose at least 2 log messages.
>
> > Is there some reason for this that I am not aware of? Have I
> > misunderstood something, or is this a bug?
>
> Not sure yet - can you please post your test server code? Feel free to
> add it as a bug on bugs.python.org, with the code attached, and I'll
> look into it. Include "logging" in the subject/summary line.
>
> Thanks,
>
> Vinay Sajip


Here is the server code. Pretty much directly copied from the example,
aside from not having the the handler loop forever, and queing the
records instead of dealing with the directly.

After further investigation, running the client with a long timeout,
without the server, so that every connection will fail, produces
results much closer to what I would expect. Connections attempted for
each message initially, but not for all of the later messages as the
retry time increases.

The point is kinda moot now, since I guess not closing the connection
is the 'right way' to do this, but I'm still interested in why I see
this behaviour when the server closes the connection.

#!/usr/bin/python

import thread
import cPickle
import logging
import logging.handlers
import SocketServer
import select
import struct
import sys
import time

port = 12345

queue = []
queue_lock = thread.allocate_lock()

class LogRecordStreamHandler(SocketServer.StreamRequestHandler):

        def handle(self):
                chunk = self.connection.recv(4)

                if len(chunk) < 4:
                        print "failed to get 4 bytes in first read"
                        return

                slen = struct.unpack(">L", chunk)[0]
                chunk = self.connection.recv(slen)

                while len(chunk) < slen:
                        chunk = chunk + self.connection.recv(slen -
len(chunk))

                obj = self.unPickle(chunk)
                record = logging.makeLogRecord(obj)

                queue_lock.acquire()
                queue.insert(0, record)
                queue_lock.release()

        def unPickle(self, data):
                return cPickle.loads(data)

class LogRecordSocketReceiver(SocketServer.ThreadingTCPServer):

        def __init__(self, host='localhost',
                        port=port, handler=LogRecordStreamHandler):
                SocketServer.ThreadingTCPServer.__init__(self, (host,
port), handler)
                self.abort = 0
                self.timeout = 1
                self.logname = None

        def serve_until_stopped(self):
                abort = 0
                while not abort:
                        rd, wr, ex =
select.select([self.socket.fileno()],
                                [], [],
                                self.timeout)
                        if rd:
                                self.handle_request()
                        abort = self.abort

def listen():

        server = LogRecordSocketReceiver()
        server.serve_until_stopped()

def main():
        global queue, queue_lock

        logger = logging.getLogger()
        logger.addHandler(logging.StreamHandler(sys.stdout))

        while True:
                record = None
                locked = queue_lock.acquire(0)

                if locked:
                        if len(queue):
                                record = queue.pop()

                        queue_lock.release()

                        if record:
                                logger.handle(record)

                        else:
                                time.sleep(1)

if __name__ == '__main__':

        thread.start_new_thread(listen, ())
        main()