From 7f28f124f8769d13fd6284cf50927dd94804ed35 Mon Sep 17 00:00:00 2001
From: Karl Semich <0xloem@gmail.com>
Date: Wed, 19 Sep 2018 14:38:31 +0000
Subject: [PATCH] Add API for releasing zero-copy USB buffers.

API functions are added to provide for the use of zero-copy buffers
beyond the lifetime of the async callback.  This allows the number of
calls to the callback to be rapidly tracked in order to measure and
predict latency and data loss, even when processing the data is slower
than receiving it, without requiring memcpy() be used to take data out
of the buffers.  This addition benefits mostly multithreaded clients.

rtlsdr_read_async_ex is added which passes a buffer handle to the
callback.  Additionally, it calls its callback even in the case of
failed bulk transfers, to inform of possible gaps in the stream.

rtlsdr_read_async_ex_release is added, which resubmits buffers to libusb
when client code has finished with them.

rtlsdr_read_async is rewritten in terms of the more general
rtlsdr_read_async_ex.

Significant documentation is added to explain how these functions relate
to latency.
---
 include/rtl-sdr.h | 82 +++++++++++++++++++++++++++++++++++++++++++++++++++++++
 src/librtlsdr.c   | 58 +++++++++++++++++++++++++++++++++++----
 2 files changed, 134 insertions(+), 6 deletions(-)

diff --git a/include/rtl-sdr.h b/include/rtl-sdr.h
index 3ed13ae..9b952fd 100644
--- a/include/rtl-sdr.h
+++ b/include/rtl-sdr.h
@@ -335,6 +335,17 @@ RTLSDR_API int rtlsdr_get_offset_tuning(rtlsdr_dev_t *dev);
 
 RTLSDR_API int rtlsdr_reset_buffer(rtlsdr_dev_t *dev);
 
+/*!
+ * Read samples from the device synchronously.  One request for data will be sent
+ * to the device and returned.  rtlsdr_read_async is preferred for streaming, as
+ * it provides for multiple buffers to be queued on the wire simultaneously.
+ *
+ * \param dev the device handle given by rtlsdr_open()
+ * \param buf buffer to be filled with samples (interleaved uint8_t IQ pairs)
+ * \param len length of buf in bytes
+ * \param n_read output location for number of bytes read
+ * \return 0 on success
+ */
 RTLSDR_API int rtlsdr_read_sync(rtlsdr_dev_t *dev, void *buf, int len, int *n_read);
 
 typedef void(*rtlsdr_read_async_cb_t)(unsigned char *buf, uint32_t len, void *ctx);
@@ -372,6 +383,77 @@ RTLSDR_API int rtlsdr_read_async(rtlsdr_dev_t *dev,
 				 uint32_t buf_num,
 				 uint32_t buf_len);
 
