dissertation-1-language-tests/cpp/main.cpp

#include <iostream>

#include <cstring>
#include <fcntl.h>

#include <linux/if.h>
#include <linux/if_tun.h>
#include <zconf.h>

#include <sys/ioctl.h>
#include <thread>
#include <csignal>
#include <queue>
#include <mutex>
#include <condition_variable>
#include <sstream>

struct Packet {
    size_t len;
    uint8_t* data;

    Packet(const uint8_t *input, size_t num_bytes) {
        len = num_bytes;

        data = new uint8_t[len];
        std::memcpy(data, input, len);
    };

    ~Packet() {
        delete[] data;
    }

    [[nodiscard]] std::string print() const {
        std::stringstream out;
        for (size_t i = 0; i < len; i++) {
            int temp = data[i];
            out << std::hex << temp << " ";
        }
        return out.str();
    }
};

template <class T> class ThreadSafeQueue {
    std::queue<T> _queue = std::queue<T>();

    std::mutex _mutex;
    std::condition_variable _cond;

public:
    ThreadSafeQueue() = default;

    void push(T item) {
        _mutex.lock();
        _queue.push(item);
        _mutex.unlock();

        _cond.notify_one();
    }

    T pop() {
        while (true) {
            std::unique_lock<std::mutex> unique(_mutex);
            _cond.wait(unique);

            if (!_queue.empty()) {
                T out = _queue.front();
                _queue.pop();
                return out;
            }
        }
    }
};

int tun_alloc(const char *dev, short flags) {
    struct ifreq ifr{};
    int fd, err;

    if( (fd = open("/dev/net/tun" , O_RDWR)) < 0 ) {
        perror("Opening /dev/net/tun");
        return fd;
    }

    memset(&ifr, 0, sizeof(ifr));

    ifr.ifr_flags = flags;

    strncpy(ifr.ifr_name, dev, IFNAMSIZ);

    if( (err = ioctl(fd, TUNSETIFF, (void *)&ifr)) < 0 ) {
        perror("ioctl(TUNSETIFF)");
        close(fd);
        return err;
    }

    return fd;
}

volatile sig_atomic_t stop = 0;

void interrupt_handler(int signum) {
    stop = 1;
}

std::mutex print_lock;

void consumer(const int index, ThreadSafeQueue<Packet*> *queue) {
    std::cout << "thread " << index << "starting" << std::endl;

    while (!stop) {
        Packet *p = queue->pop();

        print_lock.lock();
        std::cout << "thread " << index << " received a packet with content `" << p->print() << "`" << std::endl;
        print_lock.unlock();

        delete p;
    }
}

int main() {
    signal(SIGINT, interrupt_handler);

    int tun = tun_alloc("nc%d", IFF_TUN);

    auto queue = new ThreadSafeQueue<Packet*>();

    std::thread threads[10];
    for (int i = 0; i < 10; i++) {
        const int i_safe = i;
        threads[i] = std::thread ([i_safe, queue]() {
            consumer(i_safe, queue);
        });
    }

    std::thread reader([tun, queue]() {
        uint8_t buffer[1500];
        while (!stop) {
            int num_bytes = read(tun, &buffer, 1500);
            if (num_bytes != 0) {
                auto *packet = new Packet(buffer, num_bytes);
                queue->push(packet);
            } else {
                stop = 1;
            };
        }
    });

    while (!stop) pause();

    std::cout << "terminating..." << std::endl;
    std::terminate();
}