Forget unnecessary deltas

diff --git a/src/clients.c b/src/clients.c
index 5169f2d7d4867751c7d4b08989c60ce8bd572c8a..955976b73e5cc4b7d1940804425d219d01ac556c 100644 (file)
--- a/src/clients.c
+++ b/src/clients.c
@@ -22,6 +22,8 @@ struct hashable_client {
 static struct hashable_client *table;
 /** Read/write lock, which protects @table and its inhabitants. */
 static pthread_rwlock_t lock;
+/** Serial number from which deltas must be stored */
+uint32_t min_serial;
 
 int
 clients_db_init(void)
@@ -32,7 +34,7 @@ clients_db_init(void)
        error = pthread_rwlock_init(&lock, NULL);
        if (error)
                return pr_errno(error, "pthread_rwlock_init() errored");
-
+       min_serial = 0;
        return 0;
 }
 
@@ -108,6 +110,48 @@ clients_add(int fd, struct sockaddr_storage *addr, uint8_t rtr_version)
        return error;
 }
 
+void
+clients_update_serial(int fd, uint32_t serial)
+{
+       struct hashable_client *cur_client;
+
+       rwlock_write_lock(&lock);
+       HASH_FIND_INT(table, &fd, cur_client);
+       if (cur_client == NULL)
+               goto unlock;
+
+       cur_client->meat.serial_number = serial;
+
+unlock:
+       rwlock_unlock(&lock);
+}
+
+uint32_t
+clients_get_min_serial(void)
+{
+       struct hashable_client *current, *ptr;
+       uint32_t result;
+
+       rwlock_write_lock(&lock);
+       if (HASH_COUNT(table) == 0)
+               goto unlock;
+
+       HASH_ITER(hh, table, current, ptr) {
+               if (current == table) {
+                       min_serial = current->meat.serial_number;
+                       continue;
+               }
+               if (current->meat.serial_number < min_serial)
+                       min_serial = current->meat.serial_number;
+       }
+
+unlock:
+       result = min_serial;
+       rwlock_unlock(&lock);
+
+       return result;
+}
+
 void
 clients_forget(int fd)
 {

diff --git a/src/clients.h b/src/clients.h
index 9286a61ac5cb4d8b26a9de786d6da5526c5fb7a1..7be04f82a9ab72e47a184b8df7f2e640b2b921bc 100644 (file)
--- a/src/clients.h
+++ b/src/clients.h
@@ -14,14 +14,17 @@ struct client {
        };
        in_port_t sin_port;
        uint8_t rtr_version;
+       uint32_t serial_number;
 };
 
 int clients_db_init(void);
 
 int clients_add(int, struct sockaddr_storage *, uint8_t);
+void clients_update_serial(int, uint32_t);
 void clients_forget(int);
 typedef int (*clients_foreach_cb)(struct client const *, void *);
 int clients_foreach(clients_foreach_cb, void *);
+uint32_t clients_get_min_serial(void);
 
 void clients_db_destroy(void);
 

diff --git a/src/rtr/db/vrps.c b/src/rtr/db/vrps.c
index 0be01e5243ac98505114ecbfcfc8aee05fc1e447..14038ced85b478b9bc1a79740c6661432639e7bb 100644 (file)
--- a/src/rtr/db/vrps.c
+++ b/src/rtr/db/vrps.c
@@ -3,6 +3,7 @@
 #include <pthread.h>
 #include <stdbool.h>
 #include <string.h>
+#include "clients.h"
 #include "common.h"
 #include "data_structure/array_list.h"
 
@@ -218,6 +219,58 @@ vrps_foreach_delta_roa(uint32_t from, uint32_t to, vrp_foreach_cb cb, void *arg)
        return error;
 }
 
