Switch to atomics

components/rgbd-sources/include/ftl/rgbd/streamer.hpp

@@ -11,6 +11,7 @@
 #include <vector>
 #include <map>
 #include <shared_mutex>
+#include <atomic>
 
 namespace ftl {
 namespace rgbd {
@@ -29,7 +30,7 @@ static const unsigned int kTransmitted = 0x2;
 
 struct StreamSource {
 	ftl::rgbd::Source *src;
-	unsigned int state;								// Busy or ready to swap?
+	std::atomic<unsigned int> state;				// Busy or ready to swap?
 	cv::Mat rgb;									// Tx buffer
 	cv::Mat depth;									// Tx buffer
 	std::vector<detail::StreamClient> clients[10];	// One list per bitrate
@@ -104,7 +105,7 @@ class Streamer : public ftl::Configurable {
 	bool late_;
 	std::mutex job_mtx_;
 	std::condition_variable job_cv_;
-	int jobs_;
+	std::atomic<int> jobs_;
 
 	void _schedule();
 	void _swap(detail::StreamSource &);

components/rgbd-sources/src/streamer.cpp

@@ -106,6 +106,9 @@ void Streamer::add(Source *src) {
 }
 
 void Streamer::_addClient(const string &source, int N, int rate, const ftl::UUID &peer, const string &dest) {
+	StreamSource *s = nullptr;
+
+	{
 		UNIQUE_LOCK(mutex_,slk);
 		if (sources_.find(source) == sources_.end()) return;
 
@@ -114,7 +117,12 @@ void Streamer::_addClient(const string &source, int N, int rate, const ftl::UUID
 
 		DLOG(INFO) << "Adding Stream Peer: " << peer.to_string();
 
-	StreamSource *s = sources_[source];
+		s = sources_[source];
+	}
+
+	if (!s) return;
+
+	UNIQUE_LOCK(s->mutex, lk2);
 	for (int i=0; i<s->clients[rate].size(); i++) {
 		if (s->clients[rate][i].peerid == peer) {
 			StreamClient &c = s->clients[rate][i];
@@ -221,26 +229,29 @@ void Streamer::_schedule() {
 		}
 
 		// There will be two jobs for this source...
-		UNIQUE_LOCK(job_mtx_,lk);
+		//UNIQUE_LOCK(job_mtx_,lk);
 		jobs_ += 2;
-		lk.unlock();
+		//lk.unlock();
 
-		// Grab job
-		pool_.push([this,uri](int id) {
 		StreamSource *src = sources_[uri];
+		if (src == nullptr || src->state != 0) continue;
 
+		// Grab job
+		pool_.push([this,src](int id) {
+			//StreamSource *src = sources_[uri];
 			src->src->grab();
 
 			// CHECK (Nick) Can state be an atomic instead?
-			UNIQUE_LOCK(src->mutex, lk);
+			//UNIQUE_LOCK(src->mutex, lk);
 			src->state |= ftl::rgbd::detail::kGrabbed;
 			_swap(*src);
-			lk.unlock();
+			//lk.unlock();
 
 			// Mark job as finished
-			UNIQUE_LOCK(job_mtx_, ulk);
-			jobs_--;
-			ulk.unlock();
+			//UNIQUE_LOCK(job_mtx_, ulk);
+			//jobs_--;
+			//ulk.unlock();
+			--jobs_;
 			job_cv_.notify_one();
 		});
 
@@ -249,8 +260,8 @@ void Streamer::_schedule() {
 		// meaning that no lock is required here since outer shared_lock
 		// prevents addition of new clients.
 		// TODO, could do one for each bitrate...
-		pool_.push([this,uri](int id) {
-			StreamSource *src = sources_[uri];
+		pool_.push([this,src](int id) {
+			//StreamSource *src = sources_[uri];
 
 			try {
 			if (src && src->rgb.rows > 0 && src->depth.rows > 0 && src->clients[0].size() > 0) {
@@ -270,6 +281,8 @@ void Streamer::_schedule() {
 
 				//LOG(INFO) << "Data size: " << ((rgb_buf.size() + d_buf.size()) / 1024) << "kb";
 
+				{
+					UNIQUE_LOCK(src->mutex,lk);
 					auto i = src->clients[0].begin();
 					while (i != src->clients[0].end()) {
 						try {
@@ -289,6 +302,7 @@ void Streamer::_schedule() {
 						}
 					}
 				}
+			}
 			} catch(...) {
 				LOG(ERROR) << "Error in transmission loop";
 			}
@@ -298,16 +312,17 @@ void Streamer::_schedule() {
 			LOG(INFO) << "Stream Elapsed: " << elapsed.count();*/
 
 			// CHECK (Nick) Could state be an atomic?
-			UNIQUE_LOCK(src->mutex,lk);
+			//UNIQUE_LOCK(src->mutex,lk);
 			//LOG(INFO) << "Tx Frame: " << uri;
 			src->state |= ftl::rgbd::detail::kTransmitted;
 			_swap(*src);
-			lk.unlock();
+			//lk.unlock();
 
 			// Mark job as finished
-			UNIQUE_LOCK(job_mtx_,ulk);
-			jobs_--;
-			ulk.unlock();
+			//UNIQUE_LOCK(job_mtx_,ulk);
+			//jobs_--;
+			//ulk.unlock();
+			--jobs_;
 			job_cv_.notify_one();
 		});
 	}