digipeater: add one-shot packet queue and interval handling

The oneshot queue is for connection management frames that are only sent once,
at the beginning of the next burst. An example is the Connection Reset packet.

Intervals define the boundary between the handling of different connections.
The interval can be either ended by a packet with TX Request set, or by a
timeout. In either case, forwarding of packets to the current connection stops
and the connection is re-scheduled to a later point in time.

impl/src/layer2/digipeater.c

@@ -1,5 +1,4 @@
-#include <string.h>
-#include <assert.h>
+#include "layer2/packet_queue.h"
 
 #define LOGGER_MODULE_NAME "digi"
 #include "logger.h"
@@ -83,9 +82,13 @@ result_t digipeater_init(digipeater_ctx_t *ctx, const ham64_t *my_addr, digipeat
 
 	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->interval_end_time = now;
+
 	return connection_list_init(&ctx->conn_list);
 }
 
@@ -135,31 +138,40 @@ result_t digipeater_handle_packet(digipeater_ctx_t *ctx, const uint8_t *buf, siz
 		return ERR_INVALID_ADDRESS;
 	}
 
-	// FIXME: handle connection management packets here
+	// 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;
-					return connection_handle_packet_prechecked(current_conn, &header, payload, payload_len);
+					result = connection_handle_packet_prechecked(current_conn, &header, payload, payload_len);
 				} else {
 					LOG(LVL_WARN, "Digipeater in CONN state, but there is no active connection! Packet dropped.");
-					return OK;
+					result = OK;
 				}
 			}
 
 		case DIGIPEATER_STATE_BEACON:
-			return digipeater_handle_beacon_responses(ctx, &header, payload, payload_len);
+			result = digipeater_handle_beacon_responses(ctx, &header, payload, payload_len);
 	}
 
-	return OK;
+	// end the current interval if tx_request is set in an incoming packet
+	if(header.tx_request) {
+		LOG(LVL_INFO, "TX Request was received. Ending current interval.");
+		digipeater_end_interval(ctx);
+	}
+
+	return result;
 }
 
 
@@ -188,24 +200,35 @@ size_t digipeater_encode_next_packet(digipeater_ctx_t *ctx, uint8_t *buf, size_t
 
 		ctx->state = DIGIPEATER_STATE_BEACON;
 		*end_burst = true;
-		return packet_size;
-	} else {
-		// 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;
-		}
-
 		return packet_size;
 	}
+
+	// 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;
+		}
+	}
+
+	// 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;
+	}
+
+	return packet_size;
 }
 
 
@@ -226,8 +249,37 @@ bool digipeater_can_transmit(digipeater_ctx_t *ctx)
 }
 
 
+void digipeater_extend_interval(digipeater_ctx_t *ctx, uint64_t ns)
+{
+	ctx->interval_end_time += ns;
+}
+
+
+result_t digipeater_end_interval(digipeater_ctx_t *ctx)
+{
+	uint64_t now = get_hires_time();
+
+	// 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->interval_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->interval_end_time) {
+		// at the end of the interval, the next connection is activated and the
+		// current one is re-scheduled.
+		LOG(LVL_DEBUG, "Interval ended by timeout at %llu ns.", now);
+	}
+
 	switch(ctx->state) {
 		case DIGIPEATER_STATE_CONN:
 			{

impl/src/layer2/digipeater.h

@@ -13,6 +13,7 @@
 #include <stdbool.h>
 
 #include "connection_list.h"
+#include "layer2/packet_queue.h"
 
 struct digipeater_ctx_s;
 
@@ -21,17 +22,28 @@ typedef enum {
 	DIGIPEATER_STATE_CONN,   //!< Handling client connections
 } digipeater_state_t;
 
+typedef enum {
+	DIGIPEATER_EVT_INTERVAL_END, //!< The current interval has ended and new packets should be transmitted.
+} digipeater_evt_t;
+
 /*!\brief Type for a callback function that is called when a data packet was received. */
-typedef void (*digipeater_data_callback_t)(struct digipeater_ctx_s *conn, const uint8_t *data, size_t len);
+typedef void (*digipeater_data_callback_t)(struct digipeater_ctx_s *digi, const uint8_t *data, size_t len);
+
+/*!\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_data_callback_t data_cb; //!< Callback function for received data packets.
+	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 interval_end_time;  //!< Absolute timestamp of the end of the current interval.
 
 	connection_list_t conn_list; //!< List of connections.
 } digipeater_ctx_t;
@@ -85,6 +97,26 @@ size_t digipeater_encode_next_packet(digipeater_ctx_t *ctx, uint8_t *buf, size_t
  */
 bool digipeater_can_transmit(digipeater_ctx_t *ctx);
 
+/*!\brief Extend the current interval.
+ *
+ * By default, the interval 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_interval(digipeater_ctx_t *ctx, uint64_t ns);
+
+/*!\brief End the current interval.
+ *
+ * End the interval without waiting for the timeout. This switches to the next
+ * connection and stop forwarding received packets to the current one.
+ */
+result_t digipeater_end_interval(digipeater_ctx_t *ctx);
+
 /*!\brief Handle internal maintenance tasks.
  *
  * This should be called periodically to handle timeouts and retransmissions.