clearlinux/uwsgi
2
0
Fork 0
mirror of https://github.com/clearlinux/uwsgi.git synced 2026-09-04 12:41:38 +00:00
Code Issues Packages Projects Releases Wiki Activity

added body to spooler

Browse Source
This commit is contained in:
roberto@maverick64
2011-07-05 14:55:10 +02:00
parent 513c9cee3d
commit ec899a45d5
5 changed files with 63 additions and 19 deletions
Download Patch File Download Diff File Expand all files Collapse all files
decoratortest.py
+8
View File
@@ -61,6 +61,11 @@ def delayed_task(args):
# send a signal to all workers
uwsgi.signal(100)
@spool
def big_body_task(args):
print("*** managing a task with a body of %d bytes ***", len(args['body']))
print(args['body'].swapcase())
# run a task every hour
@cron(59, -1, -1, -1, -1)
def one_hour_passed(num):
@@ -110,3 +115,6 @@ 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")
uwsgi_source_file = open('uwsgi.c','r')
big_body_task.spool(priority=9,filename='uwsgi.c',body=uwsgi_source_file.read())
uwsgi_source_file.close()
plugins/python/python_plugin.c
+4 -1
View File
@@ -1023,7 +1023,7 @@ void uwsgi_python_add_item(char *key, uint16_t keylen, char *val, uint16_t valle
PyDict_SetItem(pydict, PyString_FromStringAndSize(key, keylen), PyString_FromStringAndSize(val, vallen));
}
int uwsgi_python_spooler(char *buf, uint16_t len) {
int uwsgi_python_spooler(char *buf, uint16_t len, char *body, size_t body_len) {
static int random_seed_reset = 0;
@@ -1057,6 +1057,9 @@ int uwsgi_python_spooler(char *buf, uint16_t len) {
}
pyargs = PyTuple_New(1);
if (body && body_len > 0) {
PyDict_SetItemString(spool_dict, "body", PyString_FromStringAndSize(body, body_len));
}
PyTuple_SetItem(pyargs, 0, spool_dict);
ret = python_call(spool_func, pyargs, 0);
plugins/python/uwsgi_pymodule.c
+22 -12
View File
@@ -2,8 +2,6 @@
#include "uwsgi_python.h"
char *spool_buffer = NULL;
extern struct uwsgi_server uwsgi;
extern struct uwsgi_python up;
@@ -1223,6 +1221,8 @@ PyObject *py_uwsgi_send_spool(PyObject * self, PyObject * args, PyObject *kw) {
char *priority = NULL;
long numprio = 0;
time_t at = 0;
char *body = NULL;
size_t body_len= 0;
spool_dict = PyTuple_GetItem(args, 0);
@@ -1267,12 +1267,23 @@ PyObject *py_uwsgi_send_spool(PyObject * self, PyObject * args, PyObject *kw) {
}
}
PyObject *pybody = PyDict_GetItemString(spool_dict, "body");
if (pybody) {
if (PyString_Check(pybody)) {
body = PyString_AsString(pybody);
body_len = PyString_Size(pybody);
PyDict_DelItemString(spool_dict, "body");
}
}
spool_vars = PyDict_Items(spool_dict);
if (!spool_vars) {
Py_INCREF(Py_None);
return Py_None;
}
char *spool_buffer = uwsgi_malloc(UMAX16);
cur_buf = spool_buffer;
for (i = 0; i < PyList_Size(spool_vars); i++) {
@@ -1287,7 +1298,7 @@ PyObject *py_uwsgi_send_spool(PyObject * self, PyObject * args, PyObject *kw) {
keysize = PyString_Size(key);
valsize = PyString_Size(val);
if (cur_buf + keysize + 2 + valsize + 2 <= spool_buffer + uwsgi.buffer_size) {
if (cur_buf + keysize + 2 + valsize + 2 <= spool_buffer + UMAX16) {
#ifdef __BIG_ENDIAN__
keysize = uwsgi_swap16(keysize);
@@ -1312,21 +1323,25 @@ PyObject *py_uwsgi_send_spool(PyObject * self, PyObject * args, PyObject *kw) {
}
else {
Py_DECREF(zero);
return PyErr_Format(PyExc_ValueError, "spooler packet cannot be more than %d bytes", uwsgi.buffer_size);
free(spool_buffer);
return PyErr_Format(PyExc_ValueError, "spooler packet cannot be more than %d bytes", UMAX16);
}
}
else {
Py_DECREF(zero);
free(spool_buffer);
return PyErr_Format(PyExc_ValueError, "spooler callable dictionary must contains only strings");
}
}
else {
free(spool_buffer);
Py_DECREF(zero);
Py_INCREF(Py_None);
return Py_None;
}
}
else {
free(spool_buffer);
Py_INCREF(Py_None);
return Py_None;
}
@@ -1335,10 +1350,12 @@ 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, at);
i = spool_request(spool_filename, uwsgi.workers[0].requests + 1, wsgi_req->async_id, spool_buffer, cur_buf - spool_buffer, priority, at, body, body_len);
if (priority) {
free(priority);
}
free(spool_buffer);
Py_DECREF(spool_vars);
@@ -2880,13 +2897,6 @@ void init_uwsgi_module_spooler(PyObject * current_uwsgi_module) {
exit(1);
}
spool_buffer = malloc(uwsgi.buffer_size);
if (!spool_buffer) {
uwsgi_error("malloc()");
exit(1);
}
for (uwsgi_function = uwsgi_spooler_methods; uwsgi_function->ml_name != NULL; uwsgi_function++) {
PyObject *func = PyCFunction_New(uwsgi_function, NULL);
PyDict_SetItemString(uwsgi_module_dict, uwsgi_function->ml_name, func);
spooler.c
+27 -4
View File
@@ -62,7 +62,7 @@ void destroy_spool(char *dir, char *file) {
}
int spool_request(char *filename, int rn, int core_id, char *buffer, int size, char *priority, time_t at) {
int spool_request(char *filename, int rn, int core_id, char *buffer, int size, char *priority, time_t at, char *body, size_t body_len) {
struct timeval tv;
int fd;
@@ -126,6 +126,12 @@ int spool_request(char *filename, int rn, int core_id, char *buffer, int size, c
goto clear;
}
if (body && body_len > 0) {
if ((size_t)write(fd, body, body_len) != body_len) {
goto clear;
}
}
if (at > 0) {
struct timeval tv[2];
tv[0].tv_sec = at;
@@ -139,7 +145,7 @@ int spool_request(char *filename, int rn, int core_id, char *buffer, int size, c
close(fd);
uwsgi_log("written %d bytes to spool file %s\n", size + 4, filename);
uwsgi_log("written %d bytes to spool file %s\n", size + body_len + 4, filename);
uwsgi_unlock(uwsgi.spooler_lock);
@@ -279,6 +285,8 @@ void spooler_manage_task(char *dir, char *task) {
char spool_buf[0xffff];
struct uwsgi_header uh;
char *body = NULL;
size_t body_len = 0;
int spool_fd;
@@ -348,13 +356,28 @@ void spooler_manage_task(char *dir, char *task) {
return;
}
// body available ?
if (sf_lstat.st_size > (uh.pktsize+4)) {
body_len = sf_lstat.st_size - (uh.pktsize+4);
body = uwsgi_malloc(body_len);
if ((size_t)read(spool_fd, body, body_len) != body_len) {
uwsgi_error("read()");
destroy_spool(dir, task);
close(spool_fd);
free(body);
return;
}
}
close(spool_fd);
for(i=0;i<0xff;i++) {
if (uwsgi.p[i]->spooler) {
time_t now = time(NULL);
ret = uwsgi.p[i]->spooler(spool_buf, uh.pktsize);
ret = uwsgi.p[i]->spooler(spool_buf, uh.pktsize, body, body_len);
if (body) {
free(body);
}
if (ret == 0) continue;
if (ret == -2) {
@@ -383,7 +406,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, 0);
i = spool_request(spool_filename, uwsgi.workers[0].requests + 1, wsgi_req->async_id, wsgi_req->buffer, wsgi_req->uh.pktsize, NULL, 0, NULL, 0);
wsgi_req->uh.modifier1 = 255;
wsgi_req->uh.pktsize = 0;
if (i > 0) {
uwsgi.h
+2 -2
View File
@@ -587,7 +587,7 @@ struct uwsgi_plugin {
char *(*decode_string) (void *);
int (*signal_handler) (uint8_t, void *);
int (*spooler) (char *, uint16_t);
int (*spooler) (char *, uint16_t, char *, size_t);
uint16_t(*rpc) (void *, uint8_t, char **, char *);
@@ -1512,7 +1512,7 @@ void snmp_init(void);
#endif
#ifdef UWSGI_SPOOLER
int spool_request(char *, int, int, char *, int, char *, time_t);
int spool_request(char *, int, int, char *, int, char *, time_t, char *, size_t);
void spooler(void);
pid_t spooler_start(void);
#endif