Mailman 3 October 2011 - Twisted

Re: [Twisted-Python] running 1,000,000 tasks, 40 at-a-time
by Terry Jones Oct. 26, 2011

Oct. 26, 2011

Hi Jason > I don't see anything in task.Cooperator to limit the # of > simultaneously-running tasks. Am I missing something? I'm not sure, but I think so. It does limit the number of simultaneously-running tasks. > I guess, technically, could write my own scheduler that limits the # of > simultaneously-running tasks. But, then task.Cooperator isn't really > doing anything useful for me. I think you need to look at the code more closely (and try using it). The fact that there's a generator that is yielding deferreds one by one and that you pass that generator to coop.coiterate the # of simultaneous tasks that you want has the effect of only launching that many tasks at once and the cooperator just pulls a new one from the iterator/generator when one of the previous jobs finishes. It's a bit subtle, but I think it's what you want. Give the code a try & see :-) T

1 0

Re: [Twisted-Python] running 1,000,000 tasks, 40 at-a-time
by Terry Jones Oct. 26, 2011

Oct. 26, 2011

Hi Jason > - Is there a Twisted pattern for managing tasks efficiently that I be > - missing? Sounds like you should be using a Python generator and that you're not. First off, have you read http://jcalderone.livejournal.com/24285.html ? If not, do. Take note of the line work = (callable(elem, *args, **named) for elem in iterable) work is a generator. Make sure you understand every line of that code :-) You might also find https://launchpad.net/txrdq of interest, though it's probably overkill for what you're trying to do. Terry

3 3

[Twisted-Python] perspective broker remote call returning a dictionary
by Ted Toth Oct. 26, 2011

Oct. 26, 2011

I'm trying to return a dictionary with objects as values from a remote call but it's not working. Can anyone tell me what I'm doing wrong or point me at an example that works? dserver.py -------------------------------------------------------------------------------------------------------------------------------- from twisted.spread import pb class Test(): def __init__(self, data): self.data = data class CopyTest(Test, pb.Copyable): def __init__(self, test): Test.__init__(self, test.data) class CopyDict(dict, pb.Copyable): def __init__(self, d): dict.__init__(self) for key in d.keys(): self[key] = CopyTest(d[key]) print key, self[key].data print "Dict: ", pb.jelly(self) print self class dTest(pb.Root): def __init__(self): self.d = {} self.d['A'] = Test("data for A") self.d['D'] = Test("data for D") self.d['C'] = Test("data for C") self.d['B'] = Test("data for B") def remote_getD(self): from dserver import CopyDict foo = CopyDict(self.d) return foo if __name__ == '__main__': from twisted.internet import reactor d = dTest() reactor.listenTCP(8789, pb.PBServerFactory(d)) reactor.run() --------------------------------------------------------------------------------------------------------------- dclient.py --------------------------------------------------------------------------------------------------------------- from twisted.spread import pb from twisted.internet import reactor from dserver import Test, CopyTest, CopyDict import sys class ReceiverDict(dict, pb.RemoteCopy): def setCopyableState(self, state): print "ReceiverDict: ", state self.__dict__ = state # pass pb.setUnjellyableForClass(CopyDict, ReceiverDict) class ReceiverTest(pb.RemoteCopy, Test): def setCopyableState(self, state): print "ReceiverTest: ", state self.__dict__ = state # pass pb.setUnjellyableForClass(CopyTest, ReceiverTest) class Receiver(object): def __init__(self): self.domain = None def runTests(self): self.connect( ).addCallback( lambda _: self.getD( )).addCallback( self._catchFailure).addCallback( lambda _: reactor.stop( )) def connect(self): factory = pb.PBClientFactory( ) reactor.connectTCP("localhost", 8789, factory) return factory.getRootObject( ).addCallback(self._connected) def _connected(self, rootObj): self.domain = rootObj def getD(self): print "Getting object..." return self.domain.callRemote( 'getD').addCallback( self._gotD) def _gotD(self, d): print "Got :", d for key in d.keys(): print "key: ",key print d[key] return d def _catchFailure(self, failure): print "Error:", failure #.getErrorMessage( ) t = Receiver() t.runTests() reactor.run() ----------------------------------------------------------------------------------------------------------- Debug output: [tedx@comms twisted]$ python dserver.py A data for A C data for C B data for B D data for D Dict: ['dserver.CopyDict', ['dictionary']] {'A': <dserver.CopyTest instance at 0x1a0ccb0>, 'C': <dserver.CopyTest instance at 0x1a0ccf8>, 'B': <dserver.CopyTest instance at 0x1a0cd40>, 'D': <dserver.CopyTest instance at 0x1a0cd88>} [tedx@comms twisted]$ python dclient.py Getting object... ReceiverDict: {} Got : {} Error: {}

