Implement synchronous outlet for zero-copy writes.

Author: cboulay

examples/ReceiveDataInChunks.cpp

@@ -18,11 +18,12 @@ int main(int argc, char **argv) {
 	try {
 
 		std::string name{argc > 1 ? argv[1] : "MyAudioStream"};
-		double max_buffered = argc > 2 ? std::stod(argv[2]) : 360.;
+		double max_buflen = argc > 2 ? std::stod(argv[2]) : 360.;
 		bool flush = argc > 3;
 		// resolve the stream of interest & make an inlet
+		int32_t buf_samples = (int32_t)(max_buflen * 1000);
 		lsl::stream_info inlet_info = lsl::resolve_stream("name", name).at(0);
-		lsl::stream_inlet inlet(inlet_info, (int32_t)max_buffered);
+		lsl::stream_inlet inlet(inlet_info, max_buflen, transp_bufsize_thousandths);
 
 		// Use set_postprocessing to get the timestamps in a common base clock.
 		// Do not use if this application will record timestamps to disk -- it is better to 

examples/SendDataInChunks.cpp

@@ -92,15 +92,24 @@ struct fake_device {
 
 int main(int argc, char **argv) {
 	std::cout << "SendDataInChunks" << std::endl;
-	std::cout << "SendDataInChunks StreamName StreamType samplerate n_channels max_buffered chunk_rate" << std::endl;
+	std::cout << "SendDataInChunks StreamName StreamType samplerate n_channels max_buffered "
+				 "chunk_rate nodata use_sync"
+			  << std::endl;
 	std::cout << "- max_buffered -- duration in sec (or x100 samples if samplerate is 0) to buffer for each outlet" << std::endl;
 	std::cout << "- chunk_rate -- number of chunks pushed per second. For this example, make it a common factor of samplingrate and 1000." << std::endl;
-	
+	std::cout << "- nodata -- Set non-zero to cause the fake device to not copy pattern data into "
+				 "the buffer."
+			  << std::endl;
+	std::cout << "- use_sync -- Set to non-zero to use blocking send." << std::endl;
+
 	std::string name{argc > 1 ? argv[1] : "MyAudioStream"}, type{argc > 2 ? argv[2] : "Audio"};
 	int samplingrate = argc > 3 ? std::stol(argv[3]) : 44100;  // Here we specify srate, but typically this would come from the device.
 	int n_channels = argc > 4 ? std::stol(argv[4]) : 2;        // Here we specify n_chans, but typically this would come from theh device.
 	double max_buffered = argc > 5 ? std::stod(argv[5]) : 360.;
 	int32_t chunk_rate = argc > 6 ? std::stol(argv[6]) : 10;  // Chunks per second.
+	bool nodata = argc > 7;
+	bool do_sync = argc > 8 ? (bool)std::stol(argv[8]) : true;
+
 	int32_t chunk_samples = samplingrate > 0 ? std::max((samplingrate / chunk_rate), 1) : 100;  // Samples per chunk.
 	int32_t chunk_duration = 1000 / chunk_rate;  // Milliseconds per chunk
 
@@ -118,7 +127,9 @@ int main(int argc, char **argv) {
 			chn.append_child_value("type", type);
 		}
 		int32_t buf_samples = (int32_t)(max_buffered * samplingrate);
-		lsl::stream_outlet outlet(info, chunk_samples, buf_samples);
+		auto flags = static_cast<lsl_transport_options_t>(
+			(do_sync ? transp_sync_blocking : transp_default) | transp_bufsize_samples);
+		lsl::stream_outlet outlet(info, chunk_samples, buf_samples, flags);
 		info = outlet.info(); // Refresh info with whatever the outlet captured.
 		std::cout << "Stream UID: " << info.uid() << std::endl;
 
@@ -128,6 +139,7 @@ int main(int argc, char **argv) {
 		// Prepare buffer to get data from 'device'.
 		//  The buffer should be larger than you think you need. Here we make it 4x as large.
 		std::vector<int16_t> chunk_buffer(4 * chunk_samples * n_channels);
+		std::fill(chunk_buffer.begin(), chunk_buffer.end(), 0);
 
 		std::cout << "Now sending data..." << std::endl;
 
@@ -141,7 +153,7 @@ int main(int argc, char **argv) {
 			std::this_thread::sleep_until(next_chunk_time);
 
 			// Get data from device
-			std::size_t returned_samples = my_device.get_data(chunk_buffer);
+			std::size_t returned_samples = my_device.get_data(chunk_buffer, nodata);
 
 			// send it to the outlet. push_chunk_multiplexed is one of the more complicated approaches.
 			//  other push_chunk methods are easier but slightly slower.

include/lsl/common.h

@@ -161,6 +161,9 @@ typedef enum {
 	/// The supplied max_buf should be scaled by 0.001.
 	transp_bufsize_thousandths = 2,
 
+	/// The outlet will use synchronous (blocking) calls to asio to push data
+	transp_sync_blocking = 4,
+
 	// prevent compilers from assuming an instance fits in a single byte
 	_lsl_transport_options_maxval = 0x7f000000
 } lsl_transport_options_t;

src/stream_outlet_impl.cpp

@@ -22,8 +22,14 @@ stream_outlet_impl::stream_outlet_impl(const stream_info_impl &info, int32_t chu
 	  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)),
+	  do_sync_(flags & transp_sync_blocking), io_ctx_data_(std::make_shared<asio::io_context>(1)),
 	  io_ctx_service_(std::make_shared<asio::io_context>(1)) {
+
+	if ((info.channel_format() == cft_string) && (flags & transp_sync_blocking)) {
+		LOG_F(WARNING, "sync push not supported for string-formatted streams. Reverting to async.");
+		do_sync_ = false;
+	}
+
 	ensure_lsl_initialized();
 	const api_config *cfg = api_config::get_instance();
 
@@ -147,8 +153,22 @@ void stream_outlet_impl::push_numeric_raw(const void *data, double timestamp, bo
 	if (lsl::api_config::get_instance()->force_default_timestamps()) timestamp = 0.0;
 	sample_p smp(
 		sample_factory_->new_sample(timestamp == 0.0 ? lsl_clock() : timestamp, pushthrough));
-	smp->assign_untyped(data);
-	send_buffer_->push_sample(smp);
+	if (!do_sync_) {
+		smp->assign_untyped(data); // Note: Makes a copy!
+		send_buffer_->push_sample(smp);
+	} else {
+		if (timestamp == DEDUCED_TIMESTAMP) {
+			sync_buffs_.push_back(asio::buffer(&TAG_DEDUCED_TIMESTAMP, 1));
+		} else {
+			sync_buffs_.push_back(asio::buffer(&TAG_TRANSMITTED_TIMESTAMP, 1));
+			sync_buffs_.push_back(asio::buffer(&timestamp, sizeof(timestamp)));
+		}
+		sync_buffs_.push_back(asio::buffer(data, smp->datasize()));
+		if (pushthrough) {
+			tcp_server_->write_all_blocking(sync_buffs_);
+			sync_buffs_.clear();
+		}
+	}
 }
 
 bool stream_outlet_impl::have_consumers() { return send_buffer_->have_consumers(); }
@@ -162,8 +182,22 @@ void stream_outlet_impl::enqueue(const T *data, double timestamp, bool pushthrou
 	if (lsl::api_config::get_instance()->force_default_timestamps()) timestamp = 0.0;
 	sample_p smp(
 		sample_factory_->new_sample(timestamp == 0.0 ? lsl_clock() : timestamp, pushthrough));
-	smp->assign_typed(data);
-	send_buffer_->push_sample(smp);
+	if (!do_sync_) {
+		smp->assign_typed(data);
+		send_buffer_->push_sample(smp);
+	} else {
+		if (timestamp == DEDUCED_TIMESTAMP) {
+			sync_buffs_.push_back(asio::buffer(&TAG_DEDUCED_TIMESTAMP, 1));
+		} else {
+			sync_buffs_.push_back(asio::buffer(&TAG_TRANSMITTED_TIMESTAMP, 1));
+			sync_buffs_.push_back(asio::buffer(&timestamp, sizeof(timestamp)));
+		}
+		sync_buffs_.push_back(asio::buffer(data, smp->datasize()));
+		if (pushthrough) {
+			tcp_server_->write_all_blocking(sync_buffs_);
+			sync_buffs_.clear();
+		}
+	}
 }
 
 template void stream_outlet_impl::enqueue<char>(const char *data, double, bool);

src/stream_outlet_impl.h

@@ -4,6 +4,7 @@
 #include "common.h"
 #include "forward.h"
 #include "stream_info_impl.h"
+#include <asio/buffer.hpp>
 #include <cstdint>
 #include <loguru.hpp>
 #include <memory>
@@ -319,6 +320,8 @@ class stream_outlet_impl {
 	stream_info_impl_p info_;
 	/// the single-producer, multiple-receiver send buffer
 	send_buffer_p send_buffer_;
+	/// Flag to indicate that push_* operations should be blocking synchronous. false by default.
+	bool do_sync_;
 	/// the IO service objects
 	io_context_p io_ctx_data_, io_ctx_service_;
 
@@ -331,6 +334,8 @@ class stream_outlet_impl {
 	std::vector<udp_server_p> responders_;
 	/// threads that handle the I/O operations (two per stack: one for UDP and one for TCP)
 	std::vector<thread_p> io_threads_;
+	/// buffers used in synchronous call to gather-write data directly to the socket.
+	std::vector<asio::const_buffer> sync_buffs_;
 };
 
 } // namespace lsl

src/tcp_server.cpp

@@ -143,9 +143,9 @@ class client_session : public std::enable_shared_from_this<client_session> {
 };
 
 tcp_server::tcp_server(stream_info_impl_p info, io_context_p io, send_buffer_p sendbuf,
-	factory_p factory, int chunk_size, bool allow_v4, bool allow_v6)
+	factory_p factory, int chunk_size, bool allow_v4, bool allow_v6, bool do_sync)
 	: chunk_size_(chunk_size), info_(std::move(info)), io_(std::move(io)),
-	  factory_(std::move(factory)), send_buffer_(std::move(sendbuf)) {
+	  factory_(std::move(factory)), send_buffer_(std::move(sendbuf)), transfer_is_sync_(do_sync) {
 	// assign connection-dependent fields
 	info_->session_id(api_config::get_instance()->session_id());
 	info_->reset_uid();
@@ -232,7 +232,47 @@ void tcp_server::accept_next_connection(tcp_acceptor_p &acceptor) {
 }
 
 
-// === graceful cancellation of in-flight sockets ===
+// === synchronous transfer
+
+void tcp_server::write_all_blocking(std::vector<asio::const_buffer> bufs) {
+	int writes_outstanding = 0;
+	bool any_session_broken = false;
+
+	for (auto &sock : sync_sockets_) {
+		asio::async_write(*sock, bufs,
+			[this, sock, &writes_outstanding](
+				const asio::error_code &ec, size_t bytes_transferred) {
+				writes_outstanding--;
+				switch (ec.value()) {
+				case 0: break; // success
+				case asio::error::broken_pipe:
+				case asio::error::connection_reset:
+					LOG_F(WARNING, "Broken Pipe / Connection Reset detected. Closing socket.");
+					{
+						asio::error_code close_ec;
+						sock->close(close_ec);
+					}
+					break;
+				default:
+					LOG_F(WARNING, "Unhandled write_all_blocking error: %s.", ec.message().c_str());
+				}
+			});
+		writes_outstanding++;
+	}
+	try {
+		assert(sync_transfer_io_ctx_);
+		sync_transfer_io_ctx_->restart();
+		while (writes_outstanding) sync_transfer_io_ctx_->run_one();
+		if (any_session_broken) {
+			// remove sessions whose socket was closed
+			auto new_end_it = std::remove_if(sync_sockets_.begin(), sync_sockets_.end(),
+				[](const tcp_socket_p &sock) {
+					return !sock->is_open();
+				});
+			sync_sockets_.erase(new_end_it, sync_sockets_.end());
+		}
+	} catch (std::exception &e) { LOG_F(ERROR, "Error during write_all_blocking: %s", e.what()); }
+}
 
 void tcp_server::register_inflight_session(const std::shared_ptr<client_session> &session) {
 	std::lock_guard<std::recursive_mutex> lock(inflight_mut_);
@@ -535,6 +575,16 @@ void client_session::handle_send_feedheader_outcome(err_t err, std::size_t n) {
 		// convenient for unit tests
 		if (max_buffered_ <= 0) return;
 
+		if (serv->transfer_is_sync_) {
+			LOG_F(INFO, "Using synchronous blocking transfers for new client session.");
+			asio::post(*serv->sync_transfer_io_ctx_,
+				[serv, sock_p = std::make_shared<tcp_socket>(std::move(sock_))]() {
+					serv->sync_sockets_.emplace_back(std::move(sock_p));
+				});
+			serv->unregister_inflight_session(this);
+			return;
+		}
+
 		// determine transfer parameters
 		auto queue = serv->send_buffer_->new_consumer(max_buffered_);
 

src/tcp_server.h

@@ -47,9 +47,11 @@ class tcp_server : public std::enable_shared_from_this<tcp_server> {
 	 * @param protocol The protocol (IPv4 or IPv6) that shall be serviced by this server.
 	 * @param chunk_size The preferred chunk size, in samples. If 0, the pushthrough flag determines
 	 * the effective chunking.
+	 * @param do_sync Set true to indicate data transfer should happen synchronously in a blocking
+	 * call. Default false -- asynchronous transfer in a thread (copies data).
 	 */
 	tcp_server(stream_info_impl_p info, io_context_p io, send_buffer_p sendbuf, factory_p factory,
-		int chunk_size, bool allow_v4, bool allow_v6);
+		int chunk_size, bool allow_v4, bool allow_v6, bool do_sync = false);
 
 	/**
 	 * Begin serving TCP connections.
@@ -67,6 +69,12 @@ class tcp_server : public std::enable_shared_from_this<tcp_server> {
 	 */
 	void end_serving();
 
+	/**
+	 * Write directly to each socket. This should only be used when server initialized with
+	 * do_async = false.
+	 */
+	void write_all_blocking(std::vector<asio::const_buffer> buffs);
+
 private:
 	friend class client_session;
 
@@ -95,6 +103,16 @@ class tcp_server : public std::enable_shared_from_this<tcp_server> {
 	// acceptor socket
 	tcp_acceptor_p acceptor_v4_, acceptor_v6_; // our server socket
 
+
+	// sync mode fields
+
+	// Flag to indicate that new client_sessions should use synchronous blocking data transfer.
+	bool transfer_is_sync_;
+	// sockets that should receive data in sync mode
+	std::vector<tcp_socket_p> sync_sockets_;
+	// io context for sync mode, app is responsible for running it
+	std::unique_ptr<asio::io_context> sync_transfer_io_ctx_;
+
 	// registry of in-flight asessions (for cancellation)
 	std::map<void *, std::weak_ptr<client_session>> inflight_;
 	std::recursive_mutex inflight_mut_; // mutex protecting the registry from concurrent access