Bug fix (temporary fix until we do broader refactor) - continue immediately after we free DAG runInfo that returned with an error.

Author: alonre24

src/background_workers.c

@@ -391,19 +391,16 @@ void *RedisAI_Run_ThreadMain(void *arg) {
 
                 // Run is over, now iterate over the run info structs in the batch
                 // and see if any error was generated
-                int dagError = 0;
+                int first_dag_error = 0;
                 for (long long i = 0; i < array_len(batch_rinfo); i++) {
                     RedisAI_RunInfo *rinfo = batch_rinfo[i];
-                    // We lock on the DAG because error could be set from
-                    // other threads operating on the same DAG (TODO: use atomic)
-                    dagError = __atomic_load_n(rinfo->dagError, __ATOMIC_RELAXED);
-
                     // We record that there was an error for later on
-                    run_error = dagError;
-
+                    run_error = __atomic_load_n(rinfo->dagError, __ATOMIC_RELAXED);
+                    if (i == 0 && run_error == 1)
+                        first_dag_error = 1;
                     // If there was an error and the reference count for the dag
                     // has gone to zero and the client is still around, we unblock
-                    if (dagError) {
+                    if (run_error) {
                         RedisAI_RunInfo *orig = rinfo->orig_copy;
                         long long dagRefCount = RAI_DagRunInfoFreeShallowCopy(rinfo);
                         if (dagRefCount == 0) {
@@ -415,6 +412,10 @@ void *RedisAI_Run_ThreadMain(void *arg) {
                         __atomic_add_fetch(rinfo->dagCompleteOpCount, 1, __ATOMIC_RELAXED);
                     }
                 }
+                if (first_dag_error) {
+                    run_queue_len = queueLength(run_queue_info->run_queue);
+                    continue;
+                }
             }
 
             // We initialize variables where we'll store the fact hat, after the current