clearlinux/uwsgi
2
0
Fork 0
mirror of https://github.com/clearlinux/uwsgi.git synced 2026-09-01 11:15:57 +00:00
Code Issues Packages Projects Releases Wiki Activity

prepare dor async-friendly rpc

Browse Source
This commit is contained in:
Unbit
2013-02-07 17:23:59 +01:00
parent 7d84413038
commit 9f15345183
7 changed files with 64 additions and 152 deletions
Download Patch File Download Diff File Expand all files Collapse all files
core/init.c
+3
View File
@@ -90,6 +90,9 @@ void uwsgi_init_default() {
uwsgi.cores = 1;
uwsgi.threads = 1;
// default max number of rpc slot
uwsgi.max_rpc = 64;
uwsgi.offload_threads_events = 64;
uwsgi.default_app = -1;
core/rpc.c
+6 -6
View File
@@ -14,8 +14,8 @@ int uwsgi_register_rpc(char *name, uint8_t modifier1, uint8_t args, void *func)
uwsgi_lock(uwsgi.rpc_table_lock);
if (uwsgi.shared->rpc_count < MAX_RPC) {
urpc = &uwsgi.shared->rpc_table[uwsgi.shared->rpc_count];
if (uwsgi.shared->rpc_count < uwsgi.rpc_max) {
urpc = &uwsgi.rpc_table[uwsgi.shared->rpc_count];
memcpy(urpc->name, name, strlen(name));
urpc->modifier1 = modifier1;
@@ -36,13 +36,13 @@ int uwsgi_register_rpc(char *name, uint8_t modifier1, uint8_t args, void *func)
uint16_t uwsgi_rpc(char *name, uint8_t argc, char *argv[], uint16_t argvs[], char *output) {
struct uwsgi_rpc *urpc = NULL;
int i;
uint64_t i;
uint16_t ret = 0;
for (i = 0; i < uwsgi.shared->rpc_count; i++) {
if (uwsgi.shared->rpc_table[i].name[0] != 0) {
if (!strcmp(uwsgi.shared->rpc_table[i].name, name)) {
urpc = &uwsgi.shared->rpc_table[i];
if (uwsgi.rpc_table[i].name[0] != 0) {
if (!strcmp(uwsgi.rpc_table[i].name, name)) {
urpc = &uwsgi.rpc_table[i];
break;
}
}
plugins/lua/lua_plugin.c
+32 -134
View File
@@ -14,7 +14,7 @@ struct uwsgi_lua {
#define lca(L, n) ulua_check_args(L, __FUNCTION__, n)
struct uwsgi_option uwsgi_lua_options[] = {
static struct uwsgi_option uwsgi_lua_options[] = {
{"lua", required_argument, 0, "load lua wsapi app", uwsgi_opt_set_str, &ulua.filename, 0},
@@ -24,7 +24,7 @@ struct uwsgi_option uwsgi_lua_options[] = {
static void ulua_check_args(lua_State *L, const char *func, int n) {
int args = lua_gettop(L);
char error[4096];
char error[1024];
if (args != n) {
if (n == 1) {
snprintf(error, 4096, "uwsgi.%s takes 1 parameter", func+10);
@@ -82,53 +82,6 @@ static int uwsgi_api_register_rpc(lua_State *L) {
return 1;
}
static char *encode_lua_table(lua_State *L, int index, uint16_t *size) {
char *buf, *ptrbuf;
char *key;
char *value;
size_t keylen;
size_t vallen;
*size = 0;
lua_pushnil(L);
while (lua_next(L, index) != 0) {
if (lua_isstring(L, -2) && lua_isstring(L, -1)) {
key = (char *) lua_tolstring(L, -2, &keylen);
value = (char *) lua_tolstring(L, -1, &vallen);
if (keylen > 0xffff || vallen > 0xffff) continue;
*size += (2+keylen+2+vallen);
}
lua_pop(L, 1);
}
buf = uwsgi_malloc(*size);
ptrbuf = buf;
lua_pushnil(L);
while (lua_next(L, index) != 0) {
if (lua_isstring(L, -2) && lua_isstring(L, -1)) {
key = (char *) lua_tolstring(L, -2, &keylen);
value = (char *) lua_tolstring(L, -1, &vallen);
if (keylen > 0xffff || vallen > 0xffff) continue;
*ptrbuf++ = (uint8_t) (keylen & 0xff);
*ptrbuf++ = (uint8_t) ((keylen >>8) & 0xff);
memcpy(ptrbuf, key, keylen); ptrbuf += keylen;
*ptrbuf++ = (uint8_t) (vallen & 0xff);
*ptrbuf++ = (uint8_t) ((vallen >>8) & 0xff);
memcpy(ptrbuf, value, vallen); ptrbuf += vallen;
}
lua_pop(L, 1);
}
return buf;
}
static int uwsgi_api_cache_set(lua_State *L) {
int args = lua_gettop(L);
@@ -137,6 +90,8 @@ static int uwsgi_api_cache_set(lua_State *L) {
uint64_t expires = 0;
size_t vallen;
if (!uwsgi.caches) goto error;
if (args > 1) {
@@ -146,10 +101,13 @@ static int uwsgi_api_cache_set(lua_State *L) {
expires = lua_tonumber(L, 3);
}
uwsgi_wlock(uwsgi.caches->lock);
uwsgi_cache_set((char *)key, strlen(key), (char *)value, (uint16_t) vallen, expires, 0);
uwsgi_rwunlock(uwsgi.caches->lock);
}
error:
lua_pushnil(L);
return 1;
@@ -185,94 +143,30 @@ static int uwsgi_api_cache_get(lua_State *L) {
lca(L, 1);
if (!uwsgi.caches) goto error;
if (lua_isstring(L, 1)) {
// get the key
key = lua_tolstring(L, 1, NULL);
uwsgi_rlock(uwsgi.caches->lock);
value = uwsgi_cache_get((char *)key, strlen(key), &valsize);
if (value) {
lua_pushlstring(L, value, valsize);
uwsgi_rwunlock(uwsgi.caches->lock);
return 1;
}
uwsgi_rwunlock(uwsgi.caches->lock);
}
error:
lua_pushnil(L);
return 1;
}
static int uwsgi_api_send_message(lua_State *L) {
int args = lua_gettop(L);
const char *host;
int uwsgi_fd;
uint8_t modifier1, modifier2;
char *pkt = NULL;
uint16_t pktsize = 0 ;
char buf[4096];
int rlen;
int items = 0;
int input_fd = -1, timeout = -1, input_size = 0;
// is this an fd ?
if (lua_isnumber(L, 1)) {
args = 1;
}
else if (lua_isstring(L, 1)) {
host = lua_tolstring(L, 1, NULL);
uwsgi_fd = uwsgi_connect((char *)host, timeout, 0);
modifier1 = lua_tonumber(L, 2);
modifier2 = lua_tonumber(L, 3);
if (args > 4) {
timeout = lua_tonumber(L, 5);
if (args == 7) {
input_fd = lua_tonumber(L, 6);
input_size = lua_tonumber(L, 7);
}
}
if (lua_istable(L,4)) {
// passed a table
pkt = encode_lua_table(L, 4, &pktsize);
}
if (uwsgi_send_message(uwsgi_fd, modifier1, modifier2, pkt, pktsize, input_fd, input_size, timeout) == -1) {
free(pkt);
lua_pushnil(L);
return 1;
}
free(pkt);
for(;;) {
rlen = uwsgi_waitfd(uwsgi_fd, timeout);
if (rlen > 0) {
rlen = read(uwsgi_fd, buf, 4096);
if (rlen < 0) {
uwsgi_error("read()");
break;
}
else if (rlen > 0) {
lua_pushlstring(L, buf, rlen);
items++;
}
else {
break;
}
}
else if (rlen == 0) {
uwsgi_log("uwsgi request timed out waiting for response\n");
break;
}
}
close(uwsgi_fd);
}
return items;
}
static int uwsgi_api_cl(lua_State *L) {
struct wsgi_request *wsgi_req = current_wsgi_req();
@@ -338,9 +232,7 @@ static int uwsgi_api_unlock(lua_State *L) {
static const luaL_reg uwsgi_api[] = {
{"log", uwsgi_api_log},
{"cl", uwsgi_api_cl},
{"req_fd", uwsgi_api_req_fd},
{"send_message", uwsgi_api_send_message},
{"connection_fd", uwsgi_api_req_fd},
{"cache_get", uwsgi_api_cache_get},
{"cache_set", uwsgi_api_cache_set},
{"register_signal", uwsgi_api_register_signal},
@@ -374,9 +266,9 @@ static int uwsgi_lua_input(lua_State *L) {
return 0;
}
int uwsgi_lua_init(){
static int uwsgi_lua_init(){
uwsgi_log("Initializing Lua environment... (%d cores)\n", uwsgi.cores);
uwsgi_log("Initializing Lua environment... (%d lua_States)\n", uwsgi.cores);
ulua.L = uwsgi_malloc( sizeof(lua_State*) * uwsgi.cores );
@@ -386,7 +278,7 @@ int uwsgi_lua_init(){
}
void uwsgi_lua_app() {
static void uwsgi_lua_app() {
int i;
if (ulua.filename) {
@@ -423,7 +315,7 @@ void uwsgi_lua_app() {
}
}
int uwsgi_lua_request(struct wsgi_request *wsgi_req) {
static int uwsgi_lua_request(struct wsgi_request *wsgi_req) {
int i;
const char *http, *http2;
@@ -538,13 +430,13 @@ clear:
}
void uwsgi_lua_after_request(struct wsgi_request *wsgi_req) {
static void uwsgi_lua_after_request(struct wsgi_request *wsgi_req) {
log_request(wsgi_req);
}
int uwsgi_lua_magic(char *mountpoint, char *lazy) {
static int uwsgi_lua_magic(char *mountpoint, char *lazy) {
if (!strcmp(lazy+strlen(lazy)-4, ".lua")) {
ulua.filename = lazy;
@@ -559,7 +451,7 @@ int uwsgi_lua_magic(char *mountpoint, char *lazy) {
return 0;
}
char *uwsgi_lua_code_string(char *id, char *code, char *func, char *key, uint16_t keylen) {
static char *uwsgi_lua_code_string(char *id, char *code, char *func, char *key, uint16_t keylen) {
static struct lua_State *L = NULL;
@@ -607,13 +499,15 @@ char *uwsgi_lua_code_string(char *id, char *code, char *func, char *key, uint16_
return NULL;
}
int uwsgi_lua_signal_handler(uint8_t sig, void *handler) {
static int uwsgi_lua_signal_handler(uint8_t sig, void *handler) {
struct wsgi_request *wsgi_req = current_wsgi_req();
lua_State *L = ulua.L[wsgi_req->async_id];
#ifdef UWSGI_DEBUG
uwsgi_log("managing signal handler on core %d\n", wsgi_req->async_id);
#endif
lua_rawgeti(L, LUA_REGISTRYINDEX, (long) handler);
@@ -631,7 +525,7 @@ int uwsgi_lua_signal_handler(uint8_t sig, void *handler) {
}
uint16_t uwsgi_lua_rpc(void * func, uint8_t argc, char **argv, uint16_t argvs[], char *buffer) {
static uint16_t uwsgi_lua_rpc(void * func, uint8_t argc, char **argv, uint16_t argvs[], char *buffer) {
uint8_t i;
const char *sv;
@@ -643,7 +537,9 @@ uint16_t uwsgi_lua_rpc(void * func, uint8_t argc, char **argv, uint16_t argvs[],
lua_State *L = ulua.L[wsgi_req->async_id];
#ifdef UWSGI_DEBUG
uwsgi_log("get function %d\n", ifunc);
#endif
lua_rawgeti(L, LUA_REGISTRYINDEX, ifunc);
for(i=0;i<argc;i++) {
@@ -658,7 +554,9 @@ uint16_t uwsgi_lua_rpc(void * func, uint8_t argc, char **argv, uint16_t argvs[],
sv = lua_tolstring(L, -1, &sl);
#ifdef UWSGI_DEBUG
uwsgi_log("sv = %s sl = %lu\n", sv, (unsigned long) sl);
#endif
if (sl <= 0xffff) {
memcpy(buffer, sv, sl);
return sl;
plugins/psgi/uwsgi_plmodule.c
+2 -2
View File
@@ -100,7 +100,7 @@ XS(XS_cache_set) {
STRLEN keylen;
STRLEN vallen;
if (uwsgi.cache_max_items == 0) goto clear;
if (!uwsgi.caches) goto clear;
psgi_check_args(2);
@@ -122,7 +122,7 @@ XS(XS_cache_get) {
STRLEN keylen;
uint64_t vallen;
if (uwsgi.cache_max_items == 0) goto clear;
if (!uwsgi.caches) goto clear;
psgi_check_args(1);
plugins/python/uwsgi_pymodule.c
+3 -3
View File
@@ -334,12 +334,12 @@ PyObject *py_uwsgi_call(PyObject * self, PyObject * args) {
PyObject *py_uwsgi_rpc_list(PyObject * self, PyObject * args) {
int i;
uint64_t i;
PyObject *rpc_list = PyTuple_New(uwsgi.shared->rpc_count);
for (i = 0; i < uwsgi.shared->rpc_count; i++) {
if (uwsgi.shared->rpc_table[i].name[0] != 0) {
PyTuple_SetItem(rpc_list, i, PyString_FromString(uwsgi.shared->rpc_table[i].name));
if (uwsgi.rpc_table[i].name[0] != 0) {
PyTuple_SetItem(rpc_list, i, PyString_FromString(uwsgi.rpc_table[i].name));
}
}
plugins/rpc/rpc_plugin.c
+10 -2
View File
@@ -5,13 +5,16 @@ extern struct uwsgi_server uwsgi;
int uwsgi_rpc_request(struct wsgi_request *wsgi_req) {
// this is the list of args
char *argv[256];
// this is the size of each argument
uint16_t argvs[256];
// maximum number of supported arguments
uint8_t argc = 0xff;
/* Standard RPC request */
if (!wsgi_req->uh->pktsize) {
uwsgi_log("Invalid RPC request. skip.\n");
uwsgi_log("Empty RPC request. skip.\n");
return -1;
}
@@ -34,11 +37,16 @@ int uwsgi_rpc_request(struct wsgi_request *wsgi_req) {
uwsgi_log("RPC args %d\n", argc-1);
#endif
// call the function (output will be in wsgi_req->buffer)
wsgi_req->uh->pktsize = uwsgi_rpc(argv[0], argc-1, argv+1, argvs+1, wsgi_req->buffer);
// using modifier1 we may want a raw output
if (wsgi_req->uh->modifier2 == 0) {
uwsgi_response_write_body_do(wsgi_req, (char *) wsgi_req->uh, 4);
if (uwsgi_response_write_body_do(wsgi_req, (char *) wsgi_req->uh, 4)) {
return -1;
}
}
// write the response
uwsgi_response_write_body_do(wsgi_req, wsgi_req->buffer, wsgi_req->uh->pktsize);
return 0;
uwsgi.h
+8 -5
View File
@@ -2122,6 +2122,10 @@ struct uwsgi_server {
struct uwsgi_lock_item *rpc_table_lock;
struct uwsgi_lock_item *sa_lock;
// rpc
uint64_t rpc_max;
struct uwsgi_rpc *rpc_table;
// subscription client
int subscribe_freq;
int subscription_tolerance;
@@ -2258,8 +2262,7 @@ struct uwsgi_server {
struct uwsgi_signal_rb_timer rb_timers[MAX_TIMERS];
int rb_timers_cnt;
struct uwsgi_rpc rpc_table[MAX_RPC];
int rpc_count;
uint64_t rpc_count;
int worker_log_pipe[2];
// used for request logging
@@ -2669,9 +2672,9 @@ void uwsgi_cache_start_sync_servers(void);
int uwsgi_start(void *);
int uwsgi_register_rpc(char *, uint8_t, uint8_t, void *);
uint16_t uwsgi_rpc(char *, uint8_t, char **, uint16_t *, char *);
char *uwsgi_do_rpc(char *, char *, uint8_t, char **, uint16_t *, uint16_t *);
int uwsgi_register_rpc(char *, uint8_t, uint8_t, void *);
uint16_t uwsgi_rpc(char *, uint8_t, char **, uint16_t *, char *);
char *uwsgi_do_rpc(char *, char *, uint8_t, char **, uint16_t *, uint16_t *);
char *uwsgi_cheap_string(char *, int);