* More things to be thread-safe:

 - pool allocator is now thread-safe
 - lua subsystem now holds lock to avoid lua stack corruption
 - events subsystem now using conditional variables to wait for async_threads
 - insert_result is thread-safe now

5 files changed, 39 insertions, 0 deletions

diff --git a/src/lua/lua_classifier.c b/src/lua/lua_classifier.c
index 0bf0daca8..edaf4e7a6 100644
--- a/src/lua/lua_classifier.c
+++ b/src/lua/lua_classifier.c
@@ -123,6 +123,7 @@ call_classifier_pre_callbacks (struct classifier_config *ccf, struct worker_task
 	struct classifier_callback_data *cd;
 	lua_State                       *L;
 
+
 	/* Go throught all callbacks and call them, appending results to list */
 	cur = g_list_first (ccf->pre_callbacks);
 	while (cur) {
@@ -134,6 +135,7 @@ call_classifier_pre_callbacks (struct classifier_config *ccf, struct worker_task
 		cur = g_list_next (cur);
 	}
 	
+	g_mutex_lock (lua_mtx);
 	if (res == NULL) {
 		L = task->cfg->lua_state;
 		/* Check function from global table 'classifiers' */
@@ -149,6 +151,8 @@ call_classifier_pre_callbacks (struct classifier_config *ccf, struct worker_task
 		}
 		lua_pop (L, 1);
 	}
+	g_mutex_unlock (lua_mtx);
+
 	return res;
 }
 
@@ -162,6 +166,7 @@ call_classifier_post_callbacks (struct classifier_config *ccf, struct worker_tas
 	double                          out = in;
 	GList                          *cur;
 
+	g_mutex_lock (lua_mtx);
 	/* Go throught all callbacks and call them, appending results to list */
 	cur = g_list_first (ccf->pre_callbacks);
 	while (cur) {
@@ -190,6 +195,7 @@ call_classifier_post_callbacks (struct classifier_config *ccf, struct worker_tas
 
 		cur = g_list_next (cur);
 	}
+	g_mutex_unlock (lua_mtx);
 
 	return out;
 
diff --git a/src/lua/lua_common.c b/src/lua/lua_common.c
index 00fcadc93..4d90048ca 100644
--- a/src/lua/lua_common.c
+++ b/src/lua/lua_common.c
@@ -28,6 +28,9 @@
 /* Lua module init function */
 #define MODULE_INIT_FUNC "module_init"
 
+/* Global lua mutex */
+GMutex *lua_mtx = NULL;
+
 const luaL_reg                  null_reg[] = {
 	{"__tostring", lua_class_tostring},
 	{NULL, NULL}
@@ -224,6 +227,13 @@ init_lua (struct config_file *cfg)
 	L = lua_open ();
 	luaL_openlibs (L);
 
+#if ((GLIB_MAJOR_VERSION == 2) && (GLIB_MINOR_VERSION <= 30))
+	lua_mtx = g_mutex_new ();
+#else
+	lua_mtx = g_malloc (sizeof (GMutex));
+	g_mutex_init (lua_mtx);
+#endif
+
 	(void)luaopen_rspamd (L);
 	(void)luaopen_logger (L);
 	(void)luaopen_config (L);
@@ -320,6 +330,7 @@ lua_call_filter (const gchar *function, struct worker_task *task)
 	struct worker_task            **ptask;
 	lua_State                      *L = task->cfg->lua_state;
 
+	g_mutex_lock (lua_mtx);
 	lua_getglobal (L, function);
 	ptask = lua_newuserdata (L, sizeof (struct worker_task *));
 	lua_setclass (L, "rspamd{task}", -1);
@@ -335,6 +346,8 @@ lua_call_filter (const gchar *function, struct worker_task *task)
 	}
 	result = lua_tonumber (L, -1);
 	lua_pop (L, 1);				/* pop returned value */
+	g_mutex_unlock (lua_mtx);
+
 	return result;
 }
 
@@ -345,6 +358,7 @@ lua_call_chain_filter (const gchar *function, struct worker_task *task, gint *ma
 	guint                           i;
 	lua_State                      *L = task->cfg->lua_state;
 
+	g_mutex_lock (lua_mtx);
 	lua_getglobal (L, function);
 
 	for (i = 0; i < number; i++) {
@@ -360,6 +374,8 @@ lua_call_chain_filter (const gchar *function, struct worker_task *task, gint *ma
 	}
 	result = lua_tonumber (L, -1);
 	lua_pop (L, 1);				/* pop returned value */
+	g_mutex_unlock (lua_mtx);
+
 	return result;
 }
 
@@ -374,6 +390,7 @@ lua_call_expression_func (const gchar *module, const gchar *function,
 	struct expression_argument     *arg;
 	int                             nargs = 1, pop = 0;
 
+	g_mutex_lock (lua_mtx);
 	/* Call specified function and expect result of given expected_type */
 	/* First check function in config table */
 	lua_getglobal (L, "config");
@@ -435,6 +452,7 @@ lua_call_expression_func (const gchar *module, const gchar *function,
 
 	if (lua_pcall (L, nargs, 1, 0) != 0) {
 		msg_info ("call to %s failed: %s", function, lua_tostring (L, -1));
+		g_mutex_unlock (lua_mtx);
 		return FALSE;
 	}
 	pop ++;
@@ -442,11 +460,13 @@ lua_call_expression_func (const gchar *module, const gchar *function,
 	if (!lua_isboolean (L, -1)) {
 		lua_pop (L, pop);
 		msg_info ("function %s must return a boolean", function);
+		g_mutex_unlock (lua_mtx);
 		return FALSE;
 	}
 	*res = lua_toboolean (L, -1);
 	lua_pop (L, pop);
 
+	g_mutex_unlock (lua_mtx);
 	return TRUE;
 }
 
@@ -468,6 +488,7 @@ lua_consolidation_callback (gpointer key, gpointer value, gpointer arg)
 	struct consolidation_callback_data *data = (struct consolidation_callback_data *)arg;
 	lua_State                      *L = data->task->cfg->lua_state;
 
+	g_mutex_lock (lua_mtx);
 	lua_getglobal (L, data->func);
 
 	lua_pushstring (L, (const gchar *)key);
@@ -483,6 +504,7 @@ lua_consolidation_callback (gpointer key, gpointer value, gpointer arg)
 	res = lua_tonumber (L, -1);
 	lua_pop (L, 1);				/* pop returned value */
 	data->score += res;
+	g_mutex_unlock (lua_mtx);
 }
 
 double
@@ -519,6 +541,7 @@ lua_normalizer_func (struct config_file *cfg, long double score, void *params)
         return score;
     }
 
+    g_mutex_lock (lua_mtx);
     lua_getglobal (L, p->data);
     lua_pushnumber (L, score);
 
@@ -533,6 +556,7 @@ lua_normalizer_func (struct config_file *cfg, long double score, void *params)
 	res = lua_tonumber (L, -1);
 	lua_pop (L, 1);
 
+	g_mutex_unlock (lua_mtx);
     return res;
 }
 
diff --git a/src/lua/lua_common.h b/src/lua/lua_common.h
index c1891a6a7..32604cbc4 100644
--- a/src/lua/lua_common.h
+++ b/src/lua/lua_common.h
@@ -15,6 +15,7 @@
 #define LUA_INTERFACE_DEF(class, name) { #name, lua_##class##_##name }
 
 extern const luaL_reg null_reg[];
+extern GMutex *lua_mtx;
 
 #define RSPAMD_LUA_API_VERSION 9
 
diff --git a/src/lua/lua_config.c b/src/lua/lua_config.c
index cd1287a18..bdb4ce056 100644
--- a/src/lua/lua_config.c
+++ b/src/lua/lua_config.c
@@ -318,6 +318,7 @@ lua_config_function_callback (struct worker_task *task, GList *args, void *user_
 	GList                          *cur;
 	gboolean                        res = FALSE;
 
+	g_mutex_lock (lua_mtx);
 	if (cd->cb_is_ref) {
 		lua_rawgeti (cd->L, LUA_REGISTRYINDEX, cd->callback.ref);
 	}
@@ -347,6 +348,7 @@ lua_config_function_callback (struct worker_task *task, GList *args, void *user_
 		}
 		lua_pop (cd->L, 1);
 	}
+	g_mutex_unlock (lua_mtx);
 
 	return res;
 }
@@ -439,6 +441,7 @@ lua_call_post_filters (struct worker_task *task)
 	struct worker_task            **ptask;
 	GList                          *cur;
 
+	g_mutex_lock (lua_mtx);
 	cur = task->cfg->post_filters;
 	while (cur) {
 		cd = cur->data;
@@ -458,6 +461,7 @@ lua_call_post_filters (struct worker_task *task)
 		}
 		cur = g_list_next (cur);
 	}
