Add support for more precision on buffer lengths; necessary for high bandwidth streams where 1 sec is too long.

Author: cboulay

include/lsl/common.h

@@ -150,6 +150,21 @@ typedef enum {
 	_lsl_error_code_maxval = 0x7f000000
 } lsl_error_code_t;
 
+/// Flags for outlet_ex and inlet_ex
+typedef enum {
+	/// Keep legacy behavior: max_buffered / max_buflen is in seconds; use asynch transfer.
+	transp_default = 0,
+
+	/// The supplied max_buf value is in samples.
+	transp_bufsize_samples = 1,
+
+	/// The supplied max_buf should be scaled by 0.001.
+	transp_bufsize_thousandths = 2,
+
+	// prevent compilers from assuming an instance fits in a single byte
+	_lsl_transport_options_maxval = 0x7f000000
+} lsl_transport_options_t;
+
 /// Return an explanation for the last error
 extern LIBLSL_C_API const char *lsl_last_error(void);
 

include/lsl/inlet.h

@@ -39,6 +39,12 @@
  * @return A newly created lsl_inlet handle or NULL in the event that an error occurred.
  */
 extern LIBLSL_C_API lsl_inlet lsl_create_inlet(lsl_streaminfo info, int32_t max_buflen, int32_t max_chunklen, int32_t recover);
+/** @copydoc lsl_create_inlet()
+ * @param flags An integer that is the result of bitwise OR'ing one or more options from
+ * #lsl_transport_options_t together (e.g., #transp_bufsize_samples)
+ */
+extern LIBLSL_C_API lsl_inlet lsl_create_inlet_ex(lsl_streaminfo info, int32_t max_buflen,
+	int32_t max_chunklen, int32_t recover, lsl_transport_options_t flags);
 
 /**
 * Destructor.

include/lsl/outlet.h

@@ -34,6 +34,12 @@
  * @return A newly created lsl_outlet handle or NULL in the event that an error occurred.
  */
 extern LIBLSL_C_API lsl_outlet lsl_create_outlet(lsl_streaminfo info, int32_t chunk_size, int32_t max_buffered);
+/** @copydoc lsl_create_outlet()
+ * @param flags An integer that is the result of bitwise OR'ing one or more options from
+ * #lsl_transport_options_t together (e.g., #transp_bufsize_samples|#transp_bufsize_thousandths)
+ */
+extern LIBLSL_C_API lsl_outlet lsl_create_outlet_ex(
+	lsl_streaminfo info, int32_t chunk_size, int32_t max_buffered, lsl_transport_options_t flags);
 
 /**
  * Destroy an outlet.
@@ -93,7 +99,8 @@ extern LIBLSL_C_API int32_t lsl_push_sample_vtp(lsl_outlet out, const void *data
 /** @copybrief lsl_push_sample_ftp
  * @see lsl_push_sample_ftp
  * @param out The lsl_outlet object through which to push the data.
- * @param data A pointer to values to push. The number of values pointed to must be no less than the number of channels in the sample.
+ * @param data A pointer to values to push. The number of values pointed to must be no less than the
+ * number of channels in the sample.
  * @param lengths A pointer the number of elements to push for each channel (string lengths).
  */
 extern LIBLSL_C_API int32_t lsl_push_sample_buf(lsl_outlet out, const char **data, const uint32_t *lengths);

include/lsl_cpp.h

