multiprocess spooler

lock.c

@@ -437,3 +437,35 @@ void uwsgi_setup_locking() {
 	uwsgi.rwlock_size = UWSGI_RWLOCK_SIZE;
 }
 
+
+int uwsgi_fcntl_lock(int fd) {
+	struct flock fl;
+	fl.l_type = F_WRLCK;
+	fl.l_whence = SEEK_CUR;
+	fl.l_start = 0;
+	fl.l_len = 0;
+	fl.l_pid = 0;
+
+	int ret = fcntl(fd, F_SETLKW, &fl);
+	if (ret < 0)
+		uwsgi_error("fcntl()");
+
+	return ret;
+}
+
+int uwsgi_fcntl_is_locked(int fd) {
+
+	struct flock fl;
+        fl.l_type = F_WRLCK;
+        fl.l_whence = SEEK_CUR;
+        fl.l_start = 0;
+        fl.l_len = 0;
+        fl.l_pid = 0;
+
+        if (fcntl(fd, F_SETLK, &fl)) {
+		return 1;
+	}
+
+	return 0;
+	
+}

spooler.c

@@ -3,15 +3,66 @@
 
 extern struct uwsgi_server uwsgi;
 
-static void spooler_readdir(char *);
+static void spooler_readdir(struct uwsgi_spooler *, char *dir);
 #ifdef __linux__
-static void spooler_scandir(char *);
+static void spooler_scandir(struct uwsgi_spooler *, char *dir);
 #endif
-void spooler_manage_task(char *, char *);
+static void spooler_manage_task(struct uwsgi_spooler *, char *, char *);
 
 // fake function to allow waking the spooler
 void spooler_wakeup() {}
 
