Sebastien Kirche wrote:

Hi,

I am working on a little project to list the servers of the online game «Enemy Territory» and also to list/find a known player. I try to use twisted to simplify the network access. I am new both to twisted and python, but as i am also a programmer, learning is quite easy.

The task consists in 2 times: 1) one UDP request to the master server at idsoftware that returns the list of the active (slaves) servers (currently around 2020 hosts) 2) one UDP request to *each* slave server to obtain characteristics (name, map that is played, ...) and the list of the players

For the time i managed to perform the task 1) and i can obtain the list of the hosts:ports to ask for.

If someone would like to peek an eye on my code, i uploaded it [1].

My problem is now to send around 2000+ non blocking requests and to take the answers as they come back.

Data treatment should not be a problem, but i don't really see how to perform the mass request.

I don't understand clearly the twisted terminology (protocols, factory, ...). I have seen in another script [2] a case where there is just one datagramReceived that checks on the host:port of the datagram to distinguish the answers, but that is kind of a library, and it only shows 2 requests for a test. I have no idea for the mass request.

Would someone be kind to guide me a little about it ? For example by just giving me a snippet for a multiple UDP request with the hosts are in a list, and if the datagramReceived will suit the response treatment.

I hope to be clear enough. TIA.

Ah ha. This is extremely similar to what I've been doing recently (SNMP, with many many agents, lots of concurrent UDP clients). Here's the setup I used (warning: comes with no guarantee, may end your marriage, etc. - specifically, read the warning at the bottom). This is very, very pseudo-code - my actual code has a lot of very confusing non-relevant stuff in it. It uses a queue to buffer the receive events and ensure the UDP socket queue is emptied asap (the queue function exits quickly, but reschdules itself a short time in the future; this short time is enough to let select() run and the data be received - see the recent thread on the mailing list about "scalability with hundreds of clients" and callLater(0, ...) not doing what you might think). That same queue is used to start off the clients' first xmit, meaning they'll be relatively well interspersed and you won't overload the socket *output* buffer either. class Protocol(protocol.DatagramProtocol): def __init__(self): self._queue = [] self.timeouts = [] from twisted.internet import reactor self.reactor = reactor reactor.callLater(1, self.dotimeouts) def dotimeouts(self): now = time.time() while self.timeouts: due, deferred = self.timeouts[0] if due > now: break due, deferred = self.timeouts.pop(0) # Might have been called, don't timeout if so if not deferred.called: deferred.errback(Timeout()) def queue(self, callable, pargs=tuple(), kwargs=dict()): if not self.calls: # Then we won't have schedules an dequeue either # WARNING: this number is important... self.reactor.callLater(0.001, self.dequeue) self._queue.append((callable, pargs, kwargs)) def dequeue(self): if not self.calls: # Shouldn't happen return callable, pargs, kwargs = self._queue.pop(0) callable(*pargs, **kwargs) if self.calls: # If we've more to dequeue, do so # WARNING: this number is also important... self.reactor.callLater(0.001, self.dequeue) def datagramReceived(self, data, addr): # Do stuff, then... pdu = self.parse(data) deferred = self.get_outstanding(pdu, addr) self.queue(deferred.callback, pdu) def query(self, host, op, args): # Do stuff, then pdu = self.encode(op, args) self.transport.write(pdu.bytes(), host) return self.set_outstanding(pdu, host) class Client: def __init__(self, host, protocol): self.protocol = protocol self.host = host self.protocol.queue(self.step1) def step1(self): deferred = self.protocol.query(self.host, 'op', 'args') deferred.addCallbacks(self.step2, self.fail) def step2(self, pdu): for thing, value in pdu.items(): # blah, blah pass deferred = self.protocol.query(self.host, 'op2', None) deferred.addCallbacks(self.step3, self.fail) if __name__=='__main__': import sys from twisted.internet import reactor proto = Protocol() reactor.listenUDP(0, proto) for hostname in sys.argv[1:] if ':' in hostname: hostname, port = hostname.split(':') else: port = defaultport client = Client((hostname, port), proto) reactor.run() Now, I make no claims this is the perfect Twisted app - it ain't. What it *does* show is the only way I've found (which may be entirely my lack of ability) to scalably send multiple hundreds of UDP PDUs without: a) Starving the Twisted mainloop of CPU, meaning select() doesn't get run often enough, and the UDP socket buffer overflows, dropping replies and necessitating retransmits b) Using a lot of sockets, which runs into problems with select() and poll() as well as the system fd limit (solveable with ulimit I'll grant) c) Starving the clients that are slow responders of "cpu" (queue) time The warning however: UDP, lacking flow control, is very easy to get wrong and accidentally DDoS the clients you're trying to talk to. Specifically the value in the reactor.callLater(0.001, ...) determines how often a "task" will be dequeued from the protocol, and therefore how many PDUs you'll send and receive events you'll process per second. Best to start with a) a small number of servers and b) a larger value (lower rate) for that. You should also probably implement some form of variable per-host timeout to get some kind of rate control/ However, I've had a great deal of difficultly making these points understood to other coders, for which I've got two explanations; I'm totally wrong, or it's a very subtle issue. Guess which I think it is :o)

