Items stuck at rte_hash

[Felipe Pereira <felipe@spint.com.br>]

Hello,

I'm having trouble using rte_hash, as it would 'lock' some items
It runs like a charm in the beginning, but after some time, it appears 
that some items are not being correctly removed
I've tried with rcu and locks, without rcu and not using any 
extra_flags, and lot of other things, all without success
For the first 15 seconds, everything looks good
After this, the rte_hash_count start dropping, goes bellow 0 and returns 
to UINT_MAX, but when counting the items when iterating trough the hash, 
it appears fine
But after around 4 to 5 minutes (sometimes more), the iteration count 
begins to go up, and the items not found (that i have deleted, but are 
showing when iterating) begins to increase.
At the end of the program, there are always a lot of items left in the list.
Could this be some problem with the list, maybe running out of space or 
having trouble deleting items with hash colision
Or is it something that I'm doing wrong ?

The output of my run was this:

[+ 1 s] Iter_Items: 114735 | Hash_Items 114726 | Removed: 0 | Not found: 0
[+ 2 s] Iter_Items: 231794 | Hash_Items 231785 | Removed: 0 | Not found: 0
[+ 3 s] Iter_Items: 351164 | Hash_Items 351155 | Removed: 0 | Not found: 0
[+ 4 s] Iter_Items: 473058 | Hash_Items 473049 | Removed: 0 | Not found: 0
[+ 6 s] Iter_Items: 566082 | Hash_Items 566073 | Removed: 33098 | Not 
found: 0
[+ 7 s] Iter_Items: 572731 | Hash_Items 572722 | Removed: 127875 | Not 
found: 0
[+ 8 s] Iter_Items: 570068 | Hash_Items 570059 | Removed: 138554 | Not 
found: 0
[+ 9 s] Iter_Items: 566778 | Hash_Items 566769 | Removed: 139760 | Not 
found: 0
[+11 s] Iter_Items: 563572 | Hash_Items 563563 | Removed: 139683 | Not 
found: 0
[+12 s] Iter_Items: 561858 | Hash_Items 561849 | Removed: 137552 | Not 
found: 0
[+13 s] Iter_Items: 561689 | Hash_Items 561680 | Removed: 135999 | Not 
found: 0
[+14 s] Iter_Items: 561712 | Hash_Items 561703 | Removed: 136548 | Not 
found: 0
[+16 s] Iter_Items: 561367 | Hash_Items 435598 | Removed: 136681 | Not 
found: 0
[+17 s] Iter_Items: 561991 | Hash_Items 300510 | Removed: 135721 | Not 
found: 0
[+18 s] Iter_Items: 562183 | Hash_Items 164894 | Removed: 135810 | Not 
found: 0
[+20 s] Iter_Items: 561991 | Hash_Items 28558 | Removed: 136137 | Not 
found: 0
[+21 s] Iter_Items: 561690 | Hash_Items 4294859233 | Removed: 136316 | 
Not found: 0
[+22 s] Iter_Items: 561306 | Hash_Items 4294722529 | Removed: 136316 | 
Not found: 0
[+23 s] Iter_Items: 561223 | Hash_Items 4294586142 | Removed: 136318 | 
Not found: 0
[+25 s] Iter_Items: 561345 | Hash_Items 4294449992 | Removed: 136272 | 
Not found: 0
[+26 s] Iter_Items: 560814 | Hash_Items 4294313029 | Removed: 136433 | 
Not found: 0
[+27 s] Iter_Items: 560436 | Hash_Items 4294176699 | Removed: 135949 | 
Not found: 0
[+28 s] Iter_Items: 561268 | Hash_Items 4294041867 | Removed: 135659 | 
Not found: 0
[+30 s] Iter_Items: 561344 | Hash_Items 4293905559 | Removed: 136390 | 
Not found: 0
...
[+271 s] Iter_Items: 551732 | Hash_Items 4288675841 | Removed: 134114 | 
Not found: 0
[+273 s] Iter_Items: 551692 | Hash_Items 4288675841 | Removed: 134093 | 
Not found: 0
[+274 s] Iter_Items: 551408 | Hash_Items 4288675842 | Removed: 134031 | 
Not found: 0
[+275 s] Iter_Items: 551856 | Hash_Items 4288675841 | Removed: 133723 | 
Not found: 0
[+276 s] Iter_Items: 554636 | Hash_Items 4288675843 | Removed: 131317 | 
Not found: 121
[+278 s] Iter_Items: 573530 | Hash_Items 4288675841 | Removed: 114045 | 
Not found: 6621
[+279 s] Iter_Items: 599976 | Hash_Items 4288675841 | Removed: 105931 | 
Not found: 22822
[+280 s] Iter_Items: 632009 | Hash_Items 4288675841 | Removed: 100389 | 
Not found: 17505
[+281 s] Iter_Items: 663134 | Hash_Items 4288675841 | Removed: 101299 | 
Not found: 16571
[+283 s] Iter_Items: 689778 | Hash_Items 4288675842 | Removed: 107426 | 
Not found: 17118
[+284 s] Iter_Items: 710009 | Hash_Items 4288675841 | Removed: 115318 | 
Not found: 21135
[+285 s] Iter_Items: 720070 | Hash_Items 4288675841 | Removed: 126528 | 
Not found: 33973
[+287 s] Iter_Items: 722742 | Hash_Items 4288675842 | Removed: 134906 | 
Not found: 33721
[+288 s] Iter_Items: 722672 | Hash_Items 4288675841 | Removed: 137514 | 
Not found: 34587
[+289 s] Iter_Items: 722319 | Hash_Items 4288675841 | Removed: 137900 | 
Not found: 32162
[+290 s] Iter_Items: 722133 | Hash_Items 4288675841 | Removed: 137498 | 
Not found: 31933
[+292 s] Iter_Items: 722219 | Hash_Items 4288675841 | Removed: 137711 | 
Not found: 35989
[+293 s] Iter_Items: 722527 | Hash_Items 4288675841 | Removed: 137602 | 
Not found: 34185
[+294 s] Iter_Items: 722389 | Hash_Items 4288675841 | Removed: 137727 | 
Not found: 34689
[+296 s] Iter_Items: 722108 | Hash_Items 4288675841 | Removed: 137533 | 
Not found: 32894
[+297 s] Iter_Items: 720595 | Hash_Items 4288675841 | Removed: 137628 | 
Not found: 34177
[+298 s] Iter_Items: 719408 | Hash_Items 4288675841 | Removed: 137677 | 
Not found: 38662
[+300 s] Iter_Items: 718916 | Hash_Items 4288675841 | Removed: 137589 | 
Not found: 37144
 >> Stopping insertions...
