Fix TX handling for multiple packets

- (only) time-based end-of-transmission tracking
  - removed tx_done flag
  - count zero-buffers correctly in time-tracking
  - add 10 ms of headroom so the transmission does not stop before buffer was
    completely transmitted (race condition)
  - fix race condition with tx_start_time in sdr_start_tx()
- simplified packet queuing (no chunking)
- read multiple packets before starting transmission (to fill buffers initially)

Thanks to rudi_s!

impl/src/main.c

@@ -155,49 +155,17 @@ void cb_rx(rx_evt_t evt, uint8_t *packet_data, size_t packet_len)
 
 static int debug_fd;
 
-
+static result_t transmit(sdr_ctx_t *sdr, const float complex *samples, size_t len)
-#define CHUNKSIZE_BB 16384
-#define CHUNKSIZE_RF (CHUNKSIZE_BB * SDR_OVERSAMPLING)
-static result_t transmit_in_chunks(sdr_ctx_t *sdr, const float complex *samples, size_t len)
 {
-	size_t transmitted = 0;
+	size_t to_transmit_rf = len * SDR_OVERSAMPLING;
-	unsigned retry_counter = 0;
+	float complex rf_samples[to_transmit_rf];
 
-	float complex rf_samples[CHUNKSIZE_RF];
+	RESULT_CHECK(sdr_baseband_to_rf(sdr, samples, len, rf_samples, &to_transmit_rf));
 
-	while(transmitted < len) {
+	result_t result = sdr_transmit(sdr, rf_samples, to_transmit_rf, 100000);
-		size_t to_transmit_bb = len - transmitted;
-		if(to_transmit_bb > CHUNKSIZE_BB) {
-			to_transmit_bb = CHUNKSIZE_BB;
-		}
 
-		size_t to_transmit_rf = CHUNKSIZE_RF;
+	fprintf(stderr, "t");
-
+	return result;
-		RESULT_CHECK(sdr_baseband_to_rf(sdr, samples + transmitted, to_transmit_bb, rf_samples, &to_transmit_rf));
-
-		result_t result = sdr_transmit(sdr, rf_samples, to_transmit_rf, 100000);
-
-		if(result != OK) {
-			retry_counter++;
-			fprintf(stderr, "sdr_transmit failed %d times\n", retry_counter);
-
-			if(retry_counter > 3) {
-				return result;
-			} else {
-				continue;
-			}
-		}
-
-		//write(debug_fd, rf_samples, to_transmit_rf*sizeof(rf_samples[0]));
-
-		fprintf(stderr, "t");
-
-		transmitted += to_transmit_bb;
-
-		retry_counter = 0;
-	}
-
-	return OK;
 }
 
 
@@ -311,9 +279,23 @@ int main(int argc, char **argv)
 				dump_array_cf(whole_burst, burst_len, 1.0f, "/tmp/tx.cpx");
 
 				// ensure that the buffer is full before TX is turned on to avoid transmitting empty buffers
-				RESULT_CHECK(transmit_in_chunks(&sdr, whole_burst, burst_len));
+				RESULT_CHECK(transmit(&sdr, whole_burst, burst_len));
 
 				if(!on_air) {
+					size_t buffer_used_samples = sdr_get_tx_buffer_used_space(&sdr);
+					const size_t min_required_samples = 4*128*1024;
+
+					int ret2 = poll(&pfd, 1, 0);
+					if(ret2 < 0) {
+						perror("poll");
+						break;
+					} else if(ret2 != 0 && (buffer_used_samples < min_required_samples)) {
+						// enqueue more packets before starting TX
+						fprintf(stderr, "Pre-buffering more packets: %zd / %zd samples.\n", buffer_used_samples, min_required_samples);
+						continue;
+					}
+
+					fprintf(stderr, "RX -> TX\n");
 					RESULT_CHECK(sdr_stop_rx(&sdr));
 
 					// transmit packets on the frequency where the last packet was received.
@@ -324,6 +306,7 @@ int main(int argc, char **argv)
 
 				on_air = true;
 			} else if(on_air) { // ret == 0
+				fprintf(stderr, "TX -> RX\n");
 				RESULT_CHECK(sdr_flush_tx_buffer(&sdr));
 
 				RESULT_CHECK(sdr_stop_tx(&sdr));

impl/src/sdr/sdr.c

@@ -94,14 +94,10 @@ static int tx_callback(hackrf_transfer *transfer)
 		return HACKRF_ERROR_OTHER;
 	}
 
