From dcf4217b9d495a4465af6268b18a70dc605d91e8 Mon Sep 17 00:00:00 2001 From: "roberto@debian32" Date: Wed, 26 Oct 2011 10:13:20 +0200 Subject: [PATCH] greatly improved uwsgi.mule_msg() --- mule.c | 12 +++- plugins/python/python_plugin.c | 28 ++++++++ plugins/python/uwsgi_pymodule.c | 123 +++++++++++++++++++++++--------- uwsgi.h | 1 + 4 files changed, 131 insertions(+), 33 deletions(-) diff --git a/mule.c b/mule.c index 55b5b17b..088eb041 100644 --- a/mule.c +++ b/mule.c @@ -165,7 +165,17 @@ void uwsgi_mule_handler() { uwsgi_error("read()"); } else { - uwsgi_log("*** mule %d received a %d bytes message ***\n", uwsgi.muleid, len); + int i,found = 0; + for(i=0;i<0xff;i++) { + if (uwsgi.p[i]->mule_msg) { + if (uwsgi.p[i]->mule_msg(message, len)) { + found = 1; + break; + } + } + } + if (!found) + uwsgi_log("*** mule %d received a %d bytes message ***\n", uwsgi.muleid, len); } } } diff --git a/plugins/python/python_plugin.c b/plugins/python/python_plugin.c index 2ed4623e..d142cf3f 100644 --- a/plugins/python/python_plugin.c +++ b/plugins/python/python_plugin.c @@ -1411,6 +1411,33 @@ int uwsgi_python_mule(char *opt) { } +int uwsgi_python_mule_msg(char *message, size_t len) { + + UWSGI_GET_GIL; + + PyObject *mule_msg_hook = PyDict_GetItemString(up.embedded_dict, "mule_msg_hook"); + if (!mule_msg_hook) { + // ignore + UWSGI_RELEASE_GIL; + return 0; + } + + PyObject *pyargs = PyTuple_New(1); + PyTuple_SetItem(pyargs, 0, PyString_FromStringAndSize(message, len)); + + PyObject *ret = python_call(mule_msg_hook, pyargs, 0, NULL); + Py_DECREF(pyargs); + if (ret) { + Py_DECREF(ret); + } + + if (PyErr_Occurred()) + PyErr_Print(); + + UWSGI_RELEASE_GIL; + return 1; +} + struct uwsgi_plugin python_plugin = { .name = "python", @@ -1448,6 +1475,7 @@ struct uwsgi_plugin python_plugin = { .rpc = uwsgi_python_rpc, .mule = uwsgi_python_mule, + .mule_msg = uwsgi_python_mule_msg, .spooler = uwsgi_python_spooler, diff --git a/plugins/python/uwsgi_pymodule.c b/plugins/python/uwsgi_pymodule.c index 94199935..cf332ad0 100644 --- a/plugins/python/uwsgi_pymodule.c +++ b/plugins/python/uwsgi_pymodule.c @@ -1153,23 +1153,50 @@ PyObject *py_uwsgi_mule_msg(PyObject * self, PyObject * args) { char *message = NULL; Py_ssize_t message_len = 0; - int mule_id = 0; + PyObject *mule_obj = NULL; ssize_t len; + int fd = -1; + int mule_id = -1; - if (!PyArg_ParseTuple(args, "s#|i:mule_msg", &message, &message_len, &mule_id)) { + if (!PyArg_ParseTuple(args, "s#|O:mule_msg", &message, &message_len, &mule_obj)) { return NULL; } - if (mule_id == 0) { + if (mule_obj == NULL) { len = write(uwsgi.shared->mule_queue_pipe[0], message, message_len); if (len <= 0) { uwsgi_error("write()"); } } - else if (mule_id > 0 && mule_id <= uwsgi.mules_cnt) { - len = write(uwsgi.mules[mule_id-1].queue_pipe[0], message, message_len); - if (len <= 0) { - uwsgi_error("write()"); + else { + if (PyString_Check(mule_obj)) { + struct uwsgi_farm *uf = get_farm_by_name(PyString_AsString(mule_obj)); + if (uf == NULL) { + return PyErr_Format(PyExc_ValueError, "unknown farm"); + } + fd = uf->queue_pipe[0]; + } + else if (PyInt_Check(mule_obj)) { + mule_id = PyInt_AsLong(mule_obj); + if (mule_id < 0 && mule_id > uwsgi.mules_cnt) { + return PyErr_Format(PyExc_ValueError, "invalid mule number"); + } + if (mule_id == 0) { + fd = uwsgi.shared->mule_queue_pipe[0]; + } + else { + fd = uwsgi.mules[mule_id-1].queue_pipe[0]; + } + } + else { + return PyErr_Format(PyExc_ValueError, "invalid mule"); + } + + if (fd > -1) { + len = write(fd, message, message_len); + if (len < 0) { + uwsgi_error("write()"); + } } } @@ -1183,20 +1210,23 @@ PyObject *py_uwsgi_mule_get_msg(PyObject * self, PyObject * args, PyObject *kwar ssize_t len = 0; // this buffer will be configurable char *message; - struct pollfd mulepoll[4]; + struct pollfd *mulepoll; int count = 4; + int farms_count = 0; uint8_t uwsgi_signal; PyObject *manage_signals = NULL; + PyObject *manage_farms = NULL; int buffer_size = 65536; int timeout = -1; + int i; - static char *kwlist[] = {"signals", "buffer_size", "timeout", NULL}; + static char *kwlist[] = {"signals", "buffer_size", "timeout", "farms", NULL}; if (uwsgi.muleid == 0) { return PyErr_Format(PyExc_ValueError, "you can receive mule messages only in a mule !!!"); } - if (!PyArg_ParseTupleAndKeywords(args, kwargs, "|Oii:mule_get_msg", kwlist, &manage_signals, &buffer_size, &timeout)) { + if (!PyArg_ParseTupleAndKeywords(args, kwargs, "|OOii:mule_get_msg", kwlist, &manage_signals, &manage_farms, &buffer_size, &timeout)) { return NULL; } @@ -1204,10 +1234,22 @@ PyObject *py_uwsgi_mule_get_msg(PyObject * self, PyObject * args, PyObject *kwar count = 2; } + if (manage_farms == Py_None || manage_farms == Py_False) { + goto next; + } + + for(i=0;i -1) timeout = timeout*1000; message = uwsgi_malloc(buffer_size); + mulepoll = uwsgi_malloc(sizeof(struct pollfd) * (count+farms_count)); mulepoll[0].fd = uwsgi.mules[uwsgi.muleid-1].queue_pipe[1]; mulepoll[0].events = POLLIN; @@ -1219,39 +1261,54 @@ PyObject *py_uwsgi_mule_get_msg(PyObject * self, PyObject * args, PyObject *kwar mulepoll[3].fd = uwsgi.my_signal_socket; mulepoll[3].events = POLLIN; } - int ret = poll(mulepoll, count, timeout); + + for(i=0;imule_queue_pipe[1], message, 65536); + len = read(uwsgi.shared->mule_queue_pipe[1], message, buffer_size); } - else if (count > 2) { - int interesting_fd = -1; - if (mulepoll[2].revents & POLLIN) { - interesting_fd = mulepoll[2].fd; - } - else if (mulepoll[3].revents & POLLIN) { - interesting_fd = mulepoll[3].fd; - } + else { + if (count > 2) { + int interesting_fd = -1; + if (mulepoll[2].revents & POLLIN) { + interesting_fd = mulepoll[2].fd; + } + else if (mulepoll[3].revents & POLLIN) { + interesting_fd = mulepoll[3].fd; + } - if (interesting_fd > -1) { - len = read(interesting_fd, &uwsgi_signal, 1); - if (len <= 0) { - uwsgi_log_verbose("uWSGI mule %d braying: my master died, i will follow him...\n", uwsgi.muleid); - end_me(0); - } + if (interesting_fd > -1) { + len = read(interesting_fd, &uwsgi_signal, 1); + if (len <= 0) { + uwsgi_log_verbose("uWSGI mule %d braying: my master died, i will follow him...\n", uwsgi.muleid); + end_me(0); + } #ifdef UWSGI_DEBUG - uwsgi_log_verbose("master sent signal %d to mule %d\n", uwsgi_signal, uwsgi.muleid); + uwsgi_log_verbose("master sent signal %d to mule %d\n", uwsgi_signal, uwsgi.muleid); #endif - if (uwsgi_signal_handler(uwsgi_signal)) { - uwsgi_log_verbose("error managing signal %d on mule %d\n", uwsgi_signal, uwsgi.mywid); - } - goto clear; + if (uwsgi_signal_handler(uwsgi_signal)) { + uwsgi_log_verbose("error managing signal %d on mule %d\n", uwsgi_signal, uwsgi.mywid); + } + goto clear; + } + } + + for(i=0;i