Sojus07/hackrf
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request #1168 from martinling/flush-callback

Browse Source 
Provide a callback for TX flush, rather than a wait function.
This commit is contained in:
Michael Ossmann
2022-09-23 11:12:37 -04:00
committed by GitHub
parent c0ba843085 2e73a4f2db
commit 5b98cf92d8
3 changed files with 206 additions and 149 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

121
host/hackrf-tools/src/hackrf_transfer.c
View File

@ -324,6 +324,7 @@ char* u64toa(uint64_t val, t_u64toa* str)
static volatile bool do_exit = false;
static volatile bool interrupted = false;
static volatile bool tx_complete = false;
static volatile bool flush_complete = false;
#ifdef _WIN32
static HANDLE interrupt_handle;
#endif
@ -486,20 +487,8 @@ int tx_callback(hackrf_transfer* transfer)
return -1;
}
/* Accumulate power (magnitude squared). */
uint64_t sum = 0;
for (i = 0; i < transfer->valid_length; i++) {
int8_t value = transfer->buffer[i];
sum += value * value;
}
/* Update both running totals at approximately the same time. */
byte_count += transfer->valid_length;
stream_power += sum;
/* If the last data was already buffered, stop. */
if (tx_complete) {
stop_main_loop();
return -1;
}
@ -523,7 +512,23 @@ int tx_callback(hackrf_transfer* transfer)
} else {
/* Read samples from file. */
bytes_read = fread(transfer->buffer, 1, bytes_to_read, file);
/* If no more bytes, error or file empty, terminate. */
if (bytes_read == 0) {
/* Report any error. */
if (ferror(file)) {
fprintf(stderr, "Could not read input file.\n");
stop_main_loop();
return -1;
}
if (ftell(file) < 1) {
stop_main_loop();
return -1;
}
}
}
/* Now set the valid length to the bytes we put in the buffer. */
transfer->valid_length = bytes_read;
/* If the sample limit has been reached, this is the last data. */
@ -538,26 +543,74 @@ int tx_callback(hackrf_transfer* transfer)
}
/* Otherwise, the file ran short. If not repeating, this is the last data. */
if (!repeat) {
if ((!repeat) || (ftell(file) < 1)) {
tx_complete = true;
return 0;
}
/* If we get to here, we need to repeat the file until we fill the buffer. */
while (bytes_read < bytes_to_read) {
size_t extra_bytes_read;
/* Rewind and read more samples. */
rewind(file);
bytes_read +=
extra_bytes_read =
fread(transfer->buffer + bytes_read,
1,
bytes_to_read - bytes_read,
file);
transfer->valid_length = bytes_read;
/* If no more bytes, error or file empty, use what we have. */
if (extra_bytes_read == 0) {
/* Report any error. */
if (ferror(file)) {
fprintf(stderr, "Could not read input file.\n");
tx_complete = true;
return 0;
}
if (ftell(file) < 1) {
tx_complete = true;
return 0;
}
}
bytes_read += extra_bytes_read;
transfer->valid_length += extra_bytes_read;
}
/* Then return normally. */
return 0;
}
static void tx_complete_callback(hackrf_transfer* transfer, int success)
{
// If a transfer failed to complete, stop the main loop.
if (!success) {
stop_main_loop();
return;
}
/* Accumulate power (magnitude squared). */
uint32_t i;
uint64_t sum = 0;
for (i = 0; i < transfer->valid_length; i++) {
int8_t value = transfer->buffer[i];
sum += value * value;
}
/* Update both running totals at approximately the same time. */
byte_count += transfer->valid_length;
stream_power += sum;
}
static void flush_callback(void* flush_ctx, int success)
{
if (success) {
flush_complete = true;
}
stop_main_loop();
}
static int update_stats(hackrf_device* device, hackrf_m0_state* state, stats_t* stats)
{
int result = hackrf_get_m0_state(device, state);
@ -690,7 +743,6 @@ int main(int argc, char** argv)
unsigned int lna_gain = 8, vga_gain = 20, txvga_gain = 0;
hackrf_m0_state state;
stats_t stats = {0, 0};
static int32_t preload_bytes = 0;
while ((opt = getopt(argc, argv, "Hwr:t:f:i:o:m:a:p:s:Fn:b:l:g:x:c:d:C:RS:Bh?")) !=
EOF) {
@ -1249,10 +1301,11 @@ int main(int argc, char** argv)
result |= hackrf_set_lna_gain(device, lna_gain);
result |= hackrf_start_rx(device, rx_callback, NULL);
} else {
preload_bytes = hackrf_get_transfer_queue_depth(device) *
hackrf_get_transfer_buffer_size(device);
result = hackrf_set_txvga_gain(device, txvga_gain);
result |= hackrf_enable_tx_flush(device, 1);
result |= hackrf_enable_tx_flush(device, flush_callback, NULL);
result |= hackrf_set_tx_block_complete_callback(
device,
tx_complete_callback);
result |= hackrf_start_tx(device, tx_callback, NULL);
}
@ -1281,8 +1334,7 @@ int main(int argc, char** argv)
.it_value = {.tv_sec = 1, .tv_usec = 0}};
setitimer(ITIMER_REAL, &interval_timer, NULL);
#endif
while ((hackrf_is_streaming(device) == HACKRF_TRUE) && (do_exit == false)) {
uint64_t byte_count_now;
while (!do_exit) {
struct timeval time_now;
float time_difference, rate;
if (stream_size > 0) {
@ -1315,6 +1367,7 @@ int main(int argc, char** argv)
}
#endif
} else {
uint64_t byte_count_now;
uint64_t stream_power_now;
#ifdef _WIN32
// Wait for interval timer event, or interrupt event.
@ -1332,28 +1385,11 @@ int main(int argc, char** argv)
byte_count = 0;
stream_power = 0;
/*
* The TX callback is called to preload the USB
* transfer buffers at the start of TX. This results in
* invalid statistics collected about the empty buffers
* before any USB transfer is completed. We skip these
* statistics and do not report them to the user.
*/
if (preload_bytes > 0) {
if (preload_bytes > byte_count_now) {
preload_bytes -= byte_count_now;
byte_count_now = 0;
} else {
byte_count_now -= preload_bytes;
preload_bytes = 0;
}
}
time_difference = TimevalDiff(&time_now, &time_start);
rate = (float) byte_count_now / time_difference;
if ((byte_count_now == 0) && (hw_sync)) {
fprintf(stderr, "Waiting for trigger...\n");
} else {
} else if (!((byte_count_now == 0) && (flush_complete))) {
double full_scale_ratio = (double) stream_power_now /
(byte_count_now * 127 * 127);
double dB_full_scale = 10 * log10(full_scale_ratio) + 3.0;
@ -1389,7 +1425,7 @@ int main(int argc, char** argv)
time_start = time_now;
if ((byte_count_now == 0) && (!hw_sync)) {
if ((byte_count_now == 0) && (!hw_sync) && (!flush_complete)) {
exit_code = EXIT_FAILURE;
fprintf(stderr,
"\nCouldn't transfer any bytes for one second.\n");
@ -1406,11 +1442,6 @@ int main(int argc, char** argv)
interval_timer.it_value.tv_sec = 0;
setitimer(ITIMER_REAL, &interval_timer, NULL);
#endif
if ((transmit || signalsource) && !interrupted) {
// Wait for TX to finish.
hackrf_await_tx_flush(device);
}
result = hackrf_is_streaming(device);
if (do_exit) {
fprintf(stderr, "\nExiting...\n");

222
host/libhackrf/src/hackrf.c
View File

@ -147,6 +147,9 @@ struct hackrf_device {
pthread_mutex_t all_finished_lock; /* used to protect all_finished */
bool flush;
struct libusb_transfer* flush_transfer;
hackrf_flush_cb_fn flush_callback;
hackrf_tx_block_complete_cb_fn tx_completion_callback;
void* flush_ctx;
};
typedef struct {
@ -236,6 +239,9 @@ static int cancel_transfers(hackrf_device* device)
}
}
if (device->flush_transfer != NULL)
libusb_cancel_transfer(device->flush_transfer);
device->transfers_setup = false;
device->flush = false;
@ -361,7 +367,8 @@ static int prepare_transfers(
.rx_ctx = device->rx_ctx,
.tx_ctx = device->tx_ctx,
};
if (device->callback(&transfer) == 0) {
if ((device->callback(&transfer) == 0) &&
(transfer.valid_length > 0)) {
device->transfers[transfer_index]->length =
transfer.valid_length;
ready_transfers++;
@ -724,6 +731,9 @@ static int hackrf_open_setup(libusb_device_handle* usb_device, hackrf_device** d
lib_device->active_transfers = 0;
lib_device->flush = false;
lib_device->flush_transfer = NULL;
lib_device->flush_callback = NULL;
lib_device->flush_ctx = NULL;
lib_device->tx_completion_callback = NULL;
result = pthread_mutex_init(&lib_device->transfer_lock, NULL);
if (result != 0) {
@ -1759,11 +1769,80 @@ static void* transfer_threadproc(void* arg)
return NULL;
}
static void transfer_finished(
struct hackrf_device* device,
struct libusb_transfer* finished_transfer)
static void LIBUSB_CALL hackrf_libusb_flush_callback(struct libusb_transfer* usb_transfer)
{
// If a transfer finished for any reason, we're shutting down.
bool success = usb_transfer->status == LIBUSB_TRANSFER_COMPLETED;
// All transfers have now ended, so proceed with signalling completion.
hackrf_device* device = (hackrf_device*) usb_transfer->user_data;
pthread_mutex_lock(&device->all_finished_lock);
device->flush = false;
device->active_transfers = 0;
pthread_cond_broadcast(&device->all_finished_cv);
pthread_mutex_unlock(&device->all_finished_lock);
if (device->flush_callback)
device->flush_callback(device->flush_ctx, success);
}
static void LIBUSB_CALL
hackrf_libusb_transfer_callback(struct libusb_transfer* usb_transfer)
{
hackrf_device* device = (hackrf_device*) usb_transfer->user_data;
bool success, resubmit = false;
int result;
hackrf_transfer transfer = {
.device = device,
.buffer = usb_transfer->buffer,
.buffer_length = TRANSFER_BUFFER_SIZE,
.valid_length = usb_transfer->actual_length,
.rx_ctx = device->rx_ctx,
.tx_ctx = device->tx_ctx};
success = usb_transfer->status == LIBUSB_TRANSFER_COMPLETED;
if (device->tx_completion_callback != NULL) {
device->tx_completion_callback(&transfer, success);
}
// Take lock to make sure that we don't restart a
// transfer whilst cancel_transfers() is in the middle
// of stopping them.
pthread_mutex_lock(&device->transfer_lock);
if (success) {
if (device->streaming && (device->callback(&transfer) == 0) &&
(transfer.valid_length > 0)) {
if ((resubmit = device->transfers_setup)) {
if (usb_transfer->endpoint == TX_ENDPOINT_ADDRESS) {
usb_transfer->length = transfer.valid_length;
// Pad to the next 512-byte boundary.
uint8_t* buffer = usb_transfer->buffer;
while (usb_transfer->length % 512 != 0)
buffer[usb_transfer->length++] = 0;
}
result = libusb_submit_transfer(usb_transfer);
}
} else if (device->flush) {
result = libusb_submit_transfer(device->flush_transfer);
if (result != LIBUSB_SUCCESS) {
device->streaming = false;
device->flush = false;
}
}
} else {
device->streaming = false;
device->flush = false;
}
// Now we can release the lock. Our transfer was either
// cancelled or restarted, not both.
pthread_mutex_unlock(&device->transfer_lock);
// If a data transfer was resubmitted successfully, we're done.
if (resubmit && result == LIBUSB_SUCCESS)
return;
// Otherwise, no further calls should be made to the TX callback.
device->streaming = false;
// If this is the last transfer, signal that all are now finished.
@ -1779,73 +1858,6 @@ static void transfer_finished(
pthread_mutex_unlock(&device->all_finished_lock);
}
static void LIBUSB_CALL hackrf_libusb_flush_callback(struct libusb_transfer* usb_transfer)
{
// TX buffer is now flushed, so proceed with signalling completion.
hackrf_device* device = (hackrf_device*) usb_transfer->user_data;
pthread_mutex_lock(&device->all_finished_lock);
device->flush = false;
device->active_transfers = 0;
pthread_cond_broadcast(&device->all_finished_cv);
pthread_mutex_unlock(&device->all_finished_lock);
}
static void LIBUSB_CALL
hackrf_libusb_transfer_callback(struct libusb_transfer* usb_transfer)
{
hackrf_device* device = (hackrf_device*) usb_transfer->user_data;
bool resubmit;
int result;
if (usb_transfer->status == LIBUSB_TRANSFER_COMPLETED) {
hackrf_transfer transfer = {
.device = device,
.buffer = usb_transfer->buffer,
.buffer_length = TRANSFER_BUFFER_SIZE,
.valid_length = usb_transfer->actual_length,
.rx_ctx = device->rx_ctx,
.tx_ctx = device->tx_ctx};
if (device->streaming && device->callback(&transfer) == 0) {
// Take lock to make sure that we don't restart a
// transfer whilst cancel_transfers() is in the middle
// of stopping them.
pthread_mutex_lock(&device->transfer_lock);
if ((resubmit = device->transfers_setup)) {
if (usb_transfer->endpoint == TX_ENDPOINT_ADDRESS) {
usb_transfer->length = transfer.valid_length;
// Pad to the next 512-byte boundary.
uint8_t* buffer = usb_transfer->buffer;
while (usb_transfer->length % 512 != 0)
buffer[usb_transfer->length++] = 0;
}
result = libusb_submit_transfer(usb_transfer);
}
// Now we can release the lock. Our transfer was either
// cancelled or restarted, not both.
pthread_mutex_unlock(&device->transfer_lock);
if (resubmit && result == LIBUSB_SUCCESS)
return;
} else if (device->flush) {
result = libusb_submit_transfer(device->flush_transfer);
if (result != LIBUSB_SUCCESS) {
device->streaming = false;
device->flush = false;
}
}
} else {
device->streaming = false;
device->flush = false;
}
// Unless we resubmitted this transfer and returned above,
// it's now finished.
transfer_finished(device, usb_transfer);
}
static int kill_transfer_thread(hackrf_device* device)
{
void* value;
@ -2004,45 +2016,51 @@ int ADDCALL hackrf_start_tx(
return result;
}
int ADDCALL hackrf_enable_tx_flush(hackrf_device* device, int enable)
ADDAPI int ADDCALL hackrf_set_tx_block_complete_callback(
hackrf_device* device,
hackrf_tx_block_complete_cb_fn callback)
{
if (enable) {
if (device->flush_transfer) {
return HACKRF_SUCCESS;
}
device->tx_completion_callback = callback;
return HACKRF_SUCCESS;
}
if ((device->flush_transfer = libusb_alloc_transfer(0)) == NULL) {
return HACKRF_ERROR_LIBUSB;
}
ADDAPI int ADDCALL hackrf_enable_tx_flush(
hackrf_device* device,
hackrf_flush_cb_fn callback,
void* flush_ctx)
{
device->flush_callback = callback;
device->flush_ctx = flush_ctx;
libusb_fill_bulk_transfer(
device->flush_transfer,
device->usb_device,
TX_ENDPOINT_ADDRESS,
calloc(1, DEVICE_BUFFER_SIZE),
DEVICE_BUFFER_SIZE,
hackrf_libusb_flush_callback,
device,
0);
device->flush_transfer->flags = LIBUSB_TRANSFER_FREE_BUFFER;
} else {
libusb_free_transfer(device->flush_transfer);
device->flush_transfer = NULL;
if (device->flush_transfer) {
return HACKRF_SUCCESS;
}
if ((device->flush_transfer = libusb_alloc_transfer(0)) == NULL) {
return HACKRF_ERROR_LIBUSB;
}
libusb_fill_bulk_transfer(
device->flush_transfer,
device->usb_device,
TX_ENDPOINT_ADDRESS,
calloc(1, DEVICE_BUFFER_SIZE),
DEVICE_BUFFER_SIZE,
hackrf_libusb_flush_callback,
device,
0);
device->flush_transfer->flags = LIBUSB_TRANSFER_FREE_BUFFER;
return HACKRF_SUCCESS;
}
int ADDCALL hackrf_await_tx_flush(hackrf_device* device)
ADDAPI int ADDCALL hackrf_disable_tx_flush(hackrf_device* device)
{
// Wait for the transfer thread to signal that all transfers
// have finished.
pthread_mutex_lock(&device->all_finished_lock);
while (device->active_transfers > 0) {
pthread_cond_wait(&device->all_finished_cv, &device->all_finished_lock);
}
pthread_mutex_unlock(&device->all_finished_lock);
libusb_free_transfer(device->flush_transfer);
device->flush_transfer = NULL;
device->flush_callback = NULL;
device->flush_ctx = NULL;
return HACKRF_SUCCESS;
}

12
host/libhackrf/src/hackrf.h
View File

@ -228,6 +228,8 @@ struct hackrf_device_list {
typedef struct hackrf_device_list hackrf_device_list_t;
typedef int (*hackrf_sample_block_cb_fn)(hackrf_transfer* transfer);
typedef void (*hackrf_tx_block_complete_cb_fn)(hackrf_transfer* transfer, int);
typedef void (*hackrf_flush_cb_fn)(void* flush_ctx, int);
#ifdef __cplusplus
extern "C" {
@ -270,8 +272,14 @@ extern ADDAPI int ADDCALL hackrf_start_tx(
hackrf_sample_block_cb_fn callback,
void* tx_ctx);
extern ADDAPI int ADDCALL hackrf_enable_tx_flush(hackrf_device* device, int enable);
extern ADDAPI int ADDCALL hackrf_await_tx_flush(hackrf_device* device);
extern ADDAPI int ADDCALL hackrf_set_tx_block_complete_callback(
hackrf_device* device,
hackrf_tx_block_complete_cb_fn callback);
extern ADDAPI int ADDCALL hackrf_enable_tx_flush(
hackrf_device* device,
hackrf_flush_cb_fn callback,
void* flush_ctx);
extern ADDAPI int ADDCALL hackrf_stop_tx(hackrf_device* device);