Mailman 3 April 2007 - Twisted

[Twisted-Python] qtreactor and OSX
by Black April 27, 2007

April 27, 2007

I have a program that uses QT and twisted. I've had no problem running it on a Linux machine, but when I moved it over to a Mac, I get the following error: Traceback (most recent call last): File "cursor_test.py", line 8, in <module> from twisted.internet import qtreactor File "/Library/Frameworks/Python.framework/Versions/2.5/lib/ python2.5/site-packages/twisted/internet/qtreactor.py", line 7, in <module> from twisted.plugins.qtreactor_stub import … [View More]

2 2

[Twisted-Python] newb: Twisted-Goodies asynch cluster and TaskQueue
by Ram Peters April 27, 2007

April 27, 2007

I have several log files to parse every hour. I am thinking of using Twisted-Goodies asynch cluster and TaskQueue. What I wanted to do is assign part of the file or a single file, in the taskqueue and set different client (node) to do the parsing. Number of logfiles and size of log files may increase in the future. My question is, are these right tool to use for this task? Is there any examples that uses asynch cluster and TaskQueue? Thank You.

2 2

[Twisted-Python] Different logging class for .tac file run by twistd
by Thomas Jacob April 26, 2007

April 26, 2007

I would like to replace the default logger of twistd with my own (a version of DailyLogFile). Now it's possible to use startLogging just like in a stand-alone twisted app, but twistd will still start the standard logger (twistd.log), so now I have two different log files. <.tac-file> .... application = service.Application("someapp") ... web = internet.TCPServer(1234, server.Site(..)) web.setServiceParent(application) log.startLogging(DailyLogFile(LOG_FILE, LOG_DIR, LOG_MODE)) <end .… [View More]

4 6

[Twisted-Python] Advice on Debugging Stackless/Twisted Problem involving Deferreds
by Andrew Francis April 25, 2007

April 25, 2007

