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

new Erlang subsystem

Browse Source
This commit is contained in:
roberto@maverick64
2011-02-05 15:16:22 +01:00
parent 4dea7cfb4b
commit 534ff245f2
16 changed files with 680 additions and 823 deletions
Download Patch File Download Diff File Expand all files Collapse all files
buildconf/default.ini
-1
View File
@@ -4,7 +4,6 @@ ini = true
yaml = true
snmp = true
sctp = false
erlang = false
spooler = true
embedded = true
udp = true
erlang.c
-727
View File
@@ -1,727 +0,0 @@
#ifdef UWSGI_ERLANG
#include "uwsgi.h"
extern struct uwsgi_server uwsgi;
static void erlang_log(void);
/*
PyObject *py_erlang_connect(PyObject * self, PyObject * args) {
char *erlang_node;
int fd;
if (!PyArg_ParseTuple(args, "s:erlang_connect", &erlang_node)) {
return NULL;
}
UWSGI_SET_BLOCKING;
fd = erl_connect(erlang_node);
UWSGI_UNSET_BLOCKING;
return PyInt_FromLong(fd);
}
PyObject *py_erlang_recv_message(PyObject * self, PyObject * args) {
int erfd;
struct pollfd erpoll;
ErlMessage emsg;
PyObject *pyer = NULL;
unsigned char erlang_buffer[8192];
int eret;
int timeout = uwsgi.shared->options[UWSGI_OPTION_SOCKET_TIMEOUT];
if (!PyArg_ParseTuple(args, "i|i:erlang_recv_message", &erfd, &timeout)) {
return NULL;
}
if (erfd < 0) {
goto clear;
}
erpoll.fd = erfd;
erpoll.events = POLLIN;
cycle:
memset(&emsg, 0, sizeof(ErlMessage));
UWSGI_SET_BLOCKING;
if (timeout > 0) {
eret = poll(&erpoll, 1, timeout * 1000);
if (eret < 0) {
uwsgi_error("poll()");
goto clear;
}
else if (eret == 0) {
goto clear;
}
}
if (erl_receive_msg(erfd, erlang_buffer, 8192, &emsg) == ERL_MSG) {
if (emsg.type == ERL_TICK) {
goto cycle;
}
if (emsg.msg) {
pyer = eterm_to_py(emsg.msg);
}
if (emsg.msg) {
erl_free_compound(emsg.msg);
}
if (emsg.to) {
erl_free_compound(emsg.to);
}
if (emsg.from) {
erl_free_compound(emsg.from);
}
if (!pyer) {
goto clear;
}
UWSGI_UNSET_BLOCKING;
return pyer;
}
clear:
UWSGI_UNSET_BLOCKING;
Py_INCREF(Py_None);
return Py_None;
}
PyObject *py_erlang_send_message(PyObject * self, PyObject * args) {
ETERM *pymessage;
ETERM *pid;
int erfd;
PyObject *ermessage, *erdest, *zero;
int er_number, er_serial, er_creation;
char *er_node;
if (!PyArg_ParseTuple(args, "iOO|i:erlang_send_message", &erfd, &erdest, &ermessage)) {
return NULL;
}
if (erfd < 0) {
goto clear;
}
if (!PyString_Check(erdest) && !PyDict_Check(erdest)) {
goto clear;
}
pymessage = py_to_eterm(ermessage);
if (!pymessage) {
goto clear;
}
if (PyString_Check(erdest)) {
if (!erl_reg_send(erfd, PyString_AsString(erdest), pymessage)) {
erl_err_msg("erl_reg_send()");
goto clear2;
}
}
else if (PyDict_Check(erdest)) {
zero = PyDict_GetItemString(erdest, "node");
if (!zero) {
goto clear2;
}
if (!PyString_Check(zero)) {
goto clear2;
}
er_node = PyString_AsString(zero);
zero = PyDict_GetItemString(erdest, "number");
if (!zero) {
goto clear2;
}
if (!PyInt_Check(zero)) {
goto clear2;
}
er_number = PyInt_AsLong(zero);
zero = PyDict_GetItemString(erdest, "serial");
if (!zero) {
goto clear2;
}
if (!PyInt_Check(zero)) {
goto clear2;
}
er_serial = PyInt_AsLong(zero);
zero = PyDict_GetItemString(erdest, "creation");
if (!zero) {
goto clear2;
}
if (!PyInt_Check(zero)) {
goto clear2;
}
er_creation = PyInt_AsLong(zero);
pid = erl_mk_pid((const char *) er_node, er_number, er_serial, er_creation);
if (!pid) {
goto clear2;
}
if (!erl_send(erfd, pid, pymessage)) {
erl_err_msg("erl_send()");
erl_free_term(pid);
goto clear2;
}
erl_free_term(pid);
}
else {
goto clear;
}
erl_free_compound(pymessage);
Py_INCREF(Py_True);
return Py_True;
clear2:
erl_free_compound(pymessage);
clear:
PyErr_Print();
Py_INCREF(Py_None);
return Py_None;
}
PyObject *py_erlang_rpc(PyObject * self, PyObject * args) {
int fd, timeout = uwsgi.shared->options[UWSGI_OPTION_SOCKET_TIMEOUT];
char *emod, *efun;
PyObject *eargs, *pyer = NULL;
ETERM *pyargs, *rex;
ErlMessage emsg;
int eret;
if (!PyArg_ParseTuple(args, "issO|i:erlang_rpc", &fd, &emod, &efun, &eargs, &timeout)) {
return NULL;
}
if (fd < 0) {
goto clear;
}
pyargs = py_to_eterm(eargs);
if (!pyargs) {
goto clear;
}
if (erl_rpc_to(fd, emod, efun, pyargs)) {
erl_err_msg("erl_rpc_to()");
goto clear2;
}
cycle:
memset(&emsg, 0, sizeof(ErlMessage));
UWSGI_SET_BLOCKING;
eret = erl_rpc_from(fd, timeout * 1000, &emsg);
if (eret == ERL_TICK) {
goto cycle;
}
else if (eret == ERL_MSG) {
if (emsg.msg) {
if (ERL_IS_TUPLE(emsg.msg)) {
rex = erl_element(1, emsg.msg);
if (!rex) {
goto clear2;
}
if (!strncmp("rex", ERL_ATOM_PTR(rex), ERL_ATOM_SIZE(rex))) {
erl_free_term(rex);
rex = erl_element(2, emsg.msg);
if (!rex) {
goto clear2;
}
pyer = eterm_to_py(rex);
erl_free_term(rex);
}
else {
erl_free_term(rex);
}
}
}
if (emsg.msg) {
erl_free_compound(emsg.msg);
}
if (emsg.to) {
erl_free_compound(emsg.to);
}
if (emsg.from) {
erl_free_compound(emsg.from);
}
if (!pyer) {
goto clear2;
}
erl_free_compound(pyargs);
UWSGI_UNSET_BLOCKING;
return pyer;
}
else {
erl_err_msg("erl_rpc_from()");
}
clear2:
UWSGI_UNSET_BLOCKING;
erl_free_compound(pyargs);
clear:
Py_INCREF(Py_None);
return Py_None;
}
PyObject *py_erlang_close(PyObject * self, PyObject * args) {
int fd;
if (!PyArg_ParseTuple(args, "i:erlang_close", &fd)) {
return NULL;
}
if (fd >= 0) {
erl_close_connection(fd);
}
Py_INCREF(Py_True);
return Py_True;
}
static PyMethodDef uwsgi_erlang_methods[] = {
{"erlang_connect", py_erlang_connect, METH_VARARGS, ""},
{"erlang_send_message", py_erlang_send_message, METH_VARARGS, ""},
{"erlang_recv_message", py_erlang_recv_message, METH_VARARGS, ""},
{"erlang_rpc", py_erlang_rpc, METH_VARARGS, ""},
{"erlang_close", py_erlang_close, METH_VARARGS, ""},
{NULL, NULL},
};
*/
int init_erlang(char *nodename, char *cookie) {
struct sockaddr_in e_addr;
char *ip;
char *node;
int efd;
int rlen;
char *cookiefile;
char *cookiehome;
char cookievalue[128];
int cookiefd;
//PyMethodDef *uwsgi_function;
ip = strchr(nodename, '@');
if (ip == NULL) {
uwsgi_log( "*** invalid erlang node name ***\n");
return -1;
}
uwsgi_log("using cookie %s\n", cookie);
if (cookie == NULL) {
// get the cookie from the home
cookiehome = getenv("HOME");
if (!cookiehome) {
uwsgi_log( "unable to get erlang cookie from your home.\n");
return -1;
}
cookiefile = malloc(strlen(cookiehome) + 1 + strlen(".erlang.cookie") + 1);
if (!cookiefile) {
uwsgi_error("malloc()");
}
cookiefile[0] = 0;
strcat(cookiefile, cookiehome);
strcat(cookiefile, "/.erlang.cookie");
cookiefd = open(cookiefile, O_RDONLY);
if (cookiefd < 0) {
uwsgi_error("open()");
free(cookiefile);
return -1;
}
memset(cookievalue, 0, 128);
if (read(cookiefd, cookievalue, 127) < 1) {
uwsgi_log( "invalid cookie found in %s\n", cookiefile);
close(cookiefd);
free(cookiefile);
return -1;
}
cookie = cookievalue;
close(cookiefd);
free(cookiefile);
}
node = malloc((ip - nodename) + 1);
if (node == NULL) {
uwsgi_error("malloc()");
return -1;
}
memset(node, 0, (ip - nodename) + 1);
memcpy(node, nodename, ip - nodename);
erl_init(NULL, 0);
if (erl_connect_xinit(ip + 1, node, nodename, NULL, cookie, 0) == -1) {
uwsgi_log( "*** unable to initialize erlang c-node ***\n");
return -1;
}
efd = socket(AF_INET, SOCK_STREAM, 0);
if (efd < 0) {
uwsgi_error("socket()");
return -1;
}
memset(&e_addr, 0, sizeof(struct sockaddr_in));
e_addr.sin_family = AF_INET;
e_addr.sin_addr.s_addr = inet_addr(ip + 1);
rlen = 1;
if (setsockopt(efd, SOL_SOCKET, SO_REUSEADDR, &rlen, sizeof(rlen))) {
uwsgi_error("setsockopt()");
close(efd);
return -1;
}
if (bind(efd, (struct sockaddr *) &e_addr, sizeof(struct sockaddr_in)) < 0) {
uwsgi_error("bind()");
close(efd);
return -1;
}
rlen = sizeof(struct sockaddr_in);
if (getsockname(efd, (struct sockaddr *) &e_addr, (socklen_t *) & rlen)) {
uwsgi_error("getsockname()");
close(efd);
return -1;
}
if (listen(efd, uwsgi.listen_queue)) {
uwsgi_error("listen()");
close(efd);
return -1;
}
if (erl_publish(ntohs(e_addr.sin_port)) < 0) {
uwsgi_log( "*** unable to subscribe with EPMD ***\n");
close(efd);
return -1;
}
uwsgi_log( "Erlang C-Node initialized on port %d you can access it with name %s\n", ntohs(e_addr.sin_port), nodename);
/*
for (uwsgi_function = uwsgi_erlang_methods; uwsgi_function->ml_name != NULL; uwsgi_function++) {
PyObject *func = PyCFunction_New(uwsgi_function, NULL);
PyDict_SetItemString(uwsgi.embedded_dict, uwsgi_function->ml_name, func);
Py_DECREF(func);
}
*/
return efd;
}
/*
ETERM *py_to_eterm(PyObject * pobj) {
int i;
int count;
PyObject *pobj2;
ETERM *eobj = NULL;
ETERM *eobj2 = NULL;
ETERM **eobj3;
if (pobj == NULL) {
return erl_mk_empty_list();
}
if (PyString_Check(pobj)) {
eobj = erl_mk_atom(PyString_AsString(pobj));
}
else if (PyInt_Check(pobj)) {
eobj = erl_mk_int(PyInt_AsLong(pobj));
}
else if (PyList_Check(pobj)) {
eobj = erl_mk_empty_list();
for (i = PyList_Size(pobj) - 1; i >= 0; i--) {
pobj2 = PyList_GetItem(pobj, i);
eobj2 = py_to_eterm(pobj2);
eobj = erl_cons(eobj2, eobj);
}
}
else if (PyDict_Check(pobj)) {
// a pid
char *er_node;
int er_number, er_serial, er_creation;
pobj2 = PyDict_GetItemString(pobj, "node");
if (!pobj2) {
PyErr_Print();
goto clear;
}
if (!PyString_Check(pobj2)) {
goto clear;
}
er_node = PyString_AsString(pobj2);
pobj2 = PyDict_GetItemString(pobj, "number");
if (!pobj2) {
PyErr_Print();
goto clear;
}
if (!PyInt_Check(pobj2)) {
goto clear;
}
er_number = PyInt_AsLong(pobj2);
pobj2 = PyDict_GetItemString(pobj, "serial");
if (!pobj2) {
PyErr_Print();
goto clear;
}
if (!PyInt_Check(pobj2)) {
goto clear;
}
er_serial = PyInt_AsLong(pobj2);
pobj2 = PyDict_GetItemString(pobj, "creation");
if (!pobj2) {
PyErr_Print();
goto clear;
}
if (!PyInt_Check(pobj2)) {
goto clear;
}
er_creation = PyInt_AsLong(pobj2);
eobj = erl_mk_pid(er_node, er_number, er_serial, er_creation);
}
else if (PyTuple_Check(pobj)) {
count = PyTuple_Size(pobj);
eobj3 = malloc(sizeof(ETERM *) * count);
for (i = 0; i < count; i++) {
pobj2 = PyTuple_GetItem(pobj, i);
if (!pobj2) {
break;
}
eobj3[i] = py_to_eterm(pobj2);
}
eobj = erl_mk_tuple(eobj3, count);
free(eobj3);
}
else {
uwsgi_log( "UNMANAGED PYTHON TYPE: %s\n", pobj->ob_type->tp_name);
}
clear:
if (eobj == NULL) {
return erl_mk_empty_list();
}
return eobj;
}
PyObject *eterm_to_py(ETERM * obj) {
int i;
int count;
ETERM *obj2;
PyObject *eobj = NULL;
if (obj == NULL) {
Py_INCREF(Py_None);
return Py_None;
}
switch (ERL_TYPE(obj)) {
case ERL_CONS:
case ERL_NIL:
count = erl_length(obj);
eobj = PyList_New(0);
for (i = 0; i < count; i++) {
obj2 = erl_hd(obj);
PyList_Append(eobj, eterm_to_py(obj2));
obj = erl_tl(obj);
}
break;
case ERL_TUPLE:
eobj = PyTuple_New(erl_size(obj));
for (i = 1; i <= erl_size(obj); i++) {
obj2 = erl_element(i, obj);
PyTuple_SetItem(eobj, i - 1, eterm_to_py(obj2));
}
break;
case ERL_ATOM:
eobj = PyString_FromStringAndSize(ERL_ATOM_PTR(obj), ERL_ATOM_SIZE(obj));
break;
case ERL_INTEGER:
eobj = PyInt_FromLong(ERL_INT_VALUE(obj));
break;
case ERL_BINARY:
uwsgi_log( "FOUND A BINARY %.*s\n", ERL_BIN_SIZE(obj), ERL_BIN_PTR(obj));
break;
case ERL_PID:
eobj = PyDict_New();
if (PyDict_SetItemString(eobj, "node", PyString_FromString(ERL_PID_NODE(obj)))) {
PyErr_Print();
break;
}
if (PyDict_SetItemString(eobj, "number", PyInt_FromLong(ERL_PID_NUMBER(obj)))) {
PyErr_Print();
break;
}
if (PyDict_SetItemString(eobj, "serial", PyInt_FromLong(ERL_PID_SERIAL(obj)))) {
PyErr_Print();
break;
}
if (PyDict_SetItemString(eobj, "creation", PyInt_FromLong(ERL_PID_CREATION(obj)))) {
PyErr_Print();
break;
}
default:
uwsgi_log( "UNMANAGED ETERM TYPE: %d\n", ERL_TYPE(obj));
break;
}
if (eobj == NULL) {
Py_INCREF(Py_None);
return Py_None;
}
return eobj;
}
*/
void erlang_loop(struct wsgi_request *wsgi_req) {
ErlConnect econn;
ErlMessage em;
unsigned char buffer[4096];
//ETERM *eresponse;
/*
PyObject *callable = PyDict_GetItemString(uwsgi.embedded_dict, "erlang_func");
if (!callable) {
PyErr_Print();
uwsgi_log( "- you have not defined a uwsgi.erlang_func callable, Erlang message manager will be disabled until you define it -\n");
}
PyObject *pargs = PyTuple_New(1);
if (!pargs) {
PyErr_Print();
uwsgi_log( "- error preparing arg tuple for uwsgi.erlang_func callable, Erlang message manager will be disabled -\n");
}
*/
while (uwsgi.workers[uwsgi.mywid].manage_next_request) {
uwsgi_log("entering erlang cycle\n");
UWSGI_CLEAR_STATUS;
wsgi_req->poll.fd = erl_accept(uwsgi.erlangfd, &econn);
uwsgi_log("accepted %d\n", wsgi_req->poll.fd);
if (wsgi_req->poll.fd >= 0) {
UWSGI_SET_ERLANGING;
for (;;) {
uwsgi_log("receive message\n");
if (erl_receive_msg(wsgi_req->poll.fd, buffer, 4096, &em) == ERL_MSG) {
uwsgi_log("message received\n");
if (em.type == ERL_TICK)
continue;
/*
if (!callable) {
callable = PyDict_GetItemString(uwsgi.embedded_dict, "erlang_func");
}
if (!callable) {
uwsgi_log( "- you still have not defined a uwsgi.erlang_func callable, Erlang message rejected -\n");
}
PyObject *zero = eterm_to_py(em.msg);
*/
uwsgi_log("received object %d\n", ERL_TYPE(em.msg));
if (em.msg) {
erl_free_compound(em.msg);
}
if (em.to) {
erl_free_compound(em.to);
}
/*
if (!zero) {
PyErr_Print();
continue;
}
if (PyTuple_SetItem(pargs, 0, zero)) {
PyErr_Print();
continue;
}
PyObject *erlang_result = PyEval_CallObject(callable, pargs);
//Py_DECREF(zero);
if (erlang_result) {
eresponse = py_to_eterm(erlang_result);
if (eresponse) {
erl_send(wsgi_req->poll.fd, em.from, eresponse);
erl_free_compound(eresponse);
}
Py_DECREF(erlang_result);
}
*/
if (em.from) {
erl_free_compound(em.from);
}
uwsgi.workers[0].requests++;
uwsgi.workers[uwsgi.mywid].requests++;
if (uwsgi.shared->options[UWSGI_OPTION_LOGGING])
erlang_log();
}
else {
uwsgi_log("oooops\n");
break;
}
}
erl_close_connection(wsgi_req->poll.fd);
UWSGI_UNSET_ERLANGING;
}
}
}
static void erlang_log() {
if (uwsgi.shared->options[UWSGI_OPTION_MEMORY_DEBUG]) {
get_memusage();
}
else {
uwsgi.workers[uwsgi.mywid].rss_size = 0;
uwsgi.workers[uwsgi.mywid].vsz_size = 0;
}
uwsgi_log( "[Erlang worker %d pid %d] request %llu done {rss: %llu vsz: %llu}\n", uwsgi.mywid, uwsgi.mypid, uwsgi.workers[uwsgi.mywid].requests, uwsgi.workers[uwsgi.mywid].rss_size, uwsgi.workers[uwsgi.mywid].vsz_size);
}
#else
#warning "*** Erlang support is disabled ***"
#endif
gateway.c
+27
View File
@@ -2,6 +2,33 @@
extern struct uwsgi_server uwsgi;
struct uwsgi_gateway *register_fat_gateway(char *name, void (*loop)(void)) {
struct uwsgi_gateway *ug;
int num=1,i;
if (uwsgi.gateways_cnt >= MAX_GATEWAYS) {
uwsgi_log("you can register max %d gateways\n", MAX_GATEWAYS);
return NULL;
}
for(i=0;i<uwsgi.gateways_cnt;i++) {
if (!strcmp(name, uwsgi.gateways[i].name)) {
num++;
}
}
ug = &uwsgi.gateways[uwsgi.gateways_cnt];
ug->pid = 0;
ug->name = name;
ug->loop = loop;
ug->num = num;
uwsgi.gateways_cnt++;
return ug;
}
struct uwsgi_gateway *register_gateway(char *name, void (*loop)(void)) {
pid_t gw_pid;
master.c
+7
View File
@@ -223,6 +223,13 @@ void master_loop(char **argv, char **environ) {
*/
// spawn fat gateways
for(i=0;i<uwsgi.gateways_cnt;i++) {
if (uwsgi.gateways[i].pid == 0) {
gateway_respawn(i);
}
}
for (;;) {
//uwsgi_log("ready_to_reload %d %d\n", ready_to_reload, uwsgi.numproc);
if (ready_to_die >= uwsgi.numproc && uwsgi.to_hell) {
plugins/erlang/erlang.c
+401
View File
@@ -0,0 +1,401 @@
#include "uwsgi.h"
#include "erlang.h"
extern struct uwsgi_server uwsgi;
struct uwsgi_erlang uerl;
struct option erlang_options[] = {
{"erlang", required_argument, 0, LONG_ARGS_ERLANG},
{"erlang-cookie", required_argument, 0, LONG_ARGS_ERLANG_COOKIE},
{0, 0, 0, 0},
};
void dump_eterm(ei_x_buff *x) {
int etype, esize, arity;
long long num;
char *atom;
int i;
char *binary;
long bin_size;
double fnum;
ei_get_type(x->buff, &x->index, &etype, &esize);
uwsgi_log("etype: %d/%c esize: %d\n", etype, etype, esize);
switch(etype) {
case ERL_SMALL_INTEGER_EXT:
case ERL_INTEGER_EXT:
case ERL_SMALL_BIG_EXT:
case ERL_LARGE_BIG_EXT:
ei_decode_longlong(x->buff, &x->index, &num);
uwsgi_log("num: %lu\n", num);
break;
case ERL_FLOAT_EXT:
ei_decode_double(x->buff, &x->index, &fnum);
uwsgi_log("float: %f\n", fnum);
break;
case ERL_STRING_EXT:
atom = uwsgi_malloc(esize+1);
ei_decode_string(x->buff, &x->index, atom);
uwsgi_log("string: %s\n", atom);
free(atom);
break;
case ERL_ATOM_EXT:
atom = uwsgi_malloc(esize+1);
ei_decode_atom(x->buff, &x->index, atom);
uwsgi_log("atom: %s\n", atom);
free(atom);
break;
case ERL_SMALL_TUPLE_EXT:
case ERL_LARGE_TUPLE_EXT:
ei_decode_tuple_header(x->buff, &x->index, &arity);
for(i=0;i<arity;i++) {
dump_eterm(x);
}
break;
case ERL_LIST_EXT:
case ERL_NIL_EXT:
ei_decode_list_header(x->buff, &x->index, &arity);
if (arity == 0) {
uwsgi_log("nil value\n");
break;
}
for(i=0;i<arity+1;i++) {
dump_eterm(x);
}
break;
case ERL_BINARY_EXT:
binary = uwsgi_malloc(esize);
ei_decode_binary(x->buff, &x->index, binary, &bin_size);
uwsgi_log("binary data of %d bytes\n", bin_size);
free(binary);
break;
default:
uwsgi_log("ignored...\n");
ei_skip_term(x->buff, &x->index);
break;
}
}
void uwsgi_erlang_rpc(int fd, erlang_pid *from, ei_x_buff *x) {
int etype, esize;
int arity;
char *gen_call;
char *module;
char *call;
char buffer[0xffff];
char *argv[0xff] ;
int argc;
uint16_t ret;
ei_x_buff xr;
erlang_ref eref;
ei_get_type(x->buff, &x->index, &etype, &esize);
uwsgi_log("%d %c %c %c\n", etype, etype, ERL_SMALL_TUPLE_EXT, ERL_LARGE_TUPLE_EXT);
if (etype != ERL_SMALL_TUPLE_EXT && etype != ERL_LARGE_TUPLE_EXT) return;
uwsgi_log("decode tuple\n");
ei_decode_tuple_header(x->buff, &x->index, &arity);
uwsgi_log("rpc arity %d\n", arity);
if (arity != 3) return ;
ei_get_type(x->buff, &x->index, &etype, &esize);
uwsgi_log("%d %c\n", etype, etype);
if (etype != ERL_ATOM_EXT && etype != ERL_STRING_EXT) return ;
gen_call = uwsgi_malloc(esize);
if (etype == ERL_ATOM_EXT) {
ei_decode_atom(x->buff, &x->index, gen_call);
}
else {
ei_decode_string(x->buff, &x->index, gen_call);
}
uwsgi_log("gen call = %s\n", gen_call);
ei_get_type(x->buff, &x->index, &etype, &esize);
if (etype != ERL_SMALL_TUPLE_EXT) return ;
ei_decode_tuple_header(x->buff, &x->index, &arity);
if (arity != 2) return ;
ei_get_type(x->buff, &x->index, &etype, &esize);
ei_skip_term(x->buff, &x->index);
uwsgi_log("skip0 %d %c\n", etype, etype);
ei_get_type(x->buff, &x->index, &etype, &esize);
uwsgi_log("skip1 %d %c\n", etype, etype);
ei_decode_ref(x->buff, &x->index, &eref);
ei_get_type(x->buff, &x->index, &etype, &esize);
uwsgi_log("%d %c\n", etype, etype);
module = uwsgi_malloc(esize);
if (etype == ERL_ATOM_EXT) {
ei_decode_atom(x->buff, &x->index, module);
}
else {
ei_decode_string(x->buff, &x->index, module);
}
ei_get_type(x->buff, &x->index, &etype, &esize);
uwsgi_log("%d %c\n", etype, etype);
if (etype != ERL_SMALL_TUPLE_EXT) return ;
ei_decode_tuple_header(x->buff, &x->index, &arity);
uwsgi_log("arity: %d\n", arity);
if (arity != 5) return ;
ei_get_type(x->buff, &x->index, &etype, &esize);
uwsgi_log("%d %c\n", etype, etype);
char *method = uwsgi_malloc(esize);
if (etype == ERL_ATOM_EXT) {
ei_decode_atom(x->buff, &x->index, method);
}
else {
ei_decode_string(x->buff, &x->index, method);
}
if (strcmp(method, "call")) return;
ei_get_type(x->buff, &x->index, &etype, &esize);
uwsgi_log("%d %c\n", etype, etype);
if (etype != ERL_ATOM_EXT && etype != ERL_STRING_EXT) return ;
module = uwsgi_malloc(esize);
if (etype == ERL_ATOM_EXT) {
ei_decode_atom(x->buff, &x->index, module);
}
else {
ei_decode_string(x->buff, &x->index, module);
}
ei_get_type(x->buff, &x->index, &etype, &esize);
if (etype != ERL_ATOM_EXT && etype != ERL_STRING_EXT) return ;
call = uwsgi_malloc(esize);
if (etype == ERL_ATOM_EXT) {
ei_decode_atom(x->buff, &x->index, call);
}
else {
ei_decode_string(x->buff, &x->index, call);
}
uwsgi_log("RPC %s %s\n", module, call);
ei_get_type(x->buff, &x->index, &etype, &esize);
if (etype == ERL_ATOM_EXT) {
argc = 1;
argv[0] = uwsgi_malloc(esize+1);
ei_decode_atom(x->buff, &x->index, argv[0]);
}
else if (etype == ERL_STRING_EXT) {
argc = 1;
argv[0] = uwsgi_malloc(esize+1);
ei_decode_string(x->buff, &x->index, argv[0]);
}
ret = uwsgi_rpc(call, argc, argv, buffer);
uwsgi_log("buffer: %.*s\n", ret, buffer);
ei_x_new_with_version(&xr);
ei_x_encode_tuple_header(&xr, 2);
//ei_x_encode_atom(&xr, "rex");
ei_x_encode_ref(&xr, &eref);
ei_x_encode_string_len(&xr, buffer, ret);
uwsgi_log("ei_send to %d %s %d %d %d: %d %d\n", fd, from->node, from->num , from->serial, from->creation, xr.index, ei_send(fd, from, xr.buff, xr.index));
//uwsgi_log("ei_send to %d %s %d %d %d: %d %d\n", fd, from->node, from->num , from->serial, from->creation, xr.index, ei_reg_send(&uerl.cnode, fd, "rex", xr.buff, xr.index));
}
void erlang_loop() {
ErlConnect econn;
//ErlMessage em;
erlang_msg em;
int fd;
int eversion;
ei_x_buff x, xr;
ei_x_new(&x);
ei_x_new(&xr);
/*
int fd0 = ei_connect(&uerl.cnode, "anothernode@maverick64");
uwsgi_log("fd0: %d\n", fd0);
ei_x_encode_list_header(&x, 0);
ei_rpc_to(&uerl.cnode, fd0, "erlang", "node", x.buff, x.index);
ei_rpc_from(&uerl.cnode, fd0, 10000, &em, &xr);
uwsgi_log("From: %s To: %s RegName: %s\n", em.from.node, em.to.node, em.toname);
xr.index = 0;
ei_decode_version(xr.buff, &xr.index, &eversion);
uwsgi_log("eversion: %d\n", eversion);
dump_eterm(&xr);
*/
for(;;) {
fd = ei_accept(&uerl.cnode, uerl.fd, &econn);
if (fd >= 0) {
for (;;) {
if (ei_xreceive_msg(fd, &em, &x) == ERL_MSG) {
if (em.msgtype == ERL_TICK)
continue;
uwsgi_log("From: %s To: %s RegName: %s\n", em.from.node, em.to.node, em.toname);
x.index = 0;
ei_decode_version(x.buff, &x.index, &eversion);
uwsgi_log("eversion: %d\n", eversion);
if (!strcmp(em.toname, "rex")) {
uwsgi_erlang_rpc(fd, &em.from, &x);
}
else {
dump_eterm(&x);
}
/*
if (em.msgtype) {
dump_erl_obj(em.msg);
erl_free_compound(em.msg);
}
if (em.to) {
uwsgi_log("*** TO ***\n");
dump_erl_obj(em.to);
erl_free_compound(em.to);
}
if (em.from) {
uwsgi_log("*** FROM ***\n");
dump_erl_obj(em.from);
erl_free_compound(em.from);
}
*/
}
else {
break;
}
}
close(fd);
}
}
}
int erlang_init() {
char *host;
struct sockaddr_in sin;
socklen_t slen = sizeof(struct sockaddr_in);
if (uerl.name) {
uwsgi.master_process = 1;
host = strchr(uerl.name, '@');
if (!host) {
if (ei_connect_init(&uerl.cnode, uerl.name, uerl.cookie, 0) < 0) {
uwsgi_log("unable to initialize erlang connection\n");
exit(1);
}
}
uerl.fd = bind_to_tcp("", uwsgi.listen_queue, NULL);
if (uerl.fd < 0) {
exit(1);
}
if (getsockname(uerl.fd, (struct sockaddr *) &sin, &slen)) {
uwsgi_error("getsockname()");
exit(1);
}
if (ei_publish(&uerl.cnode, ntohs(sin.sin_port)) < 0) {
uwsgi_log( "*** unable to subscribe with EPMD ***\n");
exit(1);
}
uwsgi_log("Erlang C-Node %s registered on port %d\n", ei_thisnodename(&uerl.cnode), ntohs(sin.sin_port));
if (register_fat_gateway("erlang", erlang_loop) == NULL) {
uwsgi_log("unable to register the erlang gateway\n");
exit(1);
}
}
return 0;
}
int erlang_opt(int i, char *optarg) {
switch(i) {
case LONG_ARGS_ERLANG:
uerl.name = optarg;
return 1;
case LONG_ARGS_ERLANG_COOKIE:
uerl.cookie = optarg;
return 1;
}
return 0;
}
struct uwsgi_plugin erlang_plugin = {
.options = erlang_options,
.manage_opt = erlang_opt,
.init = erlang_init,
};
plugins/erlang/erlang.h
+17
View File
@@ -0,0 +1,17 @@
#include <ei.h>
#define LONG_ARGS_ERLANG 17012
#define LONG_ARGS_ERLANG_COOKIE 17013
struct uwsgi_erlang {
ei_cnode cnode;
char *name;
char *cookie;
int fd;
};
plugins/erlang/uwsgiplugin.py
+7
View File
@@ -0,0 +1,7 @@
NAME='erlang'
CFLAGS = []
LDFLAGS = []
LIBS = ['-lei']
GCC_LIST = ['erlang']
plugins/pyerl/pyerl.c
+135
View File
@@ -0,0 +1,135 @@
#include "../erlang/erlang.h"
#include "../python/uwsgi_python.h"
extern struct uwsgi_server uwsgi;
extern struct uwsgi_erlang uerl;
extern struct uwsgi_python up;
ei_cnode *pyerl_cnode;
PyObject *pyerl_connect(PyObject * self, PyObject * args) {
Py_INCREF(Py_None);
return Py_None;
}
int py_to_erl(PyObject *, ei_x_buff*);
PyObject *pyerl_simple_send(PyObject * self, PyObject * args) {
char *node;
char *reg;
PyObject *pobj;
ei_x_buff x;
int fd;
if (!PyArg_ParseTuple(args, "ssO:erlang_simple_send", &node, &reg, &pobj)) {
return NULL;
}
fd = ei_connect(pyerl_cnode, node);
if (fd < 0) {
return PyErr_Format(PyExc_ValueError, "Unable to connect to erlang node %s", node);
}
ei_x_new_with_version(&x);
if (py_to_erl(pobj, &x) < 0) {
ei_x_free(&x);
return PyErr_Format(PyExc_ValueError, "Unsupported object in Python->Erlang translation");
}
ei_reg_send(pyerl_cnode, fd, reg, x.buff, x.index);
close(fd);
ei_x_free(&x);
Py_INCREF(Py_None);
return Py_None;
}
static PyMethodDef uwsgi_pyerl_methods[] = {
{"erlang_connect", pyerl_connect, METH_VARARGS, ""},
{"erlang_simple_send", pyerl_simple_send, METH_VARARGS, ""},
{NULL, NULL},
};
void py_erl_init_functions() {
PyMethodDef *uwsgi_function;
for (uwsgi_function = uwsgi_pyerl_methods; uwsgi_function->ml_name != NULL; uwsgi_function++) {
PyObject *func = PyCFunction_New(uwsgi_function, NULL);
PyDict_SetItemString(up.embedded_dict, uwsgi_function->ml_name, func);
Py_DECREF(func);
}
}
int py_to_erl(PyObject *pobj, ei_x_buff *x) {
int i;
PyObject *pobj2;
if (pobj == NULL || pobj == Py_None) {
ei_x_encode_empty_list(x);
}
else if (PyString_Check(pobj)) {
ei_x_encode_string(x, PyString_AsString(pobj));
}
else if (PyInt_Check(pobj)) {
ei_x_encode_long(x, PyInt_AsLong(pobj));
}
else if (PyList_Check(pobj)) {
ei_x_encode_list_header(x, PyList_Size(pobj));
for (i = 0; i < PyList_Size(pobj); i++) {
pobj2 = PyList_GetItem(pobj, i);
if (py_to_erl(pobj2, x) < 0) return -1;
}
ei_x_encode_empty_list(x);
}
else if (PyTuple_Check(pobj)) {
ei_x_encode_tuple_header(x, PyTuple_Size(pobj));
for (i = 0; i < PyTuple_Size(pobj); i++) {
pobj2 = PyTuple_GetItem(pobj, i);
if (py_to_erl(pobj2, x) < 0) return -1;
}
}
else {
return -1;
}
return x->index;
}
void pyerl_init() {
up.extension = py_erl_init_functions;
if (!uerl.name) {
pyerl_cnode = uwsgi_malloc(sizeof(ei_cnode));
memset(pyerl_cnode, 0, sizeof(ei_cnode));
if (ei_connect_init(pyerl_cnode, "uwsgi", NULL, 0) < 0) {
uwsgi_log("unable to initialize erlang connection\n");
exit(1);
}
uwsgi_log("Erlang C-Node name: %s\n",ei_thisnodename(pyerl_cnode));
}
else {
pyerl_cnode = &uerl.cnode;
}
uwsgi_log("enabled Python<->Erlang bridge\n");
}
struct uwsgi_plugin pyerl_plugin = {
.post_init = pyerl_init,
};
plugins/pyerl/uwsgiplugin.py
+8
View File
@@ -0,0 +1,8 @@
from distutils import sysconfig
NAME='pyerl'
CFLAGS = ['-I' + sysconfig.get_python_inc(), '-I' + sysconfig.get_python_inc(plat_specific=True)]
LDFLAGS = []
LIBS = []
GCC_LIST = ['pyerl']
plugins/python/python_plugin.c
+42 -33
View File
@@ -91,39 +91,6 @@ int uwsgi_python_init() {
uwsgi_log("Python main interpreter initialized at %p\n", up.main_thread);
init_pyargv();
#ifdef UWSGI_MINTERPRETERS
init_uwsgi_embedded_module();
#endif
if (up.test_module != NULL) {
if (PyImport_ImportModule(up.test_module)) {
exit(0);
}
exit(1);
}
init_uwsgi_vars();
// setup app loaders
#ifdef UWSGI_MINTERPRETERS
up.loaders[LOADER_DYN] = uwsgi_dyn_loader;
#endif
up.loaders[LOADER_UWSGI] = uwsgi_uwsgi_loader;
up.loaders[LOADER_FILE] = uwsgi_file_loader;
up.loaders[LOADER_PASTE] = uwsgi_paste_loader;
up.loaders[LOADER_EVAL] = uwsgi_eval_loader;
up.loaders[LOADER_MOUNT] = uwsgi_mount_loader;
up.loaders[LOADER_CALLABLE] = uwsgi_callable_loader;
up.loaders[LOADER_STRING_CALLABLE] = uwsgi_string_callable_loader;
// by default set a fake GIL (little impact on performance)
up.gil_get = gil_fake_get;
up.gil_release = gil_fake_release;
up.swap_ts = simple_swap_ts;
up.reset_ts = simple_reset_ts;
return 1;
}
@@ -451,10 +418,14 @@ void init_uwsgi_embedded_module() {
PyErr_Print();
exit(1);
}
/*
if (PyDict_SetItemString(up.embedded_dict, "KIND_ERLANG", PyInt_FromLong(KIND_ERLANG))) {
PyErr_Print();
exit(1);
}
*/
if (PyDict_SetItemString(up.embedded_dict, "KIND_PROXY", PyInt_FromLong(KIND_PROXY))) {
PyErr_Print();
exit(1);
@@ -578,6 +549,10 @@ void init_uwsgi_embedded_module() {
}
init_uwsgi_module_cache(new_uwsgi_module);
if (up.extension) {
up.extension();
}
}
#endif
@@ -690,6 +665,40 @@ int uwsgi_python_mount_app(char *mountpoint, char *app) {
void uwsgi_python_init_apps() {
init_pyargv();
#ifdef UWSGI_MINTERPRETERS
init_uwsgi_embedded_module();
#endif
if (up.test_module != NULL) {
if (PyImport_ImportModule(up.test_module)) {
exit(0);
}
exit(1);
}
init_uwsgi_vars();
// setup app loaders
#ifdef UWSGI_MINTERPRETERS
up.loaders[LOADER_DYN] = uwsgi_dyn_loader;
#endif
up.loaders[LOADER_UWSGI] = uwsgi_uwsgi_loader;
up.loaders[LOADER_FILE] = uwsgi_file_loader;
up.loaders[LOADER_PASTE] = uwsgi_paste_loader;
up.loaders[LOADER_EVAL] = uwsgi_eval_loader;
up.loaders[LOADER_MOUNT] = uwsgi_mount_loader;
up.loaders[LOADER_CALLABLE] = uwsgi_callable_loader;
up.loaders[LOADER_STRING_CALLABLE] = uwsgi_string_callable_loader;
// by default set a fake GIL (little impact on performance)
up.gil_get = gil_fake_get;
up.gil_release = gil_fake_release;
up.swap_ts = simple_swap_ts;
up.reset_ts = simple_reset_ts;
if (up.wsgi_config != NULL) {
init_uwsgi_app(LOADER_UWSGI, up.wsgi_config, uwsgi.wsgi_req, up.main_thread);
}
plugins/python/uwsgi_python.h
+2
View File
@@ -126,6 +126,8 @@ struct uwsgi_python {
char *pymodule_alias[MAX_PYMODULE_ALIAS];
int pymodule_alias_cnt;
void (*extension)(void);
};
socket.c
+6 -4
View File
@@ -370,11 +370,13 @@ int bind_to_tcp(char *socket_name, int listen_queue, char *tcp_port) {
struct sockaddr_in uws_addr;
int reuse = 1;
tcp_port[0] = 0;
memset(&uws_addr, 0, sizeof(struct sockaddr_in));
uws_addr.sin_family = AF_INET;
uws_addr.sin_port = htons(atoi(tcp_port + 1));
if (tcp_port) {
tcp_port[0] = 0;
uws_addr.sin_port = htons(atoi(tcp_port + 1));
}
serverfd = socket(AF_INET, SOCK_STREAM, 0);
if (serverfd < 0) {
@@ -427,8 +429,8 @@ int bind_to_tcp(char *socket_name, int listen_queue, char *tcp_port) {
exit(1);
}
tcp_port[0] = ':';
if (tcp_port) tcp_port[0] = ':';
return serverfd;
}
utils.c
+10
View File
@@ -1542,3 +1542,13 @@ int is_unix(char *socket_name, int len) {
return 1;
}
int is_a_number(char *what) {
int i;
for(i=0;i<(int)strlen(what);i++) {
if (!isdigit(what[i])) return 0;
}
return 1;
}
uwsgi.c
+15 -24
View File
@@ -1333,8 +1333,8 @@ int uwsgi_start(void *v_argv) {
}
// initialize request plugin only if workers are available
if (uwsgi.sockets_cnt) {
// initialize request plugin only if workers or master are available
if (uwsgi.sockets_cnt || uwsgi.master_process) {
for (i = 0; i < 0xFF; i++) {
if (uwsgi.p[i]->init) {
uwsgi.p[i]->init();
@@ -1343,6 +1343,19 @@ int uwsgi_start(void *v_argv) {
}
/* gp/plugin initialization */
for(i =0; i < uwsgi.gp_cnt; i++) {
if (uwsgi.gp[i]->post_init) {
uwsgi.gp[i]->post_init();
}
}
for(i =0; i < 0xff; i++) {
if (uwsgi.p[i]->post_init) {
uwsgi.p[i]->post_init();
}
}
uwsgi.current_wsgi_req = simple_current_wsgi_req;
@@ -1486,13 +1499,6 @@ uwsgi.shared->hooks[UWSGI_MODIFIER_PING] = uwsgi_request_ping; //100
uwsgi_rawlog(" ***\n");
#ifdef UWSGI_ERLANG
if (uwsgi.erlang_node) {
uwsgi.erlang_nodes = 1;
uwsgi.erlangfd = init_erlang(uwsgi.erlang_node, uwsgi.erlang_cookie);
}
#endif
#ifdef UWSGI_SNMP
if (uwsgi.snmp) {
//snmp_init();
@@ -1506,7 +1512,6 @@ uwsgi.shared->hooks[UWSGI_MODIFIER_PING] = uwsgi_request_ping; //100
}
}
/*parse xml for <app> tags */
#ifdef UWSGI_XML
if (uwsgi.xml_round2 && uwsgi.xml_config != NULL) {
@@ -1683,20 +1688,6 @@ uwsgi.shared->hooks[UWSGI_MODIFIER_PING] = uwsgi_request_ping; //100
}
}
#ifdef UWSGI_ERLANG
if (uwsgi.erlang_nodes > 0) {
if (uwsgi.numproc <= uwsgi.erlang_nodes) {
uwsgi_log("You do not have enough worker for Erlang. Please respawn with at least %d processes.\n", uwsgi.erlang_nodes + 1);
} else if (uwsgi.mywid > (uwsgi.numproc - uwsgi.erlang_nodes)) {
uwsgi_log("Erlang mode enabled for worker %d.\n", uwsgi.mywid);
erlang_loop(uwsgi.wsgi_req);
//NEVER HERE
exit(1);
}
}
close(uwsgi.erlangfd);
#endif
#ifdef UWSGI_ASYNC
uwsgi.async_running = -1;
#endif
uwsgi.h
+3 -28
View File
@@ -258,8 +258,6 @@ struct uwsgi_opt {
#define LONG_ARGS_LIMIT_AS 17009
#define LONG_ARGS_UDP 17010
#define LONG_ARGS_WSGI_FILE 17011
#define LONG_ARGS_ERLANG 17012
#define LONG_ARGS_ERLANG_COOKIE 17013
#define LONG_ARGS_BINARY_PATH 17014
#define LONG_ARGS_PROXY 17015
#define LONG_ARGS_PROXY_NODE 17016
@@ -345,11 +343,6 @@ struct uwsgi_opt {
#define UWSGI_SET_LOCKING uwsgi.workers[uwsgi.mywid].status |= UWSGI_STATUS_LOCKING
#define UWSGI_UNSET_LOCKING uwsgi.workers[uwsgi.mywid].status ^= UWSGI_STATUS_LOCKING
#define UWSGI_STATUS_ERLANGING 1 << 3
#define UWSGI_IS_ERLANGING uwsgi.workers[uwsgi.mywid].status & UWSGI_STATUS_ERLANGING
#define UWSGI_SET_ERLANGING uwsgi.workers[uwsgi.mywid].status |= UWSGI_STATUS_ERLANGING
#define UWSGI_UNSET_ERLANGING uwsgi.workers[uwsgi.mywid].status ^= UWSGI_STATUS_ERLANGING
#ifdef __linux__
#include <endian.h>
#elif __sun__
@@ -426,6 +419,7 @@ struct uwsgi_plugin {
uint8_t modifier1;
void *data;
int (*init) (void);
void (*post_init) (void);
void (*post_fork) (void);
struct option *options;
const char *short_options;
@@ -751,11 +745,6 @@ struct uwsgi_server {
char *spool_dir;
#endif
#ifdef UWSGI_ERLANG
char *erlang_node;
char *erlang_cookie;
#endif
#ifdef UWSGI_NAGIOS
int nagios;
#endif
@@ -850,11 +839,6 @@ struct uwsgi_server {
struct uwsgi_app apps[MAX_APPS];
#ifdef UWSGI_ERLANG
int erlang_nodes;
int erlangfd;
#endif
int no_orphans;
char *chdir;
@@ -959,7 +943,6 @@ struct uwsgi_lb_group {
#define KIND_WORKER 1
#define KIND_EVENT 2
#define KIND_SPOOLER 3
#define KIND_ERLANG 4
#define KIND_PROXY 5
#define KIND_MASTER 6
@@ -1173,16 +1156,6 @@ int uwsgi_parse_vars(struct wsgi_request *);
int uwsgi_enqueue_message(char *, int, uint8_t, uint8_t, char *, int, int);
#ifdef UWSGI_ERLANG
#include <erl_interface.h>
#include <ei.h>
int init_erlang(char *, char *);
void erlang_loop(struct wsgi_request *);
#endif
void manage_opt(int, char *);
void uwsgi_cluster_add_node(struct uwsgi_cluster_node *, int);
@@ -1429,6 +1402,7 @@ int uwsgi_parse_array(char *, uint16_t, char **, uint8_t *);
void log_syslog(char *);
struct uwsgi_gateway *register_gateway(char *, void (*)(void));
struct uwsgi_gateway *register_fat_gateway(char *, void (*)(void));
void gateway_respawn(int);
char *uwsgi_open_and_read(char *, int *, int, char *[]);
@@ -1465,3 +1439,4 @@ int uwsgi_simple_send_string2(char *, uint8_t, uint8_t, char *, uint16_t, char *
int uwsgi_simple_send_string(char *, uint8_t, uint8_t, char *, uint16_t, int);
int is_unix(char *, int);
int is_a_number(char *);
uwsgiconfig.py
-6
View File
@@ -420,12 +420,6 @@ class uConf(object):
self.gcc_list.append('xmlconf')
if self.get('erlang'):
self.depends_on("erlang", ['embedded'])
self.cflags.append("-DUWSGI_ERLANG")
self.libs.append(os.environ['ERLANG_LDFLAGS'])
self.gcc_list.append('erlang')
if self.get('plugin_dir'):
self.cflags.append('-DUWSGI_PLUGIN_DIR=\\"%s\\"' % self.get('plugin_dir'))