skeleton for --cache-udp-server and --cache-udp-node

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;

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;
 

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},

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;