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

http fixes and new subscriber infrastructure

Browse Source
This commit is contained in:
roberto@mrspurr
2011-02-20 20:06:26 +01:00
parent ec08632a29
commit 3b67033d65
5 changed files with 495 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files
cache.c
+294
View File
@@ -2,6 +2,150 @@
extern struct uwsgi_server uwsgi;
char *uwsgi_get_subscriber(struct uwsgi_dict *udict, char *key, uint16_t keylen, uint64_t *vallen) {
uint64_t ovl;
struct uwsgi_subscriber *usub;
char *ret = NULL;
*vallen = 0;
usub = (struct uwsgi_subscriber *) uwsgi_dict_get(udict, key, keylen, &ovl);
if (usub == NULL || !ovl) return NULL;
if (!usub->nodes) return NULL;
if (usub && ovl) {
ret = usub->name[usub->current];
// dead node
if (ret[0] == 0) {
if (usub->current == usub->nodes-1) {
usub->nodes--;
}
// retry with another node (if available)
if (usub->nodes > 0) {
usub->current++;
if (usub->current >= usub->nodes) usub->current = 0;
return uwsgi_get_subscriber(udict, key, keylen, vallen);
}
}
else {
*vallen = strlen(ret);
}
if (usub->nodes > 1) {
usub->current++;
if (usub->current >= usub->nodes) usub->current = 0;
}
}
return ret;
}
void uwsgi_add_subscriber(struct uwsgi_dict *udict, char *key, uint16_t keylen, char *address, uint64_t address_len) {
char *ptr;
uint64_t vallen = 0;
struct uwsgi_subscriber *usub, nusub;
int found = 0;
int i;
ptr = uwsgi_dict_get(udict, key, keylen, &vallen);
if (ptr && vallen) {
usub = (struct uwsgi_subscriber *) ptr;
for(i=0;i<(int)usub->nodes;i++) {
if (!uwsgi_strncmp(usub->name[i], strlen(usub->name[i]), address, address_len)) {
found = 1;
break;
}
}
if (!found) {
found = usub->nodes;
// check for unallocated slot
for(i=0;i<(int)usub->nodes;i++) {
if (usub->name[i][0] == 0) {
found = i;
break;
}
}
memcpy(usub->name[found], address, address_len);
if (found == (int) usub->nodes) {
usub->nodes++;
}
}
return;
}
else {
nusub.nodes = 1;
nusub.current = 0;
memcpy(nusub.name[0], address, address_len);
uwsgi_dict_set(udict, key, keylen, (char *) &nusub, sizeof(struct uwsgi_subscriber));
}
}
struct uwsgi_dict *uwsgi_dict_create(uint64_t items, uint64_t blocksize) {
int i;
struct uwsgi_dict *udict = (struct uwsgi_dict *) mmap(NULL, sizeof(uint64_t) * UMAX16, PROT_READ | PROT_WRITE, MAP_SHARED | MAP_ANON, -1, 0);
if (!udict) {
uwsgi_error("mmap()");
exit(1);
}
if (!blocksize) blocksize = 4096;
if (blocksize % uwsgi.page_size != 0) {
uwsgi_log("invalid shared dictionary blocksize %llu: must be a multiple of memory page size (%d bytes)\n", (unsigned long long) udict->blocksize, uwsgi.page_size);
exit(1);
}
udict->blocksize = blocksize;
udict->max_items = items;
udict->hashtable = (uint64_t *) mmap(NULL, sizeof(uint64_t) * UMAX16, PROT_READ | PROT_WRITE, MAP_SHARED | MAP_ANON, -1, 0);
if (!udict->hashtable) {
uwsgi_error("mmap()");
exit(1);
}
memset(udict->hashtable, 0, sizeof(uint64_t) * UMAX16);
udict->unused_stack = (uint64_t *) mmap(NULL, sizeof(uint64_t) * udict->max_items, PROT_READ | PROT_WRITE, MAP_SHARED | MAP_ANON, -1, 0);
if (!udict->unused_stack) {
uwsgi_error("mmap()");
exit(1);
}
memset(udict->unused_stack, 0, sizeof(uint64_t) * udict->max_items);
udict->items = (struct uwsgi_dict_item *) mmap(NULL, sizeof(struct uwsgi_dict_item) * udict->max_items, PROT_READ | PROT_WRITE, MAP_SHARED | MAP_ANON, -1, 0);
if (!udict->items) {
uwsgi_error("mmap()");
exit(1);
}
udict->data = mmap(NULL, udict->blocksize * udict->max_items, PROT_READ | PROT_WRITE, MAP_SHARED | MAP_ANON, -1, 0);
if (!udict->data) {
uwsgi_error("mmap()");
exit(1);
}
for(i=0;i< (int) udict->max_items;i++) {
memset(&udict->items[i], 0, sizeof(struct uwsgi_dict_item));
}
udict->first_available_item = 1;
udict->unused_stack_ptr = 0;
udict->lock = uwsgi_mmap_shared_lock();
uwsgi_lock_init(udict->lock);
return udict;
}
uint32_t djb33x_hash(char *key, int keylen) {
register uint32_t hash = 5381;
@@ -14,6 +158,90 @@ uint32_t djb33x_hash(char *key, int keylen) {
return hash;
}
inline uint64_t uwsgi_dict_get_index(struct uwsgi_dict *udict, char *key, uint16_t keylen) {
uint32_t hash = djb33x_hash(key, keylen);
int hash_key = hash % 0xffff;
uint64_t slot = udict->hashtable[hash_key];
struct uwsgi_dict_item *udi;
udi = &udict->items[slot];
// first round
if (udi->djbhash != hash) goto cycle;
if (udi->keysize != keylen) goto cycle;
if (memcmp(udi->key, key, keylen)) goto cycle;
return slot;
cycle:
while(udi->next) {
slot = udi->next;
udi = &udict->items[slot];
if (udi->djbhash != hash) continue;
if (udi->keysize != keylen) continue;
if (!memcmp(udi->key, key, keylen)) return slot;
}
return 0;
}
char *uwsgi_dict_get(struct uwsgi_dict *udict, char *key, uint16_t keylen, uint64_t *valsize) {
uint64_t index = uwsgi_dict_get_index(udict, key, keylen);
if (index) {
*valsize = udict->items[index].valsize;
udict->items[index].hits++;
return udict->data+(index*udict->blocksize);
}
return NULL;
}
int uwsgi_dict_del(struct uwsgi_dict *udict, char *key, uint16_t keylen) {
uint64_t index = 0;
struct uwsgi_dict_item *udi;
int ret = -1;
index = uwsgi_dict_get_index(udict, key, keylen);
if (index) {
udi = &udict->items[index] ;
udi->keysize = 0;
udi->valsize = 0;
udict->unused_stack_ptr++;
udict->unused_stack[udict->unused_stack_ptr] = index;
// try to return to initial condition...
if (index == udict->first_available_item-1) {
udict->first_available_item--;
}
ret = 0;
// relink collisioned entry
if (udi->prev) {
udict->items[udi->prev].next = udi->next;
}
if (udi->next) {
udict->items[udi->next].prev = udi->prev;
}
if (!udi->prev && !udi->next) {
// reset hashtable entry
udict->hashtable[udi->djbhash % 0xffff] = 0;
}
udi->djbhash = 0;
udi->prev = 0;
udi->next = 0;
}
return ret;
}
inline uint64_t uwsgi_cache_get_index(char *key, uint16_t keylen) {
uint32_t hash = djb33x_hash(key, keylen);
@@ -176,6 +404,72 @@ end:
}
int uwsgi_dict_set(struct uwsgi_dict *udict, char *key, uint16_t keylen, char *val, uint64_t vallen) {
uint64_t index = 0, last_index = 0 ;
struct uwsgi_dict_item *udi, *udii;
int ret = -1;
int slot;
if (!keylen || !vallen) return -1;
if (keylen > UWSGI_CACHE_MAX_KEY_SIZE) return -1;
if (udict->first_available_item >= udict->max_items && !udict->unused_stack_ptr) {
uwsgi_log("*** DANGER dictionary %p is FULL !!! ***\n", udict);
goto end;
}
index = uwsgi_dict_get_index(udict, key, keylen);
if (!index) {
if (udict->unused_stack_ptr) {
index = udict->unused_stack[udict->unused_stack_ptr];
udict->unused_stack_ptr--;
}
else {
index = udict->first_available_item;
if (udict->first_available_item < udict->max_items) {
udict->first_available_item++;
}
}
udi = &udict->items[index] ;
udi->djbhash = djb33x_hash(key, keylen);
udi->hits = 0;
memcpy(udi->key, key, keylen);
memcpy(udict->data+(index*udict->blocksize), val, vallen);
// set this as late as possibile (to reduce races risk)
udi->valsize = vallen;
udi->keysize = keylen;
ret = 0;
// now put the value in the 16bit hashtable
slot = udi->djbhash % 0xffff;
if (udict->hashtable[slot] == 0) {
udict->hashtable[slot] = index;
}
else {
// append to first available next
last_index = udict->hashtable[slot];
udii = &udict->items[ last_index ];
while(udii->next) {
last_index = udii->next;
udii = &udict->items[ last_index ];
}
udii->next = index;
udi->prev = last_index;
}
}
end:
return ret;
}
void cache_command(char *key, uint16_t keylen, char *val, uint16_t vallen, void *data) {
struct wsgi_request *wsgi_req = (struct wsgi_request *) data;
master.c
+42 -1
View File
@@ -19,7 +19,6 @@ void get_linux_tcp_info(int fd) {
}
#endif
void manage_cluster_announce(char *key, uint16_t keylen, char *val, uint16_t vallen, void *data) {
char *tmpstr;
@@ -85,6 +84,9 @@ void master_loop(char **argv, char **environ) {
char *cluster_opt_buf = NULL;
int cluster_opt_size = 4;
char subscrbuf[4096];
char *ssb;
#ifdef UWSGI_MULTICAST
char *cptrbuf;
uint16_t ustrlen;
@@ -663,6 +665,45 @@ void master_loop(char **argv, char **environ) {
uwsgi_cluster_add_me();
}
// resubscribe every 10 cycles
if (uwsgi.subscriptions_cnt > 0 && (master_cycles % 10) == 0) {
for(i=0;i<uwsgi.subscriptions_cnt;i++) {
char *udp_address = strchr(uwsgi.subscriptions[i],':');
if (!udp_address) continue;
char *subscription_key = strchr(udp_address+1, ':');
udp_address = uwsgi_concat2n(uwsgi.subscriptions[i], subscription_key-uwsgi.subscriptions[i], "", 0);
uwsgi_log("subscribe to %s with key %s\n", udp_address, subscription_key+1);
ssb = subscrbuf;
ustrlen = 3;
*ssb++ = (uint8_t) (ustrlen & 0xff);
*ssb++ = (uint8_t) ((ustrlen >>8) & 0xff);
memcpy(ssb, "key", ustrlen);
ssb+=ustrlen;
ustrlen = strlen(subscription_key+1);
*ssb++ = (uint8_t) (ustrlen & 0xff);
*ssb++ = (uint8_t) ((ustrlen >>8) & 0xff);
memcpy(ssb, subscription_key+1, ustrlen);
ssb+=ustrlen;
ustrlen = 7;
*ssb++ = (uint8_t) (ustrlen & 0xff);
*ssb++ = (uint8_t) ((ustrlen >>8) & 0xff);
memcpy(ssb, "address", ustrlen);
ssb+=ustrlen;
ustrlen = strlen(uwsgi.sockets[0].name);
*ssb++ = (uint8_t) (ustrlen & 0xff);
*ssb++ = (uint8_t) ((ustrlen >>8) & 0xff);
memcpy(ssb, uwsgi.sockets[0].name, ustrlen);
ssb+=ustrlen;
uwsgi_log("sent %d bytes\n", send_udp_message(224, udp_address, subscrbuf, ssb-subscrbuf));
free(udp_address);
}
}
#endif
plugins/http/http.c
+80 -11
View File
@@ -15,10 +15,11 @@
#define MAX_HTTP_VEC 128
#define MAX_HTTP_EXTRA_VARS 64
#define LONG_ARGS_HTTP_EVENTS 300001
#define LONG_ARGS_HTTP_USE_PATTERN 300002
#define LONG_ARGS_HTTP_USE_BASE 300003
#define LONG_ARGS_HTTP_USE_TO 300004
#define LONG_ARGS_HTTP_EVENTS 300001
#define LONG_ARGS_HTTP_USE_PATTERN 300002
#define LONG_ARGS_HTTP_USE_BASE 300003
#define LONG_ARGS_HTTP_USE_TO 300004
#define LONG_ARGS_HTTP_SUBSCRIPTION_SERVER 300005
#define HTTP_STATUS_FREE 0
#define HTTP_STATUS_CONNECTING 1
@@ -30,6 +31,8 @@ struct uwsgi_http {
int use_cache;
int nevents;
char *subscription_server;
char *pattern;
int pattern_len;
@@ -44,6 +47,8 @@ struct uwsgi_http {
uint8_t modifier1;
int load;
struct uwsgi_dict *subscription_dict;
} uhttp;
struct option http_options[] = {
@@ -55,11 +60,34 @@ struct option http_options[] = {
{"http-use-pattern", required_argument, 0, LONG_ARGS_HTTP_USE_PATTERN},
{"http-use-base", required_argument, 0, LONG_ARGS_HTTP_USE_BASE},
{"http-events", required_argument, 0, LONG_ARGS_HTTP_EVENTS},
{"http-subscription-server", required_argument, 0, LONG_ARGS_HTTP_SUBSCRIPTION_SERVER},
{0, 0, 0, 0},
};
extern struct uwsgi_server uwsgi;
void http_manage_subscription(char *key, uint16_t keylen, char *val, uint16_t vallen, void *data) {
struct uwsgi_subscribe_req *usr = (struct uwsgi_subscribe_req *) data;
if (!uwsgi_strncmp("key", 3, key, keylen)) {
usr->key = val;
usr->keylen = vallen;
}
else if (!uwsgi_strncmp("auth", 4, key, keylen)) {
usr->auth = val;
usr->auth_len = vallen;
}
else if (!uwsgi_strncmp("address", 7, key, keylen)) {
usr->address = val;
usr->address_len = vallen;
}
}
struct http_session {
int fd;
@@ -129,15 +157,15 @@ uint16_t http_add_uwsgi_header(struct http_session *h_session, struct iovec *iov
if ((*c) + 4 >= MAX_HTTP_VEC) return 0;
if (uwsgi_strncmp("CONTENT_TYPE", 12, hh, keylen) && uwsgi_strncmp("CONTENT_LENGTH", 14, hh, keylen)) {
if (!uwsgi_strncmp("HOST", 4, hh, keylen)) {
h_session->hostname = val;
h_session->hostname_len = vallen;
}
else if (uwsgi_strncmp("CONTENT_TYPE", 12, hh, keylen) && uwsgi_strncmp("CONTENT_LENGTH", 14, hh, keylen)) {
keylen += 5;
prefix = 1;
if ((*c) + 5 >= MAX_HTTP_VEC) return 0;
}
else if (uwsgi_strncmp("HOST", 4, hh, keylen)) {
h_session->hostname = val;
h_session->hostname_len = vallen;
}
strsize1[0] = (uint8_t) (keylen & 0xff);
strsize1[1] = (uint8_t) ((keylen >> 8) & 0xff);
@@ -294,6 +322,7 @@ void http_loop() {
int uhttp_queue;
int uhttp_server;
int uhttp_subserver = -1;
int nevents;
int interesting_fd;
int new_connection;
@@ -318,6 +347,7 @@ void http_loop() {
struct http_session *uhttp_session;
struct http_session *uhttp_table[2048];
struct uwsgi_subscribe_req usr;
int soopt;
socklen_t solen = sizeof(int);
@@ -328,12 +358,19 @@ void http_loop() {
uhttp_server = bind_to_tcp(uhttp.socket_name, uwsgi.listen_queue, strchr(uhttp.socket_name,':'));
uhttp_queue = event_queue_init();
events = event_queue_alloc(uhttp.nevents);
event_queue_add_fd_read(uhttp_queue, uhttp_server);
if (uhttp.subscription_server) {
uhttp_subserver = bind_to_udp(uhttp.subscription_server, 0, 0);
event_queue_add_fd_read(uhttp_queue, uhttp_subserver);
uhttp.subscription_dict = uwsgi_dict_create(100, 0);
}
if (uhttp.pattern) {
init_magic_table(magic_table);
}
@@ -373,6 +410,14 @@ void http_loop() {
event_queue_add_fd_read(uhttp_queue, new_connection);
}
else if (interesting_fd == uhttp_subserver) {
len = recv(uhttp_subserver, bbuf, 4096, 0);
if (len > 0) {
memset(&usr, 0, sizeof(struct uwsgi_subscribe_req));
uwsgi_hooked_parse(bbuf+4, len-4, http_manage_subscription, &usr);
uwsgi_add_subscriber(uhttp.subscription_dict, usr.key, usr.keylen, usr.address, usr.address_len);
}
}
else {
uhttp_session = uhttp_table[interesting_fd];
@@ -431,9 +476,13 @@ void http_loop() {
}
if (uhttp.base) {
if (uhttp.use_cache) {
uhttp_session->instance_address = uwsgi_cache_get(uhttp_session->hostname, uhttp_session->hostname_len, &uhttp_session->instance_address_len);
}
else if (uhttp.base) {
uhttp_session->instance_address = uwsgi_concat2n(uhttp.base, uhttp.base_len, uhttp_session->hostname, uhttp_session->hostname_len);
uhttp_session->instance_address_len = uhttp.base_len + uhttp_session->hostname_len;
}
else if (uhttp.pattern) {
magic_table['s'] = uwsgi_concat2n(uhttp_session->hostname, uhttp_session->hostname_len, "", 0);
int tmp_addr_len = 0;
@@ -445,16 +494,27 @@ void http_loop() {
uhttp_session->instance_address = uhttp.to;
uhttp_session->instance_address_len = uhttp.to_len;
}
else if (uhttp.subscription_server) {
uhttp_session->instance_address = uwsgi_get_subscriber(uhttp.subscription_dict, uhttp_session->hostname, uhttp_session->hostname_len, &uhttp_session->instance_address_len);
}
else if (uwsgi.sockets_cnt > 0) {
uhttp_session->instance_address = uwsgi.sockets[0].name;
uhttp_session->instance_address_len = strlen(uwsgi.sockets[0].name);
}
if (!uhttp_session->instance_address_len) {
close(uhttp_session->fd);
uhttp_table[uhttp_session->fd] = NULL;
uhttp.load--;
free(uhttp_session);
break;
}
uhttp_session->pass_fd = is_unix(uhttp_session->instance_address, uhttp_session->instance_address_len);
uhttp_session->instance_fd = uwsgi_connectn(uhttp_session->instance_address, uhttp_session->instance_address_len, 0, 1);
if (uhttp.pattern) {
if (uhttp.pattern || uhttp.base ) {
free(uhttp_session->instance_address);
}
@@ -488,6 +548,9 @@ void http_loop() {
if (getsockopt(uhttp_session->instance_fd, SOL_SOCKET, SO_ERROR, (void *) (&soopt), &solen) < 0) {
uwsgi_error("getsockopt()");
if (uhttp.subscription_server) {
uhttp_session->instance_address[0] = 0;
}
close(uhttp_session->fd);
close(uhttp_session->instance_fd);
uhttp_table[uhttp_session->fd] = NULL;
@@ -499,6 +562,9 @@ void http_loop() {
if (soopt) {
uwsgi_log("unable to connect() to uwsgi instance: %s\n", strerror(soopt));
if (uhttp.subscription_server) {
uhttp_session->instance_address[0] = 0;
}
close(uhttp_session->fd);
close(uhttp_session->instance_fd);
uhttp_table[uhttp_session->fd] = NULL;
@@ -680,6 +746,9 @@ int http_opt(int i, char *optarg) {
case LONG_ARGS_HTTP:
uhttp.socket_name = optarg;
return 1;
case LONG_ARGS_HTTP_SUBSCRIPTION_SERVER:
uhttp.subscription_server = optarg;
return 1;
case LONG_ARGS_HTTP_EVENTS:
uhttp.nevents = atoi(optarg);
return 1;
uwsgi.c
+10 -1
View File
@@ -107,6 +107,7 @@ static struct option long_base_options[] = {
#endif
{"cluster-reload", required_argument, 0, LONG_ARGS_CLUSTER_RELOAD},
{"cluster-log", required_argument, 0, LONG_ARGS_CLUSTER_LOG},
{"subscribe-to", required_argument, 0, LONG_ARGS_SUBSCRIBE_TO},
#ifdef UWSGI_SNMP
{"snmp", no_argument, 0, LONG_ARGS_SNMP},
{"snmp-community", required_argument, 0, LONG_ARGS_SNMP_COMMUNITY},
@@ -1958,13 +1959,21 @@ end:
uwsgi.check_static_len = strlen(uwsgi.check_static);
return 1;
case LONG_ARGS_ATTACH_DAEMON:
if (uwsgi.startup_daemons_cnt < 63) {
if (uwsgi.startup_daemons_cnt < MAX_DAEMONS) {
uwsgi.startup_daemons[uwsgi.startup_daemons_cnt] = optarg;
uwsgi.startup_daemons_cnt++;
} else {
uwsgi_log("you can specify at most %d --attach-daemons options\n", MAX_DAEMONS);
}
return 1;
case LONG_ARGS_SUBSCRIBE_TO:
if (uwsgi.subscriptions_cnt < MAX_SUBSCRIPTIONS) {
uwsgi.subscriptions[uwsgi.subscriptions_cnt] = optarg;
uwsgi.subscriptions_cnt++;
} else {
uwsgi_log("you can specify at most %d --attach-daemons options\n", MAX_SUBSCRIPTIONS);
}
return 1;
#ifdef __linux__
case LONG_ARGS_CGROUP:
uwsgi.cgroup = optarg;
uwsgi.h
+69
View File
@@ -19,6 +19,7 @@
#define MAX_RPC 64
#define MAX_GATEWAYS 64
#define MAX_DAEMONS 8
#define MAX_SUBSCRIPTIONS 8
#ifndef UWSGI_LOAD_EMBEDDED_PLUGINS
#define UWSGI_LOAD_EMBEDDED_PLUGINS
@@ -346,6 +347,7 @@ struct uwsgi_opt {
#define LONG_ARGS_QUEUE 17075
#define LONG_ARGS_QUEUE_BLOCKSIZE 17076
#define LONG_ARGS_ATTACH_DAEMON 17077
#define LONG_ARGS_SUBSCRIBE_TO 17078
@@ -957,6 +959,9 @@ struct uwsgi_server {
char *startup_daemons[MAX_DAEMONS];
int startup_daemons_cnt;
char *subscriptions[MAX_SUBSCRIPTIONS];
int subscriptions_cnt;
};
struct uwsgi_rpc {
@@ -971,6 +976,7 @@ struct uwsgi_lb_group {
int kind;
};
#define KIND_NULL 0
#define KIND_WORKER 1
#define KIND_EVENT 2
@@ -1461,3 +1467,66 @@ char *uwsgi_resolve_ip(char *);
char *uwsgi_queue_get(uint64_t, uint64_t *);
char *uwsgi_queue_pull(uint64_t *);
int uwsgi_queue_push(char *, uint64_t);
// maintain alignment here !!!
struct uwsgi_dict_item {
// size of the value (64bit)
uint64_t valsize;
// 64bit hits
uint64_t hits;
// previous same-hash item
uint64_t prev;
// next same-hash item
uint64_t next;
// djb hash of the key
uint32_t djbhash;
// size of the key
uint16_t keysize;
// key chracters follows...
char key[UWSGI_CACHE_MAX_KEY_SIZE];
} __attribute__((__packed__));
struct uwsgi_dict {
uint64_t blocksize;
uint64_t max_items;
uint64_t *hashtable;
uint64_t *unused_stack;
uint64_t first_available_item;
uint64_t unused_stack_ptr;
void *data;
void *lock;
struct uwsgi_dict_item *items;
};
#define SUBSCRIBER_PAGESIZE 4096
#define SUBSCRIBER_NODES (SUBSCRIBER_PAGESIZE/128)-1
struct uwsgi_subscriber {
uint64_t nodes;
uint64_t current;
// support upto md5
char auth[32];
char name[128][SUBSCRIBER_NODES];
};
struct uwsgi_subscribe_req {
char *key;
uint16_t keylen;
char *address;
uint16_t address_len;
char *auth;
uint16_t auth_len;
};
struct uwsgi_dict *uwsgi_dict_create(uint64_t, uint64_t);
void uwsgi_add_subscriber(struct uwsgi_dict *, char *, uint16_t, char *, uint64_t);
char *uwsgi_dict_get(struct uwsgi_dict *, char *, uint16_t, uint64_t *);
int uwsgi_dict_set(struct uwsgi_dict *, char *, uint16_t, char *, uint64_t);
char *uwsgi_get_subscriber(struct uwsgi_dict *, char *, uint16_t, uint64_t *);