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

refactored cache threads

Browse Source
This commit is contained in:
Roberto De Ioris
2012-11-25 17:22:44 +01:00
parent fd86c6cafa
commit a3bbfde884
3 changed files with 107 additions and 100 deletions
Download Patch File Download Diff File Expand all files Collapse all files
core/cache.c
+104
View File
@@ -551,3 +551,107 @@ void uwsgi_cache_rlock() {
void uwsgi_cache_rwunlock() {
uwsgi.lock_ops.rwunlock(uwsgi.cache_lock);
}
void *cache_udp_server_loop(void *noarg) {
// block all signals
sigset_t smask;
sigfillset(&smask);
pthread_sigmask(SIG_BLOCK, &smask, NULL);
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;
// cache set/update
if (buf[3] == 10) {
memcpy(&ss, buf + 4, 2);
if (4+ss > pktsize) continue;
uint16_t keylen = ss;
char *key = buf + 6;
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(uwsgi.cache_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(uwsgi.cache_lock);
}
// cache del
else if (buf[3] == 11) {
}
}
return NULL;
}
void *cache_sweeper_loop(void *noarg) {
int i;
// block all signals
sigset_t smask;
sigfillset(&smask);
pthread_sigmask(SIG_BLOCK, &smask, NULL);
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 < (int) uwsgi.cache_max_items; i++) {
uwsgi_wlock(uwsgi.cache_lock);
if (uwsgi.cache_items[i].expires) {
if (uwsgi.cache_items[i].expires < (uint64_t) uwsgi.current_time) {
uwsgi_cache_del(NULL, 0, i);
freed_items++;
}
}
uwsgi_rwunlock(uwsgi.cache_lock);
}
if (uwsgi.cache_report_freed_items && freed_items > 0) {
uwsgi_log("freed %llu cache items\n", (unsigned long long) freed_items);
}
};
return NULL;
}
core/master.c
-100
View File
@@ -352,106 +352,6 @@ void *logger_thread_loop(void *noarg) {
return NULL;
}
void *cache_udp_server_loop(void *noarg) {
// block all signals
sigset_t smask;
sigfillset(&smask);
pthread_sigmask(SIG_BLOCK, &smask, NULL);
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;
// cache set/update
if (buf[3] == 10) {
memcpy(&ss, buf + 4, 2);
if (4+ss > pktsize) continue;
uint16_t keylen = ss;
char *key = buf + 6;
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(uwsgi.cache_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(uwsgi.cache_lock);
}
}
return NULL;
}
void *cache_sweeper_loop(void *noarg) {
int i;
// block all signals
sigset_t smask;
sigfillset(&smask);
pthread_sigmask(SIG_BLOCK, &smask, NULL);
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 < (int) uwsgi.cache_max_items; i++) {
uwsgi_wlock(uwsgi.cache_lock);
if (uwsgi.cache_items[i].expires) {
if (uwsgi.cache_items[i].expires < (uint64_t) uwsgi.current_time) {
uwsgi_cache_del(NULL, 0, i);
freed_items++;
}
}
uwsgi_rwunlock(uwsgi.cache_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_subscribe(char *subscription, uint8_t cmd) {
int subfile_size;
uwsgi.h
+3
View File
@@ -3485,6 +3485,9 @@ void uwsgi_cache_wlock(void);
void uwsgi_cache_rlock(void);
void uwsgi_cache_rwunlock(void);
void *cache_sweeper_loop(void *);
void *cache_udp_server_loop(void *);
void uwsgi_user_lock(int);
void uwsgi_user_unlock(int);