+	g_mutex_unlock (lua_mtx);
 }
 
 static gint
@@ -583,6 +587,7 @@ lua_metric_symbol_callback (struct worker_task *task, gpointer ud)
 	struct lua_callback_data       *cd = ud;
 	struct worker_task            **ptask;
 
+	g_mutex_lock (lua_mtx);
 	if (cd->cb_is_ref) {
 		lua_rawgeti (cd->L, LUA_REGISTRYINDEX, cd->callback.ref);
 	}
@@ -597,6 +602,7 @@ lua_metric_symbol_callback (struct worker_task *task, gpointer ud)
 		msg_info ("call to %s failed: %s", cd->cb_is_ref ? "local function" :
 									cd->callback.name, lua_tostring (cd->L, -1));
 	}
+	g_mutex_unlock (lua_mtx);
 }
 
 static gint
diff --git a/src/lua/lua_task.c b/src/lua/lua_task.c
index c0b299fb7..d28d897f6 100644
--- a/src/lua/lua_task.c
+++ b/src/lua/lua_task.c
@@ -468,6 +468,7 @@ lua_dns_callback (struct rspamd_dns_reply *reply, gpointer arg)
 	union rspamd_reply_element     *elt;
 	GList                          *cur;
 
+	g_mutex_lock (lua_mtx);
 	if (cd->cb_is_ref) {
 		lua_rawgeti (cd->L, LUA_REGISTRYINDEX, cd->callback.ref);
 	}
@@ -553,6 +554,7 @@ lua_dns_callback (struct rspamd_dns_reply *reply, gpointer arg)
 	if (cd->cb_is_ref) {
 		luaL_unref (cd->L, LUA_REGISTRYINDEX, cd->callback.ref);
 	}
+	g_mutex_unlock (lua_mtx);
 }
 
 static gint