cache-server implementation

cache.c

@@ -582,3 +582,122 @@ end:
         return ret;
 
 }
+
+/* THIS PART IS HEAVILY OPTIMIZED: PERFORMANCE NOT ELEGANCE !!! */
+
+void *cache_thread_loop(void *fd_ptr) {
+
+	int fd = (int) fd_ptr;
+	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;
+
+		val = uwsgi_cache_get(key, keylen, &vallen);
+                if (val && vallen > 0) {
+                	if (write(ctl_poll.fd, val, vallen) < 0) {
+				uwsgi_error("cache write()");
+			}
+		}
+
+clear:
+		close(ctl_poll.fd);
+	}
+
+}
+
+int uwsgi_cache_server(char *socket, int threads) {
+
+        int fd;
+	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;
+}
+

uwsgi.c

@@ -88,6 +88,8 @@ static struct option long_base_options[] = {
 	{"cache-blocksize", required_argument, 0, LONG_ARGS_CACHE_BLOCKSIZE},
 	{"cache-store", required_argument, 0, LONG_ARGS_CACHE_STORE},
 	{"cache-store-sync", required_argument, 0, LONG_ARGS_CACHE_STORE_SYNC},
+	{"cache-server", required_argument, 0, LONG_ARGS_CACHE_SERVER},
+	{"cache-server-threads", required_argument, 0, LONG_ARGS_CACHE_SERVER_THREADS},
 	{"queue", required_argument, 0, LONG_ARGS_QUEUE},
 	{"queue-blocksize", required_argument, 0, LONG_ARGS_QUEUE_BLOCKSIZE},
 	{"queue-store", required_argument, 0, LONG_ARGS_QUEUE_STORE},
@@ -813,6 +815,7 @@ int main(int argc, char *argv[], char *envp[]) {
 
 	uwsgi.signal_socket = -1;
 	uwsgi.my_signal_socket = -1;
+	uwsgi.cache_server_fd = -1;
 
 	uwsgi.emperor_fd_config = -1;
 	uwsgi.emperor_pid = -1;
@@ -1631,6 +1634,10 @@ int uwsgi_start(void *v_argv) {
 				uwsgi_log("!!! unable to attach daemon %s !!!\n", uwsgi.startup_daemons[i]);
 			}
 		}
+		// create the cache server
+		if (uwsgi.cache_server) {
+			uwsgi.cache_server_fd = uwsgi_cache_server(uwsgi.cache_server, uwsgi.cache_server_threads);
+		}
 	}
 
 	/* plugin initialization */
@@ -1687,6 +1694,12 @@ int uwsgi_start(void *v_argv) {
 					if (j == uwsgi.emperor_fd)
 						continue;
 				}
+
+				if (uwsgi.cache_server && uwsgi.cache_server_fd != -1) {
+					if (j == uwsgi.cache_server_fd)
+						continue;
+				}
+
 				socket_type_len = sizeof(struct sockaddr_un);
 				gsa.sa = (struct sockaddr *) &usa;
 				if (!getsockname(j, gsa.sa, &socket_type_len)) {
@@ -2965,6 +2978,12 @@ static int manage_base_opt(int i, char *optarg) {
 	case LONG_ARGS_CHECK_INTERVAL:
 		uwsgi.shared->options[UWSGI_OPTION_MASTER_INTERVAL] = atoi(optarg);
 		return 1;
+	case LONG_ARGS_CACHE_SERVER:
+		uwsgi.cache_server = optarg;
+		return 1;
+	case LONG_ARGS_CACHE_SERVER_THREADS:
+		uwsgi.cache_server_threads = atoi(optarg);
+		return 1;
 	case LONG_ARGS_CACHE:
 		uwsgi.cache_max_items = atoi(optarg);
 		return 1;

uwsgi.h

@@ -456,6 +456,8 @@ struct uwsgi_opt {
 #define LONG_ARGS_LOGTO2		17126
 #define LONG_ARGS_VASSAL_SOS_BACKLOG	17127
 #define LONG_ARGS_EMPEROR_BROODLORD	17128
+#define LONG_ARGS_CACHE_SERVER		17129
+#define LONG_ARGS_CACHE_SERVER_THREADS	17130
 
 
 #define UWSGI_OK	0
@@ -1237,6 +1239,11 @@ struct uwsgi_server {
 	size_t cache_filesize;
 	int cache_store_sync;
 
+	char *cache_server;
+	int cache_server_threads;
+	int cache_server_fd;
+	pthread_mutex_t cache_server_lock;
+
 	uint64_t queue_size;
 	uint64_t queue_blocksize;
 	void *queue;
@@ -2073,6 +2080,8 @@ int uwsgi_endswith(char *, char *);
 
 char *uwsgi_get_var(struct wsgi_request *, char *, uint16_t *);
 
+int uwsgi_cache_server(char *, int);
+
 #ifdef __cplusplus
 }
 #endif