From f9a9ef57528c501bb2bb9e959420e99b3909e018 Mon Sep 17 00:00:00 2001 From: "roberto@maverick64" Date: Sat, 12 Feb 2011 10:33:12 +0100 Subject: [PATCH] new cache subsystem --- cache.c | 142 +++++++++++++++++++++----------- master.c | 2 + plugins/fastrouter/fastrouter.c | 2 +- plugins/http/http.c | 2 +- plugins/python/uwsgi_pymodule.c | 30 +++++-- uwsgi.c | 26 +++++- uwsgi.h | 27 +++--- 7 files changed, 163 insertions(+), 68 deletions(-) diff --git a/cache.c b/cache.c index 2272cadd..62096cb3 100644 --- a/cache.c +++ b/cache.c @@ -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; } diff --git a/master.c b/master.c index 0cc1150a..57d169cc 100644 --- a/master.c +++ b/master.c @@ -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); } } diff --git a/plugins/fastrouter/fastrouter.c b/plugins/fastrouter/fastrouter.c index 265eadf4..de65f720 100644 --- a/plugins/fastrouter/fastrouter.c +++ b/plugins/fastrouter/fastrouter.c @@ -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; }; diff --git a/plugins/http/http.c b/plugins/http/http.c index 34147762..cad56704 100644 --- a/plugins/http/http.c +++ b/plugins/http/http.c @@ -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; diff --git a/plugins/python/uwsgi_pymodule.c b/plugins/python/uwsgi_pymodule.c index f022f2e8..b4f99726 100644 --- a/plugins/python/uwsgi_pymodule.c +++ b/plugins/python/uwsgi_pymodule.c @@ -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) { diff --git a/uwsgi.c b/uwsgi.c index 51659a6a..9724f2b8 100644 --- a/uwsgi.c +++ b/uwsgi.c @@ -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; diff --git a/uwsgi.h b/uwsgi.h index 983ca979..7aceaef1 100644 --- a/uwsgi.h +++ b/uwsgi.h @@ -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 *);