From b75ba73264d4f2f40e7e6e9bfe6983228e554a73 Mon Sep 17 00:00:00 2001 From: "roberto@sirius" Date: Thu, 9 Dec 2010 19:54:37 +0100 Subject: [PATCH] caching infrastructure (need optimizations) --- gil.c | 19 ----- lock.c | 73 ++++++++++++++++++ loop.c | 2 + master.c | 85 ++++++++++++++++++--- plugins/python/python_plugin.c | 4 + plugins/python/uwsgi_pymodule.c | 125 +++++++++++++++++++++++++++---- plugins/python/uwsgi_python.h | 1 + plugins/python/wsgi_subhandler.c | 1 - utils.c | 1 + uwsgi.c | 72 +++++++++++------- uwsgi.h | 45 +++++++++++ uwsgiconfig.py | 24 +++++- websockets.py | 1 + 13 files changed, 379 insertions(+), 74 deletions(-) delete mode 100644 gil.c create mode 100644 lock.c diff --git a/gil.c b/gil.c deleted file mode 100644 index 4cddb8a7..00000000 --- a/gil.c +++ /dev/null @@ -1,19 +0,0 @@ -#include "uwsgi.h" - -extern struct uwsgi_server uwsgi; - -void gil_real_get() { - PyEval_AcquireLock(); - PyThreadState_Swap((PyThreadState *) pthread_getspecific(uwsgi.ut_save_key)); -} - -void gil_real_release() { - pthread_setspecific(uwsgi.ut_save_key, (void *) PyThreadState_Swap(NULL)); - PyEval_ReleaseLock(); -} - -struct wsgi_request* threaded_current_wsgi_req() { return pthread_getspecific(uwsgi.ut_key); } -struct wsgi_request* simple_current_wsgi_req() { return uwsgi.wsgi_req; } - -void gil_fake_get() {} -void gil_fake_release() {} diff --git a/lock.c b/lock.c new file mode 100644 index 00000000..4e8e4cb0 --- /dev/null +++ b/lock.c @@ -0,0 +1,73 @@ +#include "uwsgi.h" + + +#ifdef UWSGI_LOCK_USE_MUTEX + +// REMEMBER lock must contains space for both pthread_mutex_t and pthread_mutexattr_t !!! +void uwsgi_lock_init(void *lock) { + + if (pthread_mutexattr_init((pthread_mutexattr_t *) lock)) { + uwsgi_log("unable to allocate mutexattr structure\n"); + exit(1); + } + if (pthread_mutexattr_setpshared((pthread_mutexattr_t *) lock, PTHREAD_PROCESS_SHARED)) { + uwsgi_log("unable to share mutex\n"); + exit(1); + } + + if (pthread_mutex_init((pthread_mutex_t *) lock + sizeof(pthread_mutexattr_t), (pthread_mutexattr_t *) lock)) { + uwsgi_log("unable to initialize mutex\n"); + exit(1); + } + + +} + +void uwsgi_lock(void *lock) { + + pthread_mutex_lock((pthread_mutex_t *) lock + sizeof(pthread_mutexattr_t)); +} + +void uwsgi_unlock(void *lock) { + + pthread_mutex_unlock((pthread_mutex_t *) lock + sizeof(pthread_mutexattr_t)); +} + + +#endif + + +#ifdef UWSGI_LOCK_USE_OSX_SPINLOCK + +void uwsgi_lock_init(void *lock) { + + memset(lock, 0, sizeof(OSSpinLock)); +} + +void uwsgi_lock(void *lock) { + + OSSpinLockLock((OSSpinLock *) lock); +} + +void uwsgi_unlock(void *lock) { + + OSSpinLockUnlock((OSSpinLock *) lock); +} + + +#endif + + +#ifdef UWSGI_LOCK_USE_FLOCK + +void uwsgi_lock_init(void *lock) {} + +void uwsgi_lock(void *lock) { + if (flock((int) *lock, LOCK_EX)) { uwsgi_error("flock()"); } +} + +void uwsgi_unlock(void *lock) { + if (flock((int) *lock, LOCK_UN)) { uwsgi_error("flock()"); } +} + +#endif diff --git a/loop.c b/loop.c index 66a8477f..555e81ac 100644 --- a/loop.c +++ b/loop.c @@ -85,10 +85,12 @@ void *simple_loop(void *arg1) { continue; } + if (wsgi_req_recv(wsgi_req)) { continue; } + uwsgi_close_request(wsgi_req); } diff --git a/master.c b/master.c index a895df75..4491bb05 100644 --- a/master.c +++ b/master.c @@ -44,6 +44,8 @@ void master_loop(char **argv, char **environ) { uint64_t master_cycles = 0; uint64_t tmp_counter; + uint64_t current_time = time(NULL); + struct timeval last_respawn; pid_t pid; @@ -59,6 +61,9 @@ void master_loop(char **argv, char **environ) { int master_has_children = 0; + struct pollfd *uwsgi_signal_poll; + char uwsgi_signal; + #ifdef UWSGI_UDP struct pollfd uwsgi_poll[2]; int uwsgi_poll_size = 0; @@ -80,7 +85,7 @@ void master_loop(char **argv, char **environ) { int i,j; - struct timeval check_interval = {.tv_sec = 1,.tv_usec = 0 }; + int check_interval = 1; // release the GIL //UWSGI_RELEASE_GIL @@ -94,6 +99,19 @@ void master_loop(char **argv, char **environ) { signal(SIGUSR1, (void *) &stats); + uwsgi_signal_poll = malloc(sizeof(struct pollfd) * uwsgi.numproc); + if (!uwsgi_signal_poll) { + uwsgi_error("malloc()"); + exit(1); + } + memset(uwsgi_signal_poll, 0, sizeof(struct pollfd) * uwsgi.numproc); + + for(i=1;i<=uwsgi.numproc;i++) { + uwsgi_log("adding %d to signal poll\n", uwsgi.workers[i].pipe[0]); + uwsgi_signal_poll[i-1].fd = uwsgi.workers[i].pipe[0]; + uwsgi_signal_poll[i-1].events = POLLIN; + } + uwsgi.wsgi_req->buffer = uwsgi.async_buf[0]; #ifdef UWSGI_UDP if (uwsgi.udp_socket) { @@ -287,9 +305,9 @@ void master_loop(char **argv, char **environ) { if (diedpid == 0) { /* all processes ok, doing status scan after N seconds */ - check_interval.tv_sec = uwsgi.shared->options[UWSGI_OPTION_MASTER_INTERVAL]; - if (!check_interval.tv_sec) - check_interval.tv_sec = 1; + check_interval = uwsgi.shared->options[UWSGI_OPTION_MASTER_INTERVAL]; + if (!check_interval) + check_interval = 1; #ifdef UWSGI_UDP #ifdef UWSGI_MULTICAST @@ -297,7 +315,7 @@ void master_loop(char **argv, char **environ) { #else if ((uwsgi.udp_socket && udp_fd >= 0)) { #endif - rlen = poll(uwsgi_poll, uwsgi_poll_size, check_interval.tv_sec * 1000); + rlen = poll(uwsgi_poll, uwsgi_poll_size, check_interval * 1000); if (rlen < 0) { uwsgi_error("poll()"); } @@ -405,16 +423,38 @@ void master_loop(char **argv, char **environ) { } else { #endif - select(0, NULL, NULL, NULL, &check_interval); + rlen = poll(uwsgi_signal_poll, uwsgi.numproc, check_interval); + if (rlen < 0) { + uwsgi_error("poll()"); + continue; + } + else if (rlen > 0) { + for(i=0;i 0) { + uwsgi_log("received uwsgi signal %d from worker %d\n", uwsgi_signal, i+1); + } + else { + uwsgi_log_verbose("lost connection with worker %d\n", i+1); + } + } + } + } #ifdef UWSGI_UDP } #endif + current_time = time(NULL); // checking logsize if (uwsgi.logfile) { uwsgi.shared->logsize = lseek(2, 0, SEEK_CUR); } + master_cycles++; working_workers = 0; blocking_workers = 0; @@ -440,9 +480,20 @@ void master_loop(char **argv, char **environ) { uwsgi.workers[0].requests = tmp_counter; } - check_interval.tv_sec = uwsgi.shared->options[UWSGI_OPTION_MASTER_INTERVAL]; - if (!check_interval.tv_sec) - check_interval.tv_sec = 1; + // remove expired cache items + if (uwsgi.cache_max_items > 0) { + for(i=0;ioptions[UWSGI_OPTION_MASTER_INTERVAL]; + if (!check_interval) + check_interval = 1; #ifdef __linux__ @@ -455,7 +506,7 @@ void master_loop(char **argv, char **environ) { for (i = 1; i <= uwsgi.numproc; i++) { /* first check for harakiri */ if (uwsgi.workers[i].harakiri > 0) { - if (uwsgi.workers[i].harakiri < time(NULL)) { + if (uwsgi.workers[i].harakiri < (time_t) current_time) { /* first try to invoke the harakiri() custom handler */ /* TODO */ /* then brutally kill the worker */ @@ -496,7 +547,7 @@ void master_loop(char **argv, char **environ) { } else if (ucn->name[0] != 0 && ucn->type == CLUSTER_NODE_DYNAMIC) { // if the last_seen attr is higher than 30 secs ago, mark the node as dead - if ( (time(NULL) - ucn->last_seen) > 30) { + if ( (current_time - ucn->last_seen) > 30) { uwsgi_log_verbose("no presence announce in the last 30 seconds by node %s, i assume it is dead.\n", ucn->name); ucn->name[0] = 0 ; } @@ -588,15 +639,23 @@ void master_loop(char **argv, char **environ) { } gettimeofday(&last_respawn, NULL); uwsgi.respawn_delta = last_respawn.tv_sec; + // close the communication pipe + close(uwsgi.workers[uwsgi.mywid].pipe[0]); + if (socketpair(AF_UNIX, SOCK_STREAM, 0, uwsgi.workers[uwsgi.mywid].pipe)) { + uwsgi_error("socketpair()\n"); + continue; + } pid = fork(); if (pid == 0) { + // fix the communication pipe + close(uwsgi.workers[uwsgi.mywid].pipe[0]); uwsgi.mypid = getpid(); uwsgi.workers[uwsgi.mywid].pid = uwsgi.mypid; uwsgi.workers[uwsgi.mywid].harakiri = 0; uwsgi.workers[uwsgi.mywid].requests = 0; uwsgi.workers[uwsgi.mywid].failed_requests = 0; uwsgi.workers[uwsgi.mywid].respawn_count++; - uwsgi.workers[uwsgi.mywid].last_spawn = time(NULL); + uwsgi.workers[uwsgi.mywid].last_spawn = current_time; uwsgi.workers[uwsgi.mywid].manage_next_request = 1; break; } @@ -605,6 +664,8 @@ void master_loop(char **argv, char **environ) { } else { uwsgi_log( "Respawned uWSGI worker (new pid: %d)\n", pid); + close(uwsgi.workers[uwsgi.mywid].pipe[1]); + uwsgi_signal_poll[uwsgi.mywid-1].fd = uwsgi.workers[uwsgi.mywid].pipe[0]; #ifdef UWSGI_SPOOLER if (uwsgi.mywid <= 0 && diedpid != uwsgi.shared->spooler_pid) { #else diff --git a/plugins/python/python_plugin.c b/plugins/python/python_plugin.c index fa97c54f..6edee9db 100644 --- a/plugins/python/python_plugin.c +++ b/plugins/python/python_plugin.c @@ -645,6 +645,10 @@ void uwsgi_uwsgi_config(char *module) { if (uwsgi.sharedareasize > 0 && uwsgi.sharedarea) { init_uwsgi_module_sharedarea(new_uwsgi_module); } + + if (uwsgi.cache_max_items > 0) { + init_uwsgi_module_cache(new_uwsgi_module); + } } #endif diff --git a/plugins/python/uwsgi_pymodule.c b/plugins/python/uwsgi_pymodule.c index c2494ea5..353d6978 100644 --- a/plugins/python/uwsgi_pymodule.c +++ b/plugins/python/uwsgi_pymodule.c @@ -7,17 +7,6 @@ char *spool_buffer = NULL; extern struct uwsgi_server uwsgi; extern struct uwsgi_python up; -#ifdef __APPLE__ -#define UWSGI_LOCK OSSpinLockLock((OSSpinLock *) uwsgi.sharedareamutex); -#define UWSGI_UNLOCK OSSpinLockUnlock((OSSpinLock *) uwsgi.sharedareamutex); -#elif defined(__linux__) || defined(__sun__) || defined(__FreeBSD__) -#define UWSGI_LOCK pthread_mutex_lock((pthread_mutex_t *) uwsgi.sharedareamutex + sizeof(pthread_mutexattr_t)); -#define UWSGI_UNLOCK pthread_mutex_unlock((pthread_mutex_t *) uwsgi.sharedareamutex + sizeof(pthread_mutexattr_t)); -#else -#define UWSGI_LOCK if (flock(uwsgi.sockets[0].fd, LOCK_EX)) { uwsgi_error("flock()"); } -#define UWSGI_UNLOCK if (flock(uwsgi.sockets[0].fd, LOCK_UN)) { uwsgi_error("flock()"); } -#endif - #define UWSGI_LOGBASE "[- uWSGI -" char *uwsgi_encode_pydict(PyObject *pydict, uint16_t *size) { @@ -166,6 +155,22 @@ PyObject *py_uwsgi_close(PyObject * self, PyObject * args) { } +PyObject *py_uwsgi_signal(PyObject * self, PyObject * args) { + + char uwsgi_signal; + + if (!PyArg_ParseTuple(args, "B:signal", &uwsgi_signal)) { + return NULL; + } + + uwsgi_log("sending %d to master\n", uwsgi_signal); + write(uwsgi.workers[uwsgi.mywid].pipe[1], &uwsgi_signal, 1); + + Py_INCREF(Py_None); + return Py_None; + +} + PyObject *py_uwsgi_recv_frame(PyObject * self, PyObject * args) { struct wsgi_request *wsgi_req = current_wsgi_req(); @@ -545,8 +550,8 @@ PyObject *py_uwsgi_lock(PyObject * self, PyObject * args) { #else if (uwsgi.numproc > 1 && uwsgi.mypid != uwsgi.workers[0].pid) { #endif - UWSGI_LOCK - UWSGI_SET_LOCKING; + uwsgi_lock(uwsgi.user_lock); + UWSGI_SET_LOCKING; } Py_INCREF(Py_None); @@ -555,8 +560,8 @@ PyObject *py_uwsgi_lock(PyObject * self, PyObject * args) { PyObject *py_uwsgi_unlock(PyObject * self, PyObject * args) { - UWSGI_UNLOCK - UWSGI_UNSET_LOCKING; + uwsgi_unlock(uwsgi.user_lock); + UWSGI_UNSET_LOCKING; Py_INCREF(Py_None); return Py_None; @@ -1766,6 +1771,8 @@ static PyMethodDef uwsgi_advanced_methods[] = { {"unlock", py_uwsgi_unlock, METH_VARARGS, ""}, {"send", py_uwsgi_send, METH_VARARGS, ""}, {"cl", py_uwsgi_cl, METH_VARARGS, ""}, + + {"signal", py_uwsgi_signal, METH_VARARGS, ""}, #ifdef UWSGI_SENDFILE {"sendfile", py_uwsgi_advanced_sendfile, METH_VARARGS, ""}, #endif @@ -1813,6 +1820,77 @@ static PyMethodDef uwsgi_sa_methods[] = { {NULL, NULL}, }; +PyObject *py_uwsgi_cache_del(PyObject * self, PyObject * args) { + + char *key ; + char *value ; + + if (!PyArg_ParseTuple(args, "s:cache_del", &key, &value)) { + return NULL; + } + + + if (uwsgi_cache_del(key, strlen(key))) { + Py_INCREF(Py_None); + return Py_None; + } + + Py_INCREF(Py_True); + return Py_True; + +} + + +PyObject *py_uwsgi_cache_set(PyObject * self, PyObject * args) { + + char *key ; + char *value ; + uint64_t expires = 0; + + if (!PyArg_ParseTuple(args, "ss|i:cache_set", &key, &value, &expires)) { + return NULL; + } + + + if (uwsgi_cache_set(key, strlen(key), value, strlen(value), expires)) { + Py_INCREF(Py_None); + return Py_None; + } + + Py_INCREF(Py_True); + return Py_True; + +} + + +PyObject *py_uwsgi_cache_get(PyObject * self, PyObject * args) { + + char *key ; + uint16_t valsize; + char *value ; + + if (!PyArg_ParseTuple(args, "s:cache_get", &key)) { + return NULL; + } + + value = uwsgi_cache_get(key, strlen(key), &valsize); + + if (value) { + return PyString_FromStringAndSize(value, valsize); + } + + Py_INCREF(Py_None); + return Py_None; + +} + +static PyMethodDef uwsgi_cache_methods[] = { + {"cache_get", py_uwsgi_cache_get, METH_VARARGS, ""}, + {"cache_set", py_uwsgi_cache_set, METH_VARARGS, ""}, + {"cache_del", py_uwsgi_cache_del, METH_VARARGS, ""}, + {NULL, NULL}, +}; + #ifdef UWSGI_SPOOLER @@ -1865,6 +1943,23 @@ void init_uwsgi_module_advanced(PyObject * current_uwsgi_module) { } + void init_uwsgi_module_cache(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_cache_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 9ce6bf6f..b4cfa17d 100644 --- a/plugins/python/uwsgi_python.h +++ b/plugins/python/uwsgi_python.h @@ -195,5 +195,6 @@ void gil_fake_release(void); void init_uwsgi_module_advanced(PyObject *); void init_uwsgi_module_spooler(PyObject *); void init_uwsgi_module_sharedarea(PyObject *); +void init_uwsgi_module_cache(PyObject *); PyObject *uwsgi_pyimport_by_filename(char *, char *); diff --git a/plugins/python/wsgi_subhandler.c b/plugins/python/wsgi_subhandler.c index faa1fe46..8639bb39 100644 --- a/plugins/python/wsgi_subhandler.c +++ b/plugins/python/wsgi_subhandler.c @@ -15,7 +15,6 @@ void *uwsgi_request_subhandler_wsgi(struct wsgi_request *wsgi_req, struct uwsgi_ } */ - wsgi_socket = PyFile_FromFile(wsgi_req->async_post, "wsgi_input", "r", NULL); PyDict_SetItemString(wsgi_req->async_environ, "wsgi.input", wsgi_socket); Py_DECREF(wsgi_socket); diff --git a/utils.c b/utils.c index 4789847d..bcd43f5f 100644 --- a/utils.c +++ b/utils.c @@ -401,6 +401,7 @@ void wsgi_req_setup(struct wsgi_request *wsgi_req, int async_id) { wsgi_req->post_buffering_buf = uwsgi.async_post_buf[wsgi_req->async_id]; } + } int wsgi_req_recv(struct wsgi_request *wsgi_req) { diff --git a/uwsgi.c b/uwsgi.c index 1256f0d7..06e7b55c 100644 --- a/uwsgi.c +++ b/uwsgi.c @@ -58,6 +58,7 @@ static struct option long_base_options[] = { {"max-requests", required_argument, 0, 'R'}, {"socket-timeout", required_argument, 0, 'z'}, {"sharedarea", required_argument, 0, 'A'}, + {"cache", required_argument, 0, LONG_ARGS_CACHE}, #ifdef UWSGI_SPOOLER {"spooler", required_argument, 0, 'Q'}, #endif @@ -861,40 +862,24 @@ options_parsed: } } + uwsgi.user_lock = uwsgi_mmap_shared_lock(); + if (!uwsgi.user_lock) { + uwsgi_error("mmap()"); + exit(1); + } + uwsgi_lock_init(uwsgi.user_lock); + #ifdef UWSGI_EMBEDDED if (uwsgi.sharedareasize > 0) { -#ifndef __OpenBSD__ - uwsgi.sharedareamutex = mmap(NULL, sizeof(pthread_mutexattr_t) + sizeof(pthread_mutex_t), PROT_READ | PROT_WRITE, MAP_SHARED | MAP_ANON, -1, 0); + uwsgi.sharedareamutex = uwsgi_mmap_shared_lock(); if (!uwsgi.sharedareamutex) { uwsgi_error("mmap()"); exit(1); } -#else - uwsgi_log("***WARNING*** the sharedarea on OpenBSD is not SMP-safe. Beware of race conditions !!!\n"); -#endif uwsgi.sharedarea = mmap(NULL, uwsgi.page_size * uwsgi.sharedareasize, PROT_READ | PROT_WRITE, MAP_SHARED | MAP_ANON, -1, 0); if (uwsgi.sharedarea) { uwsgi_log("shared area mapped at %p, you can access it with uwsgi.sharedarea* functions.\n", uwsgi.sharedarea); - -#ifdef __APPLE__ - memset(uwsgi.sharedareamutex, 0, sizeof(OSSpinLock)); -#else -#if !defined(__OpenBSD__) && !defined(__NetBSD__) - if (pthread_mutexattr_init((pthread_mutexattr_t *) uwsgi.sharedareamutex)) { - uwsgi_log("unable to allocate mutexattr structure\n"); - exit(1); - } - if (pthread_mutexattr_setpshared((pthread_mutexattr_t *) uwsgi.sharedareamutex, PTHREAD_PROCESS_SHARED)) { - uwsgi_log("unable to share mutex\n"); - exit(1); - } - if (pthread_mutex_init((pthread_mutex_t *) uwsgi.sharedareamutex + sizeof(pthread_mutexattr_t), (pthread_mutexattr_t *) uwsgi.sharedareamutex)) { - uwsgi_log("unable to initialize mutex\n"); - exit(1); - } -#endif -#endif - + uwsgi_lock_init(uwsgi.sharedareamutex); } else { uwsgi_error("mmap()"); exit(1); @@ -903,6 +888,33 @@ options_parsed: } #endif + if (uwsgi.cache_max_items > 0) { + uwsgi.cache_items = mmap(NULL, sizeof(struct uwsgi_cache_item) * uwsgi.cache_max_items, PROT_READ | PROT_WRITE, MAP_SHARED | MAP_ANON, -1, 0); + if (!uwsgi.cache_items) { + uwsgi_error("mmap()"); + exit(1); + } + + uwsgi.cache = mmap(NULL, 32768 * uwsgi.cache_max_items, PROT_READ | PROT_WRITE, MAP_SHARED | MAP_ANON, -1, 0); + if (!uwsgi.cache) { + uwsgi_error("mmap()"); + exit(1); + } + + for(i=0;icache_first_available_item = 1; + + uwsgi.cache_lock = uwsgi_mmap_shared_lock(); + if (!uwsgi.cache_lock) { + uwsgi_error("mmap()"); + exit(1); + } + uwsgi_lock_init(uwsgi.cache_lock); + } uwsgi.current_wsgi_req = simple_current_wsgi_req; @@ -1298,8 +1310,14 @@ uwsgi.shared->hooks[UWSGI_MODIFIER_PING] = uwsgi_request_ping; //100 uwsgi.respawn_delta = last_respawn.tv_sec; } for (i = 2 - uwsgi.master_process; i < uwsgi.numproc + 1; i++) { + // setup internal signalling system + if (socketpair(AF_UNIX, SOCK_STREAM, 0, uwsgi.workers[i].pipe)) { + uwsgi_error("socketpair()\n"); + exit(1); + } pid = fork(); if (pid == 0) { + close(uwsgi.workers[i].pipe[0]); uwsgi.mypid = getpid(); uwsgi.workers[i].pid = uwsgi.mypid; uwsgi.workers[i].id = i; @@ -1312,6 +1330,7 @@ uwsgi.shared->hooks[UWSGI_MODIFIER_PING] = uwsgi_request_ping; //100 exit(1); } else { uwsgi_log("spawned uWSGI worker %d (pid: %d, cores: %d)\n", i, pid, uwsgi.cores); + close(uwsgi.workers[i].pipe[1]); gettimeofday(&last_respawn, NULL); uwsgi.respawn_delta = last_respawn.tv_sec; } @@ -1772,6 +1791,9 @@ end: case LONG_ARGS_CHECK_INTERVAL: uwsgi.shared->options[UWSGI_OPTION_MASTER_INTERVAL] = atoi(optarg); return 1; + case LONG_ARGS_CACHE: + uwsgi.cache_max_items = atoi(optarg); + return 1; case 'A': uwsgi.sharedareasize = atoi(optarg); return 1; diff --git a/uwsgi.h b/uwsgi.h index 399dc3a7..60349f09 100644 --- a/uwsgi.h +++ b/uwsgi.h @@ -173,6 +173,27 @@ #define UWSGI_LISTEN_QUEUE 511 #endif +#define UWSGI_CACHE_MAX_KEY_SIZE 4071 + +// maintain alignment here !!! +struct uwsgi_cache_item { + + // size of the key + uint16_t keysize; + // djb hash of the key + uint32_t djbhash; + // size of the value (max 64KB) + uint16_t valsize; + // 64bit expiration (0 for immortal) + uint64_t expires; + // 64bit hits + uint64_t hits; + // mark the end of the table + char used; + // key chracters follows... + char key[UWSGI_CACHE_MAX_KEY_SIZE]; +} __attribute__((__packed__)); + struct uwsgi_opt { char *key; char *value; @@ -244,6 +265,7 @@ struct uwsgi_opt { #define LONG_ARGS_CLUSTER 17062 #define LONG_ARGS_CLUSTER_RELOAD 17063 #define LONG_ARGS_CLUSTER_LOG 17064 +#define LONG_ARGS_CACHE 17065 @@ -809,6 +831,13 @@ struct uwsgi_server { char *cluster; int cluster_fd; struct sockaddr_in mc_cluster_addr; + + uint16_t cache_max_items; + struct uwsgi_cache_item *cache_items; + void *cache; + void *cache_lock; + + void *user_lock; }; #define CLUSTER_NODE_STATIC 0 @@ -875,6 +904,9 @@ struct uwsgi_shared { #endif + uint16_t cache_first_available_item; + uint16_t cache_first_available_item_tmp; + }; struct uwsgi_core { @@ -913,6 +945,9 @@ struct uwsgi_worker { int manage_next_request; + // this is used for the internal signalling system + int pipe[2]; + struct uwsgi_core **cores; }; @@ -1191,3 +1226,13 @@ char *generate_socket_name(char *); ssize_t uwsgi_send_message(int, uint8_t, uint8_t, char *, uint16_t, int, size_t, int); char *uwsgi_cluster_best_node(void); + +int uwsgi_cache_set(char *, uint16_t, char *, uint16_t, uint64_t); +int uwsgi_cache_del(char *, uint16_t); +char *uwsgi_cache_get(char *, uint16_t, uint16_t *); + +#define uwsgi_mmap_shared_lock() mmap(NULL, sizeof(pthread_mutexattr_t) + sizeof(pthread_mutex_t), PROT_READ | PROT_WRITE, MAP_SHARED | MAP_ANON, -1, 0) + +void uwsgi_lock_init(void *); +void uwsgi_lock(void *); +void uwsgi_unlock(void *); diff --git a/uwsgiconfig.py b/uwsgiconfig.py index 52b2caef..9ff44046 100644 --- a/uwsgiconfig.py +++ b/uwsgiconfig.py @@ -136,7 +136,7 @@ class uConf(): def __init__(self, filename): self.config = ConfigParser.ConfigParser() self.config.read(filename) - self.gcc_list = ['utils', 'protocol', 'socket', 'logging', 'master', 'plugins', 'loop', 'uwsgi'] + self.gcc_list = ['utils', 'protocol', 'socket', 'logging', 'master', 'plugins', 'lock', 'cache', 'loop', 'uwsgi'] self.cflags = ['-O2', '-Wall', '-Werror', '-D_LARGEFILE_SOURCE', '-D_FILE_OFFSET_BITS=64'] + os.environ.get("CFLAGS", "").split() gcc_version = str(spcall2("%s -v" % GCC)).split('\n')[-1].split()[2] gcc_major = int(gcc_version.split('.')[0]) @@ -153,13 +153,15 @@ class uConf(): def set(self, key, value): self.config.set('uwsgi',key, value) - def get(self,key): + def get(self,key,default=None): try: value = self.config.get('uwsgi', key) if value == "" or value == "false": return None return value except: + if default: + return default return None def depends_on(self, what, dep): @@ -186,6 +188,24 @@ class uConf(): self.libs.remove('-lpthread') self.libs.append('-lroot') + # set locking subsystem + locking_mode = self.get('locking','auto') + + print locking_mode, uwsgi_os + if locking_mode == 'auto': + if uwsgi_os == 'Linux': + locking_mode = 'pthread_mutex' + elif uwsgi_os == 'Darwin': + locking_mode = 'osx_spinlock' + + if locking_mode == 'pthread_mutex': + self.cflags.append('-DUWSGI_LOCK_USE_MUTEX') + elif locking_mode == 'osx_spinlock': + self.cflags.append('-DUWSGI_LOCK_USE_OSX_SPINLOCK') + else: + self.cflags.append('-DUWSGI_LOCK_USE_FLOCK') + + if self.get('embedded'): self.cflags.append('-DUWSGI_EMBEDDED') diff --git a/websockets.py b/websockets.py index 05fbef33..8ac1e00f 100644 --- a/websockets.py +++ b/websockets.py @@ -48,6 +48,7 @@ def application(e, start_response): message = uwsgi.recv_frame(client, '\x00', '\xff') while message: print message + uwsgi.signal(-17) yield '\x00' + message + '\xff' if len(message) == 0: raise StopIteration