clearlinux/uwsgi
2
0
Fork 0
mirror of https://github.com/clearlinux/uwsgi.git synced 2026-08-19 03:57:21 +00:00
Code Issues Packages Projects Releases Wiki Activity

added the router_http plugin

Browse Source
This commit is contained in:
roberto@quantal64
2012-09-01 11:17:20 +02:00
parent 2122bacb33
commit e8f39685cc
13 changed files with 700 additions and 335 deletions
Download Patch File Download Diff File Expand all files Collapse all files
buildconf/base.ini
+1 -1
View File
@@ -29,7 +29,7 @@ plugins =
bin_name = uwsgi
append_version =
plugin_dir = .
embedded_plugins = %(main_plugin)s, ping, cache, nagios, rrdtool, carbon, rpc, corerouter, fastrouter, http, ugreen, signal, syslog, rsyslog, logsocket, router_uwsgi, router_redirect, router_basicauth, zergpool, redislog, rawrouter
embedded_plugins = %(main_plugin)s, ping, cache, nagios, rrdtool, carbon, rpc, corerouter, fastrouter, http, ugreen, signal, syslog, rsyslog, logsocket, router_uwsgi, router_redirect, router_basicauth, zergpool, redislog, router_rewrite, router_http
as_shared_library = false
locking = auto
core/buffer.c
+68
View File
@@ -0,0 +1,68 @@
#include "uwsgi.h"
extern struct uwsgi_server uwsgi;
struct uwsgi_buffer *uwsgi_buffer_new(size_t len) {
struct uwsgi_buffer *ub = uwsgi_calloc(sizeof(struct uwsgi_buffer));
if (len) {
ub->buf = uwsgi_malloc(len);
ub->len = len;
}
return ub;
}
int uwsgi_buffer_append(struct uwsgi_buffer *ub, char *buf, size_t len) {
size_t remains = ub->len - ub->pos;
if (len > remains) {
char *new_buf = realloc(ub->buf, ub->len + UMAX(len, (size_t) uwsgi.page_size));
if (!new_buf) {
uwsgi_error("realloc()");
return -1;
}
}
memcpy(ub->buf + ub->pos, buf, len);
ub->pos += len;
return 0;
}
void uwsgi_buffer_destroy(struct uwsgi_buffer *ub) {
if (ub->buf) free(ub->buf);
free(ub);
}
int uwsgi_buffer_send(struct uwsgi_buffer *ub, int fd) {
size_t remains = ub->pos;
char *ptr = ub->buf;
while(remains > 0) {
int ret = uwsgi_waitfd_write(fd, uwsgi.shared->options[UWSGI_OPTION_SOCKET_TIMEOUT]);
if (ret > 0) {
ssize_t len = write(fd, ptr, remains);
if (len > 0) {
ptr += len;
remains -= len;
}
else if (len == 0) {
return -1;
}
else {
uwsgi_error("write()");
return -1;
}
}
else if (ret == 0) {
uwsgi_log("timeout while sending buffer !!!\n");
return -1;
}
else {
return -1;
}
}
return 0;
}
core/logging.c
+252
View File
@@ -19,6 +19,258 @@
extern struct uwsgi_server uwsgi;
//use this instead of fprintf to avoid buffering mess with udp logging
void uwsgi_log(const char *fmt, ...) {
va_list ap;
char logpkt[4096];
int rlen = 0;
int ret;
struct timeval tv;
char sftime[64];
char ctime_storage[26];
time_t now;
if (uwsgi.logdate) {
if (uwsgi.log_strftime) {
now = uwsgi_now();
rlen = strftime(sftime, 64, uwsgi.log_strftime, localtime(&now));
memcpy(logpkt, sftime, rlen);
memcpy(logpkt + rlen, " - ", 3);
rlen += 3;
}
else {
gettimeofday(&tv, NULL);
#ifdef __sun__
ctime_r((const time_t *) &tv.tv_sec, ctime_storage, 26);
#else
ctime_r((const time_t *) &tv.tv_sec, ctime_storage);
#endif
memcpy(logpkt, ctime_storage, 24);
memcpy(logpkt + 24, " - ", 3);
rlen = 24 + 3;
}
}
va_start(ap, fmt);
ret = vsnprintf(logpkt + rlen, 4096 - rlen, fmt, ap);
va_end(ap);
if (ret >= 4096) {
char *tmp_buf = uwsgi_malloc(rlen + ret + 1);
memcpy(tmp_buf, logpkt, rlen);
va_start(ap, fmt);
ret = vsnprintf(tmp_buf + rlen, ret + 1, fmt, ap);
va_end(ap);
rlen = write(2, tmp_buf, rlen + ret);
free(tmp_buf);
return;
}
rlen += ret;
// do not check for errors
rlen = write(2, logpkt, rlen);
}
void uwsgi_log_verbose(const char *fmt, ...) {
va_list ap;
char logpkt[4096];
int rlen = 0;
struct timeval tv;
char sftime[64];
time_t now;
char ctime_storage[26];
if (uwsgi.log_strftime) {
now = uwsgi_now();
rlen = strftime(sftime, 64, uwsgi.log_strftime, localtime(&now));
memcpy(logpkt, sftime, rlen);
memcpy(logpkt + rlen, " - ", 3);
rlen += 3;
}
else {
gettimeofday(&tv, NULL);
#ifdef __sun__
ctime_r((const time_t *) &tv.tv_sec, ctime_storage, 26);
#else
ctime_r((const time_t *) &tv.tv_sec, ctime_storage);
#endif
memcpy(logpkt, ctime_storage, 24);
memcpy(logpkt + 24, " - ", 3);
rlen = 24 + 3;
}
va_start(ap, fmt);
rlen += vsnprintf(logpkt + rlen, 4096 - rlen, fmt, ap);
va_end(ap);
// do not check for errors
rlen = write(2, logpkt, rlen);
}
// create the logpipe
void create_logpipe(void) {
#if defined(SOCK_SEQPACKET) && defined(__linux__)
if (socketpair(AF_UNIX, SOCK_SEQPACKET, 0, uwsgi.shared->worker_log_pipe)) {
#else
if (socketpair(AF_UNIX, SOCK_DGRAM, 0, uwsgi.shared->worker_log_pipe)) {
#endif
uwsgi_error("socketpair()\n");
exit(1);
}
uwsgi_socket_nb(uwsgi.shared->worker_log_pipe[0]);
uwsgi_socket_nb(uwsgi.shared->worker_log_pipe[1]);
if (uwsgi.shared->worker_log_pipe[1] != 1) {
if (dup2(uwsgi.shared->worker_log_pipe[1], 1) < 0) {
uwsgi_error("dup2()");
exit(1);
}
}
if (dup2(1, 2) < 0) {
uwsgi_error("dup2()");
exit(1);
}
}
#ifdef UWSGI_ZEROMQ
// the zeromq logger
ssize_t uwsgi_zeromq_logger(struct uwsgi_logger *ul, char *message, size_t len) {
if (!ul->configured) {
if (!ul->arg) {
uwsgi_log_safe("invalid zeromq syntax\n");
exit(1);
}
void *ctx = uwsgi_zeromq_init();
ul->data = zmq_socket(ctx, ZMQ_PUSH);
if (ul->data == NULL) {
uwsgi_error_safe("zmq_socket()");
exit(1);
}
if (zmq_connect(ul->data, ul->arg) < 0) {
uwsgi_error_safe("zmq_connect()");
exit(1);
}
ul->configured = 1;
}
zmq_msg_t msg;
if (zmq_msg_init_size(&msg, len) == 0) {
memcpy(zmq_msg_data(&msg), message, len);
#if ZMQ_VERSION >= ZMQ_MAKE_VERSION(3,0,0)
zmq_sendmsg(ul->data, &msg, 0);
#else
zmq_send(ul->data, &msg, 0);
#endif
zmq_msg_close(&msg);
}
return 0;
}
#endif
// log to the specified file or udp address
void logto(char *logfile) {
int fd;
#ifdef UWSGI_UDP
char *udp_port;
struct sockaddr_in udp_addr;
udp_port = strchr(logfile, ':');
if (udp_port) {
udp_port[0] = 0;
if (!udp_port[1] || !logfile[0]) {
uwsgi_log("invalid udp address\n");
exit(1);
}
fd = socket(AF_INET, SOCK_DGRAM, 0);
if (fd < 0) {
uwsgi_error("socket()");
exit(1);
}
memset(&udp_addr, 0, sizeof(struct sockaddr_in));
udp_addr.sin_family = AF_INET;
udp_addr.sin_port = htons(atoi(udp_port + 1));
char *resolved = uwsgi_resolve_ip(logfile);
if (resolved) {
udp_addr.sin_addr.s_addr = inet_addr(resolved);
}
else {
udp_addr.sin_addr.s_addr = inet_addr(logfile);
}
if (connect(fd, (const struct sockaddr *) &udp_addr, sizeof(struct sockaddr_in)) < 0) {
uwsgi_error("connect()");
exit(1);
}
}
else {
#endif
if (uwsgi.log_truncate) {
fd = open(logfile, O_RDWR | O_CREAT | O_TRUNC, S_IRUSR | S_IWUSR | S_IRGRP);
}
else {
fd = open(logfile, O_RDWR | O_CREAT | O_APPEND, S_IRUSR | S_IWUSR | S_IRGRP);
}
if (fd < 0) {
uwsgi_error_open(logfile);
exit(1);
}
uwsgi.logfile = logfile;
if (uwsgi.chmod_logfile_value) {
if (chmod(uwsgi.logfile, uwsgi.chmod_logfile_value)) {
uwsgi_error("chmod()");
}
}
#ifdef UWSGI_UDP
}
#endif
/* stdout */
if (fd != 1) {
if (dup2(fd, 1) < 0) {
uwsgi_error("dup2()");
exit(1);
}
close(fd);
}
/* stderr */
if (dup2(1, 2) < 0) {
uwsgi_error("dup2()");
exit(1);
}
}
void uwsgi_setup_log() {
if (uwsgi.daemonize) {
core/protocol.c
+2 -34
View File
@@ -412,12 +412,9 @@ int uwsgi_enqueue_message(char *host, int port, uint8_t modifier1, uint8_t modif
ssize_t uwsgi_send_message(int fd, uint8_t modifier1, uint8_t modifier2, char *message, uint16_t size, int pfd, ssize_t plen, int timeout) {
struct pollfd uwsgi_mpoll;
ssize_t cnt;
struct uwsgi_header uh;
char buffer[4096];
ssize_t ret = 0;
int pret;
struct msghdr msg;
struct iovec iov [1];
union {
@@ -478,37 +475,8 @@ ssize_t uwsgi_send_message(int fd, uint8_t modifier1, uint8_t modifier2, char *m
// transfer data from one socket to another
if (pfd >= 0 && plen > 0) {
uwsgi_mpoll.fd = pfd;
uwsgi_mpoll.events = POLLIN;
while(plen > 0) {
pret = poll(&uwsgi_mpoll, 1, timeout*1000);
if (pret < 0) {
uwsgi_error("poll()");
return -1;
}
else if (pret == 0) {
uwsgi_log("timeout waiting for socket data\n");
return -1;
}
else {
cnt = read(pfd, buffer, UMIN(4096, plen));
if (cnt < 0) {
uwsgi_error("read()");
return -1;
}
else if (cnt == 0) {
return ret;
}
// send to peer
if (write(fd, buffer, cnt) != cnt) {
uwsgi_error("write()");
return -1;
}
ret += cnt;
plen -= cnt;
}
}
ret = uwsgi_pipe_sized(pfd, fd, timeout, plen);
if (ret < 0) return -1;
}
core/utils.c
+150 -268
View File
@@ -76,6 +76,7 @@ uint64_t uwsgi_swap64(uint64_t x) {
#endif
// check if a string is a valid hex number
int check_hex(char *str, int len) {
int i;
for (i = 0; i < len; i++) {
@@ -89,6 +90,7 @@ int check_hex(char *str, int len) {
}
// increase worker harakiri
void inc_harakiri(int sec) {
if (uwsgi.master_process) {
uwsgi.workers[uwsgi.mywid].harakiri += sec;
@@ -98,6 +100,7 @@ void inc_harakiri(int sec) {
}
}
// set worker harakiri
void set_harakiri(int sec) {
if (sec == 0) {
uwsgi.workers[uwsgi.mywid].harakiri = 0;
@@ -110,6 +113,7 @@ void set_harakiri(int sec) {
}
}
// set user harakiri
void set_user_harakiri(int sec) {
if (!uwsgi.master_process) {
uwsgi_log("!!! unable to set user harakiri without the master process !!!\n");
@@ -123,6 +127,7 @@ void set_user_harakiri(int sec) {
}
}
// set mule harakiri
void set_mule_harakiri(int sec) {
if (sec == 0) {
uwsgi.mules[uwsgi.muleid - 1].harakiri = 0;
@@ -136,6 +141,7 @@ void set_mule_harakiri(int sec) {
}
#ifdef UWSGI_SPOOLER
// set spooler harakiri
void set_spooler_harakiri(int sec) {
if (sec == 0) {
uwsgi.i_am_a_spooler->harakiri = 0;
@@ -150,6 +156,7 @@ void set_spooler_harakiri(int sec) {
#endif
// daemonize to the specified logfile
void daemonize(char *logfile) {
pid_t pid;
int fdin;
@@ -212,158 +219,7 @@ void daemonize(char *logfile) {
logto(logfile);
}
void logto(char *logfile) {
int fd;
#ifdef UWSGI_UDP
char *udp_port;
struct sockaddr_in udp_addr;
udp_port = strchr(logfile, ':');
if (udp_port) {
udp_port[0] = 0;
if (!udp_port[1] || !logfile[0]) {
uwsgi_log("invalid udp address\n");
exit(1);
}
fd = socket(AF_INET, SOCK_DGRAM, 0);
if (fd < 0) {
uwsgi_error("socket()");
exit(1);
}
memset(&udp_addr, 0, sizeof(struct sockaddr_in));
udp_addr.sin_family = AF_INET;
udp_addr.sin_port = htons(atoi(udp_port + 1));
char *resolved = uwsgi_resolve_ip(logfile);
if (resolved) {
udp_addr.sin_addr.s_addr = inet_addr(resolved);
}
else {
udp_addr.sin_addr.s_addr = inet_addr(logfile);
}
if (connect(fd, (const struct sockaddr *) &udp_addr, sizeof(struct sockaddr_in)) < 0) {
uwsgi_error("connect()");
exit(1);
}
}
else {
#endif
if (uwsgi.log_truncate) {
fd = open(logfile, O_RDWR | O_CREAT | O_TRUNC, S_IRUSR | S_IWUSR | S_IRGRP);
}
else {
fd = open(logfile, O_RDWR | O_CREAT | O_APPEND, S_IRUSR | S_IWUSR | S_IRGRP);
}
if (fd < 0) {
uwsgi_error_open(logfile);
exit(1);
}
uwsgi.logfile = logfile;
if (uwsgi.chmod_logfile_value) {
if (chmod(uwsgi.logfile, uwsgi.chmod_logfile_value)) {
uwsgi_error("chmod()");
}
}
#ifdef UWSGI_UDP
}
#endif
/* stdout */
if (fd != 1) {
if (dup2(fd, 1) < 0) {
uwsgi_error("dup2()");
exit(1);
}
close(fd);
}
/* stderr */
if (dup2(1, 2) < 0) {
uwsgi_error("dup2()");
exit(1);
}
}
#ifdef UWSGI_ZEROMQ
ssize_t uwsgi_zeromq_logger(struct uwsgi_logger *ul, char *message, size_t len) {
if (!ul->configured) {
if (!ul->arg) {
uwsgi_log_safe("invalid zeromq syntax\n");
exit(1);
}
void *ctx = zmq_init(1);
if (ctx == NULL) {
uwsgi_error_safe("zmq_init()");
exit(1);
}
ul->data = zmq_socket(ctx, ZMQ_PUSH);
if (ul->data == NULL) {
uwsgi_error_safe("zmq_socket()");
exit(1);
}
if (zmq_connect(ul->data, ul->arg) < 0) {
uwsgi_error_safe("zmq_connect()");
exit(1);
}
ul->configured = 1;
}
zmq_msg_t msg;
if (zmq_msg_init_size(&msg, len) == 0) {
memcpy(zmq_msg_data(&msg), message, len);
#if ZMQ_VERSION >= ZMQ_MAKE_VERSION(3,0,0)
zmq_sendmsg(ul->data, &msg, 0);
#else
zmq_send(ul->data, &msg, 0);
#endif
zmq_msg_close(&msg);
}
return 0;
}
#endif
void create_logpipe(void) {
#if defined(SOCK_SEQPACKET) && defined(__linux__)
if (socketpair(AF_UNIX, SOCK_SEQPACKET, 0, uwsgi.shared->worker_log_pipe)) {
#else
if (socketpair(AF_UNIX, SOCK_DGRAM, 0, uwsgi.shared->worker_log_pipe)) {
#endif
uwsgi_error("socketpair()\n");
exit(1);
}
uwsgi_socket_nb(uwsgi.shared->worker_log_pipe[0]);
uwsgi_socket_nb(uwsgi.shared->worker_log_pipe[1]);
if (uwsgi.shared->worker_log_pipe[1] != 1) {
if (dup2(uwsgi.shared->worker_log_pipe[1], 1) < 0) {
uwsgi_error("dup2()");
exit(1);
}
}
if (dup2(1, 2) < 0) {
uwsgi_error("dup2()");
exit(1);
}
}
// get current working directory
char *uwsgi_get_cwd() {
// set this to static to avoid useless reallocations in stats mode
@@ -386,6 +242,7 @@ char *uwsgi_get_cwd() {
}
// generate internal server error message
void internal_server_error(struct wsgi_request *wsgi_req, char *message) {
if (uwsgi.wsgi_req->headers_size == 0) {
@@ -402,6 +259,7 @@ void internal_server_error(struct wsgi_request *wsgi_req, char *message) {
uwsgi.wsgi_req->response_size += wsgi_req->socket->proto_write(wsgi_req, message, strlen(message));
}
// check if a string_list containes an item
struct uwsgi_string_list *uwsgi_string_list_has_item(struct uwsgi_string_list *list, char *key, size_t keylen) {
struct uwsgi_string_list *usl = list;
while (usl) {
@@ -491,6 +349,7 @@ void uwsgi_set_cgroup() {
}
#endif
// drop privileges (as root)
void uwsgi_as_root() {
@@ -766,6 +625,7 @@ void uwsgi_as_root() {
}
}
// destroy a request
void uwsgi_destroy_request(struct wsgi_request *wsgi_req) {
wsgi_req->socket->proto_close(wsgi_req);
@@ -783,6 +643,7 @@ void uwsgi_destroy_request(struct wsgi_request *wsgi_req) {
}
// finalize/close/free a request
void uwsgi_close_request(struct wsgi_request *wsgi_req) {
int waitpid_status;
@@ -979,6 +840,7 @@ void uwsgi_linux_ksm_map(void) {
#endif
#endif
// setup for a new request
void wsgi_req_setup(struct wsgi_request *wsgi_req, int async_id, struct uwsgi_socket *uwsgi_sock) {
wsgi_req->poll.events = POLLIN;
@@ -1044,6 +906,7 @@ int wsgi_req_async_recv(struct wsgi_request *wsgi_req) {
}
#endif
// receive a new request
int wsgi_req_recv(struct wsgi_request *wsgi_req) {
uwsgi.workers[uwsgi.mywid].cores[wsgi_req->async_id].in_request = 1;
@@ -1075,6 +938,7 @@ int wsgi_req_recv(struct wsgi_request *wsgi_req) {
}
// accept a new request
int wsgi_req_simple_accept(struct wsgi_request *wsgi_req, int fd) {
wsgi_req->poll.fd = wsgi_req->socket->proto_accept(wsgi_req, fd);
@@ -1091,6 +955,7 @@ int wsgi_req_simple_accept(struct wsgi_request *wsgi_req, int fd) {
return 0;
}
// send heartbeat to the emperor
void uwsgi_heartbeat() {
if (!uwsgi.has_emperor) return;
@@ -1106,6 +971,7 @@ void uwsgi_heartbeat() {
}
// accept a request
int wsgi_req_accept(int queue, struct wsgi_request *wsgi_req) {
int ret;
@@ -1217,6 +1083,7 @@ int wsgi_req_accept(int queue, struct wsgi_request *wsgi_req) {
return -1;
}
// fix related options
void sanitize_args() {
if (uwsgi.async > 1) {
@@ -1256,6 +1123,7 @@ void sanitize_args() {
}
}
// translate a OS env to a uWSGI option
void env_to_arg(char *src, char *dst) {
int i;
int val = 0;
@@ -1278,6 +1146,7 @@ void env_to_arg(char *src, char *dst) {
dst[strlen(src)] = 0;
}
// lower a string
char *uwsgi_lower(char *str, size_t size) {
size_t i;
for (i = 0; i < size; i++) {
@@ -1287,6 +1156,7 @@ char *uwsgi_lower(char *str, size_t size) {
return str;
}
// parse OS envs
void parse_sys_envs(char **envs) {
char **uenvs = envs;
@@ -1309,101 +1179,7 @@ void parse_sys_envs(char **envs) {
}
//use this instead of fprintf to avoid buffering mess with udp logging
void uwsgi_log(const char *fmt, ...) {
va_list ap;
char logpkt[4096];
int rlen = 0;
int ret;
struct timeval tv;
char sftime[64];
char ctime_storage[26];
time_t now;
if (uwsgi.logdate) {
if (uwsgi.log_strftime) {
now = uwsgi_now();
rlen = strftime(sftime, 64, uwsgi.log_strftime, localtime(&now));
memcpy(logpkt, sftime, rlen);
memcpy(logpkt + rlen, " - ", 3);
rlen += 3;
}
else {
gettimeofday(&tv, NULL);
#ifdef __sun__
ctime_r((const time_t *) &tv.tv_sec, ctime_storage, 26);
#else
ctime_r((const time_t *) &tv.tv_sec, ctime_storage);
#endif
memcpy(logpkt, ctime_storage, 24);
memcpy(logpkt + 24, " - ", 3);
rlen = 24 + 3;
}
}
va_start(ap, fmt);
ret = vsnprintf(logpkt + rlen, 4096 - rlen, fmt, ap);
va_end(ap);
if (ret >= 4096) {
char *tmp_buf = uwsgi_malloc(rlen + ret + 1);
memcpy(tmp_buf, logpkt, rlen);
va_start(ap, fmt);
ret = vsnprintf(tmp_buf + rlen, ret + 1, fmt, ap);
va_end(ap);
rlen = write(2, tmp_buf, rlen + ret);
free(tmp_buf);
return;
}
rlen += ret;
// do not check for errors
rlen = write(2, logpkt, rlen);
}
void uwsgi_log_verbose(const char *fmt, ...) {
va_list ap;
char logpkt[4096];
int rlen = 0;
struct timeval tv;
char sftime[64];
time_t now;
char ctime_storage[26];
if (uwsgi.log_strftime) {
now = uwsgi_now();
rlen = strftime(sftime, 64, uwsgi.log_strftime, localtime(&now));
memcpy(logpkt, sftime, rlen);
memcpy(logpkt + rlen, " - ", 3);
rlen += 3;
}
else {
gettimeofday(&tv, NULL);
#ifdef __sun__
ctime_r((const time_t *) &tv.tv_sec, ctime_storage, 26);
#else
ctime_r((const time_t *) &tv.tv_sec, ctime_storage);
#endif
memcpy(logpkt, ctime_storage, 24);
memcpy(logpkt + 24, " - ", 3);
rlen = 24 + 3;
}
va_start(ap, fmt);
rlen += vsnprintf(logpkt + rlen, 4096 - rlen, fmt, ap);
va_end(ap);
// do not check for errors
rlen = write(2, logpkt, rlen);
}
// check if a string is contained in another one
char *uwsgi_str_contains(char *str, int slen, char what) {
int i;
@@ -1415,6 +1191,7 @@ char *uwsgi_str_contains(char *str, int slen, char what) {
return NULL;
}
// fast compare 2 sized strings
inline int uwsgi_strncmp(char *src, int slen, char *dst, int dlen) {
if (slen != dlen)
@@ -1424,6 +1201,7 @@ inline int uwsgi_strncmp(char *src, int slen, char *dst, int dlen) {
}
// fast sized check of initial part of a string
inline int uwsgi_starts_with(char *src, int slen, char *dst, int dlen) {
if (slen < dlen)
@@ -1432,6 +1210,7 @@ inline int uwsgi_starts_with(char *src, int slen, char *dst, int dlen) {
return memcmp(src, dst, dlen);
}
// unsized check
inline int uwsgi_startswith(char *src, char *what, int wlen) {
int i;
@@ -1444,6 +1223,7 @@ inline int uwsgi_startswith(char *src, char *what, int wlen) {
return 0;
}
// concatenate strings
char *uwsgi_concatn(int c, ...) {
va_list s;
@@ -1612,6 +1392,7 @@ char *uwsgi_concat4n(char *one, int s1, char *two, int s2, char *three, int s3,
// concat unsized strings
char *uwsgi_concat(int c, ...) {
va_list s;
@@ -1670,6 +1451,7 @@ char *uwsgi_strncopy(char *s, int len) {
}
// get the application id
int uwsgi_get_app_id(char *app_name, int app_name_len, int modifier1) {
int i;
@@ -2336,13 +2118,10 @@ add:
}
int uwsgi_waitfd(int fd, int timeout) {
int uwsgi_waitfd_event(int fd, int timeout, int event) {
int ret;
struct pollfd upoll[1];
char oob;
ssize_t rlen;
struct pollfd upoll;
if (!timeout)
timeout = uwsgi.shared->options[UWSGI_OPTION_SOCKET_TIMEOUT];
@@ -2351,34 +2130,24 @@ int uwsgi_waitfd(int fd, int timeout) {
if (timeout < 0)
timeout = -1;
upoll[0].fd = fd;
upoll[0].events = POLLIN | POLLPRI;
upoll[0].revents = 0;
ret = poll(upoll, 1, timeout);
upoll.fd = fd;
upoll.events = event;
upoll.revents = 0;
ret = poll(&upoll, 1, timeout);
if (ret < 0) {
uwsgi_error("poll()");
}
else if (ret > 0) {
if (upoll[0].revents & POLLIN) {
if (upoll.revents & event) {
return ret;
}
if (upoll[0].revents & POLLPRI) {
uwsgi_log("DETECTED PRI DATA\n");
rlen = recv(fd, &oob, 1, MSG_OOB);
uwsgi_log("RECEIVE OOB DATA %d !!!\n", rlen);
if (rlen < 0) {
return -1;
}
return 0;
}
return -1;
}
return ret;
}
inline void *uwsgi_malloc(size_t size) {
char *ptr = malloc(size);
@@ -4775,6 +4544,119 @@ char *uwsgi_sanitize_cert_filename(char *base, char *key, uint16_t keylen) {
#endif
ssize_t uwsgi_pipe(int src, int dst, int timeout) {
char buf[8192];
size_t written = -1;
ssize_t len;
for(;;) {
int ret = uwsgi_waitfd(src, timeout);
if (ret > 0) {
len = read(src, buf, 8192);
if (len == 0) {
return written;
}
else if (len < 0) {
uwsgi_error("read()");
return -1;
}
size_t remains = len;
while(remains > 0) {
int ret = uwsgi_waitfd_write(dst, timeout);
if (ret > 0) {
len = write(dst, buf, remains);
if (len > 0) {
remains-=len;
written+=len;
}
else if (len == 0) {
return written;
}
else {
uwsgi_error("write()");
return -1;
}
}
else if (ret == 0) {
goto timeout;
}
else {
return -1;
}
}
}
else if (ret == 0) {
goto timeout;
}
else {
return -1;
}
}
return written;
timeout:
uwsgi_log("timeout while piping from %d to %d !!!\n", src, dst);
return -1;
}
ssize_t uwsgi_pipe_sized(int src, int dst, size_t required, int timeout) {
char buf[8192];
size_t written = -1;
ssize_t len;
while(written < required) {
int ret = uwsgi_waitfd(src, timeout);
if (ret > 0) {
len = read(src, buf, UMIN(8192, required-written));
if (len == 0) {
return written;
}
else if (len < 0) {
uwsgi_error("read()");
return -1;
}
size_t remains = len;
while(remains > 0) {
int ret = uwsgi_waitfd_write(dst, timeout);
if (ret > 0) {
len = write(dst, buf, remains);
if (len > 0) {
remains-=len;
written+=len;
}
else if (len == 0) {
return written;
}
else {
uwsgi_error("write()");
return -1;
}
}
else if (ret == 0) {
goto timeout;
}
else {
return -1;
}
}
}
else if (ret == 0) {
goto timeout;
}
else {
return -1;
}
}
return written;
timeout:
uwsgi_log("timeout while piping from %d to %d !!!\n", src, dst);
return -1;
}
void uwsgi_set_cpu_affinity() {
char buf[4096];
int ret;
core/uwsgi.c
+1 -1
View File
@@ -2594,7 +2594,7 @@ next:
#ifdef UWSGI_ZEROMQ
// setup zeromq context (if required) one per-worker
if (uwsgi.zeromq) {
uwsgi_zeromq_init();
uwsgi_zeromq_init_sockets();
}
#endif
plugins/router_http/router_http.c
+93
View File
@@ -0,0 +1,93 @@
#include "../../uwsgi.h"
#ifdef UWSGI_ROUTING
extern struct uwsgi_server uwsgi;
int uwsgi_routing_func_http(struct wsgi_request *wsgi_req, struct uwsgi_route *ur) {
// mark a route request
wsgi_req->status = -17;
// get the http address from the route
char *addr = ur->data;
// connect to the http server
int http_fd = uwsgi_connect(addr, uwsgi.shared->options[UWSGI_OPTION_SOCKET_TIMEOUT], 0);
if (http_fd < 0) {
uwsgi_log("unable to connect to host %s\n", addr);
return UWSGI_ROUTE_NEXT;
}
// convert the wsgi_request to an http proxy request
struct uwsgi_buffer *ub = uwsgi_to_http(wsgi_req);
if (!ub) {
uwsgi_log("unable to generate http request for %s\n", addr);
close(http_fd);
return UWSGI_ROUTE_NEXT;
}
// send the request
if (uwsgi_buffer_send(ub, http_fd)) {
uwsgi_log("error routing request to http server %s\n", addr);
close(http_fd);
uwsgi_buffer_destroy(ub);
return UWSGI_ROUTE_NEXT;
}
ssize_t ret;
// pipe the body
if (wsgi_req->post_cl > 0) {
int post_fd = wsgi_req->poll.fd;
if (wsgi_req->async_post) {
post_fd = fileno(wsgi_req->async_post);
}
ret = uwsgi_pipe_sized(post_fd, http_fd, wsgi_req->post_cl, 0);
if (ret < 0) {
uwsgi_log("error routing request body (%llu bytes) to http server %s\n", (unsigned long long) wsgi_req->post_cl, addr);
close(http_fd);
uwsgi_buffer_destroy(ub);
return UWSGI_ROUTE_BREAK;
}
}
// pipe the response
ret = uwsgi_pipe(http_fd, wsgi_req->poll.fd, 0);
if (ret > 0) {
wsgi_req->response_size += ret;
}
else {
uwsgi_log("error routing request to http server %s\n", addr);
}
close(http_fd);
uwsgi_buffer_destroy(ub);
return UWSGI_ROUTE_BREAK;
}
int uwsgi_router_http(struct uwsgi_route *ur, char *args) {
ur->func = uwsgi_routing_func_http;
ur->data = (void *) args;
ur->data_len = strlen(args);
return 0;
}
void router_http_register(void) {
uwsgi_register_router("http", uwsgi_router_http);
}
struct uwsgi_plugin router_http_plugin = {
.name = "router_http",
.on_load = router_http_register,
};
#else
struct uwsgi_plugin router_http_plugin = {
.name = "router_http",
};
#endif
plugins/router_http/uwsgiplugin.py
+6
View File
@@ -0,0 +1,6 @@
NAME='router_http'
CFLAGS = []
LDFLAGS = []
LIBS = []
GCC_LIST = ['router_http']
plugins/router_uwsgi/router_uwsgi.c
+6 -24
View File
@@ -33,8 +33,6 @@ int uwsgi_routing_func_uwsgi_simple(struct wsgi_request *wsgi_req, struct uwsgi_
int uwsgi_routing_func_uwsgi_remote(struct wsgi_request *wsgi_req, struct uwsgi_route *ur) {
char buf[8192];
ssize_t len;
struct uwsgi_header *uh = (struct uwsgi_header *) ur->data;
char *addr = ur->data + sizeof(struct uwsgi_header);
@@ -62,29 +60,13 @@ int uwsgi_routing_func_uwsgi_remote(struct wsgi_request *wsgi_req, struct uwsgi_
return UWSGI_ROUTE_NEXT;
}
for(;;) {
int ret = uwsgi_waitfd(uwsgi_fd, uwsgi.shared->options[UWSGI_OPTION_SOCKET_TIMEOUT]);
if (ret > 0) {
len = read(uwsgi_fd, buf, 8192);
if (len == 0) {
break;
}
else if (len < 0) {
uwsgi_error("read()");
break;
}
if (write(wsgi_req->poll.fd, buf, len) != len) {
uwsgi_error("write()");
break;
}
}
else {
uwsgi_log("timeout !!!\n");
break;
}
ssize_t ret = uwsgi_pipe(uwsgi_fd, wsgi_req->poll.fd, 0);
if (ret > 0) {
wsgi_req->response_size += ret;
}
else {
uwsgi_log("unable to manage uwsgi route response for %s\n", addr);
}
close(uwsgi_fd);
return UWSGI_ROUTE_BREAK;
proto/http.c
+85
View File
@@ -369,3 +369,88 @@ int uwsgi_proto_http_parser(struct wsgi_request *wsgi_req) {
return UWSGI_AGAIN;
}
void uwsgi_httpize_var(char *buf, size_t len) {
size_t i;
int upper = 1;
for(i=0;i<len;i++) {
if (upper) {
upper = 0;
continue;
}
if (buf[i] == '_') {
buf[i] = '-';
upper = 1;
continue;
}
buf[i] = tolower( (int) buf[i]);
}
}
struct uwsgi_buffer *uwsgi_to_http(struct wsgi_request *wsgi_req) {
struct uwsgi_buffer *ub = uwsgi_buffer_new(4096);
if (uwsgi_buffer_append(ub, wsgi_req->method, wsgi_req->method_len)) goto clear;
if (uwsgi_buffer_append(ub, " ", 1)) goto clear;
if (uwsgi_buffer_append(ub, wsgi_req->uri, wsgi_req->uri_len)) goto clear;
if (uwsgi_buffer_append(ub, " ", 1)) goto clear;
if (uwsgi_buffer_append(ub, wsgi_req->protocol, wsgi_req->protocol_len)) goto clear;
if (uwsgi_buffer_append(ub, "\r\n", 2)) goto clear;
int i;
char *x_forwarded_for = NULL;
size_t x_forwarded_for_len = 0;
// starting adding headers
for(i=0;i<wsgi_req->var_cnt;i++) {
if (!uwsgi_starts_with(wsgi_req->hvec[i].iov_base, wsgi_req->hvec[i].iov_len, "HTTP_", 5)) {
char *header = wsgi_req->hvec[i].iov_base+5;
size_t header_len = wsgi_req->hvec[i].iov_len-5;
if (!uwsgi_strncmp(header, header_len, "CONNECTION", 10)) goto next;
if (!uwsgi_strncmp(header, header_len, "KEEP_ALIVE", 10)) goto next;
if (!uwsgi_strncmp(header, header_len, "X_FORWARDED_FOR", 15)) {
x_forwarded_for = wsgi_req->hvec[i+1].iov_base;
x_forwarded_for_len = wsgi_req->hvec[i+1].iov_len;
goto next;
}
if (uwsgi_buffer_append(ub, header, header_len)) goto clear;
// transofmr uwsgi var to http header
uwsgi_httpize_var((ub->buf+ub->pos) - header_len, header_len);
if (uwsgi_buffer_append(ub, ": ", 2)) goto clear;
if (uwsgi_buffer_append(ub, wsgi_req->hvec[i+1].iov_base, wsgi_req->hvec[i+1].iov_len)) goto clear;
if (uwsgi_buffer_append(ub, "\r\n", 2)) goto clear;
}
next:
i++;
}
// append required headers
if (uwsgi_buffer_append(ub, "Connection: close\r\n", 19)) goto clear;
if (uwsgi_buffer_append(ub, "X-Forwarded-For: ", 17)) goto clear;
if (x_forwarded_for_len > 0) {
if (uwsgi_buffer_append(ub, x_forwarded_for, x_forwarded_for_len)) goto clear;
if (uwsgi_buffer_append(ub, ", ", 2)) goto clear;
}
if (uwsgi_buffer_append(ub, wsgi_req->remote_addr, wsgi_req->remote_addr_len)) goto clear;
if (uwsgi_buffer_append(ub, "\r\n\r\n", 4)) goto clear;
return ub;
clear:
uwsgi_buffer_destroy(ub);
return NULL;
}
proto/zeromq.c
+12 -4
View File
@@ -2,15 +2,23 @@
extern struct uwsgi_server uwsgi;
void uwsgi_zeromq_init() {
uwsgi.zmq_context = zmq_init(1);
void *uwsgi_zeromq_init() {
if (!uwsgi.zmq_context) {
uwsgi.zmq_context = zmq_init(1);
if (uwsgi.zmq_context == NULL) {
uwsgi_error("zmq_init()");
exit(1);
}
}
return uwsgi.zmq_context;
}
struct uwsgi_socket *uwsgi_sock = uwsgi.sockets;
while(uwsgi_sock) {
void uwsgi_zeromq_init_sockets() {
uwsgi_zeromq_init();
struct uwsgi_socket *uwsgi_sock = uwsgi.sockets;
while(uwsgi_sock) {
if (!uwsgi_sock->proto_name || strcmp(uwsgi_sock->proto_name, "zmq")) {
goto zmq_next;
}
uwsgi.h
+23 -2
View File
@@ -289,6 +289,11 @@ extern int pivot_root(const char *new_root, const char *put_old);
#endif
struct uwsgi_buffer {
char *buf;
off_t pos;
size_t len;
};
struct uwsgi_string_list {
@@ -2265,7 +2270,9 @@ void add_exported_option(char *, char *, int);
ssize_t uwsgi_send_empty_pkt(int, char *, uint8_t, uint8_t);
int uwsgi_waitfd(int, int);
int uwsgi_waitfd_event(int, int, int);
#define uwsgi_waitfd(a, b) uwsgi_waitfd_event(a, b, POLLIN)
#define uwsgi_waitfd_write(a, b) uwsgi_waitfd_event(a, b, POLLOUT)
int uwsgi_hooked_parse_dict_dgram(int, char *, size_t, uint8_t, uint8_t, void (*)(char *, uint16_t, char *, uint16_t, void *), void *);
int uwsgi_hooked_parse(char *, size_t, void (*)(char *, uint16_t, char *, uint16_t, void *), void *);
@@ -2283,6 +2290,7 @@ int uwsgi_cluster_add_me(void);
char *generate_socket_name(char *);
#define UMIN(a,b) ((a)>(b)?(b):(a))
#define UMAX(a,b) ((a)<(b)?(b):(a))
ssize_t uwsgi_send_message(int, uint8_t, uint8_t, char *, uint16_t, int, ssize_t, int);
@@ -2627,7 +2635,8 @@ ssize_t uwsgi_proto_zeromq_write(struct wsgi_request *, char *, size_t);
ssize_t uwsgi_proto_zeromq_write_header(struct wsgi_request *, char *, size_t);
ssize_t uwsgi_proto_zeromq_sendfile(struct wsgi_request *);
int uwsgi_proto_zeromq_parser(struct wsgi_request *);
void uwsgi_zeromq_init(void);
void *uwsgi_zeromq_init(void);
void uwsgi_zeromq_init_sockets(void);
#endif
int uwsgi_num2str2(int, char *);
@@ -3202,6 +3211,18 @@ void uwsgi_emperor_start(void);
void uwsgi_bind_sockets(void);
void uwsgi_set_sockets_protocols(void);
struct uwsgi_buffer *uwsgi_buffer_new(size_t);
int uwsgi_buffer_append(struct uwsgi_buffer *, char *, size_t);
void uwsgi_buffer_destroy(struct uwsgi_buffer *);
void uwsgi_httpize_var(char *, size_t);
struct uwsgi_buffer *uwsgi_to_http(struct wsgi_request *);
ssize_t uwsgi_pipe(int, int, int);
ssize_t uwsgi_pipe_sized(int, int, size_t, int);
int uwsgi_buffer_send(struct uwsgi_buffer *, int);
void uwsgi_check_emperor(void);
#ifdef UWSGI_AS_SHARED_LIBRARY
int uwsgi_init(int, char **, char **);
uwsgiconfig.py
+1 -1
View File
@@ -357,7 +357,7 @@ class uConf(object):
self.config.read(filename)
self.gcc_list = ['core/utils', 'core/protocol', 'core/socket', 'core/logging', 'core/master', 'core/master_utils', 'core/emperor',
'core/notify', 'core/mule', 'core/subscription', 'core/stats',
'core/setup_utils', 'core/clock', 'core/init',
'core/setup_utils', 'core/clock', 'core/init', 'core/buffer',
'core/plugins', 'core/lock', 'core/cache',
'core/queue', 'core/event', 'core/signal', 'core/cluster',
'core/rpc', 'core/gateway', 'core/loop', 'lib/rbtree', 'core/rb_timers', 'core/uwsgi']