Implement synchronous outlet for zero-copy writes.

cboulay · cboulay · commit c64aeb9a05cd · 2021-10-04T15:42:54.000-04:00
diff --git a/examples/SendDataInChunks.cpp b/examples/SendDataInChunks.cpp
@@ -91,24 +91,25 @@ 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 nodata" << std::endl;
+	std::cout << "SendDataInChunks StreamName StreamType samplerate n_channels max_buffered chunk_rate nodata do_async" << 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 << "- do_async -- Set to 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_async = 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
 
 	try {
 		// Prepare the LSL stream.
-		lsl::stream_info info(name, type, n_channels, samplingrate, lsl::cf_int16);
-		lsl::stream_outlet outlet(info, chunk_samples, max_buffered);
+		lsl::stream_info info(name, type, n_channels, samplingrate, lsl::cf_int16, "example-SendDataInChunks");
 		lsl::xml_element desc = info.desc();
 		desc.append_child_value("manufacturer", "LSL");
 		lsl::xml_element chns = desc.append_child("channels");
@@ -118,6 +119,9 @@ int main(int argc, char **argv) {
 			chn.append_child_value("unit", "microvolts");
 			chn.append_child_value("type", "EEG");
 		}
+		lsl::stream_outlet outlet(info, chunk_samples, max_buffered, do_async);
+		info = outlet.info();  // Refresh info with whatever the outlet captured.
+		std::cout << "Stream UID: " << info.uid() << std::endl;
 
 		// Create a connection to our device.
 		fake_device my_device(n_channels, (float)samplingrate);
@@ -142,7 +146,8 @@ int main(int argc, char **argv) {
 
 			// send it to the outlet. push_chunk_multiplexed is one of the more complicated approaches.
 			//  other push_chunk methods are easier but slightly slower.
-			outlet.push_chunk_multiplexed(chunk_buffer.data(), returned_samples * n_channels, 0.0, false);
+			double ts = lsl::local_clock();
+			outlet.push_chunk_multiplexed(chunk_buffer.data(), returned_samples * n_channels, ts, true);
 		}
 
 	} catch (std::exception &e) { std::cerr << "Got an exception: " << e.what() << std::endl; }
diff --git a/include/lsl/outlet.h b/include/lsl/outlet.h
@@ -31,11 +31,14 @@
  * nominal  sampling rate, otherwise x100 in samples). A good default is 360, which corresponds to 6
  * minutes of data. Note that, for high-bandwidth data you will almost certainly want to use a lower
  * value here to avoid  running out of RAM.
