added the uWSGI emperor

emperor.c

@@ -0,0 +1,272 @@
+#include "uwsgi.h"
+
+extern struct uwsgi_server uwsgi;
+
+struct uwsgi_instance {
+	struct uwsgi_instance *ui_prev;
+	struct uwsgi_instance *ui_next;
+
+	char name[0xff];
+	pid_t pid ;
+
+	int status;
+	time_t born;
+	time_t last_mod;
+
+	uint64_t respawns;
+
+	int pipe[2];
+};
+
+struct uwsgi_instance *ui;
+
+struct uwsgi_instance *emperor_get(char *name) {
+	
+	struct uwsgi_instance *c_ui = ui;
+	
+	while(c_ui->ui_next) {
+		c_ui = c_ui->ui_next;
+
+		if (!strcmp(c_ui->name, name)) {
+			return c_ui;
+		}
+	}
+	return NULL;
+}
+
+void emperor_del(struct uwsgi_instance *c_ui) {
+
+	struct uwsgi_instance *parent_ui = c_ui->ui_prev;
+	struct uwsgi_instance *child_ui = c_ui->ui_next;
+
+	parent_ui->ui_next = child_ui;
+	if (child_ui) {
+		child_ui->ui_prev = parent_ui;
+	}
+
+	// this will destroy the whole uWSGI instance (and workers)
+	close(c_ui->pipe[0]);
+
+	uwsgi_log("removed uwsgi instance %s\n", c_ui->name);
+
+	free(c_ui);
+
+}
+
+void emperor_stop(struct uwsgi_instance *c_ui) {
+	// remove uWSGI instance
+
+	if (write(c_ui->pipe[0], "\0", 1) != 1) {
+		uwsgi_error("write()");
+	}
+
+	c_ui->status = 1;
+
+	uwsgi_log("stop the uwsgi instance %s\n", c_ui->name);
+}
+
+void emperor_respawn(struct uwsgi_instance *c_ui, time_t mod) {
+
+	// reload the uWSGI instance
+	if (write(c_ui->pipe[0], "\1", 1) != 1) {
+                uwsgi_error("write()");
+        }
+
+        c_ui->respawns++;
+	c_ui->last_mod = mod;
+
+        uwsgi_log("reload the uwsgi instance %s\n", c_ui->name);
+}
+
+void emperor_add(char *name, time_t born) {
+
+	struct uwsgi_instance *c_ui = ui;
+	struct uwsgi_instance *n_ui = NULL;
+	pid_t pid ;
+	char *argv[4];
+	char *uef ;
+
+        while(c_ui->ui_next) {
+                c_ui = c_ui->ui_next;
+        }
+
+	n_ui = uwsgi_malloc(sizeof(struct uwsgi_instance));
+	memset(n_ui, 0, sizeof(struct uwsgi_instance));
+
+	c_ui->ui_next = n_ui;
+	uwsgi_log("c_ui->ui_next = %p\n", c_ui->ui_next);
+	n_ui->ui_prev = c_ui;
+	memcpy(n_ui->name, name, strlen(name));
+	n_ui->born = born;
+	n_ui->last_mod = born;
+
+	if (socketpair(AF_UNIX, SOCK_STREAM, 0, n_ui->pipe)) {
+		uwsgi_error("socketpair()");
+		goto clear;
+	}
+
+	// a new uWSGI instance will start 
+	pid = fork();
+	if (pid < 0) {
+		uwsgi_error("fork()")
+	}
+	else if (pid > 0) {
+		n_ui->pid = pid;
+		// close the right side of the pipe
+		close(n_ui->pipe[1]);
+		return;
+	}
+	else {
+		uef = uwsgi_num2str(n_ui->pipe[1]);
+		if (setenv("UWSGI_EMPEROR_FD", uef, 1)) {
+			uwsgi_error("setenv()");
+			exit(1);
+		}
+		free(uef);
+
+		// close the left side of the pipe
+		close(n_ui->pipe[0]);
+
+		// set args
+		argv[0] = uwsgi.binary_path;
+		if (!strcmp(name+(strlen(name)-4), ".xml")) argv[1] = "--xml";
+		if (!strcmp(name+(strlen(name)-4), ".ini")) argv[1] = "--ini";
+		if (!strcmp(name+(strlen(name)-4), ".yml")) argv[1] = "--yaml";
+		if (!strcmp(name+(strlen(name)-5), ".yaml")) argv[1] = "--yaml";
+		argv[2] = name;
+		argv[3] = NULL;
+		// start !!!
+		if (execvp(argv[0], argv)) {
+			uwsgi_error("execvp()");
+		}
+		// never here
+		exit(1);
+	}
+
+clear:
+
+	free(n_ui);
+	c_ui->ui_next = NULL;
+
+}
+
+void emperor_loop() {
+
+	// monitor a directory
+
+	struct uwsgi_instance ui_base;
+	struct uwsgi_instance *ui_current;
+	struct stat st;
+
+	pid_t diedpid;
+	int waitpid_status;
+	int has_children = 0;
+	int i_am_alone = 0;
+
+	memset(&ui_base, 0, sizeof(struct uwsgi_instance));
+
+	uwsgi_log("*** starting uWSGI Emperor ***\n");
+
+	if (chdir(uwsgi.emperor_dir)) {
+		uwsgi_error("chdir()");
+		exit(1);
+	}
+	struct dirent *de;
+
+	ui = &ui_base;
+
+	for(;;) {
+
+
+		if (!i_am_alone) {
+			diedpid = waitpid(uwsgi.emperor_pid, &waitpid_status, WNOHANG);
+			if (diedpid < 0) {
+				uwsgi_error("waitpid()");
+			}
+			else if (diedpid > 0) {
+				i_am_alone = 1;
+			}
+		}
+
+		DIR *dir = opendir(".");
+		while((de = readdir(dir)) != NULL) {
+			if (!strcmp(de->d_name+(strlen(de->d_name)-4), ".xml") ||
+				!strcmp(de->d_name+(strlen(de->d_name)-4), ".ini") ||
+				!strcmp(de->d_name+(strlen(de->d_name)-4), ".yml") ||
+				!strcmp(de->d_name+(strlen(de->d_name)-5), ".yaml")
+				) {
+
+				
+				if (strlen(de->d_name) >= 0xff) continue;
+
+				if (stat(de->d_name, &st)) continue;
+
+				if (!S_ISREG(st.st_mode)) continue;
+		
+				ui_current = emperor_get(de->d_name);
+
+				if (ui_current) {
+					// check if mtime is changed and the uWSGI instance must be reloaded
+					if (st.st_mtime > ui_current->last_mod) {
+						emperor_respawn(ui_current, st.st_mtime);
+					}
+				}
+				else {
+					emperor_add(de->d_name, st.st_mtime);
+				}	
+			}
+		}
+		closedir(dir);
+
+		// check for removed instances
+
+		ui_current = ui;
+		has_children = 0;
+                while(ui_current->ui_next) {
+                	ui_current = ui_current->ui_next;
+			has_children++;
+		}
+
+		if (has_children) {
+			diedpid = waitpid(WAIT_ANY, &waitpid_status, WNOHANG);
+		}
+		else {
+			diedpid = 0;
+		}
+		if (diedpid < 0) {
+			uwsgi_error("waitpid()");
+		}
+		ui_current = ui;
+        	while(ui_current->ui_next) {
+                	ui_current = ui_current->ui_next;
+			if (ui_current->pid == diedpid) {
+				if (ui_current->status == 0) {
+					// respawn an accidentally dead instance
+					emperor_add(ui_current->name, ui_current->last_mod);
+					emperor_del(ui_current);
+					break;
+				}
+				else if (ui_current->status == 1) {
+					// remove 'marked for dead' instance
+					emperor_del(ui_current);
+					break;
+				}
+			}
+			else if (ui_current->status == 1) {
+				emperor_del(ui_current);
+				break;
+			}
+			else if (stat(ui_current->name, &st)) {
+				emperor_stop(ui_current);
+			}
+        	}
+
+		sleep(3);
+		
+	}
+	
+}
+
+
+
+

