added uwsgi queue subsystem

plugins/python/gil.c

@@ -4,11 +4,13 @@ extern struct uwsgi_server uwsgi;
 extern struct uwsgi_python up;
 
 void gil_real_get() {
+	uwsgi_log("*** REAL GIL ***\n");
 	PyEval_AcquireLock();
 	PyThreadState_Swap((PyThreadState *) pthread_getspecific(up.upt_gil_key));
 }
 
 void gil_real_release() {
+	uwsgi_log("*** REAL RELEASE ***\n");
 	pthread_setspecific(up.upt_gil_key, (void *) PyThreadState_Swap(NULL));
 	PyEval_ReleaseLock();
 }

plugins/python/python_plugin.c

@@ -93,6 +93,14 @@ int uwsgi_python_init() {
 
 	up.main_thread = PyThreadState_Get();
 
+        // by default set a fake GIL (little impact on performance)
+        up.gil_get = gil_fake_get;
+        up.gil_release = gil_fake_release;
+
+        up.swap_ts = simple_swap_ts;
+        up.reset_ts = simple_reset_ts;
+	
+
 	uwsgi_log("Python main interpreter initialized at %p\n", up.main_thread);
 
 	return 1;
@@ -133,7 +141,9 @@ void uwsgi_python_post_fork() {
 	PyErr_Clear();
 #endif
 
-UWSGI_RELEASE_GIL}
+UWSGI_RELEASE_GIL
+
+}
 
 PyObject *uwsgi_pyimport_by_filename(char *name, char *filename) {
 
@@ -556,6 +566,8 @@ void init_uwsgi_embedded_module() {
 
 	init_uwsgi_module_cache(new_uwsgi_module);
 
+	init_uwsgi_module_queue(new_uwsgi_module);
+
 	if (up.extension) {
 		up.extension();
 	}
@@ -697,13 +709,6 @@ void uwsgi_python_init_apps() {
         up.loaders[LOADER_CALLABLE] = uwsgi_callable_loader;
         up.loaders[LOADER_STRING_CALLABLE] = uwsgi_string_callable_loader;
 
-        // by default set a fake GIL (little impact on performance)
-        up.gil_get = gil_fake_get;
-        up.gil_release = gil_fake_release;
-
-        up.swap_ts = simple_swap_ts;
-        up.reset_ts = simple_reset_ts;
-	
 
 	if (up.wsgi_config != NULL) {
 		init_uwsgi_app(LOADER_UWSGI, up.wsgi_config, uwsgi.wsgi_req, up.main_thread);
@@ -736,6 +741,7 @@ void uwsgi_python_enable_threads() {
 	pthread_setspecific(up.upt_gil_key, (void *) PyThreadState_Get());
 	pthread_mutex_init(&up.lock_pyloaders, NULL);
 	pthread_atfork(uwsgi_python_pthread_prepare, uwsgi_python_pthread_parent, uwsgi_python_pthread_child);
+
 	up.gil_get = gil_real_get;
 	up.gil_release = gil_real_release;
 

plugins/python/uwsgi_pymodule.c

@@ -2450,6 +2450,95 @@ PyObject *py_uwsgi_cache_exists(PyObject * self, PyObject * args) {
 
 }
 
+PyObject *py_uwsgi_queue_push(PyObject * self, PyObject * args) {
+
+	Py_ssize_t msglen = 0;
+	char *message ;
+	PyObject *res;
+
+	if (!PyArg_ParseTuple(args, "s#:queue_push", &message, &msglen)) {
+                return NULL;
+        }
+	
+	if (uwsgi.queue_size) {
+		uwsgi_log("locking queue\n");
+                uwsgi_wlock(uwsgi.queue_lock);
+                if (uwsgi_queue_push(message, msglen)) {
+			Py_INCREF(Py_True);
+                        res = Py_True;
+                }
+                else {
+                        Py_INCREF(Py_None);
+                        res = Py_None;
+                }
+                uwsgi_rwunlock(uwsgi.queue_lock);
+		uwsgi_log("unlocked queue\n");
+                return res;
+        }
+
+        Py_INCREF(Py_None);
+        return Py_None;
+	
+}
+
+PyObject *py_uwsgi_queue_pull(PyObject * self, PyObject * args) {
+
+	char *message;
+	uint64_t size;
+	PyObject *res;
+
+	if (!PyArg_ParseTuple(args, ":queue_pull")) {
+                return NULL;
+        }
+
+	if (uwsgi.queue_size) {
+		uwsgi_wlock(uwsgi.queue_lock);
+		message = uwsgi_queue_pull(&size);
+		if (message) {
+                        res = PyString_FromStringAndSize(message, size);
+                }
+                else {
+                        Py_INCREF(Py_None);
+                        res = Py_None;
+                }
+                uwsgi_rwunlock(uwsgi.queue_lock);
+                return res;
+	}
+
+	Py_INCREF(Py_None);     
+        return Py_None;
+
+}
+
+PyObject *py_uwsgi_queue_get(PyObject * self, PyObject * args) {
+
+	long index = 0;
+	uint64_t size = 0;
+	char *message;
+	PyObject *res;
+
+	if (!PyArg_ParseTuple(args, "l:queue_get", &index)) {
+                return NULL;
+        }
+
+	if (uwsgi.queue_size) {
+		uwsgi_rlock(uwsgi.queue_lock);
+		message = uwsgi_queue_get(index, &size);
+		if (message) {
+			res = PyString_FromStringAndSize(message, size);
+		}
+		else {
+			Py_INCREF(Py_None);
+			res = Py_None;
+		}
+		uwsgi_rwunlock(uwsgi.queue_lock);
+		return res;
+	}	
+
+	Py_INCREF(Py_None);
+	return Py_None;
+}
+
 PyObject *py_uwsgi_cache_get(PyObject * self, PyObject * args) {
 
 	char *key;
@@ -2508,6 +2597,13 @@ static PyMethodDef uwsgi_cache_methods[] = {
 	{NULL, NULL},
 };
 
+static PyMethodDef uwsgi_queue_methods[] = {
+	{"queue_get", py_uwsgi_queue_get, METH_VARARGS, ""},
+	{"queue_push", py_uwsgi_queue_push, METH_VARARGS, ""},
+	{"queue_pull", py_uwsgi_queue_pull, METH_VARARGS, ""},
+	{NULL, NULL},
+};
+
 
 
 #ifdef UWSGI_SPOOLER
@@ -2577,6 +2673,24 @@ void init_uwsgi_module_cache(PyObject * current_uwsgi_module) {
 	}
 }
 
+void init_uwsgi_module_queue(PyObject * current_uwsgi_module) {
+        PyMethodDef *uwsgi_function;
+        PyObject *uwsgi_module_dict;
+
+        uwsgi_module_dict = PyModule_GetDict(current_uwsgi_module);
+        if (!uwsgi_module_dict) {
+                uwsgi_log("could not get uwsgi module __dict__\n");
+                exit(1);
+        }
+
+        for (uwsgi_function = uwsgi_queue_methods; uwsgi_function->ml_name != NULL; uwsgi_function++) {
+                PyObject *func = PyCFunction_New(uwsgi_function, NULL);
+                PyDict_SetItemString(uwsgi_module_dict, uwsgi_function->ml_name, func);
+                Py_DECREF(func);
+        }
+}
+
+
 void init_uwsgi_module_sharedarea(PyObject * current_uwsgi_module) {
 	PyMethodDef *uwsgi_function;
 	PyObject *uwsgi_module_dict;

plugins/python/uwsgi_python.h

@@ -25,8 +25,8 @@
 #endif
 
 #ifdef UWSGI_THREADING
-#define UWSGI_GET_GIL (*up.gil_get)();
-#define UWSGI_RELEASE_GIL (*up.gil_release)();
+#define UWSGI_GET_GIL up.gil_get();
+#define UWSGI_RELEASE_GIL up.gil_release();
 #else
 #define UWSGI_GET_GIL
 #define UWSGI_RELEASE_GIL
@@ -213,6 +213,7 @@ void init_uwsgi_module_advanced(PyObject *);
 void init_uwsgi_module_spooler(PyObject *);
 void init_uwsgi_module_sharedarea(PyObject *);
 void init_uwsgi_module_cache(PyObject *);
+void init_uwsgi_module_queue(PyObject *);
 
 PyObject *uwsgi_pyimport_by_filename(char *, char *);