Fix race in surbl module.

Add more debugging to destructors in pools.

src/client/rspamc.c

@@ -36,6 +36,7 @@ static gchar                   *statfile;
 static gchar                   *ip;
 static gint                     weight = 1;
 static gint                     flag;
+static gint                     timeout = 5;
 static gboolean                 pass_all;
 static gboolean                 tty = FALSE;
 static gboolean                 verbose = FALSE;
@@ -50,6 +51,7 @@ static GOptionEntry entries[] =
 		{ "pass", 'p', 0, G_OPTION_ARG_NONE, &pass_all, "Pass all filters", NULL },
 		{ "verbose", 'v', 0, G_OPTION_ARG_NONE, &verbose, "More verbose output", NULL },
 		{ "ip", 'i', 0, G_OPTION_ARG_STRING, &ip, "Emulate that message was received from specified ip address", NULL },
+		{ "timeout", 't', 0, G_OPTION_ARG_INT, &timeout, "Timeout for waiting for a reply", NULL },
 		{ NULL, 0, 0, G_OPTION_ARG_NONE, NULL, NULL, NULL }
 };
 
@@ -551,6 +553,7 @@ main (gint argc, gchar **argv, gchar **env)
 	rspamd_client_init ();
 
 	read_cmd_line (&argc, &argv);
+	rspamd_set_timeout (1000, timeout * 1000);
 	tty = isatty (STDOUT_FILENO);
 	/* Now read other args from argc and argv */
 	if (argc == 1) {

src/mem_pool.c

@@ -438,7 +438,8 @@ memory_pool_unlock_shared (memory_pool_t * pool, void *pointer)
 }
 
 void
-memory_pool_add_destructor (memory_pool_t * pool, pool_destruct_func func, void *data)
+memory_pool_add_destructor_full (memory_pool_t * pool, pool_destruct_func func, void *data,
+		const gchar *function, const gchar *line)
 {
 	struct _pool_destructors       *cur, *tmp;
 
@@ -449,6 +450,8 @@ memory_pool_add_destructor (memory_pool_t * pool, pool_destruct_func func, void
 		while (tmp) {
 			if (tmp->func == func && tmp->data == data) {
 				/* Do not add identical destructors, they must be unique */
+				msg_warn ("duplicate desctrutors detected: already have destructor from %s:%s and is trying to add from %s:%s",
+						tmp->function, tmp->loc, function, line);
 				return;
 			}
 			tmp = tmp->prev;
@@ -456,6 +459,8 @@ memory_pool_add_destructor (memory_pool_t * pool, pool_destruct_func func, void
 
 		cur->func = func;
 		cur->data = data;
+		cur->function = function;
+		cur->loc = line;
 		cur->prev = pool->destructors;
 		pool->destructors = cur;
 	}

src/mem_pool.h

@@ -62,6 +62,8 @@ struct _pool_chain_shared {
 struct _pool_destructors {
 	pool_destruct_func func;				/**< pointer to destructor					*/
 	void *data;								/**< data to free							*/
+	const gchar *function;					/**< function from which this destructor was added */
+	const gchar *loc;						/**< line number                            */
 	struct _pool_destructors *prev;			/**< chain link								*/
 };
 
@@ -165,7 +167,11 @@ void memory_pool_unlock_shared (memory_pool_t *pool, void *pointer);
  * @param func pointer to function-destructor
  * @param data pointer to data that would be passed to destructor
  */
-void memory_pool_add_destructor (memory_pool_t *pool, pool_destruct_func func, void *data);
+void memory_pool_add_destructor_full (memory_pool_t *pool, pool_destruct_func func, void *data,
+		const gchar *function, const gchar *line);
+
+/* Macros for common usage */
+#define memory_pool_add_destructor(pool, func, data) memory_pool_add_destructor_full(pool, func, data, G_STRFUNC, G_STRLOC)
 
 /**
  * Replace destructor callback to pool for specified pointer

src/plugins/surbl.c

@@ -825,8 +825,9 @@ redirector_callback (gint fd, short what, void *arg)
 			r = rspamd_snprintf (url_buf, sizeof (url_buf), "GET %s HTTP/1.0\r\n\r\n", struri (param->url));
 			if (write (param->sock, url_buf, r) == -1) {
 				msg_err ("write failed %s to %s", strerror (errno), param->redirector->name);
-				remove_normal_event (param->task->s, free_redirector_session, param);
 				upstream_fail (&param->redirector->up, param->task->tv.tv_sec);
+				remove_normal_event (param->task->s, free_redirector_session, param);
+
 				return;
 			}
 			param->state = STATE_READ;
@@ -834,8 +835,9 @@ redirector_callback (gint fd, short what, void *arg)
 		else {
 			msg_info ("<%s> connection to redirector %s timed out while waiting for write",
 					param->task->message_id, param->redirector->name);
-			remove_normal_event (param->task->s, free_redirector_session, param);
 			upstream_fail (&param->redirector->up, param->task->tv.tv_sec);
+			remove_normal_event (param->task->s, free_redirector_session, param);
+
 			return;
 		}
 		break;
@@ -844,8 +846,9 @@ redirector_callback (gint fd, short what, void *arg)
 			r = read (param->sock, url_buf, sizeof (url_buf) - 1);
 			if (r <= 0) {
 				msg_err ("read failed: %s from %s", strerror (errno), param->redirector->name);
-				remove_normal_event (param->task->s, free_redirector_session, param);
 				upstream_fail (&param->redirector->up, param->task->tv.tv_sec);
+				remove_normal_event (param->task->s, free_redirector_session, param);
+
 				return;
 			}
 
@@ -865,14 +868,14 @@ redirector_callback (gint fd, short what, void *arg)
 					parse_uri (param->url, memory_pool_strdup (param->task->task_pool, c), param->task->task_pool);
 				}
 			}
-			remove_normal_event (param->task->s, free_redirector_session, param);
 			upstream_ok (&param->redirector->up, param->task->tv.tv_sec);
+			remove_normal_event (param->task->s, free_redirector_session, param);
 		}
 		else {
 			msg_info ("<%s> reading redirector %s timed out, while waiting for read",
 					param->redirector->name, param->task->message_id);
-			remove_normal_event (param->task->s, free_redirector_session, param);
 			upstream_fail (&param->redirector->up, param->task->tv.tv_sec);
+			remove_normal_event (param->task->s, free_redirector_session, param);
 		}
 		break;
 	}