1

I'm trying to pass a TCP connection to a Twisted subprocess with adoptStreamConnection, but I can't figure out how to get the Process disposed in the main process after doing that.

My desired flow looks like this:

  • Finish writing any data the Protocol transport has waiting
  • When we know the write buffer is empty send the AMP message to transfer the socket to the subprocess
  • Dispose the Protocol instance in the main process

I tried doing nothing, loseConnection, abortConnection, and monkey patching _socketClose out and using loseConnection. See code here:

import weakref
from twisted.internet import reactor
from twisted.internet.endpoints import TCP4ServerEndpoint
from twisted.python.sendmsg import getsockfam
from twisted.internet.protocol import Factory, Protocol
import twisted.internet.abstract

class EchoProtocol(Protocol):
    def dataReceived(self, data):
        self.transport.write(data)

class EchoFactory(Factory):
    protocol = EchoProtocol

class TransferProtocol(Protocol):
    def dataReceived(self, data):
        self.transport.write('main process still listening!: %s' % (data))

    def connectionMade(self):
        self.transport.write('this message should make it to the subprocess\n')

        # attempt 1: do nothing
        #    everything works fine in the adopt (including receiving the written message), but old protocol still exists (though isn't doing anything)

        # attempt 1: try calling loseConnection
        #    we lose connection before the adopt opens the socket (presumably TCP disconnect message was sent)
        #
        # self.transport.loseConnection()

        # attempt 2: try calling abortConnection
        #    result is same as loseConnection
        #
        # self.transport.abortConnection()

        # attempt 3: try monkey patching the socket close out and calling loseConnection
        #    result: same as doing nothing-- adopt works (including receiving the written message), old protocol still exists
        #
        # def ignored(*args, **kwargs):
        #       print 'ignored :D'
        #
        # self.transport._closeSocket = ignored
        # self.transport.loseConnection()

        reactor.callLater(0, adopt, self.transport.fileno())

class ServerFactory(Factory):
    def buildProtocol(self, addr):
        p = TransferProtocol()
        self.ref = weakref.ref(p)
        return p

f = ServerFactory()

def adopt(fileno):
    print "does old protocol still exist?: %r" % (f.ref())
    reactor.adoptStreamConnection(fileno, getsockfam(fileno), EchoFactory())

port = 1337
endpoint = TCP4ServerEndpoint(reactor, port)
d = endpoint.listen(f)
reactor.run()

In all cases the Protocol object still exists in the main process after the socket has been transferred. How can I clean this up?

Thanks in advance.

2
  • Hi, I'm not really sure - but isn't this what you want? stackoverflow.com/questions/13239244/…
    – iScrE4m
    Commented Jan 9, 2016 at 19:45
  • No, the problem is nothing seems to actually get rid of the Protocol instance. (I.e. There seems to be at least one dangling reference in the Twisted reactor somewhere that's keeping the Protocol instance in scope)
    – Salis
    Commented Jan 9, 2016 at 21:34

1 Answer 1

2

Neither loseConnection nor abortConnection tell the reactor to "forget" about a connection; they close the connection, which is very different; they tell the peer that the connection has gone away.

You want to call self.transport.stopReading() and self.transport.stopWriting() to remove the references to it from the reactor.

Also, it's not valid to use a weakref to test for the remaining existence of an object unless you call gc.collect() first.

As far as making sure that all the data has been sent: the only reliable way to do that is to have an application-level acknowledgement of the data that you've sent. This is why protocols that need a handshake that involves changing protocols - say, for example, STARTTLS - have a specific handshake where the initiator says "I'm going to switch" (and then stops sending), then the peer says "OK, you can switch now". Another way to handle that in this case would be to hand the data you'd like to write to the subprocess via some other channel, instead of passing it to transport.write.

3
  • In that case how can I make sure all the data is done being written before stopping writing in the main process? The FileDescriptor class in twisted.internet.abstract seems to be the relevant implementation. It sets a disconnecting flag to 1 which causes the write function to return main.CONNECTION_DONE from _postLoseConnection. I suppose that means I can monkeypatch _postLoseConnection to serve as a psuedo-callback to know writing is finished. That is obviously not clean or reliable though. Is there a better approach?
    – Salis
    Commented Jan 9, 2016 at 22:29
  • That definitely works, at least in one test, just not clean. Proof of concept: pastebin.com/uy4ZSUtC
    – Salis
    Commented Jan 9, 2016 at 22:34
  • I've added that information to my answer instead of sticking it in a comment.
    – Glyph
    Commented Jan 11, 2016 at 10:45

Your Answer

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

Not the answer you're looking for? Browse other questions tagged or ask your own question.