+ * @param do_async Set to 1 to preserve previous behaviour of using asynchronous operations,
+ * or set to 0 to use new synchronous methods which block but are overall faster due to fewer
+ * copies.
  * @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);
 extern LIBLSL_C_API lsl_outlet lsl_create_outlet_d(
-	lsl_streaminfo info, int32_t chunk_size, double max_buffered);
+	lsl_streaminfo info, int32_t chunk_size, double max_buffered, int do_async);
 
 /**
  * Destroy an outlet.
diff --git a/include/lsl_cpp.h b/include/lsl_cpp.h
@@ -400,9 +400,9 @@ class stream_outlet {
 		: 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, double max_buffered)
+	stream_outlet(const stream_info &info, int32_t chunk_size, double max_buffered, bool do_async = true)
 		: channel_count(info.channel_count()), sample_rate(info.nominal_srate()),
-		  obj(lsl_create_outlet_d(info.handle().get(), chunk_size, max_buffered),
+		  obj(lsl_create_outlet_d(info.handle().get(), chunk_size, max_buffered, do_async),
 			  &lsl_destroy_outlet) {}
 
 	// ========================================
diff --git a/src/lsl_outlet_c.cpp b/src/lsl_outlet_c.cpp
@@ -16,17 +16,17 @@ using namespace lsl;
 
 // boilerplate wrapper code
 LIBLSL_C_API lsl_outlet lsl_create_outlet_d(
-	lsl_streaminfo info, int32_t chunk_size, double max_buffered) {
+	lsl_streaminfo info, int32_t chunk_size, double max_buffered, int do_async) {
 	double buftime = info->nominal_srate();
 	if (buftime <= 0) buftime = 100;
 	if ((buftime * max_buffered) < 1) max_buffered = 1.5 / buftime;
 	return create_object_noexcept<stream_outlet_impl>(
-		*info, chunk_size, static_cast<int>(buftime * max_buffered));
+		*info, chunk_size, static_cast<int>(buftime * max_buffered), (bool)do_async);
 }
 
 LIBLSL_C_API lsl_outlet lsl_create_outlet(
 	lsl_streaminfo info, int32_t chunk_size, int32_t max_buffered) {
-	return lsl_create_outlet_d(info, chunk_size, (double)max_buffered);
+	return lsl_create_outlet_d(info, chunk_size, (double)max_buffered, 1);
 }
 
 LIBLSL_C_API void lsl_destroy_outlet(lsl_outlet out) {
diff --git a/src/stream_outlet_impl.cpp b/src/stream_outlet_impl.cpp
@@ -12,15 +12,18 @@
 namespace lsl {
 
 stream_outlet_impl::stream_outlet_impl(
-	const stream_info_impl &info, int32_t chunk_size, int32_t max_capacity)
+	const stream_info_impl &info, int32_t chunk_size, int32_t max_capacity, bool do_async)
 	: 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)) {
+	  send_buffer_(std::make_shared<send_buffer>(max_capacity)),
+      do_async_(do_async) {
+	// TODO: assert do_async_ or info.channel_format() != cft_string,
+	//  no-copy blocking write not supported for variable-length strings.
 	ensure_lsl_initialized();
 	const api_config *cfg = api_config::get_instance();
 
@@ -71,7 +74,7 @@ void stream_outlet_impl::instantiate_stack(tcp tcp_protocol, udp udp_protocol) {
 	// create TCP data server
 	ios_.push_back(std::make_shared<asio::io_context>());
 	tcp_servers_.push_back(std::make_shared<tcp_server>(
-		info_, ios_.back(), send_buffer_, sample_factory_, tcp_protocol, chunk_size_));
+		info_, ios_.back(), send_buffer_, sample_factory_, tcp_protocol, chunk_size_, do_async_));
 	// create UDP time server
 	ios_.push_back(std::make_shared<asio::io_context>());
 	udp_servers_.push_back(std::make_shared<udp_server>(info_, *ios_.back(), udp_protocol));
@@ -143,8 +146,23 @@ 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_async_) {
+		smp->assign_untyped(data);
+		send_buffer_->push_sample(smp);
+	} else {
+		// TODO: Move the following into a common method
+		if (timestamp == DEDUCED_TIMESTAMP) {
+			sync_buffs_.push_back(lslboost::asio::buffer(&TAG_DEDUCED_TIMESTAMP, 1));
+		} else {
+			sync_buffs_.push_back(lslboost::asio::buffer(&TAG_TRANSMITTED_TIMESTAMP, 1));
+			sync_buffs_.push_back(lslboost::asio::buffer(&timestamp, sizeof(timestamp)));
+		}
+		sync_buffs_.push_back(lslboost::asio::buffer(data, smp->datasize()));
+		if (pushthrough) {
+			for (auto &tcp_server : tcp_servers_)
+				tcp_server->write_all_blocking(sync_buffs_);
+		}
+	}
 }
 
 bool stream_outlet_impl::have_consumers() { return send_buffer_->have_consumers(); }
@@ -158,8 +176,24 @@ 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_async_) {
+		smp->assign_typed(data);
+		send_buffer_->push_sample(smp);
+	} else {
+		// TODO: Move the following into a common method
+		if (timestamp == DEDUCED_TIMESTAMP) {
+			sync_buffs_.push_back(lslboost::asio::buffer(&TAG_DEDUCED_TIMESTAMP, 1));
+		} else {
+			sync_buffs_.push_back(lslboost::asio::buffer(&TAG_TRANSMITTED_TIMESTAMP, 1));
+			sync_buffs_.push_back(lslboost::asio::buffer(&timestamp, sizeof(timestamp)));
+		}
+		sync_buffs_.push_back(lslboost::asio::buffer(data, smp->datasize()));
+		if (pushthrough) {
+			for (auto &tcp_server : tcp_servers_)
+				tcp_server->write_all_blocking(sync_buffs_);
+			sync_buffs_.clear();
+		}
+	}
 }
 
 template void stream_outlet_impl::enqueue<char>(const char *data, double, bool);
diff --git a/src/stream_outlet_impl.h b/src/stream_outlet_impl.h
@@ -10,6 +10,7 @@
 #include <string>
 #include <thread>
 #include <vector>
+#include <boost/asio/buffer.hpp>
 
 using asio::ip::tcp;
 using asio::ip::udp;
@@ -35,9 +36,13 @@ class stream_outlet_impl {
 	 * @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 do_async If true (default) then send operations will happen in a secondary thread.
+	 * Set false to force synchronous-only transfers. This will slow down calls to `push_*` but
+	 * makes fewer data copies and can be overall much faster for high bandwidth data. This feature
+	 * is new and relatively untested; use at your own risk.
 	 */
 	stream_outlet_impl(
-		const stream_info_impl &info, int32_t chunk_size = 0, int32_t max_capacity = 512000);
+		const stream_info_impl &info, int32_t chunk_size = 0, int32_t max_capacity = 512000, bool do_async = true);
 
 	/**
 	 * Destructor.
@@ -317,6 +322,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 asynchronous. true by default.
+	bool do_async_;
 	/// the IO service objects (two per stack: one for UDP and one for TCP)
 	std::vector<io_context_p> ios_;
 
@@ -329,6 +336,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<lslboost::asio::const_buffer> sync_buffs_;
 };
 
 } // namespace lsl
diff --git a/src/tcp_server.cpp b/src/tcp_server.cpp
@@ -148,10 +148,11 @@ 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, tcp protocol, int chunk_size)
+	factory_p factory, tcp protocol, int chunk_size, bool do_async)
 	: chunk_size_(chunk_size), shutdown_(false), info_(std::move(info)), io_(std::move(io)),
 	  factory_(std::move(factory)), send_buffer_(std::move(sendbuf)),
-	  acceptor_(std::make_shared<tcp::acceptor>(*io_)) {
+	  acceptor_(std::make_shared<tcp::acceptor>(*io_)),
+      transfer_is_async_(do_async) {
 	// open the server connection
 	acceptor_->open(protocol);
 
@@ -222,34 +223,69 @@ void tcp_server::handle_accept_outcome(std::shared_ptr<client_session> newsessio
 	accept_next_connection();
 }
 
+// === synchronous transfer
+
+void tcp_server::write_all_blocking(std::vector<lslboost::asio::const_buffer> buffs) {
+	std::lock_guard<std::recursive_mutex> lock(inflight_mut_);
+	std::size_t bytes_sent;
+	lslboost::system::error_code ec;
+	for (const auto &x : inflight_ready_) {
+		if (x.second && x.first->is_open()) {
+			bytes_sent = x.first->send(buffs, 0, ec);
+			if (ec) {
+				switch(ec.value()) {
+				case lslboost::system::errc::broken_pipe:
+				case lslboost::system::errc::connection_reset:
+					LOG_F(WARNING, "Broken Pipe / Connection Reset detected. Closing socket.");
+					inflight_ready_[x.first] = false;
+					post(*io_, [x]() {
+						close_inflight_socket(x);
+					});
+					// We leave it up to the client_session destructor to remove the socket.
+					break;
+				default:
+					LOG_F(WARNING, "Unhandled write_all_blocking error: %s.", ec.message().c_str());
+				}
+			}
+		}
+	}
+}
+
 // === graceful cancellation of in-flight sockets ===
 
 void tcp_server::register_inflight_socket(const tcp_socket_p &sock) {
 	std::lock_guard<std::recursive_mutex> lock(inflight_mut_);
-	inflight_.insert(sock);
+	inflight_ready_.insert({sock, false});
 }
 
 void tcp_server::unregister_inflight_socket(const tcp_socket_p &sock) {
 	std::lock_guard<std::recursive_mutex> lock(inflight_mut_);
-	inflight_.erase(sock);
+	inflight_ready_[sock] = false;
+	inflight_ready_.erase(sock);
+}
+
+void tcp_server::close_inflight_socket(std::pair<tcp_socket_p, bool> x) {
+	try {
+		if (x.first->is_open()) {
+			try {
+				// (in some cases shutdown may fail)
+				x.first->shutdown(x.first->shutdown_both);
+			} catch (...) {}
+			x.first->close();
+		}
+	} catch (std::exception &e) {
+		LOG_F(WARNING, "Error during shutdown_and_close: %s", e.what());
+	}
 }
 
 void tcp_server::close_inflight_sockets() {
 	std::lock_guard<std::recursive_mutex> lock(inflight_mut_);
-	for (const auto &sock : inflight_)
-		post(*io_, [sock]() {
-			try {
-				if (sock->is_open()) {
-					try {
-						// (in some cases shutdown may fail)
-						sock->shutdown(sock->shutdown_both);
-					} catch (...) {}
-					sock->close();
-				}
-			} catch (std::exception &e) {
-				LOG_F(WARNING, "Error during shutdown_and_close: %s", e.what());
-			}
+	for (const auto &x : inflight_ready_) {
+		inflight_ready_[x.first] = false;
+		post(*io_, [x]() {
+			close_inflight_socket(x);
 		});
+	}
 }
 
 
@@ -509,8 +545,17 @@ void client_session::handle_send_feedheader_outcome(err_t err, std::size_t n) {
 		feedbuf_.consume(n);
 		// register outstanding work at the server (will be unregistered at session destruction)
 		work_ = std::make_shared<work_p::element_type>(serv_->io_->get_executor());
-		// spawn a sample transfer thread
+		serv_->inflight_ready_[sock_] = true;
 		std::thread(&client_session::transfer_samples_thread, this, shared_from_this()).detach();
+
+		/*
+		if (serv_->transfer_is_async_)
+			// spawn a sample transfer thread
+			std::thread(&client_session::transfer_samples_thread, this, shared_from_this()).detach();
+		else
+			LOG_F(WARNING, "Using synchronous-only transfer for new client session!");
+		 */
+
 	} catch (std::exception &e) {
 		LOG_F(WARNING, "Unexpected error while handling the feedheader send outcome: %s", e.what());
 	}
