diff --git a/master.c b/master.c index 95ce8d79..f6f03539 100644 --- a/master.c +++ b/master.c @@ -2,6 +2,33 @@ extern struct uwsgi_server uwsgi; +void expire_rb_timeouts(struct rb_root *root) { + + time_t current = time(NULL); + struct uwsgi_rb_timer *urbt; + struct uwsgi_signal_rb_timer *usrbt; + + for(;;) { + + urbt = uwsgi_min_rb_timer(root); + + if (urbt == NULL) return; + + if (urbt->key <= current) { + // remove the timeout and add another + usrbt = (struct uwsgi_signal_rb_timer *) urbt->data; + rb_erase(&usrbt->uwsgi_rb_timer->rbt, root); + free(usrbt->uwsgi_rb_timer); + uwsgi_route_signal(usrbt->sig); + usrbt->uwsgi_rb_timer = uwsgi_add_rb_timer(root, time(NULL) + usrbt->value, usrbt); + continue; + } + + break; + } +} + + void uwsgi_subscribe(char *subscription) { char *ssb; @@ -140,6 +167,9 @@ void master_loop(char **argv, char **environ) { int check_interval = 1; + struct uwsgi_rb_timer *min_timeout; + struct rb_root *rb_timers = uwsgi_init_rb_timer(); + // release the GIL //UWSGI_RELEASE_GIL @@ -402,7 +432,7 @@ void master_loop(char **argv, char **environ) { // add unregistered timers - // locking is not needed as monitors can only increase + // locking is not needed as timers can only increase for(i=0;itimers_cnt;i++) { if (!ushared->timers[i].registered) { ushared->timers[i].fd = event_queue_add_timer(uwsgi.master_queue, &ushared->timers[i].id, ushared->timers[i].value); @@ -410,9 +440,38 @@ void master_loop(char **argv, char **environ) { } } + // add unregistered rb_timers + // locking is not needed as rb_timers can only increase + for(i=0;irb_timers_cnt;i++) { + if (!ushared->rb_timers[i].registered) { + ushared->rb_timers[i].uwsgi_rb_timer = uwsgi_add_rb_timer(rb_timers, time(NULL) + ushared->rb_timers[i].value, &ushared->rb_timers[i]); + ushared->rb_timers[i].registered = 1; + } + } + int interesting_fd = -1; + + if (ushared->rb_timers_cnt>0) { + min_timeout = uwsgi_min_rb_timer(rb_timers); + if (min_timeout == NULL ) { + check_interval = uwsgi.shared->options[UWSGI_OPTION_MASTER_INTERVAL]; + } + else { + check_interval = min_timeout->key - time(NULL); + if (check_interval <= 0) { + expire_rb_timeouts(rb_timers); + check_interval = 0; + } + } + } rlen = event_queue_wait(uwsgi.master_queue, check_interval, &interesting_fd); + if (rlen == 0) { + if (ushared->rb_timers_cnt>0) { + expire_rb_timeouts(rb_timers); + } + } + if (rlen > 0) { if (uwsgi.log_master) { diff --git a/plugins/python/uwsgi_pymodule.c b/plugins/python/uwsgi_pymodule.c index c17c0f61..b9880cfa 100644 --- a/plugins/python/uwsgi_pymodule.c +++ b/plugins/python/uwsgi_pymodule.c @@ -197,6 +197,23 @@ PyObject *py_uwsgi_add_timer(PyObject * self, PyObject * args) { return Py_None; } +PyObject *py_uwsgi_add_rb_timer(PyObject * self, PyObject * args) { + + uint8_t uwsgi_signal; + int secs; + + if (!PyArg_ParseTuple(args, "Bi:add_rb_timer", &uwsgi_signal, &secs)) { + return NULL; + } + + if (uwsgi_signal_add_rb_timer(uwsgi_signal, secs)) + return PyErr_Format(PyExc_ValueError, "unable to add rb_timer"); + + Py_INCREF(Py_None); + return Py_None; +} + + PyObject *py_uwsgi_add_file_monitor(PyObject * self, PyObject * args) { @@ -2323,7 +2340,7 @@ static PyMethodDef uwsgi_advanced_methods[] = { {"signal_received", py_uwsgi_signal_received, METH_VARARGS, ""}, {"add_file_monitor", py_uwsgi_add_file_monitor, METH_VARARGS, ""}, {"add_timer", py_uwsgi_add_timer, METH_VARARGS, ""}, - //{"add_rb_timer", py_uwsgi_add_rb_timer, METH_VARARGS, ""}, + {"add_rb_timer", py_uwsgi_add_rb_timer, METH_VARARGS, ""}, {"register_rpc", py_uwsgi_register_rpc, METH_VARARGS, ""}, {"rpc", py_uwsgi_rpc, METH_VARARGS, ""}, diff --git a/signal.c b/signal.c index 1b865044..566e90df 100644 --- a/signal.c +++ b/signal.c @@ -91,6 +91,31 @@ int uwsgi_add_timer(uint8_t sig, int secs) { } +int uwsgi_signal_add_rb_timer(uint8_t sig, int secs) { + + uwsgi_lock(uwsgi.rb_timer_table_lock); + + if (ushared->rb_timers_cnt < 64) { + + // fill the timer table, the master will use it to add items to the event queue + ushared->rb_timers[ushared->rb_timers_cnt].value = secs; + ushared->rb_timers[ushared->rb_timers_cnt].registered = 0; + ushared->rb_timers[ushared->rb_timers_cnt].sig = sig; + ushared->rb_timers_cnt++; + } + else { + uwsgi_log("you can register max 64 rb_timers !!!\n"); + uwsgi_unlock(uwsgi.rb_timer_table_lock); + return -1; + } + + uwsgi_unlock(uwsgi.rb_timer_table_lock); + + return 0; + +} + + void uwsgi_route_signal(uint8_t sig) { diff --git a/uwsgi.c b/uwsgi.c index 40585673..73d27d78 100644 --- a/uwsgi.c +++ b/uwsgi.c @@ -1062,6 +1062,10 @@ int uwsgi_start(void *v_argv) { uwsgi.timer_table_lock = uwsgi_mmap_shared_lock(); uwsgi_lock_init(uwsgi.timer_table_lock); + // rb_timer table lock + uwsgi.rb_timer_table_lock = uwsgi_mmap_shared_lock(); + uwsgi_lock_init(uwsgi.rb_timer_table_lock); + // daemons table lock uwsgi.daemon_table_lock = uwsgi_mmap_shared_lock(); uwsgi_lock_init(uwsgi.daemon_table_lock); diff --git a/uwsgi.h b/uwsgi.h index 17aea92b..427fdac9 100644 --- a/uwsgi.h +++ b/uwsgi.h @@ -672,7 +672,6 @@ struct uwsgi_fmon { }; struct uwsgi_timer { - char svalue[0xff]; int value; int fd; int id; @@ -680,6 +679,13 @@ struct uwsgi_timer { uint8_t sig; }; +struct uwsgi_signal_rb_timer { + int value; + int registered; + uint8_t sig; + struct uwsgi_rb_timer *uwsgi_rb_timer; +}; + struct uwsgi_server { @@ -954,6 +960,7 @@ struct uwsgi_server { void *signal_table_lock; void *fmon_table_lock; void *timer_table_lock; + void *rb_timer_table_lock; void *rpc_table_lock; void *spooler_lock; @@ -1081,6 +1088,9 @@ struct uwsgi_shared { struct uwsgi_timer timers[64]; int timers_cnt; + struct uwsgi_signal_rb_timer rb_timers[64]; + int rb_timers_cnt; + struct uwsgi_rpc rpc_table[MAX_RPC]; int rpc_count; @@ -1411,6 +1421,7 @@ void *uwsgi_mmap_shared_rwlock(void); int uwsgi_register_signal(uint8_t, char *, void *, uint8_t); int uwsgi_add_file_monitor(uint8_t, char *); int uwsgi_add_timer(uint8_t, int); +int uwsgi_signal_add_rb_timer(uint8_t, int); int uwsgi_signal_handler(uint8_t); void uwsgi_route_signal(uint8_t);