From 312563437cd3fc589590a192bf153b33eb324d40 Mon Sep 17 00:00:00 2001 From: "roberto@precise64" Date: Thu, 1 Mar 2012 14:54:10 +0100 Subject: [PATCH] multiprocess spooler --- lock.c | 32 +++++++++++ spooler.c | 142 +++++++++++++++++++++++++++++++++++------------ tests/spoolme.py | 19 +++++++ utils.c | 34 ------------ uwsgi.c | 14 +---- uwsgi.h | 6 ++ 6 files changed, 164 insertions(+), 83 deletions(-) create mode 100644 tests/spoolme.py diff --git a/lock.c b/lock.c index eafef65f..9bcfb6ec 100644 --- a/lock.c +++ b/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; + +} diff --git a/spooler.c b/spooler.c index fe50d81f..a32986e2 100644 --- a/spooler.c +++ b/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;inext == 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"); } diff --git a/tests/spoolme.py b/tests/spoolme.py new file mode 100644 index 00000000..95ea8c32 --- /dev/null +++ b/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" diff --git a/utils.c b/utils.c index 261391b0..2b74eedb 100644 --- a/utils.c +++ b/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; diff --git a/uwsgi.c b/uwsgi.c index 25437358..dc79f4a4 100644 --- a/uwsgi.c +++ b/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); diff --git a/uwsgi.h b/uwsgi.h index e523cbff..2bf07e80 100644 --- a/uwsgi.h +++ b/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