diff --git a/decoratortest.py b/decoratortest.py index b18c339b..f0f1e3b1 100644 --- a/decoratortest.py +++ b/decoratortest.py @@ -48,6 +48,12 @@ def an_infinite_task(args): print("infinite: %d %s" % (i, str(args))) time.sleep(1) + +# spool a task after 60 seconds +@spool +def delayed_task(args): + print("*** I am a delayed spool job. It is %s [%s]***" % (time.asctime(), str(args))) + # run a task every hour @cron(59, -1, -1, -1, -1) def one_hour_passed(num): @@ -90,6 +96,7 @@ def locked_func(): print("done with locked function on worker %d" % uwsgi.worker_id()) a_long_task.spool({'foo':'bar'}, hello='world') -an_infinite_task.spool(foo='bar') +an_infinite_task.spool(foo='bar', priority=3) +delayed_task.spool(foo2='bar2', at=time.time()+60) a_running_thread() a_running_thread_with_args("uWSGI") diff --git a/plugins/python/python_plugin.c b/plugins/python/python_plugin.c index f8765f06..2ec80621 100644 --- a/plugins/python/python_plugin.c +++ b/plugins/python/python_plugin.c @@ -1026,6 +1026,9 @@ void uwsgi_python_add_item(char *key, uint16_t keylen, char *val, uint16_t valle int uwsgi_python_spooler(char *buf, uint16_t len) { static int random_seed_reset = 0; + + UWSGI_GET_GIL; + PyObject *spool_dict = PyDict_New(); PyObject *spool_func, *pyargs, *ret; @@ -1036,17 +1039,20 @@ int uwsgi_python_spooler(char *buf, uint16_t len) { if (!up.embedded_dict) { // ignore + UWSGI_RELEASE_GIL; return 0; } spool_func = PyDict_GetItemString(up.embedded_dict, "spooler"); if (!spool_func) { // ignore + UWSGI_RELEASE_GIL; return 0; } if (uwsgi_hooked_parse(buf, len, uwsgi_python_add_item, spool_dict)) { // malformed packet, destroy it + UWSGI_RELEASE_GIL; return -2; } @@ -1057,16 +1063,21 @@ int uwsgi_python_spooler(char *buf, uint16_t len) { if (ret) { if (!PyInt_Check(ret)) { // error, retry + UWSGI_RELEASE_GIL; return -1; } - return PyInt_AsLong(ret); + int retval = (int) PyInt_AsLong(ret); + UWSGI_RELEASE_GIL; + return retval; + } if (PyErr_Occurred()) PyErr_Print(); // error, retry + UWSGI_RELEASE_GIL; return -1; } diff --git a/plugins/python/uwsgi_pymodule.c b/plugins/python/uwsgi_pymodule.c index b3f2613f..98741d5d 100644 --- a/plugins/python/uwsgi_pymodule.c +++ b/plugins/python/uwsgi_pymodule.c @@ -1162,6 +1162,7 @@ PyObject *py_uwsgi_send_spool(PyObject * self, PyObject * args, PyObject *kw) { struct wsgi_request *wsgi_req = current_wsgi_req(); char *priority = NULL; long numprio = 0; + time_t at = 0; spool_dict = PyTuple_GetItem(args, 0); @@ -1190,6 +1191,22 @@ PyObject *py_uwsgi_send_spool(PyObject * self, PyObject * args, PyObject *kw) { } } + PyObject *pyat = PyDict_GetItemString(spool_dict, "at"); + if (pyat) { + if (PyInt_Check(pyat)) { + at = (time_t) PyInt_AsLong(pyat); + PyDict_DelItemString(spool_dict, "at"); + } + else if (PyLong_Check(pyat)) { + at = (time_t) PyLong_AsLong(pyat); + PyDict_DelItemString(spool_dict, "at"); + } + else if (PyFloat_Check(pyat)) { + at = (time_t) PyFloat_AsDouble(pyat); + PyDict_DelItemString(spool_dict, "at"); + } + } + spool_vars = PyDict_Items(spool_dict); if (!spool_vars) { Py_INCREF(Py_None); @@ -1258,7 +1275,7 @@ PyObject *py_uwsgi_send_spool(PyObject * self, PyObject * args, PyObject *kw) { if (numprio) { priority = uwsgi_num2str(numprio); } - i = spool_request(spool_filename, uwsgi.workers[0].requests + 1, wsgi_req->async_id, spool_buffer, cur_buf - spool_buffer, priority); + i = spool_request(spool_filename, uwsgi.workers[0].requests + 1, wsgi_req->async_id, spool_buffer, cur_buf - spool_buffer, priority, at); if (priority) { free(priority); } diff --git a/spooler.c b/spooler.c index f71cce1b..90127854 100644 --- a/spooler.c +++ b/spooler.c @@ -61,7 +61,7 @@ void destroy_spool(char *dir, char *file) { } -int spool_request(char *filename, int rn, int core_id, char *buffer, int size, char *priority) { +int spool_request(char *filename, int rn, int core_id, char *buffer, int size, char *priority, time_t at) { struct timeval tv; int fd; @@ -125,6 +125,17 @@ int spool_request(char *filename, int rn, int core_id, char *buffer, int size, c goto clear; } + if (at > 0) { + struct timeval tv[2]; + tv[0].tv_sec = at; + tv[0].tv_usec = 0; + tv[1].tv_sec = at; + tv[1].tv_usec = 0; + if (futimes(fd, tv)) { + uwsgi_error("futimes()"); + } + } + close(fd); uwsgi_log("written %d bytes to spool file %s\n", size + 4, filename); @@ -276,6 +287,11 @@ void spooler_manage_task(char *dir, char *task) { return; } + // a spool request for the future + if (sf_lstat.st_mtime > time(NULL)) { + return; + } + #ifdef __linux__ if (S_ISDIR(sf_lstat.st_mode) && uwsgi.spooler_ordered) { if (chdir(task)) { @@ -365,7 +381,7 @@ int uwsgi_request_spooler(struct wsgi_request *wsgi_req) { } uwsgi_log("managing spool request...\n"); - i = spool_request(spool_filename, uwsgi.workers[0].requests + 1, wsgi_req->async_id, wsgi_req->buffer, wsgi_req->uh.pktsize, NULL); + i = spool_request(spool_filename, uwsgi.workers[0].requests + 1, wsgi_req->async_id, wsgi_req->buffer, wsgi_req->uh.pktsize, NULL, 0); wsgi_req->uh.modifier1 = 255; wsgi_req->uh.pktsize = 0; if (i > 0) { diff --git a/uwsgi.c b/uwsgi.c index cf44b0e6..33bb83ce 100644 --- a/uwsgi.c +++ b/uwsgi.c @@ -2037,6 +2037,14 @@ int uwsgi_start(void *v_argv) { } } + // master fixup + for (i = 0; i < 0xFF; i++) { + if (uwsgi.p[i]->master_fixup) { + uwsgi.p[i]->master_fixup(0); + } + } + + #ifdef UWSGI_SPOOLER if (uwsgi.spool_dir != NULL && uwsgi.sockets) { @@ -2048,13 +2056,7 @@ int uwsgi_start(void *v_argv) { routing_setup(); #endif - // master fixup - for (i = 0; i < 0xFF; i++) { - if (uwsgi.p[i]->master_fixup) { - uwsgi.p[i]->master_fixup(0); - } - } if (!uwsgi.master_process) { diff --git a/uwsgi.h b/uwsgi.h index 582c6051..3b0185f5 100644 --- a/uwsgi.h +++ b/uwsgi.h @@ -1504,7 +1504,7 @@ void snmp_init(void); #endif #ifdef UWSGI_SPOOLER -int spool_request(char *, int, int, char *, int, char *); +int spool_request(char *, int, int, char *, int, char *, time_t); void spooler(void); pid_t spooler_start(void); #endif