[+301 s] Iter_Items: 516547 | Hash_Items 4288675841 | Removed: 202369 | 
Not found: 59775
[+302 s] Iter_Items: 361146 | Hash_Items 4288675841 | Removed: 155401 | 
Not found: 106743
[+303 s] Iter_Items: 242023 | Hash_Items 4288675841 | Removed: 119123 | 
Not found: 143021
[+305 s] Iter_Items: 165981 | Hash_Items 4288675841 | Removed: 76042 | 
Not found: 165972
[+306 s] Iter_Items: 165981 | Hash_Items 4288675841 | Removed: 0 | Not 
found: 165972
[+307 s] Iter_Items: 165981 | Hash_Items 4288675841 | Removed: 0 | Not 
found: 165972
[+308 s] Iter_Items: 165981 | Hash_Items 4288675841 | Removed: 0 | Not 
found: 165972
...
[+328 s] Iter_Items: 165981 | Hash_Items 4288675841 | Removed: 0 | Not 
found: 165972
[+329 s] Iter_Items: 165981 | Hash_Items 4288675841 | Removed: 0 | Not 
found: 165972
[+330 s] Iter_Items: 165981 | Hash_Items 4288675841 | Removed: 0 | Not 
found: 165972
 >> Force exiting after max time.

--- Benchmark Statistics ---
Inserts/sec: 31805132 96379.19
Deletes/sec: 31639151 95876.22
Reads/sec:   10549998738 31969693.15
Benchmark completed.

Below is my code:

#include <stdio.h>

