#!/usr/bin/env python """ StacklessTwistedReactor.py Andrew Francis August 15th, 2007 This application illustrates what I think a StacklessTwisted "reactor" or processor would look like. Amongst the things it does are: 1)a solution to the "lost" deferred problem. The key is running the Request Handler in its own tasklet. This code will also allow Twisted calls to be make and block in a request handler. 2) Deferreds are hidden away from the client code Hold You Head High - Argent """ 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 = """hello world""" from twisted.python import log, logfile from twisted.internet import task class TwistedException(Exception): def __init__(self, value): self.value = value def __repr__(self): return str(self.value) def pump(): stackless.schedule() class Response(object): def __init_(self): return class Request(object): def __init__(self): return """ run Twisted in its own tasklet """ def twistedReactor(): l = task.LoopingCall(stackless.schedule) l.start(.01) 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): log.msg( "[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): """ let us create a tasklet to do work in the request handler and allow the Reactor to continue """ def process(self): t = stackless.tasklet(self.doWork)() def doWork(self): 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() self.flag = False """ 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): log.msg("__handleResponse__ started") response = Response() response.requestId = self.requestId response.type = CLIENT_RESPONSE response.body = pageData response.error = False self.channel.send(response) log.msg("__handleResponse__ finished") return def __handleError__(self, failure): """ I don't use send_exception() because I think the parameter is a Exception string or class. I want a way of returning the failure. """ response = Response() response.requestId = self.requestId response.type = CLIENT_RESPONSE response.body = failure response.error = True self.channel.send(response) return def connect(self): log.msg( "entering connect") client.getPage(self.address).addCallback(self.__handleResponse__).addErrback(self.__handleError__) log.msg( "exiting connect") return class StacklessTwistedReactor(object): def __init__(self): self.flag = True self.channel = stackless.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 log.msg( "[requestId " + str(self.requestId) + "]") self.requests[self.requestId] = activity return self.requestId #get a request def __getRequest__(self, requestId): log.msg( "[resuming request " + str(requestId) + "]" ) return self.requests[requestId] def __removeRequest__(self, requestId): del self.requests[requestId] """ API """ def getHttpRequest(self, path): log.msg("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) log.msg("getHttpRequest finished") return result #looks like a simple client.getPage() def getPage(self, address): log.msg("getWebPage started") #pdb.set_trace() activity = self.__getTaskletEntry__() requestId = self.__addRequest__(activity) ClientConnection(self.channel, address, requestId).connect() result = activity.suspend() self.__removeRequest__(requestId) if result.error == True: raise TwistedException(result.body) log.msg("getWebPage finished") return result.body def sendHttpResponse(self, message): log.msg( "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) log.msg( "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) log.msg( "finished processing" ) return def run(self): try: log.startLogging(sys.stdout) log.msg("test starting Solution") log.msg("test starting now") """ The processor acts like a dispatcher, taking requests and waiting for responses """ p = stackless.tasklet(self.processEvents)() """ server waits for incoming HTTP requests """ s = stackless.tasklet(Server().execute)(8000, self.channel) r = stackless.tasklet(twistedReactor)() except: log.msg( sys.exc_info()) reactor.stop()