hamnet70/impl/test/layer2_over_udp/l2udptest.c

/*
 * SPDX-License-Identifier: GPL-3.0-or-later
 *
 * Copyright (C) 2024  Thomas Kolb
 */

#include <linux/if.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>

#include <arpa/inet.h>

#include <liquid/liquid.h>

#include <sys/poll.h>
#include <unistd.h>
#include <fcntl.h>
#include <poll.h>
#include <signal.h>
#include <errno.h>

#include "utils.h"

#define LOGGER_MODULE_NAME "main"
#include "logger.h"
#include "options.h"
#include "jsonlogger.h"
#include "debug_structs.h"

#include "layer2/layer2_tx.h"
#include "layer2/layer2_rx.h"

#include "layer2/tundev.h"

#include "config.h"

#define RESULT_CHECK(stmt) { \
	result_t res = stmt; \
	if(res != OK) { \
		LOG(LVL_FATAL, "Error %d in %s:%d!", res, __FILE__, __LINE__); \
		exit(1); \
	} \
}

#define BROADCAST_PORT 3737

static int  m_tunfd = -1;
static bool m_running = true;

static int m_bcast_sock = -1;

static double next_tx_switch_time = 0.0;

static rx_stats_t m_rx_stats;

static layer2_rx_t l2rx;
static layer2_tx_t l2tx;


static void signal_handler(int signal, siginfo_t *info, void *ctx)
{
	(void)signal;
	(void)info;
	(void)ctx;

	LOG(LVL_INFO, "\nGracefully shutting down on signal %d.", signal);

	m_running = false;
}

static void block_tx_for(unsigned offset_ms)
{
	next_tx_switch_time = get_hires_time() + (double)offset_ms * 0.001;
}


void handle_received_packet(uint8_t *packet_data, size_t packet_len)
{
	block_tx_for(TX_SWITCH_BACKOFF_END_OF_PACKET_MS);

	bool shall_ack;
	result_t result = layer2_rx_handle_packet(&l2rx, packet_data, packet_len, &shall_ack);
	switch(result) {
		case OK:
			layer2_tx_handle_ack(&l2tx, layer2_rx_get_last_acked_seq(&l2rx), shall_ack);
			m_rx_stats.successful_decodes++;
			break;

		case ERR_INTEGRITY:
			LOG(LVL_ERR, "Packet could not be decoded by Layer 2.");
			m_rx_stats.failed_decodes++;
			break;

		case ERR_SEQUENCE:
			LOG(LVL_ERR, "Packet not in the expected sequence.");
			break;

		default: // all other errors
			LOG(LVL_ERR, "layer2_rx_handle_packet() returned error code %u.", result);
			break;
	}
}


static result_t transmit(const uint8_t *data, size_t len)
{
	result_t result = OK;

	struct sockaddr_in bcast_addr = {0};

	bcast_addr.sin_family = AF_INET;
	bcast_addr.sin_port = htons(BROADCAST_PORT);
	bcast_addr.sin_addr.s_addr = INADDR_BROADCAST;

	int ret = sendto(m_bcast_sock, data, len, 0, (struct sockaddr*)&bcast_addr, sizeof(bcast_addr));
	if(ret < 0) {
		LOG(LVL_ERR, "sendto: %s", strerror(errno));
		exit(EXIT_FAILURE);
	}

	fprintf(stderr, "t");
	return result;
}