@@ -396,10 +396,11 @@ class stream_outlet {
 	 * @param max_buffered Optionally the maximum amount of data to buffer (in seconds if there is a
 	 * nominal sampling rate, otherwise x100 in samples). The default is 6 minutes of data.
 	 */
-	stream_outlet(const stream_info &info, int32_t chunk_size = 0, int32_t max_buffered = 360)
-		: channel_count(info.channel_count()),
-		  sample_rate(info.nominal_srate()),
-		  obj(lsl_create_outlet(info.handle().get(), chunk_size, max_buffered), &lsl_destroy_outlet) {}
+	stream_outlet(const stream_info &info, int32_t chunk_size = 0, int32_t max_buffered = 360,
+		lsl_transport_options_t flags = transp_default)
+		: channel_count(info.channel_count()), sample_rate(info.nominal_srate()),
+		  obj(lsl_create_outlet_ex(info.handle().get(), chunk_size, max_buffered, flags),
+			  &lsl_destroy_outlet) {}
 
 	// ========================================
 	// === Pushing a sample into the outlet ===
@@ -900,9 +901,10 @@ class stream_inlet {
 	 * lsl::lost_error if the stream's source is lost (e.g., due to an app or computer crash).
 	 */
 	stream_inlet(const stream_info &info, int32_t max_buflen = 360, int32_t max_chunklen = 0,
-		bool recover = true)
+		bool recover = true, lsl_transport_options_t flags = transp_default)
 		: channel_count(info.channel_count()),
-		  obj(lsl_create_inlet(info.handle().get(), max_buflen, max_chunklen, recover), &lsl_destroy_inlet) {}
+		  obj(lsl_create_inlet_ex(info.handle().get(), max_buflen, max_chunklen, recover, flags),
+			  &lsl_destroy_inlet) {}
 
 	/// Return a shared pointer to pass to C-API functions that aren't wrapped yet
 	///

src/lsl_inlet_c.cpp

@@ -14,11 +14,19 @@ extern "C" {
 
 using namespace lsl;
 
+LIBLSL_C_API lsl_inlet lsl_create_inlet_ex(lsl_streaminfo info, int32_t max_buflen,
+	int32_t max_chunklen, int32_t recover, lsl_transport_options_t flags) {
+	try {
+		int32_t buf_samples = info->calc_transport_buf_samples(max_buflen, flags);
+		return create_object_noexcept<stream_inlet_impl>(
+			*info, buf_samples, max_chunklen, recover != 0);
+	}
+	LSLCATCHANDSTORE(nullptr, std::invalid_argument, lsl_argument_error);
+	return nullptr;
+}
 LIBLSL_C_API lsl_inlet lsl_create_inlet(
 	lsl_streaminfo info, int32_t max_buflen, int32_t max_chunklen, int32_t recover) {
-	return create_object_noexcept<stream_inlet_impl>(*info,
-		(info->nominal_srate() ? (int)(info->nominal_srate() * max_buflen) : max_buflen * 100) + 1,
-		max_chunklen, recover != 0);
+	return lsl_create_inlet_ex(info, max_buflen, max_chunklen, recover, transp_default);
 }
 
 LIBLSL_C_API void lsl_destroy_inlet(lsl_inlet in) {

src/lsl_outlet_c.cpp

@@ -15,12 +15,14 @@ extern "C" {
 using namespace lsl;
 
 // boilerplate wrapper code
+LIBLSL_C_API lsl_outlet lsl_create_outlet_ex(
+	lsl_streaminfo info, int32_t chunk_size, int32_t max_buffered, lsl_transport_options_t flags) {
+	return create_object_noexcept<stream_outlet_impl>(*info, chunk_size, max_buffered, flags);
+}
+
 LIBLSL_C_API lsl_outlet lsl_create_outlet(
 	lsl_streaminfo info, int32_t chunk_size, int32_t max_buffered) {
-	double buftime = info->nominal_srate();
-	if (buftime <= 0) buftime = 100;
-	return create_object_noexcept<stream_outlet_impl>(
-		*info, chunk_size, static_cast<int>(buftime * max_buffered));
+	return lsl_create_outlet_ex(info, chunk_size, max_buffered, transp_default);
 }
 
 LIBLSL_C_API void lsl_destroy_outlet(lsl_outlet out) {

src/stream_info_impl.cpp

@@ -244,6 +244,24 @@ int stream_info_impl::channel_bytes() const {
 xml_node stream_info_impl::desc() { return doc_.child("info").child("desc"); }
 xml_node stream_info_impl::desc() const { return doc_.child("info").child("desc"); }
 
+uint32_t lsl::stream_info_impl::calc_transport_buf_samples(
+	int32_t requested_len, lsl_transport_options_t flags) const {
+	if ((flags & transp_bufsize_samples) && (flags & transp_bufsize_thousandths))
+		throw std::invalid_argument(
+			"transp_bufsize_samples and transp_bufsize_thousandths are mutually exclusive");
+
+	int32_t buf_samples;
+	if (flags & transp_bufsize_samples)
+		buf_samples = requested_len;
+	else if (nominal_srate() == LSL_IRREGULAR_RATE)
+		buf_samples = requested_len * 100;
+	else
+		buf_samples = nominal_srate() * requested_len;
+	if (flags & transp_bufsize_thousandths) buf_samples /= 1000;
+	buf_samples = (buf_samples > 0) ? buf_samples : 1;
+	return buf_samples;
+}
+
 void stream_info_impl::version(int v) {
 	version_ = v;
 	doc_.child("info").child("version").first_child().set_value(to_string(version_ / 100.).c_str());

src/stream_info_impl.h

@@ -221,6 +221,9 @@ class stream_info_impl {
 	pugi::xml_node desc();
 	pugi::xml_node desc() const;
 
+	/// helper function to calculate the buffer size in samples for inlets and outlets
+	uint32_t calc_transport_buf_samples(int32_t requested_len, lsl_transport_options_t flags) const;
+
 protected:
 	/// Create and assign the XML DOM structure based on the class fields.
 	void write_xml(pugi::xml_document &doc);

src/stream_outlet_impl.cpp

@@ -11,16 +11,17 @@
 
 namespace lsl {
 
-stream_outlet_impl::stream_outlet_impl(
-	const stream_info_impl &info, int32_t chunk_size, int32_t max_capacity)
+stream_outlet_impl::stream_outlet_impl(const stream_info_impl &info, int32_t chunk_size,
+	int32_t requested_bufsize, lsl_transport_options_t flags)
 	: sample_factory_(std::make_shared<factory>(info.channel_format(), info.channel_count(),
 		  static_cast<uint32_t>(
 			  info.nominal_srate()
 				  ? info.nominal_srate() * api_config::get_instance()->outlet_buffer_reserve_ms() /
 						1000
 				  : api_config::get_instance()->outlet_buffer_reserve_samples()))),
-	  chunk_size_(chunk_size), info_(std::make_shared<stream_info_impl>(info)),
-	  send_buffer_(std::make_shared<send_buffer>(max_capacity)),
+	  chunk_size_(info.calc_transport_buf_samples(requested_bufsize, flags)),
+	  info_(std::make_shared<stream_info_impl>(info)),
+	  send_buffer_(std::make_shared<send_buffer>(chunk_size_)),
 	  io_ctx_data_(std::make_shared<asio::io_context>(1)),
 	  io_ctx_service_(std::make_shared<asio::io_context>(1)) {
 	ensure_lsl_initialized();

src/stream_outlet_impl.h

@@ -32,12 +32,14 @@ class stream_outlet_impl {
 	 * @param chunk_size The preferred chunk size, in samples, at which data shall be transmitted
 	 * over the network. Can be selectively overridden by the inlet. If 0 (=default), the chunk size
 	 * is determined by the pushthrough flag in push_sample or push_chunk.
-	 * @param max_capacity The maximum number of samples buffered for unresponsive receivers. If
-	 * more samples get pushed, the oldest will be dropped. The default is sufficient to hold a bit
-	 * more than 15 minutes of data at 512Hz, while consuming not more than ca. 512MB of RAM.
+	 * @param requested_size The maximum number of seconds/samples buffered for unresponsive
+	 * receivers. If more samples get pushed, the oldest will be dropped. The default is sufficient
+	 * to hold a bit more than 15 minutes of data at 512Hz, while consuming not more than ca. 512MB
+	 * of RAM. Depends on `flags` as calculated in `stream_info_impl::calc_transport_buf_samples()`
+	 * @param flags Bitwise-OR'd flags from lsl_transport_options_t
 	 */
-	stream_outlet_impl(
-		const stream_info_impl &info, int32_t chunk_size = 0, int32_t max_capacity = 512000);
+	stream_outlet_impl(const stream_info_impl &info, int32_t chunk_size = 0,
+		int32_t requested_bufsize = 900, lsl_transport_options_t flags = transp_default);
 
 	/**
 	 * Destructor.