[Project] Remove unneeded phases, add safe guards, fix issues

diff --git a/src/libserver/symbols_cache.c b/src/libserver/symbols_cache.c
index 5d30f46dcb0da7411a1028cf0f828273c63a6159..c4b32d0c71e6e48adc049ab0607ef938cb6e79e0 100644 (file)
--- a/src/libserver/symbols_cache.c
+++ b/src/libserver/symbols_cache.c
@@ -155,11 +155,8 @@ struct delayed_cache_condition {
 enum rspamd_cache_savepoint_stage {
        RSPAMD_CACHE_PASS_INIT = 0,
        RSPAMD_CACHE_PASS_PREFILTERS,
-       RSPAMD_CACHE_PASS_WAIT_PREFILTERS,
        RSPAMD_CACHE_PASS_FILTERS,
-       RSPAMD_CACHE_PASS_WAIT_FILTERS,
        RSPAMD_CACHE_PASS_POSTFILTERS,
-       RSPAMD_CACHE_PASS_WAIT_POSTFILTERS,
        RSPAMD_CACHE_PASS_IDEMPOTENT,
        RSPAMD_CACHE_PASS_WAIT_IDEMPOTENT,
        RSPAMD_CACHE_PASS_DONE,
@@ -1307,6 +1304,7 @@ rspamd_symbols_cache_check_symbol (struct rspamd_task *task,
        if (item->func) {
 
                g_assert (item->func != NULL);
+               g_assert (!isset (checkpoint->processed_bits, item->id * 2));
                /* Check has been started */
                setbit (checkpoint->processed_bits, item->id * 2);
 
@@ -1351,8 +1349,10 @@ rspamd_symbols_cache_check_symbol (struct rspamd_task *task,
                        checkpoint->items_inflight ++;
                        /* Callback now must finalize itself */
                        item->func (task, item, item->user_data);
+                       checkpoint->cur_item = NULL;
 
                        if (checkpoint->items_inflight == 0) {
+
                                return TRUE;
                        }
 
@@ -1641,6 +1641,7 @@ rspamd_symbols_cache_process_symbols (struct rspamd_task * task,
        case RSPAMD_CACHE_PASS_PREFILTERS:
                /* Check for prefilters */
                saved_priority = G_MININT;
+               all_done = TRUE;
 
                for (i = 0; i < (gint)cache->prefilters->len; i ++) {
                        item = g_ptr_array_index (cache->prefilters, i);
@@ -1669,21 +1670,7 @@ rspamd_symbols_cache_process_symbols (struct rspamd_task * task,
 
                                rspamd_symbols_cache_check_symbol (task, cache, item,
                                                checkpoint);
-                       }
-               }
-
-               checkpoint->pass = RSPAMD_CACHE_PASS_WAIT_PREFILTERS;
-               break;
-
-       case RSPAMD_CACHE_PASS_WAIT_PREFILTERS:
-               all_done = TRUE;
-
-               for (i = 0; i < (gint)cache->prefilters->len; i ++) {
-                       item = g_ptr_array_index (cache->prefilters, i);
-
-                       if (!isset (checkpoint->processed_bits, item->id * 2 + 1)) {
                                all_done = FALSE;
-                               break;
                        }
                }
 
@@ -1694,13 +1681,17 @@ rspamd_symbols_cache_process_symbols (struct rspamd_task * task,
                if (stage == RSPAMD_TASK_STAGE_FILTERS) {
                        return rspamd_symbols_cache_process_symbols (task, cache, stage);
                }
+
                break;
+
        case RSPAMD_CACHE_PASS_FILTERS:
                /*
                 * On the first pass we check symbols that do not have dependencies
                 * If we figure out symbol that has no dependencies satisfied, then
                 * we just save it for another pass
                 */
+               all_done = TRUE;
+
                for (i = 0; i < (gint)checkpoint->version; i ++) {
                        if (RSPAMD_TASK_IS_SKIPPED (task)) {
                                return TRUE;
@@ -1724,11 +1715,10 @@ rspamd_symbols_cache_process_symbols (struct rspamd_task * task,
                        }
 
                        if (!isset (checkpoint->processed_bits, item->id * 2)) {
+                               all_done = FALSE;
+
                                if (!rspamd_symbols_cache_check_deps (task, cache, item,
                                                checkpoint, 0, FALSE)) {
-                                       gboolean found = FALSE;
-                                       guint j;
-                                       struct rspamd_symcache_item *tmp_it;
 
                                        msg_debug_cache_task ("blocked execution of %d(%s) unless deps are "
                                                        "resolved",
@@ -1742,12 +1732,7 @@ rspamd_symbols_cache_process_symbols (struct rspamd_task * task,
                        }
                }
 
-               checkpoint->pass = RSPAMD_CACHE_PASS_WAIT_FILTERS;
-               break;
-
-       case RSPAMD_CACHE_PASS_WAIT_FILTERS:
-               if (checkpoint->items_inflight == 0 ||
-                               stage == RSPAMD_TASK_STAGE_POST_FILTERS) {
+               if (all_done || stage == RSPAMD_TASK_STAGE_POST_FILTERS) {
                        checkpoint->pass = RSPAMD_CACHE_PASS_POSTFILTERS;
                }
 
@@ -1760,6 +1745,7 @@ rspamd_symbols_cache_process_symbols (struct rspamd_task * task,
        case RSPAMD_CACHE_PASS_POSTFILTERS:
                /* Check for postfilters */
                saved_priority = G_MININT;
+               all_done = TRUE;
 
                for (i = 0; i < (gint)cache->postfilters->len; i ++) {
                        if (RSPAMD_TASK_IS_SKIPPED (task)) {
@@ -1771,6 +1757,8 @@ rspamd_symbols_cache_process_symbols (struct rspamd_task * task,
                        if (!isset (checkpoint->processed_bits, item->id * 2) &&
                                        !isset (checkpoint->processed_bits, item->id * 2 + 1)) {
                                /* Check priorities */
+                               all_done = FALSE;
+
                                if (saved_priority == G_MININT) {
                                        saved_priority = item->priority;
                                }
@@ -1790,20 +1778,6 @@ rspamd_symbols_cache_process_symbols (struct rspamd_task * task,
                                                checkpoint);
                        }
                }
-               checkpoint->pass = RSPAMD_CACHE_PASS_WAIT_POSTFILTERS;
-               break;
-
-       case RSPAMD_CACHE_PASS_WAIT_POSTFILTERS:
-               all_done = TRUE;
-
-               for (i = 0; i < (gint)cache->postfilters->len; i ++) {
-                       item = g_ptr_array_index (cache->postfilters, i);
-
-                       if (!isset (checkpoint->processed_bits, item->id * 2 + 1)) {
-                               all_done = FALSE;
-                               break;
-                       }
-               }
 
                if (all_done) {
                        checkpoint->pass = RSPAMD_CACHE_PASS_IDEMPOTENT;
@@ -2561,7 +2535,7 @@ rspamd_symbols_cache_finalize_item (struct rspamd_task *task,
                                                                        struct rspamd_symcache_item *item)
 {
        struct cache_savepoint *checkpoint = task->checkpoint;
-       struct rspamd_symcache_item *rdep;
+       struct cache_dependency *rdep;
        gdouble total_ticks = 0, t2, diff;
        guint i;
        struct timeval tv;
@@ -2572,7 +2546,7 @@ rspamd_symbols_cache_finalize_item (struct rspamd_task *task,
        g_assert (item->async_events == 0);
 
        msg_debug_cache_task ("process finalize for item %s", item->symbol);
-       setbit (checkpoint->processed_bits, item->id + 1);
+       setbit (checkpoint->processed_bits, item->id * 2 + 1);
        checkpoint->items_inflight --;
        checkpoint->cur_item = NULL;
 
@@ -2603,9 +2577,13 @@ rspamd_symbols_cache_finalize_item (struct rspamd_task *task,
 
        /* Process all reverse dependencies */
        PTR_ARRAY_FOREACH (item->rdeps, i, rdep) {
-               rspamd_symbols_cache_check_symbol (task, task->cfg->cache,
-                               rdep,
-                               checkpoint);
+               if (rdep->item) {
+                       if (!isset (checkpoint->processed_bits, rdep->item->id * 2)) {
+                               rspamd_symbols_cache_check_symbol (task, task->cfg->cache,
+                                               rdep->item,
+                                               checkpoint);
+                       }
+               }
        }
 }