DPDK——哈希库使用
一、概述
最近项目中使用了DPDK作为底层框架，有个需求想借助DPDK的哈希库，于是便调研了下相关使用，本文将介绍哈希库的简单使用，使用的是DPDP19.08版本。
首先我们打开DPDK的说明文档（https://doc.dpdk.org/api-19.08/rte__hash_8h.html），其主要的函数接口如下：
1.1 建立哈希表
类似于文件操作，首先咱得创建一个哈希表，得到一个类似于句柄的东东：
struct rte_hash* rte_hash_create(const struct rte_hash_parameters *params)这里需要定义一个struct rte_hash_parameters的参数，其结构如下：
struct rte_hash_parameters {const char *name;/< Name of the hash. */uint32_t entries;/< Total hash table entries. */uint32_t reserved;/< Unused field. Should be set to 0 */uint32_t key_len;/< Length of hash key. */rte_hash_function hash_func;/< Primary Hash function used to calculate hash. */uint32_t hash_func_init_val;/< Init value used by hash_func. */int socket_id;/< NUMA Socket ID for memory. */uint8_t extra_flag;/< Indicate if additional parameters are present. */};其中，entries也就是哈希表的条目数最大支持1<<30（1,073,741,824）大小，extra_flag可以设置多线程和扩展桶操作，具体操作有以下几种：
1.RTE_HASH_EXTRA_FLAGS_MULTI_WRITER_ADD：插入的默认行为，允许多个线程写入哈希表，如果仅设置此标志，无法保护“读者”免受正在进行的写入操作的影响。
2.RTE_HASH_EXTRA_FLAGS_RW_CONCURRENCY：支持读写并发，使用读写锁提供并发。
3.RTE_HASH_EXTRA_FLAGS_EXT_TABLE：哈希桶将使用链表进行扩展，以插入这些失败的键。 对于插入哈希表的键个数能达到容量的100％且不能容忍任何键插入失败（即使很少）的工作负载，此功能非常重要。
4.RTE_HASH_EXTRA_FLAGS_NO_FREE_ON_DEL：在调用delete（）时不会释放哈希表中条目的位置，当启用RTE_HASH_EXTRA_FLAGS_RW_CONCURRENCY_LF 时默认启用此功能。
5.RTE_HASH_EXTRA_FLAGS_RW_CONCURRENCY_LF：提供不使用读写锁的读/写并发，目前该设置不支持可扩展存储桶表功能。
6.RTE_HASH_EXTRA_FLAGS_TRANS_MEM_SUPPORT：启用硬件事务内存支持，如果硬件支持，读写锁将使用硬件事务性内存（例如Intel®TSX）来保证线程安全。如果平台支持Intel®TSX，建议设置事务性内存标志，因为这将加快并发表操作。否则，由于软件锁定机制相关的开销，并发操作将变慢。
2.2 增、改
int32_trte_hash_add_key(const struct rte_hash *h, const void *key);intrte_hash_add_key_data(const struct rte_hash *h, const void *key, void *data);int32_trte_hash_add_key_with_hash_data(const struct rte_hash *h, const void *key,hash_sig_t sig, void *data);int32_trte_hash_add_key_with_hash_data(const struct rte_hash *h, const void *key,hash_sig_t sig, void *data);int32_trte_hash_del_key(const struct rte_hash *h, const void *key);2.3 删
int32_trte_hash_del_key(const struct rte_hash *h, const void *key);int32_trte_hash_del_key_with_hash(const struct rte_hash *h, const void *key, hash_sig_t sig);2.4 查
intrte_hash_get_key_with_position(const struct rte_hash *h, const int32_t position,void **key);intrte_hash_lookup_data(const struct rte_hash *h, const void *key, void **data);intrte_hash_lookup_with_hash_data(const struct rte_hash *h, const void *key,hash_sig_t sig, void **data);int32_trte_hash_lookup(const struct rte_hash *h, const void *key);int32_trte_hash_lookup_with_hash(const struct rte_hash *h,const void *key, hash_sig_t sig);hash_sig_trte_hash_hash(const struct rte_hash *h, const void *key);intrte_hash_lookup_bulk_data(const struct rte_hash *h, const void **keys, uint32_t num_keys, uint64_t *hit_mask, void *data[]);intrte_hash_lookup_bulk(const struct rte_hash *h, const void **keys, uint32_t num_keys, int32_t *positions); int32_trte_hash_iterate(const struct rte_hash *h, const void **key, void **data, uint32_t *next); 二、实现
上述给出了大部分的函数接口，下面我们借助其中的一部分函数测试下200万键值对的存放和查找效率，C代码如下：
#include #include #include #include #include #include #include #include #include #include #define MAX_NUM 2000000struct rte_hash *create_hash(){ struct rte_hash *pstHash;const struct rte_hash_parameters stHashParam ={.name = "hash_test",.entries = (1 << 24),.reserved = 0,.key_len = 16,.hash_func = rte_jhash,.hash_func_init_val = 0,.socket_id = rte_socket_id(),.extra_flag = 0,};pstHash = rte_hash_create(&stHashParam); if (!pstHash) { printf("rte_hash_create() fail!\n"); return NULL; } printf("create hash_table: [%s] success!\n", stHashParam.name); return pstHash;}void dbg_print(char *info, unsigned char *dat, int dat_len){ int i; printf("\n%s:%d\n", info, dat_len); for (i = 0; i < dat_len; i++) { if (i && (i % 16 == 0)) printf("\n"); printf("%02x ", dat[i]); } printf("\n");}int main(int argc, char **argv){ struct rte_hash *pstHash; void *data = NULL; char acSupi[15]; int *iData = NULL; int iRet = 0; int iSize = 0; float time_s, time_ns, time_ms; struct timespec ts1, ts2; int i = 0;iRet = rte_eal_init(argc, argv);if(iRet < 0)rte_exit(EXIT_FAILURE, "rte_eal_init()\n"); pstHash = create_hash(); iData = (int*)malloc(sizeof(int) * MAX_NUM); if(iData == NULL)return -1;#if 1 //clock_gettime(CLOCK_MONOTONIC, &ts1); //记录函数开始时间 // 测试200万存储时间 memset(acSupi, 0x00, sizeof acSupi); for(i = 0; i < MAX_NUM; ++i) { iData[i] = i + 1; sprintf(acSupi, "%d", i); iRet = rte_hash_add_key_data(pstHash, acSupi, (void*)&iData[i]);/* if(i % 100000 == 0) { dbg_print("acSupi", (unsigned char *)acSupi, sizeof(acSupi)); printf("acSupi : %s\n", acSupi); printf("iRet = %d, iData = %d\n", iRet, iData); }*/ }#endif#if 1 // 测试200万查找效率 clock_gettime(CLOCK_MONOTONIC, &ts1); //记录函数开始时间 memset(acSupi, 0x00, sizeof acSupi); for(i = 0; i < MAX_NUM; ++i) { sprintf(acSupi, "%d", i); iRet = rte_hash_lookup_data(pstHash, acSupi, &data);/*if(i % 100000 == 0){ dbg_print("acSupi", (unsigned char *)acSupi, sizeof(acSupi)); printf("iRet = %d, data = %d\n", iRet, *(int *)data);}*/ }#endifclock_gettime(CLOCK_MONOTONIC, &ts2); // 记录函数结束时间 time_s = ts2.tv_sec - ts1.tv_sec; // 得到秒的时间 time_ns = (ts2.tv_nsec - ts1.tv_nsec); // 得到纳秒的时间 time_ms = time_s * 1000 + time_ns / 1000000; printf("\n*\n");iSize = rte_hash_count(pstHash); printf("htable size = %d\n", iSize); printf("run time is %f ms \n", time_ms);printf("\n*\n"); return 0;}在虚拟机测试单线程200万连续存储和查找均不到1秒：
注意！！！
我在测试的犯了一个很粗心的错误，rte_hash_add_key_data (const struct rte_hash *h, const void *key, void *data)这个函数在存储data时，是以变量的地址为准，而我存储200万data时，只是对同一个变量设置不同的值来存入哈希表，这样导致后面使用rte_hash_lookup_data (const struct rte_hash *h, const void *key, void data)时得到的data内容都是同一个值！
参考资料：
后台服务器：https://course.0voice.com/v1/course/intro?courseId=5&agentId=0
https://doc.dpdk.org/api-19.08/rte__hash_8h.html
https://blog.csdn.net/qq_15437629/article/details/112989017?spm=1001.2101.3001.6650.7&utm_medium=distribute.pc_relevant.none-task-blog-2%7Edefault%7EBlogCommendFromBaidu%7ERate-7.pc_relevant_paycolumn_v3&depth_1-utm_source=distribute.pc_relevant.none-task-blog-2%7Edefault%7EBlogCommendFromBaidu%7ERate-7.pc_relevant_paycolumn_v3&utm_relevant_index=9
