added 'at' variable to the spooler

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")

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;
 }
 

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);
 	}

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) {

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) {

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