2 1

[Twisted-Python] running 1,000,000 tasks, 40 at-a-time
by Jason Rennie Oct. 26, 2011

Oct. 26, 2011

The background: I've been using DeferredSemaphore and DeferredList to manage the running of tasks with a resource constraint (only so many tasks can run at the same time). This worked great until I tried to use it to manage millions of tasks. Simply setting them up to run (DeferredSemaphore.run() calls) took appx. 2 hours and used ~5 gigs of ram. This was less efficient than I expected. Note that these numbers don't include time/memory for actually running the tasks, only time/memory to set up the running of the tasks. I've since written a custom task runner that has uses comparatively little setup time/memory by adding a "manager" callback to each task which starts additional tasks as appropriate. My questions: - Is the behavior I'm seeing expected? i.e. are DS/DL only recommended for task management if the # of tasks not too large? Is there a better way to use DS/DL that I might not be thinking of? - Is there a Twisted pattern for managing tasks efficiently that I might be missing? Thanks, Jason

3 2

[Twisted-Python] Weekly Bug Summary
by exarkun＠twistedmatrix.com Oct. 23, 2011

Oct. 23, 2011

1 0

Re: [Twisted-Python] Passing extra arguments along callbacks
by Terry Jones Oct. 22, 2011

Oct. 22, 2011

Hi Jonathan > def resolve(ip): > ptr = '.'.join(ip.split('.')[::-1]) + '.in-addr.arpa' > return client.lookupPointer(ptr).\ > addCallback(lambda x: (ip, x)).\ > addErrback(lambda x: (ip, x)) Your errback isn't raising and isn't returning a Failure, so the processing switches back to the callback chain. If you add an errback that's a real function and it raises something, things will work. E.g. def err(fail, ip): raise Exception(fail, ip) return client.lookupPointer(ptr).\ addCallback(lambda x: (ip, x)).\ addErrback(err, ip) > l = [resolve(ip) for ip in IPS] > d = DeferredList( l, consumeErrors=True).\ > addCallback(display_results).\ > addCallback(lambda x: reactor.stop) Try reactor.stop() Terry

2 1

[Twisted-Python] Passing extra arguments along callbacks
by Jonathan Ballet Oct. 22, 2011

Oct. 22, 2011

Hello, I'm trying to play a little bit with Twisted to discover how it works and I thought it was a nice exercise with the following case. I have a list of IP addresses that I want to get the reverse DNS value. I'm using the answer provided by JP at http://twistedmatrix.com/pipermail/twisted-python/2004-September/008715.html to perform the lookup and it's all good. I'm putting all the results of client.lookupPointer() into a DeferredList to display the result in order when everything is donew, which is also working fine. The code looks like this so far: def resolve(ip): ptr = '.'.join(ip.split('.')[::-1]) + '.in-addr.arpa' return client.lookupPointer(ptr) def display_results(result): for (succeeded, value) in result: if not succeeded: print "unknown" else: (ans, auth, add) = value name = ans[0].payload.name print name d = DeferredList([resolve(ip) for ip in IPS], consumeErrors=True).addCallback(display_results) Now, I would like to have access both to the result of the lookup but also the original IP address, so that I can print, especially, the addresses which failed to be resolved. At first, I added a callback to lookupPointer() like this: client.lookupPointer(ptr).addCallback(lambda r: (ip, r)) and it does the job: in display_results(), I now have the IP address and the result of the lookup. However, I'm stuck how to do this for the failures: I tried to add an errback the same way as I added the callback to lookupPointer(), but then, it seems that DeferredList() sees the result of the errback() as a ssuccess, so `succeeded` is True and I hav to ressort to a isinstance() call to compare to see if my item is a Failure exception or not, like this: def display_results(result): for (succeeded, value) in result: ip, answer = result if isinstance(answer, twisted.python.failure.Failure): succeeded = False ... My question is: is there a simpler way to pass values along the callbacks? Side question: why the reactor is not stopped in the final example below? Thanks! Jonathan ps: my final try looks like this: from twisted.internet import reactor from twisted.internet.defer import DeferredList from twisted.names import client import twisted.python.failure IPS = open('ips.txt').read().split()[0:10] def display_results(result): for (succeeded, value) in result: ip, answer = value if isinstance(answer, twisted.python.failure.Failure): succeeded = False if not succeeded: name = "(unknown)" else: (ans, auth, add) = answer name = str(ans[0].payload.name) print "%16s: %s" % (ip, name) def resolve(ip): ptr = '.'.join(ip.split('.')[::-1]) + '.in-addr.arpa' return client.lookupPointer(ptr).\ addCallback(lambda x: (ip, x)).\ addErrback(lambda x: (ip, x)) l = [resolve(ip) for ip in IPS] d = DeferredList( l, consumeErrors=True).\ addCallback(display_results).\ addCallback(lambda x: reactor.stop) reactor.run()

