clearlinux/uwsgi
2
0
Fork 0
mirror of https://github.com/clearlinux/uwsgi.git synced 2026-10-17 01:50:53 +00:00
Code Issues Packages Projects Releases Wiki Activity
Files
b90779aecfb91cb08bb6a175fdcbc49345f02113
uwsgi/core/cache.c
T

862 lines
24 KiB
C
Raw Blame History

#include <uwsgi.h>
extern struct uwsgi_server uwsgi;
#define cache_item(x) (struct uwsgi_cache_item *) (((char *)uc->items) + ((sizeof(struct uwsgi_cache_item)+uc->keysize) * x))
// block bitmap manager
/*
static uint64_t cache_get_block_by_bitmap(struct uwsgi_cache *uc, uint64_t len) {
// first of all, how many blocks i need ?
uint64_t blocks = len/uc->blocksize;
if (len%uc->blocksize > 0) blocks++;
uwsgi_log("searching for %d free blocks\n");
return 0;
}
*/
static void cache_unmark_blocks(struct uwsgi_cache *uc, uint64_t index, uint64_t len) {
/*
uint64_t base = index/8;
uint64_t bit = index%8;
uint64_t blocks = len/uc->blocksize;
if (len%uc->blocksize > 0) blocks++;
*/
}
static void cache_send_udp_command(char *, uint16_t, char *, uint16_t, uint64_t, uint8_t);
static void cache_sync_hook(char *k, uint16_t kl, char *v, uint16_t vl, void *data) {
struct uwsgi_cache *uc = (struct uwsgi_cache *) data;
if (!uwsgi_strncmp(k, kl, "items", 5)) {
size_t num = uwsgi_str_num(v, vl);
if (num != uc->max_items) {
uwsgi_log("[cache-sync] invalid cache size, expected %llu received %llu\n", (unsigned long long) uc->max_items, (unsigned long long) num);
exit(1);
}
}
if (!uwsgi_strncmp(k, kl, "blocksize", 9)) {
size_t num = uwsgi_str_num(v, vl);
if (num != uc->blocksize) {
uwsgi_log("[cache-sync] invalid cache block size, expected %llu received %llu\n", (unsigned long long) uc->blocksize, (unsigned long long) num);
exit(1);
}
}
}
static void uwsgi_cache_load_files(struct uwsgi_cache *uc) {
struct uwsgi_string_list *usl = uwsgi.load_file_in_cache;
while(usl) {
size_t len = 0;
char *value = uwsgi_open_and_read(usl->value, &len, 0, NULL);
if (value) {
if (!uwsgi_cache_set2(uc,usl->value, usl->len, value, len, 0, 0)) {
uwsgi_log("[cache] stored %s\n", usl->value);
}
}
usl = usl->next;
}
}
void uwsgi_cache_init(struct uwsgi_cache *uc) {
uc->hashtable = uwsgi_calloc_shared(sizeof(uint64_t) * uc->hashsize);
uc->unused_blocks_stack = uwsgi_calloc_shared(sizeof(uint64_t) * uc->blocks);
// the first cache item is always zero
uc->first_available_block = 1;
uc->unused_blocks_stack_ptr = 0;
uc->filesize = ( (sizeof(struct uwsgi_cache_item)+uc->keysize) * uc->max_items) + (uc->blocksize * uc->blocks);
if (uc->use_blocks_bitmap) {
uc->blocks_bitmap = uwsgi_calloc_shared(uc->blocks/8);
}
//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 (uc->store) {
int cache_fd;
struct stat cst;
if (stat(uc->store, &cst)) {
uwsgi_log("creating a new cache store file: %s\n", uc->store);
cache_fd = open(uc->store, O_CREAT | O_RDWR, S_IRUSR | S_IWUSR);
if (cache_fd >= 0) {
// fill the caching store
if (ftruncate(cache_fd, uc->filesize)) {
uwsgi_log("ftruncate()");
exit(1);
}
}
}
else {
if ((size_t) cst.st_size != uc->filesize || !S_ISREG(cst.st_mode)) {
uwsgi_log("invalid cache store file. Please remove it or fix cache blocksize/items to match its size\n");
exit(1);
}
cache_fd = open(uc->store, O_CREAT | O_RDWR, S_IRUSR | S_IWUSR);
uwsgi_log("recovered cache from backing store file: %s\n", uc->store);
}
if (cache_fd < 0) {
uwsgi_error_open(uc->store);
exit(1);
}
uc->items = (struct uwsgi_cache_item *) mmap(NULL, uc->filesize, PROT_READ | PROT_WRITE, MAP_SHARED, cache_fd, 0);
if (!uc->items) {
uwsgi_error("uwsgi_cache_init()/mmap() [with store]");
exit(1);
}
uwsgi_cache_fix(uc);
}
else {
uc->items = (struct uwsgi_cache_item *) mmap(NULL, uc->filesize, PROT_READ | PROT_WRITE, MAP_SHARED | MAP_ANON, -1, 0);
if (!uc->items) {
uwsgi_error("uwsgi_cache_init()/mmap()");
exit(1);
}
uint64_t i;
for (i = 0; i < uc->max_items; i++) {
// here we only need to clear the item header
memset(cache_item(i), 0, sizeof(struct uwsgi_cache_item));
}
}
uc->data = ((char *)uc->items) + ((sizeof(struct uwsgi_cache_item)+uc->keysize) * uc->max_items);
uc->lock = uwsgi_rwlock_init("cache");
uwsgi_log("*** Cache subsystem initialized: %lluMB (key: %llu bytes, keys: %llu bytes, data: %llu bytes) preallocated ***\n", (unsigned long long) uc->filesize / (1024 * 1024),
(unsigned long long) sizeof(struct uwsgi_cache_item)+uc->keysize,
(unsigned long long) ((sizeof(struct uwsgi_cache_item)+uc->keysize) * uc->max_items), (unsigned long long) (uc->blocksize * uc->max_items));
uwsgi_cache_load_files(uc);
struct uwsgi_string_list *usl = uwsgi.cache_udp_node;
while(usl) {
char *port = strchr(usl->value, ':');
if (!port) {
uwsgi_log("[cache-udp-node] invalid udp address: %s\n", usl->value);
exit(1);
}
// no need to zero the memory, socket_to_in_addr will do that
struct sockaddr_in *sin = uwsgi_malloc(sizeof(struct sockaddr_in));
usl->custom = socket_to_in_addr(usl->value, port, 0, sin);
usl->custom_ptr = sin;
uwsgi_log("added cache udp node %s\n", usl->value);
usl = usl->next;
}
uwsgi.cache_udp_node_socket = socket(AF_INET, SOCK_DGRAM, 0);
if (uwsgi.cache_udp_node_socket < 0) {
uwsgi_error("[cache-udp-node] socket()");
exit(1);
}
uwsgi_socket_nb(uwsgi.cache_udp_node_socket);
if (uwsgi.cache_sync) {
uwsgi_log("[cache-sync] getting cache dump from %s ...\n", uwsgi.cache_sync);
int fd = uwsgi_connect(uwsgi.cache_sync, 0, 0);
if (fd < 0) {
uwsgi_log("[cache-sync] unable to connect to the cache server\n");
exit(1);
}
struct uwsgi_header cuh;
cuh.modifier1 = 111;
cuh.modifier2 = 6;
cuh.pktsize = 0;
if (write(fd, &cuh, 4) != 4) {
uwsgi_log("[cache-sync] unable to write to the cache server\n");
exit(1);
}
int ret = uwsgi_read_uh(fd, &cuh, uwsgi.shared->options[UWSGI_OPTION_SOCKET_TIMEOUT]);
if (ret) {
uwsgi_log("[cache-sync] unable to read from the cache server\n");
exit(1);
}
if (cuh.modifier1 != 111 || cuh.modifier2 != 7) {
uwsgi_log("[cache-sync] invalid uwsgi packet received from the cache server\n");
exit(1);
}
char *dump_buf = uwsgi_malloc(cuh.pktsize);
ret = uwsgi_read_nb(fd, dump_buf, cuh.pktsize, uwsgi.shared->options[UWSGI_OPTION_SOCKET_TIMEOUT]);
if (ret) {
uwsgi_log("[cache-sync] unable to read from the cache server\n");
exit(1);
}
uwsgi_hooked_parse(dump_buf, cuh.pktsize, cache_sync_hook, NULL);
ret = uwsgi_read_nb(fd, (char *) uc->items, uc->filesize, uwsgi.shared->options[UWSGI_OPTION_SOCKET_TIMEOUT]);
if (ret) {
uwsgi_log("[cache-sync] unable to read from the cache server\n");
exit(1);
}
// reset the hashtable
memset(uc->hashtable, 0, sizeof(uint64_t) * UMAX16);
// re-fill the hashtable
uwsgi_cache_fix(uc);
}
}
static inline uint64_t uwsgi_cache_get_index(struct uwsgi_cache *uc, char *key, uint16_t keylen) {
uint32_t hash = uc->hash->func(key, keylen);
uint32_t hash_key = hash % uc->hashsize;
uint64_t slot = uc->hashtable[hash_key];
struct uwsgi_cache_item *uci = cache_item(slot);
uint64_t rounds = 0;
// first round
if (uci->hash % uc->hashsize != hash_key)
return 0;
if (uci->hash != 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 = cache_item(slot);
rounds++;
if (rounds > uc->max_items) {
uwsgi_log("ALARM !!! cache-loop (and potential deadlock) detected slot = %lu prev = %lu next = %lu\n", slot, uci->prev, uci->next);
// terrible case: the whole uWSGI stack can deadlock, leaving only the master alive
// if the master is avalable, trigger a brutal reload
if (uwsgi.master_process) {
kill(uwsgi.workers[0].pid, SIGTERM);
}
// otherwise kill the current worker (could be pretty useless...)
else {
exit(1);
}
}
if (uci->hash != hash)
continue;
if (uci->keysize != keylen)
continue;
if (!memcmp(uci->key, key, keylen))
return slot;
}
return 0;
}
uint32_t uwsgi_cache_exists2(struct uwsgi_cache *uc, char *key, uint16_t keylen) {
return uwsgi_cache_get_index(uc, key, keylen);
}
char *uwsgi_cache_get2(struct uwsgi_cache *uc, char *key, uint16_t keylen, uint64_t * valsize) {
uint64_t index = uwsgi_cache_get_index(uc, key, keylen);
if (index) {
struct uwsgi_cache_item *uci = cache_item(index);
if (uci->flags & UWSGI_CACHE_FLAG_UNGETTABLE)
return NULL;
*valsize = uci->valsize;
uci->hits++;
uc->hits++;
return uc->data + (index * uc->blocksize);
}
uc->miss++;
return NULL;
}
int uwsgi_cache_del2(struct uwsgi_cache *uc, char *key, uint16_t keylen, uint64_t index, uint16_t flags) {
struct uwsgi_cache_item *uci;
int ret = -1;
if (!index)
index = uwsgi_cache_get_index(uc, key, keylen);
if (index) {
uci = cache_item(index);
uci->keysize = 0;
cache_unmark_blocks(uc, index, uci->valsize);
uci->valsize = 0;
// try to return to initial condition...
if (index == uc->first_available_block - 1) {
uc->first_available_block--;
//uwsgi_log("FACI: %llu STACK PTR: %llu\n", (unsigned long long) uwsgi.shared->cache_first_available_block, (unsigned long long) uwsgi.shared->cache_unused_blocks_stack_ptr);
}
ret = 0;
// relink collisioned entry
if (uci->prev) {
struct uwsgi_cache_item *ucii = cache_item(uci->prev);
ucii->next = uci->next;
}
else {
// set next as the new entry point (could be 0)
uc->hashtable[uci->hash % uc->hashsize] = uci->next;
}
if (uci->next) {
struct uwsgi_cache_item *ucii = cache_item(uci->next);
ucii->prev = uci->prev;
}
if (!uci->prev && !uci->next) {
// reset hashtable entry
//uwsgi_log("!!! resetted hashtable entry !!!\n");
uc->hashtable[uci->hash % uc->hashsize] = 0;
}
uci->hash = 0;
uci->prev = 0;
uci->next = 0;
uci->expires = 0;
uc->n_items--;
}
if (uwsgi.cache_udp_node && ret == 0 && !(flags & UWSGI_CACHE_FLAG_LOCAL)) {
cache_send_udp_command(key, keylen, NULL, 0, 0, 11);
}
return ret;
}
void uwsgi_cache_fix(struct uwsgi_cache *uc) {
uint64_t i;
unsigned long long restored = 0;
for (i = 0; i < uc->max_items; i++) {
// valid record ?
struct uwsgi_cache_item *uci = cache_item(i);
if (uci->keysize) {
if (!uci->prev) {
// put value in hash_table
uc->hashtable[uci->hash % uc->hashsize] = i;
restored++;
}
}
else {
// put this record in unused stack
uc->first_available_block = i;
uc->unused_blocks_stack_ptr++;
uc->unused_blocks_stack[uc->unused_blocks_stack_ptr] = i;
}
}
uwsgi_log("[uwsgi-cache] restored %llu items\n", restored);
}
int uwsgi_cache_set2(struct uwsgi_cache *uc, char *key, uint16_t keylen, char *val, uint64_t vallen, uint64_t expires, uint64_t flags) {
uint64_t index = 0, last_index = 0;
struct uwsgi_cache_item *uci, *ucii;
int ret = -1;
if (!keylen || !vallen)
return -1;
if (keylen > uc->keysize)
return -1;
if (uc->first_available_block >= uc->max_items && !uc->unused_blocks_stack_ptr) {
uwsgi_log("*** DANGER cache is FULL !!! ***\n");
uc->full++;
goto end;
}
//uwsgi_log("putting cache data in key %.*s %d\n", keylen, key, vallen);
index = uwsgi_cache_get_index(uc, key, keylen);
if (!index) {
if (uc->unused_blocks_stack_ptr) {
//uwsgi_log("!!! REUSING CACHE SLOT !!! (faci: %llu)\n", (unsigned long long) uwsgi.shared->cache_first_available_block);
index = uc->unused_blocks_stack[uc->unused_blocks_stack_ptr];
uc->unused_blocks_stack_ptr--;
}
else {
index = uc->first_available_block;
if (uc->first_available_block < uc->max_items) {
uc->first_available_block++;
}
}
uci = cache_item(index);
if (expires && !(flags & UWSGI_CACHE_FLAG_ABSEXPIRE))
expires += uwsgi_now();
uci->expires = expires;
uci->hash = uc->hash->func(key, keylen);
uci->hits = 0;
uci->flags = flags;
memcpy(uci->key, key, keylen);
memcpy(((char *) uc->data) + (index * uc->blocksize), val, vallen);
// set this as late as possibile (to reduce races risk)
uci->valsize = vallen;
uci->keysize = keylen;
ret = 0;
// now put the value in the hashtable
uint32_t slot = uci->hash % uc->hashsize;
// reset values
uci->prev = 0;
uci->next = 0;
last_index = uc->hashtable[slot];
if (last_index == 0) {
uc->hashtable[slot] = index;
}
else {
// append to first available next
ucii = cache_item(last_index);
while (ucii->next) {
last_index = ucii->next;
ucii = cache_item(last_index);
}
ucii->next = index;
uci->prev = last_index;
}
uc->n_items++ ;
}
else if (flags & UWSGI_CACHE_FLAG_UPDATE) {
uci = cache_item(index);
if (expires && !(flags & UWSGI_CACHE_FLAG_ABSEXPIRE)) {
expires += uwsgi_now();
uci->expires = expires;
}
memcpy(uc->data + (index * uc->blocksize), val, vallen);
uci->valsize = vallen;
ret = 0;
}
if (uwsgi.cache_udp_node && ret == 0 && !(flags & UWSGI_CACHE_FLAG_LOCAL)) {
cache_send_udp_command(key, keylen, val, vallen, expires, 10);
}
end:
return ret;
}
static void cache_send_udp_command(char *key, uint16_t keylen, char *val, uint16_t vallen, uint64_t expires, uint8_t cmd) {
struct uwsgi_header uh;
uint8_t u_k[2];
uint8_t u_v[2];
uint8_t u_e[2];
uint16_t vallen16 = vallen;
struct iovec iov[7];
struct msghdr mh;
memset(&mh, 0, sizeof(struct msghdr));
mh.msg_iov = iov;
mh.msg_iovlen = 3;
if (cmd == 10) {
mh.msg_iovlen = 7;
}
iov[0].iov_base = &uh;
iov[0].iov_len = 4;
u_k[0] = (uint8_t) (keylen & 0xff);
u_k[1] = (uint8_t) ((keylen >> 8) & 0xff);
iov[1].iov_base = u_k;
iov[1].iov_len = 2;
iov[2].iov_base = key;
iov[2].iov_len = keylen;
uh.pktsize = 2 + keylen;
if (cmd == 10) {
u_v[0] = (uint8_t) (vallen16 & 0xff);
u_v[1] = (uint8_t) ((vallen16 >> 8) & 0xff);
iov[3].iov_base = u_v;
iov[3].iov_len = 2;
iov[4].iov_base = val;
iov[4].iov_len = vallen16;
char es[sizeof(UMAX64_STR) + 1];
uint16_t es_size = uwsgi_long2str2n(expires, es, sizeof(UMAX64_STR));
u_e[0] = (uint8_t) (es_size & 0xff);
u_e[1] = (uint8_t) ((es_size >> 8) & 0xff);
iov[5].iov_base = u_e;
iov[5].iov_len = 2;
iov[6].iov_base = es;
iov[6].iov_len = es_size;
uh.pktsize += 2 + vallen16 + 2 + es_size;
}
uh.modifier1 = 111;
uh.modifier2 = cmd;
struct uwsgi_string_list *usl = uwsgi.cache_udp_node;
while(usl) {
mh.msg_name = usl->custom_ptr;
mh.msg_namelen = usl->custom;
if (sendmsg(uwsgi.cache_udp_node_socket, &mh, 0) <= 0) {
uwsgi_error("[cache-udp-node] sendmsg()");
}
usl = usl->next;
}
}
/* THIS PART IS HEAVILY OPTIMIZED: PERFORMANCE NOT ELEGANCE !!! */
void *cache_thread_loop(void *ucache) {
struct uwsgi_cache *uc = (struct uwsgi_cache *) ucache;
int fd = uc->thread_server_fd;
int i;
ssize_t len;
char uwsgi_packet[UMAX16 + 4];
struct uwsgi_header *uh = (struct uwsgi_header *) uwsgi_packet;
char *val;
uint64_t vallen;
char *key;
uint16_t keylen;
struct pollfd ctl_poll;
char *watermark;
struct sockaddr_un ctl_sun;
socklen_t ctl_sun_len;
ctl_poll.events = POLLIN;
for (;;) {
ctl_sun_len = sizeof(struct sockaddr_un);
pthread_mutex_lock(&uwsgi.cache_server_lock);
ctl_poll.fd = accept(fd, (struct sockaddr *) &ctl_sun, &ctl_sun_len);
pthread_mutex_unlock(&uwsgi.cache_server_lock);
if (ctl_poll.fd < 0) {
uwsgi_error("cache accept()");
continue;
}
i = 0;
while (i < 4) {
len = poll(&ctl_poll, 1, uwsgi.shared->options[UWSGI_OPTION_SOCKET_TIMEOUT]);
if (len <= 0) {
uwsgi_error("cache poll()");
goto clear;
}
len = read(ctl_poll.fd, uwsgi_packet + i, 4 - i);
if (len < 0) {
uwsgi_error("cache read()");
goto clear;
}
i += len;
}
if (uh->pktsize == 0)
goto clear;
while (i < 4 + uh->pktsize) {
len = poll(&ctl_poll, 1, uwsgi.shared->options[UWSGI_OPTION_SOCKET_TIMEOUT]);
if (len <= 0) {
uwsgi_error("cache poll()");
goto clear;
}
len = read(ctl_poll.fd, uwsgi_packet + i, (4 + uh->pktsize) - i);
if (len < 0) {
uwsgi_error("cache read()");
goto clear;
}
i += len;
}
watermark = uwsgi_packet + 4 + uh->pktsize;
// get first parameter
memcpy(&keylen, uwsgi_packet + 4, 2);
#ifdef __BIG_ENDIAN__
keylen = uwsgi_swap16(keylen);
#endif
if (uwsgi_packet + 6 + keylen > watermark)
goto clear;
key = uwsgi_packet + 6 + keylen + 2;
memcpy(&keylen, key - 2, 2);
#ifdef __BIG_ENDIAN__
keylen = uwsgi_swap16(keylen);
#endif
if (key + keylen > watermark)
goto clear;
uwsgi_rlock(uc->lock);
val = uwsgi_cache_get2(uc, key, keylen, &vallen);
if (val && vallen > 0) {
if (write(ctl_poll.fd, val, vallen) != (int64_t) vallen) {
uwsgi_error("cache write()");
}
}
uwsgi_rwunlock(uc->lock);
clear:
close(ctl_poll.fd);
}
return NULL;
}
int uwsgi_cache_server(char *socket, int threads) {
int *fd = uwsgi_malloc(sizeof(int));
int i;
pthread_t thread_id;
char *tcp_port = strchr(socket, ':');
if (tcp_port) {
*fd = bind_to_tcp(socket, uwsgi.listen_queue, tcp_port);
}
else {
*fd = bind_to_unix(socket, uwsgi.listen_queue, uwsgi.chmod_socket, uwsgi.abstract_socket);
}
pthread_mutex_init(&uwsgi.cache_server_lock, NULL);
if (threads < 1)
threads = 1;
uwsgi_log("*** cache-optimized server enabled on fd %d (%d threads) ***\n", *fd, threads);
for (i = 0; i < threads; i++) {
pthread_create(&thread_id, NULL, cache_thread_loop, (void *) fd);
}
return *fd;
}
void uwsgi_cache_wlock(struct uwsgi_cache *uc) {
uwsgi.lock_ops.wlock(uc->lock);
}
void uwsgi_cache_rlock(struct uwsgi_cache *uc) {
uwsgi.lock_ops.rlock(uc->lock);
}
void uwsgi_cache_rwunlock(struct uwsgi_cache *uc) {
uwsgi.lock_ops.rwunlock(uc->lock);
}
void *cache_udp_server_loop(void *ucache) {
// block all signals
sigset_t smask;
sigfillset(&smask);
pthread_sigmask(SIG_BLOCK, &smask, NULL);
struct uwsgi_cache *uc = (struct uwsgi_cache *) ucache;
int queue = event_queue_init();
struct uwsgi_string_list *usl = uwsgi.cache_udp_server;
while(usl) {
if (strchr(usl->value, ':')) {
int fd = bind_to_udp(usl->value, 0, 0);
if (fd < 0) {
uwsgi_log("[cache-udp-server] cannot bind to %s\n", usl->value);
exit(1);
}
uwsgi_socket_nb(fd);
event_queue_add_fd_read(queue, fd);
uwsgi_log("*** cache udp server running on %s ***\n", usl->value);
}
usl = usl->next;
}
// allocate 64k chunk to receive messages
char *buf = uwsgi_malloc(UMAX16);
for(;;) {
uint16_t pktsize = 0, ss = 0;
int interesting_fd = -1;
int rlen = event_queue_wait(queue, -1, &interesting_fd);
if (rlen <= 0) continue;
if (interesting_fd < 0) continue;
ssize_t len = read(interesting_fd, buf, UMAX16);
if (len <= 7) {
uwsgi_error("[cache-udp-server] read()");
}
if (buf[0] != 111) continue;
memcpy(&pktsize, buf+1, 2);
if (pktsize != len-4) continue;
memcpy(&ss, buf + 4, 2);
if (4+ss > pktsize) continue;
uint16_t keylen = ss;
char *key = buf + 6;
// cache set/update
if (buf[3] == 10) {
if (keylen + 2 + 2 > pktsize) continue;
memcpy(&ss, buf + 6 + keylen, 2);
if (4+keylen+ss > pktsize) continue;
uint16_t vallen = ss;
char *val = buf + 8 + keylen;
uint64_t expires = 0;
if (2 + keylen + 2 + vallen + 2 < pktsize) {
memcpy(&ss, buf + 8 + keylen + vallen , 2);
if (6+keylen+vallen+ss > pktsize) continue;
expires = uwsgi_str_num(buf + 10 + keylen+vallen, ss);
}
uwsgi_wlock(uc->lock);
if (uwsgi_cache_set(key, keylen, val, vallen, expires, UWSGI_CACHE_FLAG_UPDATE|UWSGI_CACHE_FLAG_LOCAL|UWSGI_CACHE_FLAG_ABSEXPIRE)) {
uwsgi_log("[cache-udp-server] unable to update cache\n");
}
uwsgi_rwunlock(uc->lock);
}
// cache del
else if (buf[3] == 11) {
uwsgi_wlock(uc->lock);
if (uwsgi_cache_del(key, keylen, 0, UWSGI_CACHE_FLAG_LOCAL)) {
uwsgi_log("[cache-udp-server] unable to update cache\n");
}
uwsgi_rwunlock(uc->lock);
}
}
return NULL;
}
void *cache_sweeper_loop(void *ucache) {
uint64_t i;
// block all signals
sigset_t smask;
sigfillset(&smask);
pthread_sigmask(SIG_BLOCK, &smask, NULL);
struct uwsgi_cache *uc = (struct uwsgi_cache *) ucache;
if (!uwsgi.cache_expire_freq)
uwsgi.cache_expire_freq = 3;
// remove expired cache items TODO use rb_tree timeouts
for (;;) {
sleep(uwsgi.cache_expire_freq);
uint64_t freed_items = 0;
// skip the first slot
for (i = 1; i < uc->max_items; i++) {
uwsgi_wlock(uc->lock);
struct uwsgi_cache_item *uci = cache_item(i);
if (uci->expires) {
if (uci->expires < (uint64_t) uwsgi.current_time) {
uwsgi_cache_del2(uc, NULL, 0, i, UWSGI_CACHE_FLAG_LOCAL);
freed_items++;
}
}
uwsgi_rwunlock(uc->lock);
}
if (uwsgi.cache_report_freed_items && freed_items > 0) {
uwsgi_log("freed %llu cache items\n", (unsigned long long) freed_items);
}
};
return NULL;
}
void uwsgi_cache_sync_all() {
/*
if (uwsgi.cache_store && uwsgi.cache_filesize) {
if (msync(uwsgi.cache_items, uwsgi.cache_filesize, MS_ASYNC)) {
uwsgi_error("msync()");
}
}
if (uwsgi.cache_store && uwsgi.cache_filesize && uwsgi.cache_store_sync && ((uwsgi.master_cycles % uwsgi.cache_store_sync) == 0)) {
if (msync(uwsgi.cache_items, uwsgi.cache_filesize, MS_ASYNC)) {
uwsgi_error("msync()");
}
}
*/
}
void uwsgi_cache_start_sweepers() {
/*
if (uwsgi.caches > 0 && !uwsgi.cache_no_expire) {
if (pthread_create(&cache_sweeper, NULL, cache_sweeper_loop, NULL)) {
uwsgi_error("pthread_create()");
uwsgi_log("unable to run the cache sweeper !!!\n");
}
else {
uwsgi_log("cache sweeper thread enabled\n");
}
}
*/
}
void uwsgi_cache_start_sync_servers() {
/*
if (uwsgi.cache_max_items > 0 && uwsgi.cache_udp_server) {
if (pthread_create(&cache_udp_server, NULL, cache_udp_server_loop, NULL)) {
uwsgi_error("pthread_create()");
uwsgi_log("unable to run the cache udp server !!!\n");
}
else {
uwsgi_log("cache udp server thread enabled\n");
}
}
*/
}
void uwsgi_cache_create(char *arg) {
struct uwsgi_cache *old_uc = NULL, *uc = uwsgi.caches;
while(uc) {
old_uc = uc;
uc = uc->next;
}
uc = uwsgi_calloc_shared(sizeof(struct uwsgi_cache));
if (old_uc) {
old_uc->next = uc;
}
else {
uwsgi.caches = uc;
}
// default (old-stye) cache ?
if (!arg) {
uc->blocksize = uwsgi.cache_blocksize;
if (!uc->blocksize) uc->blocksize = UMAX16;
uc->max_items = uwsgi.cache_max_items;
uc->blocks = uwsgi.cache_max_items;
uc->keysize = 2048;
uc->hashsize = UMAX16;
uc->hash = uwsgi_hash_algo_get("djb33x");
}
else {
}
uwsgi_cache_init(uc);
}
Reference in New Issue View Git Blame Copy Permalink