Remove data callbacks in connection and digipeater modules

Interval suggests a regular timing structure which is not intended in this
construct and therefore is misleading.

.forgejo/workflows/doc.yaml

@@ -3,7 +3,7 @@ jobs:
   build-doc:
     runs-on: docker
     container:
-      image: git.tkolb.de/amateurfunk/hamnet70/asciidoctor:1.7
+      image: git.tkolb.de/amateurfunk/hamnet70/asciidoctor:1.6
     steps:
       - uses: actions/checkout@v4
       - run: cd doc && make
@@ -16,7 +16,7 @@ jobs:
     runs-on: docker
     if: github.ref == 'refs/heads/main'
     container:
-      image: git.tkolb.de/amateurfunk/hamnet70/asciidoctor:1.7
+      image: git.tkolb.de/amateurfunk/hamnet70/asciidoctor:1.6
     steps:
       - run: mkdir ~/.ssh && echo "${{ secrets.SSH_KEY }}" > ~/.ssh/id_ed25519 && chmod 0600 ~/.ssh/id_ed25519 && echo "${{ secrets.SSH_KNOWN_HOST }}" > ~/.ssh/known_hosts
       - uses: actions/download-artifact@v3

doc/hamnet70.adoc

@@ -10,7 +10,7 @@ Thomas Kolb DL5TKL
 
 == Introduction
 
-https://git.tkolb.de/amateurfunk/hamnet70[Hamnet70] intends to provide a network system in the 70 cm amateur radio band that can transfer Internet Protocol (IP) packets at a high speed.
+Hamnet70 intends to provide a network system in the 70 cm amateur radio band that can transfer Internet Protocol (IP) packets at a high speed.
 The protocols defined here are inspired by the Packet Radio network and Bluetooth Low Energy.
 
 The protocols are designed primarily for centralized infrastructure as they were common in the Packet Radio network.
@@ -238,49 +238,6 @@ Due to the long possible bursts, this system can achieve high throughput if ther
 
 === Frame Definitions
 
-==== Data Frame
-
-Data Frames carry all higher-layer data in a connection.
-Each Data Frame transfers a single layer-3 packet.
-
-The layer 2 header of Data Frames is filled as follows:
-
-- Message Type: `000` (Data Frame)
-- TX Request: `1` if this is the last packet in the burst, `0` otherwise
-- Source Address: the transmitter’s HAM-64 address
-- Destination Address: the target station’s HAM-64 address
-- TX sequence number: as required by Go-Back-N
-- RX sequence number: as required by Go-Back-N
-
-To identify how the encoded packet should be handled, the layer 3 protocol is encoded in the first byte of the layer 2 payload.
-The full layer 2 payload therefore is composed as follows:
-
-- Layer 3 protocol ID (1 Byte)
-- Layer 3 packet data (variable length)
-
-So far, the following protocols are defined and supported:
-
-[cols="2,1,1", options="header"]
-.Layer 3 protocol identifiers. EtherType is given as reference.
-|===
-| Protocol | Hamnet70 ID | EtherType
-
-| IPv6
-| `0x00`
-| `0x86DD`
-
-| IPv4
-| `0x10`
-| `0x0800`
-
-| _undefined/auto_
-| `0xFF`
-| -
-
-3+|_All other values are reserved._
-
-|===
-
 ==== Empty Frame
 
 The Empty Frame does not contain any data and therefore only consists of the header and the CRC.
@@ -448,104 +405,15 @@ The following Configuration Types are defined:
 If the client receives an unknown Configuration Type the corresponding block shall be skipped.
 The remaining blocks shall be parsed as usual.
 
-===== Connection Reset
+=== Ideas
 
-A Connection Reset is set by the digipeater if it receives unexpected packets from stations that are not connected.
+To be defined:
 
-The layer 2 header is filled as follows:
-
-- Message Type: `001` (Connection Management)
-- TX Request: `1` if this is the last packet in the burst, `0` otherwise
-- Source Address: the digipeater’s HAM-64 address
-- Destination Address: the new client’s HAM-64 address
-- TX sequence number: reserved, always 0
-- RX sequence number: reserved, always 0
-
-The message contains exactly 1 data byte: `0x03` to indicate that this is a Connection Reset packet.
-
-A Connection Request is sent in a regular burst and can be in any position.
-
-This message means that the digipeater does not have any information about a connection with the addressed client.
-Therefore, when a client receives a Connection Reset, it shall drop its complete connection state and start a new connection procedure if desired.
-
-===== Disconnect Request
-
-A Disconnect Request is set by the digipeater when it wants to orderly shut down a client connection.
-It is the last packet that the digipeater sends in a connection.
-
-The layer 2 header is filled as follows:
-
-- Message Type: `001` (Connection Management)
-- TX Request: `1` if this is the last packet in the burst, `0` otherwise
-- Source Address: the digipeater’s HAM-64 address
-- Destination Address: the client’s HAM-64 address
-- TX sequence number: as required by Go-Back-N
-- RX sequence number: as required by Go-Back-N
-
-The message contains exactly 1 data byte: `0x04` to indicate that this is a Disconnect Request packet.
-
-A Disconnect Request is sent in a regular burst and can be in any position.
-The sequence numbers are sent and handled the same way as in regular connection packets.
-
-When a client receives a Disconnect Request, it may transmit the remainder of its current queue, but must not queue new data packets.
-After the last packet is transmitted, the client shall send a Disconnect packet, which confirms the end of the connection.
-
-The digipeater must keep the connection state in memory until either the Disconnect packet is received from the client or the connection times out.
-
-===== Disconnect
-
-A Disconnect packet is sent by the client to terminate the connection.
-It is the last packet in a connection and is not confirmed by the digipeater.
-
-The layer 2 header is filled as follows:
-
-- Message Type: `001` (Connection Management)
-- TX Request: `0`
-- Source Address: the client’s HAM-64 address
-- Destination Address: the digipeater’s HAM-64 address
-- TX sequence number: as required by Go-Back-N
-- RX sequence number: as required by Go-Back-N
-
-The message contains exactly 1 data byte: `0x05` to indicate that this is a Disconnect packet.
-
-A Disconnect is sent always as the final packet of a burst.
-The sequence numbers are sent and handled the same way as in regular connection packets.
-
-When the digipeater receives a disconnect packet in the regular packet flow (i.e. no previous packets are lost), it will immediately drop the connection state and not call this client again.
-Therefore, if the client wants to ensure that all previous packets are transmitted, it must wait until the digipeater confirms that by sending the corresponding RX sequence number before sending the Disconnect packet.
-
-==== Connectionless Frame
-
-Connectionless Frames are used to transfer packets between unconnected notes.
-They can be used to implement custom protocols (similar to APRS, which is implemented on top of AX.25).
-
-The layer 2 header of Connectionless Frames is filled as follows:
-
-- Message Type: `100` (Connectionless Frame)
-- TX Request: `1` if this is the last packet in the burst, `0` otherwise
-- Source Address: the transmitter’s HAM-64 address
-- Destination Address: the target station’s HAM-64 address
-- TX sequence number: user-defined
-- RX sequence number: user-defined
-
-The sequence numbers can be used in any way that is useful for the custom protocol.
-
-It is required that the first byte of each Connectionless Frame identify the protocol being used.
-Protocol numbers are centrally assigned and are listed below.
-Some protocol numbers are reserved for experimentation and development and can be self-assigned temporarily.
-
-[cols="1,3", options="header"]
-.Connectionless Frame protocol IDs
-|===
-|Protocol IDs
-|Description
-
-|`0x00 .. 0xF7`
-|_reserved_
-
-|`0xF8 .. 0xFF`
-|Available for experimentation and developmentfootnote:[If you are developing a new protocol, you can freely pick a number from this range. Please check which temporary IDs are already used around your location and pick a free one. When your protocol reaches a sufficiently stable state, please request an official ID assignment].
-|===
+- connection establishment procedure (request, response)
+- how do clients get an IP(v6) address? -> should be derived from the call sign
+- when are new clients allowed to connect? -> base station calls for any new stations in regular intervals
+- handling of packets from unknown clients that are not connection requests
+- signal quality handling
 
 === Message Sequence Charts
 

impl/setup_tundev.sh

@@ -13,7 +13,7 @@ HOSTID="$1"
 
 DEV=hamnet70
 
-sudo ip tuntap add dev $DEV mode tun user $(whoami)
+sudo ip tuntap add dev $DEV mode tun pi user $(whoami)
 sudo ip link set dev $DEV txqueuelen 16  # default is 500
 sudo ip link set dev $DEV up
 sudo ip address add dev $DEV fd73::$HOSTID/64

impl/src/layer2/connection.c

@@ -1,15 +1,26 @@
+#include "layer2/packet_structs.h"
 #include <string.h>
 #include <assert.h>
 
+#define LOGGER_MODULE_NAME "conn"
+#include "logger.h"
+
 #include "connection.h"
 
 #include "config.h"
 #include "layer2/ham64.h"
+#include "layer2/packet_queue.h"
 #include "results.h"
+#include "utils.h"
 
 #define SEQ_NR_MASK 0xF
 
