From 84374ba4f99e74d6c8bfdc67a257e7dc64e27b52 Mon Sep 17 00:00:00 2001 From: Roberto De Ioris Date: Wed, 6 May 2015 09:11:35 +0200 Subject: [PATCH] implementation of #904 --- core/master.c | 7 +++++ core/master_checks.c | 12 ++++++-- core/spooler.c | 66 ++++++++++++++++++++++++++++++++++++++------ core/uwsgi.c | 3 ++ t/spooler/cheap.py | 21 ++++++++++++++ uwsgi.h | 4 +++ 6 files changed, 101 insertions(+), 12 deletions(-) create mode 100644 t/spooler/cheap.py diff --git a/core/master.c b/core/master.c index dc967aad..af6ca2c6 100644 --- a/core/master.c +++ b/core/master.c @@ -669,6 +669,13 @@ int master_loop(char **argv, char **environ) { return 0; } + // spooler cheap management + if (uwsgi.spooler_cheap) { + if ((uwsgi.master_cycles % uwsgi.spooler_frequency) == 0) { + uwsgi_spooler_cheap_check(); + } + } + // check if someone is dead diedpid = waitpid(WAIT_ANY, &waitpid_status, WNOHANG); diff --git a/core/master_checks.c b/core/master_checks.c index 65c50520..8797a948 100644 --- a/core/master_checks.c +++ b/core/master_checks.c @@ -302,9 +302,15 @@ int uwsgi_master_check_spoolers_death(int diedpid) { struct uwsgi_spooler *uspool = uwsgi.spoolers; while (uspool) { if (uspool->pid > 0 && diedpid == uspool->pid) { - uwsgi_log("OOOPS the spooler is no more...trying respawn...\n"); - uspool->respawned++; - uspool->pid = spooler_start(uspool); + if (uwsgi.spooler_cheap) { + uwsgi_log_verbose("spooler %s ended\n", uspool->dir); + uspool->pid = 0; + } + else { + uwsgi_log("OOOPS the spooler is no more...trying respawn...\n"); + uspool->respawned++; + uspool->pid = spooler_start(uspool); + } return -1; } uspool = uspool->next; diff --git a/core/spooler.c b/core/spooler.c index 15f29243..b54d6750 100755 --- a/core/spooler.c +++ b/core/spooler.c @@ -431,8 +431,9 @@ void spooler(struct uwsgi_spooler *uspool) { // reset the tasks counter uspool->tasks = 0; - for (;;) { + time_t last_task_managed = 0; + for (;;) { if (chdir(uspool->dir)) { uwsgi_error("chdir()"); @@ -446,6 +447,16 @@ void spooler(struct uwsgi_spooler *uspool) { spooler_readdir(uspool, NULL); } + // here we check (if in cheap mode), if the spooler has done its job + if (uwsgi.spooler_cheap) { + if (last_task_managed == uspool->last_task_managed) { + uwsgi_log_verbose("cheaping spooler %s ...\n", uspool->dir); + exit(0); + } + last_task_managed = uspool->last_task_managed; + } + + int timeout = uwsgi.shared->spooler_frequency ? uwsgi.shared->spooler_frequency : uwsgi.spooler_frequency; if (wakeup > 0) { timeout = 0; @@ -508,7 +519,7 @@ static void spooler_readdir(struct uwsgi_spooler *uspool, char *dir) { closedir(sdir); } else { - uwsgi_error("opendir()"); + uwsgi_error("spooler_readdir()/opendir()"); } } @@ -528,6 +539,7 @@ void spooler_manage_task(struct uwsgi_spooler *uspool, char *dir, char *task) { if (!strncmp("uwsgi_spoolfile_on_", task, 19) || (uwsgi.spooler_ordered && is_a_number(task))) { struct stat sf_lstat; + if (lstat(task, &sf_lstat)) { return; } @@ -539,7 +551,7 @@ void spooler_manage_task(struct uwsgi_spooler *uspool, char *dir, char *task) { if (S_ISDIR(sf_lstat.st_mode) && uwsgi.spooler_ordered) { if (chdir(task)) { - uwsgi_error("chdir()"); + uwsgi_error("spooler_manage_task()/chdir()"); return; } #ifdef __UCLIBC__ @@ -551,7 +563,7 @@ void spooler_manage_task(struct uwsgi_spooler *uspool, char *dir, char *task) { spooler_scandir(uspool, prio_path); free(prio_path); if (chdir(dir)) { - uwsgi_error("chdir()"); + uwsgi_error("spooler_manage_task()/chdir()"); } return; } @@ -560,6 +572,7 @@ void spooler_manage_task(struct uwsgi_spooler *uspool, char *dir, char *task) { } if (!access(task, R_OK | W_OK)) { + spool_fd = open(task, O_RDWR); if (spool_fd < 0) { @@ -580,12 +593,13 @@ void spooler_manage_task(struct uwsgi_spooler *uspool, char *dir, char *task) { return; } + ssize_t rlen = uwsgi_protected_read(spool_fd, &uh, 4); if (rlen != 4) { // it could be here for broken file or just opened one if (rlen < 0) - uwsgi_error("read()"); + uwsgi_error("spooler_manage_task()/read()"); uwsgi_protected_close(spool_fd); return; } @@ -595,7 +609,7 @@ void spooler_manage_task(struct uwsgi_spooler *uspool, char *dir, char *task) { #endif if (uwsgi_protected_read(spool_fd, spool_buf, uh._pktsize) != uh._pktsize) { - uwsgi_error("read()"); + uwsgi_error("spooler_manage_task()/read()"); destroy_spool(dir, task); uwsgi_protected_close(spool_fd); return; @@ -606,7 +620,7 @@ void spooler_manage_task(struct uwsgi_spooler *uspool, char *dir, char *task) { body_len = sf_lstat.st_size - (uh._pktsize + 4); body = uwsgi_malloc(body_len); if ((size_t) uwsgi_protected_read(spool_fd, body, body_len) != body_len) { - uwsgi_error("read()"); + uwsgi_error("spooler_manage_task()/read()"); destroy_spool(dir, task); uwsgi_protected_close(spool_fd); free(body); @@ -614,8 +628,10 @@ void spooler_manage_task(struct uwsgi_spooler *uspool, char *dir, char *task) { } } - // now the task is running and should not be waken up + // now the task is running and should not be woken up uspool->running = 1; + // this is used in cheap mode for making decision about who must die + uspool->last_task_managed = uwsgi_now(); if (!uwsgi.spooler_quiet) uwsgi_log("[spooler %s pid: %d] managing request %s ...\n", uspool->dir, (int) uwsgi.mypid, task); @@ -624,7 +640,7 @@ void spooler_manage_task(struct uwsgi_spooler *uspool, char *dir, char *task) { // chdir before running the task (if requested) if (uwsgi.spooler_chdir) { if (chdir(uwsgi.spooler_chdir)) { - uwsgi_error("chdir()"); + uwsgi_error("spooler_manage_task()/chdir()"); } } @@ -682,3 +698,35 @@ void spooler_manage_task(struct uwsgi_spooler *uspool, char *dir, char *task) { } } } + +// this function checks which spooler should be spawned +void uwsgi_spooler_cheap_check() { + struct uwsgi_spooler *uspool = uwsgi.spoolers; + char *last_managed = NULL; + while(uspool) { + // skip already active spoolers + if (uspool->pid > 0) goto next; + // spooler dir names (in multiprocess mode, are ordered, so we can use + // this trick for avoiding spawning multiple processes for the same dir + // in the same cycle + if (!last_managed || strcmp(last_managed, uspool->dir)) { + // unfortunately, reusing readdir/scandir of the spooler is too dungeorus + // as the code is run in the master, let's do a simpler check + struct dirent *dp; + DIR *sdir = opendir(uspool->dir); + if (!sdir) goto next; + while ((dp = readdir(sdir)) != NULL) { + if (strncmp("uwsgi_spoolfile_on_", dp->d_name, 19)) continue; + // a uwsgi_spoolfile_on_* file has been found... + uspool->respawned++; + // spawn a new spooler + uspool->pid = spooler_start(uspool); + last_managed = uspool->dir; + break; + } + closedir(sdir); + } +next: + uspool = uspool->next; + } +} diff --git a/core/uwsgi.c b/core/uwsgi.c index d56bf7ff..74864183 100755 --- a/core/uwsgi.c +++ b/core/uwsgi.c @@ -328,6 +328,7 @@ static struct uwsgi_option uwsgi_base_options[] = { {"spooler-harakiri", required_argument, 0, "set harakiri timeout for spooler tasks", uwsgi_opt_set_int, &uwsgi.harakiri_options.spoolers, 0}, {"spooler-frequency", required_argument, 0, "set spooler frequency", uwsgi_opt_set_int, &uwsgi.spooler_frequency, 0}, {"spooler-freq", required_argument, 0, "set spooler frequency", uwsgi_opt_set_int, &uwsgi.spooler_frequency, 0}, + {"spooler-cheap", no_argument, 0, "set spooler cheap mode", uwsgi_opt_true, &uwsgi.spooler_cheap, 0}, {"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}, @@ -3242,6 +3243,8 @@ next: while (uspool) { if (uspool->mode == UWSGI_SPOOLER_EXTERNAL) goto next2; + // skip spooler start on cheap mode + if (uwsgi.spooler_cheap) goto next2; uspool->pid = spooler_start(uspool); next2: uspool = uspool->next; diff --git a/t/spooler/cheap.py b/t/spooler/cheap.py new file mode 100644 index 00000000..1ea585ea --- /dev/null +++ b/t/spooler/cheap.py @@ -0,0 +1,21 @@ +# uwsgi --spooler spool1 --spooler spool2 --spooler-cheap --spooler-frequency 5 --spooler-processes 4 --mule --shared-py-import=t/spooler/cheap.py --stats :5000 +from uwsgidecorators import * +import time +import random +import os + +def fake(args): + time.sleep(6) + return uwsgi.SPOOL_OK + +uwsgi.spooler = fake + +base = os.getcwd() +spoolers = [base + '/spool1', base + '/spool2'] + +@mule(1) +def spooler_enqueuer(): + while True: + print("enqueuing task...") + uwsgi.spool({'one':'two', 'spooler': random.choice(spoolers)}) + time.sleep(random.randrange(1, 15)) diff --git a/uwsgi.h b/uwsgi.h index fe1e8cd4..8d1bb43c 100755 --- a/uwsgi.h +++ b/uwsgi.h @@ -1146,6 +1146,8 @@ struct uwsgi_spooler { int signal_pipe[2]; struct uwsgi_spooler *next; + + time_t last_task_managed; }; #ifdef UWSGI_ROUTING @@ -2798,6 +2800,7 @@ struct uwsgi_server { char *safe_pidfile2; int die_on_no_workers; + int spooler_cheap; }; struct uwsgi_rpc { @@ -4964,6 +4967,7 @@ int vassal_attr_get_multi(struct uwsgi_instance *, char *, int (*)(struct uwsgi_ int uwsgi_zeus_spawn_instance(struct uwsgi_instance *); time_t uwsgi_parse_http_date(char *, uint16_t); +void uwsgi_spooler_cheap_check(void); #ifdef __cplusplus } #endif