#include <stdint.h>
#include <stdlib.h>
#include <stdbool.h>
#include <unistd.h>
#include <time.h>

#include <rte_eal.h>
#include <rte_lcore.h>
#include <rte_hash.h>
#include <rte_jhash.h>
#include <rte_malloc.h>
#include <rte_cycles.h>
#include <rte_rcu_qsbr.h>
#include <rte_atomic.h>
#include <rte_spinlock.h>

#define BENCH_TIME 300
#define BENCH_MAX_TIME 330
#define CLEANUP_TIMEOUT_CYCLES (5 * rte_get_timer_hz())
#define HASH_CAPACITY (10 * 1024 * 1024)
#define MAX_EXPIRED_KEYS (256 * 1024)

struct item
{
     uint64_t key;
     uint64_t timestamp;
};

struct thread_data
{
     uint64_t rand_seed;
     uint64_t rand_counter;
};

static struct rte_hash *hash_table;
static struct rte_rcu_qsbr *qsbr;
static volatile bool stop_insert = false;
static volatile bool stop_all = false;

static rte_atomic64_t total_inserts;
static rte_atomic64_t total_reads;
static rte_atomic64_t total_deletes;
static rte_atomic32_t itens_in_hash;
static rte_atomic32_t not_found_keys;

static rte_spinlock_t insert_lock;

static uint64_t rand_key(struct thread_data *tdata)
{
     if (tdata->rand_counter == 0)
     {
         tdata->rand_seed = ((uint64_t)rand() << 32) | rand();
         tdata->rand_counter = 1000;
     }
     uint64_t key = tdata->rand_seed;
     tdata->rand_seed += 3;
     tdata->rand_counter--;
     return key;
}

static int worker_fn(__rte_unused void *arg)
{
     unsigned int lcore_id = rte_lcore_id();
     int count = 0;
     struct thread_data tdata = {.rand_seed = 1234, .rand_counter = 1000};
     uint32_t current_items;

     rte_rcu_qsbr_thread_register(qsbr, lcore_id);
     rte_rcu_qsbr_thread_online(qsbr, lcore_id);

     while (!stop_all)
     {
         uint64_t key = rand_key(&tdata);
         rte_hash_lookup_data(hash_table, &key, NULL);
         rte_atomic64_inc(&total_reads);
         count++;

         if (count >= 300)
         {
             count = 0;
             current_items = rte_atomic32_read(&itens_in_hash);
             if (!stop_insert && current_items < HASH_CAPACITY * 0.75)
             {
                 struct item *itm = rte_malloc(NULL, sizeof(struct 
item), 0);
                 if (!itm)
                 {
                     fprintf(stderr, "Memory allocation failed\n");
                     continue;
                 }
                 itm->key = key;
                 itm->timestamp = rte_get_timer_cycles();

                 rte_spinlock_lock(&insert_lock);
                 rte_hash_add_key_data(hash_table, &itm->key, itm);
                 rte_spinlock_unlock(&insert_lock);

                 rte_atomic64_inc(&total_inserts);
                 rte_atomic32_inc(&itens_in_hash);
             }
         }

         rte_rcu_qsbr_quiescent(qsbr, lcore_id);
     }

     rte_rcu_qsbr_thread_offline(qsbr, lcore_id);
     rte_rcu_qsbr_thread_unregister(qsbr, lcore_id);
     return 0;
}

static int cleanup_timeout()
{
     const uint64_t now = rte_get_timer_cycles();
     const uint64_t cutoff = now - CLEANUP_TIMEOUT_CYCLES;

     uint32_t iter = 0;
     const void *key;
     void *data;
     uint64_t expired_keys[MAX_EXPIRED_KEYS];
     int expired_count = 0;
     rte_atomic32_set(&not_found_keys, 0);

     while (rte_hash_iterate(hash_table, &key, &data, &iter) >= 0)
     {
         struct item *itm = (struct item *)data;
         if ((itm->timestamp < cutoff || stop_insert) && expired_count < 
MAX_EXPIRED_KEYS)
         {
             expired_keys[expired_count++] = *(const uint64_t *)key;
         }
     }

     int removed = 0;
     for (int i = 0; i < expired_count; ++i)
     {
         if (rte_hash_lookup_data(hash_table, &expired_keys[i], &data) >= 0)
         {
             if (rte_hash_del_key(hash_table, &expired_keys[i]) >= 0)
             {
                 rte_rcu_qsbr_synchronize(qsbr, RTE_QSBR_THRID_INVALID);
                 rte_free(data);
                 removed++;
                 rte_atomic32_dec(&itens_in_hash);
             }
         }
         else
         {
             rte_atomic32_add(&not_found_keys, 1);
         }
     }

     rte_atomic64_add(&total_deletes, removed);
     return removed;
}

