Блог

Редактирование формулировки проблемы, чтобы подробнее осветить основную логику

Мы наблюдаем проблемы с производительностью при ограничении скорости на основе назначения. Мы поддерживаем состояние для каждой пары {назначение-src} (максимум 100 назначений и 2 ^ 16 источников). У нас есть массив из 100 узлов, и на каждом узле у нас есть rte_hash *. Эта хэш-таблица будет поддерживать состояние каждого IP-адреса источника, видимого этим назначением. У нас есть сопоставление для каждого видимого назначения (от 0 до 100), и это используется для индексации в массив. Если конкретный источник за секунду превышает пороговое значение, определенное для этого назначения, мы блокируем источник, в противном случае мы разрешаем источник. Во время выполнения, когда мы видим только трафик для 2 или 3 назначений, проблем нет, но когда мы выходим за пределы 5, мы видим много падений. Наша функция должна выполнить поиск и определить поток, соответствующий dest_ip и src_ip. Обработайте поток и решите, нужно ли его удалять. Если поток не найден, добавьте его в хэш.

 struct flow_state {
    struct rte_hash* hash;    
};

struct flow_state flow_state_arr[100];
  

// я собираюсь создать эти хэш-таблицы, используя rte_hash_create в pipeline_init и освободить их во время pipeline_free.

Я описываю, что мы делаем в псевдокоде.

 run()
{
    1) do rx
    2) from the pkt, get index into the flow_state_arr and retrieve the rte_hash* handle    
    3) rte_hash_lookup_data(hash, src_ip,flow_data)
    4) if entry found, take decision on the flow (the decision is simply say rate limiting the flow)
    5) else rte_hash_add_data(hash,src_ip,new_flow_data) to add the flow to table and forward
}
  

Пожалуйста, укажите, можем ли мы использовать эти несколько объектов хэш-таблицы в пути к данным или каков наилучший способ, если нам нужно обрабатывать состояния для каждого назначения отдельно.

Редактировать
Спасибо, что ответили. Я буду рад поделиться фрагментами кода и нашими собранными результатами. У меня нет результатов сравнения для других версий DPDK, но ниже приведены некоторые результаты наших тестов с использованием 17.11.1.

Тестовая настройка
Я использую IXIA traffic gen (используя два канала 10G для генерации 12Mpps) для 3 назначений 14.143.156.x (в данном случае — 101,102,103). Трафик каждого назначения поступает из 2 ^ 16 разных источников. Это настройка генерации трафика.

