diff --git a/contrib/spoolqueue/producer.py b/contrib/spoolqueue/producer.py new file mode 100644 index 00000000..ac6008a7 --- /dev/null +++ b/contrib/spoolqueue/producer.py @@ -0,0 +1,11 @@ +from tasksconsumer import enqueue + +def application(env, sr): + + sr('200 OK', [('Content-Type','text/html')]) + + enqueue(queue='fast', pippo='pluto') + + return "Task enqueued" + + diff --git a/contrib/spoolqueue/tasks.py b/contrib/spoolqueue/tasks.py new file mode 100644 index 00000000..5d6f55f0 --- /dev/null +++ b/contrib/spoolqueue/tasks.py @@ -0,0 +1,9 @@ +from tasksconsumer import * + +@queueconsumer('fast', 4) +def fast_queue(arguments): + print "fast", arguments + +@queueconsumer('slow') +def slow_queue(arguments): + print "foobar", arguments diff --git a/contrib/spoolqueue/tasksconsumer.py b/contrib/spoolqueue/tasksconsumer.py new file mode 100644 index 00000000..3647670b --- /dev/null +++ b/contrib/spoolqueue/tasksconsumer.py @@ -0,0 +1,44 @@ +from uwsgidecorators import * +import Queue +from threading import Thread + +queues = {} + +class queueconsumer(object): + + def __init__(self, name, num=1, **kwargs): + self.name = name + self.num = num + self.queue = Queue.Queue() + self.threads = [] + self.func = None + queues[self.name] = self + + + @staticmethod + def consumer(self): + while True: + req = self.queue.get() + print req + self.func(req) + self.queue.task_done() + + def __call__(self, f): + self.func = f + for i in range(self.num): + t = Thread(target=self.consumer,args=(self,)) + self.threads.append(t) + t.daemon = True + t.start() + +@spool +def spooler_enqueuer(arguments): + if 'queue' in arguments: + queue = arguments['queue'] + queues[queue].queue.put(arguments) + else: + raise Exception("You have to specify a queue name") + + +def enqueue(*args, **kwargs): + return spooler_enqueuer.spool(*args, **kwargs) diff --git a/plugins/python/pyutils.c b/plugins/python/pyutils.c index 6bebc625..3784efa2 100644 --- a/plugins/python/pyutils.c +++ b/plugins/python/pyutils.c @@ -22,7 +22,9 @@ PyObject *python_call(PyObject *callable, PyObject *args, int catch) { if (PyErr_ExceptionMatches(PyExc_MemoryError)) { uwsgi_log("Memory Error detected !!!\n"); } - uwsgi.workers[uwsgi.mywid].exceptions++; + // this can be in a spooler or in the master + if (uwsgi.mywid > 0) + uwsgi.workers[uwsgi.mywid].exceptions++; if (!catch) { PyErr_Print(); } diff --git a/spooler.c b/spooler.c index a66fd002..aa696b42 100644 --- a/spooler.c +++ b/spooler.c @@ -46,6 +46,12 @@ pid_t spooler_start() { } } } + + for (i = 0; i < 0xFF; i++) { + if (uwsgi.p[i]->post_fork) { + uwsgi.p[i]->post_fork(); + } + } uwsgi.signal_socket = uwsgi.shared->spooler_signal_pipe[1]; for (i = 0; i < 0xFF; i++) { if (uwsgi.p[i]->spooler_init) {