一次奇怪的现象...

在上周的测试过程中，发现随着程序运行时间的增长[速度不快]，每个port的数据包miss高达千万级别的，也就是能处理就几十万的数据包，由处理千万到几十万，这个现象特别奇怪；然后多次运行一段时间也是如此，就记着这个问题待解决，然后又通过不同的源[速度非常快]一会造成了上述现象，猜测是不是由同一个原因引起的。

虽然业务代码比较复杂，从收队列中取数据包，处理，再决定是转发还是丢弃，input到发队列中和丢弃也是有可能的。测试了纯转发和丢弃数据包并不费时，剩下的就是处理了。仔细分析处理部分代码，在可疑的地方加上了cpu转数，然后分析后发现求hash计算只消耗几百到几千的转数，然后进行hash查找的时候也就几百几千，但在有些情况下达到一两万左右，可能是因为在key-data插入过程中有冲突，所以在查找时比较次数有点多和key的长度有关。然后在申请资源前直接返回了，发现这个问题不存在了，于是猜测是申请资源的时候造成竞争导致的。通过perf工具也能发现问题所在，但是为啥对于变化比较慢的不同源一开始没有出现那种情况，而且是越往后才出现的。

后来分析相关申请资源代码后：

 68 void *
 69 rte_malloc_socket(const char *type, size_t size, unsigned align, int socket_arg)
 70 {
 71     struct rte_mem_config *mcfg = rte_eal_get_configuration()->mem_config;
 72     int socket, i;
 73     void *ret;
 74 
 75     /* return NULL if size is 0 or alignment is not power-of-2 */
 76     if (size == 0 || (align && !rte_is_power_of_2(align)))
 77         return NULL;
 78 
 79     if (!rte_eal_has_hugepages())
 80         socket_arg = SOCKET_ID_ANY;
 81 
 82     if (socket_arg == SOCKET_ID_ANY)
 83         socket = malloc_get_numa_socket();
 84     else
 85         socket = socket_arg;
 86 
 87     /* Check socket parameter */
 88     if (socket >= RTE_MAX_NUMA_NODES)
 89         return NULL;
 90 
 91     ret = malloc_heap_alloc(&mcfg->malloc_heaps[socket], type,
 92                 size, 0, align == 0 ? 1 : align, 0);
 93     if (ret != NULL || socket_arg != SOCKET_ID_ANY)
 94         return ret;
 95 
 96     /* try other heaps */
 97     for (i = 0; i < RTE_MAX_NUMA_NODES; i++) {
 98         /* we already tried this one */
 99         if (i == socket)
100             continue;
101 
102         ret = malloc_heap_alloc(&mcfg->malloc_heaps[i], type,
103                     size, 0, align == 0 ? 1 : align, 0);
104         if (ret != NULL)
105             return ret;
106     }
107 
108     return NULL;
109 }

148 /*  
149  * Main function to allocate a block of memory from the heap.
150  * It locks the free list, scans it, and adds a new memseg if the
151  * scan fails. Once the new memseg is added, it re-scans and should return
152  * the new element after releasing the lock.
153  */             
154 void * 
155 malloc_heap_alloc(struct malloc_heap *heap,
156         const char *type __attribute__((unused)), size_t size, unsigned flags,
157         size_t align, size_t bound)
158 {   
159     struct malloc_elem *elem;
160         
161     size = RTE_CACHE_LINE_ROUNDUP(size);
162     align = RTE_CACHE_LINE_ROUNDUP(align);
163         
164     rte_spinlock_lock(&heap->lock);
165         
166     elem = find_suitable_element(heap, size, flags, align, bound);
167     if (elem != NULL) {
168         elem = malloc_elem_alloc(elem, size, align, bound);
169         /* increase heap's count of allocated elements */
170         heap->alloc_count++;
171     }
172     rte_spinlock_unlock(&heap->lock);
173 
174     return elem == NULL ? NULL : (void *)(&elem[1]);
175 }

如上，因为内存资源是加了自旋锁rte_spinlock_lock，故多个线程在申请的时候进行互斥，当时知道这里是这么实现的，一开始跑的时候并不会出现那种情况，是在运行一段时间后出现的，这就有问题了。那为什么在以非常快的速度变化源的时候一下子就出现问题了呢？这个问题也好分析。

后来写了个资源管理，本身资源管理不会包括调用rte_malloc_socket和rte_free，里面使用到的一定的技巧。包括申请与释放，完全不会造成资源泄露和把竞争的概率降的基几没有，也在一些点面作了平衡[比如申请多少，什么时候释放]。后来用了多种不同的极端测试，以上问题都解决了，回归了正常。

由于在一秒内每个线程可能需要处理三四百万个数据包，为每个数据包分配一个资源，加上一定的超时时间，累计的可能达到八九千万，那对于超时处理，在这方面也作了优化，避免了不会超时的资源在其他tick处理。因为如果每次都去操作超时资源，那么工作线程就不用做其他事了，那样miss就又回到原来的情况了。

纵观整个逻辑，在工作线程处理过程中，基本没有线程间资源的竞争，尽量减少不必要的计算，把资源局部化，对上一次的结果[比较耗时的]作了缓存或索引，那么下一次在进行计算时直接把可能要进行下一步计算的结果给直接取出来了。