[Twisted-Python] Looking for some advice debugging issue with transport.write calls

[Brian Costlow]

Summary: I am having an issue writing data from an object based on
twisted's LineReceiver. Calling self.transport.write from the protocol,
some data makes it through, some does not. Using tcpdump, I am not seeing
the missing data cross the network interface.

Ubuntu LTS 14.04
Python 2.7.6
Twisted 15.0.0

The app recieves a bunch of messages from various upstream servers (not
mine) and either discards—or bends, folds, spindles and muilates and sends
the modified message to a downstream server. The message format is
text-based, proprietary and not mine.

The way it works:

1. The app puts messages, consisting of a tuple of two strings: an id and
the message data, on a queue. The queue is actually a deque on the factory.
2. There's one client protocol instance for sending, it holds a connection
to downstream server. The factory is a reconnecting one, but only if the
connection drops. (I'm not creating a new protocol instance/connection per
message)
3. There's a method on the protocol that checks for messages in the
factory's queue, and sends them, one at a time. It waits for a short ack
text message back from the receiving end, or times out (my app's timeout,
not a network stack one) before sending the next message, if any. Code for
the sending piece is below.
4. The downstream end (an app not under my control) is not getting all the
messages. If my timeout check runs, and there's not been a returned ack
message yet, invariably the sysadmins for the downstream system report they
never got that message.

def check_for_send(self):
    if not self.in_send_message:
        try:
            self.in_send_message = True
            msg = self.factory.queue.popleft()
            try:
                self.waiting_id = msg[0]
                log.msg("Sending Message: {0}".format(msg[0]))
                raw_msg = "".join([self.MSG_START, msg[1], self.MSG_END,
self.delimiter])
                threads.deferToThread(self._dump_raw_message, msg[0],
raw_msg)
                self.transport.write(raw_msg)
                reactor.callLater(30, self.check_for_timeout, msg[0])
            except:
                log.msg("Error sending Message: {0}".format(msg[0]))
                log.err()
                self.factory.add_message(msg)
                self.waiting_id = ""
                self.in_send_message = False
        except IndexError:
            self.in_send_message = False
        finally:
            reactor.callLater(0, self.check_for_send)
    else:
        reactor.callLater(1, self.check_for_send)

Mostly looking for some advice on further debugging this, as whenever I've
called self.transport.write on a lineReceiver over tcp before, it "just
worked."

Observations.

1. The event loop doesn't appear to be getting stuck. Messages that don't
arrive downstream are interleaved with those that do. And the other part of
the app that is receiving messages from upstream keeps right on trucking.

2. I am now writing copies of the messages to disk, that call is right
before the transport.write call, and all the "missing" messages do end up
on disk.

3. I also fire the callLater for the timeout function on each message. The
disk write and the callLater both happen for missing messages.

4. I ran tcpdump on the interface for about 30 minutes and matched up with
log statements. If I see my protocol timed out waiting for an ack message
in my app's log, I can't find the outgoing message on the interface.

5. I am not seeing any errors in my logs.
-------------- next part --------------
An HTML attachment was scrubbed...
URL: <http://twistedmatrix.com/pipermail/twisted-python/attachments/20150427/c84aa1e0/attachment.html>