Sebastien Kirche wrote:

Hi,

I am working on a little project to list the servers of the online game «Enemy Territory» and also to list/find a known player. I try to use twisted to simplify the network access. I am new both to twisted and python, but as i am also a programmer, learning is quite easy.

The task consists in 2 times: 1) one UDP request to the master server at idsoftware that returns the list of the active (slaves) servers (currently around 2020 hosts) 2) one UDP request to *each* slave server to obtain characteristics (name, map that is played, ...) and the list of the players

For the time i managed to perform the task 1) and i can obtain the list of the hosts:ports to ask for.

If someone would like to peek an eye on my code, i uploaded it [1].

My problem is now to send around 2000+ non blocking requests and to take the answers as they come back.

Data treatment should not be a problem, but i don't really see how to perform the mass request.

I don't understand clearly the twisted terminology (protocols, factory, ...). I have seen in another script [2] a case where there is just one datagramReceived that checks on the host:port of the datagram to distinguish the answers, but that is kind of a library, and it only shows 2 requests for a test. I have no idea for the mass request.

Would someone be kind to guide me a little about it ? For example by just giving me a snippet for a multiple UDP request with the hosts are in a list, and if the datagramReceived will suit the response treatment.

I hope to be clear enough. TIA.

Ah ha. This is extremely similar to what I've been doing recently (SNMP, with many many agents, lots of concurrent UDP clients). Here's the setup I used (warning: comes with no guarantee, may end your marriage, etc. - specifically, read the warning at the bottom). This is very, very pseudo-code - my actual code has a lot of very confusing non-relevant stuff in it. It uses a queue to buffer the receive events and ensure the UDP socket queue is emptied asap (the queue function exits quickly, but reschdules itself a short time in the future; this short time is enough to let select() run and the data be received - see the recent thread on the mailing list about "scalability with hundreds of clients" and callLater(0, ...) not doing what you might think). That same queue is used to start off the clients' first xmit, meaning they'll be relatively well interspersed and you won't overload the socket *output* buffer either. class Protocol(protocol.DatagramProtocol): def __init__(self): self._queue = [] self.timeouts = [] from twisted.internet import reactor self.reactor = reactor reactor.callLater(1, self.dotimeouts) def dotimeouts(self): now = time.time() while self.timeouts: due, deferred = self.timeouts[0] if due > now: break due, deferred = self.timeouts.pop(0) # Might have been called, don't timeout if so if not deferred.called: deferred.errback(Timeout()) def queue(self, callable, pargs=tuple(), kwargs=dict()): if not self.calls: # Then we won't have schedules an dequeue either # WARNING: this number is important... self.reactor.callLater(0.001, self.dequeue) self._queue.append((callable, pargs, kwargs)) def dequeue(self): if not self.calls: # Shouldn't happen return callable, pargs, kwargs = self._queue.pop(0) callable(*pargs, **kwargs) if self.calls: # If we've more to dequeue, do so # WARNING: this number is also important... self.reactor.callLater(0.001, self.dequeue) def datagramReceived(self, data, addr): # Do stuff, then... pdu = self.parse(data) deferred = self.get_outstanding(pdu, addr) self.queue(deferred.callback, pdu) def query(self, host, op, args): # Do stuff, then pdu = self.encode(op, args) self.transport.write(pdu.bytes(), host) return self.set_outstanding(pdu, host) class Client: def __init__(self, host, protocol): self.protocol = protocol self.host = host self.protocol.queue(self.step1) def step1(self): deferred = self.protocol.query(self.host, 'op', 'args') deferred.addCallbacks(self.step2, self.fail) def step2(self, pdu): for thing, value in pdu.items(): # blah, blah pass deferred = self.protocol.query(self.host, 'op2', None) deferred.addCallbacks(self.step3, self.fail) if __name__=='__main__': import sys from twisted.internet import reactor proto = Protocol() reactor.listenUDP(0, proto) for hostname in sys.argv[1:] if ':' in hostname: hostname, port = hostname.split(':') else: port = defaultport client = Client((hostname, port), proto) reactor.run() Now, I make no claims this is the perfect Twisted app - it ain't. What it *does* show is the only way I've found (which may be entirely my lack of ability) to scalably send multiple hundreds of UDP PDUs without: a) Starving the Twisted mainloop of CPU, meaning select() doesn't get run often enough, and the UDP socket buffer overflows, dropping replies and necessitating retransmits b) Using a lot of sockets, which runs into problems with select() and poll() as well as the system fd limit (solveable with ulimit I'll grant) c) Starving the clients that are slow responders of "cpu" (queue) time The warning however: UDP, lacking flow control, is very easy to get wrong and accidentally DDoS the clients you're trying to talk to. Specifically the value in the reactor.callLater(0.001, ...) determines how often a "task" will be dequeued from the protocol, and therefore how many PDUs you'll send and receive events you'll process per second. Best to start with a) a small number of servers and b) a larger value (lower rate) for that. You should also probably implement some form of variable per-host timeout to get some kind of rate control/ However, I've had a great deal of difficultly making these points understood to other coders, for which I've got two explanations; I'm totally wrong, or it's a very subtle issue. Guess which I think it is :o)