updated decorators with @farm([name])

.hgtags

@@ -33,3 +33,4 @@ fb168b0b86169219aa9b8e40f0caa6297cf34dbc 0.9.9-rc1
 6aa667612019fcff8e68af9469f73aff6fa378a8 0.9.9-rc2
 376ffd8f190623e5e2521fe9fa64bf03c31e71be 0.9.9
 7048ae11cfc8453c6bb599af3c0dfacd7ebae43c 1.0-rc1
+7a4e021ea7ba2953e474326612c3bb82d6817f0e 1.0-rc2

logging.c

@@ -129,22 +129,22 @@ void log_request(struct wsgi_request *wsgi_req) {
 	rlen = writev(2, logvec, logvecpos+1);
 }
 
-void get_memusage() {
+void get_memusage(uint64_t *rss, uint64_t *vsz) {
 
 #ifdef UNBIT
-	uwsgi.workers[uwsgi.mywid].vsz_size = syscall(356);
+	*vsz = syscall(356);
 #elif defined(__linux__)
 	FILE *procfile;
 	int i;
 	procfile = fopen("/proc/self/stat", "r");
 	if (procfile) {
-		i = fscanf(procfile, "%*s %*s %*s %*s %*s %*s %*s %*s %*s %*s %*s %*s %*s %*s %*s %*s %*s %*s %*s %*s %*s %*s %llu %lld", (unsigned long long *) &uwsgi.workers[uwsgi.mywid].vsz_size, (unsigned long long *) &uwsgi.workers[uwsgi.mywid].rss_size);
+		i = fscanf(procfile, "%*s %*s %*s %*s %*s %*s %*s %*s %*s %*s %*s %*s %*s %*s %*s %*s %*s %*s %*s %*s %*s %*s %llu %lld", (unsigned long long *) vsz, (unsigned long long *) rss);
 		if (i != 2) {
 			uwsgi_log( "warning: invalid record in /proc/self/stat\n");
 		}
 		fclose(procfile);
 	}
-	uwsgi.workers[uwsgi.mywid].rss_size = uwsgi.workers[uwsgi.mywid].rss_size * uwsgi.page_size;
+	*rss = *rss * uwsgi.page_size;
 #elif defined (__sun__)
 	psinfo_t info;
 	int procfd;
@@ -152,8 +152,8 @@ void get_memusage() {
 	procfd = open("/proc/self/psinfo", O_RDONLY);
 	if (procfd >= 0) {
 		if ( read(procfd, (char *) &info, sizeof(info)) > 0) {
-			uwsgi.workers[uwsgi.mywid].rss_size = (uint64_t) info.pr_rssize * 1024;
-			uwsgi.workers[uwsgi.mywid].vsz_size = (uint64_t) info.pr_size * 1024;
+			*rss = (uint64_t) info.pr_rssize * 1024;
+			*vsz = (uint64_t) info.pr_size * 1024;
 		}
 		close(procfd);
 	}
@@ -164,8 +164,8 @@ void get_memusage() {
 	mach_msg_type_number_t t_size = sizeof(struct task_basic_info);
 
 	if (task_info(mach_task_self(), TASK_BASIC_INFO, (task_info_t) & t_info, &t_size) == KERN_SUCCESS) {
-		uwsgi.workers[uwsgi.mywid].rss_size = t_info.resident_size;
-		uwsgi.workers[uwsgi.mywid].vsz_size = t_info.virtual_size;
+		*rss = t_info.resident_size;
+		*vsz = t_info.virtual_size;
 	}
 #elif defined(__FreeBSD__) || defined(__NetBSD__) || defined(__DragonFly__) || defined(__OpenBSD__)
 	kvm_t *kv;
@@ -184,8 +184,8 @@ void get_memusage() {
 		struct kinfo_proc *kproc;
 		kproc = kvm_getprocs(kv, KERN_PROC_PID, uwsgi.mypid, &cnt);
 		if (kproc && cnt > 0) {
-			uwsgi.workers[uwsgi.mywid].vsz_size = kproc->ki_size;
-			uwsgi.workers[uwsgi.mywid].rss_size = kproc->ki_rssize * uwsgi.page_size;
+			*vsz = kproc->ki_size;
+			*rss = kproc->ki_rssize * uwsgi.page_size;
 		}
 #elif defined(__NetBSD__) || defined(__OpenBSD__)
 		struct kinfo_proc2 *kproc2;
@@ -193,11 +193,11 @@ void get_memusage() {
 		kproc2 = kvm_getproc2(kv, KERN_PROC_PID, uwsgi.mypid, sizeof(struct kinfo_proc2), &cnt);
 		if (kproc2 && cnt > 0) {
 #ifdef __OpenBSD__
-			uwsgi.workers[uwsgi.mywid].vsz_size = (kproc2->p_vm_dsize + kproc2->p_vm_ssize + kproc2->p_vm_tsize) * uwsgi.page_size;
+			*vsz = (kproc2->p_vm_dsize + kproc2->p_vm_ssize + kproc2->p_vm_tsize) * uwsgi.page_size;
 #else
-			uwsgi.workers[uwsgi.mywid].vsz_size = kproc2->p_vm_msize * uwsgi.page_size;
+			*vsz = kproc2->p_vm_msize * uwsgi.page_size;
 #endif
-			uwsgi.workers[uwsgi.mywid].rss_size = kproc2->p_vm_rssize * uwsgi.page_size;
+			*rss = kproc2->p_vm_rssize * uwsgi.page_size;
 		}
 #endif
 
@@ -207,13 +207,14 @@ void get_memusage() {
 	area_info ai;
 	int32 cookie;
 
-	uwsgi.workers[uwsgi.mywid].vsz_size = 0;
-	uwsgi.workers[uwsgi.mywid].rss_size = 0;
+	*vsz = 0;
+	*rss = 0;
 	while(get_next_area_info(0, &cookie, &ai) == B_OK) {
-		uwsgi.workers[uwsgi.mywid].vsz_size += ai.ram_size;
+		*vsz += ai.ram_size;
 		if ( (ai.protection & B_WRITE_AREA) != 0) {
-			uwsgi.workers[uwsgi.mywid].rss_size += ai.ram_size;
+			*rss += ai.ram_size;
 		}
 	}
 #endif
+
 }

master_utils.c

@@ -71,6 +71,15 @@ void uwsgi_fixup_fds(int wid, int muleid) {
 				}
                         }
 
+			for(i=0;i<uwsgi.farms_cnt;i++) {
+                                if (uwsgi.farms[i].signal_pipe[0] != -1) close(uwsgi.farms[i].signal_pipe[0]);
+
+				if (muleid == 0) {
+                                	if (uwsgi.farms[i].signal_pipe[1] != -1) close(uwsgi.farms[i].signal_pipe[1]);
+                                	if (uwsgi.farms[i].queue_pipe[1] != -1) close(uwsgi.farms[i].queue_pipe[1]);
+				}
+			}
+
                 }
 
 	

mule.c

@@ -63,6 +63,64 @@ void uwsgi_mule(int id) {
 	}
 }
 
+int uwsgi_farm_has_mule(struct uwsgi_farm *farm, int muleid) {
+
+	struct uwsgi_mule_farm *umf = farm->mules;
+
+	while(umf) {
+		if (umf->mule->id == muleid) {
+			return 1;
+		}
+		umf = umf->next;
+	}
+
+	return 0;
+}
+
+int farm_has_signaled(int fd) {
+
+	int i;
+	for(i=0;i<uwsgi.farms_cnt;i++) {
+		struct uwsgi_mule_farm *umf = uwsgi.farms[i].mules;
+		while(umf) {
+			if (umf->mule->id == uwsgi.muleid && uwsgi.farms[i].signal_pipe[1] == fd) {
+				return 1;
+			}
+			umf = umf->next;
+		}
+	}
+	
+	return 0;
+}
+
+int farm_has_msg(int fd) {
+
+        int i;
+        for(i=0;i<uwsgi.farms_cnt;i++) {
+                struct uwsgi_mule_farm *umf = uwsgi.farms[i].mules;
+                while(umf) {
+                        if (umf->mule->id == uwsgi.muleid && uwsgi.farms[i].queue_pipe[1] == fd) {
+                                return 1;
+                        }
+                        umf = umf->next;
+                }
+        }
+
+        return 0;
+}
+
+
+void uwsgi_mule_add_farm_to_queue(int queue) {
+
+	int i;
+	for(i=0;i<uwsgi.farms_cnt;i++) {
+		if (uwsgi_farm_has_mule(&uwsgi.farms[i], uwsgi.muleid)) {
+			event_queue_add_fd_read(queue, uwsgi.farms[i].signal_pipe[1]);	
+			event_queue_add_fd_read(queue, uwsgi.farms[i].queue_pipe[1]);	
+		}
+	}
+}
+
 void uwsgi_mule_handler() {
 	
 	ssize_t len;
@@ -79,13 +137,15 @@ void uwsgi_mule_handler() {
 	event_queue_add_fd_read(mule_queue, uwsgi.my_signal_socket);
 	event_queue_add_fd_read(mule_queue, uwsgi.mules[uwsgi.muleid-1].queue_pipe[1]);
 
+	uwsgi_mule_add_farm_to_queue(mule_queue);
+
 	for(;;) {
 		rlen = event_queue_wait(mule_queue, -1, &interesting_fd);
 		if (rlen <= 0) {
 			continue;
 		}
 
-		if (interesting_fd == uwsgi.signal_socket || interesting_fd == uwsgi.my_signal_socket) {
+		if (interesting_fd == uwsgi.signal_socket || interesting_fd == uwsgi.my_signal_socket || farm_has_signaled(interesting_fd)) {
 			len = read(interesting_fd, &uwsgi_signal, 1);
 			if (len <= 0) {
                 		uwsgi_log_verbose("uWSGI mule %d braying: my master died, i will follow him...\n", uwsgi.muleid);
@@ -98,8 +158,8 @@ void uwsgi_mule_handler() {
                 		uwsgi_log_verbose("error managing signal %d on mule %d\n", uwsgi_signal, uwsgi.mywid);
                 	}
 		}
-		else if (interesting_fd == uwsgi.mules[uwsgi.muleid-1].queue_pipe[1]) {
-			len = read(uwsgi.mules[uwsgi.muleid-1].queue_pipe[1], message, 65536);
+		else if (interesting_fd == uwsgi.mules[uwsgi.muleid-1].queue_pipe[1] || farm_has_msg(interesting_fd)) {
+			len = read(interesting_fd, message, 65536);
 			if (len < 0) {
 				uwsgi_error("read()");
 			}	
@@ -107,9 +167,47 @@ void uwsgi_mule_handler() {
 				exit(1);
 			}
 			else {
-				uwsgi_log("*** mule %d received a %d bytes message\n", uwsgi.muleid, len);
+				uwsgi_log("*** mule %d received a %d bytes message ***\n", uwsgi.muleid, len);
 			}
 		}
 	}
 
 }
+
+struct uwsgi_mule *get_mule_by_id(int id) {
+
+	int i;
+
+	for(i=0;i<uwsgi.mules_cnt;i++) {
+		if (uwsgi.mules[i].id == id) {
+			return &uwsgi.mules[i];
+		}
+	}
+
+	return NULL;
+}
+
+struct uwsgi_mule_farm *uwsgi_mule_farm_new(struct uwsgi_mule_farm **umf, struct uwsgi_mule *um) {
+
+        struct uwsgi_mule_farm *uwsgi_mf = *umf, *old_umf;
+
+        if (!uwsgi_mf) {
+                *umf = uwsgi_malloc(sizeof(struct uwsgi_mule_farm));
+                uwsgi_mf = *umf;
+        }
+        else {
+                while(uwsgi_mf) {
+                        old_umf = uwsgi_mf;
+			uwsgi_mf = uwsgi_mf->next;
+                }
+
+                uwsgi_mf = uwsgi_malloc(sizeof(struct uwsgi_mule_farm));
+                old_umf->next = uwsgi_mf;
+        }
+
+        uwsgi_mf->mule = um;
+        uwsgi_mf->next = NULL;
+
+        return uwsgi_mf;
+}
+

plugins/psgi/psgi.h

@@ -34,7 +34,7 @@ struct uwsgi_perl {
 
 void init_perl_embedded_module(void);
 void uwsgi_psgi_app(void);
-int psgi_response(struct wsgi_request *, PerlInterpreter *, AV*);
+int psgi_response(struct wsgi_request *, AV*);
 
 #define psgi_xs(func) newXS("uwsgi::" #func, XS_##func, "uwsgi")
 #define psgi_check_args(x) if (items < x) Perl_croak(aTHX_ "Usage: uwsgi::%s takes %d arguments", __FUNCTION__ + 3, x)

plugins/psgi/psgi_loader.c

@@ -41,19 +41,11 @@ XS(XS_stream)
     AV *response = (AV* ) SvREFCNT_inc(SvRV(ST(0))) ;
 
 	if (av_len(response) == 2) {
-
-#ifdef my_perl
-		while (psgi_response(wsgi_req, my_perl, response) != UWSGI_OK);
-#else
-		while (psgi_response(wsgi_req, uperl.main, response) != UWSGI_OK);
-#endif
+		while (psgi_response(wsgi_req, response) != UWSGI_OK);
 	}
 	else if (av_len(response) == 1) {
-#ifdef my_perl
-		while (psgi_response(wsgi_req, my_perl, response) != UWSGI_OK);
-#else
-		while (psgi_response(wsgi_req, uperl.main, response) != UWSGI_OK);
-#endif
+		while (psgi_response(wsgi_req, response) != UWSGI_OK);
+
 		SvREFCNT_dec(response);
                 ST(0) = sv_bless(newRV(sv_newmortal()), uperl.streaming_stash);
                 XSRETURN(1);
@@ -236,6 +228,7 @@ xs_init(pTHX)
 void uwsgi_psgi_app() {
 
         struct stat stat_psgi;
+	int id = uwsgi_apps_cnt;
 
         if (uperl.psgi) {
 
@@ -291,7 +284,6 @@ void uwsgi_psgi_app() {
 
                 uperl.psgibuffer[stat_psgi.st_size] = 0;
 
-                if (uwsgi.threads < 2) {
                         uperl.psgi_main = perl_eval_pv(uwsgi_concat4("#line 1 ", uperl.psgi, "\n", uperl.psgibuffer), 0);
                         if (!uperl.psgi_main) {
                                 uwsgi_log("unable to find PSGI function entry point.\n");
@@ -303,11 +295,31 @@ void uwsgi_psgi_app() {
                                 exit(1);
                         }
 
+                if (uwsgi.threads < 2) {
                         free(uperl.psgibuffer);
                         close(uperl.fd);
-                }
+		}
+
+		struct uwsgi_app *wi = &uwsgi_apps[id];
+		memset(wi, 0, sizeof(struct uwsgi_app));
+
+		wi->modifier1 = 5;
+		wi->mountpoint = "";
+		wi->mountpoint_len = 0;
+
+                uwsgi_log("PSGI app %d (%s) loaded at %p\n", id, uperl.psgi, uperl.psgi_main);
+
+		uwsgi_apps_cnt++;
+	
+		// check if we need to emulate fork() COW
+		int i;
+        	if (uwsgi.mywid == 0) {
+                	for(i=1;i<=uwsgi.numproc;i++) {
+                        	memcpy(&uwsgi.workers[i].apps[id], &uwsgi.workers[0].apps[id], sizeof(struct uwsgi_app));
+                        	uwsgi.workers[i].apps_cnt = uwsgi_apps_cnt;
+                	}
+        	}
 
-                uwsgi_log("PSGI app (%s) loaded at %p\n", uperl.psgi, uperl.psgi_main);
         }
 
 

plugins/psgi/psgi_plugin.c

@@ -23,7 +23,7 @@ SV *uwsgi_perl_obj_new(char *class, size_t class_len) {
 
 	SV *newobj;
 	// set current context ?
-	//dTHX;
+	dTHX;
 	dSP;
 
 	ENTER;
@@ -49,7 +49,7 @@ SV *uwsgi_perl_call_stream(SV *func) {
 
 	SV *ret = NULL;
         // set current context ?
-        //dTHX;
+        dTHX;
         dSP;
 
         ENTER;
@@ -79,7 +79,7 @@ int uwsgi_perl_obj_can(SV *obj, char *method, size_t len) {
 
 	int ret;
         // set current context ? needed for threading
-        //dTHX;
+        dTHX;
         dSP;
 
         ENTER;
@@ -107,6 +107,7 @@ SV *uwsgi_perl_obj_call(SV *obj, char *method) {
 
         SV *ret = NULL;
 
+	dTHX;
         dSP;
 
         ENTER;
@@ -140,6 +141,7 @@ AV *psgi_call(struct wsgi_request *wsgi_req, SV *psgi_func, SV *env) {
 
 	AV *ret = NULL;
 
+	dTHX;
         dSP;
 
         ENTER;
@@ -149,6 +151,7 @@ AV *psgi_call(struct wsgi_request *wsgi_req, SV *psgi_func, SV *env) {
 	PUTBACK;
 
 	call_sv(psgi_func, G_SCALAR | G_EVAL);
+
 	SPAGAIN;
 
         if(SvTRUE(ERRSV)) {
@@ -326,51 +329,19 @@ int uwsgi_perl_init(){
 
 }
 
-#ifdef my_perl
+#ifdef USE_ITHREADS
 void uwsgi_perl_enable_threads() {
 	
-	int i;
-
 	if (pthread_key_create(&uperl.u_interpreter, NULL)) {
         	uwsgi_error("pthread_key_create()");
                 exit(1);
 	}
 
-	uperl.interp = malloc( sizeof(PerlInterpreter*) * uwsgi.threads );
-	if (!uperl.interp) {
-		uwsgi_error("malloc()");
-		exit(1);
-	}
+	uperl.interp = uwsgi_malloc( sizeof(PerlInterpreter*) * uwsgi.threads );
+	uperl.interp[0] = uperl.main;
 
-	for(i=1;i<uwsgi.threads;i++) {
-		uperl.interp[i] = perl_clone(uperl.main, CLONEf_KEEP_PTR_TABLE);
-		if (!uperl.interp[i]) {
-			uwsgi_log("unable to create new perl interpreter\n");
-			exit(1);
-		}
-	}
+	uperl.psgi_func = uwsgi_malloc( sizeof(SV*) * uwsgi.threads );
 
-	uperl.psgi_func = malloc( sizeof(SV*) * uwsgi.threads );
-	if (!uperl.psgi_func) {
-		uwsgi_error("malloc()");
-		exit(1);
-	}
-	
-
-	dTHXa(uperl.main);
-	PERL_SET_CONTEXT(uperl.main);
-
-	uperl.psgi_main = perl_eval_pv(uperl.psgibuffer, 0);
-	if (!uperl.psgi_main) {
-		uwsgi_log("unable to find PSGI function entry point.\n");
-		exit(1);
-	}
-
-	if(SvTRUE(ERRSV)) {
-		uwsgi_log("%s\n", SvPV_nolen(ERRSV));
-		exit(1);
-	}
-	
 
 }
 #endif
@@ -381,12 +352,10 @@ int uwsgi_perl_request(struct wsgi_request *wsgi_req) {
 
 	SV **harakiri;
 	SV *psgi_func = uperl.psgi_main;
-	// ugly hack
-	register PerlInterpreter *my_perl = uperl.main;
 
 #ifdef UWSGI_ASYNC
 	if (wsgi_req->async_status == UWSGI_AGAIN) {
-		return psgi_response(wsgi_req, my_perl, wsgi_req->async_placeholder);	
+		return psgi_response(wsgi_req, wsgi_req->async_placeholder);	
 	}
 #endif
 
@@ -401,19 +370,14 @@ int uwsgi_perl_request(struct wsgi_request *wsgi_req) {
 		return -1;
 	}
 
-
-#ifdef UWSGI_THREADING
-	if (uwsgi.threads > 1 && wsgi_req->async_id > 0) {
+	if (uwsgi.threads > 1) {
 		psgi_func = uperl.psgi_func[wsgi_req->async_id];
-		my_perl = pthread_getspecific(uperl.u_interpreter);
 	}
-#endif
-
 
+	
 	ENTER;
 	SAVETMPS;
 
-
 	wsgi_req->async_environ = build_psgi_env(wsgi_req);
 	if (!wsgi_req->async_environ) goto clear;
 
@@ -431,7 +395,7 @@ int uwsgi_perl_request(struct wsgi_request *wsgi_req) {
 		goto clear2;
 	}
 
-	while (psgi_response(wsgi_req, my_perl, wsgi_req->async_result) != UWSGI_OK) {
+	while (psgi_response(wsgi_req, wsgi_req->async_result) != UWSGI_OK) {
 #ifdef UWSGI_ASYNC
 		if (uwsgi.async > 1) {
 			FREETMPS;
@@ -471,26 +435,41 @@ void uwsgi_perl_after_request(struct wsgi_request *wsgi_req) {
 
 }
 
+#ifdef USE_ITHREADS
 void uwsgi_perl_init_thread(int core_id) {
 
 
-	pthread_setspecific(uperl.u_interpreter, uperl.interp[core_id]);
+
+	if (core_id > 0) {
+		uperl.interp[core_id] = perl_clone(uperl.main, CLONEf_KEEP_PTR_TABLE);
+                if (!uperl.interp[core_id]) {
+                        uwsgi_log("unable to create new perl interpreter\n");
+                        exit(1);
+                }
+	}
+
 	dTHXa(uperl.interp[core_id]);
 	PERL_SET_CONTEXT(uperl.interp[core_id]);
 
-	uperl.psgi_func[core_id] = perl_eval_pv(uperl.psgibuffer, 0);
-        if (!uperl.psgi_func[core_id]) {
-        	uwsgi_log("unable to find PSGI function entry point.\n");
-		exit(1);
-        }
+	pthread_setspecific(uperl.u_interpreter, uperl.interp[core_id]);
 
-        if(SvTRUE(ERRSV)) {
-        	uwsgi_log("%s\n", SvPV_nolen(ERRSV));
-		exit(1);
-        }
+	uperl.psgi_func[0] = uperl.psgi_main;
+	if (core_id > 0) {
+		uperl.psgi_func[core_id] = perl_eval_pv(uwsgi_concat4("#line 1 ", uperl.psgi, "\n", uperl.psgibuffer), 0);
 
+        	if (!uperl.psgi_func[core_id]) {
+        		uwsgi_log("unable to find PSGI function entry point.\n");
+                	exit(1);
+        	}
+
+        	if(SvTRUE(ERRSV)) {
+        		uwsgi_log("%s\n", SvPV_nolen(ERRSV));
+                	exit(1);
+        	}
+	}
 
 }
+#endif
 
 int uwsgi_perl_manage_options(int i, char *optarg) {
 
@@ -531,11 +510,11 @@ struct uwsgi_plugin psgi_plugin = {
 	.init_apps = uwsgi_psgi_app,
 	//.magic = uwsgi_perl_magic,
 	//.help = uwsgi_perl_help,
-#ifdef my_perl
+#ifdef USE_ITHREADS
 	.enable_threads = uwsgi_perl_enable_threads,
+	.init_thread = uwsgi_perl_init_thread,
 #endif
 	.manage_opt = uwsgi_perl_manage_options,
-	.init_thread = uwsgi_perl_init_thread,
 	.request = uwsgi_perl_request,
 	.after_request = uwsgi_perl_after_request,
 

plugins/psgi/psgi_response.c

@@ -4,7 +4,7 @@ extern struct uwsgi_server uwsgi;
 
 extern struct http_status_codes hsc[];
 
-int psgi_response(struct wsgi_request *wsgi_req, PerlInterpreter *my_perl, AV *response) {
+int psgi_response(struct wsgi_request *wsgi_req, AV *response) {
 
 	SV **status_code, **hitem ;
 	AV *headers, *body =NULL;

plugins/python/uwsgi_pymodule.c

@@ -1085,6 +1085,70 @@ PyObject *py_uwsgi_setprocname(PyObject * self, PyObject * args) {
 	return Py_None;
 }
 
+PyObject *py_uwsgi_in_farm(PyObject * self, PyObject * args) {
+
+	char *farm_name = NULL;
+	int i;
+
+	if (!PyArg_ParseTuple(args, "|s:in_farm", &farm_name)) {
+                return NULL;
+        }
+
+	if (uwsgi.muleid == 0) goto none;
+
+	for(i=0;i<uwsgi.farms_cnt;i++) {
+		if (!farm_name) {
+			if (uwsgi_farm_has_mule(&uwsgi.farms[i], uwsgi.muleid)) {
+				Py_INCREF(Py_True);
+				return Py_True;
+			}
+		}
+		else {
+			if (!strcmp(farm_name, uwsgi.farms[i].name)) {
+				if (uwsgi_farm_has_mule(&uwsgi.farms[i], uwsgi.muleid)) {
+					Py_INCREF(Py_True);
+					return Py_True;
+				}
+			}
+		}
+	}
+none:
+
+	Py_INCREF(Py_None);
+	return Py_None;
+   
+}
+
+PyObject *py_uwsgi_farm_msg(PyObject * self, PyObject * args) {
+
+        char *message = NULL;
+        Py_ssize_t message_len = 0;
+	char *farm_name = NULL;
+        ssize_t len;
+	int i;
+
+        if (!PyArg_ParseTuple(args, "ss#:farm_msg", &farm_name, &message, &message_len)) {
+                return NULL;
+        }
+
+	for(i=0;i<uwsgi.farms_cnt;i++) {
+	
+		if (!strcmp(farm_name, uwsgi.farms[i].name)) {
+                	len = write(uwsgi.farms[i].queue_pipe[0], message, message_len);
+                	if (len <= 0) {
+                        	uwsgi_error("write()");
+                	}
+			break;
+		}
+	
+        }
+
+        Py_INCREF(Py_None);
+        return Py_None;
+
+}
+
+
 PyObject *py_uwsgi_mule_msg(PyObject * self, PyObject * args) {
 
 	char *message = NULL;
@@ -1131,6 +1195,57 @@ PyObject *py_uwsgi_mule_get_msg(PyObject * self, PyObject * args) {
 	return PyString_FromStringAndSize(message, len);
 }
 
+PyObject *py_uwsgi_farm_get_msg(PyObject * self, PyObject * args) {
+
+        ssize_t len = 0;
+        // this buffer will be configurable
+        char message[65536];
+	int i, count = 0, pos = 0, ret;
+	struct pollfd *farmpoll;
+
+        if (uwsgi.muleid == 0) {
+                return PyErr_Format(PyExc_ValueError, "you can receive mule messages only in a mule !!!");
+        }
+        UWSGI_RELEASE_GIL;
+	for(i=0;i<uwsgi.farms_cnt;i++) {	
+		if (uwsgi_farm_has_mule(&uwsgi.farms[i], uwsgi.muleid)) count++;
+	}
+	farmpoll = uwsgi_malloc( sizeof(struct pollfd) * count);
+	for(i=0;i<uwsgi.farms_cnt;i++) {
+		if (uwsgi_farm_has_mule(&uwsgi.farms[i], uwsgi.muleid)) {
+			farmpoll[pos].fd = uwsgi.farms[i].queue_pipe[1];
+			farmpoll[pos].events = POLLIN;
+			pos++;
+		}
+	}
+
+	ret = poll(farmpoll, count, -1);
+	if (ret <= 0) {
+		uwsgi_error("poll()");
+		free(farmpoll);
+		Py_INCREF(Py_None);
+		return Py_None;
+	}
+
+	for(i=0;i<count;i++) {
+		if (farmpoll[i].revents & POLLIN) {
+        		len = read(farmpoll[i].fd, message, 65536);
+			break;
+		}
+	}
+        UWSGI_GET_GIL;
+        if (len <= 0) {
+                uwsgi_error("read()");
+		free(farmpoll);
+                Py_INCREF(Py_None);
+                return Py_None;
+        }
+
+	free(farmpoll);
+        return PyString_FromStringAndSize(message, len);
+}
+
+
 PyObject *py_uwsgi_extract(PyObject * self, PyObject * args) {
 
         char *name;
@@ -2492,12 +2607,13 @@ PyObject *py_uwsgi_logsize(PyObject * self, PyObject * args) {
 
 PyObject *py_uwsgi_mem(PyObject * self, PyObject * args) {
 
+	uint64_t rss=0, vsz = 0;
 	PyObject *ml = PyTuple_New(2);
 
-	get_memusage();
+	get_memusage(&rss, &vsz);
 
-	PyTuple_SetItem(ml, 0, PyLong_FromUnsignedLongLong(uwsgi.workers[uwsgi.mywid].rss_size));
-	PyTuple_SetItem(ml, 1, PyLong_FromUnsignedLongLong(uwsgi.workers[uwsgi.mywid].vsz_size));
+	PyTuple_SetItem(ml, 0, PyLong_FromUnsignedLongLong(rss));
+	PyTuple_SetItem(ml, 1, PyLong_FromUnsignedLongLong(vsz));
 
 	return ml;
 
@@ -2884,7 +3000,10 @@ static PyMethodDef uwsgi_advanced_methods[] = {
 	{"extract", py_uwsgi_extract, METH_VARARGS, ""},
 
 	{"mule_msg", py_uwsgi_mule_msg, METH_VARARGS, ""},
+	{"farm_msg", py_uwsgi_farm_msg, METH_VARARGS, ""},
 	{"mule_get_msg", py_uwsgi_mule_get_msg, METH_VARARGS, ""},
+	{"farm_get_msg", py_uwsgi_farm_get_msg, METH_VARARGS, ""},
+	{"in_farm", py_uwsgi_in_farm, METH_VARARGS, ""},
 	//{"call_hook", py_uwsgi_call_hook, METH_VARARGS, ""},
 
 	{NULL, NULL},

signal.c

@@ -271,6 +271,20 @@ void uwsgi_route_signal(uint8_t sig) {
                         }
 		}
 	}
+
+	else if (!strncmp(use->receiver, "farm", 4)) {
+		i = atoi(use->receiver+4);
+		if (i > uwsgi.farms_cnt || i <= 0) {
+			uwsgi_log("invalid signal target: %s\n", use->receiver);
+		}
+		else {
+			if (write(uwsgi.farms[i-1].signal_pipe[0], &sig, 1) != 1) {
+                                uwsgi_error("write()");
+                                uwsgi_log("could not deliver signal %d to farm %d (%s)\n", sig, i, uwsgi.farms[i-1].name);
+                        }
+		}
+	}
+
 	else {
 		// unregistered signal, sending it to all the workers
 		uwsgi_log("^^^ UNSUPPORTED SIGNAL TARGET: %s ^^^\n", use->receiver);

utils.c

@@ -753,7 +753,7 @@ void uwsgi_close_request(struct wsgi_request *wsgi_req) {
 
 	int waitpid_status;
 	int tmp_id;
-	uint64_t tmp_rt;
+	uint64_t tmp_rt, rss = 0, vsz = 0;
 
 	gettimeofday(&wsgi_req->end_of_request, NULL);
 	
@@ -763,8 +763,11 @@ void uwsgi_close_request(struct wsgi_request *wsgi_req) {
 	uwsgi.workers[uwsgi.mywid].avg_response_time = (uwsgi.workers[uwsgi.mywid].avg_response_time+tmp_rt)/2;
 
 	// get memory usage
-	if (uwsgi.shared->options[UWSGI_OPTION_MEMORY_DEBUG] == 1 || uwsgi.force_get_memusage )
-		get_memusage();
+	if (uwsgi.shared->options[UWSGI_OPTION_MEMORY_DEBUG] == 1 || uwsgi.force_get_memusage ) {
+		get_memusage(&rss, &vsz);
+		uwsgi.workers[uwsgi.mywid].vsz_size = vsz;
+		uwsgi.workers[uwsgi.mywid].rss_size = rss;
+	}
 
 
 	// close the connection with the webserver
@@ -813,11 +816,11 @@ void uwsgi_close_request(struct wsgi_request *wsgi_req) {
 		goodbye_cruel_world();
 	}
 
-	if (uwsgi.reload_on_as && (rlim_t) uwsgi.workers[uwsgi.mywid].vsz_size >= uwsgi.reload_on_as) {
+	if (uwsgi.reload_on_as && (rlim_t) vsz >= uwsgi.reload_on_as) {
 		goodbye_cruel_world();
 	}
 
-	if (uwsgi.reload_on_rss && (rlim_t) uwsgi.workers[uwsgi.mywid].rss_size >= uwsgi.reload_on_rss) {
+	if (uwsgi.reload_on_rss && (rlim_t) rss >= uwsgi.reload_on_rss) {
 		goodbye_cruel_world();
 	}
 

uwsgi.c

@@ -126,6 +126,7 @@ static struct option long_base_options[] = {
 	{"spooler-chdir", required_argument, 0, LONG_ARGS_SPOOLER_CHDIR},
 #endif
 	{"mule", optional_argument, 0, LONG_ARGS_MULE},
+	{"farm", required_argument, 0, LONG_ARGS_FARM},
 	{"disable-logging", no_argument, 0, 'L'},
 
 	{"pidfile", required_argument, 0, LONG_ARGS_PIDFILE},
@@ -2253,10 +2254,67 @@ skipzero:
 				exit(1);
 			}
 
+			uwsgi.mules[i].id = i+1;
+
 			snprintf(uwsgi.mules[i].name, 0xff, "uWSGI mule %d", i+1);
 		}
 	}
 
+	if (uwsgi.farms_cnt > 0) {
+		uwsgi.farms = (struct uwsgi_farm *) mmap(NULL, sizeof(struct uwsgi_farm) * uwsgi.farms_cnt, PROT_READ | PROT_WRITE, MAP_SHARED | MAP_ANON, -1, 0);
+		if (!uwsgi.farms) {
+			uwsgi_error("mmap()");
+                	exit(1);
+		}
+		memset(uwsgi.farms, 0, sizeof(struct uwsgi_farm) * uwsgi.farms_cnt);
+
+		struct uwsgi_string_list *farm_name = uwsgi.farms_list;
+		for(i=0;i<uwsgi.farms_cnt;i++) {
+
+			char *farm_value = uwsgi_str(farm_name->value);
+
+			char *mules_list = strchr(farm_value, ':');
+			if (!mules_list) {
+				uwsgi_log("invalid farm value (%s) must be in the form name:mule[,muleN].\n", farm_value);
+				exit(1);
+			}
+
+			mules_list[0] = 0;
+			mules_list++;
+
+			strncpy(uwsgi.farms[i].name, farm_value, 0xff);
+
+			// create the socket pipe
+        		if (socketpair(AF_UNIX, SOCK_STREAM, 0, uwsgi.farms[i].signal_pipe)) {
+                		uwsgi_error("socketpair()\n");
+        		}
+
+			if (socketpair(AF_UNIX, SOCK_DGRAM, 0, uwsgi.farms[i].queue_pipe)) {
+				uwsgi_error("socketpair()");
+				exit(1);
+			}
+
+
+			char *p = strtok(mules_list, ",");
+			while(p != NULL) {
+				struct uwsgi_mule *um = get_mule_by_id( atoi( p ) );
+				if (!um) {
+					uwsgi_log("invalid mule id: %s\n", p);
+					exit(1);
+				}
+
+				uwsgi_mule_farm_new(&uwsgi.farms[i].mules, um);
+
+				p = strtok(NULL, ",");
+			}
+			uwsgi_log("created farm %d name: %s mules:%s\n", i+1, uwsgi.farms[i].name, strchr(farm_name->value, ':')+1);
+
+			farm_name = farm_name->next;
+
+		}
+		
+	}
+
 	/*
 
 	   uwsgi.shared->hooks[0] = uwsgi_request_wsgi;
@@ -3377,6 +3435,11 @@ static int manage_base_opt(int i, char *optarg) {
 		uwsgi.mules_cnt++;
 		uwsgi_string_new_list(&uwsgi.mules_patches, optarg);
 		return 1;
+	case LONG_ARGS_FARM:
+		uwsgi.master_process = 1;
+		uwsgi.farms_cnt++;
+		uwsgi_string_new_list(&uwsgi.farms_list, optarg);
+		return 1;
 	case LONG_ARGS_SOCKET_PROTOCOL:
 		// TODO map each socket to a specific protocol
 		return 1;

uwsgi.h

@@ -532,6 +532,7 @@ struct uwsgi_opt {
 #define LONG_ARGS_PROCNAME_APPEND	17154
 #define LONG_ARGS_PROCNAME		17155
 #define LONG_ARGS_PROCNAME_MASTER	17156
+#define LONG_ARGS_FARM			17157
 
 
 #define UWSGI_OK	0
@@ -1294,12 +1295,15 @@ struct uwsgi_server {
 	/* the list of mules */
 	struct uwsgi_string_list *mules_patches;
 	struct uwsgi_mule *mules;
+	struct uwsgi_string_list *farms_list;
+	struct uwsgi_farm *farms;
 
 	pid_t mypid;
 	int mywid;
 
 	int muleid;
 	int mules_cnt;
+	int farms_cnt;
 
 	rlim_t max_fd;
 
@@ -1731,6 +1735,7 @@ struct uwsgi_worker {
 	char name[0xff];
 };
 
+
 struct uwsgi_mule {
 	int id;
 	pid_t pid;
@@ -1753,6 +1758,23 @@ struct uwsgi_mule {
 	char name[0xff];
 };
 
+struct uwsgi_mule_farm {
+	struct uwsgi_mule *mule;
+	struct uwsgi_mule_farm *next;
+};
+
+struct uwsgi_farm {
+	int id;
+	char name[0xff];
+
+	int signal_pipe[2];
+	int queue_pipe[2];
+
+	struct uwsgi_mule_farm *mules;
+
+};
+
+
 
 char *uwsgi_get_cwd(void);
 
@@ -1780,7 +1802,7 @@ void daemonize(char *);
 void logto(char *);
 
 void log_request(struct wsgi_request *);
-void get_memusage(void);
+void get_memusage(uint64_t *, uint64_t *);
 void harakiri(void);
 
 void stats(int);
@@ -2395,6 +2417,11 @@ void http_url_decode(char *, uint16_t *, char *);
 
 pid_t uwsgi_fork(char *);
 
+struct uwsgi_mule *get_mule_by_id(int);
+struct uwsgi_mule_farm *uwsgi_mule_farm_new(struct uwsgi_mule_farm **, struct uwsgi_mule *);
+
+int uwsgi_farm_has_mule(struct uwsgi_farm *, int);
+
 #ifdef UWSGI_CAP
 void uwsgi_build_cap(char *);
 #endif

uwsgidecorators.py

@@ -84,6 +84,30 @@ class rpc(object):
         uwsgi.register_rpc(self.name, f)
         return f
 
+class farm_loop(object):
+
+    def __init__(self, f, farm):
+        self.f = f
+        self.farm = farm
+    
+    def __call__(self):
+        if uwsgi.mule_id() == 0:
+            return
+        if not uwsgi.in_farm(self.farm):
+            return
+        while True:
+            message = uwsgi.farm_get_msg()
+            if message:
+                self.f(message)
+
+class farm(object):
+
+    def __init__(self, name=None, **kwargs):
+        self.name = name
+
+    def __call__(self, f):
+        postfork_chain.append(farm_loop(f, self.name))
+
 class signal(object):
 
     def __init__(self, num, **kwargs):