Фрагмент кода

     struct flow_state_t {
        struct rte_hash* hash;
        uint32_t size;
        uint64_t threshold;
    };
    struct flow_data_t {
        uint8_t curr_state; // 0 if blocked, 1 if allowed
        uint64_t pps_count;
        uint64_t src_first_seen;
    };
    struct pipeline_ratelimit {
        struct pipeline p;
        struct pipeline_ratelimit_params params;
        rte_table_hash_op_hash f_hash;
        uint32_t swap_field0_offset[SWAP_DIM];
        uint32_t swap_field1_offset[SWAP_DIM];
        uint64_t swap_field_mask[SWAP_DIM];
        uint32_t swap_n_fields;
        pipeline_msg_req_handler custom_handlers[2]; // handlers for add and del
        struct flow_state_t flow_state_arr[100];
        struct flow_data_t flows[100][65536];
    } __rte_cache_aligned;
    
    /*
      add_handler(pipeline,msg) -- msg includes index and threshold
      In the add handler
      a rule/ threshold is added for a destination
      rte_hash_create and store rte_hash* in flow_state_arr[index]
      max of 100 destinations or rules are allowed
      previous pipelines add the ID (index) to the packet to look in to the
      flow_state_arr for the rule
    */
    
    /*
      del_handler(pipeline,msg) -- msg includes index
      In the del handler
      a rule/ threshold @index is deleted
      the associated rte_hash* is also freed
      the slot is made free
    */
    
    #define ALLOWED 1
    #define BLOCKED 0
    #define TABLE_MAX_CAPACITY 65536
    int do_rate_limit(struct pipeline_ratelimit* ps, uint32_t id, unsigned char* pkt)
    {
        uint64_t curr_time_stamp = rte_get_timer_cycles();
        struct iphdr* iph = (struct iphdr*)pkt;
        uint32_t src_ip = rte_be_to_cpu_32(iph->saddr);
    
        struct flow_state_t* node = amp;ps->flow_state_arr[id];
        struct flow_data_t* flow = NULL
        rte_hash_lookup_data(node->hash, amp;src_ip, (void**)amp;flow);
        if (flow != NULL)
        {
            if (flow->curr_state == ALLOWED)
            {
                if (flow->pps_count   > node->threshold)
                {
                    uint64_t seconds_elapsed = (curr_time_stamp - flow->src_first_seen) / CYCLES_IN_1SEC;
                    if (seconds_elapsed)
                    {
                        flow->src_first_seen  = seconds_elapsed * CYCLES_IN_1_SEC;
                        flow->pps_count = 1;
                        return ALLOWED;
                    }
                    else
                    {
                        flow->pps_count = 0;
                        flow->curr_state = BLOCKED;
                        return BLOCKED;
                    }
                }
                return ALLOWED;
            }
            else
            {
                uint64_t seconds_elapsed = (curr_time_stamp - flow->src_first_seen) / CYCLES_IN_1SEC;
                if (seconds_elapsed > 120)
                {
                    flow->curr_state = ALLOWED;
                    flow->pps_count = 0;
                    flow->src_first_seen  = seconds_elapsed * CYCLES_IN_1_SEC;
                    return ALLOWED;
                }
                return BLOCKED;
            }
        }
        int index = node->size;
        // If entry not found and we have reached capacity
        // Remove the rear element and mark it as the index for the new node    
        if (node->size == TABLE_MAX_CAPACITY)
        {
            rte_hash_reset(node->hash);
            index = node->size = 0;
        }
    
        // Add new element @packet_flows[mit_id][index]
        struct flow_data_t* flow_data = amp;ps->flows[id][index]; 
        *flow_data = { ALLOWED, 1, curr_time_stamp };
        node->size  ;
    
        // Add the new key to hash
        rte_hash_add_key_data(node->hash, (void*)amp;src_ip, (void*)flow_data);    
        return ALLOWED;
    }
    static int pipeline_ratelimit_run(void* pipeline)
    {
        struct pipeline_ratelimit* ps = (struct pipeline_ratelimit*)pipeline;
    
        struct rte_port_in* port_in = p->port_in_next;
        struct rte_port_out* port_out = amp;p->ports_out[0];
        struct rte_port_out* port_drop = amp;p->ports_out[2];
    
        uint8_t valid_pkt_cnt = 0, invalid_pkt_cnt = 0;
        struct rte_mbuf* valid_pkts[RTE_PORT_IN_BURST_SIZE_MAX];
        struct rte_mbuf* invalid_pkts[RTE_PORT_IN_BURST_SIZE_MAX];
    
        memset(valid_pkts, 0, sizeof(valid_pkts));
        memset(invalid_pkts, 0, sizeof(invalid_pkts));
    
        uint64_t n_pkts;
    
        if (unlikely(port_in == NULL)) {
            return 0;
        }
    
        /* Input port RX */
        n_pkts = port_in->ops.f_rx(port_in->h_port, p->pkts,
            port_in->burst_size);
    
        if (n_pkts == 0)
        {
            p->port_in_next = port_in->next;
            return 0;
        }
    
        uint32_t rc = 0;
        char* rx_pkt = NULL;
    
        for (j = 0; j < n_pkts; j  ) {
    
            struct rte_mbuf* m = p->pkts[j];
            rx_pkt = rte_pktmbuf_mtod(m, char*);
            uint32_t id = rte_be_to_cpu_32(*(uint32_t*)(rx_pkt - sizeof(uint32_t)));
            unsigned short packet_len = rte_be_to_cpu_16(*((unsigned short*)(rx_pkt   16)));
    
            struct flow_state_t* node = amp;(ps->flow_state_arr[id]);
    
            if (node->hash amp;amp; node->threshold != 0)
            {
                // Decide whether to allow of drop the packet
                // returns allow - 1, drop - 0
                if (do_rate_limit(ps, id, (unsigned char*)(rx_pkt   14)))
                    valid_pkts[valid_pkt_count  ] = m;
                else
                    invalid_pkts[invalid_pkt_count  ] = m;
            }
            else
                valid_pkts[valid_pkt_count  ] = m;
    
            if (invalid_pkt_cnt) {
                p->pkts_mask = 0;
                rte_memcpy(p->pkts, invalid_pkts, sizeof(invalid_pkts));
                p->pkts_mask = RTE_LEN2MASK(invalid_pkt_cnt, uint64_t);
                rte_pipeline_action_handler_port_bulk_mod(p, p->pkts_mask, port_drop);
            }
    
            p->pkts_mask = 0;
            memset(p->pkts, 0, sizeof(p->pkts));
    
            if (valid_pkt_cnt != 0)
            {
                rte_memcpy(p->pkts, valid_pkts, sizeof(valid_pkts));
                p->pkts_mask = RTE_LEN2MASK(valid_pkt_cnt, uint64_t);
            }
    
            rte_pipeline_action_handler_port_bulk_mod(p, p->pkts_mask, port_out);
    
            /* Pick candidate for next port IN to serve */
            p->port_in_next = port_in->next;
            return (int)n_pkts;
        }
}
  

Результаты

1. При генерации трафика только для одного назначения из 60000 источников с порогом 14Mpps, не было никаких падений. Мы смогли отправить 12Mpps из IXIA и recv 12Mpps
2. Снижение наблюдалось после добавления 3 или более назначений (каждое настроено на получение трафика из 60000 источников). Пропускная способность составляла всего 8-9 Mpps. При отправке для 100 назначений (60000 src каждый) было обработано всего 6,4 Mpps. было замечено падение на 50%.
3. При запуске через vtune-profiler он сообщил, что rte_hash_lookup_data является точкой доступа и в основном привязан к памяти (привязан к DRAM). Я скоро прикреплю отчет vtune.

Основываясь на обновлении, полученном в результате внутреннего тестирования, rte_hash библиотека не вызывает снижения производительности. Следовательно, как предложено в комментарии, это, скорее всего, связано с текущим шаблоном и дизайном алгоритма, которые могут приводить к промахам кэша и меньшему количеству инструкций за цикл.

Чтобы определить, является ли это остановкой интерфейса, остановкой конвейера серверной части или остановкой памяти, пожалуйста, используйте perf или vtune . Также постарайтесь минимизировать ветвление и использовать больше likely и prefetch тоже.