1 0

[Twisted-Python] Anyone interested in being release manager for 11.1?
by Itamar Turner-Trauring Oct. 21, 2011

Oct. 21, 2011

Twisted is basically at a point where a new version can be released... given a release manager. I'm told it's pretty automated and only takes 45 minutes or so these days. Any volunteers?

5 4

[Twisted-Python] Twisted enterprise deployment models
by Brad Milne Oct. 21, 2011

Oct. 21, 2011

We have a core message-handling platform which encompasses an ever-increasing array of modules and services run inside one Twisted process. This is good as it means the reactor is in control of most of the stuff relevant to it. However, as the number of deployed services within it grows, it becomes increasingly burdensome to have all the modules in a single deployed codebase running in a single launched process. Java has JEE; erlang has hot code loading; x has y; what does Twisted have? pb? To date we have begun introducing RabbitMQ (AMQP), with separate Twisted processes running as adapters at either end of queues. On the surface this fits nicely as it provides distributed architecture, resilience through isolation, and decoupled upgrade paths. On the flipside though, it means more technology and more processes running outside of a singleton Twisted reactor (read: OS process management, which Twisted's designed to be better than). So my questions: 1) I'm right, right? Twisted doesn't have mechanisms or a library for hot swapping code modules or similar solutions for low impact upgrades and high availability? (I originally read about Twisted *plugins* hoping for this, but was disappointed to find how basic they actually were) 2) Is RabbitMQ with txAMQP-empowered Twisted instances on the ends of the queues a sensible, compatible, best-practice approach as agreed by the community? Looking forward to your thoughts Brad

5 8

[Twisted-Python] libamp - A C library implementation of AMP
by Eric P. Mangold Oct. 19, 2011

Oct. 19, 2011

Hello, Peter le Bek and I have been working on a C library implementation of AMP for about 4 months now, and I was hoping to get some feedback and constructive criticism prior to a release. And I need help answering one big question (below). http://amp-protocol.net/Libamp libamp is currently quite usable, for clients and servers, and is geared towards being used with an async I/O library such as libevent, but it could also be driving with synchronous I/O if there was some reason to. And we have very good test coverage! The test suite runs valgrind-clean and includes malloc()-failure tests, so pretty much all error-handling cases are covered and won't leak memory. http://teratorn.org/code/libamp/html_coverage/code/libamp/index.html A number of basic AMP types are currently supported, but support for more advanced types, like DateTime, Decimal, ListOf and AmpList is still needed. The best way to get a feel for the current API would be to read the example programs, e.g.: http://teratorn.org/code/darcsweb/darcsweb.cgi?r=libamp;a=headblob;f=/examp… http://teratorn.org/code/darcsweb/darcsweb.cgi?r=libamp;a=headblob;f=/examp… and have a look at amp.h: http://teratorn.org/code/darcsweb/darcsweb.cgi?r=libamp;a=headblob;f=/amp.h The API centers around AMP_Proto objects, which you create for each AMP peer you're talking to. An AMP_Proto includes a function pointer, which you set with amp_set_write_handler(), that knows how to deliver (or queue) bytes to the other side (be in stdout, a socket, etc). Then you may use amp_call() to call remote commands. When new bytes come in, e.g. via libevent, you drive the AMP_Proto with a call to amp_consume_bytes() which parses the incoming data and performs callbacks to 1) deliver the result of a previous amp_call() 2) invoke a registered responder for a command. amp_add_responder() adds a callback function for handling specific AMP command requests. The basic API for actually handling a command request, or for constructing a response, is low-level, but usable - which leads up to the big question I'm trying to answer: Does AMP need a declarative syntax for defining AMP commands? Think e.g. protobuf's .proto files. With libamp you deal directly with AMP boxes - sets of key/value pairs, and you need to manually verify that a command was called with all of the keys that it expects, and that the values given coerce to the type of data you expect. Viewing amp.h and the examples, you can see how the amp_put_*() and amp_get_*() family of functions is used to encode and decode C types to and from AMP_Box's. This isn't terribly difficult and works well enough in the simple example programs, but I feel there must be a better way, that alleviates some of the manual error checking. Right now I'm leaning toward coming up with some kind of .amp file format, and an `ampc' tool to compile it to C/libamp code. Support for .amp files could be added in Twisted, and you could then share Command declarations between clients and servers written in different languages. What about the next step of declaring AMP "services" - a set of Commands available via a known URL. Why do crummy web-services get to have all the fun? All thoughts and suggestions are welcome. -Eric P.S. Anyone wanting to help with the C code is free to join #amp on Freenode and lend a hand. It could use a once-over by folks familiar with writing C libraries and best-practices thereof.

1 0