[Web-SIG] Could WSGI handle Asynchronous response?

Tue Jul 29 01:57:19 CEST 2008

On Jul 28, 2008, at 12:52 PM, Etienne Robillard wrote:

> On Mon, 18 Feb 2008 04:23:38 -0800 (PST)
> est <electronixtar at gmail.com> wrote:
>
>> I am writing a small 'comet'-like app using flup, something like
>> this:
<snip>
>> So is WSGI really synchronous? How can I handle asynchronous outputs
>> with flup/WSGI ?

WSGI says that the entire body should be written by the time the wsgi  
application returns. So yes it is really synchronous; as Manlio  
Perillo said in another message it is possible to abuse generators to  
allow a wsgi application to operate in the fashion you desire, but  
both the server and the application have to know how to do this and  
there is no standardization yet.

> maybe start by looking here: http://twistedmatrix.com/trac/browser/trunk/twisted/web2/wsgi.py

web2.wsgi's server doesn't really get around the problem. While it  
does non-blocking i/o for the http request and response, it actually  
calls the wsgi application in a threadpool, because there's no way for  
the wsgi application to return before having generated all of the  
response, and even if there were people's wsgi applications don't work  
this way.

You might want to check out orbited (http://www.orbited.org/), which  
doesn't have anything to do with wsgi, but is a Python comet server  
implemented entirely with non-blocking i/o (using libevent).

However, if you are willing to spend some time getting a custom comet  
server up and running, you could take a look at eventlet (http://pypi.python.org/pypi/eventlet/ 
) and spawning (http://pypi.python.org/pypi/Spawning/). I've been  
working on eventlet for a couple of years precisely to make  
implementing scalable and easy to maintain comet applications possible.

Here's a simple Comet server that uses spawning and eventlet. This  
will give you a comet server that scales to tons of simultaneous  
connections, because eventlet mashes together greenlet (coroutines, or  
light-weight cooperative threads) with non-blocking i/o (select, poll,  
libevent, or libev). This is how Spawning can be used to get around  
the wsgi restriction that the entire body should be written by the  
time the wsgi application returns; since spawning uses greenlets  
instead of posix threads for each wsgi request when --threads=0 is  
passed, many simultaneous wsgi applications can be running waiting for  
Comet events with very little memory and CPU overhead.

Save it in a file called spawningcomet.py and run it with:

	spawn spawningcomet.wsgi_application --threads=0

Then, visit http://localhost:8080 in your browser and run this in  
another terminal:

	python spawningcomet.py hello world

## spawningcomet.py

import struct
import sys
import uuid

from eventlet import api
from eventlet import coros

SEND_EVENT_INTERFACE = ''
SEND_EVENT_PORT = 4200

HTML_TEMPLATE = """<html>
     <head>
         <script type="text/javascript">
<!--
function make_request(event_id) {
     var req = new XMLHttpRequest();

     req.onreadystatechange = function() {
         if (req.readyState == 4) {
             var newdiv = document.createElement("div");

newdiv.appendChild(document.createTextNode(req.responseText));
             document.getElementById("body").appendChild(newdiv);
             var next_event = req.getResponseHeader("X-Next-Event");
             if (next_event) {
                 make_request(next_event);
             }
         }
     }
     req.open("GET", event_id);
     req.send(null);
}

make_request("%s");
-->
         </script>
     </head>
     <body id="body">
         <h1>Dynamic content will appear below</h1>
     </body>
</html>
"""

class Comet(object):
     def __init__(self):
         api.spawn(
             api.tcp_server,
             api.tcp_listener((SEND_EVENT_INTERFACE, SEND_EVENT_PORT)),
             self.read_events_forever)

         self.current_event = {'event': coros.event(), 'next': None}
         self.first_event_id = str(uuid.uuid1())
         self.events = {self.first_event_id: self.current_event}

     def read_events_forever(self, (sock, addr)):
         reader = sock.makefile('r')
         try:
             while True:
                 ## Read the next event value out of the socket
                 valuelen = reader.read(4)
                 if not valuelen:
                     break

                 valuelen, = struct.unpack('!L', valuelen)
                 value = reader.read(valuelen)

                 ## Make a new event and link the current event to it
                 old_event = self.current_event
                 old_event['next'] = str(uuid.uuid1())
                 self.current_event = {
                     'event': coros.event(), 'next': None}
                 self.events[old_event['next']] = self.current_event

                 ## Send the event value to any waiting http requests
                 old_event['event'].send(value)
         finally:
             reader.close()
             sock.close()

     def __call__(self, env, start_response):
         if env['REQUEST_METHOD'] != 'GET':
             start_response('405 Method Not Allowed', [('Content- 
type', 'text/plain')])
             return ['Method Not Allowed\n']

         if not env['PATH_INFO'] or env['PATH_INFO'] == '/':
             start_response('200 OK', [('Content-type', 'text/html')])
             return HTML_TEMPLATE % (self.first_event_id, )

         event = self.events.get(env['PATH_INFO'][1:], None)
         if event is None:
             start_response('404 Not Found', [('Content-type', 'text/ 
plain')])
             return ['Not Found\n']

         value = event['event'].wait()
         start_response('200 OK', [
             ('Content-type', 'text/plain'),
             ('X-Next-Event', event['next'])])
         return [value, '\n']

def send_event(where, value):
     sock = api.connect_tcp(where)
     writer = sock.makefile('w')
     writer.write('%s%s' % (struct.pack('!L', len(value)), value))

if __name__ == '__main__':
     if len(sys.argv) > 1:
         value = ' '.join(sys.argv[1:])
     else:
         value = sys.stdin.read()
     send_event((SEND_EVENT_INTERFACE, SEND_EVENT_PORT), value)
else:
     wsgi_application = Comet()