better neighbor sortingHEADmrow

1 files changed, 261 insertions, 44 deletions

diff --git a/src/server.c b/src/server.c
index 7dc7382..06c5bd3 100644
--- a/src/server.c
+++ b/src/server.c
@@ -177,7 +177,7 @@ void* propagate_packet(void* _packet) {
     pthread_mutex_unlock(&neighbor_storage_mutex);
 
     for (uint64_t i = 0; i < active_count; i++) { pthread_join(neighbors_threads[ i ], NULL); }
-    
+
     log(_debug, "done");
 
     free(send_data);
@@ -1107,64 +1107,281 @@ void* cleanup_packets(void* arg) {
     return NULL;
 }
 
+static inline int neighbor_equal(const struct relay_neighbor* a, const struct relay_neighbor* b) {
+    if (a == NULL || b == NULL) return 0;
+    if (a->port != b->port) return 0;
+    if (a->has_ipv4 != b->has_ipv4) return 0;
+    if (a->has_ipv4) {
+        if (memcmp(a->v4_addr, b->v4_addr, 4) != 0) return 0;
+    }
+    if (a->has_ipv6 != b->has_ipv6) return 0;
+    if (a->has_ipv6) {
+        if (memcmp(a->v6_addr, b->v6_addr, 16) != 0) return 0;
+    }
+    return 1;
+}
+
+static inline int in_good_neighbors(
+    const struct relay_neighbor* cur,
+    const struct relay_neighbor* good_neighbors,
+    uint64_t                     good_count) {
+    uint64_t i;
+    for (i = 0; i < good_count; ++i) {
+        if (neighbor_equal(cur, &good_neighbors[ i ])) return 1;
+    }
+    return 0;
+}
+
+static inline int slot_matches_dead_snapshot(
+    uint64_t                     slot_idx,
+    const struct relay_neighbor* candidates,
+    const uint64_t*              candidate_indices,
+    const bool*                  candidate_dead,
+    uint64_t                     cc) {
+    uint64_t k;
+    for (k = 0; k < cc; ++k) {
+        if (!candidate_dead[ k ]) continue;
+        if (candidate_indices[ k ] != slot_idx) continue;
+        return 1;
+    }
+    return 0;
+}
+
 void* check_neighbors(void* arg) {
     sleep(5 * 30);
 
+    const int64_t PENALTY_CANNOT_STORE     = (int64_t) 1e9;
+    const int64_t WEIGHT_HELD_PACKETS      = 1000;
+    const int64_t WEIGHT_LEADING_DIFF      = 5000;
+    const int64_t PENALTY_NOT_ACCEPT_PLAIN = 10000;
+    const int     MAX_MOVES_PER_ITEM       = 8;
+
     while (neighbor_check_thread_running) {
         pthread_mutex_lock(&neighbor_storage_mutex);
-        for (uint64_t i = 0; i < max_neighbors; i++) {
-            if (!empty_neighbor_slots[ i ]) {
-                struct relay_neighbor neighbor = neighbor_storage[ i ];
-
-                byte*  response_data = NULL;
-                size_t response_len  = 0;
-                int    result        = -1;
-
-                byte audit_request[ 1 ] = { audit_server };
-
-                if (neighbor.has_ipv4) {
-                    char ip_string[ 16 ];
-                    sprintf(
-                        ip_string,
-                        "%d.%d.%d.%d",
-                        neighbor.v4_addr[ 0 ],
-                        neighbor.v4_addr[ 1 ],
-                        neighbor.v4_addr[ 2 ],
-                        neighbor.v4_addr[ 3 ]);
-                    result = study(
-                        ip_string, neighbor.port, audit_request, 1, &response_data, &response_len);
-                } else if (neighbor.has_ipv6) {
-                    char ip_string[ 46 ];
-                    int  offset = 0;
-
-                    for (int j = 0; j < 8; j++) {
-                        uint16_t word
-                            = (neighbor.v6_addr[ j * 2 ] << 8) | neighbor.v6_addr[ j * 2 + 1 ];
-                        offset += snprintf(
-                            ip_string + offset,
-                            sizeof(ip_string) - offset,
-                            "%x%s",
-                            word,
-                            (j < 7) ? ":" : "");
-                    }
 
-                    result = study(
-                        ip_string, neighbor.port, audit_request, 1, &response_data, &response_len);
-                }
+        uint64_t first_empty = max_neighbors;
+        int64_t  last_full   = -1;
+        for (uint64_t i = 0; i < max_neighbors; ++i) {
+            if (likely(!empty_neighbor_slots[ i ])) {
+                if ((int64_t) i > last_full) last_full = (int64_t) i;
+            } else {
+                if (first_empty == max_neighbors) first_empty = i;
+            }
+        }
 
-                if (result != 0 || response_data[ 0 ] != audit_response) {
-                    empty_neighbor_slots[ i ] = true;
-                    neighbor_count--;
-                    log(_warn, "removed dead neighbor");
+        if (last_full < 0) {
+            pthread_mutex_unlock(&neighbor_storage_mutex);
+            sleep(60 * 5);
+            continue;
+        }
+
+        uint64_t scan_start = (first_empty < (uint64_t) last_full) ? first_empty : 0;
+        uint64_t scan_end   = (uint64_t) last_full;
+
+        uint64_t candidate_count = 0;
+        for (uint64_t i = scan_start; i <= scan_end; ++i) {
+            if (unlikely(empty_neighbor_slots[ i ])) continue;
+            candidate_count++;
+        }
+
+        if (candidate_count == 0) {
+            pthread_mutex_unlock(&neighbor_storage_mutex);
+            sleep(60 * 5);
+            continue;
+        }
+
+        struct relay_neighbor* candidates = malloc(sizeof(struct relay_neighbor) * candidate_count);
+        uint64_t*              candidate_indices = malloc(sizeof(uint64_t) * candidate_count);
+        if (!candidates || !candidate_indices) {
+            free(candidates);
+            free(candidate_indices);
+            pthread_mutex_unlock(&neighbor_storage_mutex);
+            sleep(60 * 5);
+            continue;
+        }
+
+        uint64_t cc = 0;
+        for (uint64_t i = scan_start; i <= scan_end; ++i) {
+            if (unlikely(empty_neighbor_slots[ i ])) continue;
+            candidates[ cc ]        = neighbor_storage[ i ];
+            candidate_indices[ cc ] = i;
+            cc++;
+        }
+
+        pthread_mutex_unlock(&neighbor_storage_mutex);
+
+        struct relay_neighbor* good_neighbors = malloc(sizeof(struct relay_neighbor) * cc);
+        int64_t*               scores         = malloc(sizeof(int64_t) * cc);
+        bool*                  candidate_dead = calloc(cc, sizeof(bool));
+
+        if (!good_neighbors || !scores || !candidate_dead) {
+            free(candidates);
+            free(candidate_indices);
+            free(good_neighbors);
+            free(scores);
+            free(candidate_dead);
+            sleep(60 * 5);
+            continue;
+        }
+
+        uint64_t good_count = 0;
+        for (uint64_t i = 0; i < cc; ++i) {
+            struct relay_neighbor n = candidates[ i ];
+            char                  ip_string[ INET6_ADDRSTRLEN ];
+            ip_string[ 0 ] = '\0';
+
+            if (likely(n.has_ipv4)) {
+                snprintf(
+                    ip_string,
+                    sizeof(ip_string),
+                    "%d.%d.%d.%d",
+                    n.v4_addr[ 0 ],
+                    n.v4_addr[ 1 ],
+                    n.v4_addr[ 2 ],
+                    n.v4_addr[ 3 ]);
+            } else if (n.has_ipv6) {
+                int offset = 0;
+                for (int j = 0; j < 8; ++j) {
+                    uint16_t word = (n.v6_addr[ j * 2 ] << 8) | n.v6_addr[ j * 2 + 1 ];
+                    offset += snprintf(
+                        ip_string + offset,
+                        sizeof(ip_string) - offset,
+                        "%x%s",
+                        word,
+                        (j < 7) ? ":" : "");
                 }
+            } else {
+                candidate_dead[ i ] = true;
+                continue;
+            }
+
+            uint64_t          t0      = get_current_time_ms();
+            struct relay_info audit   = run_server_audit(ip_string, n.port);
+            uint64_t          t1      = get_current_time_ms();
+            int64_t           latency = (int64_t) ((t1 >= t0) ? (t1 - t0) : 0);
+
+            int all_zero_audit
+                = (audit.major_version == 0 && audit.minor_version == 0
+                   && audit.capabilities.can_store == 0 && audit.capabilities.leading_zeros == 0
+                   && audit.capabilities.accepts_plain == 0 && audit.statistics.neighbors == 0
+                   && audit.statistics.held_packets == 0);
+            if (all_zero_audit) {
+                candidate_dead[ i ] = true;
+                log(_warn,
+                    "neighbor %s:%u appears dead (all-zero audit), marking snapshot dead",
+                    ip_string,
+                    n.port);
+                continue;
+            }
+
+            int64_t score = latency;
+            score += (int64_t) audit.statistics.held_packets * WEIGHT_HELD_PACKETS;
+
+            int64_t leading_diff
+                = (int64_t) audit.capabilities.leading_zeros - (int64_t) server_stats.leading_zeros;
+            if (leading_diff < 0) leading_diff = -leading_diff;
+            score += leading_diff * WEIGHT_LEADING_DIFF;
+
+            if (!audit.capabilities.can_store) score += PENALTY_CANNOT_STORE;
+            if (!audit.capabilities.accepts_plain) score += PENALTY_NOT_ACCEPT_PLAIN;
+
+            if (audit.major_version != server_stats.major_version
+                || audit.minor_version != server_stats.minor_version) {
+                score += 20000;
+            }
+
+            good_neighbors[ good_count ] = n;
+            scores[ good_count ]         = score;
+            good_count++;
+
+            log(_info,
+                "neighbor-audit %s:%u latency %lld held %u leads %u can_store %d accept_plain %d "
+                "score %lld",
+                ip_string,
+                n.port,
+                (long long) latency,
+                (unsigned) audit.statistics.held_packets,
+                (unsigned) audit.capabilities.leading_zeros,
+                audit.capabilities.can_store ? 1 : 0,
+                audit.capabilities.accepts_plain ? 1 : 0,
+                (long long) score);
+        }
 
-                if (response_data) { free(response_data); }
+        for (uint64_t k = 1; k < good_count; ++k) {
+            uint64_t              j     = k;
+            int                   moves = 0;
+            struct relay_neighbor nv    = good_neighbors[ k ];
+            int64_t               ns    = scores[ k ];
+            while (j > 0 && ns < scores[ j - 1 ] && moves < MAX_MOVES_PER_ITEM) {
+                good_neighbors[ j ] = good_neighbors[ j - 1 ];
+                scores[ j ]         = scores[ j - 1 ];
+                --j;
+                ++moves;
             }
+            good_neighbors[ j ] = nv;
+            scores[ j ]         = ns;
+        }
+
+        pthread_mutex_lock(&neighbor_storage_mutex);
+
+        struct relay_neighbor* final_neighbors
+            = malloc(sizeof(struct relay_neighbor) * max_neighbors);
+        if (!final_neighbors) {
+            pthread_mutex_unlock(&neighbor_storage_mutex);
+            free(candidates);
+            free(candidate_indices);
+            free(good_neighbors);
+            free(scores);
+            free(candidate_dead);
+            free(final_neighbors);
+            sleep(60 * 5);
+            continue;
         }
+
+        uint64_t final_count = 0;
+        for (uint64_t i = 0; i < good_count && final_count < max_neighbors; ++i) {
+            final_neighbors[ final_count++ ] = good_neighbors[ i ];
+        }
+
+        for (uint64_t i = 0; i < max_neighbors && final_count < max_neighbors; ++i) {
+            if (unlikely(empty_neighbor_slots[ i ])) continue;
+            struct relay_neighbor* cur = &neighbor_storage[ i ];
+
+            if (in_good_neighbors(cur, final_neighbors, final_count)) continue;
+
+            int matched_dead_snapshot = 0;
+            for (uint64_t k = 0; k < cc; ++k) {
+                if (!candidate_dead[ k ]) continue;
+                if (candidate_indices[ k ] != i) continue;
+                if (neighbor_equal(cur, &candidates[ k ])) {
+                    matched_dead_snapshot = 1;
+                    break;
+                }
+            }
+            if (matched_dead_snapshot) { continue; }
+
+            final_neighbors[ final_count++ ] = *cur;
+        }
+
+        for (uint64_t i = 0; i < final_count; ++i) {
+            neighbor_storage[ i ]     = final_neighbors[ i ];
+            empty_neighbor_slots[ i ] = false;
+        }
+        for (uint64_t i = final_count; i < max_neighbors; ++i) { empty_neighbor_slots[ i ] = true; }
+        neighbor_count = final_count;
+
         pthread_mutex_unlock(&neighbor_storage_mutex);
 
+        free(final_neighbors);
+        free(candidates);
+        free(candidate_indices);
+        free(good_neighbors);
+        free(scores);
+        free(candidate_dead);
+
         sleep(60 * 5);
     }
+
     return NULL;
 }