+typedef struct libusb_transfer * buf_handle_t;
+typedef void(*rtlsdr_read_async_ex_cb_t)(buf_handle_t release_hdl, unsigned char *buf, uint32_t len, void *ctx);
+
+/*!
+ * Read samples from the device asynchronously using long-lived DMA buffers.
+ * This function will block until it is canceled using rtlsdr_cancel_async().
+ *
+ * The memory buffers passed to the callback will continue to be valid, and may
+ * be read from in other threads, until released with rtlsdr_async_ex_release().
+ * Because buffers will not be reused until released, rtlsdr_async_ex_release()
+ * must be called within an average of buf_len / sample_rate seconds of every
+ * callback, or there will be gaps in the stream.
+ *
+ * Unlike rtlsdr_read_async(), rtlsdr_read_async_ex() will additionally call
+ * the callback in the case of failed USB transfers, passing len = 0.  These
+ * could indicate a gap in the stream.
+ *
+ * \param dev the device handle given by rtlsdr_open()
+ * \param cb callback function to return received samples.
+ *      each call will be in the same thread, in the order of receipt
+ * \param ctx user specific context to pass via the callback function
+ * \param buf_num optional total number of zerocopy buffers to maintain.
+ *      once these are exhausted, data will no longer stream until
+ *      they are passed to rtlsdr_async_ex_release() for reuse.
+ *		  set to 0 for default buffer count (15)
+ * \param buf_len optional buffer length, must be multiple of 512,
+ *		  should be a multiple of 16384 (URB size), set to 0
+ *		  for default buffer length (16 * 32 * 512)
+ * \return 0 on success
+ */
+RTLSDR_API int rtlsdr_read_async_ex(rtlsdr_dev_t *dev,
+                                    rtlsdr_read_async_ex_cb_t cb,
+                                    void *ctx,
+                                    uint32_t buf_num,
+                                    uint32_t buf_len);
+
+/*!
+ * Release a long-lived DMA buffer that was passed to a callback during a run
+ * of rtlsdr_read_async_ex().  Each zerocopy buffer passed to the callback has
+ * been reserved for that block of data, and will not be reused until this
+ * function is called to release it.
+ *
+ * There must be enough released buffers submitted to libusb to accommodate
+ * driver and hardware communication latency.  The total number of buffers
+ * filling this role, and the total number of buffers unreleased, can be
+ * calculated:
+ *
+ *   buf_num = value passed to rtlsdr_read_async_ex
+ *   callback_calls = number of times the callback function has been called
+ *   buf_releases = number of successful calls to rtlsdr_async_ex_release
+ *
+ *   bufs_reserved = callback_calls - buf_releases
+ *   bufs_in_flight = buf_num - bufs_reserved
+ *
+ * Note that this calculation assumes that all reserved buffers have been
+ * passed to the callback already; hence it is not be valid after making a
+ * blocking call inside the callback, as there may be more reserved buffers
+ * queued up that will be delivered after the callback returns.
+ *
+ * Once bufs_in_flight falls too low, there may be gaps in the data stream.
+ * The cutoff point for this value depends on latency in the operating system
+ * and hardware.
+ *
+ * Once bufs_in_flight hits zero, streaming will stop completely.
+ *
+ * \param dev the device handle given by rtlsdr_open()
+ * \param release_hdl the buffer passed to the callback
+ * \return 0 on success
+ */
+RTLSDR_API int rtlsdr_async_ex_release(rtlsdr_dev_t *dev, buf_handle_t release_hdl);
+
 /*!
  * Cancel all pending asynchronous operations on the device.
  *
diff --git a/src/librtlsdr.c b/src/librtlsdr.c
index 433ed5b..51cc3fc 100644
--- a/src/librtlsdr.c
+++ b/src/librtlsdr.c
@@ -99,7 +99,7 @@ struct rtlsdr_dev {
 	uint32_t xfer_buf_len;
 	struct libusb_transfer **xfer;
 	unsigned char **xfer_buf;
-	rtlsdr_read_async_cb_t cb;
+	rtlsdr_read_async_ex_cb_t cb;
 	void *cb_ctx;
 	enum rtlsdr_async_status async_status;
 	int async_cancel;
@@ -1697,14 +1697,16 @@ int rtlsdr_read_sync(rtlsdr_dev_t *dev, void *buf, int len, int *n_read)
 static void LIBUSB_CALL _libusb_callback(struct libusb_transfer *xfer)
 {
 	rtlsdr_dev_t *dev = (rtlsdr_dev_t *)xfer->user_data;
+	uint32_t passed_len;
 
-	if (LIBUSB_TRANSFER_COMPLETED == xfer->status) {
-		if (dev->cb)
-			dev->cb(xfer->buffer, xfer->actual_length, dev->cb_ctx);
+	if (xfer->status == LIBUSB_TRANSFER_CANCELLED)
+		return;
 
-		libusb_submit_transfer(xfer); /* resubmit transfer */
+	if (LIBUSB_TRANSFER_COMPLETED == xfer->status) {
 		dev->xfer_errors = 0;
-	} else if (LIBUSB_TRANSFER_CANCELLED != xfer->status) {
+
+		passed_len = xfer->actual_length;
+	} else {
 #ifndef _WIN32
 		if (LIBUSB_TRANSFER_ERROR == xfer->status)
 			dev->xfer_errors++;
@@ -1716,9 +1718,18 @@ static void LIBUSB_CALL _libusb_callback(struct libusb_transfer *xfer)
 			rtlsdr_cancel_async(dev);
 			fprintf(stderr, "cb transfer status: %d, "
 				"canceling...\n", xfer->status);
+
+      return;
 #ifndef _WIN32
 		}
 #endif
+		passed_len = 0;
+	}
+
+	if (dev->cb) {
+		dev->cb(xfer, xfer->buffer, passed_len, dev->cb_ctx);
+	} else {
+		rtlsdr_async_ex_release(dev, xfer);
 	}
 }
 
@@ -1830,8 +1841,38 @@ static int _rtlsdr_free_async_buffers(rtlsdr_dev_t *dev)
 	return 0;
 }
 
+struct rtlsdr_read_async_cb_data
+{
+	rtlsdr_dev_t * dev;
+	rtlsdr_read_async_cb_t cb;
+	void * ctx;
+};
+
+void rtlsdr_read_async_cb_wrapper(buf_handle_t release_hdl, unsigned char *buf,
+				  uint32_t len, void *ctx)
+{
+	struct rtlsdr_read_async_cb_data * data = ctx;
+
+	if (data->cb && len > 0) {
+		data->cb(buf, len, data->ctx);
+	}
+
+	rtlsdr_async_ex_release(data->dev, release_hdl);
+}
+
 int rtlsdr_read_async(rtlsdr_dev_t *dev, rtlsdr_read_async_cb_t cb, void *ctx,
 			  uint32_t buf_num, uint32_t buf_len)
+{
+	struct rtlsdr_read_async_cb_data data;
+	data.dev = dev;
+	data.cb = cb;
+	data.ctx = ctx;
+
+	return rtlsdr_read_async_ex(dev, rtlsdr_read_async_cb_wrapper, &data,
+				    buf_num, buf_len);
+}
+int rtlsdr_read_async_ex(rtlsdr_dev_t *dev, rtlsdr_read_async_ex_cb_t cb, void *ctx,
+			 uint32_t buf_num, uint32_t buf_len)
 {
 	unsigned int i;
 	int r = 0;
@@ -1939,6 +1980,11 @@ int rtlsdr_read_async(rtlsdr_dev_t *dev, rtlsdr_read_async_cb_t cb, void *ctx,
 	return r;
 }
 
+int rtlsdr_async_ex_release(rtlsdr_dev_t *dev, buf_handle_t release_hdl)
+{
+	return libusb_submit_transfer(release_hdl); /* resubmit transfer */
+}
+
 int rtlsdr_cancel_async(rtlsdr_dev_t *dev)
 {
 	if (!dev)
-- 
2.15.1