-result_t connection_init(connection_ctx_t *ctx, const ham64_t *my_addr, const ham64_t *peer_addr)
+result_t connection_init(
+		connection_ctx_t *ctx,
+		const ham64_t *my_addr,
+		const ham64_t *peer_addr,
+		connection_event_callback_t event_cb,
+		void *user_ctx)
 {
 	ctx->last_acked_seq = 0;
 	ctx->next_expected_seq = 0;
@@ -21,6 +32,35 @@ result_t connection_init(connection_ctx_t *ctx, const ham64_t *my_addr, const ha
 	ctx->my_addr = *my_addr;
 	ctx->peer_addr = *peer_addr;
 
+	ctx->event_cb = event_cb;
+
+	uint64_t now = get_hires_time();
+	ctx->last_rx_time = now;
+	ctx->retransmit_time = 0;
+
+	ctx->user_context = user_ctx;
+
+	// calculate IPv6 address
+	struct in6_addr net_addr;
+	inet_pton(AF_INET6, IPV6_NET, &net_addr);
+
+	memset(ctx->peer_ipv6_addr.s6_addr, 0, sizeof(ctx->peer_ipv6_addr));
+	memcpy(ctx->peer_ipv6_addr.s6_addr, net_addr.s6_addr, 8); // copy the network part
+
+	// fill the host part from the peer’s ham64 address. The bytes are filled in
+	// reverse order to make the „readable“ IPv6 address as short as possible.
+	for(uint8_t i = 0; i < peer_addr->length; i++) {
+		ctx->peer_ipv6_addr.s6_addr[15 - 2*i] = (peer_addr->addr[i] >> 8) & 0xFF;
+		ctx->peer_ipv6_addr.s6_addr[14 - 2*i] = (peer_addr->addr[i] >> 0) & 0xFF;
+	}
+
+	// print the address for debugging
+	char ipv6_str[INET6_ADDRSTRLEN];
+	char ham64_str[HAM64_FMT_MAX_LEN];
+	ham64_format(peer_addr, ham64_str);
+	inet_ntop(AF_INET6, &ctx->peer_ipv6_addr, ipv6_str, sizeof(ipv6_str));
+	LOG(LVL_DEBUG, "IPv6 address assigned to %s is %s.", ham64_str, ipv6_str);
+
 	ctx->conn_state = CONN_STATE_INITIALIZED;
 
 	return OK;
@@ -38,7 +78,7 @@ void connection_destroy(connection_ctx_t *ctx)
 }
 
 
-result_t connection_handle_packet(connection_ctx_t *ctx, const uint8_t *buf, size_t buf_len)
+result_t connection_handle_packet(connection_ctx_t *ctx, const uint8_t *buf, size_t buf_len, layer2_data_packet_t *data_packet)
 {
 	// check the connection state
 	switch(ctx->conn_state) {
@@ -71,17 +111,9 @@ result_t connection_handle_packet(connection_ctx_t *ctx, const uint8_t *buf, siz
 	}
 
 	// check if the packet really should be handled by us
-	if(!ham64_is_equal(&header.src_addr, &ctx->peer_addr)) {
-		char fmt_src_addr[HAM64_FMT_MAX_LEN];
-		char fmt_peer_addr[HAM64_FMT_MAX_LEN];
-
-		ham64_format(&header.src_addr, fmt_src_addr);
-		ham64_format(&ctx->peer_addr, fmt_peer_addr);
-
-		LOG(LVL_ERR, "Packet has the wrong source address: got %s, expected %s",
-				fmt_src_addr, fmt_peer_addr);
-
-		return ERR_INVALID_ADDRESS;
+	if(ham64_is_equal(&ctx->my_addr, &header.src_addr)) {
+		LOG(LVL_DEBUG, "Packet is from ourselves. Ignored.");
+		return OK;
 	}
 
 	if(!ham64_is_equal(&header.dst_addr, &ctx->my_addr)) {
@@ -97,50 +129,96 @@ result_t connection_handle_packet(connection_ctx_t *ctx, const uint8_t *buf, siz
 		return ERR_INVALID_ADDRESS;
 	}
 
+	size_t header_size = layer2_get_encoded_header_size(&header);
+
+	const uint8_t *payload = buf + header_size;
+	size_t payload_len = packet_size - header_size;
+
+	return connection_handle_packet_prechecked(ctx, &header, payload, payload_len, data_packet);
+}
+
+result_t connection_handle_packet_prechecked(
+		connection_ctx_t *ctx,
+		const layer2_packet_header_t *header,
+		const uint8_t *payload, size_t payload_len,
+		layer2_data_packet_t *data_packet)
+{
+	data_packet->payload_type = L2_PAYLOAD_TYPE_INVALID;
+	data_packet->payload_len = 0;
+
+	// check the connection state
+	switch(ctx->conn_state) {
+		case CONN_STATE_UNINITIALIZED:
+		case CONN_STATE_INITIALIZED:
+		case CONN_STATE_CLOSED:
+			LOG(LVL_ERR, "Trying to pass packet to connection in state %u", ctx->conn_state);
+			return ERR_INVALID_STATE;
+
+		case CONN_STATE_CONNECTING:
+		case CONN_STATE_ESTABLISHED:
+			// in these states, packets can be handled
+			break;
+	}
+
+	// check if this packet is from our designated peer
+	if(!ham64_is_equal(&header->src_addr, &ctx->peer_addr)) {
+		char fmt_src_addr[HAM64_FMT_MAX_LEN];
+		char fmt_peer_addr[HAM64_FMT_MAX_LEN];
+
+		ham64_format(&header->src_addr, fmt_src_addr);
+		ham64_format(&ctx->peer_addr, fmt_peer_addr);
+
+		LOG(LVL_ERR, "Packet has the wrong source address: got %s, expected %s",
+				fmt_src_addr, fmt_peer_addr);
+
+		return ERR_INVALID_ADDRESS;
+	}
+
 	LOG(LVL_DEBUG, "Handling %s packet with rx_seq_nr %u, tx_seq_nr %u.",
-			layer2_msg_type_to_string(header.msg_type), header.rx_seq_nr, header.tx_seq_nr);
+			layer2_msg_type_to_string(header->msg_type), header->rx_seq_nr, header->tx_seq_nr);
 
-	ctx->last_acked_seq = header.rx_seq_nr;
+	ctx->last_acked_seq = header->rx_seq_nr;
 
 
-	switch(header.msg_type) {
+	switch(header->msg_type) {
 		case L2_MSG_TYPE_EMPTY:
 			LOG(LVL_DEBUG, "Empty packet: accepted ACK for %u.", ctx->last_acked_seq);
-			return OK; // do not ACK and call back
+			// handle the acknowledgement internally
+			connection_handle_ack(ctx, header->rx_seq_nr, false);
+			return OK; // do not ACK
 
 		case L2_MSG_TYPE_CONN_MGMT:
 		case L2_MSG_TYPE_CONNECTIONLESS:
-			LOG(LVL_WARN, "Message type %s is not implemented yet.", layer2_msg_type_to_string(header.msg_type));
+			LOG(LVL_WARN, "Message type %s is not implemented yet.", layer2_msg_type_to_string(header->msg_type));
 			return OK;
 
 		case L2_MSG_TYPE_DATA:
 			break;
 
 		default:
-			LOG(LVL_ERR, "Invalid message type %d.", header.msg_type);
+			LOG(LVL_ERR, "Invalid message type %d.", header->msg_type);
 			return ERR_INVALID_STATE;
 	}
 
-	if(ctx->next_expected_seq != header.tx_seq_nr) {
-		LOG(LVL_ERR, "Expected sequence number %u, received %u.", ctx->next_expected_seq, header.tx_seq_nr);
+	if(ctx->next_expected_seq != header->tx_seq_nr) {
+		LOG(LVL_ERR, "Expected sequence number %u, received %u.", ctx->next_expected_seq, header->tx_seq_nr);
 		return ERR_SEQUENCE;
 	}
 
 	ctx->next_expected_seq++;
 	ctx->next_expected_seq &= 0xF;
 
-	LOG(LVL_INFO, "Received ACK for seq_nr %u in packet seq_nr %u.", header.rx_seq_nr, header.tx_seq_nr);
+	ctx->last_rx_time = get_hires_time();
+
+	LOG(LVL_INFO, "Received ACK for seq_nr %u in packet seq_nr %u.", header->rx_seq_nr, header->tx_seq_nr);
 
 	// handle the acknowledgement internally
-	connection_handle_ack(ctx, header.rx_seq_nr);
+	connection_handle_ack(ctx, header->rx_seq_nr, true);
 
-	size_t header_size = layer2_get_encoded_header_size(&header);
-
-	// extract the payload and forward it to the tun device
-	const uint8_t *payload = buf + header_size;
-	size_t payload_len = packet_size - header_size;
-
-	ctx->data_cb(ctx, payload, payload_len);
+	// pass the decoded data back to the user
+	data_packet->payload_type = payload[0];
+	data_packet->payload      = payload + 1;
+	data_packet->payload_len  = payload_len - 1;
 
 	return OK;
 }
@@ -209,6 +287,12 @@ result_t connection_enqueue_packet(connection_ctx_t *ctx, uint8_t *buf, size_t b
 }
 
 
+bool connection_can_enqueue_packet(const connection_ctx_t *ctx)
+{
+	return packet_queue_get_free_space(&ctx->packet_queue) > 0;
+}
+
+
 result_t connection_add_empty_packet(connection_ctx_t *ctx, bool tx_request)
 {
 	// check the connection state
@@ -227,10 +311,8 @@ result_t connection_add_empty_packet(connection_ctx_t *ctx, bool tx_request)
 
 	layer2_packet_header_t header;
 
-	header.dst_addr.addr[0] = 0xFFFF;
-	header.dst_addr.length = 1;
-	header.src_addr.addr[0] = 0x0001;
-	header.src_addr.length = 1;
+	header.dst_addr = ctx->peer_addr;
+	header.src_addr = ctx->my_addr;
 	header.msg_type = L2_MSG_TYPE_EMPTY;
 	header.rx_seq_nr = 0; // will be filled in layer2_tx_encode_next_packet()
 	header.tx_seq_nr = 0; // not used in empty packets
@@ -244,7 +326,56 @@ result_t connection_add_empty_packet(connection_ctx_t *ctx, bool tx_request)
 }
 
 
-size_t connection_encode_next_packet(connection_ctx_t *ctx, uint8_t ack_seq_nr, uint8_t *buf, size_t buf_len)
+result_t connection_send_parameters(connection_ctx_t *ctx)
+{
+	// check the connection state
+	switch(ctx->conn_state) {
+		case CONN_STATE_UNINITIALIZED:
+		case CONN_STATE_CLOSED:
+		case CONN_STATE_CONNECTING:
+		case CONN_STATE_ESTABLISHED:
+			LOG(LVL_ERR, "Trying to send connection parameters in state %u", ctx->conn_state);
+			return ERR_INVALID_STATE;
+
+		case CONN_STATE_INITIALIZED:
+			// in these states, packets can be handled
+			break;
+	}
+
+	layer2_packet_header_t header;
+
+	header.dst_addr = ctx->peer_addr;
+	header.src_addr = ctx->my_addr;
+	header.msg_type = L2_MSG_TYPE_CONN_MGMT;
+	header.rx_seq_nr = 0; // will be filled in layer2_tx_encode_next_packet()
+	header.tx_seq_nr = ctx->next_seq_nr;
+	header.tx_request = 1;
+
+	size_t payload_len = 1;
+	uint8_t *payload = malloc(payload_len);
+	if(!payload) {
+		return ERR_NO_MEM;
+	}
+
+	payload[0] = CONN_MGMT_TYPE_CONNECTION_PARAMETERS;
+
+	// TODO: calculate IP addresses from client’s HAM64 address
+
+	if (!packet_queue_add(&ctx->packet_queue, &header, payload, payload_len)) {
+		return ERR_NO_MEM;
+	}
+
+	ctx->next_seq_nr++;
+	ctx->next_seq_nr &= SEQ_NR_MASK;
+
+	// connection is considered established after the connection parameters are sent.
+	ctx->conn_state = CONN_STATE_ESTABLISHED;
+
+	return OK;
+}
+
+
+size_t connection_encode_next_packet(connection_ctx_t *ctx, uint8_t *buf, size_t buf_len, bool *end_burst)
 {
 	// check the connection state
 	switch(ctx->conn_state) {
@@ -252,7 +383,7 @@ size_t connection_encode_next_packet(connection_ctx_t *ctx, uint8_t ack_seq_nr,
 		case CONN_STATE_INITIALIZED:
 		case CONN_STATE_CLOSED:
 			LOG(LVL_ERR, "Trying to encode packet in inactive state %u", ctx->conn_state);
-			return ERR_INVALID_STATE;
+			return 0;
 
 		case CONN_STATE_CONNECTING:
 		case CONN_STATE_ESTABLISHED:
@@ -267,32 +398,21 @@ size_t connection_encode_next_packet(connection_ctx_t *ctx, uint8_t ack_seq_nr,
 		return 0;
 	}
 
-	unsigned int crc_size = crc_sizeof_key(PAYLOAD_CRC_SCHEME);
-
-	assert(buf_len >= LAYER2_PACKET_HEADER_ENCODED_SIZE_MAX + crc_size + entry->data_len);
-
 	layer2_packet_header_t header = entry->header;
-	header.rx_seq_nr = ack_seq_nr;
+	header.rx_seq_nr = ctx->next_expected_seq;
 
 	// encode the header
 	LOG(LVL_DEBUG, "Encoding packet with rx_seq_nr %u, tx_seq_nr %u.", header.rx_seq_nr, header.tx_seq_nr);
 
-	size_t packet_size = layer2_encode_packet_header(&header, buf);
-
-	// add the payload data
-	if(entry->data) {
-		memcpy(buf + packet_size, entry->data, entry->data_len);
+	size_t packet_size = layer2_encode_packet(&header, entry->data, entry->data_len, buf, buf_len);
+	if(packet_size == 0) {
+		LOG(LVL_ERR, "Buffer too small for encoded packet!");
+		return 0;
 	}
 
-	packet_size += entry->data_len;
-
-	// calculate CRC of everything and append it to the packet
-	crc_append_key(PAYLOAD_CRC_SCHEME, buf, packet_size);
-
-	packet_size += crc_size;
-
 	ctx->next_packet_index++;
 
+	*end_burst = header.tx_request;
 	return packet_size;
 }
 
@@ -318,7 +438,7 @@ void connection_tx_clean_empty_packet(connection_ctx_t *ctx)
 }
 
 
-void connection_handle_ack(connection_ctx_t *ctx, uint8_t acked_seq)
+void connection_handle_ack(connection_ctx_t *ctx, uint8_t acked_seq, bool do_ack)
 {
 	// check the connection state
 	switch(ctx->conn_state) {
@@ -334,8 +454,6 @@ void connection_handle_ack(connection_ctx_t *ctx, uint8_t acked_seq)
 			break;
 	}
 
-	ctx->next_packet_index = 0;
-
 	size_t packets_to_remove = 0;
 	size_t packets_available = packet_queue_get_used_space(&ctx->packet_queue);
 
@@ -349,13 +467,23 @@ void connection_handle_ack(connection_ctx_t *ctx, uint8_t acked_seq)
 		packets_to_remove++;
 	}
 
-	packet_queue_delete(&ctx->packet_queue, packets_to_remove);
+	if(packets_to_remove != 0) {
+		packet_queue_delete(&ctx->packet_queue, packets_to_remove);
 
-	packets_available = packet_queue_get_used_space(&ctx->packet_queue);
+		// send the next requested packet (all previous ones were deleted above).
+		assert(ctx->next_packet_index >= packets_to_remove);
+		ctx->next_packet_index -= packets_to_remove;
 
-	LOG(LVL_DEBUG, "handling ack for seq_nr %u, removing %zu packets, %zu packets remaining.", acked_seq, packets_to_remove, packets_available);
+		packets_available = packet_queue_get_used_space(&ctx->packet_queue);
 
-	if(packets_available == 0) {
+		LOG(LVL_DEBUG, "handling ack for seq_nr %u, removed %zu packets, %zu packets remaining.", acked_seq, packets_to_remove, packets_available);
+
+		ctx->retransmit_time = get_hires_time() + HRTIME_MS(RETRANSMIT_TIMEOUT_MS);
+	} else {
+		LOG(LVL_DEBUG, "duplicate ACK for seq_nr %u", acked_seq);
+	}
+
+	if(do_ack && packets_available == 0) {
 		// no packets left in queue, but an acknowledgement must be
 		// transmitted. Add an empty packet to do that.
 		result_t res = connection_add_empty_packet(ctx, false);
@@ -373,3 +501,52 @@ bool connection_can_transmit(const connection_ctx_t *ctx)
 	return (packet_queue_get_used_space(&ctx->packet_queue) != 0)
 		&& (packet_queue_get(&ctx->packet_queue, ctx->next_packet_index) != NULL);
 }
+
+
+bool connection_is_closed(const connection_ctx_t *ctx)
+{
+	switch(ctx->conn_state) {
+		case CONN_STATE_UNINITIALIZED:
+		case CONN_STATE_INITIALIZED:
+		case CONN_STATE_CLOSED:
+			return true;
+
+		case CONN_STATE_CONNECTING:
+		case CONN_STATE_ESTABLISHED:
+			return false;
+	}
+}
+
+
+result_t connection_maintain(connection_ctx_t *ctx)
+{
+	uint64_t now = get_hires_time();
+
+	if(now > ctx->last_rx_time + HRTIME_MS(CONNECTION_TIMEOUT_MS)) {
+		LOG(LVL_INFO, "Connection timed out.");
+		ctx->conn_state = CONN_STATE_CLOSED;
+		ctx->event_cb(ctx, CONN_EVT_TIMEOUT, ctx->user_context);
+		return OK;
+	}
+
+	if(ctx->retransmit_time != 0 && now >= ctx->retransmit_time) {
+		LOG(LVL_INFO, "Retransmit triggered.");
+		ctx->retransmit_time = 0;
+		connection_restart_tx(ctx);
+		ctx->event_cb(ctx, CONN_EVT_RETRANSMIT, ctx->user_context);
+	}
+
+	return OK;
+}
+
+
+bool connection_has_ipv6_peer_address(connection_ctx_t *ctx, const uint8_t *address_to_check)
+{
+	for(size_t i = 0; i < 16; i++) {
+		if(address_to_check[i] != ctx->peer_ipv6_addr.s6_addr[i]) {
+			return false;
+		}
+	}
+
+	return true;
+}

impl/src/layer2/connection.h

@@ -14,6 +14,8 @@
 
 #include "packet_queue.h"
 
+#include <arpa/inet.h>
+
 struct connection_ctx_s;
 
 typedef enum {
@@ -24,17 +26,24 @@ typedef enum {
 	CONN_STATE_CLOSED         //!< Connection has been closed (gracefully or by timeout)
 } connection_state_t;
 
-/*!\brief Type for a callback function that is called when a data packet was received. */
-typedef void (*connection_data_callback_t)(struct connection_ctx_s *conn, const uint8_t *data, size_t len);
+typedef enum {
+	CONN_EVT_TIMEOUT,    //!< The connection timed out because no packets were received
+	CONN_EVT_RETRANSMIT, //!< Packet queue transmission is restarted
+} connection_evt_t;
+
+/*!\brief Type for a callback function that is called on various connection events. */
+typedef void (*connection_event_callback_t)(struct connection_ctx_s *conn, connection_evt_t evt, void *user_ctx);
 
 typedef struct connection_ctx_s {
 	connection_state_t conn_state; //!< State of the connection.
 
-	connection_data_callback_t data_cb; //!< Callback function for received data packets.
+	connection_event_callback_t event_cb; //!< Callback function for event signalling.
 
 	ham64_t my_addr;             //!< The local link layer address.
 	ham64_t peer_addr;           //!< The link layer address of the peer.
 
+	struct in6_addr peer_ipv6_addr;  //!< The peer’s IPv6 address (generated from LL address)
+
 	uint8_t last_acked_seq;      //!< Next sequence number expected by the peer (from last Ack).
 	uint8_t next_expected_seq;   //!< Next sequence number expected by us.
 
@@ -42,6 +51,11 @@ typedef struct connection_ctx_s {
 
 	size_t next_packet_index;    //!< Index in the packet queue of the next packet to transmit.
 	uint8_t next_seq_nr;         //!< Sequence number to tag the next transmitted packet with.
+
+	uint64_t retransmit_time;    //!< Time when a retransmit shall be triggered.
+	uint64_t last_rx_time;       //!< Time when a packet was last received and decoded.
+
+	void *user_context;          //!< An arbitrary pointer that can set by the user.
 } connection_ctx_t;
 
 
@@ -50,9 +64,16 @@ typedef struct connection_ctx_s {
  * \param ctx       The connection context to initialize.
  * \param my_addr   The local link layer address.
  * \param peer_addr The remote link layer address.
+ * \param event_cb  Callback for connection events.
+ * \param user_ctx  User context pointer (for arbitrary data).
  * \returns OK if everything worked or a fitting error code.
  */
-result_t connection_init(connection_ctx_t *ctx, const ham64_t *my_addr, const ham64_t *peer_addr);
+result_t connection_init(
+		connection_ctx_t *ctx,
+		const ham64_t *my_addr,
+		const ham64_t *peer_addr,
+		connection_event_callback_t event_cb,
+		void *user_ctx);
 
 /*!\brief Destroy the given layer 2 connection context.
  */
@@ -60,12 +81,37 @@ void connection_destroy(connection_ctx_t *ctx);
 
 /*!\brief Handle a received packet.
  *
- * \param ctx         The connection context.
- * \param buf         Pointer to the packet data.
- * \param buf_len     Length of the packet.
+ * \param[inout] ctx           The connection context.
+ * \param[in]    buf           Pointer to the packet data.
+ * \param[in]    buf_len       Length of the packet.
+ * \param[out]   data_packet   Structure will be filled with a received data packet.
  * \returns           A result code from the packet handling procedure.
  */
-result_t connection_handle_packet(connection_ctx_t *ctx, const uint8_t *buf, size_t buf_len);
+result_t connection_handle_packet(
+		connection_ctx_t *ctx,
+		const uint8_t *buf,
+		size_t buf_len,
+		layer2_data_packet_t *data_packet);
+
+/*!\brief Handle a received packet where the header has already been decoded.
+ *
+ * This function assumes that the following basic checks were already done:
+ * - CRC is correct
+ * - Header can be decoded
+ * - Destination address is the local address
+ *
+ * \param[inout] ctx           The connection context.
+ * \param[in]    header        Pointer to the decoded header structure.
+ * \param[in]    payload       Pointer to the payload data.
+ * \param[in]    payload_len   Length of the payload data.
+ * \param[out]   data_packet   Structure will be filled with a received data packet.
+ * \returns           A result code from the packet handling procedure.
+ */
+result_t connection_handle_packet_prechecked(
+		connection_ctx_t *ctx,
+		const layer2_packet_header_t *header,
+		const uint8_t *payload, size_t payload_len,
+		layer2_data_packet_t *data_packet);
 
 /*!\brief Return the sequence number expected next by our side.
  */
@@ -80,12 +126,25 @@ uint8_t connection_get_last_acked_seq(const connection_ctx_t *ctx);
  */
 result_t connection_enqueue_packet(connection_ctx_t *ctx, uint8_t *buf, size_t buf_len);
 
+/*!\brief Check if there is free space in the TX packet queue.
+ * \param ctx       The connection context.
+ */
+bool connection_can_enqueue_packet(const connection_ctx_t *ctx);
+
 /*!\brief Add an empty packet to ensure an acknowledgement is sent.
  * \param ctx         The connection context.
  * \param tx_request  Value of the TX Request field in the packet.
  */
 result_t connection_add_empty_packet(connection_ctx_t *ctx, bool tx_request);
 
+/*!\brief Send connection parameters.
+ *
+ * This packet accepts an incoming connection request.
+ *
+ * \param ctx         The connection context.
+ */
+result_t connection_send_parameters(connection_ctx_t *ctx);
+
 /*!\brief Encode the next packet for transmission.
  *
  * \note
@@ -94,12 +153,12 @@ result_t connection_add_empty_packet(connection_ctx_t *ctx, bool tx_request);
  * called to handle retransmits correctly.
  *
  * \param ctx         The connection context.
- * \param ack_seq_nr  The received sequence number to send as an acknowledgement.
  * \param buf         Where to write the encoded packet data.
  * \param buf_len     Space available in the buffer.
+ * \param end_burst   Output parameter that is set to true if this is the last packet in a burst.
  * \returns           The number of bytes written to buf or zero if no packet was available.
  */
-size_t connection_encode_next_packet(connection_ctx_t *ctx, uint8_t ack_seq_nr, uint8_t *buf, size_t buf_len);
+size_t connection_encode_next_packet(connection_ctx_t *ctx, uint8_t *buf, size_t buf_len, bool *end_burst);
 
 /*!\brief Restart the transmission from the beginning of the packet queue.
  */
@@ -117,10 +176,27 @@ void connection_tx_clean_empty_packet(connection_ctx_t *ctx);
  * \param acked_seq The acknowledged (= next expected) sequence number.
  * \param do_ack    Whether an empty packet shall be generated if the queue is empty.
  */
-void connection_handle_ack(connection_ctx_t *ctx, uint8_t acked_seq);
+void connection_handle_ack(connection_ctx_t *ctx, uint8_t acked_seq, bool do_ack);
 
 /*!\brief Check if there are packets queued for transmission.
  */
 bool connection_can_transmit(const connection_ctx_t *ctx);
 
+/*!\brief Check if this connection is closed.
+ */
+bool connection_is_closed(const connection_ctx_t *ctx);
+
+/*!\brief Handle internal maintenance tasks.
+ *
+ * This should be called periodically to handle timeouts and retransmissions.
+ */
+result_t connection_maintain(connection_ctx_t *ctx);
+
+/*!\brief Check if the given IPv6 peer address belongs to this connection.
+ *
+ * \param address_to_check   Pointer to the IPv6 address to check. Must point to 16 bytes of data.
+ * \returns True if this address is handled by this connection, false otherwise.
+ */
+bool connection_has_ipv6_peer_address(connection_ctx_t *ctx, const uint8_t *address_to_check);
+
 #endif // CONNECTION_H

impl/src/layer2/connection_list.c

@@ -0,0 +1,237 @@
+#include <string.h>
+#include <assert.h>
+
+#include "connection_list.h"
+
+#include "layer2/connection.h"
+#include "results.h"
+
+#define SEQ_NR_MASK 0xF
+
+static void destroy_entry(connection_list_entry_t *entry)
+{
+	connection_destroy(&entry->connection);
+	free(entry);
+}
+
+// find the location where the timestamp should be inserted to maintain
+// ascending timestamp order and return the then-previous entry.
+//
+// If NULL is returned, the list is either empty or the entry should be
+// inserted before the head. This must be checked externally.
+static connection_list_entry_t* find_prev_for_timestamp(connection_list_t *list, uint64_t timestamp)
+{
+	if(!list->head) {
+		return NULL;
+	}
+
+	if(timestamp < list->head->next_access_time) {
+		return NULL;
+	}
+
+	connection_list_entry_t *prev = list->head;
+	while(prev->next) {
+		if(prev->next_access_time < timestamp
+				&& prev->next->next_access_time >= timestamp) {
+			// location found!
+			break;
+		}
+
+		prev = prev->next;
+	}
+
+	// either the correct location was found or prev now points to the last entry
+	// in the list.
+	return prev;
+}
+
+result_t connection_list_init(connection_list_t *list)
+{
+	list->head = NULL;
+	return OK;
+}
+
+
+void connection_list_destroy(connection_list_t *list)
+{
+	// delete all list entries
+	while(list->head) {
+		connection_list_entry_t *next = list->head->next;
+		destroy_entry(list->head);
+		list->head = next;
+	}
+}
+
+
+result_t connection_list_insert(connection_list_t *list, const connection_ctx_t *conn, uint64_t next_access_time)
+{
+	connection_list_entry_t *new_entry = malloc(sizeof(connection_list_entry_t));
+	if(!new_entry) {
+		return ERR_NO_MEM;
+	}
+
+	new_entry->connection = *conn;
+	new_entry->next_access_time = next_access_time;
+
+	connection_list_entry_t *prev = find_prev_for_timestamp(list, next_access_time);
+
+	if(!prev) {
+		// the new entry should become the list’s head (also works if the list is empty!)
+		new_entry->next = list->head;
+		list->head = new_entry;
+	} else {
+		// insert after prev
+		new_entry->next = prev->next;
+		prev->next = new_entry;
+	}
+
+	return OK;
+}
+
+
+connection_list_entry_t* connection_list_get_head(connection_list_t *list)
+{
+	return list->head;
+}
+
+
+result_t connection_list_reschedule_head(connection_list_t *list, uint64_t next_access_timestamp)
+{
+	if(!list->head) {
+		return ERR_INVALID_STATE;
+	}
+
+	if(!list->head->next) {
+		// nothing to do because there is only one entry.
+		return OK;
+	}
+
+	if(next_access_timestamp < list->head->next->next_access_time) {
+		// the head does not need to be moved because the new timestamp is still
+		// smaller than that of the second entry => only update the timestamp.
+		list->head->next_access_time = next_access_timestamp;
+		return OK;
+	}
+
+	// detach the entry from the list
+	connection_list_entry_t *reloc_entry = list->head;
+	list->head = list->head->next;
+
+	// find the location where the entry should be reinserted
+	connection_list_entry_t *prev = find_prev_for_timestamp(list, next_access_timestamp);
+
+	if(!prev) {
+		// the relocated entry should become the list’s head (also works if the list is empty!)
+		reloc_entry->next = list->head;
+		list->head = reloc_entry;
+	} else {
+		// insert after prev
+		reloc_entry->next = prev->next;
+		prev->next = reloc_entry;
+	}
+
+	return OK;
+}
+
+
+result_t connection_list_delete_head(connection_list_t *list)
+{
+	if(!list->head) {
+		return ERR_INVALID_STATE;
+	}
+
+	connection_list_entry_t *new_head = list->head->next;
+
+	destroy_entry(list->head);
+	list->head = new_head;
+
+	return OK;
+}
+
+
+result_t connection_list_delete_closed(connection_list_t *list)
+{
+	if(!list->head) {
+		return OK;
+	}
+
+	connection_list_entry_t *prev = NULL;
+	connection_list_entry_t *cur = list->head;
+
+	while(cur) {
+		if(connection_is_closed(&cur->connection)) {
+			connection_list_entry_t *to_delete = cur;
+			cur = cur->next;
+
+			if(prev) {
+				prev->next = to_delete->next;
+			} else {
+				list->head = to_delete->next;
+			}
+
+			destroy_entry(to_delete);
+		} else {
+			prev = cur;
+			cur = cur->next;
+		}
+	}
+
+	return OK;
+}
+
+
+result_t connection_list_enqueue_packet(connection_list_t *list, uint8_t *data, size_t data_len)
+{
+	if(data_len < 40) {
+		// packet not large enough for an IPv6 header
+		LOG(LVL_DEBUG, "Packet size too small: %zu bytes given, 40 bytes needed.", data_len);
+		return ERR_INVALID_PARAM;
+	}
+
+	uint8_t version = data[0] >> 4;
+	if(version != 6) {
+		LOG(LVL_DEBUG, "IP version (%i) is not 6.", version);
+		return ERR_INVALID_PARAM;
+	}
+
+	uint8_t *dest_addr = data + 24;
+
+	// search the list for the destination address
+	connection_list_entry_t *ptr = list->head;
+
+	while(ptr) {
+		if(connection_has_ipv6_peer_address(&ptr->connection, dest_addr)) {
+			// found it!
+			break;
+		}
+
+		ptr = ptr->next;
+	}
+
+	if(!ptr) {
+		// address not found in any connection
+		return ERR_INVALID_ADDRESS;
+	}
+
+	return connection_enqueue_packet(&ptr->connection, data, data_len);
+}
+
+
+bool connection_list_can_enqueue_packet(connection_list_t *list)
+{
+	if(!list->head) {
+		return false; // no entries -> no free space
+	}
+
+	connection_list_entry_t *entry = list->head;
+
+	while(entry) {
+		if(!connection_can_enqueue_packet(&entry->connection)) {
+			return false;
+		}
+
+		entry = entry->next;
+	}
+
+	return true;
+}

impl/src/layer2/connection_list.h

@@ -0,0 +1,109 @@
+/*
+ * This file contains functions to manage a list of layer 2 connections for
+ * scheduling purposes.
+ *
+ * SPDX-License-Identifier: GPL-3.0-or-later
+ *
+ * Copyright (C) 2024  Thomas Kolb
+ */
+
+#ifndef CONNECTION_LIST_H
+#define CONNECTION_LIST_H
+
+#include <results.h>
+#include <stdbool.h>
+
+#include "connection.h"
+
+typedef struct connection_list_entry_s {
+	uint64_t next_access_time; //!< When to next activate this connection
+	connection_ctx_t connection; //!< The actual connection entry
+
+	struct connection_list_entry_s *next; //!< pointer to the next list element
+} connection_list_entry_t;
+
+typedef struct connection_list_s {
+	connection_list_entry_t *head; //!< pointer to the first list element
+} connection_list_t;
+
+
+/*!\brief Initialize a connection list.
+ *
+ * \param list      The list to initialize.
+ * \returns OK if everything worked or a fitting error code.
+ */
+result_t connection_list_init(connection_list_t *list);
+
+/*!\brief Destroy the given layer 2 connection list, deleting all internal data.
+ */
+void connection_list_destroy(connection_list_t *list);
+
+/*!\brief Insert a connection context into the list.
+ *
+ * The connection context will be copied internally. The original copy should
+ * no longer be used after the context was inserted into the list.
+ *
+ * \param list                Pointer to the list where the entry shall be inserted.
+ * \param conn                Pointer to the connection context.
+ * \param next_access_time    Timestamp when this connection should be activated.
+ * \returns OK if everything worked or a fitting error code.
+ */
+result_t connection_list_insert(connection_list_t *list, const connection_ctx_t *conn, uint64_t next_access_time);
+
+/*!\brief Get the head (first entry) of the list.
+ *
+ * \param list     Pointer to the list to access.
+ *
+ * \returns A pointer to the head entry or NULL if the list is empty.
+ */
+connection_list_entry_t* connection_list_get_head(connection_list_t *list);
+
+/*!\brief Re-schedule the current head entry.
+ *
+ * \param list                Pointer to the list to access.
+ * \param next_access_time    Timestamp when this connection should be activated again.
+ *
+ * \returns OK if everything worked or a fitting error code.
+ */
+result_t connection_list_reschedule_head(connection_list_t *list, uint64_t next_access_timestamp);
+
+/*!\brief Delete the current head entry.
+ *
+ * \param list                Pointer to the list to access.
+ *
+ * \returns OK if everything worked or a fitting error code.
+ */
+result_t connection_list_delete_head(connection_list_t *list);
+
+/*!\brief Delete all connections that are in a closed state.
+ *
+ * \param list                Pointer to the list to clean.
+ *
+ * \returns OK if everything worked or a fitting error code.
+ */
+result_t connection_list_delete_closed(connection_list_t *list);
+
+/*!\brief Insert a packet in the appropriate connection’s queue.
+ *
+ * The appropriate connection is selected by extracting the destination IP
+ * address from the packet and comparing it to the peer address of the
+ * connection.
+ *
+ * \param list         Pointer to the connection list to operate on.
+ * \param packet       Pointer to the raw packet data (no TUNTAP header!).
+ * \param packet_len   Length of the packet data.
+ *
+ * \retval OK if the packet was inserted.
+ * \retval ERR_NO_MEM if the target queue was full.
+ * \retval ERR_INVALID_ADDRESS if the destination address does not match any connection.
+ * \retval ERR_INVALID_PARAM if the packet format is not supported.
+ */
+result_t connection_list_enqueue_packet(connection_list_t *list, uint8_t *data, size_t data_len);
+
+/*!\brief Determine if a packet can be enqueued in all queues.
+ *
+ * \returns False if any queue is full, true otherwise.
+ */
+bool connection_list_can_enqueue_packet(connection_list_t *list);
+
+#endif // CONNECTION_LIST_H

impl/src/layer2/digipeater.c

@@ -0,0 +1,379 @@
+#include <poll.h>
+#include <string.h>
+#include <errno.h>
+#include <unistd.h>
+
+#define LOGGER_MODULE_NAME "digi"
+#include "logger.h"
+
+#include "digipeater.h"
+
+#include "config.h"
+#include "layer2/connection.h"
+#include "layer2/connection_list.h"
+#include "layer2/ham64.h"
+#include "layer2/packet_structs.h"
+#include "layer2/packet_queue.h"
+#include "results.h"
+#include "utils.h"
+
+void conn_event_cb(struct connection_ctx_s *conn, connection_evt_t evt, void *user_ctx)
+{
+	(void)conn;
+	digipeater_ctx_t *digi_ctx = user_ctx;
+
+	switch(evt) {
+		case CONN_EVT_TIMEOUT:
+			// connection has been closed by timeout -> clean up the list
+			connection_list_delete_closed(&digi_ctx->conn_list);
+			break;
+
+		default:
+			// do nothing
+			break;
+	}
+}
+
+static result_t digipeater_handle_beacon_responses(digipeater_ctx_t *ctx, const layer2_packet_header_t *header, const uint8_t *buf, size_t buf_len)
+{
+	LOG(LVL_DEBUG, "Handling beacon response packet.");
+	layer2_dump_packet_header(LVL_DUMP, header);
+
+	if(header->msg_type != L2_MSG_TYPE_CONN_MGMT) {
+		LOG(LVL_ERR, "Beacon response with invalid message type %i", header->msg_type);
+		return ERR_INVALID_PARAM;
+	}
+
+	if(buf_len == 0) {
+		LOG(LVL_ERR, "Missing payload in beacon response");
+		return ERR_INVALID_PARAM;
+	}
+
+	uint8_t conn_mgmt_type = buf[0];
+
+	if(conn_mgmt_type != CONN_MGMT_TYPE_CONNECTION_REQUEST) {
+		LOG(LVL_ERR, "Unexpected connection management type in beacon response: 0x%02x", conn_mgmt_type);
+		return ERR_INVALID_PARAM;
+	}
+
+	// packet is valid -> create a new connection, enqueue the parameters message
+	// and add it at the beginning of the connection list.
+
+	connection_ctx_t new_conn;
+	ERR_CHECK(connection_init(&new_conn, &ctx->my_addr, &header->src_addr, conn_event_cb, ctx));
+
+	ERR_CHECK(connection_send_parameters(&new_conn));
+
+	uint64_t now = get_hires_time();
+	ERR_CHECK(connection_list_insert(&ctx->conn_list, &new_conn, now));
+
+	return OK;
+}
+
+
+static size_t encode_beacon_packet(digipeater_ctx_t *ctx, uint8_t *buf, size_t buf_len)
+{
+	// build a beacon packet
+	layer2_packet_header_t header;
+
+	ham64_t broadcast = {{0xFFFF, 0, 0, 0}, 1};
+
+	header.dst_addr = broadcast;
+	header.src_addr = ctx->my_addr;
+	header.msg_type = L2_MSG_TYPE_CONN_MGMT;
+	header.rx_seq_nr = 0; // unused
+	header.tx_seq_nr = 0; // unused
+	header.tx_request = 1;
+
+	uint8_t payload[1] = {CONN_MGMT_TYPE_BEACON};
+	return layer2_encode_packet(&header, payload, 1, buf, buf_len);
+}
+
+
+result_t digipeater_init(digipeater_ctx_t *ctx, const ham64_t *my_addr)
+{
+	ctx->my_addr = *my_addr;
+
+	ctx->state = DIGIPEATER_STATE_CONN;
+
+	packet_queue_init(&ctx->oneshot_queue);
+
+	uint64_t now = get_hires_time();
+	ctx->next_beacon_time = now + HRTIME_MS(BEACON_INTERVAL_MS);
+
+	ctx->cycle_end_time = now;
+
+	return connection_list_init(&ctx->conn_list);
+}
+
+
+void digipeater_destroy(digipeater_ctx_t *ctx)
+{
+	connection_list_destroy(&ctx->conn_list);
+}
+
+
+result_t digipeater_handle_packet(
+		digipeater_ctx_t *ctx,
+		const uint8_t *buf,
+		size_t buf_len,
+		layer2_data_packet_t *data_packet)
+{
+	data_packet->payload_type = L2_PAYLOAD_TYPE_INVALID;
+	data_packet->payload_len = 0;
+
+	// check the CRC
+	size_t packet_size = buf_len - crc_sizeof_key(PAYLOAD_CRC_SCHEME);
+
+	if(!crc_check_key(PAYLOAD_CRC_SCHEME, (unsigned char*)buf, packet_size)) {
+		LOG(LVL_ERR, "CRC check failed!");
+		return ERR_INTEGRITY;
+	}
+
+	// decode the header
+	layer2_packet_header_t header;
+
+	if(!layer2_decode_packet_header(buf, buf_len, &header)) {
+		LOG(LVL_ERR, "Header could not be decoded!");
+		return ERR_INTEGRITY;
+	}
+
+	layer2_dump_packet_header(LVL_DUMP, &header);
+
+	// check if the packet really should be handled by us
+	if(ham64_is_equal(&ctx->my_addr, &header.src_addr)) {
+		LOG(LVL_DEBUG, "Packet is from ourselves. Ignored.");
+		return OK;
+	}
+
+	if(!ham64_is_equal(&header.dst_addr, &ctx->my_addr)) {
+		char fmt_dst_addr[HAM64_FMT_MAX_LEN];
+		char fmt_my_addr[HAM64_FMT_MAX_LEN];
+
+		ham64_format(&header.dst_addr, fmt_dst_addr);
+		ham64_format(&ctx->my_addr, fmt_my_addr);
+
+		LOG(LVL_ERR, "Packet has the wrong destination address: got %s, expected %s",
+				fmt_dst_addr, fmt_my_addr);
+
+		return ERR_INVALID_ADDRESS;
+	}
+
+	// FIXME: handle connection management packets here (or somewhere else):
+	// - Disconnect Request
+
+	size_t header_size = layer2_get_encoded_header_size(&header);
+
+	const uint8_t *payload = buf + header_size;
+	size_t payload_len = packet_size - header_size;
+
+	result_t result = OK;
+
+	switch(ctx->state) {
+		case DIGIPEATER_STATE_CONN:
+			{
+				connection_list_entry_t *head = connection_list_get_head(&ctx->conn_list);
+				if(head) {
+					connection_ctx_t *current_conn = &head->connection;
+					result = connection_handle_packet_prechecked(
+							current_conn, &header, payload, payload_len, data_packet);
+				} else {
+					LOG(LVL_WARN, "Digipeater in CONN state, but there is no active connection! Packet dropped.");
+					result = OK;
+				}
+			}
+			break;
+
+		case DIGIPEATER_STATE_BEACON:
+			result = digipeater_handle_beacon_responses(ctx, &header, payload, payload_len);
+	}
+
+	// end the current cycle if tx_request is set in an incoming packet
+	if(header.tx_request) {
+		LOG(LVL_INFO, "TX Request was received. Ending current cycle.");
+		digipeater_end_cycle(ctx);
+	}
+
+	return result;
+}
+
+
+result_t digipeater_fill_packet_queues_from_tundev(digipeater_ctx_t *ctx, int tun_fd)
+{
+	// first check if any queue is already full, so we don't have to drop packets
+	// from the TUN device queue.
+	if(!connection_list_can_enqueue_packet(&ctx->conn_list)) {
+		LOG(LVL_DEBUG, "No free space in queues.");
+		return OK; // do nothing
+	}
+
+	struct pollfd pfd_tun;
+	memset(&pfd_tun, 0, sizeof(pfd_tun));
+
+	pfd_tun.fd = tun_fd;
+	pfd_tun.events = POLLIN;
+
+	while(true) {
+		int ret = poll(&pfd_tun, 1, 0 /* timeout */);
+		if(ret < 0) {
+			LOG(LVL_ERR, "poll: %s", strerror(errno));
+			return ERR_SYSCALL;
+		} else if(ret == 0) {
+			// no more packets
+			break;
+		} else {
+			// a packet is available -> read it
+			static const size_t packetbuf_size = 2048;
+			uint8_t packetbuf[packetbuf_size];
+			ret = read(tun_fd, packetbuf, packetbuf_size);
+			if(ret < 0) {
+				LOG(LVL_ERR, "read: %s", strerror(errno));
+				return ERR_SYSCALL;
+			} else if(ret == 0) {
+				// no more data, should not happen
+				break;
+			}
+
+			uint16_t flags = *(uint16_t*)packetbuf;
+			uint16_t proto = *((uint16_t*)packetbuf + 1);
+			LOG(LVL_DUMP, "TUN Flags: 0x%04x", flags);
+			LOG(LVL_DUMP, "TUN Proto: 0x%04x", proto);
+
+			switch(proto) {
+				case 0x86dd: // IPv6
+					ERR_CHECK(connection_list_enqueue_packet(&ctx->conn_list, packetbuf, ret));
+					break;
+
+				default:
+					LOG(LVL_WARN, "Unsupported Protocol 0x%04x. Packet dropped.", proto);
+					continue;
+			}
+		}
+	}
+
+	return OK;
+}
+
+
+size_t digipeater_encode_next_packet(digipeater_ctx_t *ctx, uint8_t *buf, size_t buf_len, bool *end_burst)
+{
+	size_t packet_size = 0;
+
+	*end_burst = false;
+
+	// send packets from the one-shot queue
+	const packet_queue_entry_t *queue_entry = packet_queue_get(&ctx->oneshot_queue, 0);
+	if(queue_entry) {
+		*end_burst = queue_entry->header.tx_request == 1;
+		packet_size = layer2_encode_packet(&queue_entry->header, queue_entry->data, queue_entry->data_len, buf, buf_len);
+		if(packet_size) {
+			packet_queue_delete(&ctx->oneshot_queue, 1);
+			return packet_size;
+		}
+	}
+
+	switch(ctx->state) {
+		case DIGIPEATER_STATE_BEACON:
+			packet_size = encode_beacon_packet(ctx, buf, buf_len);
+			*end_burst = true;
+			break;
+
+		case DIGIPEATER_STATE_CONN: {
+				// pull packets from the current connection
+				connection_list_entry_t *head = connection_list_get_head(&ctx->conn_list);
+				if(!head) {
+					return 0;
+				}
+
+				connection_ctx_t *conn = &head->connection;
+				if(connection_can_transmit(conn)) {
+					packet_size = connection_encode_next_packet(conn, buf, buf_len, end_burst);
+
+					ctx->state = DIGIPEATER_STATE_CONN;
+				}
+			}
+			break;
+	}
+
+	return packet_size;
+}
+
+
+bool digipeater_can_transmit(digipeater_ctx_t *ctx)
+{
+	uint64_t now = get_hires_time();
+
+	if(now > ctx->next_beacon_time) {
+		return true;
+	}
+
+	if(packet_queue_get_used_space(&ctx->oneshot_queue) != 0) {
+		return true;
+	}
+
+	connection_list_entry_t *head = connection_list_get_head(&ctx->conn_list);
+	if(head) {
+		return connection_can_transmit(&head->connection);
+	}
+
+	return false;
+}
+
+
+void digipeater_extend_cycle(digipeater_ctx_t *ctx, uint64_t ns)
+{
+	ctx->cycle_end_time += ns;
+}
+
+
+result_t digipeater_end_cycle(digipeater_ctx_t *ctx)
+{
+	uint64_t now = get_hires_time();
+
+	if(now >= ctx->next_beacon_time) {
+		ctx->next_beacon_time += HRTIME_MS(BEACON_INTERVAL_MS);
+		ctx->state = DIGIPEATER_STATE_BEACON;
+	} else {
+		// TODO: adjust the time based on connection activity; right now this results
+		// in round-robin scheduling
+		connection_list_reschedule_head(&ctx->conn_list, now + HRTIME_MS(MIN_INTERVAL_TIME_MS));
+
+		ctx->state = DIGIPEATER_STATE_CONN;
+		ctx->cycle_end_time = now + HRTIME_MS(MIN_INTERVAL_TIME_MS);
+	}
+
+	return OK;
+}
+
+
+result_t digipeater_maintain(digipeater_ctx_t *ctx)
+{
+	uint64_t now = get_hires_time();
+
+	if(now > ctx->cycle_end_time) {
+		// at the end of the cycle, the next connection is activated and the
+		// current one is re-scheduled.
+		LOG(LVL_DEBUG, "Interval ended by timeout at %llu ns.", now);
+		digipeater_end_cycle(ctx);
+	}
+
+	switch(ctx->state) {
+		case DIGIPEATER_STATE_CONN:
+			{
+				connection_list_entry_t *head = connection_list_get_head(&ctx->conn_list);
+				if(!head) {
+					LOG(LVL_INFO, "No active connection -> force beacon state for packet handling.");
+					ctx->state = DIGIPEATER_STATE_BEACON;
+				} else {
+					ERR_CHECK(connection_maintain(&head->connection));
+				}
+				break;
+			}
+
+		case DIGIPEATER_STATE_BEACON:
+			// nothing to do here
+			break;
+	}
+
+	return OK;
+}

impl/src/layer2/digipeater.h

@@ -0,0 +1,126 @@
+/*
+ * This file contains functions to handle a single layer 2 digipeater.
+ *
+ * SPDX-License-Identifier: GPL-3.0-or-later
+ *
+ * Copyright (C) 2024  Thomas Kolb
+ */
+
+#ifndef DIGIPEATER_H
+#define DIGIPEATER_H
+
+#include <results.h>
+#include <stdbool.h>
+
+#include "connection_list.h"
+#include "layer2/packet_queue.h"
+
+struct digipeater_ctx_s;
+
+typedef enum {
+	DIGIPEATER_STATE_BEACON, //!< Beacon sent, waiting for connection requests
+	DIGIPEATER_STATE_CONN,   //!< Handling client connections
+} digipeater_state_t;
+
+typedef enum {
+	DIGIPEATER_EVT_INTERVAL_END, //!< The current cycle has ended and new packets should be transmitted.
+} digipeater_evt_t;
+
+/*!\brief Type for a callback function that is called when certain events occur. */
+typedef void (*digipeater_evt_callback_t)(struct digipeater_ctx_s *digi, digipeater_evt_t evt);
+
+typedef struct digipeater_ctx_s {
+	digipeater_state_t state; //!< Current operating state
+
+	digipeater_evt_callback_t event_cb; //!< Callback function for events.
+
+	ham64_t my_addr;             //!< The local link layer address.
+
+	packet_queue_t oneshot_queue; //!< Queue for packets that are sent once and connection-independent
+
+	uint64_t next_beacon_time;   //!< Absolute timestamp of the next beacon transmission.
+	uint64_t cycle_end_time;  //!< Absolute timestamp of the end of the current cycle.
+
+	connection_list_t conn_list; //!< List of connections.
+} digipeater_ctx_t;
+
+
+/*!\brief Initialize the digipeater context.
+ *
+ * \param ctx       The digipeater context to initialize.
+ * \param my_addr   The local link layer address.
+ * \returns OK if everything worked or a fitting error code.
+ */
+result_t digipeater_init(
+		digipeater_ctx_t *ctx,
+		const ham64_t *my_addr);
+
+/*!\brief Destroy the given digipeater context.
+ */
+void digipeater_destroy(digipeater_ctx_t *ctx);
+
+/*!\brief Handle a received packet.
+ *
+ * \param[inout] ctx           The digipeater context.
+ * \param[in]    buf           Pointer to the packet data.
+ * \param[in]    buf_len       Length of the packet.
+ * \param[out]   data_packet   Structure will be filled with a received data packet.
+ * \returns           A result code from the packet handling procedure.
+ */
+result_t digipeater_handle_packet(
+		digipeater_ctx_t *ctx,
+		const uint8_t *buf,
+		size_t buf_len,
+		layer2_data_packet_t *data_packet);
+
+/*!\brief Enqueue a packet for transmission.
+ * \param ctx       The digipeater context.
+ * \param tun_fd    File descriptor for an open TUN device.
+ */
+result_t digipeater_fill_packet_queues_from_tundev(digipeater_ctx_t *ctx, int tun_fd);
+
+/*!\brief Encode the next packet for transmission.
+ *
+ * \note
+ * If no packets are currently available for transmission, this function returns zero.
+ *
+ * \param ctx         The digipeater context.
+ * \param buf         Where to write the encoded packet data.
+ * \param buf_len     Space available in the buffer.
+ * \param end_burst   Set to true if this packet should end the current burst and start the transmission.
+ * \returns           The number of bytes written to buf or zero if no packet was available.
+ */
+size_t digipeater_encode_next_packet(digipeater_ctx_t *ctx, uint8_t *buf, size_t buf_len, bool *end_burst);
+
+/*!\brief Check if there are packets queued for transmission.
+ */
+bool digipeater_can_transmit(digipeater_ctx_t *ctx);
+
+/*!\brief Extend the current cycle.
+ *
+ * By default, the cycle duration is set to a minimum length. It must be
+ * extended by the time needed to transmit and receive packets. As the time
+ * necessary for packet transfer is unknown to the Layer 2, it must be
+ * calculated externally.
+ *
+ * This function should be called whenever
+ * - A packet is transmitted from digipeater_encode_next_packet(), or
+ * - A packet is being received
+ */
+void digipeater_extend_cycle(digipeater_ctx_t *ctx, uint64_t ns);
+
+/*!\brief End the current cycle.
+ *
+ * End the cycle without waiting for the timeout. This switches to the next
+ * connection or transmits a beacon. In any case, it stops forwarding received
+ * packets to the current connection.
+ */
+result_t digipeater_end_cycle(digipeater_ctx_t *ctx);
+
+/*!\brief Handle internal maintenance tasks.
+ *
+ * This should be called periodically to handle timeouts and retransmissions.
+ */
+result_t digipeater_maintain(digipeater_ctx_t *ctx);
+
+#endif // DIGIPEATER_H

impl/src/layer2/packet_structs.c

@@ -8,9 +8,13 @@
 #include <assert.h>
 #include <string.h>
 
+#include <liquid/liquid.h>
+
 #define LOGGER_MODULE_NAME "l2ps"
 #include "logger.h"
 
+#include "config.h"
+
 #include "packet_structs.h"
 
 
@@ -129,3 +133,36 @@ void layer2_dump_packet_header(int level, const layer2_packet_header_t *header)
 	}
 
 }
+
+
+size_t layer2_encode_packet(
+		const layer2_packet_header_t *header,
+		uint8_t *payload, size_t payload_len,
+		uint8_t *data_out, size_t data_out_len)
+{
+	unsigned int crc_size = crc_sizeof_key(PAYLOAD_CRC_SCHEME);
+
+	if(data_out_len < LAYER2_PACKET_HEADER_ENCODED_SIZE_MAX) {
+		return 0;
+	}
+
+	size_t packet_size = layer2_encode_packet_header(header, data_out);
+
+	if(data_out_len < packet_size + crc_size + payload_len) {
+		return 0;
+	}
+
+	// add the payload data
+	if(payload) {
+		memcpy(data_out + packet_size, payload, payload_len);
+	}
+
+	packet_size += payload_len;
+
+	// calculate CRC of everything and append it to the packet
+	crc_append_key(PAYLOAD_CRC_SCHEME, data_out, packet_size);
+
+	packet_size += crc_size;
+
+	return packet_size;
+}

impl/src/layer2/packet_structs.h

@@ -33,6 +33,7 @@ typedef struct layer2_packet_header_s {
 	ham64_t dst_addr;               //!< destination HAM-64 address
 } layer2_packet_header_t;
 
+
 // maximum header size
 // - 1 byte packet info
 // - 1 byte sequence numbers
@@ -74,6 +75,19 @@ bool layer2_decode_packet_header(const uint8_t *encoded, size_t encoded_len, lay
  */
 void layer2_dump_packet_header(int level, const layer2_packet_header_t *header);
 
+/*!\brief Encode a complete packet (with header and CRC)
+ * \param header       The header structure to encode.
+ * \param payload      The payload data to encode.
+ * \param payload_len  The length of the payload data.
+ * \param data_out     The buffer for the encoded packet data.
+ * \param data_out_len The size of the output buffer.
+ * \returns The number of bytes written to the output buffer or 0 if not enough space was available.
+ */
+size_t layer2_encode_packet(
+		const layer2_packet_header_t *header,
+		uint8_t *payload, size_t payload_len,
+		uint8_t *data_out, size_t data_out_len);
+
 /*!\brief Get a string representation of the given message type.
  */
 const char* layer2_msg_type_to_string(layer2_message_type_t type);
@@ -82,15 +96,36 @@ const char* layer2_msg_type_to_string(layer2_message_type_t type);
 
 typedef enum {
 	L2_PAYLOAD_TYPE_IPV4 = 0x00,
-	L2_PAYLOAD_TYPE_IPV6 = 0x01
+	L2_PAYLOAD_TYPE_IPV6 = 0x01,
+
+	L2_PAYLOAD_TYPE_INVALID = 0x7FFFFFFF
 } layer2_payload_type_t;
 
-typedef struct layer2_data_header_s {
+typedef struct layer2_data_packet_s {
 	layer2_payload_type_t payload_type; //!< Type of the contained layer 3 packet
-} layer2_data_header_t;
+
+	const uint8_t        *payload;      //!< Pointer to the payload data
+	size_t                payload_len;  //!< Length of the payload data
+} layer2_data_packet_t;
 
 /* Connection Management Structs */
 
-// TODO
+typedef enum {
+	CONN_MGMT_TYPE_BEACON                = 0x00,
+	CONN_MGMT_TYPE_CONNECTION_REQUEST    = 0x01,
+	CONN_MGMT_TYPE_CONNECTION_PARAMETERS = 0x02,
+	CONN_MGMT_TYPE_CONNECTION_RESET      = 0x03,
+	CONN_MGMT_TYPE_DISCONNECT_REQUEST    = 0x04,
+	CONN_MGMT_TYPE_DISCONNECT            = 0x05,
+} conn_mgmt_type_t;
+
+typedef enum {
+	CONN_PARAM_TYPE_IPV6_ADDRESS          = 0x00,
+	CONN_PARAM_TYPE_IPV6_GATEWAY          = 0x01,
+	CONN_PARAM_TYPE_IPV6_DNS              = 0x02,
+	CONN_PARAM_TYPE_IPV4_ADDRESS          = 0x08,
+	CONN_PARAM_TYPE_IPV4_GATEWAY          = 0x09,
+	CONN_PARAM_TYPE_IPV4_DNS              = 0x0A,
+} conn_param_type_t;
 
 #endif // PACKET_STRUCTS_H

impl/src/layer2/tundev.c

@@ -29,7 +29,7 @@ int tundev_open(char *dev)
 	struct ifreq ifr;
 	memset(&ifr, 0, sizeof(ifr));
 
-	ifr.ifr_flags = IFF_TUN | IFF_NO_PI;
+	ifr.ifr_flags = IFF_TUN;
 
 	strncpy(ifr.ifr_name, dev, IFNAMSIZ);
 

impl/src/main.c

@@ -50,7 +50,7 @@
 static int  m_tunfd = -1;
 static bool m_running = true;
 
-static double next_tx_switch_time = 0.0;
+static uint64_t next_tx_switch_time = 0;
 
 static rx_stats_t m_rx_stats;
 
@@ -71,7 +71,7 @@ static void signal_handler(int signal, siginfo_t *info, void *ctx)
 
 static void block_tx_for(unsigned offset_ms)
 {
-	next_tx_switch_time = get_hires_time() + (double)offset_ms * 0.001;
+	next_tx_switch_time = get_hires_time() + HRTIME_MS(offset_ms);
 }
 
 void print_complex_array(const char *varname, float complex const *array, size_t len)
@@ -186,7 +186,7 @@ int main(int argc, char **argv)
 
 	bool on_air = true;
 
-	srand((int)(get_hires_time() * 1e6));
+	srand(get_hires_time());
 
 	// ** Initialize **
 
@@ -233,18 +233,18 @@ int main(int argc, char **argv)
 
 	unsigned rx_retries = 0;
 
-	double old = get_hires_time();
+	uint64_t old = get_hires_time();
 	size_t total_samples = 0;
-	double next_stats_print_time = old + 0.5;
+	uint64_t next_stats_print_time = old + HRTIME_MS(500);
 
-	double retransmit_time = 0.0;
+	uint64_t retransmit_time = 0;
 
 	while(m_running) {
-		double now = get_hires_time();
+		uint64_t now = get_hires_time();
 
-		if(retransmit_time != 0.0 && now >= retransmit_time) {
+		if(retransmit_time != 0 && now >= retransmit_time) {
 			LOG(LVL_INFO, "Retransmit triggered.");
-			retransmit_time = 0.0;
+			retransmit_time = 0;
 			layer2_tx_restart(&l2tx);
 		}
 
@@ -325,7 +325,7 @@ int main(int argc, char **argv)
 				RESULT_CHECK(sdr_start_rx(&sdr));
 				on_air = false;
 
-				retransmit_time = get_hires_time() + 1.0 + 1.0 * rand() / RAND_MAX;
+				retransmit_time = get_hires_time() + HRTIME_SEC(1) + HRTIME_SEC(1.0 * rand() / RAND_MAX);
 
 				block_tx_for(TX_SWITCH_BACKOFF_AFTER_RX_ON);
 			}
@@ -360,9 +360,9 @@ int main(int argc, char **argv)
 
 			total_samples += n_rf_samples;
 
-			double new = get_hires_time();
+			uint64_t new = get_hires_time();
 			if(new >= next_stats_print_time) {
-				double rate = total_samples / (new - old);
+				double rate = total_samples * 1e9 / (new - old);
 				LOG(LVL_INFO, "\nEstimated rate: %.3f MS/s", rate / 1e6);
 				LOG(LVL_INFO, "Receiver statistics:");
 				LOG(LVL_INFO, "  Preambles found:    %8zd", m_rx_stats.preambles_found);

impl/src/sdr/sdr.c

@@ -99,9 +99,9 @@ static int tx_callback(hackrf_transfer *transfer)
 		return HACKRF_ERROR_OTHER;
 	}
 
-	if(sdr_ctx->tx_start_time == 0.0) {
+	if(sdr_ctx->tx_start_time == 0) {
 		sdr_ctx->tx_start_time = get_hires_time();
-		sdr_ctx->tx_duration = 10e-3; // give a little headroom
+		sdr_ctx->tx_duration = HRTIME_MS(10); // give a little headroom
 		LOG(LVL_INFO, "TX time tracking reset: start = %.3f.", sdr_ctx->tx_start_time);
 	}
 
@@ -112,7 +112,7 @@ static int tx_callback(hackrf_transfer *transfer)
 
 	if(samples_read != 0) {
 		// only add time if any actual samples were transmitted
-		sdr_ctx->tx_duration += (double)samples_requested / SDR_TX_SAMPLING_RATE;
+		sdr_ctx->tx_duration += HRTIME_SEC((double)samples_requested / SDR_TX_SAMPLING_RATE);
 	}
 
 	LOG(LVL_DEBUG, "copied %u samples to HackRF.", samples_read);
@@ -377,8 +377,8 @@ result_t sdr_flush_tx_buffer(sdr_ctx_t *ctx)
 			return 0;
 		}
 
-		double now = get_hires_time();
-		double end = ctx->tx_start_time + ctx->tx_duration;
+		uint64_t now = get_hires_time();
+		uint64_t end = ctx->tx_start_time + ctx->tx_duration;
 
 		if(sem_post(&ctx->buf_sem) < 0) {
 			LOG(LVL_ERR, "sem_post: %s", strerror(errno));

impl/src/utils.c

@@ -76,11 +76,23 @@ err_close:
 	return false;
 }
 
-double get_hires_time(void)
+uint64_t get_hires_time(void)
 {
 	struct timespec clk;
 	clock_gettime(CLOCK_MONOTONIC, &clk);
-	return clk.tv_sec + 1e-9 * clk.tv_nsec;
+	return clk.tv_sec * 1000000000ULL + (uint64_t)clk.tv_nsec;
+}
+
+void sleep_until(uint64_t hires_time)
+{
+	struct timespec tv;
+	int ret;
+
+	tv.tv_sec = hires_time / 1000000000ULL;
+	tv.tv_nsec = hires_time % 1000000000ULL;
+	do {
+		ret = clock_nanosleep(CLOCK_MONOTONIC, TIMER_ABSTIME, &tv, NULL);
+	} while(ret == EINTR);
 }
 
 void fsleep(double d)
@@ -93,18 +105,6 @@ void fsleep(double d)
 	nanosleep(&ts, NULL);
 }
 
-void sleep_until(double hires_time)
-{
-	struct timespec tv;
-	int ret;
-
-	tv.tv_sec = hires_time;
-	tv.tv_nsec = (uint64_t)(1e9 * hires_time) % 1000000000;
-	do {
-		ret = clock_nanosleep(CLOCK_MONOTONIC, TIMER_ABSTIME, &tv, NULL);
-	} while(ret == EINTR);
-}
-
 void hexdump(const uint8_t *data, size_t len)
 {
 	static const char lut[16] = "0123456789ABCDEF";

impl/src/utils.h

@@ -11,6 +11,10 @@
 #include <stdbool.h>
 #include <liquid/liquid.h>
 
+#define HRTIME_US(x) ((uint64_t)(1000ULL * (x)))
+#define HRTIME_MS(x) ((uint64_t)(1000000ULL * (x)))
+#define HRTIME_SEC(x) ((uint64_t)(1000000000ULL * (x)))
+
 /*! Dump a array of complex numbers.
  *
  * \param data      Pointer to the data to dump.
@@ -30,9 +34,26 @@ bool dump_array_cf(const float complex *data, size_t n, float T, const char *fil
  */
 bool dump_array_f(const float *data, size_t n, float T, const char *filename);
 
-void sleep_until(double hires_time);
+
+/*! Sleep until the given absolute timestamp in ns.
+ *
+ * The current timestamp can be retrieved using \ref get_hires_time().
+ *
+ * \param hires_time     The resume timestamp in ns.
+ */
+void sleep_until(uint64_t hires_time);
+
+
+/*! Returns the current high-resulution timestamp.
+ *
+ * This timestamp comes from the CLOCK_MONOTONIC source and has no defined relation to the wall-clock time. It can be used to calculate intervals, though.
+ *
+ * \returns A timestamp in nanosecond resolution.
+ */
+uint64_t get_hires_time(void);
+
+
 void fsleep(double d);
-double get_hires_time(void);
 
 void hexdump(const uint8_t *data, size_t len);
 

impl/test/CMakeLists.txt

@@ -222,3 +222,7 @@ target_link_libraries(
 	test_interleaver
 	m
 )
+
+#------------------------------------
+
+add_subdirectory(layer2_over_udp)

impl/test/layer2_over_udp/CMakeLists.txt

@@ -0,0 +1,73 @@
+add_executable(
+	l2udptest_client
+	../../src/utils.c
+	../../src/utils.h
+	../../src/logger.c
+	../../src/logger.h
+	../../src/options.c
+	../../src/options.h
+	../../src/var_array.c
+	../../src/var_array.h
+	../../src/config.h
+	../../src/jsonlogger.c
+	../../src/jsonlogger.h
+	../../src/debug_structs.h
+	../../src/layer2/packet_structs.c
+	../../src/layer2/packet_structs.h
+	../../src/layer2/ham64.c
+	../../src/layer2/ham64.h
+	../../src/layer2/packet_queue.c
+	../../src/layer2/packet_queue.h
+	../../src/layer2/connection.c
+	../../src/layer2/connection.h
+	../../src/layer2/tundev.c
+	../../src/layer2/tundev.h
+	l2udptest_client.c
+)
+
+target_link_libraries(
+	l2udptest_client
+	fec
+	m
+	liquid
+)
+
+#---------------------------
+
+add_executable(
+	l2udptest_digipeater
+	../../src/utils.c
+	../../src/utils.h
+	../../src/logger.c
+	../../src/logger.h
+	../../src/options.c
+	../../src/options.h
+	../../src/var_array.c
+	../../src/var_array.h
+	../../src/config.h
+	../../src/jsonlogger.c
+	../../src/jsonlogger.h
+	../../src/debug_structs.h
+	../../src/layer2/packet_structs.c
+	../../src/layer2/packet_structs.h
+	../../src/layer2/ham64.c
+	../../src/layer2/ham64.h
+	../../src/layer2/packet_queue.c
+	../../src/layer2/packet_queue.h
+	../../src/layer2/connection.c
+	../../src/layer2/connection.h
+	../../src/layer2/connection_list.c
+	../../src/layer2/connection_list.h
+	../../src/layer2/digipeater.c
+	../../src/layer2/digipeater.h
+	../../src/layer2/tundev.c
+	../../src/layer2/tundev.h
+	l2udptest_digipeater.c
+)
+
+target_link_libraries(
+	l2udptest_digipeater
+	fec
+	m
+	liquid
+)

impl/test/layer2_over_udp/l2udptest_client.c

@@ -0,0 +1,395 @@
+/*
+ * 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 "layer2/ham64.h"
+#include "utils.h"
+
+#define LOGGER_MODULE_NAME "main"
+#include "logger.h"
+#include "jsonlogger.h"
+#include "debug_structs.h"
+
+#include "layer2/connection.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 uint64_t next_tx_switch_time = 0;
+
+static rx_stats_t m_rx_stats;
+
+static connection_ctx_t l2conn;
+
+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() + HRTIME_MS(offset_ms);
+}
+
+
+void handle_received_packet(uint8_t *packet_data, size_t packet_len)
+{
+}
+
+
+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;
+}
+
+
+void rx_data_to_tun(const layer2_data_packet_t *data_packet)
+{
+	uint8_t tun_packet[4 + data_packet->payload_len];
+
+	// flags
+	tun_packet[0] = 0;
+	tun_packet[1] = 0;
+
+	switch(data_packet->payload_type) {
+		case L2_PAYLOAD_TYPE_IPV6:
+			*(uint16_t*)(tun_packet+2) = 0x86dd;
+			break;
+		case L2_PAYLOAD_TYPE_IPV4:
+			*(uint16_t*)(tun_packet+2) = 0x0800;
+			break;
+		default:
+			LOG(LVL_ERR, "Unsupported payload type: 0x%08x.", data_packet->payload_type);
+			return;
+	}
+
+	memcpy(tun_packet+4, data_packet->payload, data_packet->payload_len);
+
+	int ret = write(m_tunfd, tun_packet, sizeof(tun_packet));
+	if(ret < 0) {
+		LOG(LVL_ERR, "write(tun): %s", strerror(errno));
+	}
+}
+
+
+void conn_evt_cb(struct connection_ctx_s *conn, connection_evt_t evt, void *user_ctx)
+{
+	(void)conn;
+	(void)evt;
+	(void)user_ctx;
+}
+
+
+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(get_hires_time());
+
+	// ** Initialize **
+
+	char devname[IFNAMSIZ] = "hamnet70";
+	m_tunfd = tundev_open(devname);
+
+	if(m_tunfd < 0) {
+		return 1;
+	}
+
+	ham64_t my_address, peer_address;
+	ham64_encode(MY_CALL, &my_address);
+	ham64_encode(PEER_CALL, &peer_address);
+	RESULT_CHECK(connection_init(&l2conn, &my_address, &peer_address, conn_evt_cb, NULL));
+
+	// force connection into the established state
+	l2conn.conn_state = CONN_STATE_ESTABLISHED;
+
+	// ** 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_tun;
+	memset(&pfd_tun, 0, sizeof(pfd_tun));
+
+	pfd_tun.fd = m_tunfd;
+	pfd_tun.events = POLLIN;
+
+	struct pollfd pfd_bcast;
+	memset(&pfd_bcast, 0, sizeof(pfd_bcast));
+
+	pfd_bcast.fd = m_bcast_sock;
+	pfd_bcast.events = POLLIN;
+
+	uint64_t old = get_hires_time();
+	size_t total_bytes = 0;
+	uint64_t next_stats_print_time = old + HRTIME_MS(500);
+
+	// TODO: wait for beacon
+	// TODO: send connection request
+
+	while(m_running) {
+		uint64_t now = get_hires_time();
+
+		RESULT_CHECK(connection_maintain(&l2conn));
+
+		// fill the TX queue from the TUN device
+		while(connection_can_enqueue_packet(&l2conn)) {
+			int ret = poll(&pfd_tun, 1, 0 /* timeout */);
+			if(ret < 0) {
+				LOG(LVL_ERR, "poll: %s", strerror(errno));
+				break;
+			} else if(ret == 0) {
+				// no more packets
+				break;
+			} else {
+				// a packet is available -> move it to the queue
+				static const size_t packetbuf_size = 2048;
+				uint8_t packetbuf[packetbuf_size];
+				ret = read(m_tunfd, packetbuf, packetbuf_size);
+				if(ret < 0) {
+					LOG(LVL_ERR, "read: %s", strerror(errno));
+					return ERR_SYSCALL;
+				} else if(ret == 0) {
+					// no more data, should not happen
+					break;
+				}
+
+				LOG(LVL_DUMP, "TUN Flags: 0x%04x", *(uint16_t*)packetbuf);
+				LOG(LVL_DUMP, "TUN Proto: 0x%04x", *((uint16_t*)packetbuf + 1));
+
+				RESULT_CHECK(connection_enqueue_packet(&l2conn, packetbuf, ret));
+			}
+		}
+
+		// transmit one burst
+		if(connection_can_transmit(&l2conn)) {
+			// 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;
+				bool end_burst;
+
+				packet_size = connection_encode_next_packet(&l2conn,
+						packet_buf, sizeof(packet_buf), &end_burst);
+
+				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));
+
+				if(end_burst) {
+					LOG(LVL_DEBUG, "Ending burst on request.");
+					break;
+				}
+			}
+
+			connection_tx_clean_empty_packet(&l2conn);
+
+			LOG(LVL_DEBUG, "Burst finished: %zd packets sent.", burst_len);
+		}
+
+		// make sure the receiver runs for a minimum amount of time
+		block_tx_for(TX_SWITCH_BACKOFF_AFTER_RX_ON);
+
+		while(get_hires_time() < next_tx_switch_time) {
+			// ** Receive signal **
+
+			int ret = poll(&pfd_bcast, 1, 10);
+			if(ret < 0) {
+				LOG(LVL_ERR, "poll: %s", strerror(errno));
+				return EXIT_FAILURE;
+			}
+
+			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));
+				return EXIT_FAILURE;
+			} else if(ret == 0) {
+				LOG(LVL_ERR, "recv() returned zero.");
+				return EXIT_FAILURE;
+			}
+
+			block_tx_for(TX_SWITCH_BACKOFF_END_OF_PACKET_MS);
+
+			layer2_data_packet_t data_packet;
+
+			result_t result = connection_handle_packet(&l2conn, packetbuf, ret, &data_packet);
+			switch(result) {
+				case OK:
+					m_rx_stats.successful_decodes++;
+
+					if(data_packet.payload_len != 0) {
+						rx_data_to_tun(&data_packet);
+					}
+					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;
+			}
+
+			total_bytes += ret;
+
+			uint64_t new = get_hires_time();
+			if(new >= next_stats_print_time) {
+				double rate = total_bytes * 1e9 / (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 += HRTIME_MS(500);
+
+				total_bytes = 0;
+				old = new;
+			}
+
+			fprintf(stderr, "r");
+		}
+	}
+
+	// ** Cleanup **
+
+	close(m_bcast_sock);
+
+	connection_destroy(&l2conn);
+
+	jsonlogger_shutdown();
+
+	LOG(LVL_INFO, "Done.");
+
+	logger_shutdown();
+}
+

impl/test/layer2_over_udp/l2udptest_digipeater.c

@@ -0,0 +1,348 @@
+/*
+ * 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 "layer2/ham64.h"
+#include "utils.h"
+
+#define LOGGER_MODULE_NAME "main"
+#include "logger.h"
+#include "jsonlogger.h"
+#include "debug_structs.h"
+
+#include "layer2/digipeater.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 uint64_t next_tx_switch_time = 0;
+
+static rx_stats_t m_rx_stats;
+
+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() + HRTIME_MS(offset_ms);
+}
+
+
+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;
+}
+
+
+void rx_data_to_tun(const layer2_data_packet_t *data_packet)
+{
+	uint8_t tun_packet[4 + data_packet->payload_len];
+
+	// flags
+	tun_packet[0] = 0;
+	tun_packet[1] = 0;
+
+	switch(data_packet->payload_type) {
+		case L2_PAYLOAD_TYPE_IPV6:
+			*(uint16_t*)(tun_packet+2) = 0x86dd;
+			break;
+		case L2_PAYLOAD_TYPE_IPV4:
+			*(uint16_t*)(tun_packet+2) = 0x0800;
+			break;
+		default:
+			LOG(LVL_ERR, "Unsupported payload type: 0x%08x.", data_packet->payload_type);
+			return;
+	}
+
+	memcpy(tun_packet+4, data_packet->payload, data_packet->payload_len);
+
+	int ret = write(m_tunfd, tun_packet, sizeof(tun_packet));
+	if(ret < 0) {
+		LOG(LVL_ERR, "write(tun): %s", strerror(errno));
+	}
+}
+
+
+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;
+	}
+
+	srand(get_hires_time());
+
+	// ** Initialize **
+
+	char devname[IFNAMSIZ] = "hamnet70";
+	m_tunfd = tundev_open(devname);
+
+	if(m_tunfd < 0) {
+		return 1;
+	}
+
+	ham64_t my_address, peer_address;
+	ham64_encode(MY_CALL, &my_address);
+	ham64_encode(PEER_CALL, &peer_address);
+
+	digipeater_ctx_t digipeater;
+	RESULT_CHECK(digipeater_init(&digipeater, &my_address));
+
+	// ** 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_tun;
+	memset(&pfd_tun, 0, sizeof(pfd_tun));
+
+	pfd_tun.fd = m_tunfd;
+	pfd_tun.events = POLLIN;
+
+	struct pollfd pfd_bcast;
+	memset(&pfd_bcast, 0, sizeof(pfd_bcast));
+
+	pfd_bcast.fd = m_bcast_sock;
+	pfd_bcast.events = POLLIN;
+
+	uint64_t old = get_hires_time();
+	size_t total_bytes = 0;
+	uint64_t next_stats_print_time = old + HRTIME_MS(500);
+
+	while(m_running) {
+		RESULT_CHECK(digipeater_maintain(&digipeater));
+
+		RESULT_CHECK(digipeater_fill_packet_queues_from_tundev(&digipeater, m_tunfd));
+
+		// transmit anything available
+		if(digipeater_can_transmit(&digipeater)) {
+			// 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;
+
+				bool end_burst;
+
+				packet_size = digipeater_encode_next_packet(&digipeater,
+						packet_buf, sizeof(packet_buf), &end_burst);
+
+				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));
+
+				if(end_burst) {
+					LOG(LVL_DEBUG, "Ending burst on request.");
+					break;
+				}
+			}
+
+			// FIXME: where to put this in the digipeater?
+			//connection_tx_clean_empty_packet(&l2conn);
+
+			LOG(LVL_DEBUG, "Burst finished: %zd packets sent.", burst_len);
+		}
+
+		// make sure the receiver runs for a minimum amount of time
+		block_tx_for(TX_SWITCH_BACKOFF_AFTER_RX_ON);
+
+		// receive response
+		while(get_hires_time() < next_tx_switch_time) {
+			// ** Receive packets from the broadcast socket **
+
+			int ret = poll(&pfd_bcast, 1, 10);
+			if(ret < 0) {
+				LOG(LVL_ERR, "poll: %s", strerror(errno));
+				return EXIT_FAILURE;
+			}
+
+			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));
+				return EXIT_FAILURE;
+			} else if(ret == 0) {
+				LOG(LVL_ERR, "recv() returned zero.");
+				return EXIT_FAILURE;
+			}
+
+			block_tx_for(TX_SWITCH_BACKOFF_END_OF_PACKET_MS);
+
+			layer2_data_packet_t data_packet;
+
+			result_t result = digipeater_handle_packet(&digipeater, packetbuf, ret, &data_packet);
+			switch(result) {
+				case OK:
+					m_rx_stats.successful_decodes++;
+
+					if(data_packet.payload_len != 0) {
+						rx_data_to_tun(&data_packet);
+					}
+					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;
+			}
+
+			total_bytes += ret;
+
+			uint64_t new = get_hires_time();
+			if(new >= next_stats_print_time) {
+				double rate = total_bytes * 1e9 / (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 += HRTIME_MS(500);
+
+				total_bytes = 0;
+				old = new;
+			}
+
+			fprintf(stderr, "r");
+		}
+	}
+
+	// ** Cleanup **
+
+	close(m_bcast_sock);
+
+	digipeater_destroy(&digipeater);
+
+	jsonlogger_shutdown();
+
+	LOG(LVL_INFO, "Done.");
+
+	logger_shutdown();
+}
+