Implement watching for async symbols.

2 files changed, 23 insertions, 2 deletions

diff --git a/src/libserver/symbols_cache.c b/src/libserver/symbols_cache.c
index 40a54ad06..b74f2c7d7 100644
--- a/src/libserver/symbols_cache.c
+++ b/src/libserver/symbols_cache.c
@@ -850,6 +850,19 @@ rspamd_symbols_cache_metric_limit (struct rspamd_task *task,
 	return FALSE;
 }
 
+static void
+rspamd_symbols_cache_watcher_cb (gpointer sessiond, gpointer ud)
+{
+	struct rspamd_task *task = sessiond;
+	struct cache_item *item = ud;
+	struct cache_savepoint *checkpoint;
+
+	checkpoint = task->checkpoint;
+
+	/* Specify that we are done with this item */
+	setbit (checkpoint->processed_bits, item->id * 2 + 1);
+}
+
 static gboolean
 rspamd_symbols_cache_check_symbol (struct rspamd_task *task,
 		struct symbols_cache *cache,
@@ -867,6 +880,9 @@ rspamd_symbols_cache_check_symbol (struct rspamd_task *task,
 		setbit (checkpoint->processed_bits, item->id * 2);
 		t1 = rspamd_get_ticks ();
 		pending_before = rspamd_session_events_pending (task->s);
+		/* Watch for events appeared */
+		rspamd_session_watch_start (task->s, rspamd_symbols_cache_watcher_cb,
+				item);
 
 		if (item->symbol != NULL &&
 				G_UNLIKELY (check_debug_symbol (task->cfg, item->symbol))) {
@@ -879,10 +895,10 @@ rspamd_symbols_cache_check_symbol (struct rspamd_task *task,
 		}
 
 		t2 = rspamd_get_ticks ();
-		pending_after = rspamd_session_events_pending (task->s);
-
 		diff = (t2 - t1) * 1000000;
 		rspamd_set_counter (item, diff);
+		rspamd_session_watch_stop (task->s);
+		pending_after = rspamd_session_events_pending (task->s);
 
 		if (pending_before == pending_after) {
 			/* No new events registered */
diff --git a/src/libserver/task.c b/src/libserver/task.c
index e063f0f5b..1638a46f3 100644
--- a/src/libserver/task.c
+++ b/src/libserver/task.c
@@ -135,6 +135,11 @@ rspamd_task_fin (void *arg)
 		return TRUE;
 	}
 
+	if (RSPAMD_TASK_IS_PROCESSED (task)) {
+		rspamd_task_reply (task);
+		return TRUE;
+	}
+
 	/* One more iteration */
 	return FALSE;
 }