I'm writing a small FIX client in Python (2.7, constraint I cannot waive for now, unfortunately).

There will be multiple sockets, connecting to different ports, so I thought I'd create a SocketManager to keep a list of sockets and have select.epoll register() them all.

Each socket is wrapped by a ThinSocketWrapper class, which basically adds a lock, two string buffers (to receive and send) and a Queue (to receive messages):

class ThinSocketWrapper(socket):

    def __init__(self, *args, **kwargs):
        self.BUFLEN = 4096
        self.lock = RLock()

        self.recvBuffer = ''
        self.sendBuffer = ''

        self.recvQueue = kwargs.pop('recvQueue')
        super(ThinSocketWrapper, self).__init__(*args, **kwargs)


  1. The queue is created by the caller and passed in as an argument. This seems a bit messy to me. Would it be better to create the queue from within this class and make it accessible (e.g. through a getRecvQueue method)?

  2. The same way I could create a sendQueue in ThinSocketWrapper and make it accessible through getSendQueue(). The purpose of this queue is that it will be swept by a background thread looking for messages to send. Where should this thread live? Within the ThinSocketWrapper class? Are there better ways of sweeping a queue? Like a select.epoll for queues?

    The SocketManager is basically a thread, whose main loop looks like this:

    class SocketManager(Thread):
        def run(self):
            while True:
                    events = self.epoll.poll(TIMEOUT)
                except IOError as e:
                    if e.errno == errno.EINTR:
                for socket, event in events:
                        if event & (select.EPOLLIN | select.EPOLLPRI):
                        elif event & select.EPOLLOUT:
                        elif event & (select.EPOLLHUP | select.EPOLLERR):

    readEvent is defined as follows:

        def readEvent(self, socket):
                while True:
                    data = socket.recv(socket.BUFLEN)
                    if not data:
                        raise RuntimeError('Socket closed (read)')
                    socket.recvBuffer += data
            except socket.error as e:
                if e.errno != errno.EAGAIN:
                socket.recvBuffer = ''
  3. At the end, when we're done receiving, we basically convert whatever we received in pseudo-messages, that is we copy everything in recvBuffer to recvQueue. I say "pseudo" because it might well be that the endpoint sends us multiple messages back to back, which would be impossible to split unless we look at the content: in that situation, we would end up with multiple messages in the same item of the queue.

    This tells me that packing what has been received into messages is not this class' responsibility (in fact it should be a layer higher, from the point of view of the networking stack). What would be a nicer alternative?



Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Browse other questions tagged or ask your own question.