[Twisted-Python] Returning a deferred from buildProtocol t.i.p.Factory
Hi all I'm building a simple TCP load balancer based on a code snippet from Glyph on SO: http://stackoverflow.com/questions/4096061/general-question-regarding-wether... It's served me well but I can't work out how to convert Glyphs round robin retrieval of the server endpoint into an async balancing decision in the buildProtocol method of the Factory. If I return a deferred here it fails with an AttributeError: Deferred instance has no attribute 'makeConnection'. Currently I'm working around this by running a separate management loop that periodically updates a dictionary with all the data necessary to make my routing decision so that I can do it without a deferred. This worries me because I may be making my decision on slightly stale data and I'd really like this to be a real time decision as the connection comes in. Does anyone have a clever way of doing this? An example is below. The hashed out buildProtocol is a synchronous decision which works. Thanks in advance! from twisted.internet.protocol import Factory from twisted.protocols.portforward import ProxyFactory from twisted.internet import reactor, defer import random from twisted.python import log import sys log.startLogging(sys.stderr) local_ports = set([1024, 1025]) def port_routing_decision_sync(): return random.choice(list(local_ports)) def port_routing_decision_async(): d = defer.Deferred() reactor.callLater(1, d.callback, port_routing_decision_sync()) return d class Balancer(Factory): # def buildProtocol(self, addr): # port = port_routing_decision_sync() # print "connecting to local port {}".format(port) # return ProxyFactory("127.0.0.1", port).buildProtocol(addr) @defer.inlineCallbacks def buildProtocol(self, addr): port = yield port_routing_decision_async() print "connecting to local port {}".format(port) defer.returnValue(ProxyFactory("127.0.0.1", port).buildProtocol(addr)) def main(): factory = Balancer() reactor.listenTCP(5678, factory) reactor.run() if __name__ == "__main__": main()
Hi Tom, On 11/16/13 8:09 AM, Tom van Neerijnen wrote:
Hi all
I'm building a simple TCP load balancer based on a code snippet from Glyph on SO: http://stackoverflow.com/questions/4096061/general-question-regarding-wether...
It's served me well but I can't work out how to convert Glyphs round robin retrieval of the server endpoint into an async balancing decision in the buildProtocol method of the Factory. If I return a deferred here it fails with an AttributeError: Deferred instance has no attribute 'makeConnection'. [SNIP]
Have you considered using https://pypi.python.org/pypi/txLoadBalancer as the basis for your load-balancer? It supports random, round-robin, least-connections, and weighted, so perhaps it would suit your needs. Hope this helps, L. Daniel Burr
On Nov 16, 2013, at 7:09 AM, Tom van Neerijnen wrote:
Hi all
I'm building a simple TCP load balancer based on a code snippet from Glyph on SO: http://stackoverflow.com/questions/4096061/general-question-regarding-wether...
It's served me well but I can't work out how to convert Glyphs round robin retrieval of the server endpoint into an async balancing decision in the buildProtocol method of the Factory. If I return a deferred here it fails with an AttributeError: Deferred instance has no attribute 'makeConnection'.
Currently I'm working around this by running a separate management loop that periodically updates a dictionary with all the data necessary to make my routing decision so that I can do it without a deferred. This worries me because I may be making my decision on slightly stale data and I'd really like this to be a real time decision as the connection comes in. Does anyone have a clever way of doing this?
Hi Tom, One possibly unexpected aspect of using @inlineCallbacks is that the decorated function itself returns a Deferred. This is why you see the AttributeError...the machinery calling buildProtocol expects an IProtocol instance (or None), and the function is returning a Deferred. `defer.returnValue()` is provided to the callback on that Deferred, not as a direct return value from the decorated function. If you want to make the routing decision when the client connects, then you could push the decision-making process down into the Protocol itself. Here's a quick mockup overriding connectionMade in a ProxyServer protocol subclass. It calls the factory routing function (which may or may not return a deferred), and connects the proxy once the decision has been made. from twisted.internet.protocol import Factory from twisted.protocols.portforward import ProxyServer class Balancer(Factory): protocol = RoutingProxyServer routing_func = port_routing_decision_async class RoutingProxyServer(ProxyServer): def connectionMade(self): # Don't read anything from the connecting client until we have # somewhere to send it to. self.transport.pauseProducing() client = self.clientProtocolFactory() client.setServer(self) if self.reactor is None: from twisted.internet import reactor self.reactor = reactor def connectProxy(host, port): self.reactor.connectTCP(host, port, client) d = maybeDeferred(self.factory.routing_func) d.addCallback(connectProxy) d.addErrback(log.err) Lucas
An example is below. The hashed out buildProtocol is a synchronous decision which works. Thanks in advance!
from twisted.internet.protocol import Factory from twisted.protocols.portforward import ProxyFactory from twisted.internet import reactor, defer import random
from twisted.python import log import sys log.startLogging(sys.stderr)
local_ports = set([1024, 1025])
def port_routing_decision_sync(): return random.choice(list(local_ports))
def port_routing_decision_async(): d = defer.Deferred() reactor.callLater(1, d.callback, port_routing_decision_sync()) return d
class Balancer(Factory): # def buildProtocol(self, addr): # port = port_routing_decision_sync() # print "connecting to local port {}".format(port) # return ProxyFactory("127.0.0.1", port).buildProtocol(addr)
@defer.inlineCallbacks def buildProtocol(self, addr): port = yield port_routing_decision_async() print "connecting to local port {}".format(port) defer.returnValue(ProxyFactory("127.0.0.1", port).buildProtocol(addr))
def main(): factory = Balancer() reactor.listenTCP(5678, factory) reactor.run()
if __name__ == "__main__": main()
Thanks for both those suggestions.
I'll be taking a closer look at txLoadbabancer when I get time as it looks
like it'll take care of a lot of my desired functionality out the box.
To get started tho I'll move my async routing decision call into the
protocol as suggested.
Is there any reason why the internal calls to buildProtocol shouldn't be
wrapped in a maybeDeferred?
On Sat, Nov 16, 2013 at 7:05 PM, Lucas Taylor
On Nov 16, 2013, at 7:09 AM, Tom van Neerijnen wrote:
Hi all
I'm building a simple TCP load balancer based on a code snippet from Glyph on SO: http://stackoverflow.com/questions/4096061/general-question-regarding-wether...
It's served me well but I can't work out how to convert Glyphs round robin retrieval of the server endpoint into an async balancing decision in the buildProtocol method of the Factory. If I return a deferred here it fails with an AttributeError: Deferred instance has no attribute 'makeConnection'.
Currently I'm working around this by running a separate management loop that periodically updates a dictionary with all the data necessary to make my routing decision so that I can do it without a deferred. This worries me because I may be making my decision on slightly stale data and I'd really like this to be a real time decision as the connection comes in. Does anyone have a clever way of doing this?
Hi Tom,
One possibly unexpected aspect of using @inlineCallbacks is that the decorated function itself returns a Deferred. This is why you see the AttributeError...the machinery calling buildProtocol expects an IProtocol instance (or None), and the function is returning a Deferred. `defer.returnValue()` is provided to the callback on that Deferred, not as a direct return value from the decorated function.
If you want to make the routing decision when the client connects, then you could push the decision-making process down into the Protocol itself.
Here's a quick mockup overriding connectionMade in a ProxyServer protocol subclass. It calls the factory routing function (which may or may not return a deferred), and connects the proxy once the decision has been made.
from twisted.internet.protocol import Factory from twisted.protocols.portforward import ProxyServer
class Balancer(Factory): protocol = RoutingProxyServer routing_func = port_routing_decision_async
class RoutingProxyServer(ProxyServer):
def connectionMade(self): # Don't read anything from the connecting client until we have # somewhere to send it to. self.transport.pauseProducing()
client = self.clientProtocolFactory() client.setServer(self)
if self.reactor is None: from twisted.internet import reactor self.reactor = reactor
def connectProxy(host, port): self.reactor.connectTCP(host, port, client)
d = maybeDeferred(self.factory.routing_func) d.addCallback(connectProxy) d.addErrback(log.err)
Lucas
An example is below. The hashed out buildProtocol is a synchronous decision which works. Thanks in advance!
from twisted.internet.protocol import Factory from twisted.protocols.portforward import ProxyFactory from twisted.internet import reactor, defer import random
from twisted.python import log import sys log.startLogging(sys.stderr)
local_ports = set([1024, 1025])
def port_routing_decision_sync(): return random.choice(list(local_ports))
def port_routing_decision_async(): d = defer.Deferred() reactor.callLater(1, d.callback, port_routing_decision_sync()) return d
class Balancer(Factory): # def buildProtocol(self, addr): # port = port_routing_decision_sync() # print "connecting to local port {}".format(port) # return ProxyFactory("127.0.0.1", port).buildProtocol(addr)
@defer.inlineCallbacks def buildProtocol(self, addr): port = yield port_routing_decision_async() print "connecting to local port {}".format(port) defer.returnValue(ProxyFactory("127.0.0.1", port).buildProtocol(addr))
def main(): factory = Balancer() reactor.listenTCP(5678, factory) reactor.run()
if __name__ == "__main__": main()
_______________________________________________ Twisted-Python mailing list Twisted-Python@twistedmatrix.com http://twistedmatrix.com/cgi-bin/mailman/listinfo/twisted-python
On 05:44 pm, twisted@tomvn.com wrote:
Thanks for both those suggestions. I'll be taking a closer look at txLoadbabancer when I get time as it looks like it'll take care of a lot of my desired functionality out the box. To get started tho I'll move my async routing decision call into the protocol as suggested.
Is there any reason why the internal calls to buildProtocol shouldn't be wrapped in a maybeDeferred?
I'm not sure what you mean by "internal" here. I think you might mean "calls to buildProtocol made by reactor implementations". It is a mistake to take this perspective, though. If buildProtocol is allowed to return a Deferred sometimes then *all* callers have to be prepared to handle a Deferred (by using `maybeDeferred` or using some other strategy). This points to the reason *buildProtocol* can't be allowed to return a Deferred. It is already defined as not returning a Deferred and changing this definition would potentially break every call, both those in reactor implementations and elsewhere (and there are plenty of other places that call `buildProtocol`). Whether there is any good reason we could not introduce a new interface that is like `buildProtocol` but is also allowed to return a Deferred is another (more interesting :) question. I can't think of any offhand. The reactor would probably want to avoid monitoring new connections for read or write events until this Deferred fired (but bonus points if it still monitors it for connection lost and cancels the Deferred if this happens before it fires). That's all relatively straightforward to implement though, for someone sufficiently motivated. Jean-Paul
participants (4)
-
exarkun＠twistedmatrix.com -
L. Daniel Burr -
Lucas Taylor -
Tom van Neerijnen