int main(int argc, char **argv)
{
     int ret = rte_eal_init(argc, argv);
     if (ret < 0)
     {
         fprintf(stderr, "EAL initialization failed\n");
         return 1;
     }

     struct rte_hash_parameters params = {
         .name = "benchmark_hash",
         .entries = HASH_CAPACITY,
         .key_len = sizeof(uint64_t),
         .hash_func = rte_jhash,
         .hash_func_init_val = 0,
         .socket_id = SOCKET_ID_ANY,
         .extra_flag = 0 // no RW_CONCURRENCY
     };

     hash_table = rte_hash_create(&params);
     if (!hash_table)
     {
         fprintf(stderr, "Failed to create hash table\n");
         return 1;
     }

     qsbr = rte_zmalloc(NULL, 
rte_rcu_qsbr_get_memsize(rte_lcore_count()), RTE_CACHE_LINE_SIZE);
     if (!qsbr || rte_rcu_qsbr_init(qsbr, rte_lcore_count()) != 0)
     {
         fprintf(stderr, "Failed to initialize QSBR\n");
         return 1;
     }

     struct rte_hash_rcu_config rcu_cfg = {
         .v = qsbr,
         .trigger_reclaim_limit = 1000000};
     rte_hash_rcu_qsbr_add(hash_table, &rcu_cfg);

     rte_atomic64_init(&total_inserts);
     rte_atomic64_init(&total_reads);
     rte_atomic64_init(&total_deletes);
     rte_atomic32_init(&itens_in_hash);
     rte_spinlock_init(&insert_lock);

     unsigned lcore_id;
     RTE_LCORE_FOREACH_WORKER(lcore_id)
     {
         rte_eal_remote_launch(worker_fn, NULL, lcore_id);
     }

     time_t start = time(NULL);
     while (true)
     {
         sleep(1);
         int removed = cleanup_timeout();
         time_t now = time(NULL);
         uint32_t current_items = rte_atomic32_read(&itens_in_hash);
         uint32_t not_found = rte_atomic32_read(&not_found_keys);

         printf("[+%2ld s] Iter_Items: %u | Hash_Items %u | Removed: %d 
| Not found: %u\n",
                now - start, current_items, rte_hash_count(hash_table), 
removed, not_found);

         if (!stop_insert && now - start >= BENCH_TIME)
         {
             stop_insert = true;
             printf(">> Stopping insertions...\n");
         }

         if (stop_insert && current_items == 0)
             break;

         if (now - start >= BENCH_MAX_TIME)
         {
             printf(">> Force exiting after max time.\n");
             break;
         }
     }

     stop_all = true;
     RTE_LCORE_FOREACH_WORKER(lcore_id)
     {
         rte_eal_wait_lcore(lcore_id);
     }

     cleanup_timeout();
     rte_hash_free(hash_table);

     double elapsed = difftime(time(NULL), start);
     uint64_t inserts = rte_atomic64_read(&total_inserts);
     uint64_t deletes = rte_atomic64_read(&total_deletes);
     uint64_t reads = rte_atomic64_read(&total_reads);

     printf("\n--- Benchmark Statistics ---\n");
     printf("Inserts/sec: %lu %.2f\n", inserts, inserts / elapsed);
     printf("Deletes/sec: %lu %.2f\n", deletes, deletes / elapsed);
     printf("Reads/sec:   %lu %.2f\n", reads, reads / elapsed);
     printf("Benchmark completed.\n");

     return 0;
}