FIxed dead RD detection

resource_system/src/arbiter/arbiter.c

@@ -92,11 +92,12 @@ void gruntTrackBegin (Grunt_Tracker *t, Grunt g)
 internal_function
 void gruntTrackEnd (Grunt_Tracker *t, Char *grunt_id)
 {
-    Size index = htLookup(&t->map, hashString(grunt_id));
+    U64 hash = hashString(grunt_id);
+    Size index = htLookup(&t->map, hash);
     sbufAdd(t->free_list, index);
     free(t->grunts[index].id);
     t->grunts[index] = (Grunt){0};
-    htRemove(&t->map, index);
+    htRemove(&t->map, hash);
 }
 
 # if defined(COMPILER_CLANG)
@@ -184,14 +185,18 @@ Sint main (Sint argc, Char *argv[])
 
     Sint time_of_launch = (Sint)time(0);
 
+    U64 time_passed_last = timeMilli();
     while (global_keep_running) {
         // NOTE(naman): Get the fd's that are ready
         rd_kafka_message_t *kafka_message_read = rd_kafka_consumer_poll(kafka.reader, 0);
 
+        U64 time_passed_now = timeMilli();
+        U64 time_passed = time_passed_now - time_passed_last;
+        time_passed_last = time_passed_now;
+
         for (Size j = 0; j < gt.map.slot_count; j++) {
-            if (gt.map.values[j] != 0) {
-                Grunt g = gt.grunts[gt.map.values[j]];
-                g.time_to_die -= 10;
+            if (gt.map.keys[j] != 0) {
+                gt.grunts[gt.map.values[j]].time_to_die -= time_passed;
             }
         }
 
@@ -256,7 +261,7 @@ Sint main (Sint argc, Char *argv[])
                     }
                 } else if (kafka_message_read->rkt == topic_join_g2a) {
                     Char *id = strdup(cJSON_GetObjectItem(root, "node_id")->valuestring);
-                    Grunt grunt = {.id = id, .time_to_die = 1000};
+                    Grunt grunt = {.id = id, .time_to_die = 2000};
 
                     logMessage("Join G2A:\tid: %s", id);
 
@@ -269,7 +274,7 @@ Sint main (Sint argc, Char *argv[])
                     U64 index = htLookup(&gt.map, hashString(id));
                     if (index != 0) { // Prevent any left over message
                         // TODO(naman): Add any new resource fields here
-                        gt.grunts[index].time_to_die = 1000;
+                        gt.grunts[index].time_to_die = 2000;
 
                         gt.grunts[index].memory = cJSON_GetObjectItem(root, "memory")->valueint;
                         logMessage("Beat G2A:\tid: %s (Memory %d)", id, gt.grunts[index].memory);
@@ -322,6 +327,7 @@ Sint main (Sint argc, Char *argv[])
                 Size index = gt.map.values[j];
                 Grunt g = gt.grunts[index];
                 if (g.time_to_die <= 0) {
+                    printf("Deleting grunt: %s\n", g.id);
                     gruntTrackEnd(&gt, g.id);
                 }
             }
@@ -334,7 +340,7 @@ Sint main (Sint argc, Char *argv[])
                 U64 milli_new = timeMilli();
                 gs->milli_passed += milli_new - gs->milli_last;
                 gs->milli_last = milli_new;
-                if (gs->milli_passed >= 1000) {
+                if (gs->milli_passed >= 2000) {
                     Command c = {.kind = Command_RESPONSE_ARBITER_2_DM};
                     c.resource_id = gs->resource_id;