+void uwsgi_opt_add_spooler(char *opt, char *directory, void *none) {
+
+	int i;
+
+        if (access(directory, R_OK | W_OK | X_OK)) {
+                uwsgi_error("[spooler directory] access()");
+                exit(1);
+        }
+
+	if (uwsgi.spooler_numproc > 0) {
+		for(i=0;i<uwsgi.spooler_numproc;i++) {
+			uwsgi_new_spooler(directory);
+		}
+	}
+	else {
+        	uwsgi_new_spooler(directory);
+	}
+
+}
+
+
+struct uwsgi_spooler *uwsgi_new_spooler(char *dir) {
+
+        struct uwsgi_spooler *uspool = uwsgi.spoolers;
+
+        if (!uspool) {
+                uwsgi.spoolers = uwsgi_malloc_shared(sizeof(struct uwsgi_spooler));
+                uspool = uwsgi.spoolers;
+        }
+        else {
+                while(uspool) {
+                        if (uspool->next == NULL) {
+                                uspool->next = uwsgi_malloc_shared(sizeof(struct uwsgi_spooler));
+                                uspool = uspool->next;
+                                break;
+                        }
+                        uspool = uspool->next;
+                }
+        }
+
+        if (!realpath(dir, uspool->dir)) {
+                uwsgi_error("[spooler] realpath()");
+                exit(1);
+        }
+
+        uspool->next = NULL;
+
+        return uspool;
+}
+
+
 struct uwsgi_spooler *uwsgi_get_spooler_by_name(char *name) {
 
 	struct uwsgi_spooler *uspool = uwsgi.spoolers;
@@ -103,6 +154,7 @@ int spool_request(struct uwsgi_spooler *uspool, char *filename, int rn, int core
 		uspool = uwsgi.spoolers;
 	}
 
+	// this lock is for threads, the pid value in filename will avoid multiprocess races
 	uwsgi_lock(uspool->lock);
 
 	gettimeofday(&tv, NULL);
@@ -134,13 +186,10 @@ int spool_request(struct uwsgi_spooler *uspool, char *filename, int rn, int core
 		return 0;
 	}
 
-#ifdef __sun__
-	if (lockf(fd, F_LOCK, 0)) {
-		uwsgi_error("lockf()");
-#else
-	if (flock(fd, LOCK_EX)) {
-		uwsgi_error("flock()");
-#endif
+	// now lock the file, it will no be runnable, until the lock is not removed
+	// a race could come if the spooler take the file before fcntl is called
+        // in such case the spooler will detect a zeroed file and will retry later
+	if (uwsgi_fcntl_lock(fd)) {
 		close(fd);
 		uwsgi_unlock(uspool->lock);
 		return 0;
@@ -182,15 +231,27 @@ int spool_request(struct uwsgi_spooler *uspool, char *filename, int rn, int core
 		}
 	}
 
+	// here the file will be unlocked too
 	close(fd);
 
 	uwsgi_log("[spooler] written %d bytes to file %s\n", size + body_len + 4, filename);
 	
+	// and here waiting threads can continue
 	uwsgi_unlock(uspool->lock);
 
-/*	wake up the spooler ... (HACKY) */
-	if (uspool->pid > 0 ) {
-		(void) kill(uspool->pid, SIGUSR1);
+/*	wake up the spoolers attached to the specified dir ... (HACKY) 
+	no need to fear races, as USR1 is harmless an all of the uWSGI processes...
+	it could be a problem if a new process takes the old pid, but modern systems should avoid that
+*/
+
+	struct uwsgi_spooler *spoolers = uwsgi.spoolers;
+	while(spoolers) {
+		if (!strcmp(spoolers->dir, uspool->dir)) {
+			if (spoolers->pid > 0 && spoolers->running == 0) {
+				(void) kill(spoolers->pid, SIGUSR1);
+			}
+		}
+		spoolers = spoolers->next;
 	}
 
 	return 1;
@@ -202,6 +263,7 @@ int spool_request(struct uwsgi_spooler *uspool, char *filename, int rn, int core
 	if (unlink(filename)) {
 		uwsgi_error("unlink()");
 	}
+	// unlock the file too
 	close(fd);
 	return 0;
 }
@@ -254,13 +316,13 @@ void spooler(struct uwsgi_spooler *uspool) {
 
 		if (uwsgi.spooler_ordered) {
 #ifdef __linux__
-			spooler_scandir(uspool->dir);
+			spooler_scandir(uspool, NULL);
 #else
-			spooler_readdir(uspool->dir);
+			spooler_readdir(uspool, NULL);
 #endif
 		}
 		else {
-			spooler_readdir(uspool->dir);
+			spooler_readdir(uspool, NULL);
 		}
 
 		if (event_queue_wait(spooler_event_queue, uwsgi.shared->spooler_frequency, &interesting_fd) > 0) {
@@ -287,11 +349,13 @@ void spooler(struct uwsgi_spooler *uspool) {
 }
 
 #ifdef __linux__
-static void spooler_scandir(char *dir) {
+static void spooler_scandir(struct uwsgi_spooler *uspool, char *dir) {
 
 	struct dirent **tasklist;
 	int n;
 
+	if (!dir) dir = uspool->dir;
+
 	n = scandir(dir, &tasklist, 0, versionsort);
 	if (n < 0) {
 		uwsgi_error("scandir()");
@@ -299,7 +363,7 @@ static void spooler_scandir(char *dir) {
 	}
 	
 	while(n--) {
-		spooler_manage_task(dir, tasklist[n]->d_name);
+		spooler_manage_task(uspool, dir, tasklist[n]->d_name);
 		free(tasklist[n]);	
 	}
 
@@ -308,15 +372,17 @@ static void spooler_scandir(char *dir) {
 #endif
 
 
-static void spooler_readdir(char *dir) {
+static void spooler_readdir(struct uwsgi_spooler *uspool, char *dir) {
 
 	DIR *sdir;
 	struct dirent *dp;
 
+	if (!dir) dir = uspool->dir;
+
 	sdir = opendir(dir);
 	if (sdir) {
 		while ((dp = readdir(sdir)) != NULL) {
-			spooler_manage_task(dir, dp->d_name);
+			spooler_manage_task(uspool, dir, dp->d_name);
 		}
 		closedir(sdir);
 	}
@@ -325,7 +391,7 @@ static void spooler_readdir(char *dir) {
 	}
 }
 
-void spooler_manage_task(char *dir, char *task) {
+void spooler_manage_task(struct uwsgi_spooler *uspool, char *dir, char *task) {
 
 	int i, ret;
 
@@ -336,6 +402,8 @@ void spooler_manage_task(char *dir, char *task) {
 
 	int spool_fd;
 
+	if (!dir) dir = uspool->dir;
+
 	if (!strncmp("uwsgi_spoolfile_on_", task, 19) || (uwsgi.spooler_ordered && is_a_number(task))) {
 		struct stat sf_lstat;
 		if (lstat(task, &sf_lstat)) {
@@ -354,7 +422,7 @@ void spooler_manage_task(char *dir, char *task) {
 				return;
 			}
 			char *prio_path = realpath(".", NULL);
-			spooler_scandir(prio_path);
+			spooler_scandir(uspool, prio_path);
 			free(prio_path);
 			if (chdir(dir)) {
 				uwsgi_error("chdir()");
@@ -366,31 +434,22 @@ void spooler_manage_task(char *dir, char *task) {
 			return;
 		}
 		if (!access(task, R_OK | W_OK)) {
-			uwsgi_log("[spooler] managing request %s ...\n", task);
 
-#ifdef __sun__
-			// lockf needs write permission
 			spool_fd = open(task, O_RDWR);
-#else
-			spool_fd = open(task, O_RDONLY);
-#endif
+
 			if (spool_fd < 0) {
 				uwsgi_error_open(task);
 				return;
 			}
 
-#ifdef __sun__
-			if (lockf(spool_fd, F_LOCK, 0)) {
-				uwsgi_error("lockf()");
-#else
-			if (flock(spool_fd, LOCK_EX)) {
-				uwsgi_error("flock()");
-#endif
+			// check if the file is locked by anther process
+			if (uwsgi_fcntl_is_locked(spool_fd)) {
 				close(spool_fd);
 				return;
 			}
 
 			if (read(spool_fd, &uh, 4) != 4) {
+				// it could be here for broken file or just opened one
 				uwsgi_error("read()");
 				close(spool_fd);
 				return;
@@ -420,8 +479,13 @@ void spooler_manage_task(char *dir, char *task) {
 				}
 			}
 
-			close(spool_fd);
+			// not the task is running and should not be waken
+			uspool->running = 1;
 
+			uwsgi_log("[spooler %s pid: %d] managing request %s ...\n", uspool->dir, (int) uwsgi.mypid, task);
+
+
+			// chdir before running the task (if requested)
 			if (uwsgi.spooler_chdir) {
 				if (chdir(uwsgi.spooler_chdir)) {
 					uwsgi_error("chdir()");
@@ -442,7 +506,7 @@ void spooler_manage_task(char *dir, char *task) {
 					if (ret == 0) continue;
 					callable_found = 1;
 					if (ret == -2) {
-						uwsgi_log("[spooler] done with task %s after %d seconds\n", task, time(NULL)-now);
+						uwsgi_log("[spooler %s pid: %d] done with task %s after %d seconds\n", uspool->dir, (int) uwsgi.mypid, task, time(NULL)-now);
 						destroy_spool(dir, task);	
 					}
 					// re-spool it
@@ -453,6 +517,10 @@ void spooler_manage_task(char *dir, char *task) {
 			if (body)
 				free(body);
 
+			// here we free and unlock the task
+			close(spool_fd);
+			uspool->running = 0;
+
 			if (!callable_found) {
 				uwsgi_log("unable to find the spooler function, have you loaded it into the spooler process ?\n");
 			}

tests/spoolme.py

@@ -0,0 +1,19 @@
+import uwsgi
+import time
+
+def slow_task(args):
+    time.sleep(10)
+    return uwsgi.SPOOL_OK
+
+uwsgi.spooler = slow_task
+
+
+
+def application(env, start_response):
+
+    name = uwsgi.spool({'Hello':'World', 'I am a':'long running task'})
+    print("spooled as %s" % name)
+
+    start_response('200 Ok', [('Content-Type','text/plain'),('uWSGI-Status', 'spooled')])
+
+    return "task spooled"

utils.c

@@ -3063,40 +3063,6 @@ void *uwsgi_malloc_shared(size_t size) {
 }
 
 
-#ifdef UWSGI_SPOOLER
-struct uwsgi_spooler *uwsgi_new_spooler(char *dir) {
-
-        struct uwsgi_spooler *uspool = uwsgi.spoolers;
-
-        if (!uspool) {
-                uwsgi.spoolers = uwsgi_malloc_shared(sizeof(struct uwsgi_spooler));
-                uspool = uwsgi.spoolers;
-        }
-        else {
-                while(uspool) {
-			if (uspool->next == NULL) {
-				uspool->next = uwsgi_malloc_shared(sizeof(struct uwsgi_spooler));
-				uspool = uspool->next;
-				break;
-			}
-			uspool = uspool->next;
-                }
-        }
-
-	if (!realpath(dir, uspool->dir)) {
-		uwsgi_error("[spooler] realpath()");
-		exit(1);
-	}
-
-        uspool->lock = uwsgi_lock_init( uwsgi_concat2("spooler on ", uspool->dir) );
-
-        uspool->next = NULL;
-
-        return uspool;
-}
-#endif
-
-
 struct uwsgi_string_list *uwsgi_string_new_list(struct uwsgi_string_list **list, char *value) {
 
 	struct uwsgi_string_list *uwsgi_string = *list, *old_uwsgi_string;

uwsgi.c

@@ -161,6 +161,7 @@ static struct uwsgi_option uwsgi_base_options[] = {
 	{"spooler", required_argument, 'Q', "run a spooler on the specified directory", uwsgi_opt_add_spooler, NULL, UWSGI_OPT_MASTER},
 	{"spooler-ordered", no_argument, 0, "try to order the execution of spooler tasks", uwsgi_opt_true, &uwsgi.spooler_ordered,0},
 	{"spooler-chdir", required_argument, 0, "chdir() to specified directory before each spooler task", uwsgi_opt_set_str, &uwsgi.spooler_chdir,0},
+	{"spooler-processes", required_argument, 0, "set the number of processes for spoolers", uwsgi_opt_set_int, &uwsgi.spooler_numproc, 0},
 #endif
 	{"mule", optional_argument, 0, "add a mule", uwsgi_opt_add_mule, NULL, UWSGI_OPT_MASTER},
 	{"mules", required_argument, 0, "add the specified number of mules", uwsgi_opt_add_mules, NULL, UWSGI_OPT_MASTER},
@@ -2639,6 +2640,7 @@ skipzero:
 		struct uwsgi_spooler *uspool = uwsgi.spoolers;
 		while(uspool) {
 			create_signal_pipe(uspool->signal_pipe);
+			uspool->lock = uwsgi_lock_init( uwsgi_concat2("spooler on ", uspool->dir) );
 			uspool->pid = spooler_start(uspool);
 			uspool = uspool->next;
 		}
@@ -3475,18 +3477,6 @@ void uwsgi_opt_print(char *opt, char *value, void *str) {
 	fprintf(stdout, "%s\n", value);
 }
 
-#ifdef UWSGI_SPOOLER
-void uwsgi_opt_add_spooler(char *opt, char *directory, void *none) {
-
-	if (access(directory, R_OK | W_OK | X_OK)) {
-        	uwsgi_error("[spooler directory] access()");
-                exit(1);
-	}
-	uwsgi_new_spooler(directory);
-	
-}
-#endif
-
 void uwsgi_opt_set_uid(char *opt, char *value, void *none) {
 
 	uwsgi.uid = atoi(value);

uwsgi.h

@@ -703,6 +703,8 @@ struct uwsgi_spooler {
 	struct uwsgi_lock_item *lock;
 	time_t harakiri;
 
+	int running;
+
 	int signal_pipe[2];
 
 	struct uwsgi_spooler *next;
@@ -1303,6 +1305,7 @@ struct uwsgi_server {
 
 #ifdef UWSGI_SPOOLER
 	struct uwsgi_spooler *spoolers;
+	int spooler_numproc;
 	struct uwsgi_spooler *i_am_a_spooler;
 	char *spooler_chdir;
 	int spooler_ordered;
@@ -2692,9 +2695,12 @@ void uwsgi_flush_logs(void);
 void uwsgi_register_cheaper_algo(char *, int(*) (void));
 
 void uwsgi_setup_locking(void);
+int uwsgi_fcntl_lock(int);
+int uwsgi_fcntl_is_locked(int);
 
 void uwsgi_emulate_cow_for_apps(int);
 
+
 #ifdef UWSGI_AS_SHARED_LIBRARY
 int uwsgi_init(int, char **, char **);
 #endif