-	if(samples_read == 0) {
-		// buffer has run empty, so this is the end of the transmission
-		sdr_ctx->tx_done = true;
-	}
-
 	if(sdr_ctx->tx_start_time == 0.0) {
 		sdr_ctx->tx_start_time = get_hires_time();
-		sdr_ctx->tx_duration = 0.0f;
+		sdr_ctx->tx_duration = 10e-3; // give a little headroom
+		fprintf(stderr, "TX time tracking reset: start = %.3f.\n", sdr_ctx->tx_start_time);
 	}
 
 	for(size_t i = 0; i < samples_read; i++) {
@@ -109,7 +105,10 @@ static int tx_callback(hackrf_transfer *transfer)
 		transfer->buffer[2*i + 1] = clamp_float2int8(cimag(samples[i]));
 	}
 
-	sdr_ctx->tx_duration += (double)samples_read / SDR_TX_SAMPLING_RATE;
+	if(samples_read != 0) {
+		// only add time if any actual samples were transmitted
+		sdr_ctx->tx_duration += (double)samples_requested / SDR_TX_SAMPLING_RATE;
+	}
 
 	fprintf(stderr, "copied %u samples to HackRF.\n", samples_read);
 
@@ -266,11 +265,12 @@ result_t sdr_start_tx(sdr_ctx_t *ctx, size_t burst_size)
 	result = hackrf_set_txvga_gain(ctx->hackrf, SDR_GAIN_TX_VGA);
 	CHECK_HACKRF_RESULT(result, "hackrf_set_txvga_gain");
 
+	ctx->tx_start_time = 0.0f; // will be updated by tx_callback
+
 	result = hackrf_start_tx(ctx->hackrf, tx_callback, ctx);
 	CHECK_HACKRF_RESULT(result, "hackrf_start_tx");
 
 	ctx->status = SDR_STATUS_TX;
-	ctx->tx_start_time = 0.0f; // will be updated by tx_callback
 
 	return OK;
 }
@@ -304,8 +304,6 @@ result_t sdr_transmit(sdr_ctx_t *ctx, const float complex *samples, size_t nsamp
 		return ERR_LIQUID;
 	}
 
-	ctx->tx_done = false;
-
 	if(sem_post(&ctx->buf_sem) < 0) {
 		perror("sem_post");
 		return ERR_SDR;
@@ -358,8 +356,25 @@ result_t sdr_receive(sdr_ctx_t *ctx, float complex *samples, size_t *nsamples, l
 result_t sdr_flush_tx_buffer(sdr_ctx_t *ctx)
 {
 	// block until all samples have been transmitted
-	while(!ctx->tx_done || (get_hires_time() < (ctx->tx_start_time + ctx->tx_duration))) {
+	while(true) {
-		fsleep(1e-3);
+		if(sem_wait(&ctx->buf_sem) < 0) {
+			perror("sem_wait");
+			return 0;
+		}
+
+		double now = get_hires_time();
+		double end = ctx->tx_start_time + ctx->tx_duration;
+
+		if(sem_post(&ctx->buf_sem) < 0) {
+			perror("sem_post");
+			return 0;
+		}
+
+		if(now >= end) {
+			break;
+		}
+
+		sleep_until(end);
 	}
 
 	return OK;
@@ -386,6 +401,26 @@ size_t sdr_get_tx_buffer_free_space(sdr_ctx_t *ctx)
 }
 
 
+size_t sdr_get_tx_buffer_used_space(sdr_ctx_t *ctx)
+{
+	size_t used_samples = 0;
+
+	if(sem_wait(&ctx->buf_sem) < 0) {
+		perror("sem_wait");
+		return 0;
+	}
+
+	used_samples = cbuffercf_size(ctx->tx_buf);
+
+	if(sem_post(&ctx->buf_sem) < 0) {
+		perror("sem_post");
+		return 0;
+	}
+
+	return used_samples;
+}
+
+
 result_t sdr_rf_to_baseband(sdr_ctx_t *ctx,
 		const float complex *rf_samples, size_t nrf,
 		float complex *bb_samples, size_t *nbb)

impl/src/sdr/sdr.h

@@ -31,7 +31,6 @@ typedef struct {
 	sem_t          buf_sem;
 
 	sdr_status_t   status;
-	bool           tx_done;
 	double         tx_duration;
 	double         tx_start_time; // time when tx_callback() was first called
 } sdr_ctx_t;
@@ -50,6 +49,7 @@ result_t sdr_receive(sdr_ctx_t *ctx, float complex *samples, size_t *nsamples, l
 
 result_t sdr_flush_tx_buffer(sdr_ctx_t *ctx);
 size_t   sdr_get_tx_buffer_free_space(sdr_ctx_t *ctx);
+size_t   sdr_get_tx_buffer_used_space(sdr_ctx_t *ctx);
 
 /*!
  * \brief Convert and resample a received signal to baseband.