[Fix] Fix crashes on task cleanup

src/libserver/dns.c

@@ -122,7 +122,7 @@ make_dns_request (struct rspamd_dns_resolver *resolver,
 		return FALSE;
 	}
 
-	if (session && rspamd_session_is_destroying (session)) {
+	if (session && rspamd_session_blocked (session)) {
 		return FALSE;
 	}
 

src/libserver/events.c

@@ -21,9 +21,10 @@
 
 #define RSPAMD_SESSION_FLAG_WATCHING (1 << 0)
 #define RSPAMD_SESSION_FLAG_DESTROYING (1 << 1)
+#define RSPAMD_SESSION_FLAG_CLEANUP (1 << 2)
 
 #define RSPAMD_SESSION_IS_WATCHING(s) ((s)->flags & RSPAMD_SESSION_FLAG_WATCHING)
-#define RSPAMD_SESSION_IS_DESTROYING(s) ((s)->flags & RSPAMD_SESSION_FLAG_DESTROYING)
+#define RSPAMD_SESSION_CAN_ADD_EVENT(s) (!((s)->flags & (RSPAMD_SESSION_FLAG_DESTROYING|RSPAMD_SESSION_FLAG_CLEANUP)))
 
 #define msg_err_session(...) rspamd_default_log_function(G_LOG_LEVEL_CRITICAL, \
         "events", session->pool->tag.uid, \
@@ -172,8 +173,9 @@ rspamd_session_add_event (struct rspamd_async_session *session,
 		g_assert_not_reached ();
 	}
 
-	if (RSPAMD_SESSION_IS_DESTROYING (session)) {
-		msg_debug_session ("skip adding event subsystem: %s: session is destroying",
+	if (!RSPAMD_SESSION_CAN_ADD_EVENT (session)) {
+		msg_debug_session ("skip adding event subsystem: %s: "
+					 "session is destroying/cleaning",
 				g_quark_to_string (subsystem));
 
 		return NULL;
@@ -293,7 +295,7 @@ rspamd_session_destroy (struct rspamd_async_session *session)
 		return FALSE;
 	}
 
-	if (!(session->flags & RSPAMD_SESSION_FLAG_DESTROYING)) {
+	if (!rspamd_session_blocked (session)) {
 		session->flags |= RSPAMD_SESSION_FLAG_DESTROYING;
 		rspamd_session_cleanup (session);
 
@@ -315,6 +317,8 @@ rspamd_session_cleanup (struct rspamd_async_session *session)
 		return;
 	}
 
+	session->flags |= RSPAMD_SESSION_FLAG_CLEANUP;
+
 	kh_foreach_key (session->events, ev, {
 		/* Call event's finalizer */
 		msg_debug_session ("removed event on destroy: %p, subsystem: %s",
@@ -327,6 +331,8 @@ rspamd_session_cleanup (struct rspamd_async_session *session)
 	});
 
 	kh_clear (rspamd_events_hash, session->events);
+
+	session->flags &= ~RSPAMD_SESSION_FLAG_CLEANUP;
 }
 
 gboolean
@@ -509,9 +515,9 @@ rspamd_session_mempool (struct rspamd_async_session *session)
 }
 
 gboolean
-rspamd_session_is_destroying (struct rspamd_async_session *session)
+rspamd_session_blocked (struct rspamd_async_session *session)
 {
 	g_assert (session != NULL);
 
-	return RSPAMD_SESSION_IS_DESTROYING (session);
+	return !RSPAMD_SESSION_CAN_ADD_EVENT (session);
 }

src/libserver/events.h

@@ -153,6 +153,6 @@ struct rspamd_async_watcher* rspamd_session_get_watcher (
  * @param s
  * @return
  */
-gboolean rspamd_session_is_destroying (struct rspamd_async_session *s);
+gboolean rspamd_session_blocked (struct rspamd_async_session *s);
 
 #endif /* RSPAMD_EVENTS_H */

src/libstat/backends/redis_backend.c

@@ -1572,7 +1572,7 @@ rspamd_redis_process_tokens (struct rspamd_task *task,
 	gint ret;
 	const gchar *learned_key = "learns";
 
-	if (rspamd_session_is_destroying (task->s)) {
+	if (rspamd_session_blocked (task->s)) {
 		return FALSE;
 	}
 
@@ -1667,7 +1667,7 @@ rspamd_redis_learn_tokens (struct rspamd_task *task, GPtrArray *tokens,
 	goffset off;
 	const gchar *learned_key = "learns";
 
-	if (rspamd_session_is_destroying (task->s)) {
+	if (rspamd_session_blocked (task->s)) {
 		return FALSE;
 	}
 

src/libstat/learn_cache/redis_cache.c

@@ -438,7 +438,7 @@ rspamd_stat_cache_redis_check (struct rspamd_task *task,
 	struct timeval tv;
 	gchar *h;
 
-	if (rspamd_session_is_destroying (task->s)) {
+	if (rspamd_session_blocked (task->s)) {
 		return RSPAMD_LEARN_INGORE;
 	}
 
@@ -473,7 +473,7 @@ rspamd_stat_cache_redis_learn (struct rspamd_task *task,
 	gchar *h;
 	gint flag;
 
-	if (rspamd_session_is_destroying (task->s)) {
+	if (rspamd_session_blocked (task->s)) {
 		return RSPAMD_LEARN_INGORE;
 	}
 

src/lua/lua_http.c

@@ -809,7 +809,7 @@ lua_http_request (lua_State *L)
 		return 1;
 	}
 
-	if (session && rspamd_session_is_destroying (session)) {
+	if (session && rspamd_session_blocked (session)) {
 		lua_pushboolean (L, FALSE);
 
 		return 1;

src/lua/lua_redis.c

@@ -686,7 +686,7 @@ rspamd_lua_redis_prepare_connection (lua_State *L, gint *pcbref)
 
 		lua_pop (L, 1); /* table */
 
-		if (session && rspamd_session_is_destroying (session)) {
+		if (session && rspamd_session_blocked (session)) {
 			ret = FALSE;
 		}
 
@@ -1215,7 +1215,7 @@ lua_redis_add_cmd (lua_State *L)
 
 			LL_PREPEND (sp_ud->c->specific, sp_ud);
 
-			if (ud->s && rspamd_session_is_destroying (ud->s)) {
+			if (ud->s && rspamd_session_blocked (ud->s)) {
 				lua_pushboolean (L, 0);
 				lua_pushstring (L, "session is terminating");
 

src/lua/lua_tcp.c

@@ -1573,7 +1573,7 @@ lua_tcp_request (lua_State *L)
 	if (session) {
 		cbd->session = session;
 
-		if (rspamd_session_is_destroying (session)) {
+		if (rspamd_session_blocked (session)) {
 			TCP_RELEASE (cbd);
 			lua_pushboolean (L, FALSE);
 
@@ -1737,7 +1737,7 @@ lua_tcp_connect_sync (lua_State *L)
 	if (session) {
 		cbd->session = session;
 
-		if (rspamd_session_is_destroying (session)) {
+		if (rspamd_session_blocked (session)) {
 			TCP_RELEASE (cbd);
 			lua_pushboolean (L, FALSE);
 			lua_pushliteral (L, "Session is being destroyed, requests are not allowed");

src/plugins/fuzzy_check.c

@@ -2830,7 +2830,7 @@ register_fuzzy_client_call (struct rspamd_task *task,
 	rspamd_inet_addr_t *addr;
 	gint sock;
 
-	if (!rspamd_session_is_destroying (task->s)) {
+	if (!rspamd_session_blocked (task->s)) {
 		/* Get upstream */
 		selected = rspamd_upstream_get (rule->servers, RSPAMD_UPSTREAM_ROUND_ROBIN,
 				NULL, 0);
@@ -3311,7 +3311,7 @@ fuzzy_check_send_lua_learn (struct fuzzy_rule *rule,
 	gint ret = -1;
 
 	/* Get upstream */
-	if (!rspamd_session_is_destroying (task->s)) {
+	if (!rspamd_session_blocked (task->s)) {
 		while ((selected = rspamd_upstream_get (rule->servers,
 				RSPAMD_UPSTREAM_SEQUENTIAL, NULL, 0))) {
 			/* Create UDP socket */

src/plugins/surbl.c

@@ -1638,7 +1638,7 @@ register_redirector_call (struct rspamd_url *url, struct rspamd_task *task,
 	struct rspamd_http_message *msg;
 	struct surbl_ctx *surbl_module_ctx = surbl_get_context (task->cfg);
 
-	if (!rspamd_session_is_destroying (task->s)) {
+	if (!rspamd_session_blocked (task->s)) {
 
 		selected = rspamd_upstream_get (surbl_module_ctx->redirectors,
 				RSPAMD_UPSTREAM_ROUND_ROBIN, url->host, url->hostlen);