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

greatly improved subscription system

Browse Source
This commit is contained in:
roberto@debian32
2011-11-21 16:23:24 +01:00
parent 317df5d06b
commit ffcef81dd0
7 changed files with 242 additions and 140 deletions
Download Patch File Download Diff File Expand all files Collapse all files
master.c
+6 -55
View File
@@ -67,56 +67,7 @@ void expire_rb_timeouts(struct rb_root *root) {
}
}
void uwsgi_send_subscription(char *udp_address, char *key, size_t keysize, char *modifier1, size_t modifier1_len) {
char *ssb;
char subscrbuf[4096];
uint16_t ustrlen;
ssb = subscrbuf;
ustrlen = 3;
*ssb++ = (uint8_t) (ustrlen & 0xff);
*ssb++ = (uint8_t) ((ustrlen >>8) & 0xff);
memcpy(ssb, "key", ustrlen);
ssb+=ustrlen;
ustrlen = keysize;
*ssb++ = (uint8_t) (ustrlen & 0xff);
*ssb++ = (uint8_t) ((ustrlen >>8) & 0xff);
memcpy(ssb, key, 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->name);
*ssb++ = (uint8_t) (ustrlen & 0xff);
*ssb++ = (uint8_t) ((ustrlen >>8) & 0xff);
memcpy(ssb, uwsgi.sockets->name, ustrlen);
ssb+=ustrlen;
if (modifier1) {
ustrlen = 9;
*ssb++ = (uint8_t) (ustrlen & 0xff);
*ssb++ = (uint8_t) ((ustrlen >>8) & 0xff);
memcpy(ssb, "modifier1", ustrlen);
ssb+=ustrlen;
ustrlen = modifier1_len;
*ssb++ = (uint8_t) (ustrlen & 0xff);
*ssb++ = (uint8_t) ((ustrlen >>8) & 0xff);
memcpy(ssb, modifier1, ustrlen);
ssb+=ustrlen;
}
send_udp_message(224, udp_address, subscrbuf, ssb-subscrbuf);
}
void uwsgi_subscribe(char *subscription) {
void uwsgi_subscribe(char *subscription, uint8_t cmd) {
int subfile_size;
int i;
@@ -149,7 +100,7 @@ void uwsgi_subscribe(char *subscription) {
modifier1_len = strlen(modifier1);
keysize = strlen(key);
}
uwsgi_send_subscription(udp_address, key, keysize, modifier1, modifier1_len);
uwsgi_send_subscription(udp_address, key, keysize, modifier1, modifier1_len, cmd);
modifier1 = NULL;
modifier1_len = 0;
}
@@ -167,7 +118,7 @@ void uwsgi_subscribe(char *subscription) {
modifier1_len = strlen(modifier1);
keysize = strlen(key);
}
uwsgi_send_subscription(udp_address, key, keysize, modifier1, modifier1_len);
uwsgi_send_subscription(udp_address, key, keysize, modifier1, modifier1_len, cmd);
modifier1 = NULL;
modifier1_len = 0;
lines[i] = '\n';
@@ -191,7 +142,7 @@ void uwsgi_subscribe(char *subscription) {
modifier1_len = strlen(modifier1);
}
uwsgi_send_subscription(udp_address, subscription_key+1, strlen(subscription_key+1), modifier1, modifier1_len);
uwsgi_send_subscription(udp_address, subscription_key+1, strlen(subscription_key+1), modifier1, modifier1_len, cmd);
if (modifier1)
modifier1[-1] = ',';
}
@@ -506,7 +457,7 @@ int master_loop(char **argv, char **environ) {
// first subscription
struct uwsgi_string_list *subscriptions = uwsgi.subscriptions;
while(subscriptions) {
uwsgi_subscribe(subscriptions->value);
uwsgi_subscribe(subscriptions->value, 0);
subscriptions = subscriptions->next;
}
@@ -1424,7 +1375,7 @@ healthy:
if (uwsgi.subscriptions && ((uwsgi.master_cycles % 10) == 0 || uwsgi.master_cycles == 1)) {
struct uwsgi_string_list *subscriptions = uwsgi.subscriptions;
while(subscriptions) {
uwsgi_subscribe(subscriptions->value);
uwsgi_subscribe(subscriptions->value, 0);
subscriptions = subscriptions->next;
}
}
plugins/fastrouter/fastrouter.c
+59 -56
View File
@@ -73,6 +73,17 @@ struct uwsgi_fastrouter {
int i_am_cheap;
} ufr;
static void fastrouter_go_cheap(void) {
uwsgi_log("[uwsgi-fastrouter] no more nodes available. Going cheap...\n");
struct uwsgi_fastrouter_socket *ufr_sock = ufr.sockets;
while(ufr_sock) {
event_queue_del_fd(ufr.queue, ufr_sock->fd, event_queue_read());
ufr_sock = ufr_sock->next;
}
ufr.i_am_cheap = 1;
}
static struct uwsgi_fastrouter_socket *uwsgi_fastrouter_new_socket(char *name) {
@@ -165,41 +176,28 @@ struct fastrouter_session {
static void close_session(struct fastrouter_session **fr_table, struct fastrouter_session *fr_session) {
// check timeout expired
if (fr_session == NULL) {
if (ufr.subscription_server) {
//time_t current_time = time(NULL);
uwsgi_log("checking for node health\n");
struct uwsgi_subscribe_slot *slot = ufr.subscriptions;
while(slot) {
struct uwsgi_subscribe_node *node = slot->nodes;
while(node) {
uwsgi_log("%.*s (hits: %llu) %.*s\n", slot->keylen, slot->key, slot->hits, node->len, node->name);
node = node->next;
}
slot = slot->next;
}
del_check_timeout(ufr.subscriptions_check);
ufr.subscriptions_check = add_check_timeout(10);
}
return;
}
close(fr_session->fd);
fr_table[fr_session->fd] = NULL;
if (fr_session->instance_fd != -1) {
if (ufr.subscriptions && (fr_session->instance_failed || fr_session->status == FASTROUTER_STATUS_CONNECTING)) {
if (fr_session->un && fr_session->un->len > 0) {
uwsgi_log("[uwsgi-fastrouter] %.*s => marking %.*s as failed\n", (int) fr_session->hostname_len, fr_session->hostname, (int) fr_session->instance_address_len,fr_session->instance_address);
uwsgi_remove_subscribe_node(&ufr.subscriptions, fr_session->un);
if (ufr.subscriptions && fr_session->un && fr_session->un->len > 0) {
// decrease reference count
#ifdef UWSGI_DEBUG
uwsgi_log("[1] node %.*s refcnt: %llu\n", fr_session->un->len, fr_session->un->name, fr_session->un->reference);
#endif
fr_session->un->reference--;
#ifdef UWSGI_DEBUG
uwsgi_log("[2] node %.*s refcnt: %llu\n", fr_session->un->len, fr_session->un->name, fr_session->un->reference);
#endif
if (fr_session->instance_failed || fr_session->status == FASTROUTER_STATUS_CONNECTING) {
if (fr_session->un->death_mark == 0)
uwsgi_log("[uwsgi-fastrouter] %.*s => marking %.*s as failed\n", (int) fr_session->hostname_len, fr_session->hostname, (int) fr_session->instance_address_len,fr_session->instance_address);
fr_session->un->death_mark = 1;
// check if i can remove the node
if (fr_session->un->reference == 0) {
uwsgi_remove_subscribe_node(&ufr.subscriptions, fr_session->un);
}
if (ufr.subscriptions == NULL && ufr.cheap && !ufr.i_am_cheap) {
uwsgi_log("[uwsgi-fastrouter] no more nodes available. Going cheap...\n");
struct uwsgi_fastrouter_socket *ufr_sock = ufr.sockets;
while(ufr_sock) {
event_queue_del_fd(ufr.queue, ufr_sock->fd, event_queue_read());
ufr_sock = ufr_sock->next;
}
ufr.i_am_cheap = 1;
fastrouter_go_cheap();
}
}
@@ -437,15 +435,34 @@ void fastrouter_loop() {
if (len > 0) {
memset(&usr, 0, sizeof(struct uwsgi_subscribe_req));
uwsgi_hooked_parse(bbuf+4, len-4, fastrouter_manage_subscription, &usr);
if (uwsgi_add_subscribe_node(&ufr.subscriptions, &usr, ufr.subscription_regexp) && ufr.i_am_cheap) {
struct uwsgi_fastrouter_socket *ufr_sock = ufr.sockets;
while(ufr_sock) {
event_queue_add_fd_read(ufr.queue, ufr_sock->fd);
ufr_sock = ufr_sock->next;
}
ufr.i_am_cheap = 0;
uwsgi_log("[uwsgi-fastrouter] leaving cheap mode...\n");
// subscribe request ?
if (bbuf[3] == 0) {
if (uwsgi_add_subscribe_node(&ufr.subscriptions, &usr, ufr.subscription_regexp) && ufr.i_am_cheap) {
struct uwsgi_fastrouter_socket *ufr_sock = ufr.sockets;
while(ufr_sock) {
event_queue_add_fd_read(ufr.queue, ufr_sock->fd);
ufr_sock = ufr_sock->next;
}
ufr.i_am_cheap = 0;
uwsgi_log("[uwsgi-fastrouter] leaving cheap mode...\n");
}
}
//unsubscribe
else {
struct uwsgi_subscribe_node *node = uwsgi_get_subscribe_node_by_name(&ufr.subscriptions, usr.key, usr.keylen, usr.address, usr.address_len, ufr.subscription_regexp);
if (node && node->len) {
if (node->death_mark == 0)
uwsgi_log("[uwsgi-fastrouter] %.*s => marking %.*s as failed\n", (int) usr.keylen, usr.key, (int) usr.address_len, usr.address);
node->death_mark = 1;
// check if i can remove the node
if (node->reference == 0) {
uwsgi_remove_subscribe_node(&ufr.subscriptions, node);
}
if (ufr.subscriptions == NULL && ufr.cheap && !ufr.i_am_cheap) {
fastrouter_go_cheap();
}
}
}
}
}
else {
@@ -524,6 +541,9 @@ void fastrouter_loop() {
fr_session->instance_address_len = fr_session->un->len;
fr_session->modifier1 = fr_session->un->modifier1;
}
else if (ufr.subscriptions == NULL && ufr.cheap && !ufr.i_am_cheap) {
fastrouter_go_cheap();
}
}
else if (ufr.base) {
tmp_socket_name = uwsgi_concat2nn(ufr.base, ufr.base_len, fr_session->hostname, fr_session->hostname_len, &tmp_socket_name_len);
@@ -558,23 +578,6 @@ void fastrouter_loop() {
if (tmp_socket_name) free(tmp_socket_name);
if (fr_session->instance_fd < 0) {
/*
if (ufr.subscription_server) {
if (fr_session->un && fr_session->un->len > 0) {
uwsgi_log("[uwsgi-fastrouter] %.*s => marking %.*s as failed\n", (int) fr_session->hostname_len, fr_session->hostname, (int) fr_session->instance_address_len,fr_session->instance_address);
uwsgi_remove_subscribe_node(&ufr.subscriptions, fr_session->un);
if (ufr.subscriptions == NULL && ufr.cheap && !ufr.i_am_cheap) {
uwsgi_log("[uwsgi-fastrouter] no more nodes available. Going cheap...\n");
struct uwsgi_fastrouter_socket *ufr_sock = ufr.sockets;
while(ufr_sock) {
event_queue_del_fd(ufr.queue, ufr_sock->fd, event_queue_read());
ufr_sock = ufr_sock->next;
}
ufr.i_am_cheap = 1;
}
}
}
*/
fr_session->instance_failed = 1;
close_session(fr_table, fr_session);
break;
plugins/python/uwsgi_pymodule.c
+7 -2
View File
@@ -1942,13 +1942,18 @@ PyObject *py_uwsgi_load_plugin(PyObject * self, PyObject * args) {
PyObject *py_uwsgi_multicast(PyObject * self, PyObject * args) {
char *host, *message;
Py_ssize_t message_len;
ssize_t ret;
char *uwsgi_message;
if (!PyArg_ParseTuple(args, "ss:send_multicast_message", &host, &message)) {
if (!PyArg_ParseTuple(args, "ss#:send_multicast_message", &host, &message, &message_len)) {
return NULL;
}
ret = send_udp_message(UWSGI_MODIFIER_MULTICAST, host, message, strlen(message));
uwsgi_message = uwsgi_malloc(message_len+4);
memcpy(uwsgi_message+4, message, message_len);
ret = send_udp_message(UWSGI_MODIFIER_MULTICAST, 0, host, uwsgi_message, message_len);
free(uwsgi_message);
if (ret <= 0) {
Py_INCREF(Py_None);
protocol.c
+19 -17
View File
@@ -123,16 +123,21 @@ time_t parse_http_date(char *date, uint16_t len) {
}
#ifdef UWSGI_UDP
ssize_t send_udp_message(uint8_t modifier1, char *host, char *message, uint16_t message_size) {
ssize_t send_udp_message(uint8_t modifier1, uint8_t modifier2, char *host, char *message, uint16_t message_size) {
int fd;
struct sockaddr_in udp_addr;
char *udp_port;
ssize_t ret;
char udpbuff[1024];
if (message_size + 4 > 1024)
return -1;
struct uwsgi_header *uh;
if (message) {
uh = (struct uwsgi_header *) message;
}
else {
uh = (struct uwsgi_header *) uwsgi_malloc(4);
}
udp_port = strchr(host, ':');
if (udp_port == NULL) {
@@ -152,22 +157,15 @@ ssize_t send_udp_message(uint8_t modifier1, char *host, char *message, uint16_t
udp_addr.sin_port = htons(atoi(udp_port+1));
udp_addr.sin_addr.s_addr = inet_addr(host);
udpbuff[0] = modifier1;
uh->modifier1 = modifier1;
#ifdef __BIG_ENDIAN__
message_size = uwsgi_swap16(message_size);
uh->pktsize = uwsgi_swap16(message_size);
#else
uh->pktsize = message_size;
#endif
uh->modifier2 = modifier2;
memcpy(udpbuff+1, &message_size, 2);
udpbuff[3] = 0;
#ifdef __BIG_ENDIAN__
message_size = uwsgi_swap16(message_size);
#endif
memcpy(udpbuff+4, message, message_size);
ret = sendto(fd, udpbuff, message_size+4, 0, (struct sockaddr *) &udp_addr, sizeof(udp_addr));
ret = sendto(fd, (char *) uh, message_size+4, 0, (struct sockaddr *) &udp_addr, sizeof(udp_addr));
if (ret < 0) {
uwsgi_error("sendto()");
}
@@ -175,6 +173,10 @@ ssize_t send_udp_message(uint8_t modifier1, char *host, char *message, uint16_t
udp_port[0] = ':';
if ((char *)uh != message) {
free(uh);
}
return ret;
}
subscription.c
+109 -2
View File
@@ -14,8 +14,12 @@
This system is not mean to run on shared memory. If you have multiple processes for the same app, you have to create
a new subscriptions slot list.
To avoid removal of already using nodes, a reference count system has been implemented
*/
extern struct uwsgi_server uwsgi;
struct uwsgi_subscribe_slot *uwsgi_get_subscribe_slot(struct uwsgi_subscribe_slot **slot, char *key, uint16_t keylen, int regexp) {
struct uwsgi_subscribe_slot *current_slot = *slot;
@@ -72,23 +76,61 @@ struct uwsgi_subscribe_node *uwsgi_get_subscribe_node(struct uwsgi_subscribe_slo
if (current_slot) {
// node found, move up in the list increasing hits
current_slot->hits++;
time_t current = time(NULL);
struct uwsgi_subscribe_node *node = current_slot->nodes;
while(node) {
while(current_slot && node) {
// is the node alive ?
if (current - node->last_check > 10) {
node->death_mark = 1;
}
if (node->death_mark && node->reference == 0) {
// remove the node and move to next
struct uwsgi_subscribe_node *dead_node = node;
node = node->next;
// if the slot has been removed, return NULL;
if (uwsgi_remove_subscribe_node(slot, dead_node) == 1) {
return NULL;
}
continue;
}
if (rr_pos == current_slot->rr) {
current_slot->rr++;
node->reference++;
return node;
}
node = node->next;
rr_pos++;
}
current_slot->rr = 0;
if (current_slot->nodes) {
current_slot->nodes->reference++;
}
return current_slot->nodes;
}
return NULL;
}
void uwsgi_remove_subscribe_node(struct uwsgi_subscribe_slot **slot, struct uwsgi_subscribe_node *node) {
struct uwsgi_subscribe_node *uwsgi_get_subscribe_node_by_name(struct uwsgi_subscribe_slot **slot, char *key, uint16_t keylen, char *val, uint16_t vallen, int regexp) {
if (keylen > 0xff) return NULL;
struct uwsgi_subscribe_slot *current_slot = uwsgi_get_subscribe_slot(slot, key, keylen, regexp);
if (current_slot) {
struct uwsgi_subscribe_node *node = current_slot->nodes;
while(node) {
if (!uwsgi_strncmp(val, vallen, node->name, node->len)) {
return node;
}
node = node->next;
}
}
return NULL;
}
int uwsgi_remove_subscribe_node(struct uwsgi_subscribe_slot **slot, struct uwsgi_subscribe_node *node) {
int ret = 0;
struct uwsgi_subscribe_node *a_node;
struct uwsgi_subscribe_slot *node_slot = node->slot;
@@ -132,12 +174,15 @@ void uwsgi_remove_subscribe_node(struct uwsgi_subscribe_slot **slot, struct uwsg
}
#endif
ret = 1;
free(node_slot);
// am i the only slot ?
if (!prev_slot && !next_slot) {
*slot = NULL;
}
}
return ret;
}
struct uwsgi_subscribe_node *uwsgi_add_subscribe_node(struct uwsgi_subscribe_slot **slot, struct uwsgi_subscribe_req *usr, int regexp) {
@@ -151,6 +196,8 @@ struct uwsgi_subscribe_node *uwsgi_add_subscribe_node(struct uwsgi_subscribe_slo
node = current_slot->nodes;
while(node) {
if (!uwsgi_strncmp(node->name, node->len, usr->address, usr->address_len)) {
// remove death mark
node->death_mark = 0;
node->last_check = time(NULL);
return node;
}
@@ -162,6 +209,8 @@ struct uwsgi_subscribe_node *uwsgi_add_subscribe_node(struct uwsgi_subscribe_slo
node->len = usr->address_len;
node->modifier1 = usr->modifier1;
node->modifier2 = usr->modifier2;
node->reference = 0;
node->death_mark = 0;
node->last_check = time(NULL);
node->slot = current_slot;
memcpy(node->name, usr->address, usr->address_len);
@@ -195,6 +244,8 @@ struct uwsgi_subscribe_node *uwsgi_add_subscribe_node(struct uwsgi_subscribe_slo
current_slot->nodes = uwsgi_malloc(sizeof(struct uwsgi_subscribe_node));
current_slot->nodes->slot = current_slot;
current_slot->nodes->len = usr->address_len;
current_slot->nodes->reference = 0;
current_slot->nodes->death_mark = 0;
current_slot->nodes->modifier1 = usr->modifier1;
current_slot->nodes->modifier2 = usr->modifier2;
memcpy(current_slot->nodes->name, usr->address, usr->address_len);
@@ -272,3 +323,59 @@ struct uwsgi_subscribe_node *uwsgi_add_subscribe_node(struct uwsgi_subscribe_slo
}
void uwsgi_send_subscription(char *udp_address, char *key, size_t keysize, char *modifier1, size_t modifier1_len, uint8_t cmd) {
size_t ssb_size = 4 + (2 + 3) + (2 + keysize) + (2 + 7) + (2 + strlen(uwsgi.sockets->name));
if (modifier1) {
ssb_size += (2 + 9) + (2 + modifier1_len);
}
char *subscrbuf = uwsgi_malloc(ssb_size);
// leave space for uwsgi header
char *ssb = subscrbuf+4;
// key = "domain"
uint16_t ustrlen = 3;
*ssb++ = (uint8_t) (ustrlen & 0xff);
*ssb++ = (uint8_t) ((ustrlen >>8) & 0xff);
memcpy(ssb, "key", ustrlen);
ssb+=ustrlen;
ustrlen = keysize;
*ssb++ = (uint8_t) (ustrlen & 0xff);
*ssb++ = (uint8_t) ((ustrlen >>8) & 0xff);
memcpy(ssb, key, ustrlen);
ssb+=ustrlen;
// address = "first uwsgi socket"
ustrlen = 7;
*ssb++ = (uint8_t) (ustrlen & 0xff);
*ssb++ = (uint8_t) ((ustrlen >>8) & 0xff);
memcpy(ssb, "address", ustrlen);
ssb+=ustrlen;
ustrlen = strlen(uwsgi.sockets->name);
*ssb++ = (uint8_t) (ustrlen & 0xff);
*ssb++ = (uint8_t) ((ustrlen >>8) & 0xff);
memcpy(ssb, uwsgi.sockets->name, ustrlen);
ssb+=ustrlen;
// modifier1 = "modifier1"
if (modifier1) {
ustrlen = 9;
*ssb++ = (uint8_t) (ustrlen & 0xff);
*ssb++ = (uint8_t) ((ustrlen >>8) & 0xff);
memcpy(ssb, "modifier1", ustrlen);
ssb+=ustrlen;
ustrlen = modifier1_len;
*ssb++ = (uint8_t) (ustrlen & 0xff);
*ssb++ = (uint8_t) ((ustrlen >>8) & 0xff);
memcpy(ssb, modifier1, ustrlen);
ssb+=ustrlen;
}
send_udp_message(224, cmd, udp_address, subscrbuf, ssb_size-4);
}
uwsgi.c
+32 -6
View File
@@ -523,6 +523,15 @@ void kill_them_all(int signum) {
}
uwsgi_log("SIGINT/SIGQUIT received...killing workers...\n");
// unsubscribe if needed
struct uwsgi_string_list *subscriptions = uwsgi.subscriptions;
while(subscriptions) {
uwsgi_subscribe(subscriptions->value, 1);
subscriptions = subscriptions->next;
}
for (i = 1; i <= uwsgi.numproc; i++) {
if (uwsgi.workers[i].pid > 0)
kill(uwsgi.workers[i].pid, SIGINT);
@@ -608,6 +617,14 @@ void grace_them_all(int signum) {
uwsgi_log("...gracefully killing workers...\n");
// unsubscribe if needed
struct uwsgi_string_list *subscriptions = uwsgi.subscriptions;
while(subscriptions) {
uwsgi_subscribe(subscriptions->value, 1);
subscriptions = subscriptions->next;
}
for (i = 1; i <= uwsgi.numproc; i++) {
if (uwsgi.auto_snapshot) {
if (uwsgi.workers[i].snapshot > 0) {
@@ -689,6 +706,14 @@ void reap_them_all(int signum) {
return;
uwsgi_log("...brutally killing workers...\n");
// unsubscribe if needed
struct uwsgi_string_list *subscriptions = uwsgi.subscriptions;
while(subscriptions) {
uwsgi_subscribe(subscriptions->value, 1);
subscriptions = subscriptions->next;
}
for (i = 1; i <= uwsgi.numproc; i++) {
if (uwsgi.workers[i].pid > 0)
kill(uwsgi.workers[i].pid, SIGTERM);
@@ -2963,7 +2988,7 @@ static int manage_base_opt(int i, char *optarg) {
return 1;
#ifdef UWSGI_UDP
case LONG_ARGS_CLUSTER_RELOAD:
send_udp_message(98, optarg, "", 0);
send_udp_message(98, 0, optarg, NULL, 0);
break;
case LONG_ARGS_CLUSTER_LOG:
uwsgi_stdin_sendto(optarg, 96, 0);
@@ -4013,8 +4038,9 @@ void uwsgi_stdin_sendto(char *socket_name, uint8_t modifier1, uint8_t modifier2)
char buf[4096];
ssize_t rlen;
size_t delta = 4096;
char *ptr = buf;
size_t delta = 4096-4;
// leave space for uwsgi header
char *ptr = buf+4;
rlen = read(0, ptr, delta);
while (rlen > 0) {
@@ -4026,9 +4052,9 @@ void uwsgi_stdin_sendto(char *socket_name, uint8_t modifier1, uint8_t modifier2)
rlen = read(0, ptr, delta);
}
if (ptr > buf) {
send_udp_message(modifier1, socket_name, buf, ptr - buf);
uwsgi_log("sent string \"%.*s\" to cluster node %s", ptr - buf, buf, socket_name);
if (ptr > buf+4) {
send_udp_message(modifier1, modifier2, socket_name, buf, (ptr - buf)-4);
uwsgi_log("sent string \"%.*s\" to cluster node %s", (ptr - buf)-4, buf+4, socket_name);
}
}
uwsgi.h
+10 -2
View File
@@ -1857,7 +1857,7 @@ uint64_t uwsgi_swap64(uint64_t);
#endif
#ifdef UWSGI_UDP
ssize_t send_udp_message(uint8_t, char *, char *, uint16_t);
ssize_t send_udp_message(uint8_t, uint8_t, char *, char *, uint16_t);
#endif
int uwsgi_parse_packet(struct wsgi_request *, int);
@@ -2428,6 +2428,9 @@ struct uwsgi_subscribe_node {
uint64_t requests;
uint64_t transferred;
int death_mark;
uint64_t reference;
struct uwsgi_subscribe_slot *slot;
struct uwsgi_subscribe_node *next;
@@ -2458,8 +2461,9 @@ void mule_send_msg(int, char *, size_t);
void create_signal_pipe(int *);
struct uwsgi_subscribe_slot *uwsgi_get_subscribe_slot(struct uwsgi_subscribe_slot **, char *, uint16_t, int);
struct uwsgi_subscribe_node *uwsgi_get_subscribe_node_by_name(struct uwsgi_subscribe_slot **, char *, uint16_t, char *, uint16_t, int);
struct uwsgi_subscribe_node *uwsgi_get_subscribe_node(struct uwsgi_subscribe_slot **, char *, uint16_t, int);
void uwsgi_remove_subscribe_node(struct uwsgi_subscribe_slot **, struct uwsgi_subscribe_node *);
int uwsgi_remove_subscribe_node(struct uwsgi_subscribe_slot **, struct uwsgi_subscribe_node *);
struct uwsgi_subscribe_node *uwsgi_add_subscribe_node(struct uwsgi_subscribe_slot **, struct uwsgi_subscribe_req *, int);
ssize_t uwsgi_mule_get_msg(int, int, char *, size_t, int);
@@ -2476,6 +2480,10 @@ void manage_cluster_announce(char *, uint16_t, char *, uint16_t, void *);
int uwsgi_read_response(int, struct uwsgi_header *, int, char **);
char *uwsgi_simple_file_read(char *);
void uwsgi_send_subscription(char *, char *, size_t , char *, size_t, uint8_t);
void uwsgi_subscribe(char *, uint8_t);
#ifdef __linux__
#ifdef MADV_MERGEABLE
void uwsgi_linux_ksm_map(void);