[Project] Another try to deal with final events

5 files changed, 57 insertions, 22 deletions

diff --git a/contrib/libev/ev.c b/contrib/libev/ev.c
index cb8127fc5..82d2fa8a9 100644
--- a/contrib/libev/ev.c
+++ b/contrib/libev/ev.c
@@ -3777,6 +3777,12 @@ ev_unref (EV_P) EV_NOEXCEPT
   --activecnt;
 }
 
+int
+ev_active_cnt (EV_P) EV_NOEXCEPT
+{
+	return activecnt;
+}
+
 void
 ev_now_update (EV_P) EV_NOEXCEPT
 {
diff --git a/contrib/libev/ev.h b/contrib/libev/ev.h
index cb7b2e479..b27a2fdad 100644
--- a/contrib/libev/ev.h
+++ b/contrib/libev/ev.h
@@ -630,6 +630,7 @@ EV_API_DECL void ev_break (EV_P_ int how EV_CPP (= EVBREAK_ONE)) EV_NOEXCEPT; /*
  */
 EV_API_DECL void ev_ref   (EV_P) EV_NOEXCEPT;
 EV_API_DECL void ev_unref (EV_P) EV_NOEXCEPT;
+EV_API_DECL int ev_active_cnt (EV_P) EV_NOEXCEPT;
 
 /*
  * convenience function, wait for a single event, without registering an event watcher
diff --git a/src/controller.c b/src/controller.c
index 374880952..f24269999 100644
--- a/src/controller.c
+++ b/src/controller.c
@@ -3564,7 +3564,7 @@ lua_csession_send_string (lua_State *L)
 	return 0;
 }
 
-static gboolean
+static void
 rspamd_controller_on_terminate (struct rspamd_worker *worker)
 {
 	struct rspamd_controller_worker_ctx *ctx = worker->ctx;
@@ -3576,8 +3576,6 @@ rspamd_controller_on_terminate (struct rspamd_worker *worker)
 		ev_timer_stop (ctx->event_loop, &ctx->rrd_event);
 		rspamd_rrd_close (ctx->rrd);
 	}
-
-	return FALSE;
 }
 
 static void
@@ -3732,8 +3730,6 @@ start_controller_worker (struct rspamd_worker *worker)
 				DEFAULT_STATS_PATH);
 	}
 
-	g_ptr_array_add (worker->finish_actions,
-			(gpointer)rspamd_controller_on_terminate);
 	rspamd_controller_load_saved_stats (ctx);
 	ctx->lang_det = ctx->cfg->lang_det;
 
@@ -3917,6 +3913,7 @@ start_controller_worker (struct rspamd_worker *worker)
 	/* Start event loop */
 	ev_loop (ctx->event_loop, 0);
 	rspamd_worker_block_signals ();
+	rspamd_controller_on_terminate (worker);
 
 	rspamd_stat_close ();
 	rspamd_http_router_free (ctx->http);
diff --git a/src/libserver/worker_util.c b/src/libserver/worker_util.c
index 078de3c8f..cc89c210b 100644
--- a/src/libserver/worker_util.c
+++ b/src/libserver/worker_util.c
@@ -84,21 +84,60 @@ rspamd_get_worker_by_type (struct rspamd_config *cfg, GQuark type)
 	return NULL;
 }
 
-static gboolean
+static void
+rspamd_worker_check_finished (EV_P_ ev_timer *w, int revents)
+{
+	int *pnchecks = (int *)w->data;
+
+	if (*pnchecks > SOFT_SHUTDOWN_TIME * 10) {
+		msg_warn ("terminating worker before finishing of terminate handlers");
+		ev_break (EV_A_ EVBREAK_ONE);
+	}
+	else {
+		int refcount = ev_active_cnt (EV_A);
+
+		if (refcount == 1) {
+			ev_break (EV_A_ EVBREAK_ONE);
+		}
+	}
+}
+
+static void
 rspamd_worker_terminate_handlers (struct rspamd_worker *w)
 {
 	guint i;
 	gboolean (*cb)(struct rspamd_worker *);
-	gboolean ret = FALSE;
+	struct rspamd_abstract_worker_ctx *actx;
+	struct ev_loop *final_gift, *orig_loop;
+	static ev_timer margin_call;
+	static int nchecks = 0;
+
+	actx = (struct rspamd_abstract_worker_ctx *)w->ctx;
+
+	/*
+	 * Here are dragons:
+	 * - we create a new loop
+	 * - we set a new ev_loop for worker via injection over rspamd_abstract_worker_ctx
+	 * - then we run finish actions
+	 * - then we create a special timer to kill worker if it fails to finish
+	 */
+	final_gift = ev_loop_new (EVBACKEND_ALL);
+	orig_loop = actx->event_loop;
+	actx->event_loop = final_gift;
+	margin_call.data = &nchecks;
+	ev_timer_init (&margin_call, rspamd_worker_check_finished, 0.1,
+			0.1);
+	ev_timer_start (final_gift, &margin_call);
 
 	for (i = 0; i < w->finish_actions->len; i ++) {
 		cb = g_ptr_array_index (w->finish_actions, i);
-		if (cb (w)) {
-			ret = TRUE;
-		}
+		cb (w);
 	}
 
-	return ret;
+	ev_run (final_gift, 0);
+	ev_loop_destroy (final_gift);
+	/* Restore original loop */
+	actx->event_loop = orig_loop;
 }
 
 static void
@@ -159,8 +198,6 @@ rspamd_worker_usr1_handler (struct rspamd_worker_signal_handler *sigh, void *arg
 static gboolean
 rspamd_worker_term_handler (struct rspamd_worker_signal_handler *sigh, void *arg)
 {
-	ev_tstamp delay;
-
 	if (!sigh->worker->wanna_die) {
 		static ev_timer shutdown_ev;
 
@@ -172,16 +209,10 @@ rspamd_worker_term_handler (struct rspamd_worker_signal_handler *sigh, void *arg
 				"terminating after receiving signal %s",
 				g_strsignal (sigh->signo));
 
-		if (rspamd_worker_terminate_handlers (sigh->worker)) {
-			delay = SOFT_SHUTDOWN_TIME;
-		}
-		else {
-			delay = 0;
-		}
-
+		rspamd_worker_terminate_handlers (sigh->worker);
 		sigh->worker->wanna_die = 1;
 		ev_timer_init (&shutdown_ev, rspamd_worker_on_delayed_shutdown,
-				SOFT_SHUTDOWN_TIME, 0.0);
+				0.0, 0.0);
 		ev_timer_start (sigh->event_loop, &shutdown_ev);
 		rspamd_worker_stop_accept (sigh->worker);
 	}
diff --git a/src/rspamd.c b/src/rspamd.c
index 00995c470..2c27b85cc 100644
--- a/src/rspamd.c
+++ b/src/rspamd.c
@@ -967,7 +967,7 @@ rspamd_term_handler (struct ev_loop *loop, ev_signal *w, int revents)
 		rspamd_log_nolock (rspamd_main->logger);
 		/* Stop srv events to avoid false notifications */
 		g_hash_table_foreach (rspamd_main->workers, stop_srv_ev, rspamd_main);
-		rspamd_pass_signal (rspamd_main->workers, w->signum);
+		rspamd_pass_signal (rspamd_main->workers, SIGTERM);
 
 		if (control_fd != -1) {
 			ev_io_stop (rspamd_main->event_loop, &control_ev);