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

new cache subsystem

Browse Source
This commit is contained in:
roberto@maverick64
2011-02-12 10:33:12 +01:00
parent 6e5e357a36
commit f9a9ef5752
7 changed files with 163 additions and 68 deletions
Download Patch File Download Diff File Expand all files Collapse all files
cache.c
+93 -49
View File
@@ -14,25 +14,34 @@ uint32_t djb33x_hash(char *key, int keylen) {
return hash;
}
inline uint32_t uwsgi_cache_get_index(char *key, uint16_t keylen) {
inline uint64_t uwsgi_cache_get_index(char *key, uint16_t keylen) {
uint32_t hash = djb33x_hash(key, keylen);
uint32_t i;
int hash_key = hash % 0xffff;
for(i=1;i< uwsgi.cache_max_items;i++) {
// end of table ?
//uwsgi_log("cache item %d -> %d %d %d %.*s %d\n", i, uwsgi.cache_items[i].used, uwsgi.cache_items[i].djbhash, uwsgi.cache_items[i].keysize, uwsgi.cache_items[i].keysize, uwsgi.cache_items[i].key, uwsgi.cache_items[i].valsize);
if (uwsgi.cache_items[i].used == 0) break;
uint64_t slot = uwsgi.cache_hashtable[hash_key];
// hash comparison
if (uwsgi.cache_items[i].djbhash != hash) continue;
struct uwsgi_cache_item *uci;
// keysize comparison
if (uwsgi.cache_items[i].keysize != keylen) continue ;
//uwsgi_log("found slot %d for key %d\n", slot, hash_key);
// key comparison
if (!memcmp(uwsgi.cache_items[i].key, key, keylen)) return i;
uci = &uwsgi.cache_items[slot];
// first round
if (uci->djbhash != hash) goto cycle;
if (uci->keysize != keylen) goto cycle;
if (memcmp(uci->key, key, keylen)) goto cycle;
return slot;
cycle:
while(uci->next) {
slot = uci->next;
uci = &uwsgi.cache_items[slot];
if (uci->djbhash != hash) continue;
if (uci->keysize != keylen) continue;
if (!memcmp(uci->key, key, keylen)) return slot;
}
return 0;
@@ -43,23 +52,14 @@ uint32_t uwsgi_cache_exists(char *key, uint16_t keylen) {
return uwsgi_cache_get_index(key, keylen);
}
char *uwsgi_cache_get(char *key, uint16_t keylen, uint16_t *valsize) {
char *uwsgi_cache_get(char *key, uint16_t keylen, uint64_t *valsize) {
uint32_t index = uwsgi_cache_get_index(key, keylen);
uint64_t index = uwsgi_cache_get_index(key, keylen);
/* is locking needed for reading ?
I do not think so:
case1 => reading during delete -> the worst case is receiving corrupted data for a item that must not exists
case2 => reading during set -> the worst case is receiving corrupted data for an incomplete item
*/
if (index) {
*valsize = uwsgi.cache_items[index].valsize;
// no locking needed, as this is only an hint
uwsgi.cache_items[index].hits++;
return uwsgi.cache+(index*0xffff);
return uwsgi.cache+(index*uwsgi.cache_blocksize);
}
return NULL;
@@ -67,70 +67,114 @@ char *uwsgi_cache_get(char *key, uint16_t keylen, uint16_t *valsize) {
int uwsgi_cache_del(char *key, uint16_t keylen) {
uint32_t index = 0;
uint64_t index = 0;
struct uwsgi_cache_item *uci;
int ret = -1;
uwsgi_lock(uwsgi.cache_lock);
index = uwsgi_cache_get_index(key, keylen);
if (index) {
uci = &uwsgi.cache_items[index] ;
uci->keysize = 0;
uwsgi.shared->cache_first_available_item_tmp = uwsgi.shared->cache_first_available_item;
uwsgi.shared->cache_first_available_item = index;
uci->valsize = 0;
uwsgi.shared->cache_unused_stack_ptr++;
uwsgi.cache_unused_stack[uwsgi.shared->cache_unused_stack_ptr] = index;
// try to return to initial condition...
if (index == uwsgi.shared->cache_first_available_item-1) {
uwsgi.shared->cache_first_available_item--;
//uwsgi_log("FACI: %llu STACK PTR: %llu\n", (unsigned long long) uwsgi.shared->cache_first_available_item, (unsigned long long) uwsgi.shared->cache_unused_stack_ptr);
}
ret = 0;
// relink collisioned entry
if (uci->prev) {
uwsgi.cache_items[uci->prev].next = uci->next;
}
if (uci->next) {
uwsgi.cache_items[uci->next].prev = uci->prev;
}
if (!uci->prev && !uci->next) {
// reset hashtable entry
//uwsgi_log("!!! resetted hashtable entry !!!\n");
uwsgi.cache_hashtable[uci->djbhash % 0xffff] = 0;
}
uci->djbhash = 0;
uci->prev = 0;
uci->next = 0;
}
uwsgi_unlock(uwsgi.cache_lock);
return ret;
}
int uwsgi_cache_set(char *key, uint16_t keylen, char *val, uint16_t vallen, uint64_t expires) {
int uwsgi_cache_set(char *key, uint16_t keylen, char *val, uint64_t vallen, uint64_t expires) {
uint64_t index = 0, last_index = 0 ;
struct uwsgi_cache_item *uci, *ucii;
uint32_t index = 0 ;
struct uwsgi_cache_item *uci;
int ret = -1;
int slot;
if (!keylen || !vallen) return -1;
if (keylen > UWSGI_CACHE_MAX_KEY_SIZE) return -1;
uwsgi_lock(uwsgi.cache_lock);
if (uwsgi.shared->cache_first_available_item >= uwsgi.cache_max_items) goto end;
if (uwsgi.shared->cache_first_available_item >= uwsgi.cache_max_items && !uwsgi.shared->cache_unused_stack_ptr) {
uwsgi_log("*** DANGER cache is FULL !!! ***\n");
goto end;
}
//uwsgi_log("putting cache data in key %.*s %d\n", keylen, key, vallen);
index = uwsgi_cache_get_index(key, keylen);
if (!index) {
index = uwsgi.shared->cache_first_available_item;
if (uwsgi.shared->cache_unused_stack_ptr) {
//uwsgi_log("!!! REUSING CACHE SLOT !!! (faci: %llu)\n", (unsigned long long) uwsgi.shared->cache_first_available_item);
index = uwsgi.cache_unused_stack[uwsgi.shared->cache_unused_stack_ptr];
uwsgi.shared->cache_unused_stack_ptr--;
}
else {
index = uwsgi.shared->cache_first_available_item;
if (uwsgi.shared->cache_first_available_item < uwsgi.cache_max_items) {
uwsgi.shared->cache_first_available_item++;
}
}
uci = &uwsgi.cache_items[index] ;
if (expires) expires += time(NULL);
uci->expires = expires;
uci->djbhash = djb33x_hash(key, keylen);
uci->hits = 0;
uci->used = 1;
memcpy(uci->key, key, keylen);
memcpy(uwsgi.cache+(index*0xffff), val, vallen);
// set this as late as possibile (to minimize locking)
memcpy(uwsgi.cache+(index*uwsgi.cache_blocksize), val, vallen);
// set this as late as possibile (to reduce races risk)
uci->valsize = vallen;
uci->keysize = keylen;
ret = 0;
if (uwsgi.shared->cache_first_available_item_tmp) {
uwsgi.shared->cache_first_available_item = uwsgi.shared->cache_first_available_item_tmp ;
uwsgi.shared->cache_first_available_item_tmp = 0;
// now put the value in the 16bit hashtable
slot = uci->djbhash % 0xffff;
if (uwsgi.cache_hashtable[slot] == 0) {
uwsgi.cache_hashtable[slot] = index;
}
else {
uwsgi.shared->cache_first_available_item++;
//uwsgi_log("HASH COLLISION !!!!\n");
// append to first available next
last_index = uwsgi.cache_hashtable[slot];
ucii = &uwsgi.cache_items[ last_index ];
while(ucii->next) {
last_index = ucii->next;
ucii = &uwsgi.cache_items[ last_index ];
}
ucii->next = index;
uci->prev = last_index;
}
}
end:
uwsgi_unlock(uwsgi.cache_lock);
return ret;
}
void cache_command(char *key, uint16_t keylen, char *val, uint16_t vallen, void *data) {
void cache_command(char *key, uint16_t keylen, char *val, uint64_t vallen, void *data) {
struct wsgi_request *wsgi_req = (struct wsgi_request *) data;
@@ -147,7 +191,7 @@ void cache_command(char *key, uint16_t keylen, char *val, uint16_t vallen, void
int uwsgi_cache_request(struct wsgi_request *wsgi_req) {
uint16_t vallen = 0;
uint64_t vallen = 0;
char *value;
char *argv[3];
uint8_t argc = 0;
@@ -184,7 +228,7 @@ int uwsgi_cache_request(struct wsgi_request *wsgi_req) {
case 4:
// dict
if (wsgi_req->uh.pktsize > 0) {
uwsgi_hooked_parse(wsgi_req->buffer, wsgi_req->uh.pktsize, cache_command, (void *) wsgi_req);
//uwsgi_hooked_parse(wsgi_req->buffer, wsgi_req->uh.pktsize, cache_command, (void *) wsgi_req);
}
break;
}
master.c
+2
View File
@@ -584,11 +584,13 @@ void master_loop(char **argv, char **environ) {
// remove expired cache items
if (uwsgi.cache_max_items > 0) {
for(i=0;i< (int)uwsgi.cache_max_items;i++) {
uwsgi_lock(uwsgi.cache_lock);
if (uwsgi.cache_items[i].expires) {
if (uwsgi.cache_items[i].expires < current_time) {
uwsgi_cache_del(uwsgi.cache_items[i].key, uwsgi.cache_items[i].keysize);
}
}
uwsgi_unlock(uwsgi.cache_lock);
}
}
plugins/fastrouter/fastrouter.c
+1 -1
View File
@@ -60,7 +60,7 @@ struct fastrouter_session {
uint16_t hostname_len;
char *instance_address;
uint16_t instance_address_len;
uint64_t instance_address_len;
int pass_fd;
};
plugins/http/http.c
+1 -1
View File
@@ -77,7 +77,7 @@ struct http_session {
uint16_t hostname_len;
char *instance_address;
uint16_t instance_address_len;
uint64_t instance_address_len;
int pass_fd;
plugins/python/uwsgi_pymodule.c
+24 -6
View File
@@ -2368,10 +2368,13 @@ PyObject *py_uwsgi_cache_del(PyObject * self, PyObject * args) {
uwsgi_simple_send_string(remote, 111, 2, key, keylen, uwsgi.shared->options[UWSGI_OPTION_SOCKET_TIMEOUT]);
}
else if (uwsgi.cache_max_items) {
uwsgi_lock(uwsgi.cache_lock);
if (uwsgi_cache_del(key, strlen(key))) {
uwsgi_unlock(uwsgi.cache_lock);
Py_INCREF(Py_None);
return Py_None;
}
uwsgi_unlock(uwsgi.cache_lock);
}
Py_INCREF(Py_True);
@@ -2394,18 +2397,21 @@ PyObject *py_uwsgi_cache_set(PyObject * self, PyObject * args) {
return NULL;
}
if (vallen > 0xffff) {
return PyErr_Format(PyExc_ValueError, "uWSGI cache items size must be < 64K, requested %d bytes", (int) vallen);
if ((uint64_t)vallen > uwsgi.cache_blocksize) {
return PyErr_Format(PyExc_ValueError, "uWSGI cache items size must be < %llu, requested %d bytes", (unsigned long long)uwsgi.cache_blocksize, (int) vallen);
}
if (remote && strlen(remote) > 0) {
uwsgi_simple_send_string2(remote, 111, 1, key, keylen, value, vallen, uwsgi.shared->options[UWSGI_OPTION_SOCKET_TIMEOUT]);
}
else if (uwsgi.cache_max_items) {
if (uwsgi_cache_set(key, strlen(key), value, vallen, expires)) {
uwsgi_lock(uwsgi.cache_lock);
if (uwsgi_cache_set(key, keylen, value, vallen, expires)) {
uwsgi_unlock(uwsgi.cache_lock);
Py_INCREF(Py_None);
return Py_None;
}
uwsgi_unlock(uwsgi.cache_lock);
}
Py_INCREF(Py_True);
@@ -2447,24 +2453,36 @@ PyObject *py_uwsgi_cache_exists(PyObject * self, PyObject * args) {
PyObject *py_uwsgi_cache_get(PyObject * self, PyObject * args) {
char *key;
uint16_t valsize;
uint64_t valsize;
Py_ssize_t keylen = 0;
char *value = NULL;
char *remote = NULL;
char buffer[0xffff];
PyObject *res;
struct timeval tv, tv2;
if (!PyArg_ParseTuple(args, "s#|s:cache_get", &key, &keylen, &remote)) {
return NULL;
}
if (remote && strlen(remote) > 0) {
uwsgi_simple_message_string(remote, 111, 0, key, keylen, buffer, &valsize, uwsgi.shared->options[UWSGI_OPTION_SOCKET_TIMEOUT]);
//uwsgi_simple_message_string(remote, 111, 0, key, keylen, buffer, &valsize, uwsgi.shared->options[UWSGI_OPTION_SOCKET_TIMEOUT]);
if (valsize > 0) {
value = buffer;
}
}
else if (uwsgi.cache_max_items) {
value = uwsgi_cache_get(key, strlen(key), &valsize);
gettimeofday(&tv, NULL);
uwsgi_lock(uwsgi.cache_lock);
value = uwsgi_cache_get(key, keylen, &valsize);
res = PyString_FromStringAndSize(value, valsize);
gettimeofday(&tv2, NULL);
if ((tv2.tv_sec* (1000*1000) + tv2.tv_usec) - (tv.tv_sec* (1000*1000) + tv.tv_usec) > 30000) {
uwsgi_log("[slow] cache get done in %d microseconds (%llu bytes value)\n", (tv2.tv_sec* (1000*1000) + tv2.tv_usec) - (tv.tv_sec* (1000*1000) + tv.tv_usec), (unsigned long long) valsize);
}
uwsgi_unlock(uwsgi.cache_lock);
return res;
}
if (value) {
uwsgi.c
+25 -1
View File
@@ -63,6 +63,7 @@ static struct option long_base_options[] = {
{"socket-timeout", required_argument, 0, 'z'},
{"sharedarea", required_argument, 0, 'A'},
{"cache", required_argument, 0, LONG_ARGS_CACHE},
{"cache-blocksize", required_argument, 0, LONG_ARGS_CACHE_BLOCKSIZE},
#ifdef UWSGI_SPOOLER
{"spooler", required_argument, 0, 'Q'},
#endif
@@ -1061,13 +1062,32 @@ int uwsgi_start(void *v_argv) {
}
if (uwsgi.cache_max_items > 0) {
if (!uwsgi.cache_blocksize) uwsgi.cache_blocksize = 0xffff;
uwsgi.cache_hashtable = (uint64_t *) mmap(NULL, sizeof(uint64_t) * 0xffff, PROT_READ | PROT_WRITE, MAP_SHARED | MAP_ANON, -1, 0);
if (!uwsgi.cache_hashtable) {
uwsgi_error("mmap()");
exit(1);
}
memset(uwsgi.cache_hashtable, 0, sizeof(uint64_t) * 0xffff);
uwsgi.cache_unused_stack = (uint64_t *) mmap(NULL, sizeof(uint64_t) * uwsgi.cache_max_items, PROT_READ | PROT_WRITE, MAP_SHARED | MAP_ANON, -1, 0);
if (!uwsgi.cache_unused_stack) {
uwsgi_error("mmap()");
exit(1);
}
memset(uwsgi.cache_unused_stack, 0, sizeof(uint64_t) * uwsgi.cache_max_items);
uwsgi.cache_items = (struct uwsgi_cache_item *) 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, 0xffff * uwsgi.cache_max_items, PROT_READ | PROT_WRITE, MAP_SHARED | MAP_ANON, -1, 0);
uwsgi.cache = mmap(NULL, uwsgi.cache_blocksize * uwsgi.cache_max_items, PROT_READ | PROT_WRITE, MAP_SHARED | MAP_ANON, -1, 0);
if (!uwsgi.cache) {
uwsgi_error("mmap()");
exit(1);
@@ -1079,6 +1099,7 @@ int uwsgi_start(void *v_argv) {
// the first cache item is always zero
uwsgi.shared->cache_first_available_item = 1;
uwsgi.shared->cache_unused_stack_ptr = 0;
uwsgi.cache_lock = uwsgi_mmap_shared_lock();
uwsgi_lock_init(uwsgi.cache_lock);
@@ -1929,6 +1950,9 @@ end:
case LONG_ARGS_CACHE:
uwsgi.cache_max_items = atoi(optarg);
return 1;
case LONG_ARGS_CACHE_BLOCKSIZE:
uwsgi.cache_blocksize = atoi(optarg);
return 1;
case 'A':
uwsgi.sharedareasize = atoi(optarg);
return 1;
uwsgi.h
+17 -10
View File
@@ -190,7 +190,7 @@ extern int pivot_root(const char * new_root, const char * put_old);
#define UWSGI_LISTEN_QUEUE 511
#endif
#define UWSGI_CACHE_MAX_KEY_SIZE 4071
#define UWSGI_CACHE_MAX_KEY_SIZE 2048
union uwsgi_sockaddr {
struct sockaddr sa;
@@ -232,19 +232,22 @@ struct uwsgi_daemon {
// maintain alignment here !!!
struct uwsgi_cache_item {
// unused
uint16_t unused;
// size of the key
uint16_t keysize;
// djb hash of the key
uint32_t djbhash;
// size of the value (max 64KB)
uint16_t valsize;
// size of the value (64bit)
uint64_t valsize;
// 64bit expiration (0 for immortal)
uint64_t expires;
// 64bit hits
uint64_t hits;
// mark the end of the table
char used;
// previous same-hash item
uint64_t prev;
// next same-hash item
uint64_t next;
// key chracters follows...
char key[UWSGI_CACHE_MAX_KEY_SIZE];
} __attribute__((__packed__));
@@ -327,6 +330,7 @@ struct uwsgi_opt {
#define LONG_ARGS_WORKER_EXEC 17071
#define LONG_ARGS_EMPEROR 17072
#define LONG_ARGS_PRINT 17073
#define LONG_ARGS_CACHE_BLOCKSIZE 17074
@@ -913,6 +917,9 @@ struct uwsgi_server {
struct sockaddr_in mc_cluster_addr;
uint32_t cache_max_items;
uint64_t *cache_hashtable;
uint64_t *cache_unused_stack;
uint64_t cache_blocksize;
struct uwsgi_cache_item *cache_items;
void *cache;
@@ -1030,8 +1037,8 @@ struct uwsgi_shared {
#endif
uint16_t cache_first_available_item;
uint16_t cache_first_available_item_tmp;
uint64_t cache_first_available_item;
uint64_t cache_unused_stack_ptr;
int worker_signal_pipe[2];
@@ -1328,9 +1335,9 @@ ssize_t uwsgi_send_message(int, uint8_t, uint8_t, char *, uint16_t, int, ssize_t
char *uwsgi_cluster_best_node(void);
int uwsgi_cache_set(char *, uint16_t, char *, uint16_t, uint64_t);
int uwsgi_cache_set(char *, uint16_t, char *, uint64_t, uint64_t);
int uwsgi_cache_del(char *, uint16_t);
char *uwsgi_cache_get(char *, uint16_t, uint16_t *);
char *uwsgi_cache_get(char *, uint16_t, uint64_t *);
uint32_t uwsgi_cache_exists(char *, uint16_t);
void uwsgi_lock_init(void *);