diff --git a/core/cache.c b/core/cache.c index 4cc9bc91..43950df9 100644 --- a/core/cache.c +++ b/core/cache.c @@ -319,6 +319,14 @@ int uwsgi_cache_set(char *key, uint16_t keylen, char *val, uint64_t vallen, uint uci->valsize = vallen; ret = 0; } + + if (uwsgi.cache_udp_node && ret == 0) { + struct uwsgi_string_list *usl = uwsgi.cache_udp_node; + while(usl) { + uwsgi_log("sending cache update to %s\n", usl->value); + usl = usl->next; + } + } end: return ret; diff --git a/core/master.c b/core/master.c index 14a480bb..ae83f878 100644 --- a/core/master.c +++ b/core/master.c @@ -352,6 +352,46 @@ 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(;;) { + 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 <= 0) { + uwsgi_error("[cache-udp-server] read()"); + } + uwsgi_log("received %llu bytes\n", len); + } + + return NULL; +} + void *cache_sweeper_loop(void *noarg) { int i; @@ -602,6 +642,7 @@ int master_loop(char **argv, char **environ) { pthread_t logger_thread; pthread_t cache_sweeper; + pthread_t cache_udp_server; #ifdef UWSGI_UDP int udp_fd = -1; @@ -707,6 +748,16 @@ int master_loop(char **argv, char **environ) { } } + 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"); + } + } + uwsgi.wsgi_req->buffer = uwsgi.workers[0].cores[0].buffer; diff --git a/core/uwsgi.c b/core/uwsgi.c index 094791de..21e29223 100644 --- a/core/uwsgi.c +++ b/core/uwsgi.c @@ -184,6 +184,8 @@ static struct uwsgi_option uwsgi_base_options[] = { {"cache-no-expire", no_argument, 0, "disable auto sweep of expired items", uwsgi_opt_true, &uwsgi.cache_no_expire, 0}, {"cache-expire-freq", required_argument, 0, "set the frequency of cache sweeper scans (default 3 seconds)", uwsgi_opt_set_int, &uwsgi.cache_expire_freq, 0}, {"cache-report-freed-items", no_argument, 0, "constantly report the cache item freed by the sweeper (use only for debug)", uwsgi_opt_true, &uwsgi.cache_report_freed_items, 0}, + {"cache-udp-server", required_argument, 0, "bind the cache udp server (used only for set/update/delete) to the specified socket", uwsgi_opt_add_string_list, &uwsgi.cache_udp_server, UWSGI_OPT_MASTER}, + {"cache-udp-node", required_argument, 0, "send cache update/deletion to the specified cache udp server", uwsgi_opt_add_string_list, &uwsgi.cache_udp_node, UWSGI_OPT_MASTER}, {"queue", required_argument, 0, "enable shared queue", uwsgi_opt_set_int, &uwsgi.queue_size, 0}, {"queue-blocksize", required_argument, 0, "set queue blocksize", uwsgi_opt_set_int, &uwsgi.queue_store_sync, 0}, diff --git a/uwsgi.h b/uwsgi.h index 90d476bb..16ee7476 100644 --- a/uwsgi.h +++ b/uwsgi.h @@ -1833,6 +1833,9 @@ struct uwsgi_server { int cache_expire_freq; int cache_report_freed_items; + struct uwsgi_string_list *cache_udp_server; + struct uwsgi_string_list *cache_udp_node; + char *cache_server; int cache_server_threads; int cache_server_fd;