From f82dbfd3005270276dde67730856216467ec915a Mon Sep 17 00:00:00 2001 From: "roberto@maverick64" Date: Tue, 25 Jan 2011 16:20:11 +0100 Subject: [PATCH] support BROADCAST for clustering --- master.c | 18 +++++++--- plugins/python/uwsgi_pymodule.c | 21 +++++++++++- protocol.c | 2 ++ socket.c | 17 +++++++-- utils.c | 61 +++++++++++++++++++++++++++++++++ uwsgi.c | 50 +++++++++++++++++++++++++-- uwsgi.h | 28 ++++++++++++++- 7 files changed, 185 insertions(+), 12 deletions(-) diff --git a/master.c b/master.c index 1496ae86..460df74f 100644 --- a/master.c +++ b/master.c @@ -124,7 +124,7 @@ void master_loop(char **argv, char **environ) { uwsgi.wsgi_req->buffer = uwsgi.async_buf[0]; #ifdef UWSGI_UDP if (uwsgi.udp_socket) { - udp_fd = bind_to_udp(uwsgi.udp_socket, 0); + udp_fd = bind_to_udp(uwsgi.udp_socket, 0, 0); uwsgi_poll[uwsgi_poll_size].fd = udp_fd; if (uwsgi_poll[uwsgi_poll_size].fd < 0) { uwsgi_log( "unable to bind to udp socket. SNMP and cluster management services will be disabled.\n"); @@ -271,13 +271,13 @@ void master_loop(char **argv, char **environ) { uwsgi_log( "running %s\n", uwsgi.binary_path); argv[0] = uwsgi.binary_path; //strcpy (argv[0], uwsgi.binary_path); - execv(uwsgi.binary_path, argv); - uwsgi_error("execv()"); + execvp(uwsgi.binary_path, argv); + uwsgi_error("execvp()"); // never here exit(1); } - if (uwsgi.numproc > 0 || uwsgi.gateways_cnt > 0) { + if (uwsgi.numproc > 0 || uwsgi.gateways_cnt > 0 || ushared->daemons_cnt > 0) { master_has_children = 1; } #ifdef UWSGI_SPOOLER @@ -323,6 +323,16 @@ void master_loop(char **argv, char **environ) { } } + // add unregistered daemons + // locking is not needed as daemons can only increase (for now) + for(i=0;idaemons_cnt;i++) { + if (!ushared->daemons[i].registered) { + uwsgi_log("running daemon %s\n", ushared->daemons[i].command); + spawn_daemon(&ushared->daemons[i]); + ushared->daemons[i].registered = 1; + } + } + // add unregistered timers // locking is not needed as monitors can only increase diff --git a/plugins/python/uwsgi_pymodule.c b/plugins/python/uwsgi_pymodule.c index 8529c324..bd59e2e8 100644 --- a/plugins/python/uwsgi_pymodule.c +++ b/plugins/python/uwsgi_pymodule.c @@ -352,7 +352,7 @@ PyObject *py_uwsgi_register_rpc(PyObject * self, PyObject * args) { char *name; PyObject *func; - if (!PyArg_ParseTuple(args, "sO|B:register_signal", &name, &func, &argc)) { + if (!PyArg_ParseTuple(args, "sO|B:register_rpc", &name, &func, &argc)) { return NULL; } @@ -366,6 +366,23 @@ PyObject *py_uwsgi_register_rpc(PyObject * self, PyObject * args) { return Py_True; } +PyObject *py_uwsgi_attach_daemon(PyObject * self, PyObject * args) { + + char *command = NULL; + + if (!PyArg_ParseTuple(args, "s:attach_daemon", &command)) { + return NULL; + } + + if (uwsgi_attach_daemon(command)) { + Py_INCREF(Py_None); + return Py_None; + } + + Py_INCREF(Py_True); + return Py_True; +} + PyObject *py_uwsgi_register_signal(PyObject * self, PyObject * args) { uint8_t uwsgi_signal; @@ -2251,6 +2268,8 @@ static PyMethodDef uwsgi_advanced_methods[] = { {"send", py_uwsgi_send, METH_VARARGS, ""}, {"cl", py_uwsgi_cl, METH_VARARGS, ""}, + {"attach_daemon", py_uwsgi_attach_daemon, METH_VARARGS, ""}, + {"register_signal", py_uwsgi_register_signal, METH_VARARGS, ""}, {"signal", py_uwsgi_signal, METH_VARARGS, ""}, {"register_file_monitor", py_uwsgi_register_file_monitor, METH_VARARGS, ""}, diff --git a/protocol.c b/protocol.c index 1d9cf4cf..8867fd48 100644 --- a/protocol.c +++ b/protocol.c @@ -934,6 +934,8 @@ uint16_t fcgi_get_record(int fd, char *buf) { ptr += len; } + if (fr.type != 6) return 0; + return ntohs(*rs); } diff --git a/socket.c b/socket.c index 108472cf..38f248d3 100644 --- a/socket.c +++ b/socket.c @@ -132,10 +132,11 @@ int bind_to_unix(char *socket_name, int listen_queue, int chmod_socket, int abst #endif #ifdef UWSGI_UDP - int bind_to_udp(char *socket_name, int multicast) { + int bind_to_udp(char *socket_name, int multicast, int broadcast) { int serverfd; struct sockaddr_in uws_addr; char *udp_port; + int bcast = 1; #ifdef UWSGI_MULTICAST struct ip_mreq mc; @@ -157,7 +158,10 @@ int bind_to_unix(char *socket_name, int listen_queue, int chmod_socket, int abst uws_addr.sin_family = AF_INET; uws_addr.sin_port = htons(atoi(udp_port + 1)); - if (socket_name[0] != 0) { + if (broadcast) { + uws_addr.sin_addr.s_addr = INADDR_BROADCAST; + } + else if (socket_name[0] != 0) { uws_addr.sin_addr.s_addr = inet_addr(socket_name); } else { @@ -165,7 +169,6 @@ int bind_to_unix(char *socket_name, int listen_queue, int chmod_socket, int abst } - serverfd = socket(AF_INET, SOCK_DGRAM, 0); if (serverfd < 0) { uwsgi_error("socket()"); @@ -181,6 +184,14 @@ int bind_to_unix(char *socket_name, int listen_queue, int chmod_socket, int abst } #endif + if (broadcast) { + if (setsockopt(serverfd, SOL_SOCKET, SO_BROADCAST, &bcast, sizeof(bcast))) { + perror("setsockopt"); + close(serverfd); + return -1; + } + } + if (bind(serverfd, (struct sockaddr *) &uws_addr, sizeof(uws_addr)) != 0) { uwsgi_error("bind()"); close(serverfd); diff --git a/utils.c b/utils.c index a6cadb5a..8af75084 100644 --- a/utils.c +++ b/utils.c @@ -1403,3 +1403,64 @@ char *uwsgi_get_last_char(char *what, char c) { return ptr; } + +int uwsgi_attach_daemon(char *command) { + + struct uwsgi_daemon *d; + int ret = -1; + + uwsgi_lock(uwsgi.daemon_table_lock); + + if (uwsgi.shared->daemons_cnt < MAX_DAEMONS) { + d = &uwsgi.shared->daemons[uwsgi.shared->daemons_cnt]; + + memcpy(d->command, command, UMIN(strlen(command), 0xff-1)); + d->registered = 0; + d->status = 0; + + uwsgi.shared->daemons_cnt++; + + ret = 0; + uwsgi_log("registered daemon %s\n", command); + } + + uwsgi_unlock(uwsgi.daemon_table_lock); + + return ret; + +} + +void spawn_daemon(struct uwsgi_daemon *ud) { + + char *argv[2]; + pid_t pid = fork(); + if (pid < 0) { + uwsgi_error("fork()"); + return; + } + + if (pid > 0) { + ud->pid = pid; + ud->status = 1; + if (ud->respawns == 0) { + ud->born = time(NULL); + } + + ud->respawns++; + ud->last_spawn = time(NULL); + + } + else { + argv[0] = ud->command; + argv[1] = NULL; + + if (execvp(argv[0], argv)) { + uwsgi_error("execvp()"); + } + + // never here; + exit(1); + } + + return; +} diff --git a/uwsgi.c b/uwsgi.c index ca922143..1ea36b32 100644 --- a/uwsgi.c +++ b/uwsgi.c @@ -157,6 +157,7 @@ static struct option long_base_options[] = { {"ns", required_argument, 0, LONG_ARGS_LINUX_NS}, #endif {"loop", required_argument, 0, LONG_ARGS_LOOP}, + {"worker-exec", required_argument, 0, LONG_ARGS_WORKER_EXEC}, {"plugins", required_argument, 0, LONG_ARGS_PLUGINS}, {"remap-modifier", required_argument, 0, LONG_ARGS_REMAP_MODIFIER}, {"dump-options", no_argument, &uwsgi.dump_options, 1}, @@ -605,8 +606,9 @@ int main(int argc, char *argv[], char *envp[]) if (uwsgi.cluster != NULL) { // get multicast socket - uwsgi_log("CLUSTER: %s\n", uwsgi.cluster); uwsgi.cluster_fd = uwsgi_cluster_join(uwsgi.cluster); + + uwsgi_log("JOINED CLUSTER: %s\n", uwsgi.cluster); // ask for cluster options only if bot pre-existent options are set if (uwsgi.exported_opts_cnt == 1) { @@ -1002,6 +1004,10 @@ int uwsgi_start(void *v_argv) { // timer table lock uwsgi.timer_table_lock = uwsgi_mmap_shared_lock(); uwsgi_lock_init(uwsgi.timer_table_lock); + + // daemons table lock + uwsgi.daemon_table_lock = uwsgi_mmap_shared_lock(); + uwsgi_lock_init(uwsgi.daemon_table_lock); } uwsgi.rpc_table_lock = uwsgi_mmap_shared_lock(); @@ -1499,6 +1505,29 @@ uwsgi.shared->hooks[UWSGI_MODIFIER_PING] = uwsgi_request_ping; //100 master_loop(argv, environ); //from now on the process is a real worker } + + if (uwsgi.worker_exec) { + char *w_argv[2]; + 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_error("fcntl()"); + exit(1); + } + + if (uwsgi.sockets[0].fd != 0) { + if (dup2(uwsgi.sockets[0].fd, 0)) { + uwsgi_error("dup2()"); + } + } + execvp(w_argv[0], w_argv); + // never here + uwsgi_error("execvp()"); + exit(1); + } + for (i = 0; i < 0xFF; i++) { if (uwsgi.p[i]->post_fork) { uwsgi.p[i]->post_fork(); @@ -1684,6 +1713,9 @@ end: case LONG_ARGS_LOOP: uwsgi.loop = optarg; return 1; + case LONG_ARGS_WORKER_EXEC: + uwsgi.worker_exec = optarg; + return 1; case LONG_ARGS_REMAP_MODIFIER: uwsgi.remap_modifier = optarg; return 1; @@ -2484,15 +2516,27 @@ int uwsgi_cluster_join(char *name) { int fd ; char *cp; + int broadcast = 0; - fd = bind_to_udp(name, 1); + if (name[0] == ':') { + fd = bind_to_udp(name, 0, 1); + broadcast = 1; + } + else { + fd = bind_to_udp(name, 1, 0); + } if (fd >= 0) { cp = strchr(name,':'); cp[0] = 0; uwsgi.mc_cluster_addr.sin_family=AF_INET; - uwsgi.mc_cluster_addr.sin_addr.s_addr=inet_addr(name); + if (broadcast) { + uwsgi.mc_cluster_addr.sin_addr.s_addr=INADDR_BROADCAST; + } + else { + uwsgi.mc_cluster_addr.sin_addr.s_addr=inet_addr(name); + } uwsgi.mc_cluster_addr.sin_port=htons(atoi(cp+1)); cp[0] = ':'; diff --git a/uwsgi.h b/uwsgi.h index 1c05b692..ed2243d2 100644 --- a/uwsgi.h +++ b/uwsgi.h @@ -16,6 +16,7 @@ #define MAX_GENERIC_PLUGINS 64 #define MAX_RPC 64 #define MAX_GATEWAYS 64 +#define MAX_DAEMONS 8 #ifndef UWSGI_LOAD_EMBEDDED_PLUGINS #define UWSGI_LOAD_EMBEDDED_PLUGINS @@ -211,6 +212,19 @@ struct uwsgi_gateway { int use_signals; }; + +// Daemons are external processes maintained by the master + +struct uwsgi_daemon { + char command[0xff]; + pid_t pid; + uint64_t respawns; + time_t born; + time_t last_spawn; + int status; + int registered; +}; + // maintain alignment here !!! struct uwsgi_cache_item { @@ -307,6 +321,7 @@ struct uwsgi_opt { #define LONG_ARGS_LOG_SYSLOG 17068 #define LONG_ARGS_LOG_MASTER 17069 #define LONG_ARGS_CHECK_STATIC 17070 +#define LONG_ARGS_WORKER_EXEC 17071 @@ -667,9 +682,11 @@ struct uwsgi_server { char *mode; + char *worker_exec; struct uwsgi_gateway gateways[MAX_GATEWAYS]; int gateways_cnt; + #ifdef UWSGI_HTTP char *http; char *http_server_name; @@ -905,6 +922,7 @@ struct uwsgi_server { struct uwsgi_cache_item *cache_items; void *cache; + void *cache_lock; void *user_lock; void *signal_table_lock; @@ -912,6 +930,8 @@ struct uwsgi_server { void *timer_table_lock; void *rpc_table_lock; + void *daemon_table_lock; + }; @@ -1033,6 +1053,9 @@ struct uwsgi_shared { int rpc_count; int worker_log_pipe[2]; + + struct uwsgi_daemon daemons[MAX_DAEMONS]; + int daemons_cnt; }; struct uwsgi_core { @@ -1088,7 +1111,7 @@ void reload_me(void); void end_me(void); int bind_to_unix(char *, int, int, int); int bind_to_tcp(char *, int, char *); -int bind_to_udp(char *, int); +int bind_to_udp(char *, int, int); int timed_connect(struct pollfd *, const struct sockaddr *, int, int, int); int uwsgi_connect(char *, int, int); int connect_to_tcp(char *, int, int, int); @@ -1411,3 +1434,6 @@ struct fcgi_record { ssize_t fcgi_send_record(int, uint8_t, uint16_t, char *); ssize_t fcgi_send_param(int, char *, uint16_t, char *, uint16_t); uint16_t fcgi_get_record(int, char *); + +int uwsgi_attach_daemon(char *); +void spawn_daemon(struct uwsgi_daemon *);