Hello Colleagues: I am try to debug a problem involving Stackless and Twisted (once again). I don't see why Twisted and Stackless cannot inter-operate. I enclosed an example to illustrate the problem. I use Twisted 2.5 and Stackless 2.5 3.1b3 on Windows XP. The goal of the enclosed programme is to suspend and resume Stackless tasklets making network calls without blocking the entire Stackless Python application. Under the hood, Twisted is used. The problem is that programme prematurely … [View More]terminates - the deferred connected to client.getPage() is never called. I want to figure out exactly what is going on (in the past I have gone on wild goose chases involving threads and reactors). Using print statements and pdb (I find it rough using pdb with Stackless Python), the Twisted symptoms are: 1) ConnectTCP is successfully called. However the actual call is never made (i.e., I monitored my Apache web server's access.log). As a side note, I haven't quite figured out how the HTTPClientFactory interacts with the Twisted reactor. 2) The deferred is never called. 3) Twisted seems to stop in MainLoop(), I am not sure where. 4) No errors are generated. The Stackless symptoms are: A) The programme terminates, implying all the tasklets except the main tasklet are blocked (or disappeared). B) (Connected to A) All the channels are waiting for writers (I look at the channel.__reduce__()). Any insights would help. At this point, I like to focus my attention on the Twisted side since this is where my knowledge is the weakest. I am still reading Twisted code. I am trying to get Manhole running but I am having problems with the gtk library. It would help greatly if it could be pointed out where I should place set_trace()/break points and print statements to see what is happening to the deferred call. If someone is set up for both Twisted and Stackless and run the example, that would be super! On the Stackless side, what am I missing? I haven't detected any TaskletExit exceptions being thrown. I was thinking deadlock, but this seems unlikely. Is there a way to conduct a post-mortem on tasklets? Hopefully I have made a silly mistake... Cheers, Andrew ~~~ #!/usr/bin/env python """ TestProcessor.py April 24th, 2007 The purpose of this programme is to suspend a request handler, make a series of calls involving deferreds, and resume the request handler. This test does not work. The main tasklet makes a series of synchronous API calls : the first is to wait for a http request. The second is to get a web page. The last is to send a http reply. Under the hood, a tasklet "the processor" suspends and resumes tasklets waiting for IO events. The programme works fine for: def execute(self): self.processor.getWebPage() (the programme will just run) and def execute(self): self.processor.getHttpRequest() self.processor.sendHttpResponse() (use a HTML form with method POST) However when def execute(self): self.processor.getHttpRequest() self.processor.getWebPage() self.processor.sendHttpResponse() combination are done, something bad happens and the programme ends pre-maturely """ from twisted.internet.defer import Deferred from twisted.python.failure import Failure from twisted.internet import reactor from twisted.web import client from twisted.web import http import stackless import pdb import time import sys MESSAGE = {0 : "client_response", 1 : "receive_response", 2 : "reply_response" } CLIENT_RESPONSE = 0 RECEIVE_RESPONSE = 1 REPLY_RESPONSE = 2 message = """<html><head></head><body>hello world</body></html>""" def pump(): while (1): stackless.schedule() class Response(object): def __init_(self): return class Request(object): def __init__(self): return """ run Twisted in its own tasklet """ def twistedReactor(): reactor.run() """ a housekeeping data structure to hide how tasklets are suspended and resumed. For now, we use channels. Not sure how easy it is to replace this with capture/remove/insert """ class Activity(object): def __init__(self): self.channel = stackless.channel() return def __repr__(self): return str(self.channel.__reduce__()) def resume(self, response): self.channel.send(response) def suspend(self): print "[suspending]" return self.channel.receive() """ The Twisted Web server """ class Server(object): """ Twisted sends network events to the processor via a channel """ def execute(self, port, requestChannel): MyRequestHandler.requestChannel = requestChannel reactor.listenTCP(port, MyHttpFactory()) return class MyRequestHandler(http.Request): def process(self): print "request handler :", stackless.getcurrent() myChannel = stackless.channel() response = Response() response.type = RECEIVE_RESPONSE response.path = self.path response.channel = myChannel response.body = self.content """ send information back to the processor about the HTTP request. Include a private channel so the processor can send back a reply """ MyRequestHandler.requestChannel.send(response) """ send the reply to the client """ reply = myChannel.receive() self.write(reply.message) self.finish() """ tell the processor that request handler has finished """ ack = Response() ack.type = REPLY_RESPONSE ack.body = None ack.requestId = reply.requestId MyRequestHandler.requestChannel.send(ack) return class MyHttp(http.HTTPChannel): requestFactory = MyRequestHandler class MyHttpFactory(http.HTTPFactory): protocol = MyHttp """ The ClientConnection represents a deferred Twisted call """ class ClientConnection(object): """ include a requestId so the processor can associate completed messages with outstanding requests """ def __init__(self, channel, address, requestId): self.address = address self.channel = channel self.requestId = requestId return def __handleResponse__(self, pageData): print "__handleResponse__ started" print "deferred tasklet :", stackless.getcurrent() response = Response() response.requestId = self.requestId response.type = CLIENT_RESPONSE response.body = pageData self.channel.send(response) print "__handleResponse__ finished" return #ignore errors for now def __handleError__(self, failure): print "__handleError__" return def connect(self): client.getPage(self.address).addCallback(self.__handleResponse__).addErrback(self.__handleError__) return """ The TestTasklet makes a series of calls to the processor """ class TestTasklet(object): def __init__(self, processor): self.processor = processor return """ note - don't care about what is returned. """ def execute(self): try: """ the processor ensures that one call must complete before the other starts. Besides they all run in the same tasklet. """ self.processor.getHttpRequest("/") self.processor.getWebPage('http://localhost') self.processor.sendHttpResponse(message) except : print sys.exc_info() return class TestProcessor(object): def __init__(self, channel): self.flag = True self.channel = channel self.tasklets = {} self.requests = {} self.requestId = 0 self.serverChannel = None self.receiveActivity = None return """ just return the tasklet associated with the getHttpRequest() """ def __match__(self): return self.receiveActivity def __getTaskletEntry__(self): tasklet = stackless.getcurrent() if not self.tasklets.has_key(tasklet): self.tasklets[tasklet] = Activity() return self.tasklets[tasklet] #associate a request with a tasklet #for now don't worry about set_atomic def __addRequest__(self, activity): self.requestId = self.requestId + 1 print "[requestId " + str(self.requestId) + "]" self.requests[self.requestId] = activity return self.requestId #get a request def __getRequest__(self, requestId): print "[resuming request " + str(requestId) + "]" return self.requests[requestId] def __removeRequest__(self, requestId): del self.requests[requestId] """ API """ def getHttpRequest(self, path): print "getHttpRequest started" #pdb.set_trace() activity = self.__getTaskletEntry__() requestId = self.__addRequest__(activity) self.receiveActivity = (requestId, activity) result = activity.suspend() self.serverChannel = result.channel self.__removeRequest__(requestId) print "getHttpRequest finished" return result def getWebPage(self, address): print "getWebPage started" #pdb.set_trace() activity = self.__getTaskletEntry__() requestId = self.__addRequest__(activity) ClientConnection(self.channel, address, requestId).connect() result = activity.suspend() self.__removeRequest__(requestId) print "getWebPage finished" return result def sendHttpResponse(self, message): print "sendHttpResponse started" activity = self.__getTaskletEntry__() requestId = self.__addRequest__(activity) request = Request() request.message = message request.requestId = requestId self.serverChannel.send(request) result = activity.suspend() self.__removeRequest__(requestId) print "sendHttpResponse finished" return """ Event processing """ def processEvents(self): while (self.flag): response = self.channel.receive() if response.type == RECEIVE_RESPONSE: """ for now just blindly match any HTTP request """ requestId, activity = self.__match__() response.requestId = requestId """ get the tasklet associated with the event and resume it, giving it a result in the process """ activity = self.__getRequest__(response.requestId) activity.resume(response) print "finished processing" return try: print "test starting" channel = stackless.channel() #stackless.tasklet(pump)() processor = TestProcessor(channel) print "processor", stackless.tasklet(processor.processEvents)() print "TaskTasklet", stackless.tasklet(TestTasklet(processor).execute)() print "Server", stackless.tasklet(Server().execute)(8000, channel) print "reactor tasklet: ", stackless.tasklet(twistedReactor)() except: print sys.exc_info() reactor.stop() """ should run indefinitely """ while (stackless.getruncount() > 1): stackless.schedule() print "this is the channel count and blocked tasklets:", channel.__reduce__() print "dump request table" for activityObject in processor.requests.values(): print activityObject print "dump RH channel" print processor.receiveActivity __________________________________________________ Do You Yahoo!? Tired of spam? Yahoo! Mail has the best spam protection around http://mail.yahoo.com [View Less]