+/**
+ * Reallocate the array of @db starting at @start, the length and capacity are
+ * calculated according to the new start.
+ */
+static void
+resize_deltas_db(struct deltas_db *db, struct delta *start)
+{
+       struct delta *tmp;
+
+       db->len -= (start - db->array);
+       while (db->len < db->capacity / 2)
+               db->capacity /= 2;
+       tmp = malloc(sizeof(struct delta) * db->capacity);
+       if (tmp == NULL) {
+               pr_enomem();
+               return;
+       }
+       memcpy(tmp, start, db->len * sizeof(struct delta));
+       free(db->array);
+       db->array = tmp;
+}
+
+void
+vrps_purge(void)
+{
+       struct delta *d;
+       uint32_t min_serial;
+       int error;
+
+       min_serial = clients_get_min_serial();
+       error = rwlock_read_lock(&lock);
+       if (error) {
+               pr_err("Couldn't lock deltas DB to purge it");
+               return;
+       }
+
+       /** Assume is ordered by serial, so get the new initial pointer */
+       ARRAYLIST_FOREACH(&state.deltas, d)
+               if (d->serial >= min_serial)
+                       break;
+
+       /** Is the first element or reached end, nothing to purge */
+       if (d == state.deltas.array ||
+           (d - state.deltas.array) == state.deltas.len)
+               goto rlock_succeed;
+
+       resize_deltas_db(&state.deltas, d);
+
+rlock_succeed:
+       rwlock_unlock(&lock);
+}
+
 int
 get_last_serial_number(uint32_t *result)
 {

diff --git a/src/rtr/db/vrps.h b/src/rtr/db/vrps.h
index 77fe1223d58c219e787181078e57a53a41307491..92734d239aa3be0a206bc6785f4932fcab2bc1cb 100644 (file)
--- a/src/rtr/db/vrps.h
+++ b/src/rtr/db/vrps.h
@@ -26,6 +26,7 @@ int deltas_db_status(uint32_t *, enum delta_status *);
 
 int vrps_foreach_base_roa(vrp_foreach_cb, void *);
 int vrps_foreach_delta_roa(uint32_t, uint32_t, vrp_foreach_cb, void *);
+void vrps_purge(void);
 
 int get_last_serial_number(uint32_t *);
 uint16_t get_current_session_id(uint8_t);

diff --git a/src/rtr/pdu_sender.c b/src/rtr/pdu_sender.c
index 00f18aea2421034ec1557576330343d91050b6c4..e3f0244afa2b05c24d89835af16fe658bf77a779 100644 (file)
--- a/src/rtr/pdu_sender.c
+++ b/src/rtr/pdu_sender.c
@@ -7,6 +7,7 @@
 #include <unistd.h>
 #include <sys/queue.h>
 
+#include "clients.h"
 #include "config.h"
 #include "log.h"
 #include "rtr/pdu_serializer.h"
@@ -348,6 +349,7 @@ send_end_of_data_pdu(struct sender_common *common)
        struct end_of_data_pdu pdu;
        unsigned char data[BUFFER_SIZE];
        size_t len;
+       int error;
 
        set_header_values(&pdu.header, common->version, PDU_TYPE_END_OF_DATA,
            *common->session_id);
@@ -361,7 +363,12 @@ send_end_of_data_pdu(struct sender_common *common)
 
        len = serialize_end_of_data_pdu(&pdu, data);
 
-       return send_response(common->fd, data, len);
+       error = send_response(common->fd, data, len);
+       if (error)
+               return error;
+
+       clients_update_serial(common->fd, pdu.serial_number);
+       return error;
 }
 
 int

diff --git a/src/updates_daemon.c b/src/updates_daemon.c
index 2dea6b4097572e8f317638206d9386d39ab259c2..bf282dfefff1d187bda34327dfa35267e7ab408b 100644 (file)
--- a/src/updates_daemon.c
+++ b/src/updates_daemon.c
@@ -99,6 +99,10 @@ check_vrps_updates(void *param_void)
                }
 
                old_roas = validation_handler.arg;
+
+               /** Remove unnecessary deltas */
+               vrps_purge();
+
                error = notify_clients();
                if (error)
                        pr_debug("Could not notify clients of the new VRPs. (Error code %d.) Sleeping...",