Refactor Packet I/O

src/packet_io/dumpfile_io.cpp

@@ -0,0 +1,252 @@
+#include <pcap/pcap.h>
+#include <pthread.h>
+#include <unistd.h>
+#include <string.h>
+#include <stdlib.h>
+#include <assert.h>
+
+#include "stellar.h"
+#include "file_scan.h"
+#include "packet_private.h"
+#include "lock_free_queue.h"
+#include "dumpfile_io.h"
+
+#define MAX_PACKET_QUEUE_SIZE (4096 * 1000)
+
+struct dumpfile_io
+{
+    uint8_t nr_threads;
+    char directory[256];
+
+    pcap_t *pcap;
+    struct lock_free_queue *queue[MAX_THREAD_NUM];
+    struct packet_stat stat;
+    uint64_t io_thread_need_exit;
+    uint64_t io_thread_is_runing;
+};
+
+struct pcap_pkt
+{
+    char *data;
+    int len;
+};
+
+/******************************************************************************
+ * Private API
+ ******************************************************************************/
+
+static void pcap_handle(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];
+    lock_free_queue_push(queue, pcap_pkt);
+
+    if (ATOMIC_READ(&handle->io_thread_need_exit))
+    {
+        PACKET_IO_LOG_STATE("dumpfile io thread need exit");
+        pcap_breakloop(handle->pcap);
+    }
+}
+
+static int dumpfile_handle(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_handle, (u_char *)handle);
+    pcap_close(handle->pcap);
+
+    PACKET_IO_LOG_STATE("dumpfile %s processed", resolved_path)
+
+    return 0;
+}
+
+static void *dumpfile_thread_cycle(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");
+
+    file_scan(handle->directory, dumpfile_handle, 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 is exiting");
+    ATOMIC_SET(&handle->io_thread_is_runing, 0);
+
+    return NULL;
+}
+
+/******************************************************************************
+ * Public API
+ ******************************************************************************/
+
+struct dumpfile_io *dumpfile_io_new(const char *directory, uint8_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, strlen(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_cycle, (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);
+        }
+
+        for (uint16_t i = 0; i < handle->nr_threads; i++)
+        {
+            lock_free_queue_free(handle->queue[i]);
+        }
+        free(handle);
+        handle = NULL;
+    }
+}
+
+struct packet_stat *dumpfile_io_stat(struct dumpfile_io *handle)
+{
+    return &handle->stat;
+}
+
+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 pcap_pkt *pcap_pkt = NULL;
+    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
+        {
+            ATOMIC_ADD(&handle->stat.dev_rx_pkts, 1);
+            ATOMIC_ADD(&handle->stat.dev_rx_bytes, pcap_pkt->len);
+
+            ATOMIC_ADD(&handle->stat.raw_rx_pkts, 1);
+            ATOMIC_ADD(&handle->stat.raw_rx_bytes, pcap_pkt->len);
+
+            struct packet *pkt = &pkts[nr_parsed++];
+            memset(pkt, 0, sizeof(struct packet));
+            packet_parse(pkt, pcap_pkt->data, pcap_pkt->len);
+            packet_set_io_ctx(pkt, pcap_pkt);
+            packet_set_type(pkt, PACKET_TYPE_DATA);
+            packet_set_action(pkt, PACKET_ACTION_FORWARD);
+        }
+    }
+
+    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;
+    for (int i = 0; i < nr_pkts; i++)
+    {
+        pkt = &pkts[i];
+        len = packet_get_len(pkt);
+
+        ATOMIC_ADD(&handle->stat.dev_tx_pkts, 1);
+        ATOMIC_ADD(&handle->stat.dev_tx_bytes, len);
+
+        ATOMIC_ADD(&handle->stat.raw_tx_pkts, 1);
+        ATOMIC_ADD(&handle->stat.raw_tx_bytes, len);
+
+        struct pcap_pkt *pcap_pkt = (struct pcap_pkt *)packet_get_io_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;
+    for (int i = 0; i < nr_pkts; i++)
+    {
+        pkt = &pkts[i];
+        struct pcap_pkt *pcap_pkt = (struct pcap_pkt *)packet_get_io_ctx(pkt);
+        if (pcap_pkt)
+        {
+            free(pcap_pkt);
+        }
+        packet_free(pkt);
+    }
+}