master.c

@@ -122,6 +122,10 @@ void master_loop(char **argv, char **environ) {
 	
 
 	uwsgi.wsgi_req->buffer = uwsgi.async_buf[0];
+
+	if (uwsgi.has_emperor) {
+		event_queue_add_fd_read(uwsgi.master_queue, uwsgi.emperor_fd);
+	}
 #ifdef UWSGI_UDP
 	if (uwsgi.udp_socket) {
 		udp_fd = bind_to_udp(uwsgi.udp_socket, 0, 0);
@@ -263,6 +267,14 @@ void master_loop(char **argv, char **environ) {
 						break;
 					}
 				}
+
+				if (!found) {
+					if (uwsgi.has_emperor) {
+						if (i == uwsgi.emperor_fd) {
+							found = 1;
+						}
+					}
+				}
 				if (!found) {
 					close(i);
 				}
@@ -360,6 +372,30 @@ void master_loop(char **argv, char **environ) {
 						}
 					}
 
+					if (uwsgi.has_emperor) {
+						if (interesting_fd == uwsgi.emperor_fd) {
+							char byte;
+							rlen = read(uwsgi.emperor_fd, &byte, 1);
+                                                        if (rlen > 0) {
+								uwsgi_log("received message %d from emperor\n", byte);
+								// remove me
+								if (byte == 0) {
+									close(uwsgi.emperor_fd);
+									kill_them_all();
+								}
+								// reload me
+								else if (byte == 1) {
+									grace_them_all();
+								}
+                                                        }
+							else {
+								uwsgi_log("lost connection with my emperor !!!\n");
+								close(uwsgi.emperor_fd);
+								kill_them_all();
+							}
+						}
+					}
+
 #ifdef UWSGI_UDP
 					if (uwsgi.udp_socket && interesting_fd == udp_fd) {
 						udp_len = sizeof(udp_client);
@@ -500,7 +536,7 @@ void master_loop(char **argv, char **environ) {
 			if (uwsgi.logfile) {
 				uwsgi.shared->logsize = lseek(2, 0, SEEK_CUR);
 				if (uwsgi.shared->logsize > 8192) {
-					uwsgi_log("logsize: %d\n", uwsgi.shared->logsize);
+					//uwsgi_log("logsize: %d\n", uwsgi.shared->logsize);
 					char *new_logfile = uwsgi_malloc(strlen(uwsgi.logfile) + 14 + 1);
 					memset(new_logfile, 0, strlen(uwsgi.logfile) + 14 + 1);    
 					if (!rename(uwsgi.logfile, new_logfile)) {

utils.c

@@ -1464,3 +1464,11 @@ void spawn_daemon(struct uwsgi_daemon *ud) {
 
 	return;
 }
+
+char *uwsgi_num2str(int num) {
+	
+	char *str = uwsgi_malloc(11);	
+
+	snprintf(str, 11, "%d", num);
+	return str;
+}

uwsgi.c

@@ -56,6 +56,7 @@ static struct option long_base_options[] = {
 #endif
 	{"single-interpreter", no_argument, 0, 'i'},
 	{"master", no_argument, 0, 'M'},
+	{"emperor", required_argument, 0, LONG_ARGS_EMPEROR},
 	{"help", no_argument, 0, 'h'},
 	{"reaper", no_argument, 0, 'r'},
 	{"max-requests", required_argument, 0, 'R'},
@@ -384,6 +385,8 @@ int main(int argc, char *argv[], char *envp[])
 	struct utsname uuts;
 #endif
 
+	char *emperor_env;
+
 	signal(SIGHUP, SIG_IGN);
 	signal(SIGTERM, SIG_IGN);
 
@@ -458,6 +461,14 @@ int main(int argc, char *argv[], char *envp[])
 	uwsgi.rl.rlim_cur = 0;
 	uwsgi.rl.rlim_max = 0;
 
+	emperor_env = getenv("UWSGI_EMPEROR_FD");
+	if (emperor_env) {
+		uwsgi.has_emperor = 1;
+		uwsgi.emperor_fd = atoi(emperor_env);
+		uwsgi.master_process = 1;
+		uwsgi.no_orphans = 1;
+		uwsgi_log("*** has_emperor mode detected (fd: %d) ***\n", uwsgi.emperor_fd);
+	}
 
 	env_reloads = getenv("UWSGI_RELOADS");
 	if (env_reloads) {
@@ -481,9 +492,6 @@ int main(int argc, char *argv[], char *envp[])
 
 	uwsgi.binary_path = argv[0];
 
-	uwsgi_register_loop("simple", simple_loop);
-	uwsgi_register_loop("async", complex_loop);
-
 	//initialize embedded plugins
 	UWSGI_LOAD_EMBEDDED_PLUGINS
 
@@ -529,7 +537,6 @@ int main(int argc, char *argv[], char *envp[])
 	}
 
 
-
 	   if (optind < argc) {
 	   	char *lazy = argv[optind];
 		if (lazy[0] != '[') {
@@ -601,6 +608,9 @@ int main(int argc, char *argv[], char *envp[])
 		uwsgi_error("gethostname()");
 	}
 
+
+
+
 #ifdef UWSGI_UDP
         // get cluster configuration
 	if (uwsgi.cluster != NULL) {
@@ -928,6 +938,25 @@ int uwsgi_start(void *v_argv) {
 	}
 #endif
 
+	// end of generic initialization
+
+	// start the Emperor if needed
+	if (uwsgi.emperor_dir) {
+		uwsgi.emperor_pid = fork();
+		if (uwsgi.emperor_pid < 0) {
+			uwsgi_error("pid()");
+			exit(1);
+		}
+		else if (uwsgi.emperor_pid > 0) {
+			emperor_loop();
+			// never here
+			exit(1);
+		}
+	}
+
+
+	uwsgi_register_loop("simple", simple_loop);
+	uwsgi_register_loop("async", complex_loop);
 
 	if (uwsgi.async > 1) {
 		if (!getrlimit(RLIMIT_NOFILE, &uwsgi.rl)) {
@@ -1156,6 +1185,7 @@ int uwsgi_start(void *v_argv) {
 #ifdef UWSGI_MULTICAST
 					if (j == uwsgi.cluster_fd) continue;
 #endif
+					if (uwsgi.has_emperor) { if (j == uwsgi.emperor_fd) continue; }
 					socket_type_len = sizeof(struct sockaddr_un);
 					gsa.sa = (struct sockaddr *) & usa;
 					if (!getsockname(j, gsa.sa, &socket_type_len)) {
@@ -1773,6 +1803,9 @@ end:
 		case LONG_ARGS_LOGTO:
 			logto(optarg);
 			return 1;
+		case LONG_ARGS_EMPEROR:
+			uwsgi.emperor_dir = optarg;
+			return 1;
 		case LONG_ARGS_LOG_MASTER:
 			uwsgi.log_master = 1;
 			return 1;

uwsgi.h

@@ -322,6 +322,7 @@ struct uwsgi_opt {
 #define LONG_ARGS_LOG_MASTER		17069
 #define LONG_ARGS_CHECK_STATIC		17070
 #define LONG_ARGS_WORKER_EXEC		17071
+#define LONG_ARGS_EMPEROR		17072
 
 
 
@@ -663,6 +664,11 @@ struct uwsgi_server {
 	int             apps_cnt;
 	int             default_app;
 
+	int		has_emperor;
+	int emperor_fd;
+	char *emperor_dir;
+	pid_t emperor_pid;
+
 	int option_index;
 	struct option *long_options;
 	struct uwsgi_opt **exported_opts;
@@ -1437,3 +1443,6 @@ uint16_t fcgi_get_record(int, char *);
 
 int uwsgi_attach_daemon(char *);
 void spawn_daemon(struct uwsgi_daemon *);
+
+void emperor_loop(void);
+char *uwsgi_num2str(int);

uwsgiconfig.py

@@ -148,7 +148,7 @@ class uConf(object):
         self.config = ConfigParser.ConfigParser()
         print("using profile: %s" % filename)
         self.config.read(filename)
-        self.gcc_list = ['utils', 'protocol', 'socket', 'logging', 'master',
+        self.gcc_list = ['utils', 'protocol', 'socket', 'logging', 'master', 'emperor',
             'plugins', 'lock', 'cache', 'event', 'signal', 'rpc', 'gateway', 'loop', 'uwsgi']
         self.cflags = ['-O2', '-Wall', '-Werror', '-D_LARGEFILE_SOURCE', '-D_FILE_OFFSET_BITS=64'] + os.environ.get("CFLAGS", "").split()
         try: