diff --git a/src/main.cpp b/src/main.cpp index 13cf682..41b26e0 100644 --- a/src/main.cpp +++ b/src/main.cpp @@ -31,7 +31,8 @@ struct worker_thread_ctx { pthread_t tid; - struct packet_io_device *dev; + size_t dev_num; + struct packet_io_device **pdev; struct session_manager *session_mgr; struct plugin_manager *plugin_mgr; int thread_id; @@ -39,57 +40,76 @@ struct worker_thread_ctx void *worker_thread_cycle(void *arg) { - struct stellar_packet *rx_pkt; + //uint64_t counter = 0; + //size_t data_len = 0; + //pid_t tid = syscall(SYS_gettid); + struct stellar_packet *rx_pkt[64]; struct stellar_session *session; struct worker_thread_ctx *thread_arg = (struct worker_thread_ctx *)arg; - while (1) - { - if (packet_io_device_rx(thread_arg->dev, thread_arg->thread_id, &rx_pkt, 1) > 0) - { - session = session_manager_commit(thread_arg->session_mgr, rx_pkt, thread_arg->thread_id); - while (session) - { - plugin_manager_dispatch(thread_arg->plugin_mgr, session, thread_arg->thread_id); - session = session_manager_fetch_session(thread_arg->session_mgr, session, thread_arg->thread_id); + while (1) { + for (size_t i = 0; i < thread_arg->dev_num; i++) { + ssize_t fetch_num = packet_io_device_rx(thread_arg->pdev[i], thread_arg->thread_id, rx_pkt, 5); + for (ssize_t j = 0; j < fetch_num; j++) { +/* + char *buf_addr = get_stellar_packet_data(rx_pkt[j], &data_len); + struct ethhdr *eth_hdr = (struct ethhdr *)buf_addr; + uint16_t eth_type = ntohs(eth_hdr->h_proto); + if (eth_type == ETHERNET_TYPE_IP) { + struct ip *ip_hdr = (struct ip *)(eth_hdr + 1); + char sip_str[64] = {0}; + char dip_str[64] = {0}; + strncpy_safe(sip_str, inet_ntoa(ip_hdr->ip_src), sizeof(sip_str)); + strncpy_safe(dip_str, inet_ntoa(ip_hdr->ip_dst), sizeof(dip_str)); + + //printf("sip: %s dip: %s, ip_len:%u\n", sip_str, dip_str, ntohs(ip_hdr->ip_len)); + } else if (eth_type == ETHERNET_TYPE_IPV6) { + } + } + counter += fetch_num; + printf("tid: %d counter:%lu\n", tid, counter); + if (fetch_num == 0) { + sleep(1); + continue; + } + + packet_io_pkts_free(thread_arg->pdev[i], thread_arg->tid, rx_pkt, fetch_num); +*/ + session = session_manager_commit(thread_arg->session_mgr, rx_pkt[j], thread_arg->thread_id); + while(session) { + plugin_manager_dispatch(thread_arg->plugin_mgr, session, thread_arg->thread_id); + session = session_manager_fetch_session(thread_arg->session_mgr, session, thread_arg->thread_id); + } + + //clean session_manager event queue + packet_io_device_tx(thread_arg->pdev[i], thread_arg->thread_id, &rx_pkt[j], 1); + } + + if (fetch_num == 0) { + printf("no fetch num\n"); + sleep(1); + //dispatch to time event + //dispatch to trigger polling event } - // clean session_manager event queue - packet_io_device_tx(thread_arg->dev, thread_arg->thread_id, &rx_pkt, 1); - } - else - { - printf("no fetch num\n"); - // dispatch to time event - // dispatch to trigger polling event - } -#endif } } + return nullptr; } -struct packet_io_device *packet_io_init(const char *instance_name, const enum packet_io_run_mode mode, const int thread_num) -{ - struct packet_io_instance *ppio_inst = packet_io_instance_create(instance_name, mode); - if (nullptr == ppio_inst) - { - log_error(ST_ERR_PIO_INSTANCE, "packet_io instance init failed."); - return nullptr; - } - - struct packet_io_device *ppio_dev = packet_io_device_open(ppio_inst, "eth1", thread_num, thread_num); - if (nullptr == ppio_dev) - { - log_error(ST_ERR_PIO_DEVICE, "packet_io device open failed."); - } - return ppio_dev; -} - int main(int argc, char **argv) { - int thread_num = 1; + int thread_num = 2; char file_path[] = "./plugs/plugins.inf"; - struct packet_io_device *dev = packet_io_init("stellar", PACKET_IO_RUN_MODE_PCAP_LIVE, thread_num); + char packet_io_conf[PATH_MAX] = "./conf/packet_io/packet_io.toml"; + struct packet_io_device *devices[64] = {0}; + size_t nr_devices = 0; + struct packet_io_instance *pinst = packet_io_init("stellar", packet_io_conf, devices, &nr_devices); + if (nullptr == pinst) { + log_error(ST_ERR_PIO_INSTANCE, "packet_io init failed."); + return -1; + } + struct session_manager *session_mgr = session_manager_create(thread_num); struct plugin_manager *plugin_mgr = plugin_manager_create(thread_num); @@ -100,7 +120,8 @@ int main(int argc, char **argv) for (int i = 0; i < thread_num; i++) { - workers[i].dev = dev; + workers[i].pdev = devices; + workers[i].dev_num = nr_devices; workers[i].session_mgr = session_mgr; workers[i].plugin_mgr = plugin_mgr; workers[i].thread_id = i;