meinheld continuation and threading sample

meinheld_blocking.py

#!/usr/bin/env python

import Queue
import meinheld.middleware
import meinheld.server
import os
import signal
import sys
import threading
import time

class WorkerThread(threading.Thread):
    def __init__(self, queue=None, queue_timeout=None, **kwargs):
        super(WorkerThread, self).__init__(**kwargs)
        self._queue = queue if queue else Queue.Queue()
        self._queue_timeout = queue_timeout if queue_timeout else 1.0
        self._stop = threading.Event()

    def stop(self):
        self._stop.set()

    def stopped(self):
        return self._stop.isSet()

    def run(self):
        while not self.stopped():
            try:
                args = self._queue.get(True, timeout=self._queue_timeout)
                if args:
                    self.something_so_heavy(*args) # do something so heavy and is blocking
            except Queue.Empty:
                pass

    def something_so_heavy(self, timestamp, callback=None):
        print(timestamp)
        time.sleep(3.0)
        if callable(callback):
            callback()

threads = []
queue = Queue.Queue()

def start_threads(n=1):
    global queue, threads
    for i in range(n):
        t = WorkerThread(name="thread-%d" % (i,), queue=queue)
        threads.append(t)
        t.start()

def stop_threads():
    global threads
    for t in threads:
        t.stop()

def _application(environ, start_response):
    global queue

    # enqueue current continuation and then suspend
    c = environ.get(meinheld.middleware.CONTINUATION_KEY, None)
    queue.put((time.time(), c.resume))
    c.suspend()

    body = "hello, world\n"
    headers = [
            ("Content-Type", "text/plain"),
            ("Content-Length", str(len(body))),
    ]

    start_response("200 OK", headers)
    return [body]

application = meinheld.middleware.ContinuationMiddleware(_application)

def main():
    try:
        meinheld.server.listen(("127.0.0.1", 8080))
        meinheld.server.run(application)
    except KeyboardInterrupt: # trap SIGINT
        stop_threads()

cpucount = os.sysconf('SC_NPROCESSORS_ONLN')
start_threads(cpucount * 2)

if __name__ == '__main__':
    main()

run.sh

#!/bin/sh -e

#BIND="127.0.0.1:8080"
BIND="unix:/tmp/gunicorn.sock"

gunicorn -b "${BIND}"  -w 2 --worker-class=egg:meinheld#gunicorn_worker meinheld_blocking:application