From 8b55460156dbebd90fdbbedf9bc05e226146a396 Mon Sep 17 00:00:00 2001 From: "roberto@mrspurr" Date: Fri, 18 Mar 2011 20:20:36 +0100 Subject: [PATCH] added persistence for queue --- master.c | 12 ++++++++ plugins/python/uwsgi_pymodule.c | 53 +++++++++++++++++++++++++++++++-- queue.c | 21 +++++++++++++ uwsgi.c | 52 +++++++++++++++++++++++++++++--- uwsgi.h | 7 +++++ 5 files changed, 139 insertions(+), 6 deletions(-) diff --git a/master.c b/master.c index 38e219ea..ef93b7ae 100644 --- a/master.c +++ b/master.c @@ -337,6 +337,12 @@ void master_loop(char **argv, char **environ) { } } + if (uwsgi.queue_store && uwsgi.queue_filesize) { + if (msync(uwsgi.queue, uwsgi.queue_filesize, MS_ASYNC)) { + uwsgi_error("msync()"); + } + } + for (;;) { //uwsgi_log("ready_to_reload %d %d\n", ready_to_reload, uwsgi.numproc); @@ -950,6 +956,12 @@ void master_loop(char **argv, char **environ) { } } + if (uwsgi.queue_store && uwsgi.queue_filesize && uwsgi.queue_store_sync && ((master_cycles % uwsgi.queue_store_sync) == 0)) { + if (msync(uwsgi.queue, uwsgi.queue_filesize, MS_ASYNC)) { + uwsgi_error("msync()"); + } + } + #endif diff --git a/plugins/python/uwsgi_pymodule.c b/plugins/python/uwsgi_pymodule.c index 8b4aa75f..0eaca7d2 100644 --- a/plugins/python/uwsgi_pymodule.c +++ b/plugins/python/uwsgi_pymodule.c @@ -2528,7 +2528,6 @@ PyObject *py_uwsgi_queue_push(PyObject * self, PyObject * args) { } if (uwsgi.queue_size) { - uwsgi_log("locking queue\n"); uwsgi_wlock(uwsgi.queue_lock); if (uwsgi_queue_push(message, msglen)) { Py_INCREF(Py_True); @@ -2539,7 +2538,6 @@ PyObject *py_uwsgi_queue_push(PyObject * self, PyObject * args) { res = Py_None; } uwsgi_rwunlock(uwsgi.queue_lock); - uwsgi_log("unlocked queue\n"); return res; } @@ -2611,6 +2609,56 @@ PyObject *py_uwsgi_queue_get(PyObject * self, PyObject * args) { return Py_None; } +PyObject *py_uwsgi_queue_last(PyObject * self, PyObject * args) { + + long num = 0; + uint64_t size = 0; + char *message; + PyObject *res, *zero; + uint64_t base; + + if (!PyArg_ParseTuple(args, "l:queue_last", &num)) { + return NULL; + } + + if (uwsgi.queue_size) { + res = PyList_New(0); + uwsgi_rlock(uwsgi.queue_lock); + if (uwsgi.shared->queue_pos > 0) { + base = uwsgi.shared->queue_pos-1; + } + else { + base = uwsgi.queue_size-1; + } + if (num > (long)uwsgi.queue_size) num = uwsgi.queue_size; + while(num) { + message = uwsgi_queue_get(base, &size); + if (message && size) { + zero = PyString_FromStringAndSize(message, size); + PyList_Append(res, zero); + Py_DECREF(zero); + } + else { + uwsgi_rwunlock(uwsgi.queue_lock); + return res; + } + if (base > 0) { + base--; + } + else { + base = uwsgi.queue_size-1; + } + num--; + } + 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; @@ -2676,6 +2724,7 @@ static PyMethodDef uwsgi_cache_methods[] = { static PyMethodDef uwsgi_queue_methods[] = { {"queue_get", py_uwsgi_queue_get, METH_VARARGS, ""}, + {"queue_last", py_uwsgi_queue_last, METH_VARARGS, ""}, {"queue_push", py_uwsgi_queue_push, METH_VARARGS, ""}, {"queue_pull", py_uwsgi_queue_pull, METH_VARARGS, ""}, {"queue_slot", py_uwsgi_queue_slot, METH_VARARGS, ""}, diff --git a/queue.c b/queue.c index 5b434220..76af634b 100644 --- a/queue.c +++ b/queue.c @@ -19,6 +19,25 @@ char *uwsgi_queue_get(uint64_t index, uint64_t *size) { } +void uwsgi_queue_fix() { + + uint64_t i; + char *value; + uint64_t size; + + for(i=0;i< uwsgi.queue_size;i++) { + // valid record ? + value = uwsgi_queue_get(i, &size); + if (value && size) { + uwsgi.shared->queue_pos++; + } + else { + return; + } + } +} + + char *uwsgi_queue_pop(uint64_t *size) { struct uwsgi_queue_item *uqi; @@ -70,6 +89,8 @@ int uwsgi_queue_push(char *message, uint64_t size) { if (size > uwsgi.queue_blocksize + sizeof(struct uwsgi_queue_item)) return 0; + if (!size) return 0; + ptr = ptr + (uwsgi.queue_blocksize*uwsgi.shared->queue_pos); uqi = (struct uwsgi_queue_item *) ptr; diff --git a/uwsgi.c b/uwsgi.c index 92e4e7ac..49e66459 100644 --- a/uwsgi.c +++ b/uwsgi.c @@ -71,6 +71,8 @@ static struct option long_base_options[] = { {"cache-store-sync", required_argument, 0, LONG_ARGS_CACHE_STORE_SYNC}, {"queue", required_argument, 0, LONG_ARGS_QUEUE}, {"queue-blocksize", required_argument, 0, LONG_ARGS_QUEUE_BLOCKSIZE}, + {"queue-store", required_argument, 0, LONG_ARGS_QUEUE_STORE}, + {"queue-store-sync", required_argument, 0, LONG_ARGS_QUEUE_STORE_SYNC}, #ifdef UWSGI_SPOOLER {"spooler", required_argument, 0, 'Q'}, #endif @@ -1309,15 +1311,50 @@ int uwsgi_start(void *v_argv) { exit(1); } - uwsgi.queue = mmap(NULL, uwsgi.queue_blocksize * uwsgi.queue_size, PROT_READ | PROT_WRITE, MAP_SHARED | MAP_ANON, -1, 0); + + uwsgi.shared->queue_pos = 0; + uwsgi.shared->queue_pull_pos = 0; + + if (uwsgi.queue_store) { + uwsgi.queue_filesize = uwsgi.queue_blocksize * uwsgi.queue_size; + int queue_fd; + struct stat qst; + + if (stat(uwsgi.queue_store, &qst)) { + uwsgi_log("creating a new queue store file: %s\n", uwsgi.queue_store); + queue_fd = open(uwsgi.queue_store, O_CREAT | O_RDWR, S_IRUSR | S_IWUSR ); + if (queue_fd >= 0) { + // fill the queue store + if (ftruncate(queue_fd, uwsgi.queue_filesize)) { + uwsgi_log("ftruncate()"); + exit(1); + } + } + } + else { + if ((size_t)qst.st_size != uwsgi.queue_filesize || !S_ISREG(qst.st_mode)) { + uwsgi_log("invalid queue store file. Please remove it or fix queue blocksize/items to match its size\n"); + exit(1); + } + queue_fd = open(uwsgi.queue_store, O_CREAT | O_RDWR, S_IRUSR | S_IWUSR ); + uwsgi_log("recovered queue from backing store file: %s\n", uwsgi.queue_store); + } + + if (queue_fd < 0) { + uwsgi_error_open(uwsgi.queue_store); + exit(1); + } + uwsgi.queue = mmap(NULL, uwsgi.queue_filesize, PROT_READ | PROT_WRITE, MAP_SHARED, queue_fd, 0); + uwsgi_queue_fix(); + } + else { + 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); @@ -2352,6 +2389,13 @@ end: case LONG_ARGS_CACHE_BLOCKSIZE: uwsgi.cache_blocksize = atoi(optarg); return 1; + case LONG_ARGS_QUEUE_STORE: + uwsgi.queue_store = optarg; + uwsgi.master_process = 1; + return 1; + case LONG_ARGS_QUEUE_STORE_SYNC: + uwsgi.queue_store_sync = atoi(optarg); + return 1; case LONG_ARGS_QUEUE: uwsgi.queue_size = atoi(optarg); return 1; diff --git a/uwsgi.h b/uwsgi.h index feca046e..26b70c52 100644 --- a/uwsgi.h +++ b/uwsgi.h @@ -374,6 +374,8 @@ struct uwsgi_opt { #define LONG_ARGS_CPU_AFFINITY 17083 #define LONG_ARGS_CACHE_STORE 17084 #define LONG_ARGS_CACHE_STORE_SYNC 17085 +#define LONG_ARGS_QUEUE_STORE 17086 +#define LONG_ARGS_QUEUE_STORE_SYNC 17087 #define UWSGI_OK 0 @@ -1003,6 +1005,9 @@ struct uwsgi_server { uint64_t queue_size; uint64_t queue_blocksize; void *queue; + char *queue_store; + size_t queue_filesize; + int queue_store_sync; void *cache_lock; void *queue_lock; @@ -1651,3 +1656,5 @@ inline int event_queue_read(void); inline int event_queue_write(void); void uwsgi_help(void); + +void uwsgi_queue_fix(void);