Python: Queue.get() blocks code from checking socket connection

0
votes

I'm running a thread which grabs a message off a queue, sends it to a client, and receives confirmation. If the client disconnects, the thread catches the socket error and terminates. The issue is if the msgQ is empty, the thread never checks the socket connection. Is there a way to set this code up so even if the queue is empty, the socket is checked? (issue being, without a message, there is nothing to send)

Do I need to send a special ping?pong! message if msgQ is empty (and check on the client whether the message is log data or a ping?)? Any help would be appreciated. 

def run(self): 
    while not self._terminate: 
        try: 
            msgs = self.msgQ.get()

            self.sock.send(pickle.dumps(msgs))
            rdy = pickle.loads(self.sock.recv(2097152))
        except socket.error, EOFError: 
            print 'log socketmanager closing'
            self.terminate()
            break
        except Empty: pass
3
pythonsocketsqueue
self.sock.recv() is a blocking call unless you have non-blocking sockets. That means if there is no message on the queue, then issuing a self.sock.recv() will block and will not continue processing until something is received. Is that what you want to do? - Mahmoud Abdelkader

3 Answers

1
votes

If msgQ is empty, then the call to self.msgQ.get() raises an Empty exception and skips the calls to self.sock.send() and self.sock.recv() entirely. Your exception handler for the Empty exception does nothing, so your code will busy wait until something shows up in msgQ without ever calling send or recv.

One possible solution is to use the python select module to check the socket in your exception handler. Something along these lines:

def run(self): 
    while not self._terminate: 
        try: 
            msgs = self.msgQ.get()

            self.sock.send(pickle.dumps(msgs))
            rdy = pickle.loads(self.sock.recv(2097152))
        except socket.error, EOFError: 
            print 'log socketmanager closing'
            self.terminate()
            break
        except Empty:
            results = select.select([], [], [self.sock], 0.5) # timeout of 0.5 seconds
            if self.sock in results[2]:
                print 'exceptional condition on socket'
                self.terminate()
                break
0
votes

If you're using Queue.Queue you can call Queue.get() with optional block and timeout args, or call Queue.get_nowait() which is equivalent to Queue.get(block=False) and will immediately return if the queue is empty:

try:
    # wait for 1/10 second then return
    msgs = self.msgQ.get(timeout=0.1)
except Queue.Empty, qe:
    # handle empty queue
0
votes

I got an idea that is wrapping Queue and Pipe into a class. And then using select module to check socket and pipe, would not use polling to monitor event.

The following sample code for wrapping Queue and Pipe:

class MyQueue(object):
"""docstring for MyQueue"""
def __init__(self, arg):
    super(MyQueue, self).__init__()
    # self.arg = arg
    self._queue = Queue.Queue()
    self._rdfd, self._wrfd = os.pipe()
    return

def enQ(self, item=None, block=True, timeout=None):
    os.write(self._wrfd, '+')
    self._queue.put(item)
    return

def deQ(self, block=True, timeout=None):
    _itm = self._queue.get(block=block, timeout=timeout)
    _msg = os.read(self._rdfd, 1)
    return _itm

def get_notify(self):
    return self._rdfd

    print 'append queue'
    self.inp.append(self._myq.get_notify())
    print 'append socket'
    self.inp.append(self.sock)
    print self.inp

    _slp = self.slp
    _inp = self.inp
    _out = self.out
    _err = self.err
    _in, _out, _err = select.select(_inp, _out, _err, _slp)