Merge branch 'feature/170/concurrentencode' into 'master'

Implements #170 concurrent encoding

Closes #170

See merge request nicolas.pope/ftl!201

components/codecs/include/ftl/codecs/hevc.hpp

@@ -97,6 +97,10 @@ inline NALType getNALType(const std::vector<uint8_t> &data) {
 	return static_cast<NALType>((data[4] >> 1) & 0x3F);
 }
 
+inline bool validNAL(const std::vector<uint8_t> &data) {
+	return data[0] == 0 && data[1] == 0 && data[2] == 0 && data[3] == 1;
+}
+
 /**
  * Check the HEVC bitstream for an I-Frame. With NvPipe, all I-Frames start
  * with a VPS NAL unit so just check for this.

components/codecs/src/nvpipe_decoder.cpp

@@ -37,6 +37,7 @@ bool NvPipeDecoder::decode(const ftl::codecs::Packet &pkt, cv::cuda::GpuMat &out
 	is_float_channel_ = is_float_frame;
 	last_definition_ = pkt.definition;
 
+	//LOG(INFO) << "DECODE OUT: " << out.rows << ", " << out.type();
 	//LOG(INFO) << "DECODE RESOLUTION: (" << (int)pkt.definition << ") " << ftl::codecs::getWidth(pkt.definition) << "x" << ftl::codecs::getHeight(pkt.definition);
 
 	// Build a decoder instance of the correct kind
@@ -49,8 +50,6 @@ bool NvPipeDecoder::decode(const ftl::codecs::Packet &pkt, cv::cuda::GpuMat &out
 		if (!nv_decoder_) {
 			//LOG(INFO) << "Bitrate=" << (int)bitrate << " width=" << ABRController::getColourWidth(bitrate);
 			LOG(FATAL) << "Could not create decoder: " << NvPipe_GetError(NULL);
-		} else {
-			DLOG(INFO) << "Decoder created";
 		}
 
 		seen_iframe_ = false;
@@ -60,38 +59,46 @@ bool NvPipeDecoder::decode(const ftl::codecs::Packet &pkt, cv::cuda::GpuMat &out
 	tmp_.create(cv::Size(ftl::codecs::getWidth(pkt.definition),ftl::codecs::getHeight(pkt.definition)), (is_float_frame) ? CV_16U : CV_8UC4);
 
 	// Check for an I-Frame
-	if (pkt.codec == ftl::codecs::codec_t::HEVC) {
-		if (ftl::codecs::hevc::isIFrame(pkt.data)) seen_iframe_ = true;
-	} else if (pkt.codec == ftl::codecs::codec_t::H264) {
-		if (ftl::codecs::h264::isIFrame(pkt.data)) seen_iframe_ = true;
+	if (!seen_iframe_) {
+		if (pkt.codec == ftl::codecs::codec_t::HEVC) {
+			if (ftl::codecs::hevc::isIFrame(pkt.data)) seen_iframe_ = true;
+		} else if (pkt.codec == ftl::codecs::codec_t::H264) {
+			if (ftl::codecs::h264::isIFrame(pkt.data)) seen_iframe_ = true;
+		}
 	}
 
 	// No I-Frame yet so don't attempt to decode P-Frames.
 	if (!seen_iframe_) return false;
 
+	// Final checks for validity
+	if (pkt.data.size() == 0 || tmp_.size() != out.size()) { // || !ftl::codecs::hevc::validNAL(pkt.data)) {
+		LOG(ERROR) << "Failed to decode packet";
+		return false;
+	}
+
 	int rc = NvPipe_Decode(nv_decoder_, pkt.data.data(), pkt.data.size(), tmp_.data, tmp_.cols, tmp_.rows, tmp_.step);
 	if (rc == 0) LOG(ERROR) << "NvPipe decode error: " << NvPipe_GetError(nv_decoder_);
 
 	if (is_float_frame) {
 		// Is the received frame the same size as requested output?
-		if (out.rows == ftl::codecs::getHeight(pkt.definition)) {
+		//if (out.rows == ftl::codecs::getHeight(pkt.definition)) {
 			tmp_.convertTo(out, CV_32FC1, 1.0f/1000.0f, stream_);
-		} else {
+		/*} else {
 			LOG(WARNING) << "Resizing decoded frame from " << tmp_.size() << " to " << out.size();
 			// FIXME: This won't work on GPU
 			tmp_.convertTo(tmp_, CV_32FC1, 1.0f/1000.0f, stream_);
 			cv::cuda::resize(tmp_, out, out.size(), 0, 0, cv::INTER_NEAREST, stream_);
-		}
+		}*/
 	} else {
 		// Is the received frame the same size as requested output?
-		if (out.rows == ftl::codecs::getHeight(pkt.definition)) {
+		//if (out.rows == ftl::codecs::getHeight(pkt.definition)) {
 			// Flag 0x1 means frame is in RGB so needs conversion to BGR
 			if (pkt.flags & 0x1) {
 				cv::cuda::cvtColor(tmp_, out, cv::COLOR_RGBA2BGR, 0, stream_);
 			} else {
 				cv::cuda::cvtColor(tmp_, out, cv::COLOR_BGRA2BGR, 0, stream_);
 			}
-		} else {
+		/*} else {
 			LOG(WARNING) << "Resizing decoded frame from " << tmp_.size() << " to " << out.size();
 			// FIXME: This won't work on GPU, plus it allocates extra memory...
 			// Flag 0x1 means frame is in RGB so needs conversion to BGR
@@ -101,7 +108,7 @@ bool NvPipeDecoder::decode(const ftl::codecs::Packet &pkt, cv::cuda::GpuMat &out
 				cv::cuda::cvtColor(tmp_, tmp_, cv::COLOR_BGRA2BGR, 0, stream_);
 			}
 			cv::cuda::resize(tmp_, out, out.size(), 0.0, 0.0, cv::INTER_LINEAR, stream_);
-		}
+		}*/
 	}
 
 	stream_.waitForCompletion();

components/codecs/src/nvpipe_encoder.cpp

@@ -123,7 +123,7 @@ bool NvPipeEncoder::encode(const cv::cuda::GpuMat &in, definition_t odefinition,
 	pkt.data.resize(cs);
 	was_reset_ = false;
 
-	if (cs == 0) {
+	if (cs == 0 || cs >= ftl::codecs::kVideoBufferSize) {
 		LOG(ERROR) << "Could not encode video frame: " << NvPipe_GetError(nvenc_);
 		return false;
 	} else {

components/codecs/test/nvpipe_codec_unit.cpp

@@ -81,7 +81,8 @@ TEST_CASE( "NvPipeDecoder::decode() - A colour test image" ) {
 		});
 	}
 
-	SECTION("Full HD in, 720 out, FHD encoding") {
+	// No longer supported
+	/*SECTION("Full HD in, 720 out, FHD encoding") {
 		in = cv::cuda::GpuMat(cv::Size(1920,1080), CV_8UC3, cv::Scalar(255,0,0));
 		out = cv::cuda::GpuMat(cv::Size(1280,720), CV_8UC3, cv::Scalar(0,0,0));
 
@@ -90,9 +91,10 @@ TEST_CASE( "NvPipeDecoder::decode() - A colour test image" ) {
 		});
 
 		REQUIRE( (out.rows == 720) );
-	}
+	}*/
 
-	SECTION("HHD in, FHD out, FHD encoding") {
+	// No longer supported
+	/*SECTION("HHD in, FHD out, FHD encoding") {
 		in = cv::cuda::GpuMat(cv::Size(1280,720), CV_8UC3, cv::Scalar(255,0,0));
 		out = cv::cuda::GpuMat(cv::Size(1920,1080), CV_8UC3, cv::Scalar(0,0,0));
 
@@ -101,9 +103,10 @@ TEST_CASE( "NvPipeDecoder::decode() - A colour test image" ) {
 		});
 
 		REQUIRE( (out.rows == 1080) );
-	}
+	}*/
 
-	SECTION("FHD in, HHD out, SD encoding") {
+	// No longer supported
+	/*SECTION("FHD in, HHD out, SD encoding") {
 		in = cv::cuda::GpuMat(cv::Size(1920,1080), CV_8UC3, cv::Scalar(255,0,0));
 		out = cv::cuda::GpuMat(cv::Size(1280,720), CV_8UC3, cv::Scalar(0,0,0));
 
@@ -112,7 +115,7 @@ TEST_CASE( "NvPipeDecoder::decode() - A colour test image" ) {
 		});
 
 		REQUIRE( (out.rows == 720) );
-	}
+	}*/
 
 	REQUIRE( r );
 	REQUIRE( (cv::cuda::sum(out) != cv::Scalar(0,0,0)) );

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

@@ -223,7 +223,7 @@ ftl::cuda::TextureObject<T> &Frame::createTexture(ftl::codecs::Channel c, const
 		//LOG(INFO) << "Creating texture object";
 		m.tex = ftl::cuda::TextureObject<T>(m.gpu, interpolated);
 	} else if (m.tex.cvType() != ftl::traits::OpenCVType<T>::value || m.tex.width() != m.gpu.cols || m.tex.height() != m.gpu.rows) {
-		LOG(INFO) << "Recreating texture object for '" << ftl::codecs::name(c) << "'";
+		//LOG(INFO) << "Recreating texture object for '" << ftl::codecs::name(c) << "'";
 		m.tex.free();
 		m.tex = ftl::cuda::TextureObject<T>(m.gpu, interpolated);
 	}
@@ -256,7 +256,7 @@ ftl::cuda::TextureObject<T> &Frame::createTexture(ftl::codecs::Channel c, bool i
 		//LOG(INFO) << "Creating texture object";
 		m.tex = ftl::cuda::TextureObject<T>(m.gpu, interpolated);
 	} else if (m.tex.cvType() != ftl::traits::OpenCVType<T>::value || m.tex.width() != m.gpu.cols || m.tex.height() != m.gpu.rows || m.tex.devicePtr() != m.gpu.data) {
-		LOG(INFO) << "Recreating texture object for '" << ftl::codecs::name(c) << "'.";
+		//LOG(INFO) << "Recreating texture object for '" << ftl::codecs::name(c) << "'.";
 		m.tex.free();
 		m.tex = ftl::cuda::TextureObject<T>(m.gpu, interpolated);
 	}

components/rgbd-sources/src/abr.cpp

@@ -41,7 +41,7 @@ bitrate_t ABRController::selectBitrate(const NetFrame &frame) {
 
     float actual_mbps = (float(frame.tx_size) * 8.0f * (1000.0f / float(frame.tx_latency))) / 1048576.0f;
     float min_mbps = (float(frame.tx_size) * 8.0f * (1000.0f / float(ftl::timer::getInterval()))) / 1048576.0f;
-    //LOG(INFO) << "Bitrate = " << actual_mbps << "Mbps, min required = " << min_mbps << "Mbps";
+    if (actual_mbps < min_mbps) LOG(WARNING) << "Bitrate = " << actual_mbps << "Mbps, min required = " << min_mbps << "Mbps";
     float ratio = actual_mbps / min_mbps;
     //LOG(INFO) << "Rate Ratio = " << frame.tx_latency;
 

components/rgbd-sources/src/group.cpp

@@ -214,8 +214,8 @@ void Group::sync(std::function<bool(ftl::rgbd::FrameSet &)> cb) {
 				try {
 					cb(*fs);
 					//LOG(INFO) << "Frameset processed (" << name_ << "): " << fs->timestamp;
-				} catch(...) {
-					LOG(ERROR) << "Exception in group sync callback";
+				} catch(std::exception &e) {
+					LOG(ERROR) << "Exception in group sync callback: " << e.what();
 				}
 
 				// The buffers are invalid after callback so mark stale

components/rgbd-sources/src/sources/net/net.cpp

@@ -276,10 +276,37 @@ void NetSource::_recvPacket(short ttimeoff, const ftl::codecs::StreamPacket &spk
 		LOG(WARNING) << "Missing calibration, skipping frame";
 		return;
 	}
+
+	//LOG(INFO) << "PACKET: " << spkt.timestamp << ", " << (int)spkt.channel << ", " << (int)pkt.codec;
 	
 	const cv::Size size = cv::Size(ftl::codecs::getWidth(pkt.definition), ftl::codecs::getHeight(pkt.definition));
 	NetFrame &frame = queue_.getFrame(spkt.timestamp, size, CV_8UC3, (isFloatChannel(chan) ? CV_32FC1 : CV_8UC3));
 
+	if (timestamp_ > 0 && frame.timestamp <= timestamp_) {
+		LOG(ERROR) << "Duplicate frame - " << frame.timestamp << " received=" << int(rchan) << " uri=" << uri_;
+		return;
+	}
+
+	// Calculate how many packets to expect for this channel
+	if (frame.chunk_total[channum] == 0) {
+		frame.chunk_total[channum] = pkt.block_total;
+	}
+
+	// Capture tx time of first received chunk
+	if (frame.chunk_count[0] == 0 && frame.chunk_count[1] == 0) {
+		UNIQUE_LOCK(frame.mtx, flk);
+		if (frame.chunk_count[0] == 0 && frame.chunk_count[1] == 0) {
+			frame.tx_latency = int64_t(ttimeoff);
+		}
+	}	
+
+	++frame.chunk_count[channum];
+	if (frame.chunk_count[channum] == frame.chunk_total[channum]) ++frame.channel_count;
+	if (frame.chunk_count[channum] > frame.chunk_total[channum]) {
+		LOG(WARNING) << "Too many channel packets received, discarding";
+		return;
+	}
+
 	// Update frame statistics
 	frame.tx_size += pkt.data.size();
 
@@ -309,29 +336,6 @@ void NetSource::_recvPacket(short ttimeoff, const ftl::codecs::StreamPacket &spk
 
 	// TODO:(Nick) Decode directly into double buffer if no scaling
 
-	if (timestamp_ > 0 && frame.timestamp <= timestamp_) {
-		LOG(ERROR) << "BAD DUPLICATE FRAME - " << frame.timestamp << " received=" << int(rchan) << " uri=" << uri_;
-		return;
-	}
-
-	// Calculate how many packets to expect for this channel
-	if (frame.chunk_total[channum] == 0) {
-		frame.chunk_total[channum] = pkt.block_total;
-	}		
-
-	++frame.chunk_count[channum];
-	if (frame.chunk_count[channum] == frame.chunk_total[channum]) ++frame.channel_count;
-	if (frame.chunk_count[channum] > frame.chunk_total[channum]) LOG(FATAL) << "TOO MANY CHUNKS";
-
-	// Capture tx time of first received chunk
-	// FIXME: This seems broken
-	if (channum == 1 && frame.chunk_count[channum] == 1) {
-		UNIQUE_LOCK(frame.mtx, flk);
-		if (frame.chunk_count[channum] == 1) {
-			frame.tx_latency = int64_t(ttimeoff);
-		}
-	}
-
 	// Last chunk of both channels now received, so we are done.
 	if (frame.channel_count == spkt.channel_count) {
 		_completeFrame(frame, now-(spkt.timestamp+frame.tx_latency));

components/rgbd-sources/src/streamer.cpp

@@ -464,27 +464,43 @@ void Streamer::_process(ftl::rgbd::FrameSet &fs) {
 				auto *enc1 = src->hq_encoder_c1;
 				auto *enc2 = src->hq_encoder_c2;
 
-				// Important to send channel 2 first if needed...
-				// Receiver only waits for channel 1 by default
-				// TODO: Each encode could be done in own thread
-				if (hasChan2) {
-					// TODO: Stagger the reset between nodes... random phasing
-					if (fs.timestamp % (10*ftl::timer::getInterval()) == 0) enc2->reset();
+				MUTEX mtx;
+				std::condition_variable cv;
+				bool chan2done = false;
 
-					auto chan = fs.sources[j]->getChannel();
-
-					enc2->encode(fs.frames[j].get<cv::cuda::GpuMat>(chan), src->hq_bitrate, [this,src,hasChan2,chan](const ftl::codecs::Packet &blk){
-						_transmitPacket(src, blk, chan, hasChan2, Quality::High);
+				if (hasChan2) {
+					ftl::pool.push([this,&fs,enc2,src,hasChan2,&cv,j,&chan2done](int id) {
+						// TODO: Stagger the reset between nodes... random phasing
+						if (fs.timestamp % (10*ftl::timer::getInterval()) == 0) enc2->reset();
+
+						auto chan = fs.sources[j]->getChannel();
+
+						try {
+							enc2->encode(fs.frames[j].get<cv::cuda::GpuMat>(chan), src->hq_bitrate, [this,src,hasChan2,chan,&cv,&chan2done](const ftl::codecs::Packet &blk){
+								_transmitPacket(src, blk, chan, hasChan2, Quality::High);
+								chan2done = true;
+								cv.notify_one();
+							});
+						} catch (std::exception &e) {
+							LOG(ERROR) << "Exception in encoder: " << e.what();
+							chan2done = true;
+							cv.notify_one();
+						}
 					});
 				} else {
 					if (enc2) enc2->reset();
+					chan2done = true;
 				}
 
 				// TODO: Stagger the reset between nodes... random phasing
 				if (fs.timestamp % (10*ftl::timer::getInterval()) == 0) enc1->reset();
-				enc1->encode(fs.frames[j].get<cv::cuda::GpuMat>(Channel::Colour), src->hq_bitrate, [this,src,hasChan2](const ftl::codecs::Packet &blk){
+				enc1->encode(fs.frames[j].get<cv::cuda::GpuMat>(Channel::Colour), src->hq_bitrate, [this,src,hasChan2,&mtx](const ftl::codecs::Packet &blk){
 					_transmitPacket(src, blk, Channel::Colour, hasChan2, Quality::High);
 				});
+
+				// Ensure both channels have been completed.
+				std::unique_lock<std::mutex> lk(mtx);
+				cv.wait(lk, [&chan2done]{ return chan2done; });
 			}
 		}