3 3

Re: [Twisted-Python] Advice on Debugging Stackless/Twisted
by Andrew Francis April 25, 2007

April 25, 2007

Hello Jean-Paul: >Could you re-send the program as an attachment? It >seems to have been mangled fairly badly in the >original >email. Sorry about that. I have enclosed an attachment. Thanks! Cheers, Andrew __________________________________________________ Do You Yahoo!? Tired of spam? Yahoo! Mail has the best spam protection around http://mail.yahoo.com

1 0

[Twisted-Python] WANTED: Conch use-cases!
by Paul Swartz April 23, 2007

April 23, 2007

Hi everyone! It's z3p, the prodigal Conch developer. As part of Google's Summer of Code, I'll be fixing bugs, making the code nicer, and adding some functionality. The big piece of functionality I'll be adding is a connectSSH() function (see http://twistedmatrix.com/trac/ticket/1975). Unfortunately, I haven't been really using Conch for a while, and I'm not sure how actual people are using it. If you're using Conch (or another SSH library, like Paramiko), I'd really appreciate it if you … [View More]

4 4

[Twisted-Python] Weekly Bug Summary
by exarkun＠twistedmatrix.com April 22, 2007

April 22, 2007

1 0

[Twisted-Python] Specialized mail server
by Rob Hoadley April 20, 2007

April 20, 2007

Hello, I'm building a new specialized mail server using twisted.mail and wanted to get some feedback from the twisted community. Some background: The goal of the mail server is to answer and categorize all our bounces. This includes bounces from our outbound mail servers that never make it to a remote server or queue timeouts (i call these conversational bounces) and bounces via the return-path (non-conversational bounces) that made it to the remote mailserver but get bounced after remote … [View More]mailserver acceptance. If you are familiar with non-conversational bounces you'll know that you are not guaranteed to get a bounce email in a standard format. It all depends on the foreign mailserver software and policies. To handle the variations in bounce messages, I'm going to make a bounce server that is configurable via addition of new rules similar to the ipchains. Each rule is a regex and the rules will look similar to the Mailman bounce code regex's. These rules are stored in a db with a whole test system written around the rules to protect against bad rule entry and fallback points to previous rule sets. For the purposes of the server, the idea is to start at the top of the rule chain and work your way down. So really, I'm just talking about a bunch of regex's w/processing if a regex is hit or a final action if the none of the regex's are hit. I've already built a specialized twisted mail server with a db lookup on incoming mail that handles some of our other email (roughly 7-10K emails per day) so I'm pretty familiar with the way twisted.mail works. Where I need some help is in the area of higher-volume matters. I anticipate this mailserver could get between 40-80K emails per day. ( I work for a Email Service Provider, our bounce percentage rates are low ). My questions are as follows: 1) I've used qmail in the past. If you are familiar with qmail, you configure a locals number which tells you the number of concurrent emails qmail can process locally. This is different from the number of smtp connections ( I'm not so much worried about that ). qmail just forks a new process up to the locals number and queues up other mail. in eomRecieved what is the best manner to safeguard that I don't knock over the server by processing 100 emails in short succession with the regex's? 2) Are there any examples of a high volume email server that somewhat follows my processing requirements? I've googled around and not found any examples. It feels like I should be looking for a queue implementation within a twisted.mail server. I've looked at some of the Queue classes within the twisted.mail code and I wanted to know if this is the right direction? 3) If you have experience using twisted.mail have you had any problems with this volume of mail with some sort of minor/major processing on each incoming mail. If you've just saved to disk have you had problems with a load of say 200-300K emails per day? I appreciate any help or comments. thanks -rob [View Less]

1 0

[Twisted-Python] Re: [Twisted-commits] r20027 - Merge forward so that we can run trial.
by Jonathan Lange April 20, 2007

April 20, 2007

On 4/20/07, David Reid <dreid(a)wolfwood.twistedmatrix.com> wrote: > Author: dreid > Date: Thu Apr 19 22:25:02 2007 > New Revision: 20027 > > Added: > branches/endpoints-1442-2/twisted/internet/endpoints.py > - copied unchanged from r20025, /branches/endpoints-1442/twisted/internet/endpoints.py > branches/endpoints-1442-2/twisted/test/test_address.py > - copied unchanged from r20025, /branches/endpoints-1442/twisted/test/test_address.py > … [View More]

1 0

[Twisted-Python] Reactor hangs on second call to reactor.run
by Ed Manlove April 19, 2007

April 19, 2007

I am creating a Plone product which uses Twisted's nntp module to fetch newsgroup postings and display them within my Plone site. Within a Plone template I call the Plone External Method, XXXnewsreader. The first time I view the page containing this external method it runs fine but the second time it hangs. I've run the python debugger and it appears to be running an infinite loop within the function mainloop within the class PosixReactorBase. Can anyone share some ideas about what I … [View More]may be doing wrong here or why the reactor appears to be stuck in the mainlop func? Thanks for any assistance in advance. I am running under Windows XP using Twisted 2.5.0 and using Plone 2.1.3, Zope 2.8.7-final, python 2.3.5. Ed def XXXnewsreader(self): print self.REQUEST.form n = XXXNewsGroup('localhost','local.programming') n.run_XXXNewsGroup() return n.result class XXXDefaultNewsGroupPageProcedure(nntp.NNTPClient): def connectionMade(self): #pdb.set_trace() nntp.NNTPClient.connectionMade(self) self.fetchGroup(self.factory.newsgroup) def gotGroup(self, groupInfo): estNumMsgs, first, last, groupName, type, status = groupInfo first = int(first) last = int(last) start = max(first, last-self.factory.headerCount); self.headersToFetch = range(start, last+1) self.headerCount = len(self.headersToFetch) self.fetchNextHeader() def fetchNextHeader(self): if self.headersToFetch: nextHeaderId = self.headersToFetch.pop(0) print "Fetching header %i of %i..." % ( self.headerCount-len(self.headersToFetch), self.headerCount), self.fetchHead(nextHeaderId) else: self.quit() self.factory.deferred.callback(self.factory.headers) def gotHead(self, header): print "OK" self.factory.handleHeader(header) self.fetchNextHeader() def getHeadFailed(self, errorMessage): print errorMessage self.fetchNextHeader() def getGroupFailed(self, errorMessage): self.factory.defered.errback(Exception(errorMessage)) self.quit() self.transport.loseConnection() def connectionLost(self, error): if not self.factory.deferred.called: self.factory.deferred.errback(error) class XXXDefaultNewsGroupPageFactory(protocol.ClientFactory): protocol = XXXDefaultNewsGroupPageProcedure def __init__(self, deferred, newsgroup, headerCount=5): self.newsgroup = newsgroup self.headerCount = headerCount self.headers = [] self.deferred = deferred # defer.Deferred() def handleHeader(self, headerData): header = HeaderParser(Message).parsestr(headerData, True); self.headers.append(header) temail = parseaddr(header['from']) realName, emailAddress = temail class XXXNewsGroup: def __init__(self, nntpserver, newsgroup): self.server = nntpserver self.newsgroup = newsgroup self.deferred = defer.Deferred() self.result = '' def run_XXXNewsGroup(self): #pdb.set_trace() factory = XXXDefaultNewsGroupPageFactory(self.deferred, self.newsgroup) self.deferred.addCallback( self._toHTMLsuccess).addErrback( self.handleError) self.deferred.addCallback( self.handleSuccess).addErrback( self.handleError) reactor.connectTCP(self.server, 119, factory) reactor.run(installSignalHandlers=0) return self.result def _toHTMLsuccess(self, results): str = '<ul>\n' for header in results: temail = parseaddr(header['from']) realName, emailAddress = temail str += ' <ul>\n' str += ' <li>%s</li>\n' % header['subject'] str += ' <li>%s</li>\n' % realName str += ' <li>%s</li>\n' % header['date'] str += ' </ul>\n' str += '</ul>\n' return str def handleSuccess(self, result): self.result = result pdb.set_trace() reactor.stop() def handleError(self, error): self.result = error.getErrorMessage() print self.result reactor.stop() [View Less]

2 3