From afba91ad619b766abfe9c68fb0942225820b884c Mon Sep 17 00:00:00 2001 From: "roberto@goyle" Date: Wed, 16 Feb 2011 08:12:21 +0100 Subject: [PATCH] added uwsgi queue subsystem --- lock.c | 9 +-- plugins/python/gil.c | 2 + plugins/python/python_plugin.c | 22 +++--- plugins/python/uwsgi_pymodule.c | 114 ++++++++++++++++++++++++++++++++ plugins/python/uwsgi_python.h | 5 +- queue.c | 64 ++++++++++++++++++ signal.c | 2 +- uwsgi.c | 32 +++++++++ uwsgi.h | 17 +++++ uwsgiconfig.py | 2 +- 10 files changed, 253 insertions(+), 16 deletions(-) create mode 100644 queue.c diff --git a/lock.c b/lock.c index cbed570f..13a95971 100644 --- a/lock.c +++ b/lock.c @@ -104,10 +104,6 @@ void uwsgi_unlock(void *lock) { #define UWSGI_LOCK_SIZE sizeof(OSSpinLock) #define UWSGI_RWLOCK_SIZE sizeof(OSSpinLock) -void uwsgi_rwlock_init(void *lock) { uwsgi_lock_init(lock) ;} -void uwsgi_rlock(void *lock) { uwsgi_lock(lock);} -void uwsgi_wlock(void *lock) { uwsgi_lock(lock);} -void uwsgi_rwunlock(void *lock) { uwsgi_unlock(lock); } void uwsgi_lock_init(void *lock) { @@ -124,6 +120,11 @@ void uwsgi_unlock(void *lock) { OSSpinLockUnlock((OSSpinLock *) lock); } +void uwsgi_rwlock_init(void *lock) { uwsgi_lock_init(lock) ;} +void uwsgi_rlock(void *lock) { uwsgi_lock(lock);} +void uwsgi_wlock(void *lock) { uwsgi_lock(lock);} +void uwsgi_rwunlock(void *lock) { uwsgi_unlock(lock); } + #endif diff --git a/plugins/python/gil.c b/plugins/python/gil.c index dfcd0d7f..fd7141e2 100644 --- a/plugins/python/gil.c +++ b/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(); } diff --git a/plugins/python/python_plugin.c b/plugins/python/python_plugin.c index 61e147c2..605adc52 100644 --- a/plugins/python/python_plugin.c +++ b/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; diff --git a/plugins/python/uwsgi_pymodule.c b/plugins/python/uwsgi_pymodule.c index d1c0fd59..11354273 100644 --- a/plugins/python/uwsgi_pymodule.c +++ b/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; diff --git a/plugins/python/uwsgi_python.h b/plugins/python/uwsgi_python.h index c6393013..10e6a15c 100644 --- a/plugins/python/uwsgi_python.h +++ b/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 *); diff --git a/queue.c b/queue.c new file mode 100644 index 00000000..a05e1514 --- /dev/null +++ b/queue.c @@ -0,0 +1,64 @@ +#include "uwsgi.h" + +extern struct uwsgi_server uwsgi; + +char *uwsgi_queue_get(uint64_t index, uint64_t *size) { + + struct uwsgi_queue_item *uqi; + char *ptr = (char *) uwsgi.queue; + + if (index >= uwsgi.queue_size) return NULL; + + ptr = ptr + (uwsgi.queue_blocksize*index); + + uqi = (struct uwsgi_queue_item *) ptr; + + *size = uqi->size; + + return ptr + sizeof(struct uwsgi_queue_item); + +} + +char *uwsgi_queue_pull(uint64_t *size) { + + struct uwsgi_queue_item *uqi; + char *ptr = (char *) uwsgi.queue; + + ptr = ptr + (uwsgi.queue_blocksize*uwsgi.shared->queue_pull_pos); + uqi = (struct uwsgi_queue_item *) ptr; + + if (!uqi->size) return NULL; + + *size = uqi->size; + + uwsgi.shared->queue_pull_pos++; + + if (uwsgi.shared->queue_pull_pos >= uwsgi.queue_size) uwsgi.shared->queue_pull_pos = 0; + + return ptr + sizeof(struct uwsgi_queue_item); + +} + +int uwsgi_queue_push(char *message, uint64_t size) { + + struct uwsgi_queue_item *uqi; + char *ptr = (char *) uwsgi.queue; + + if (size > uwsgi.queue_blocksize + sizeof(struct uwsgi_queue_item)) + return 0; + + ptr = ptr + (uwsgi.queue_blocksize*uwsgi.shared->queue_pos); + uqi = (struct uwsgi_queue_item *) ptr; + + ptr += sizeof(struct uwsgi_queue_item); + + uqi->size = size; + uqi->ts = time(NULL); + memcpy(ptr, message, size); + + uwsgi.shared->queue_pos++; + + if (uwsgi.shared->queue_pos >= uwsgi.queue_size) uwsgi.shared->queue_pos = 0; + + return 1; +} diff --git a/signal.c b/signal.c index d79a242d..362d7f2e 100644 --- a/signal.c +++ b/signal.c @@ -112,6 +112,6 @@ void uwsgi_route_signal(uint8_t sig) { else if (!strcmp(use->receiver, "master")) { } // route to subscribed - else if (!strcmp(use->receiver, "master")) { + else if (!strcmp(use->receiver, "subscribed")) { } } diff --git a/uwsgi.c b/uwsgi.c index dddc3210..08e0deeb 100644 --- a/uwsgi.c +++ b/uwsgi.c @@ -64,6 +64,8 @@ static struct option long_base_options[] = { {"sharedarea", required_argument, 0, 'A'}, {"cache", required_argument, 0, LONG_ARGS_CACHE}, {"cache-blocksize", required_argument, 0, LONG_ARGS_CACHE_BLOCKSIZE}, + {"queue", required_argument, 0, LONG_ARGS_QUEUE}, + {"queue-blocksize", required_argument, 0, LONG_ARGS_QUEUE_BLOCKSIZE}, #ifdef UWSGI_SPOOLER {"spooler", required_argument, 0, 'Q'}, #endif @@ -1061,6 +1063,30 @@ int uwsgi_start(void *v_argv) { } + if (uwsgi.queue_size > 0) { + if (!uwsgi.queue_blocksize) uwsgi.queue_blocksize = 8192; + + if (uwsgi.queue_blocksize % uwsgi.page_size != 0) { + uwsgi_log("invalid queue blocksize %llu: must be a multiple of memory page size (%d bytes)\n", (unsigned long long) uwsgi.queue_blocksize, uwsgi.page_size); + exit(1); + } + + uwsgi.queue = mmap(NULL, uwsgi.queue_blocksize * uwsgi.queue_size, PROT_READ | PROT_WRITE, MAP_SHARED | MAP_ANON, -1, 0); + if (!uwsgi.queue) { + uwsgi_error("mmap()"); + exit(1); + } + + uwsgi.shared->queue_pos = 0; + uwsgi.shared->queue_pull_pos = 0; + + uwsgi.queue_lock = uwsgi_mmap_shared_rwlock(); + uwsgi_rwlock_init(uwsgi.queue_lock); + + uwsgi_log("*** Queue subsystem initialized: %dMB preallocated ***\n", (uwsgi.queue_blocksize * uwsgi.queue_size) / (1024*1024)); + + } + if (uwsgi.cache_max_items > 0) { if (!uwsgi.cache_blocksize) uwsgi.cache_blocksize = UMAX16; @@ -1960,6 +1986,12 @@ end: case LONG_ARGS_CACHE_BLOCKSIZE: uwsgi.cache_blocksize = atoi(optarg); return 1; + case LONG_ARGS_QUEUE: + uwsgi.queue_size = atoi(optarg); + return 1; + case LONG_ARGS_QUEUE_BLOCKSIZE: + uwsgi.queue_blocksize = atoi(optarg); + return 1; case 'A': uwsgi.sharedareasize = atoi(optarg); return 1; diff --git a/uwsgi.h b/uwsgi.h index a87c20db..f3abee4a 100644 --- a/uwsgi.h +++ b/uwsgi.h @@ -232,6 +232,11 @@ struct uwsgi_daemon { int registered; }; +struct uwsgi_queue_item { + uint64_t size; + time_t ts; +}; + // maintain alignment here !!! struct uwsgi_cache_item { // unused @@ -333,6 +338,8 @@ struct uwsgi_opt { #define LONG_ARGS_EMPEROR 17072 #define LONG_ARGS_PRINT 17073 #define LONG_ARGS_CACHE_BLOCKSIZE 17074 +#define LONG_ARGS_QUEUE 17075 +#define LONG_ARGS_QUEUE_BLOCKSIZE 17076 @@ -925,8 +932,12 @@ struct uwsgi_server { struct uwsgi_cache_item *cache_items; void *cache; + uint64_t queue_size; + uint64_t queue_blocksize; + void *queue; void *cache_lock; + void *queue_lock; void *user_lock; void *signal_table_lock; void *fmon_table_lock; @@ -1042,6 +1053,8 @@ struct uwsgi_shared { uint64_t cache_first_available_item; uint64_t cache_unused_stack_ptr; + uint64_t queue_pos; + uint64_t queue_pull_pos; int worker_signal_pipe[2]; struct uwsgi_signal_entry signal_table[0xff]; @@ -1435,3 +1448,7 @@ int is_unix(char *, int); int is_a_number(char *); char *uwsgi_resolve_ip(char *); + +char *uwsgi_queue_get(uint64_t, uint64_t *); +char *uwsgi_queue_pull(uint64_t *); +int uwsgi_queue_push(char *, uint64_t); diff --git a/uwsgiconfig.py b/uwsgiconfig.py index 1130e106..af61f65e 100644 --- a/uwsgiconfig.py +++ b/uwsgiconfig.py @@ -149,7 +149,7 @@ class uConf(object): print("using profile: %s" % filename) self.config.read(filename) self.gcc_list = ['utils', 'protocol', 'socket', 'logging', 'master', 'emperor', - 'plugins', 'lock', 'cache', 'event', 'signal', 'rpc', 'gateway', 'loop', 'uwsgi'] + 'plugins', 'lock', 'cache', 'queue', 'event', 'signal', 'rpc', 'gateway', 'loop', 'uwsgi'] self.cflags = ['-O2', '-Wall', '-Werror', '-D_LARGEFILE_SOURCE', '-D_FILE_OFFSET_BITS=64'] + os.environ.get("CFLAGS", "").split() try: gcc_version = str(spcall("%s -dumpversion" % GCC))