diff --git a/plugins/gevent/gevent.c b/plugins/gevent/gevent.c index 00ff0df3..ad5d6020 100644 --- a/plugins/gevent/gevent.c +++ b/plugins/gevent/gevent.c @@ -28,6 +28,22 @@ struct uwsgi_gevent { PyObject **watchers; } ugevent; +void uwsgi_opt_setup_gevent(char *opt, char *value, void *null) { + + // set async mode + uwsgi_opt_set_int(opt, value, &uwsgi.async); + // set loop engine + uwsgi.loop = "gevent"; + +} + +struct uwsgi_option gevent_options[] = { + {"gevent", required_argument, 0, "a shortcut enabling gevent loop engine with the specified number of async cores and optimal parameters", uwsgi_opt_setup_gevent, NULL, UWSGI_OPT_THREADS}, + {0, 0, 0, 0, 0, 0, 0}, + +}; + + PyObject *py_uwsgi_gevent_graceful(PyObject *self, PyObject *args) { @@ -120,8 +136,9 @@ PyObject *py_uwsgi_gevent_main(PyObject * self, PyObject * args) { // hack to retrieve the socket address PyObject *py_uwsgi_sock = PyTuple_GetItem(args, 0); struct uwsgi_socket *uwsgi_sock = (struct uwsgi_socket *) PyLong_AsLong(py_uwsgi_sock); - - struct wsgi_request *wsgi_req = find_first_available_wsgi_req(); + struct wsgi_request *wsgi_req = NULL; +edge: + wsgi_req = find_first_available_wsgi_req(); if (wsgi_req == NULL) { uwsgi_log("async queue is full !!!\n"); @@ -154,6 +171,13 @@ PyObject *py_uwsgi_gevent_main(PyObject * self, PyObject * args) { PyObject *new_gl = python_call(ugevent.spawn, ugevent.greenlet_args, 0, NULL); Py_DECREF(new_gl); + if (uwsgi_sock->edge_trigger) { +#ifdef UWSGI_DEBUG + uwsgi_log("i am a edge triggered socket !!!\n"); +#endif + goto edge; + } + clear: Py_INCREF(Py_None); return Py_None; @@ -198,6 +222,9 @@ PyObject *py_uwsgi_gevent_request(PyObject * self, PyObject * args) { if (!timer) goto clear0; for(;;) { +wait: + // if in edge-triggered mode read from socket now, and eventually come back to wait... + if (wsgi_req->socket->edge_trigger) goto edge; // wait for data in the socket PyObject *ret = uwsgi_gevent_wait(watcher, timer, greenlet_switch); if (!ret) goto clear_and_stop; @@ -209,8 +236,13 @@ PyObject *py_uwsgi_gevent_request(PyObject * self, PyObject * args) { goto clear_and_stop; } else if (ret == watcher) { +edge: status = wsgi_req->socket->proto(wsgi_req); if (status < 0) { + // if in edge-triggered, came back to wait-mode + if (wsgi_req->socket->edge_trigger && errno == EAGAIN) { + goto wait; + } goto clear_and_stop; } else if (status == 0) { @@ -419,5 +451,6 @@ int gevent_init() { struct uwsgi_plugin gevent_plugin = { .name = "gevent", + .options = gevent_options, .init = gevent_init, }; diff --git a/setup_utils.c b/setup_utils.c new file mode 100644 index 00000000..9b2f8516 --- /dev/null +++ b/setup_utils.c @@ -0,0 +1,197 @@ +#include "uwsgi.h" + +extern struct uwsgi_server uwsgi; + + +void uwsgi_setup_systemd() { + struct uwsgi_socket *uwsgi_sock = NULL; + int i; + + char *listen_pid = getenv("LISTEN_PID"); + if (listen_pid) { + if (atoi(listen_pid) == (int) getpid()) { + char *listen_fds = getenv("LISTEN_FDS"); + if (listen_fds) { + int systemd_fds = atoi(listen_fds); + if (systemd_fds > 0) { + uwsgi_log("- SystemD socket activation detected -\n"); + for (i = 3; i < 3 + systemd_fds; i++) { + uwsgi_sock = uwsgi_new_socket(NULL); + uwsgi_add_socket_from_fd(uwsgi_sock, i); + } + uwsgi.skip_zero = 1; + } + unsetenv("LISTEN_PID"); + unsetenv("LISTEN_FDS"); + } + } + } + +} + +void uwsgi_setup_upstart() { + + struct uwsgi_socket *uwsgi_sock = NULL; + + char *upstart_events = getenv("UPSTART_EVENTS"); + if (upstart_events && !strcmp(upstart_events, "socket")) { + char *upstart_fds = getenv("UPSTART_FDS"); + if (upstart_fds) { + uwsgi_log("- Upstart socket bridge detected (job: %s) -\n", getenv("UPSTART_JOB")); + uwsgi_sock = uwsgi_new_socket(NULL); + uwsgi_add_socket_from_fd(uwsgi_sock, atoi(upstart_fds)); + uwsgi.skip_zero = 1; + } + unsetenv("UPSTART_EVENTS"); + unsetenv("UPSTART_FDS"); + } + +} + + +void uwsgi_setup_zerg() { + + struct uwsgi_socket *uwsgi_sock = NULL; + int i; + + struct uwsgi_string_list *zn = uwsgi.zerg_node; + while (zn) { + if (uwsgi_zerg_attach(zn->value)) { + if (!uwsgi.zerg_fallback) { + exit(1); + } + } + zn = zn->next; + } + + + + if (uwsgi.zerg) { +#ifdef UWSGI_DEBUG + uwsgi_log("attaching zerg sockets...\n"); +#endif + int zerg_fd; + i = 0; + for (;;) { + zerg_fd = uwsgi.zerg[i]; + if (zerg_fd == -1) { + break; + } + uwsgi_sock = uwsgi_new_socket(NULL); + uwsgi_add_socket_from_fd(uwsgi_sock, zerg_fd); + i++; + } + + uwsgi_log("zerg sockets attached\n"); + } +} + +void uwsgi_setup_inherited_sockets() { + + int j; + union uwsgi_sockaddr usa; + union uwsgi_sockaddr_ptr gsa; + + struct uwsgi_socket *uwsgi_sock = uwsgi.sockets; + while (uwsgi_sock) { + //a bit overengineering + if (uwsgi_sock->name[0] != 0 && !uwsgi_sock->bound) { + for (j = 3; j < (int) uwsgi.max_fd; j++) { + uwsgi_add_socket_from_fd(uwsgi_sock, j); + } + } + uwsgi_sock = uwsgi_sock->next; + } + + //now close all the unbound fd + for (j = 3; j < (int) uwsgi.max_fd; j++) { + int useless = 1; +#ifdef UWSGI_MULTICAST + if (j == uwsgi.cluster_fd) + continue; +#endif + if (uwsgi.has_emperor) { + if (j == uwsgi.emperor_fd) + continue; + } + + if (uwsgi.shared->worker_log_pipe[0] > -1) { + if (j == uwsgi.shared->worker_log_pipe[0]) + continue; + } + + if (uwsgi.shared->worker_log_pipe[1] > -1) { + if (j == uwsgi.shared->worker_log_pipe[1]) + continue; + } + + if (uwsgi.original_log_fd > -1) { + if (j == uwsgi.original_log_fd) + continue; + } + + if (uwsgi.cache_server && uwsgi.cache_server_fd != -1) { + if (j == uwsgi.cache_server_fd) + continue; + } + + struct uwsgi_gateway_socket *ugs = uwsgi.gateway_sockets; + int found = 0; + while (ugs) { + if (ugs->fd == j) { + found = 1; + break; + } + ugs = ugs->next; + } + if (found) + continue; + + + int y; + found = 0; + for (y = 0; y < ushared->gateways_cnt; y++) { + if (ushared->gateways[y].internal_subscription_pipe[0] == j) { + found = 1; + break; + } + if (ushared->gateways[y].internal_subscription_pipe[1] == j) { + found = 1; + break; + } + } + + if (found) + continue; + + + socklen_t socket_type_len = sizeof(struct sockaddr_un); + gsa.sa = (struct sockaddr *) &usa; + if (!getsockname(j, gsa.sa, &socket_type_len)) { + 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) { + uwsgi_sock = uwsgi.shared_sockets; + while (uwsgi_sock) { + if (uwsgi_sock->fd == j && uwsgi_sock->bound) { + useless = 0; + break; + } + uwsgi_sock = uwsgi_sock->next; + } + } + } + + if (useless) { + close(j); + } + } + +} diff --git a/uwsgi.c b/uwsgi.c index 85a03a21..edcbc1f3 100644 --- a/uwsgi.c +++ b/uwsgi.c @@ -736,7 +736,7 @@ void simple_goodbye_cruel_world() { void goodbye_cruel_world() { if (!uwsgi.gbcw_hook) { - simple_goodbye_cruel_world(); + simple_goodbye_cruel_world(); } else { uwsgi.gbcw_hook(); @@ -906,8 +906,9 @@ void grace_them_all(int signum) { } if (uwsgi.auto_snapshot) { - uwsgi.respawn_workers = uwsgi.numproc-uwsgi.auto_snapshot; - if (!uwsgi.respawn_workers) uwsgi.respawn_workers = 1; + uwsgi.respawn_workers = uwsgi.numproc - uwsgi.auto_snapshot; + if (!uwsgi.respawn_workers) + uwsgi.respawn_workers = 1; } } @@ -2332,177 +2333,15 @@ int uwsgi_start(void *v_argv) { // systemd/upstart/zerg socket activation if (!uwsgi.is_a_reload) { - char *listen_pid = getenv("LISTEN_PID"); - if (listen_pid) { - if (atoi(listen_pid) == (int) getpid()) { - char *listen_fds = getenv("LISTEN_FDS"); - if (listen_fds) { - int systemd_fds = atoi(listen_fds); - if (systemd_fds > 0) { - uwsgi_log("- SystemD socket activation detected -\n"); - for (i = 3; i < 3 + systemd_fds; i++) { - uwsgi_sock = uwsgi_new_socket(NULL); - uwsgi_add_socket_from_fd(uwsgi_sock, i); - } - uwsgi.skip_zero = 1; - } - unsetenv("LISTEN_PID"); - unsetenv("LISTEN_FDS"); - } - } - } - - char *upstart_events = getenv("UPSTART_EVENTS"); - if (upstart_events && !strcmp(upstart_events, "socket")) { - char *upstart_fds = getenv("UPSTART_FDS"); - if (upstart_fds) { - uwsgi_log("- Upstart socket bridge detected (job: %s) -\n", getenv("UPSTART_JOB")); - uwsgi_sock = uwsgi_new_socket(NULL); - uwsgi_add_socket_from_fd(uwsgi_sock, atoi(upstart_fds)); - uwsgi.skip_zero = 1; - } - unsetenv("UPSTART_EVENTS"); - unsetenv("UPSTART_FDS"); - } - - - struct uwsgi_string_list *zn = uwsgi.zerg_node; - while (zn) { - if (uwsgi_zerg_attach(zn->value)) { - if (!uwsgi.zerg_fallback) { - exit(1); - } - } - zn = zn->next; - } - - - - if (uwsgi.zerg) { -#ifdef UWSGI_DEBUG - uwsgi_log("attaching zerg sockets...\n"); -#endif - int zerg_fd; - i = 0; - for (;;) { - zerg_fd = uwsgi.zerg[i]; - if (zerg_fd == -1) { - break; - } - uwsgi_sock = uwsgi_new_socket(NULL); - uwsgi_add_socket_from_fd(uwsgi_sock, zerg_fd); - i++; - } - - uwsgi_log("zerg sockets attached\n"); - } + uwsgi_setup_systemd(); + uwsgi_setup_upstart(); + uwsgi_setup_zerg(); } //check for inherited sockets if (uwsgi.is_a_reload) { - - uwsgi_sock = uwsgi.sockets; - while (uwsgi_sock) { - //a bit overengineering - if (uwsgi_sock->name[0] != 0 && !uwsgi_sock->bound) { - for (j = 3; j < (int) uwsgi.max_fd; j++) { - uwsgi_add_socket_from_fd(uwsgi_sock, j); - } - } - uwsgi_sock = uwsgi_sock->next; - } - - //now close all the unbound fd - for (j = 3; j < (int) uwsgi.max_fd; j++) { - int useless = 1; -#ifdef UWSGI_MULTICAST - if (j == uwsgi.cluster_fd) - continue; -#endif - if (uwsgi.has_emperor) { - if (j == uwsgi.emperor_fd) - continue; - } - - if (uwsgi.shared->worker_log_pipe[0] > -1) { - if (j == uwsgi.shared->worker_log_pipe[0]) - continue; - } - - if (uwsgi.shared->worker_log_pipe[1] > -1) { - if (j == uwsgi.shared->worker_log_pipe[1]) - continue; - } - - if (uwsgi.original_log_fd > -1) { - if (j == uwsgi.original_log_fd) - continue; - } - - if (uwsgi.cache_server && uwsgi.cache_server_fd != -1) { - if (j == uwsgi.cache_server_fd) - continue; - } - - struct uwsgi_gateway_socket *ugs = uwsgi.gateway_sockets; - int found = 0; - while (ugs) { - if (ugs->fd == j) { - found = 1; - break; - } - ugs = ugs->next; - } - if (found) - continue; - - - int y; - found = 0; - for (y = 0; y < ushared->gateways_cnt; y++) { - if (ushared->gateways[y].internal_subscription_pipe[0] == j) { - found = 1; - break; - } - if (ushared->gateways[y].internal_subscription_pipe[1] == j) { - found = 1; - break; - } - } - - if (found) - continue; - - - socket_type_len = sizeof(struct sockaddr_un); - gsa.sa = (struct sockaddr *) &usa; - if (!getsockname(j, gsa.sa, &socket_type_len)) { - 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) { - uwsgi_sock = uwsgi.shared_sockets; - while (uwsgi_sock) { - if (uwsgi_sock->fd == j && uwsgi_sock->bound) { - useless = 0; - break; - } - uwsgi_sock = uwsgi_sock->next; - } - } - } - - if (useless) { - close(j); - } - } + uwsgi_setup_inherited_sockets(); } diff --git a/uwsgi.h b/uwsgi.h index bae92084..c102c1bd 100644 --- a/uwsgi.h +++ b/uwsgi.h @@ -2976,6 +2976,11 @@ struct uwsgi_string_list *uwsgi_string_list_has_item(struct uwsgi_string_list *, void trigger_harakiri(int); +void uwsgi_setup_systemd(); +void uwsgi_setup_upstart(); +void uwsgi_setup_zerg(); +void uwsgi_setup_inherited_sockets(); + #ifdef UWSGI_SSL void uwsgi_ssl_init(void); SSL_CTX *uwsgi_ssl_new_server_context(char *, char *, char *, char *, char *); diff --git a/uwsgiconfig.py b/uwsgiconfig.py index a8a4ac4a..d371ee5c 100644 --- a/uwsgiconfig.py +++ b/uwsgiconfig.py @@ -343,6 +343,7 @@ class uConf(object): self.config.read(filename) self.gcc_list = ['utils', 'protocol', 'socket', 'logging', 'master', 'master_utils', 'emperor', 'notify', 'mule', 'subscription', 'stats', + 'setup_utils', 'plugins', 'lock', 'cache', 'queue', 'event', 'signal', 'cluster', 'rpc', 'gateway', 'loop', 'lib/rbtree', 'lib/amqp', 'rb_timers', 'uwsgi'] # add protocols self.gcc_list.append('proto/base')