diff --git a/src/tcp_server.h b/src/tcp_server.h
@@ -47,10 +47,11 @@ class tcp_server : public std::enable_shared_from_this<tcp_server> {
 	 * @param factory A sample factory that is shared with other server objects.
 	 * @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
+	 * @param do_async Flag to indicate the data transfer should happen asynchronously in a thread.
 	 * the effective chunking.
 	 */
 	tcp_server(stream_info_impl_p info, io_context_p io, send_buffer_p sendbuf, factory_p factory,
-		tcp protocol, int chunk_size);
+		tcp protocol, int chunk_size, bool do_async = true);
 
 	/**
 	 * Begin serving TCP connections.
@@ -68,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<lslboost::asio::const_buffer> buffs);
+
 private:
 	friend class client_session;
 	/// Start accepting a new connection.
@@ -83,6 +90,9 @@ class tcp_server : public std::enable_shared_from_this<tcp_server> {
 	/// Unregister an in-flight session socket.
 	void unregister_inflight_socket(const tcp_socket_p &sock);
 
+	/// Post a close of a single in-flight socket
+	static void close_inflight_socket(std::pair<tcp_socket_p, bool> x);
+
 	/// Post a close of all in-flight sockets.
 	void close_inflight_sockets();
 
@@ -101,8 +111,12 @@ class tcp_server : public std::enable_shared_from_this<tcp_server> {
 	// acceptor socket
 	tcp_acceptor_p acceptor_; // our server socket
 
+	// Flag to indicate that new client_sessions should setup an async data transfer thread.
+	bool transfer_is_async_;
+
 	// registry of in-flight client sockets (for cancellation)
-	std::set<tcp_socket_p> inflight_;		 // registry of currently in-flight sockets
+	// std::set<tcp_socket_p> inflight_;		 // registry of currently in-flight sockets
+	std::map<tcp_socket_p, bool> inflight_ready_;
 	std::recursive_mutex inflight_mut_;		 // mutex protecting the registry from concurrent access
 
 	// some cached data
