#include #include #include #include #include #include #include #include #include #include #include "macro.h" #include "dumpfile_io.h" #include "packet_priv.h" #include "lock_free_queue.h" #define MAX_PACKET_QUEUE_SIZE (4096 * 1000) struct dumpfile_io { uint16_t nr_threads; char directory[256]; pcap_t *pcap; struct lock_free_queue *queue[MAX_THREAD_NUM]; struct io_stat stat[MAX_THREAD_NUM]; uint64_t io_thread_need_exit; uint64_t io_thread_is_runing; }; struct pcap_pkt { char *data; int len; }; /****************************************************************************** * Private API ******************************************************************************/ typedef int file_handle(const char *file, void *arg); static int scan_directory(const char *dir, file_handle *handler, void *arg) { struct stat statbuf; struct dirent *entry; DIR *dp = opendir(dir); if (NULL == dp) { PACKET_IO_LOG_ERROR("opendir %s failed, %s", dir, strerror(errno)); return -1; } if (chdir(dir) == -1) { PACKET_IO_LOG_ERROR("chdir %s failed, %s", dir, strerror(errno)); goto error_out; } while ((entry = readdir(dp))) { if (lstat(entry->d_name, &statbuf) == -1) { PACKET_IO_LOG_ERROR("lstat %s failed, %s", entry->d_name, strerror(errno)); goto error_out; } if (S_IFDIR & statbuf.st_mode) { if (!strcmp(".", entry->d_name) || !strcmp("..", entry->d_name)) { continue; } if (scan_directory(entry->d_name, handler, arg) == -1) { goto error_out; } } else { if (handler(entry->d_name, arg) == -1) { goto error_out; } } } if (chdir("..") == -1) { PACKET_IO_LOG_ERROR("chdir .. failed, %s", strerror(errno)); goto error_out; } closedir(dp); return 0; error_out: closedir(dp); return -1; } static void pcap_packet_handler(u_char *user, const struct pcap_pkthdr *h, const u_char *bytes) { struct dumpfile_io *handle = (struct dumpfile_io *)user; // copy packet data to new memory struct pcap_pkt *pcap_pkt = (struct pcap_pkt *)calloc(1, sizeof(struct pcap_pkt) + h->caplen); if (pcap_pkt == NULL) { PACKET_IO_LOG_ERROR("unable to alloc packet"); return; } pcap_pkt->data = (char *)pcap_pkt + sizeof(struct pcap_pkt); pcap_pkt->len = h->caplen; memcpy((char *)pcap_pkt->data, bytes, h->caplen); // calculate packet hash struct packet pkt; packet_parse(&pkt, pcap_pkt->data, pcap_pkt->len); uint64_t hash = packet_get_hash(&pkt, LDBC_METHOD_HASH_INT_IP_AND_EXT_IP, 0); // push packet to queue struct lock_free_queue *queue = handle->queue[hash % handle->nr_threads]; while (lock_free_queue_push(queue, pcap_pkt) == -1) { if (ATOMIC_READ(&handle->io_thread_need_exit)) { free(pcap_pkt); PACKET_IO_LOG_STATE("dumpfile io thread need exit"); pcap_breakloop(handle->pcap); break; } usleep(1000); } if (ATOMIC_READ(&handle->io_thread_need_exit)) { PACKET_IO_LOG_STATE("dumpfile io thread need exit"); pcap_breakloop(handle->pcap); } } static int dumpfile_handler(const char *file, void *arg) { char resolved_path[256]; char pcap_errbuf[PCAP_ERRBUF_SIZE]; struct dumpfile_io *handle = (struct dumpfile_io *)arg; realpath(file, resolved_path); PACKET_IO_LOG_STATE("dumpfile %s in-processing", resolved_path) handle->pcap = pcap_open_offline(file, pcap_errbuf); if (handle->pcap == NULL) { PACKET_IO_LOG_ERROR("unable to open pcap file: %s, %s", resolved_path, pcap_errbuf); return -1; } pcap_loop(handle->pcap, -1, pcap_packet_handler, (u_char *)handle); pcap_close(handle->pcap); PACKET_IO_LOG_STATE("dumpfile %s processed", resolved_path) return 0; } static void *dumpfile_thread(void *arg) { struct dumpfile_io *handle = (struct dumpfile_io *)arg; ATOMIC_SET(&handle->io_thread_is_runing, 1); PACKET_IO_LOG_STATE("dumpfile io thread is running"); scan_directory(handle->directory, dumpfile_handler, arg); while (ATOMIC_READ(&handle->io_thread_need_exit) == 0) { PACKET_IO_LOG_STATE("dumpfile io thread waiting"); sleep(1); } PACKET_IO_LOG_STATE("dumpfile io thread exit !!!"); ATOMIC_SET(&handle->io_thread_is_runing, 0); return NULL; } /****************************************************************************** * Public API ******************************************************************************/ struct dumpfile_io *dumpfile_io_new(const char *directory, uint16_t nr_threads) { pthread_t tid; struct dumpfile_io *handle = (struct dumpfile_io *)calloc(1, sizeof(struct dumpfile_io)); if (handle == NULL) { PACKET_IO_LOG_ERROR("unable to allocate memory for dumpfile_io"); return NULL; } handle->nr_threads = nr_threads; strncpy(handle->directory, directory, MIN(strlen(directory), sizeof(handle->directory))); for (uint16_t i = 0; i < handle->nr_threads; i++) { handle->queue[i] = lock_free_queue_new(MAX_PACKET_QUEUE_SIZE); if (handle->queue[i] == NULL) { PACKET_IO_LOG_ERROR("unable to create packet queue"); goto error_out; } } if (pthread_create(&tid, NULL, dumpfile_thread, (void *)handle) != 0) { PACKET_IO_LOG_ERROR("unable to create packet io thread"); goto error_out; } return handle; error_out: dumpfile_io_free(handle); return NULL; } void dumpfile_io_free(struct dumpfile_io *handle) { if (handle) { ATOMIC_SET(&handle->io_thread_need_exit, 1); while (ATOMIC_READ(&handle->io_thread_is_runing)) { usleep(1000); } struct pcap_pkt *pcap_pkt = NULL; for (uint16_t i = 0; i < handle->nr_threads; i++) { while (1) { lock_free_queue_pop(handle->queue[i], (void **)&pcap_pkt); if (pcap_pkt) { free(pcap_pkt); } else { break; } } lock_free_queue_free(handle->queue[i]); } free(handle); handle = NULL; } } int dumpfile_io_init(struct dumpfile_io *handle, uint16_t thr_idx) { return 0; } int dumpfile_io_ingress(struct dumpfile_io *handle, uint16_t thr_idx, struct packet *pkts, int nr_pkts) { struct lock_free_queue *queue = handle->queue[thr_idx]; struct io_stat *stat = &handle->stat[thr_idx]; struct pcap_pkt *pcap_pkt = NULL; struct packet *pkt; int nr_parsed = 0; for (int i = 0; i < nr_pkts; i++) { lock_free_queue_pop(queue, (void **)&pcap_pkt); if (pcap_pkt == NULL) { break; } else { stat->dev_rx_pkts++; stat->dev_rx_bytes += pcap_pkt->len; stat->raw_rx_pkts++; stat->raw_rx_bytes += pcap_pkt->len; pkt = &pkts[nr_parsed]; memset(pkt, 0, sizeof(struct packet)); packet_parse(pkt, pcap_pkt->data, pcap_pkt->len); packet_set_origin_ctx(pkt, pcap_pkt); packet_set_origin(pkt, PACKET_ORIGIN_DUMPFILE); packet_set_action(pkt, PACKET_ACTION_FORWARD); nr_parsed++; } } return nr_parsed; } void dumpfile_io_egress(struct dumpfile_io *handle, uint16_t thr_idx, struct packet *pkts, int nr_pkts) { int len; struct packet *pkt = NULL; struct io_stat *stat = &handle->stat[thr_idx]; for (int i = 0; i < nr_pkts; i++) { pkt = &pkts[i]; len = packet_get_len(pkt); stat->dev_tx_pkts++; stat->dev_tx_bytes += len; stat->raw_tx_pkts++; stat->raw_tx_bytes += len; struct pcap_pkt *pcap_pkt = (struct pcap_pkt *)packet_get_origin_ctx(pkt); if (pcap_pkt) { free(pcap_pkt); } packet_free(pkt); } } void dumpfile_io_drop(struct dumpfile_io *handle, uint16_t thr_idx, struct packet *pkts, int nr_pkts) { struct packet *pkt = NULL; struct io_stat *stat = &handle->stat[thr_idx]; for (int i = 0; i < nr_pkts; i++) { pkt = &pkts[i]; struct pcap_pkt *pcap_pkt = (struct pcap_pkt *)packet_get_origin_ctx(pkt); if (pcap_pkt) { stat->drop_pkts++; stat->drop_bytes += packet_get_len(pkt); free(pcap_pkt); } packet_free(pkt); } } int dumpfile_io_inject(struct dumpfile_io *handle, uint16_t thr_idx, struct packet *pkts, int nr_pkts) { int len; struct packet *pkt = NULL; struct io_stat *stat = &handle->stat[thr_idx]; for (int i = 0; i < nr_pkts; i++) { pkt = &pkts[i]; len = packet_get_len(pkt); stat->inject_pkts++; stat->inject_bytes += len; stat->raw_tx_pkts++; stat->raw_tx_bytes += len; stat->dev_tx_pkts++; stat->dev_tx_bytes += len; packet_free(pkt); } return nr_pkts; } void dumpfile_io_yield(struct dumpfile_io *handle, uint16_t thr_idx, uint64_t timeout_ms) { return; } struct io_stat *dumpfile_io_stat(struct dumpfile_io *handle, uint16_t thr_idx) { return &handle->stat[thr_idx]; }