From 4c90de22c9a238be1bb691129be855d2a0d6ac3a Mon Sep 17 00:00:00 2001 From: Adam Kiss Date: Sat, 1 Aug 2026 18:45:18 +0200 Subject: [PATCH 2/2] ARM compatibility changes New fl2k_start_tx function to provide better buffer configuration to the users. Redesign thread synchronization algorithms to ensure the right memory access order using barriers on platforms that require. A small optimization for finding an empty buffer. --- CMakeLists.txt | 6 +- debian/changelog | 10 +++ include/osmo-fl2k.h | 16 +++-- src/fl2k_file.c | 2 +- src/fl2k_fm.c | 2 +- src/fl2k_tcp.c | 2 +- src/fl2k_test.c | 2 +- src/libosmo-fl2k.c | 164 +++++++++++++++++++++++++++++++++++++------- 8 files changed, 169 insertions(+), 35 deletions(-) diff --git a/CMakeLists.txt b/CMakeLists.txt index 8003c87..52c3bc0 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -45,7 +45,7 @@ endif() # Set the version information here set(VERSION_INFO_MAJOR_VERSION 0) # increment major on api compatibility changes set(VERSION_INFO_MINOR_VERSION 2) # increment minor on feature-level changes -set(VERSION_INFO_PATCH_VERSION 0) # increment patch for bug fixes and docs +set(VERSION_INFO_PATCH_VERSION 2) # increment patch for bug fixes and docs include(Version) # setup version info ######################################################################## @@ -62,6 +62,10 @@ if(CMAKE_COMPILER_IS_GNUCC AND NOT WIN32) add_definitions(-fvisibility=hidden) endif() +set(CMAKE_C_STANDARD 11) +set(CMAKE_C_STANDARD_REQUIRED ON) +set(CMAKE_C_EXTENSIONS ON) + ######################################################################## # Find build dependencies ######################################################################## diff --git a/debian/changelog b/debian/changelog index b12cee0..a352b07 100644 --- a/debian/changelog +++ b/debian/changelog @@ -1,3 +1,13 @@ +osmo-fl2k (0.2.2) unstable; urgency=medium + + [ Ádám Kiss ] + * libosmo_fl2k.c: ARM memory ordering changes for thread safety + * osmo_fl2k.h: Specify the number stalling buffers as a new argument in fl2k_start_tx function + * fl2k_file.c: Fix 0 return value of read at the end of file + * CMakeLists: C11 standard specification + + -- Oliver Smith Tue, 02 Sep 2025 08:19:06 +0200 + osmo-fl2k (0.2.1) unstable; urgency=medium [ Harald Welte ] diff --git a/include/osmo-fl2k.h b/include/osmo-fl2k.h index 02ad4ad..e8a3c6d 100644 --- a/include/osmo-fl2k.h +++ b/include/osmo-fl2k.h @@ -83,8 +83,8 @@ FL2K_API int fl2k_close(fl2k_dev_t *dev); * Set the sample rate (pixel clock) for the device * * \param dev the device handle given by fl2k_open() - * \param samp_rate the sample rate to be set, maximum value depends - * on host and USB controller + * \param samp_rate the sample rate to be set, maximum value is limited + * by the host and USB controller * \return 0 on success, -EINVAL on invalid rate */ FL2K_API int fl2k_set_sample_rate(fl2k_dev_t *dev, uint32_t target_freq); @@ -102,17 +102,21 @@ FL2K_API uint32_t fl2k_get_sample_rate(fl2k_dev_t *dev); typedef void(*fl2k_tx_cb_t)(fl2k_data_info_t *data_info); /*! - * Starts the tx thread. This function will block until - * it is being canceled using fl2k_stop_tx() + * Starts the tx thread. This function returns immediately + * after dropping the two worker threads. + * The threads could be canceled using fl2k_stop_tx() * * \param dev the device handle given by fl2k_open() * \param ctx user specific context to pass via the callback function - * \param buf_num optional buffer count, buf_num * FL2K_BUF_LEN = overall buffer size + * \param transfer_num optional buffer count, + * number of queued USB transfers * set to 0 for default buffer count (4) + * \param buf_num optional buffer count, buf_num * FL2K_BUF_LEN = overall buffer size + * set to 0 for default buffer count (transfer_num+2) * \return 0 on success */ FL2K_API int fl2k_start_tx(fl2k_dev_t *dev, fl2k_tx_cb_t cb, - void *ctx, uint32_t buf_num); + void *ctx, uint32_t transfer_num, uint32_t buf_num); /*! * Cancel all pending asynchronous operations on the device. diff --git a/src/fl2k_file.c b/src/fl2k_file.c index b0f5f07..cc3f263 100644 --- a/src/fl2k_file.c +++ b/src/fl2k_file.c @@ -178,7 +178,7 @@ int main(int argc, char **argv) goto out; } - r = fl2k_start_tx(dev, fl2k_callback, NULL, 0); + r = fl2k_start_tx(dev, fl2k_callback, NULL, 0, 0); /* Set the sample rate */ r = fl2k_set_sample_rate(dev, samp_rate); diff --git a/src/fl2k_fm.c b/src/fl2k_fm.c index f947fda..9efc2f8 100644 --- a/src/fl2k_fm.c +++ b/src/fl2k_fm.c @@ -565,7 +565,7 @@ int main(int argc, char **argv) } pthread_attr_destroy(&attr); - r = fl2k_start_tx(dev, fl2k_callback, NULL, 0); + r = fl2k_start_tx(dev, fl2k_callback, NULL, 0, 0); /* Set the sample rate */ r = fl2k_set_sample_rate(dev, samp_rate); diff --git a/src/fl2k_tcp.c b/src/fl2k_tcp.c index 871ac26..6b71644 100644 --- a/src/fl2k_tcp.c +++ b/src/fl2k_tcp.c @@ -202,7 +202,7 @@ int main(int argc, char **argv) exit(1); } - r = fl2k_start_tx(dev, fl2k_callback, NULL, buf_num); + r = fl2k_start_tx(dev, fl2k_callback, NULL, buf_num, buf_num+2); /* Set the sample rate */ r = fl2k_set_sample_rate(dev, samp_rate); diff --git a/src/fl2k_test.c b/src/fl2k_test.c index b166dda..a5b0765 100644 --- a/src/fl2k_test.c +++ b/src/fl2k_test.c @@ -285,7 +285,7 @@ int main(int argc, char **argv) buffer[i+1] = 0xff; } - r = fl2k_start_tx(dev, fl2k_callback, NULL, 0); + r = fl2k_start_tx(dev, fl2k_callback, NULL, 0, 0); /* Set the sample rate */ r = fl2k_set_sample_rate(dev, samp_rate); diff --git a/src/libosmo-fl2k.c b/src/libosmo-fl2k.c index 99dcd33..d65376d 100644 --- a/src/libosmo-fl2k.c +++ b/src/libosmo-fl2k.c @@ -28,6 +28,7 @@ #include #include #include +#include #ifndef _WIN32 #include @@ -61,7 +62,7 @@ enum fl2k_async_status { FL2K_RUNNING }; -typedef enum fl2k_buf_state { +typedef enum fl2k_buf_state : uint32_t { //Specify size to be a single memory operation on ARM BUF_EMPTY = 0, BUF_SUBMITTED, BUF_FILLED, @@ -71,6 +72,8 @@ typedef struct fl2k_xfer_info { fl2k_dev_t *dev; uint64_t seq; fl2k_buf_state_t state; + pthread_mutex_t xfer_mutex; + unsigned int id; } fl2k_xfer_info_t; struct fl2k_dev { @@ -112,11 +115,11 @@ typedef struct fl2k_dongle { const char *name; } fl2k_dongle_t; -static fl2k_dongle_t known_devices[] = { +static const fl2k_dongle_t known_devices[] = { { 0x1d5c, 0x2000, "FL2000DX OEM" }, }; -#define DEFAULT_BUF_NUMBER 4 +#define DEFAULT_XFER_NUMBER 4 #define CTRL_IN (LIBUSB_REQUEST_TYPE_VENDOR | LIBUSB_ENDPOINT_IN) #define CTRL_OUT (LIBUSB_REQUEST_TYPE_VENDOR | LIBUSB_ENDPOINT_OUT) @@ -277,10 +280,10 @@ uint32_t fl2k_get_sample_rate(fl2k_dev_t *dev) return (uint32_t)dev->rate; } -static fl2k_dongle_t *find_known_device(uint16_t vid, uint16_t pid) +static const fl2k_dongle_t *find_known_device(uint16_t vid, uint16_t pid) { unsigned int i; - fl2k_dongle_t *device = NULL; + const fl2k_dongle_t * device = NULL; for (i = 0; i < sizeof(known_devices)/sizeof(fl2k_dongle_t); i++ ) { if (known_devices[i].vid == vid && known_devices[i].pid == pid) { @@ -327,7 +330,7 @@ const char *fl2k_get_device_name(uint32_t index) libusb_context *ctx; libusb_device **list; struct libusb_device_descriptor dd; - fl2k_dongle_t *device = NULL; + const fl2k_dongle_t *device = NULL; uint32_t device_count = 0; ssize_t cnt; @@ -499,35 +502,121 @@ int fl2k_close(fl2k_dev_t *dev) return 0; } -static struct libusb_transfer *fl2k_get_next_xfer(fl2k_dev_t *dev, - fl2k_buf_state_t state) + +static pthread_mutex_t last_freed_xfer_buffer_mutex; +static int last_freed_xfer_id = -1; + +/* + * Looks for a EMPTY transfer buffer with the smallest sequence number + * Returns NULL if not found + */ +static inline struct libusb_transfer *fl2k_get_next_empty_xfer(fl2k_dev_t *dev) { unsigned int i; int next_buf = -1; uint64_t next_seq = 0; fl2k_xfer_info_t *xfer_info; + { + pthread_mutex_lock(&last_freed_xfer_buffer_mutex); + next_buf = last_freed_xfer_id; + pthread_mutex_unlock(&last_freed_xfer_buffer_mutex); + + xfer_info = (fl2k_xfer_info_t *)dev->xfer[next_buf]->user_data; + + if(xfer_info){ + pthread_mutex_lock(&xfer_info->xfer_mutex); + if (xfer_info->state == BUF_EMPTY) { + pthread_mutex_unlock(&xfer_info->xfer_mutex); + return dev->xfer[next_buf]; + } + pthread_mutex_unlock(&xfer_info->xfer_mutex); + } + } + for (i = 0; i < dev->xfer_buf_num; i++) { xfer_info = (fl2k_xfer_info_t *)dev->xfer[i]->user_data; if (!xfer_info) continue; - if (xfer_info->state == state) { - if (state == BUF_EMPTY) { + /* Sample worker thread searches for this; + * which changes the seq; so no need to + * wait for a stable seq here + * Anyway, do it! + */ + pthread_mutex_lock(&xfer_info->xfer_mutex); + if (xfer_info->state == BUF_EMPTY) { + pthread_mutex_unlock(&xfer_info->xfer_mutex); return dev->xfer[i]; - } else if ((xfer_info->seq < next_seq) || next_buf < 0) { + } + pthread_mutex_unlock(&xfer_info->xfer_mutex); + } + + return NULL; +} + +/* + * Looks for a FILLED transfer buffer with the smallest sequence number + * Returns NULL if not found + */ +static inline struct libusb_transfer *fl2k_get_next_filled_xfer_single(fl2k_dev_t *dev) +{ + unsigned int i; + int next_buf = -1; + uint64_t next_seq = 0; + fl2k_xfer_info_t *xfer_info; + + for (i = 0; i < dev->xfer_buf_num; i++) { + xfer_info = (fl2k_xfer_info_t *)dev->xfer[i]->user_data; + if (!xfer_info) + continue; + + pthread_mutex_lock(&xfer_info->xfer_mutex); + if (xfer_info->state == BUF_FILLED) { + if ((xfer_info->seq < next_seq) || next_buf < 0) { next_seq = xfer_info->seq; next_buf = i; } } + pthread_mutex_unlock(&xfer_info->xfer_mutex); } - if ((state == BUF_FILLED) && (next_buf >= 0)) + if (next_buf >= 0) return dev->xfer[next_buf]; else return NULL; } +/* + * Looks for a FILLED transfer buffer with the smallest sequence number + * Returns NULL if not found + * One retry if not found + */ +static inline struct libusb_transfer *fl2k_get_next_filled_xfer(fl2k_dev_t *dev) +{ + struct libusb_transfer* ret = fl2k_get_next_filled_xfer_single(dev); + + if (ret == NULL){ + fputs("Second try filledxfer search\n", stderr); + return fl2k_get_next_filled_xfer_single(dev); //Once again; just in hope + } + else + return ret; +} + +static struct libusb_transfer *fl2k_get_next_xfer(fl2k_dev_t *dev, + fl2k_buf_state_t current_state) +__attribute__ ((deprecated)); //Deprecated function; less branches if separated + +static struct libusb_transfer *fl2k_get_next_xfer(fl2k_dev_t *dev, + fl2k_buf_state_t current_state) +{ + switch (current_state){ + case BUF_EMPTY: return fl2k_get_next_empty_xfer(dev); + default: return fl2k_get_next_filled_xfer(dev); + } +} + static void LIBUSB_CALL _libusb_callback(struct libusb_transfer *xfer) { fl2k_xfer_info_t *xfer_info = (fl2k_xfer_info_t *)xfer->user_data; @@ -540,16 +629,23 @@ static void LIBUSB_CALL _libusb_callback(struct libusb_transfer *xfer) /* resubmit transfer */ if (FL2K_RUNNING == dev->async_status) { /* get next transfer */ - next_xfer = fl2k_get_next_xfer(dev, BUF_FILLED); + next_xfer = fl2k_get_next_filled_xfer(dev); if (next_xfer) { next_xfer_info = (fl2k_xfer_info_t *) next_xfer->user_data; /* Submit next filled transfer */ - next_xfer_info->state = BUF_SUBMITTED; + next_xfer_info->state = BUF_SUBMITTED; //Only this thread reads FILLED (neither SUBMITTED), the execution order must not interfere (architecture garantee) + //Basically this thread reads it next time. The other thread only looks for EMPTY; no metter which version (Submitted or filled) it sees. + r = libusb_submit_transfer(next_xfer); + pthread_mutex_lock(&xfer_info->xfer_mutex); + pthread_mutex_lock(&last_freed_xfer_buffer_mutex); + last_freed_xfer_id = xfer_info->id; xfer_info->state = BUF_EMPTY; pthread_cond_signal(&dev->buf_cond); + pthread_mutex_unlock(&last_freed_xfer_buffer_mutex); + pthread_mutex_unlock(&xfer_info->xfer_mutex); //Then the sampling thread reads the state, it must be set regardless of execution order } else { /* We need to re-submit the transfer * in any case, as otherwise the device @@ -652,6 +748,9 @@ static int fl2k_alloc_submit_transfers(fl2k_dev_t *dev) } } + /* init global mutex */ + pthread_mutex_init(&last_freed_xfer_buffer_mutex, NULL); + /* fill transfers */ for (i = 0; i < dev->xfer_buf_num; ++i) { libusb_fill_bulk_transfer(dev->xfer[i], @@ -665,6 +764,8 @@ static int fl2k_alloc_submit_transfers(fl2k_dev_t *dev) dev->xfer_info[i].dev = dev; dev->xfer_info[i].state = BUF_EMPTY; + dev->xfer_info[i].id = i; + pthread_mutex_init(&dev->xfer_info[i].xfer_mutex, NULL); /* if we allocate the memory through the Kernel, it is * already cleared */ @@ -672,10 +773,12 @@ static int fl2k_alloc_submit_transfers(fl2k_dev_t *dev) memset(dev->xfer_buf[i], 0, dev->xfer_buf_len); } + last_freed_xfer_id = dev->xfer_num; //No other threads, no mutex + /* submit transfers */ for (i = 0; i < dev->xfer_num; ++i) { - r = libusb_submit_transfer(dev->xfer[i]); dev->xfer_info[i].state = BUF_SUBMITTED; + r = libusb_submit_transfer(dev->xfer[i]); if (r < 0) { fprintf(stderr, "Failed to submit transfer %i\n%s", @@ -894,7 +997,7 @@ static void *fl2k_sample_worker(void *arg) if (dev->cb) dev->cb(&data_info); - xfer = fl2k_get_next_xfer(dev, BUF_EMPTY); + xfer = fl2k_get_next_empty_xfer(dev); if (!xfer) { pthread_cond_wait(&dev->buf_cond, &dev->buf_mutex); @@ -902,7 +1005,7 @@ static void *fl2k_sample_worker(void *arg) if (FL2K_RUNNING != dev->async_status) break; - xfer = fl2k_get_next_xfer(dev, BUF_EMPTY); + xfer = fl2k_get_next_empty_xfer(dev); if (!xfer) { fprintf(stderr, "no free transfer, skipping" " input buffer\n"); @@ -924,8 +1027,10 @@ static void *fl2k_sample_worker(void *arg) fl2k_convert_b(out_buf, data_info.b_buf, dev->xfer_buf_len, data_info.sampletype_signed ? 128 : 0); + pthread_mutex_lock(&xfer_info->xfer_mutex); xfer_info->seq = buf_cnt++; xfer_info->state = BUF_FILLED; + pthread_mutex_unlock(&xfer_info->xfer_mutex); } /* notify application if we've lost the device */ @@ -939,7 +1044,7 @@ static void *fl2k_sample_worker(void *arg) int fl2k_start_tx(fl2k_dev_t *dev, fl2k_tx_cb_t cb, void *ctx, - uint32_t buf_num) + uint32_t transfer_num, uint32_t buf_num) { int r = 0; int i; @@ -954,14 +1059,25 @@ int fl2k_start_tx(fl2k_dev_t *dev, fl2k_tx_cb_t cb, void *ctx, dev->cb = cb; dev->cb_ctx = ctx; - if (buf_num > 0) - dev->xfer_num = buf_num; + if (transfer_num > 0) + dev->xfer_num = transfer_num; else - dev->xfer_num = DEFAULT_BUF_NUMBER; + dev->xfer_num = DEFAULT_XFER_NUMBER; + + + if (buf_num > 0){ + dev->xfer_buf_num = buf_num; + } else { + /* have two spare buffers that can be filled while the + * others are submitted */ + dev->xfer_buf_num = dev->xfer_num + 2; + } + + if(dev->xfer_buf_num < dev->xfer_num){ + fprintf(stderr, "Logical error between active and stalling transfers\n"); + goto cleanup; + } - /* have two spare buffers that can be filled while the - * others are submitted */ - dev->xfer_buf_num = dev->xfer_num + 2; dev->xfer_buf_len = FL2K_XFER_LEN; r = fl2k_alloc_submit_transfers(dev); -- 2.54.0