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

various fixes in multiple sockets and support for file piped via stdin

Browse Source
This commit is contained in:
roberto@natty32
2011-04-27 14:15:01 +02:00
parent b4ce175bdf
commit 80a6d2ccfb
13 changed files with 358 additions and 269 deletions
Download Patch File Download Diff File Expand all files Collapse all files
async.c
+6 -5
View File
@@ -199,13 +199,13 @@ void *async_loop(void *arg1) {
int interesting_fd, i;
struct uwsgi_rb_timer *min_timeout;
int timeout;
int j;
int is_a_new_connection;
int proto_parser_status;
static struct uwsgi_async_request *current_request = NULL, *next_async_request = NULL;
void *events = event_queue_alloc(64);
struct uwsgi_socket *uwsgi_sock = uwsgi.sockets;
uwsgi.async_runqueue = NULL;
uwsgi.async_runqueue_cnt = 0;
@@ -251,9 +251,10 @@ void *async_loop(void *arg1) {
// new request coming in ?
for(j=0;j<uwsgi.sockets_cnt;j++) {
uwsgi_sock = uwsgi.sockets;
while(uwsgi_sock) {
if (interesting_fd == uwsgi.sockets[j].fd) {
if (interesting_fd == uwsgi_sock->fd) {
is_a_new_connection = 1;
@@ -263,7 +264,7 @@ void *async_loop(void *arg1) {
break;;
}
wsgi_req_setup(uwsgi.wsgi_req, uwsgi.wsgi_req->async_id, j );
wsgi_req_setup(uwsgi.wsgi_req, uwsgi.wsgi_req->async_id, uwsgi_sock );
if (wsgi_req_simple_accept(uwsgi.wsgi_req, interesting_fd)) {
#ifdef UWSGI_EVENT_USE_PORT
event_queue_add_fd_read(uwsgi.async_queue, interesting_fd);
@@ -280,7 +281,7 @@ void *async_loop(void *arg1) {
#ifndef __linux__
if (uwsgi.numproc > 1) {
/* re-set blocking socket */
if (fcntl(uwsgi.wsgi_req->poll.fd, F_SETFL, uwsgi.sockets[j].arg) < 0) {
if (fcntl(uwsgi.wsgi_req->poll.fd, F_SETFL, uwsgi_sock->arg) < 0) {
uwsgi_error("fcntl()");
uwsgi.async_queue_unused_ptr++;
uwsgi.async_queue_unused[uwsgi.async_queue_unused_ptr] = uwsgi.wsgi_req;
loop.c
+60 -6
View File
@@ -63,13 +63,25 @@ void *simple_loop(void *arg1) {
}
#endif
// initialize the main event queue to monitor sockets
uwsgi.main_queue = event_queue_init();
struct uwsgi_socket *uwsgi_sock = uwsgi.sockets;
while(uwsgi_sock) {
event_queue_add_fd_read(uwsgi.main_queue, uwsgi_sock->fd);
uwsgi_sock = uwsgi_sock->next;
}
if (uwsgi.signal_socket > -1) {
event_queue_add_fd_read(uwsgi.main_queue, uwsgi.signal_socket);
}
while (uwsgi.workers[uwsgi.mywid].manage_next_request) {
UWSGI_CLEAR_STATUS;
wsgi_req_setup(wsgi_req, core_id, -1);
wsgi_req_setup(wsgi_req, core_id, NULL);
if (wsgi_req_accept(wsgi_req)) {
continue;
@@ -99,6 +111,8 @@ void *zeromq_loop(void *arg1) {
struct wsgi_request *wsgi_req = uwsgi.wsgi_requests[core_id];
uwsgi.zeromq_recv_flag = 0;
zmq_pollitem_t zmq_poll_items[2];
char uwsgi_signal;
if (uwsgi.threads > 1) {
@@ -131,17 +145,57 @@ void *zeromq_loop(void *arg1) {
}
if (uwsgi.signal_socket > -1) {
zmq_poll_items[0].socket = pthread_getspecific(uwsgi.zmq_pull);
zmq_poll_items[0].fd = -1;
zmq_poll_items[0].events = ZMQ_POLLIN;
zmq_poll_items[1].socket = NULL;
zmq_poll_items[1].fd = uwsgi.signal_socket;
zmq_poll_items[1].events = ZMQ_POLLIN;
}
while (uwsgi.workers[uwsgi.mywid].manage_next_request) {
UWSGI_CLEAR_STATUS;
wsgi_req_setup(wsgi_req, core_id, -1);
wsgi_req_setup(wsgi_req, core_id, NULL);
uwsgi.edge_triggered = 1;
int socket_id = uwsgi.zmq_socket;
wsgi_req->socket = &uwsgi.sockets[socket_id];
wsgi_req->poll.fd = wsgi_req->socket->proto_accept(wsgi_req, uwsgi.sockets[socket_id].fd);
wsgi_req->socket = uwsgi.zmq_socket;
if (uwsgi.signal_socket > -1) {
if (zmq_poll(zmq_poll_items, 2, -1) < 0) {
uwsgi_error("zmq_poll()");
continue;
}
if (zmq_poll_items[1].revents & ZMQ_POLLIN) {
if (read(uwsgi.signal_socket, &uwsgi_signal, 1) <= 0) {
if (uwsgi.no_orphans) {
uwsgi_log_verbose("uWSGI worker %d screams: UAAAAAAH my master died, i will follow him...\n", uwsgi.mywid);
end_me(0);
}
}
else {
uwsgi_log_verbose("master sent signal %d to worker %d\n", uwsgi_signal, uwsgi.mywid);
if (uwsgi_signal_handler(uwsgi_signal)) {
uwsgi_log_verbose("error managing signal %d on worker %d\n", uwsgi_signal, uwsgi.mywid);
}
}
continue;
}
if (zmq_poll_items[0].revents & ZMQ_POLLIN) {
wsgi_req->poll.fd = wsgi_req->socket->proto_accept(wsgi_req, uwsgi.zmq_socket->fd);
}
}
else {
wsgi_req->poll.fd = wsgi_req->socket->proto_accept(wsgi_req, uwsgi.zmq_socket->fd);
}
if (wsgi_req->poll.fd >= 0) {
wsgi_req_recv(wsgi_req);
master.c
+14 -9
View File
@@ -65,10 +65,10 @@ void uwsgi_subscribe(char *subscription) {
memcpy(ssb, "address", ustrlen);
ssb+=ustrlen;
ustrlen = strlen(uwsgi.sockets[0].name);
ustrlen = strlen(uwsgi.sockets->name);
*ssb++ = (uint8_t) (ustrlen & 0xff);
*ssb++ = (uint8_t) ((ustrlen >>8) & 0xff);
memcpy(ssb, uwsgi.sockets[0].name, ustrlen);
memcpy(ssb, uwsgi.sockets->name, ustrlen);
ssb+=ustrlen;
send_udp_message(224, udp_address, subscrbuf, ssb-subscrbuf);
@@ -168,7 +168,7 @@ void master_loop(char **argv, char **environ) {
int snmp_fd = -1;
#endif
int i=0,j;
int i=0;
int rlen;
int check_interval = 1;
@@ -422,12 +422,14 @@ void master_loop(char **argv, char **environ) {
uwsgi_log( "closing all non-uwsgi socket fds > 2 (_SC_OPEN_MAX = %ld)...\n", sysconf(_SC_OPEN_MAX));
for (i = 3; i < sysconf(_SC_OPEN_MAX); i++) {
int found = 0;
for(j=0;j<uwsgi.sockets_cnt;j++) {
if (i == uwsgi.sockets[j].fd) {
uwsgi_log("found fd %d mapped to socket %d (%s)\n", i, j, uwsgi.sockets[j].name);
struct uwsgi_socket *uwsgi_sock = uwsgi.sockets;
while(uwsgi_sock) {
if (i == uwsgi_sock->fd) {
uwsgi_log("found fd %d mapped to socket %d (%s)\n", i, uwsgi_get_socket_num(uwsgi_sock), uwsgi_sock->name);
found = 1;
break;
}
uwsgi_sock = uwsgi_sock->next;
}
if (!found) {
@@ -879,9 +881,12 @@ void master_loop(char **argv, char **environ) {
#ifdef __linux__
for(i=0;i<uwsgi.sockets_cnt;i++) {
if (uwsgi.sockets[i].family != AF_INET) continue;
get_linux_tcp_info(uwsgi.sockets[i].fd);
struct uwsgi_socket *uwsgi_sock = uwsgi.sockets;
while(uwsgi_sock) {
if (uwsgi_sock->family == AF_INET) {
get_linux_tcp_info(uwsgi_sock->fd);
}
uwsgi_sock = uwsgi_sock->next;
}
#endif
mongrel2-uwsgi.conf
+4 -4
View File
@@ -3,15 +3,15 @@ main = Server(
access_log="/logs/access.log",
error_log="/logs/error.log",
chroot="./",
default_host="192.168.173.11",
default_host="192.168.173.13",
name="test",
pid_file="/run/mongrel2.pid",
port=6767,
hosts = [
Host(name="192.168.173.11", routes={
'/': Handler(send_spec='tcp://192.168.173.11:9999',
Host(name="192.168.173.13", routes={
'/': Handler(send_spec='tcp://192.168.173.13:9999',
send_ident='54c6755b-9628-40a4-9a2d-cc82a816345e',
recv_spec='tcp://192.168.173.11:9998', recv_ident='')
recv_spec='tcp://192.168.173.13:9998', recv_ident='')
})
]
)
plugins/http/http.c
+7 -7
View File
@@ -628,9 +628,9 @@ void http_loop() {
uhttp_session->instance_address_len = uhttp_session->un->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);
else if (uwsgi.sockets) {
uhttp_session->instance_address = uwsgi.sockets->name;
uhttp_session->instance_address_len = strlen(uwsgi.sockets->name);
}
if (!uhttp_session->instance_address_len) {
@@ -868,10 +868,10 @@ int http_init() {
if (!uhttp.nevents) uhttp.nevents = 64;
if (!uhttp.base && !uhttp.use_cache && !uhttp.to && !uwsgi.sockets_cnt && !uhttp.subscription_server && !uhttp.use_cluster) {
uwsgi.sockets[0].name = uwsgi_malloc(64);
uwsgi.sockets_cnt++;
snprintf(uwsgi.sockets[0].name, 64, "%d_%d.sock", (int) time(NULL), (int) getpid());
if (!uhttp.base && !uhttp.use_cache && !uhttp.to && !uwsgi.sockets && !uhttp.subscription_server && !uhttp.use_cluster) {
char *rand_name = uwsgi_malloc(64);
snprintf(rand_name, 64, "%d_%d.sock", (int) time(NULL), (int) getpid());
uwsgi.sockets = uwsgi_new_socket(rand_name);
}
uhttp.server = bind_to_tcp(uhttp.socket_name, uwsgi.listen_queue, strchr(uhttp.socket_name,':'));
plugins/nagios/nagios.c
+3 -3
View File
@@ -22,11 +22,11 @@ int nagios() {
if (!use_nagios) {
return 1;
}
if (uwsgi.sockets[0].name == NULL) {
if (!uwsgi.sockets) {
fprintf(stdout, "UWSGI UNKNOWN: you have specified an invalid socket\n");
exit(3);
}
tcp_port = strchr(uwsgi.sockets[0].name, ':');
tcp_port = strchr(uwsgi.sockets->name, ':');
if (tcp_port == NULL) {
fprintf(stdout, "UWSGI UNKNOWN: you have specified an invalid socket\n");
exit(3);
@@ -34,7 +34,7 @@ int nagios() {
tcp_port[0] = 0;
nagios_poll.fd = connect_to_tcp(uwsgi.sockets[0].name, atoi(tcp_port + 1), uwsgi.shared->options[UWSGI_OPTION_SOCKET_TIMEOUT], 0);
nagios_poll.fd = connect_to_tcp(uwsgi.sockets->name, atoi(tcp_port + 1), uwsgi.shared->options[UWSGI_OPTION_SOCKET_TIMEOUT], 0);
if (nagios_poll.fd < 0) {
fprintf(stdout, "UWSGI CRITICAL: could not connect() to workers\n");
exit(2);
plugins/python/uwsgi_pymodule.c
+1 -4
View File
@@ -2177,7 +2177,6 @@ PyObject *py_uwsgi_parse_file(PyObject * self, PyObject * args) {
PyObject *py_uwsgi_grunt(PyObject * self, PyObject * args) {
pid_t grunt_pid;
int i;
struct wsgi_request *wsgi_req = current_wsgi_req();
if (uwsgi.grunt) {
@@ -2194,9 +2193,7 @@ PyObject *py_uwsgi_grunt(PyObject * self, PyObject * args) {
goto clear;
}
else if (grunt_pid == 0) {
for (i = 0; i < uwsgi.sockets_cnt; i++) {
close(uwsgi.sockets[i].fd);
}
uwsgi_close_all_sockets();
// create a new session
setsid();
// exit on SIGPIPE
proto/zeromq.c
+5 -5
View File
@@ -383,11 +383,11 @@ void uwsgi_proto_zeromq_close(struct wsgi_request *wsgi_req) {
return;
zmq_msg_init_data(&reply, wsgi_req->proto_parser_buf, wsgi_req->proto_parser_pos, uwsgi_proto_zeromq_free, NULL);
pthread_mutex_lock(&uwsgi.zmq_lock);
if (uwsgi.threads > 1) pthread_mutex_lock(&uwsgi.zmq_lock);
if (zmq_send(uwsgi.zmq_pub, &reply, 0)) {
uwsgi_error("zmq_send()");
}
pthread_mutex_unlock(&uwsgi.zmq_lock);
if (uwsgi.threads > 1) pthread_mutex_unlock(&uwsgi.zmq_lock);
zmq_msg_close(&reply);
if (wsgi_req->async_post && wsgi_req->body_as_file) {
@@ -429,14 +429,14 @@ ssize_t uwsgi_proto_zeromq_write(struct wsgi_request * wsgi_req, char *buf, size
//uwsgi_log("|%.*s|\n", (int)wsgi_req->proto_parser_pos+len, zmq_body);
zmq_msg_init_data(&reply, zmq_body, wsgi_req->proto_parser_pos + len, uwsgi_proto_zeromq_free, NULL);
pthread_mutex_lock(&uwsgi.zmq_lock);
if (uwsgi.threads > 1) pthread_mutex_lock(&uwsgi.zmq_lock);
if (zmq_send(uwsgi.zmq_pub, &reply, 0)) {
uwsgi_error("zmq_send()");
pthread_mutex_unlock(&uwsgi.zmq_lock);
if (uwsgi.threads > 1) pthread_mutex_unlock(&uwsgi.zmq_lock);
zmq_msg_close(&reply);
return -1;
}
pthread_mutex_unlock(&uwsgi.zmq_lock);
if (uwsgi.threads > 1) pthread_mutex_unlock(&uwsgi.zmq_lock);
zmq_msg_close(&reply);
return len;
socket.c
+63 -18
View File
@@ -536,7 +536,47 @@ int timed_connect(struct pollfd *fdpoll, const struct sockaddr *addr, int addr_s
}
void uwsgi_add_socket_from_fd(int j, int i) {
int uwsgi_get_socket_num(struct uwsgi_socket *uwsgi_sock) {
int count = 0;
struct uwsgi_socket *current_sock = uwsgi.sockets;
while(current_sock) {
if (uwsgi_sock == current_sock) {
return count;
}
count++;
current_sock = current_sock->next;
}
return -1;
}
struct uwsgi_socket *uwsgi_new_socket(char *name) {
struct uwsgi_socket *uwsgi_sock = uwsgi.sockets, *old_uwsgi_sock;
if (!uwsgi_sock) {
uwsgi.sockets = uwsgi_malloc(sizeof(struct uwsgi_socket));
uwsgi_sock = uwsgi.sockets;
}
else {
while(uwsgi_sock) {
old_uwsgi_sock = uwsgi_sock;
uwsgi_sock = uwsgi_sock->next;
}
uwsgi_sock = uwsgi_malloc(sizeof(struct uwsgi_socket));
old_uwsgi_sock->next = uwsgi_sock;
}
memset(uwsgi_sock, 0, sizeof(struct uwsgi_socket));
uwsgi_sock->name = name;
return uwsgi_sock;
}
void uwsgi_add_socket_from_fd(struct uwsgi_socket *uwsgi_sock, int fd) {
socklen_t socket_type_len;
union uwsgi_sockaddr_ptr gsa, isa;
@@ -544,19 +584,17 @@ void uwsgi_add_socket_from_fd(int j, int i) {
socket_type_len = sizeof(struct sockaddr_un);
gsa.sa = &usa.sa;
if (!getsockname(j, gsa.sa, &socket_type_len)) {
if (!getsockname(fd, gsa.sa, &socket_type_len)) {
if (socket_type_len <= 2) {
// unbound socket
return;
}
if (gsa.sa->sa_family == AF_UNIX) {
if (!strcmp(usa.sa_un.sun_path, uwsgi.sockets[i].name)) {
uwsgi.sockets[i].fd = j;
uwsgi.sockets[i].family = AF_UNIX;
uwsgi.sockets[i].bound = 1;
uwsgi.sockets_poll[i].fd = uwsgi.sockets[i].fd;
uwsgi.sockets_poll[i].events = POLLIN;
uwsgi_log("uwsgi socket %d inherited UNIX address %s fd %d\n", i, uwsgi.sockets[i].name, uwsgi.sockets[i].fd);
if (!strcmp(usa.sa_un.sun_path, uwsgi_sock->name)) {
uwsgi_sock->fd = fd;
uwsgi_sock->family = AF_UNIX;
uwsgi_sock->bound = 1;
uwsgi_log("uwsgi socket %d inherited UNIX address %s fd %d\n", uwsgi_get_socket_num(uwsgi_sock), uwsgi_sock->name, uwsgi_sock->fd);
}
}
else if (gsa.sa->sa_family == AF_INET) {
@@ -577,23 +615,21 @@ void uwsgi_add_socket_from_fd(int j, int i) {
else {
computed_addr = uwsgi_concat3(ipv4a, ":", computed_port);
}
char *asterisk = strchr(uwsgi.sockets[i].name, '*');
char *asterisk = strchr(uwsgi_sock->name, '*');
int match = 1;
if (asterisk) {
asterisk[0] = 0;
match = strncmp(computed_addr, uwsgi.sockets[i].name, strlen(uwsgi.sockets[i].name));
match = strncmp(computed_addr, uwsgi_sock->name, strlen(uwsgi_sock->name));
asterisk[0] = '*';
}
else {
match = strcmp(computed_addr, uwsgi.sockets[i].name);
match = strcmp(computed_addr, uwsgi_sock->name);
}
if (!match) {
uwsgi.sockets[i].fd = j;
uwsgi.sockets[i].family = AF_INET;
uwsgi.sockets[i].bound = 1;
uwsgi.sockets_poll[i].fd = uwsgi.sockets[i].fd;
uwsgi.sockets_poll[i].events = POLLIN;
uwsgi_log("uwsgi socket %d inherited INET address %s fd %d\n", i, uwsgi.sockets[i].name, uwsgi.sockets[i].fd);
uwsgi_sock->fd = fd;
uwsgi_sock->family = AF_INET;
uwsgi_sock->bound = 1;
uwsgi_log("uwsgi socket %d inherited INET address %s fd %d\n", uwsgi_get_socket_num(uwsgi_sock), uwsgi_sock->name, uwsgi_sock->fd);
}
free(computed_addr);
}
@@ -602,3 +638,12 @@ void uwsgi_add_socket_from_fd(int j, int i) {
}
}
void uwsgi_close_all_sockets() {
struct uwsgi_socket *uwsgi_sock = uwsgi.sockets;
while(uwsgi_sock) {
close(uwsgi_sock->fd);
uwsgi_sock = uwsgi_sock->next;
}
}
spooler.c
+5 -5
View File
@@ -4,17 +4,17 @@
extern struct uwsgi_server uwsgi;
pid_t spooler_start() {
int i;
pid_t pid;
pid = fork();
pid_t pid = fork();
if (pid < 0) {
uwsgi_error("fork()");
exit(1);
}
else if (pid == 0) {
for (i = 0; i < uwsgi.sockets_cnt; i++) {
close(uwsgi.sockets[i].fd);
struct uwsgi_socket *uwsgi_sock = uwsgi.sockets;
while(uwsgi_sock) {
close(uwsgi_sock->fd);
uwsgi_sock = uwsgi_sock->next;
}
spooler();
}
utils.c
+41 -51
View File
@@ -505,7 +505,7 @@ void uwsgi_close_request(struct wsgi_request *wsgi_req) {
}
void wsgi_req_setup(struct wsgi_request *wsgi_req, int async_id, int socket_id) {
void wsgi_req_setup(struct wsgi_request *wsgi_req, int async_id, struct uwsgi_socket *uwsgi_sock) {
wsgi_req->poll.events = POLLIN;
wsgi_req->app_id = uwsgi.default_app;
@@ -525,8 +525,8 @@ void wsgi_req_setup(struct wsgi_request *wsgi_req, int async_id, int socket_id)
wsgi_req->post_buffering_buf = uwsgi.async_post_buf[wsgi_req->async_id];
}
if (socket_id > -1) {
wsgi_req->socket = &uwsgi.sockets[socket_id];
if (uwsgi_sock) {
wsgi_req->socket = uwsgi_sock;
}
}
@@ -594,35 +594,18 @@ int wsgi_req_simple_accept(struct wsgi_request *wsgi_req, int fd) {
int wsgi_req_accept(struct wsgi_request *wsgi_req) {
int i;
int ret;
int interesting_fd;
char uwsgi_signal;
struct uwsgi_socket *uwsgi_sock = uwsgi.sockets;
/*
if (uwsgi.edge_triggered) {
for(i=0;i<uwsgi.sockets_cnt;i++) {
if (uwsgi.sockets[i].edge_trigger) {
uwsgi.sockets_poll[i].revents = POLLIN;
}
else {
uwsgi.sockets_poll[i].revents = 0;
}
}
goto edgetrigger;
}
*/
polling:
uwsgi.edge_triggered = 1;
ret = poll(uwsgi.sockets_poll, uwsgi.sockets_cnt + uwsgi.master_process, uwsgi.edge_triggered - 1);
ret = event_queue_wait(uwsgi.main_queue, uwsgi.edge_triggered - 1, &interesting_fd);
if (ret < 0) {
uwsgi_error("poll()");
return -1;
}
if (uwsgi.master_process && uwsgi.sockets_poll[uwsgi.sockets_cnt].revents) {
if (read(uwsgi.sockets_poll[uwsgi.sockets_cnt].fd, &uwsgi_signal, 1) <= 0) {
if (uwsgi.signal_socket > -1 && interesting_fd == uwsgi.signal_socket) {
if (read(interesting_fd, &uwsgi_signal, 1) <= 0) {
if (uwsgi.no_orphans) {
uwsgi_log_verbose("uWSGI worker %d screams: UAAAAAAH my master died, i will follow him...\n", uwsgi.mywid);
end_me(0);
@@ -637,33 +620,19 @@ int wsgi_req_accept(struct wsgi_request *wsgi_req) {
}
//edgetrigger:
for (i = 0; i < uwsgi.sockets_cnt; i++) {
if (uwsgi.sockets_poll[i].revents & POLLIN || (uwsgi.edge_triggered && uwsgi.sockets[i].edge_trigger)) {
int socket_id = i;
wsgi_req->socket = &uwsgi.sockets[socket_id];
wsgi_req->poll.fd = wsgi_req->socket->proto_accept(wsgi_req, uwsgi.sockets_poll[i].fd);
while(uwsgi_sock) {
if (interesting_fd == uwsgi_sock->fd || (uwsgi.edge_triggered && uwsgi_sock->edge_trigger)) {
wsgi_req->socket = uwsgi_sock;
wsgi_req->poll.fd = wsgi_req->socket->proto_accept(wsgi_req, interesting_fd);
if (wsgi_req->poll.fd < 0) {
return -1;
if (uwsgi.sockets[i].edge_trigger) {
return -1;
}
if (errno == EWOULDBLOCK) {
goto polling;
}
uwsgi_error("accept()");
return -1;
}
if (!uwsgi.sockets[socket_id].edge_trigger) {
if (!uwsgi_sock->edge_trigger) {
// in Linux, new sockets do not inherit attributes
#ifndef __linux__
/* re-set blocking socket */
int arg = uwsgi.sockets[i].arg;
int arg = uwsgi_sock->arg;
arg &= (~O_NONBLOCK);
if (fcntl(wsgi_req->poll.fd, F_SETFL, arg) < 0) {
uwsgi_error("fcntl()");
@@ -677,9 +646,10 @@ int wsgi_req_accept(struct wsgi_request *wsgi_req) {
}
}
return 0;
}
uwsgi_sock = uwsgi_sock->next;
}
return -1;
@@ -1419,9 +1389,28 @@ char *uwsgi_open_and_read(char *url, int *size, int add_zero, char *magic_table[
int body = 0;
char *magic_buf;
// http url ?
// stdin ?
if (!strcmp(url, "-")) {
char stack_buf[4096];
len = 1;
while(len > 0) {
len = read(0, stack_buf, 4096);
if (len > 0) {
*size += len;
buffer = realloc(buffer, *size);
memcpy(buffer+(*size-len), stack_buf, len);
}
}
if (!strncmp("http://", url, 7)) {
if (add_zero) {
*size = *size+1;
buffer = realloc(buffer, *size);
buffer[*size-1] = 0;
}
}
// http url ?
else if (!strncmp("http://", url, 7)) {
domain = url + 7;
uri = strchr(domain, '/');
if (!uri) {
@@ -1685,7 +1674,6 @@ int uwsgi_attach_daemon(char *command) {
void spawn_daemon(struct uwsgi_daemon *ud) {
int i;
char *argv[64];
char *a;
int cnt = 1;
@@ -1716,8 +1704,10 @@ void spawn_daemon(struct uwsgi_daemon *ud) {
else {
// close uwsgi sockets
for (i = 0; i < uwsgi.sockets_cnt; i++) {
close(uwsgi.sockets[i].fd);
struct uwsgi_socket *uwsgi_sock = uwsgi.sockets;
while(uwsgi_sock) {
close(uwsgi_sock->fd);
uwsgi_sock = uwsgi_sock->next;
}
close(ud->pipe[0]);
uwsgi.c
+132 -142
View File
@@ -471,7 +471,7 @@ struct uwsgi_plugin unconfigured_plugin = {
static void vacuum(void) {
int i;
struct uwsgi_socket *uwsgi_sock = uwsgi.sockets;
if (uwsgi.vacuum) {
if (getpid() == masterpid) {
@@ -499,15 +499,16 @@ static void vacuum(void) {
uwsgi_error("chdir()");
}
}
for (i = 0; i < uwsgi.sockets_cnt; i++) {
if (uwsgi.sockets[i].family == AF_UNIX) {
if (unlink(uwsgi.sockets[i].name)) {
while(uwsgi_sock) {
if (uwsgi_sock->family == AF_UNIX) {
if (unlink(uwsgi_sock->name)) {
uwsgi_error("unlink()");
}
else {
uwsgi_log("VACUUM: unix socket %s removed.\n", uwsgi.sockets[i].name);
uwsgi_log("VACUUM: unix socket %s removed.\n", uwsgi_sock->name);
}
}
uwsgi_sock = uwsgi_sock->next;
}
}
}
@@ -585,6 +586,8 @@ int main(int argc, char *argv[], char *envp[]) {
uwsgi.master_queue = -1;
uwsgi.signal_socket = -1;
uwsgi.emperor_fd_config = -1;
uwsgi.cluster_fd = -1;
@@ -988,6 +991,7 @@ int main(int argc, char *argv[], char *envp[]) {
fclose(pidfile);
}
/*
for (i = 0; i < uwsgi.shared_sockets_cnt; i++) {
char *tcp_port = strchr(uwsgi.shared_sockets[i].name, ':');
if (tcp_port == NULL) {
@@ -1007,6 +1011,7 @@ int main(int argc, char *argv[], char *envp[]) {
}
uwsgi.shared_sockets[i].bound = 1;
}
*/
// call jail systems
for (i = 0; i < uwsgi.gp_cnt; i++) {
@@ -1052,6 +1057,7 @@ int uwsgi_start(void *v_argv) {
socklen_t socket_type_len;
int emperor_pipe[2];
struct uwsgi_socket *uwsgi_sock;
#ifdef __linux__
if (uwsgi.ns) {
@@ -1154,7 +1160,7 @@ int uwsgi_start(void *v_argv) {
}
// do not go on if no socket or gateway is defined
if (!uwsgi.sockets_cnt && !uwsgi.gateways_cnt) {
if (!uwsgi.sockets && !uwsgi.gateways_cnt) {
exit(0);
}
close(emperor_pipe[0]);
@@ -1334,14 +1340,15 @@ int uwsgi_start(void *v_argv) {
//check for inherited sockets
if (uwsgi.is_a_reload) {
for (i = 0; i < uwsgi.sockets_cnt; i++) {
uwsgi_sock = uwsgi.sockets;
while(uwsgi_sock) {
//a bit overengineering
if (uwsgi.sockets[i].name != NULL) {
if (uwsgi_sock->name[0] != 0) {
for (j = 3; j < sysconf(_SC_OPEN_MAX); j++) {
uwsgi_add_socket_from_fd(j, i);
uwsgi_add_socket_from_fd(uwsgi_sock, j);
}
}
uwsgi_sock = uwsgi_sock->next;
}
//now close all the unbound fd
@@ -1358,11 +1365,13 @@ int uwsgi_start(void *v_argv) {
socket_type_len = sizeof(struct sockaddr_un);
gsa.sa = (struct sockaddr *) &usa;
if (!getsockname(j, gsa.sa, &socket_type_len)) {
for (i = 0; i < uwsgi.sockets_cnt; i++) {
if (uwsgi.sockets[i].fd == j && uwsgi.sockets[i].bound) {
uwsgi_sock = uwsgi.sockets;
while(uwsgi_sock) {
if (uwsgi_sock->fd == j && uwsgi_sock->bound) {
useless = 0;
break;
}
uwsgi_sock = uwsgi_sock->next;
}
}
if (useless)
@@ -1370,59 +1379,52 @@ int uwsgi_start(void *v_argv) {
}
}
//now bind all the unbound sockets
for (i = 0; i < uwsgi.sockets_cnt; i++) {
if (!uwsgi.sockets[i].bound) {
char *tcp_port = strchr(uwsgi.sockets[i].name, ':');
uwsgi_sock = uwsgi.sockets;
while(uwsgi_sock) {
if (!uwsgi_sock->bound) {
char *tcp_port = strchr(uwsgi_sock->name, ':');
if (tcp_port == NULL) {
uwsgi.sockets[i].fd = bind_to_unix(uwsgi.sockets[i].name, uwsgi.listen_queue, uwsgi.chmod_socket, uwsgi.abstract_socket);
uwsgi.sockets[i].family = AF_UNIX;
uwsgi_log("uwsgi socket %d bound to UNIX address %s fd %d\n", i, uwsgi.sockets[i].name, uwsgi.sockets[i].fd);
uwsgi_sock->fd = bind_to_unix(uwsgi_sock->name, uwsgi.listen_queue, uwsgi.chmod_socket, uwsgi.abstract_socket);
uwsgi_sock->family = AF_UNIX;
uwsgi_log("uwsgi socket %d bound to UNIX address %s fd %d\n", uwsgi_get_socket_num(uwsgi_sock), uwsgi_sock->name, uwsgi_sock->fd);
}
else {
uwsgi.sockets[i].fd = bind_to_tcp(uwsgi.sockets[i].name, uwsgi.listen_queue, tcp_port);
uwsgi.sockets[i].family = AF_INET;
uwsgi_log("uwsgi socket %d bound to TCP address %s fd %d\n", i, uwsgi.sockets[i].name, uwsgi.sockets[i].fd);
uwsgi_sock->fd = bind_to_tcp(uwsgi_sock->name, uwsgi.listen_queue, tcp_port);
uwsgi_sock->family = AF_INET;
uwsgi_log("uwsgi socket %d bound to TCP address %s fd %d\n", uwsgi_get_socket_num(uwsgi_sock), uwsgi_sock->name, uwsgi_sock->fd);
}
if (uwsgi.sockets[i].fd < 0) {
uwsgi_log("unable to create server socket on: %s\n", uwsgi.sockets[i].name);
if (uwsgi_sock->fd < 0) {
uwsgi_log("unable to create server socket on: %s\n", uwsgi_sock->name);
exit(1);
}
}
uwsgi.sockets[i].bound = 1;
uwsgi.sockets_poll[i].fd = uwsgi.sockets[i].fd;
uwsgi.sockets_poll[i].events = POLLIN;
uwsgi_sock->bound = 1;
uwsgi_sock = uwsgi_sock->next;
}
int zero_used = 0;
for (i = 0; i < uwsgi.sockets_cnt; i++) {
if (uwsgi.sockets[i].bound && uwsgi.sockets[i].fd == 0) {
uwsgi_sock = uwsgi.sockets;
while(uwsgi_sock) {
if (uwsgi_sock->bound && uwsgi_sock->fd == 0) {
zero_used = 1;
break;
}
uwsgi_sock = uwsgi_sock->next;
}
if (!zero_used) {
socket_type_len = sizeof(struct sockaddr_un);
gsa.sa = (struct sockaddr *) &usa;
if (!getsockname(0, gsa.sa, &socket_type_len)) {
if (uwsgi.sockets_cnt < 8) {
uwsgi.sockets_cnt++;
uwsgi.sockets[uwsgi.sockets_cnt - 1].fd = 0;
uwsgi.sockets[uwsgi.sockets_cnt - 1].bound = 1;
uwsgi.sockets[uwsgi.sockets_cnt - 1].family = gsa.sa->sa_family;
if (uwsgi.sockets[uwsgi.sockets_cnt - 1].family == AF_UNIX) {
uwsgi.sockets[uwsgi.sockets_cnt - 1].name = usa.sa_un.sun_path;
uwsgi_log("uwsgi socket %d inherited UNIX address %s fd 0\n", uwsgi.sockets_cnt - 1, uwsgi.sockets[uwsgi.sockets_cnt - 1].name);
if (gsa.sa->sa_family == AF_UNIX) {
uwsgi_sock = uwsgi_new_socket(usa.sa_un.sun_path);
uwsgi_log("uwsgi socket %d inherited UNIX address %s fd 0\n", uwsgi_get_socket_num(uwsgi_sock), uwsgi_sock->name);
}
else {
uwsgi_log("uwsgi socket %d inherited INET address %s fd 0\n", uwsgi.sockets_cnt - 1, uwsgi.sockets[uwsgi.sockets_cnt - 1].name);
uwsgi_sock = uwsgi_new_socket(":0");
uwsgi_log("uwsgi socket %d inherited INET address %s fd 0\n", uwsgi_get_socket_num(uwsgi_sock), uwsgi_sock->name);
}
uwsgi.sockets_poll[uwsgi.sockets_cnt - 1].fd = 0;
uwsgi.sockets_poll[uwsgi.sockets_cnt - 1].events = POLLIN;
}
else {
uwsgi_log("too many socket defined, i cannot map fd 0\n");
}
}
else {
@@ -1443,57 +1445,56 @@ int uwsgi_start(void *v_argv) {
// put listening socket in non-blocking state and set the protocol
for (i = 0; i < uwsgi.sockets_cnt; i++) {
uwsgi.sockets[i].arg = fcntl(uwsgi.sockets[i].fd, F_GETFL, NULL);
if (uwsgi.sockets[i].arg < 0) {
uwsgi_sock = uwsgi.sockets;
while(uwsgi_sock) {
uwsgi_sock->arg = fcntl(uwsgi_sock->fd, F_GETFL, NULL);
if (uwsgi_sock->arg < 0) {
uwsgi_error("fcntl()");
exit(1);
}
uwsgi.sockets[i].arg |= O_NONBLOCK;
if (fcntl(uwsgi.sockets[i].fd, F_SETFL, uwsgi.sockets[i].arg) < 0) {
uwsgi_sock->arg |= O_NONBLOCK;
if (fcntl(uwsgi_sock->fd, F_SETFL, uwsgi_sock->arg) < 0) {
uwsgi_error("fcntl()");
exit(1);
}
if (uwsgi.protocol && !strcmp("http", uwsgi.protocol)) {
uwsgi.sockets[i].proto = uwsgi_proto_http_parser;
uwsgi.sockets[i].proto_accept = uwsgi_proto_base_accept;
uwsgi.sockets[i].proto_write = uwsgi_proto_http_write;
uwsgi.sockets[i].proto_writev = uwsgi_proto_http_writev;
uwsgi.sockets[i].proto_write_header = uwsgi_proto_http_write_header;
uwsgi.sockets[i].proto_writev_header = uwsgi_proto_http_writev_header;
uwsgi.sockets[i].proto_sendfile = NULL;
uwsgi.sockets[i].proto_close = uwsgi_proto_base_close;
uwsgi_sock->proto = uwsgi_proto_http_parser;
uwsgi_sock->proto_accept = uwsgi_proto_base_accept;
uwsgi_sock->proto_write = uwsgi_proto_http_write;
uwsgi_sock->proto_writev = uwsgi_proto_http_writev;
uwsgi_sock->proto_write_header = uwsgi_proto_http_write_header;
uwsgi_sock->proto_writev_header = uwsgi_proto_http_writev_header;
uwsgi_sock->proto_sendfile = NULL;
uwsgi_sock->proto_close = uwsgi_proto_base_close;
}
else if (uwsgi.protocol && (!strcmp("fastcgi", uwsgi.protocol) || !strcmp("fcgi", uwsgi.protocol))) {
uwsgi.shared->options[UWSGI_OPTION_CGI_MODE] = 1;
uwsgi.sockets[i].proto = uwsgi_proto_fastcgi_parser;
uwsgi.sockets[i].proto_accept = uwsgi_proto_base_accept;
uwsgi.sockets[i].proto_write = uwsgi_proto_fastcgi_write;
uwsgi.sockets[i].proto_writev = uwsgi_proto_fastcgi_writev;
uwsgi.sockets[i].proto_write_header = uwsgi_proto_fastcgi_write_header;
uwsgi.sockets[i].proto_writev_header = uwsgi_proto_fastcgi_writev_header;
uwsgi.sockets[i].proto_sendfile = uwsgi_proto_fastcgi_sendfile;
uwsgi.sockets[i].proto_close = uwsgi_proto_fastcgi_close;
uwsgi_sock->proto = uwsgi_proto_fastcgi_parser;
uwsgi_sock->proto_accept = uwsgi_proto_base_accept;
uwsgi_sock->proto_write = uwsgi_proto_fastcgi_write;
uwsgi_sock->proto_writev = uwsgi_proto_fastcgi_writev;
uwsgi_sock->proto_write_header = uwsgi_proto_fastcgi_write_header;
uwsgi_sock->proto_writev_header = uwsgi_proto_fastcgi_writev_header;
uwsgi_sock->proto_sendfile = uwsgi_proto_fastcgi_sendfile;
uwsgi_sock->proto_close = uwsgi_proto_fastcgi_close;
}
else {
uwsgi.sockets[i].proto = uwsgi_proto_uwsgi_parser;
uwsgi.sockets[i].proto_accept = uwsgi_proto_base_accept;
uwsgi.sockets[i].proto_write = uwsgi_proto_uwsgi_write;
uwsgi.sockets[i].proto_writev = uwsgi_proto_uwsgi_writev;
uwsgi.sockets[i].proto_write_header = uwsgi_proto_uwsgi_write_header;
uwsgi.sockets[i].proto_writev_header = uwsgi_proto_uwsgi_writev_header;
uwsgi.sockets[i].proto_sendfile = NULL;
uwsgi.sockets[i].proto_close = uwsgi_proto_base_close;
uwsgi_sock->proto = uwsgi_proto_uwsgi_parser;
uwsgi_sock->proto_accept = uwsgi_proto_base_accept;
uwsgi_sock->proto_write = uwsgi_proto_uwsgi_write;
uwsgi_sock->proto_writev = uwsgi_proto_uwsgi_writev;
uwsgi_sock->proto_write_header = uwsgi_proto_uwsgi_write_header;
uwsgi_sock->proto_writev_header = uwsgi_proto_uwsgi_writev_header;
uwsgi_sock->proto_sendfile = NULL;
uwsgi_sock->proto_close = uwsgi_proto_base_close;
}
uwsgi_sock = uwsgi_sock->next;
}
#ifdef UWSGI_ZEROMQ
if (uwsgi.zeromq) {
if (uwsgi.sockets_cnt + 1 > MAX_SOCKETS) {
uwsgi_log("too much sockets defined, cannot add a zeromq receiver\n");
exit(1);
}
uwsgi.zmq_responder = strchr(uwsgi.zeromq, ',');
if (!uwsgi.zmq_responder) {
uwsgi_log("invalid zeromq address\n");
@@ -1503,15 +1504,15 @@ int uwsgi_start(void *v_argv) {
uwsgi.zmq_responder++;
uwsgi_log("zmq receiver: %s\n", uwsgi.zmq_receiver);
uwsgi_log("zmq responder: %s\n", uwsgi.zmq_responder);
uwsgi.zmq_socket = uwsgi.sockets_cnt++;
uwsgi.zmq_socket = uwsgi_new_socket(uwsgi.zmq_receiver);
}
#endif
}
// initialize request plugin only if workers or master are available
if (uwsgi.sockets_cnt || uwsgi.master_process) {
if (uwsgi.sockets || uwsgi.master_process) {
for (i = 0; i < 0xFF; i++) {
if (uwsgi.p[i]->init) {
uwsgi.p[i]->init();
@@ -1537,7 +1538,7 @@ int uwsgi_start(void *v_argv) {
#ifdef UWSGI_THREADING
if (uwsgi.sockets_cnt) {
if (uwsgi.sockets) {
if (uwsgi.has_threads) {
if (uwsgi.threads > 1)
uwsgi.current_wsgi_req = threaded_current_wsgi_req;
@@ -1549,21 +1550,22 @@ int uwsgi_start(void *v_argv) {
}
#endif
if (!uwsgi.sockets_cnt && !uwsgi.gateways_cnt && !uwsgi.no_server && !uwsgi.udp_socket) {
if (!uwsgi.sockets && !uwsgi.gateways_cnt && !uwsgi.no_server && !uwsgi.udp_socket) {
uwsgi_log("The -s/--socket option is missing and stdin is not a socket.\n");
exit(1);
}
else if (!uwsgi.sockets_cnt && uwsgi.gateways_cnt && !uwsgi.no_server && !uwsgi.master_process) {
else if (!uwsgi.sockets && uwsgi.gateways_cnt && !uwsgi.no_server && !uwsgi.master_process) {
exit(0);
}
if (!uwsgi.sockets_cnt)
if (!uwsgi.sockets)
uwsgi.numproc = 0;
#ifdef UWSGI_DEBUG
for (i = 0; i < uwsgi.sockets_cnt; i++) {
uwsgi_sock = uwsgi.sockets;
while(uwsgi_sock) {
so_bufsize_len = sizeof(int);
if (getsockopt(uwsgi.sockets[i].fd, SOL_SOCKET, SO_RCVBUF, &so_bufsize, &so_bufsize_len)) {
if (getsockopt(uwsgi_sock->fd, SOL_SOCKET, SO_RCVBUF, &so_bufsize, &so_bufsize_len)) {
uwsgi_error("getsockopt()");
}
else {
@@ -1571,12 +1573,13 @@ int uwsgi_start(void *v_argv) {
}
so_bufsize_len = sizeof(int);
if (getsockopt(uwsgi.sockets[i].fd, SOL_SOCKET, SO_SNDBUF, &so_bufsize, &so_bufsize_len)) {
if (getsockopt(uwsgi_sock->fd, SOL_SOCKET, SO_SNDBUF, &so_bufsize, &so_bufsize_len)) {
uwsgi_error("getsockopt()");
}
else {
uwsgi_debug("uwsgi socket %d SO_SNDBUF size: %d\n", i, so_bufsize);
}
uwsgi_sock = uwsgi_sock->next;
}
#endif
@@ -1595,6 +1598,7 @@ int uwsgi_start(void *v_argv) {
}
memset(uwsgi.workers, 0, sizeof(struct uwsgi_worker) * uwsgi.numproc + 1);
/*
for (i = 0; i < MAX_SOCKETS; i++) {
if (!uwsgi.map_socket[i])
continue;
@@ -1606,11 +1610,12 @@ int uwsgi_start(void *v_argv) {
exit(1);
}
uwsgi.workers[w].sockets_mask[i] = 1;
uwsgi_log("mapped socket %d (%s) to worker %d\n", i, uwsgi.sockets[i].name, w);
uwsgi_log("mapped socket %d (%s) to worker %d\n", i, uwsgi_sock->name, w);
p = strtok(NULL, ",");
}
}
*/
uwsgi.mypid = getpid();
masterpid = uwsgi.mypid;
@@ -1768,7 +1773,7 @@ int uwsgi_start(void *v_argv) {
#ifdef UWSGI_SPOOLER
if (uwsgi.spool_dir != NULL && uwsgi.sockets_cnt > 0) {
if (uwsgi.spool_dir != NULL && uwsgi.sockets) {
uwsgi.shared->spooler_pid = spooler_start();
}
#endif
@@ -1830,21 +1835,22 @@ int uwsgi_start(void *v_argv) {
//from now on the process is a real worker
}
/*
for (i = 0; i < uwsgi.sockets_cnt; i++) {
if (uwsgi.workers[uwsgi.mywid].sockets_mask[i]) {
// disable the socket for this worker
#ifdef UWSGI_DEBUG
uwsgi_log("switching off socket %d (%d) on worker %d\n", i, uwsgi.sockets[i].fd, uwsgi.mywid);
uwsgi_log("switching off socket %d (%d) on worker %d\n", i, uwsgi_sock->fd, uwsgi.mywid);
#endif
int fd = uwsgi.sockets[i].fd;
int fd = uwsgi_sock->fd;
close(fd);
fd = open("/dev/null", O_RDONLY);
if (fd < 0) {
uwsgi_error_open("/dev/null");
exit(1);
}
if (fd != uwsgi.sockets[i].fd) {
if (dup2(fd, uwsgi.sockets[i].fd)) {
if (fd != uwsgi_sock->fd) {
if (dup2(fd, uwsgi_sock->fd)) {
uwsgi_error("dup2()");
exit(1);
}
@@ -1854,6 +1860,7 @@ int uwsgi_start(void *v_argv) {
uwsgi.sockets_poll[i].events = 0;
}
}
*/
if (uwsgi.cpu_affinity) {
#ifdef __linux__
@@ -1884,14 +1891,14 @@ int uwsgi_start(void *v_argv) {
w_argv[0] = uwsgi.worker_exec;
w_argv[1] = NULL;
uwsgi.sockets[0].arg &= (~O_NONBLOCK);
if (fcntl(uwsgi.sockets[i].fd, F_SETFL, uwsgi.sockets[i].arg) < 0) {
uwsgi.sockets->arg &= (~O_NONBLOCK);
if (fcntl(uwsgi.sockets->fd, F_SETFL, uwsgi.sockets->arg) < 0) {
uwsgi_error("fcntl()");
exit(1);
}
if (uwsgi.sockets[0].fd != 0) {
if (dup2(uwsgi.sockets[0].fd, 0)) {
if (uwsgi.sockets->fd != 0) {
if (dup2(uwsgi.sockets->fd, 0)) {
uwsgi_error("dup2()");
}
}
@@ -1945,22 +1952,16 @@ int uwsgi_start(void *v_argv) {
exit(1);
}
uwsgi.sockets[uwsgi.zmq_socket].name = uwsgi.zmq_receiver;
uwsgi.sockets[uwsgi.zmq_socket].name_len = strlen(uwsgi.zmq_receiver);
uwsgi.zmq_socket->proto = uwsgi_proto_zeromq_parser;
uwsgi.zmq_socket->proto_accept = uwsgi_proto_zeromq_accept;
uwsgi.zmq_socket->proto_close = uwsgi_proto_zeromq_close;
uwsgi.zmq_socket->proto_write = uwsgi_proto_zeromq_write;
uwsgi.zmq_socket->proto_writev = uwsgi_proto_zeromq_writev;
uwsgi.zmq_socket->proto_write_header = uwsgi_proto_zeromq_write_header;
uwsgi.zmq_socket->proto_writev_header = uwsgi_proto_zeromq_writev_header;
uwsgi.zmq_socket->proto_sendfile = uwsgi_proto_zeromq_sendfile;
uwsgi.sockets[uwsgi.zmq_socket].proto = uwsgi_proto_zeromq_parser;
uwsgi.sockets[uwsgi.zmq_socket].proto_accept = uwsgi_proto_zeromq_accept;
uwsgi.sockets[uwsgi.zmq_socket].proto_close = uwsgi_proto_zeromq_close;
uwsgi.sockets[uwsgi.zmq_socket].proto_write = uwsgi_proto_zeromq_write;
uwsgi.sockets[uwsgi.zmq_socket].proto_writev = uwsgi_proto_zeromq_writev;
uwsgi.sockets[uwsgi.zmq_socket].proto_write_header = uwsgi_proto_zeromq_write_header;
uwsgi.sockets[uwsgi.zmq_socket].proto_writev_header = uwsgi_proto_zeromq_writev_header;
uwsgi.sockets[uwsgi.zmq_socket].proto_sendfile = uwsgi_proto_zeromq_sendfile;
uwsgi.sockets[uwsgi.zmq_socket].edge_trigger = 1;
uwsgi.zmq_socket->edge_trigger = 1;
if (pthread_key_create(&uwsgi.zmq_pull, NULL)) {
uwsgi_error("pthread_key_create()");
@@ -1981,19 +1982,15 @@ int uwsgi_start(void *v_argv) {
#ifdef ZMQ_FD
size_t zmq_socket_len = sizeof(int);
if (zmq_getsockopt(pthread_getspecific(uwsgi.zmq_pull), ZMQ_FD, &uwsgi.sockets[uwsgi.zmq_socket].fd, &zmq_socket_len) < 0) {
if (zmq_getsockopt(pthread_getspecific(uwsgi.zmq_pull), ZMQ_FD, &uwsgi.zmq_socket->fd, &zmq_socket_len) < 0) {
uwsgi_error("zmq_getsockopt()");
exit(1);
}
#else
uwsgi.sockets[uwsgi.zmq_socket].fd = -1;
uwsgi.zmq_socket->fd = -1;
#endif
uwsgi.sockets_poll[uwsgi.zmq_socket].fd = uwsgi.sockets[uwsgi.zmq_socket].fd;
uwsgi.sockets_poll[uwsgi.zmq_socket].events = POLLIN;
uwsgi.sockets[uwsgi.zmq_socket].bound = 1;
uwsgi.zmq_socket->bound = 1;
uwsgi.zeromq_recv_flag = ZMQ_NOBLOCK;
}
#endif
@@ -2007,8 +2004,10 @@ int uwsgi_start(void *v_argv) {
exit(1);
}
for (i = 0; i < uwsgi.sockets_cnt; i++) {
event_queue_add_fd_read(uwsgi.async_queue, uwsgi.sockets[i].fd);
uwsgi_sock = uwsgi.sockets;
while(uwsgi_sock) {
event_queue_add_fd_read(uwsgi.async_queue, uwsgi_sock->fd);
uwsgi_sock = uwsgi_sock->next;
}
}
@@ -2089,12 +2088,11 @@ int uwsgi_start(void *v_argv) {
if (uwsgi.master_process) {
uwsgi.sockets_poll[uwsgi.sockets_cnt].fd = uwsgi.shared->worker_signal_pipe[1];
uwsgi.sockets_poll[uwsgi.sockets_cnt].events = POLLIN;
uwsgi.signal_socket = uwsgi.shared->worker_signal_pipe[1];
#ifdef UWSGI_ASYNC
// add uwsgi signal fd to async queue
if (uwsgi.async > 1) {
event_queue_add_fd_read(uwsgi.async_queue, uwsgi.sockets_poll[uwsgi.sockets_cnt].fd);
event_queue_add_fd_read(uwsgi.async_queue, uwsgi.signal_socket);
}
#endif
}
@@ -2108,7 +2106,7 @@ int uwsgi_start(void *v_argv) {
}
else {
#ifdef UWSGI_ZEROMQ
if (uwsgi.zeromq && uwsgi.async < 2 && uwsgi.sockets_cnt == 1) {
if (uwsgi.zeromq && uwsgi.async < 2 && !uwsgi.sockets->next) {
pthread_attr_t pa;
pthread_t *a_thread;
@@ -2579,10 +2577,6 @@ static int manage_base_opt(int i, char *optarg) {
}
p[0] = 0;
int sn = atoi(optarg);
if (sn < 0 || sn >= MAX_SOCKETS) {
uwsgi_log("invalid socket number in map-socket\n");
exit(1);
}
uwsgi.map_socket[sn] = p + 1;
p[0] = ':';
return 1;
@@ -2669,15 +2663,10 @@ static int manage_base_opt(int i, char *optarg) {
}
return 1;
case 's':
if (uwsgi.sockets_cnt < 8) {
uwsgi.sockets[uwsgi.sockets_cnt].name = generate_socket_name(optarg);
uwsgi.sockets_cnt++;
}
else {
uwsgi_log("you can specify at most 8 --socket options\n");
}
uwsgi_new_socket(generate_socket_name(optarg));
return 1;
case LONG_ARGS_SHARED_SOCKET:
/*
if (uwsgi.shared_sockets_cnt < MAX_SOCKETS) {
uwsgi.shared_sockets[uwsgi.shared_sockets_cnt].name = generate_socket_name(optarg);
uwsgi.shared_sockets_cnt++;
@@ -2685,6 +2674,7 @@ static int manage_base_opt(int i, char *optarg) {
else {
uwsgi_log("you can specify at most 8 --socket options\n");
}
*/
return 1;
#ifdef UWSGI_XML
case 'x':
@@ -3103,9 +3093,9 @@ int uwsgi_cluster_add_me() {
char numproc[6];
#ifdef UWSGI_ZEROMQ
if (!uwsgi.sockets[0].name && !uwsgi.zeromq) {
if (!uwsgi.sockets && !uwsgi.zeromq) {
#else
if (!uwsgi.sockets[0].name) {
if (!uwsgi.sockets) {
#endif
uwsgi_log("you need to specify at least a socket to start a uWSGI cluster\n");
exit(1);
@@ -3115,8 +3105,8 @@ int uwsgi_cluster_add_me() {
size_t len;
if (uwsgi.sockets[0].name) {
len = 2 + strlen(key1) + 2 + strlen(uwsgi.hostname) + 2 + strlen(key2) + 2 + strlen(uwsgi.sockets[0].name) + 2 + strlen(key3) + 2 + strlen(numproc) + 2 + strlen(key4) + 2 + 1;
if (uwsgi.sockets) {
len = 2 + strlen(key1) + 2 + strlen(uwsgi.hostname) + 2 + strlen(key2) + 2 + strlen(uwsgi.sockets->name) + 2 + strlen(key3) + 2 + strlen(numproc) + 2 + strlen(key4) + 2 + 1;
}
else {
len = 2 + strlen(key1) + 2 + strlen(uwsgi.hostname) + 2 + strlen(key3) + 2 + strlen(numproc) + 2 + strlen(key4) + 2 + 1;
@@ -3138,18 +3128,18 @@ int uwsgi_cluster_add_me() {
ptrbuf += strlen(uwsgi.hostname);
if (uwsgi.sockets[0].name) {
if (uwsgi.sockets->name) {
ustrlen = strlen(key2);
*ptrbuf++ = (uint8_t) (ustrlen & 0xff);
*ptrbuf++ = (uint8_t) ((ustrlen >> 8) & 0xff);
memcpy(ptrbuf, key2, strlen(key2));
ptrbuf += strlen(key2);
ustrlen = strlen(uwsgi.sockets[0].name);
ustrlen = strlen(uwsgi.sockets->name);
*ptrbuf++ = (uint8_t) (ustrlen & 0xff);
*ptrbuf++ = (uint8_t) ((ustrlen >> 8) & 0xff);
memcpy(ptrbuf, uwsgi.sockets[0].name, strlen(uwsgi.sockets[0].name));
ptrbuf += strlen(uwsgi.sockets[0].name);
memcpy(ptrbuf, uwsgi.sockets->name, strlen(uwsgi.sockets->name));
ptrbuf += strlen(uwsgi.sockets->name);
}
uwsgi.h
+17 -10
View File
@@ -15,7 +15,6 @@
#define ushared uwsgi.shared
#define MAX_SOCKETS 8
#define MAX_APPS 64
#define MAX_GENERIC_PLUGINS 64
#define MAX_RPC 64
@@ -526,6 +525,8 @@ struct uwsgi_socket {
ssize_t (*proto_sendfile)(struct wsgi_request *);
void (*proto_close)(struct wsgi_request *);
int edge_trigger;
struct uwsgi_socket *next;
};
struct uwsgi_server;
@@ -971,6 +972,8 @@ struct uwsgi_server {
int harakiri_verbose;
int main_queue;
int numproc;
int async;
int async_running;
@@ -1069,13 +1072,13 @@ struct uwsgi_server {
char *protocol;
int sockets_cnt;
int signal_socket;
#ifdef UWSGI_ZEROMQ
char *zeromq;
char *zmq_receiver;
char *zmq_responder;
int zmq_socket;
struct uwsgi_socket *zmq_socket;
void *zmq_context;
//void *zmq_pull;
void *zmq_pub;
@@ -1084,14 +1087,13 @@ struct uwsgi_server {
pthread_key_t zmq_pull;
void *zmq_log_socket;
#endif
struct uwsgi_socket sockets[MAX_SOCKETS];
struct uwsgi_socket *sockets;
// leave a slot for no-orphan mode
struct pollfd sockets_poll[9];
int shared_sockets_cnt;
struct uwsgi_socket shared_sockets[MAX_SOCKETS];
struct uwsgi_socket *shared_sockets;
char *map_socket[MAX_SOCKETS];
char **map_socket;
time_t respawn_delta;
@@ -1357,7 +1359,7 @@ struct uwsgi_worker {
uint64_t exceptions;
char sockets_mask[MAX_SOCKETS];
char *sockets_mask;
};
@@ -1472,7 +1474,7 @@ void nagios(void);
void uwsgi_close_request(struct wsgi_request *);
void wsgi_req_setup(struct wsgi_request *, int, int);
void wsgi_req_setup(struct wsgi_request *, int, struct uwsgi_socket *);
int wsgi_req_recv(struct wsgi_request *);
int wsgi_req_async_recv(struct wsgi_request *);
int wsgi_req_accept(struct wsgi_request *);
@@ -1892,9 +1894,14 @@ int uwsgi_proto_zeromq_parser(struct wsgi_request *);
int uwsgi_num2str2(int, char *);
void uwsgi_add_socket_from_fd(int, int);
void uwsgi_add_socket_from_fd(struct uwsgi_socket *, int);
char *uwsgi_split3(char *, size_t, char, char **, size_t *, char **, size_t *, char **, size_t *);
char *uwsgi_split4(char *, size_t, char, char **, size_t *, char **, size_t *, char **, size_t *, char **, size_t *);
char *uwsgi_netstring(char *, size_t, char **, size_t *);
int uwsgi_get_socket_num(struct uwsgi_socket *);
struct uwsgi_socket *uwsgi_new_socket(char *);
void uwsgi_close_all_sockets(void);