clearlinux/uwsgi
2
0
Fork 0
mirror of https://github.com/clearlinux/uwsgi.git synced 2026-09-06 21:51:30 +00:00
Code Issues Packages Projects Releases Wiki Activity

added persistence for queue

Browse Source
This commit is contained in:
roberto@mrspurr
2011-03-18 20:20:36 +01:00
parent a61d48a451
commit 8b55460156
5 changed files with 139 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files
master.c
+12
View File
@@ -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
plugins/python/uwsgi_pymodule.c
+51 -2
View File
@@ -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, ""},
queue.c
+21
View File
@@ -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;
uwsgi.c
+48 -4
View File
@@ -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;
uwsgi.h
+7
View File
@@ -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);