int main(int argc, char **argv)
{
	// initialize the console logger
	logger_init();

	if(!jsonlogger_init("jsonlog.fifo")) {
		LOG(LVL_FATAL, "Could not initialize JSON logger.");
		return EXIT_FAILURE;
	}

	bool on_air = true;

	srand((int)(get_hires_time() * 1e6));

	// ** Initialize **

	char devname[IFNAMSIZ] = "hamnet70";
	m_tunfd = tundev_open(devname);

	if(m_tunfd < 0) {
		return 1;
	}

	RESULT_CHECK(layer2_tx_init(&l2tx, m_tunfd));
	RESULT_CHECK(layer2_rx_init(&l2rx, m_tunfd));

	// ** Set up signal handling

	struct sigaction term_action = {0};
	term_action.sa_sigaction = signal_handler;

	if(sigaction(SIGTERM, &term_action, NULL) < 0) {
		LOG(LVL_ERR, "sigaction: %s", strerror(errno));
		exit(EXIT_FAILURE);
	}

	if(sigaction(SIGINT, &term_action, NULL) < 0) {
		LOG(LVL_ERR, "sigaction: %s", strerror(errno));
		exit(EXIT_FAILURE);
	}

	// ** Set up UDP socket

	m_bcast_sock = socket(AF_INET, SOCK_DGRAM, 0);
	if(m_bcast_sock < 0) {
		LOG(LVL_ERR, "socket: %s", strerror(errno));
		exit(EXIT_FAILURE);
	}

	int broadcastEnable=1;
	int ret = setsockopt(m_bcast_sock, SOL_SOCKET, SO_BROADCAST, &broadcastEnable, sizeof(broadcastEnable));
	if(ret < 0) {
		LOG(LVL_ERR, "setsockopt: %s", strerror(errno));
		exit(EXIT_FAILURE);
	}

	struct sockaddr_in bind_addr = {0};

	bind_addr.sin_family = AF_INET;
	bind_addr.sin_port = htons(BROADCAST_PORT);
	bind_addr.sin_addr.s_addr = INADDR_ANY;

	ret = bind(m_bcast_sock, (struct sockaddr*)&bind_addr, sizeof(bind_addr));
	if(ret < 0) {
		LOG(LVL_ERR, "bind: %s", strerror(errno));
		exit(EXIT_FAILURE);
	}

	// ** Process packets **

	struct pollfd pfd_bcast;
	memset(&pfd_bcast, 0, sizeof(pfd_bcast));

	pfd_bcast.fd = m_bcast_sock;
	pfd_bcast.events = POLLIN;

	unsigned rx_retries = 0;

	double old = get_hires_time();
	size_t total_bytes = 0;
	double next_stats_print_time = old + 0.5;

	double retransmit_time = 0.0;

	while(m_running) {
		double now = get_hires_time();

		if(retransmit_time != 0.0 && now >= retransmit_time) {
			LOG(LVL_INFO, "Retransmit triggered.");
			retransmit_time = 0.0;
			layer2_tx_restart(&l2tx);
		}

		// fill the TX queue from the TUN device
		RESULT_CHECK(layer2_tx_fill_packet_queue(&l2tx));

		if((now > next_tx_switch_time)) {
			if(layer2_tx_can_transmit(&l2tx)) {
				// there is a packet to be (re)transmitted.

				LOG(LVL_DEBUG, "Starting new burst.");

				size_t burst_len = 0;

				// add packets to the burst until only 50000 samples remain free in the SDR buffer
				while(true) {
					uint8_t packet_buf[2048];
					size_t packet_size;

					packet_size = layer2_tx_encode_next_packet(&l2tx,
							layer2_rx_get_next_expected_seq(&l2rx),
							packet_buf, sizeof(packet_buf));

					if(packet_size == 0) {
						// no more packets available
						LOG(LVL_DEBUG, "Ending burst due to empty packet queue.");
						break;
					}

					LOG(LVL_DEBUG, "Adding packet with %d bytes to burst.", packet_size);

					burst_len++;
					RESULT_CHECK(transmit(packet_buf, packet_size));
				}

				LOG(LVL_DEBUG, "Burst finished: %zd packets sent.", burst_len);

				if(!on_air) {
					LOG(LVL_INFO, "RX -> TX");
				}

				on_air = true;
			} else if(on_air) { // TX on, but no more bursts to send
				LOG(LVL_INFO, "TX -> RX");
				on_air = false;

				retransmit_time = get_hires_time() + 1.0 + 1.0 * rand() / RAND_MAX;

				block_tx_for(TX_SWITCH_BACKOFF_AFTER_RX_ON);
			}
		}

		if(!on_air) {
			// ** Receive signal **

			int ret = poll(&pfd_bcast, 1, 10);
			if(ret < 0) {
				LOG(LVL_ERR, "poll: %s", strerror(errno));
				break;
			}

			if(ret == 0) {
				continue;
			}

			uint8_t packetbuf[65536];
			ret = recv(m_bcast_sock, packetbuf, sizeof(packetbuf), 0);
			if(ret < 0) {
				LOG(LVL_ERR, "recv: %s", strerror(errno));
			}

			if(ret <= 0) {
				break;
			}

			total_bytes += ret;

			double new = get_hires_time();
			if(new >= next_stats_print_time) {
				double rate = total_bytes / (new - old);
				LOG(LVL_INFO, "\nEstimated rate: %.3f kB/s", rate / 1e3);
				LOG(LVL_INFO, "Receiver statistics:");
				LOG(LVL_INFO, "  Preambles found:    %8zd", m_rx_stats.preambles_found);
				LOG(LVL_INFO, "  Successful decodes: %8zd (%6.2f %%)",
						m_rx_stats.successful_decodes, m_rx_stats.successful_decodes * 100.0f / m_rx_stats.preambles_found);
				LOG(LVL_INFO, "  Header errors:      %8zd (%6.2f %%)",
						m_rx_stats.header_errors, m_rx_stats.header_errors * 100.0f / m_rx_stats.preambles_found);
				LOG(LVL_INFO, "  Failed decodes:     %8zd (%6.2f %%)",
						m_rx_stats.failed_decodes, m_rx_stats.failed_decodes * 100.0f / m_rx_stats.preambles_found);
				next_stats_print_time += 0.5;

				total_bytes = 0;
				old = new;
			}

			rx_retries = 0;

			fprintf(stderr, "r");
		}
	}

	// ** Cleanup **

	close(m_bcast_sock);

	layer2_tx_destroy(&l2tx);
	layer2_rx_destroy(&l2rx);

	jsonlogger_shutdown();

	LOG(LVL_INFO, "Done.");

	logger_shutdown();
}