Update videocapture to use common threading, so all projects will be …

…on the same plate, WiP #73

Capture/capturelib/include/CaptureLib.h

@@ -2,6 +2,7 @@
 #define REPROSTIM_CAPTURELIB_H
 
 #include <string>
+#include <functional>
 #include <iostream>
 #include <vector>
 #include <chrono>
@@ -67,7 +68,8 @@ namespace reprostim {
 
 	std::string chiToString(MWCAP_CHANNEL_INFO &info);
 
-	std::string exec(bool verbose, const std::string &cmd, bool showStdout = false);
+	std::string exec(bool verbose, const std::string &cmd, bool showStdout = false,
+					 std::function<bool()> isTerminated = [](){ return false; });
 
 	bool findTargetVideoDevice(bool verbose, const std::string &serialNumber, VideoDevice &vd);
 

Capture/capturelib/include/CaptureThreading.h

@@ -140,6 +140,7 @@ namespace reprostim {
 		SingleThreadExecutor();
 		~SingleThreadExecutor();
 
+		T* getCurrentThread() const;
 		void schedule(T* pThread);
 		void shutdown();
 	};
@@ -156,6 +157,11 @@ namespace reprostim {
 		safeDelete(m_pPrev);
 	}
 
+	template<typename T>
+	T* SingleThreadExecutor<T>::getCurrentThread() const {
+		return m_pCur;
+	}
+
 	template<typename T>
 	inline void SingleThreadExecutor<T>::safeDelete(T* &p) {
 		if( p ) {

Capture/capturelib/src/CaptureLib.cpp

@@ -71,15 +71,19 @@ namespace reprostim {
 		return s.str();
 	}
 
-	std::string exec(bool verbose, const std::string &cmd, bool showStdout) {
+	std::string exec(bool verbose,
+					 const std::string &cmd,
+					 bool showStdout,
+					 std::function<bool()> isTerminated
+					 ) {
 		std::array<char, 128> buffer;
 		std::string result;
 		std::unique_ptr<FILE, decltype(&pclose)> pipe(popen(cmd.c_str(), "r"), pclose);
 		if (!pipe) {
 			_ERROR("popen() failed for cmd: " << cmd);
 			throw std::runtime_error("popen() failed!");
 		}
-		while (fgets(buffer.data(), buffer.size(), pipe.get()) != nullptr) {
+		while (!isTerminated() && fgets(buffer.data(), buffer.size(), pipe.get()) != nullptr) {
 			result += buffer.data();
 			if (showStdout) {
 				_INFO_RAW(buffer.data());

Capture/screencapture/RecordingThread.h

@@ -10,15 +10,15 @@ using namespace reprostim;
 // Screen capture params shared between threads
 // make sure it's thread-safe in usage
 struct RecordingParams {
-	bool verbose;
-	int sessionId;
-	int cx;
-	int cy;
-	int threshold;
-	std::string outPath;
-	std::string videoDevPath;
-	bool dumpRawFrame;
-	int intervalMs;
+	const bool verbose;
+	const int sessionId;
+	const int cx;
+	const int cy;
+	const int threshold;
+	const std::string outPath;
+	const std::string videoDevPath;
+	const bool dumpRawFrame;
+	const int intervalMs;
 };
 
 using RecordingThread = WorkerThread<RecordingParams>;

Capture/screencapture/ScreenCapture.cpp

@@ -46,7 +46,7 @@ void ScreenCaptureApp::onCaptureStart() {
 
 void ScreenCaptureApp::onCaptureStop(const std::string& message) {
 	if( recording>0 ) {
-		_INFO("Stop recording snapshots for session " << g_activeSessionId );
+		_INFO("Stop recording snapshots for session " << g_activeSessionId << ". " << message);
 		m_recExec.schedule(nullptr);
 		recording = 0;
 		SLEEP_SEC(1);

Capture/version.txt

@@ -1 +1 @@
-1.1.0.27
+1.1.0.32

Capture/videocapture/VideoCapture.cpp

@@ -59,11 +59,57 @@ using namespace reprostim;
 ////////////////////////////////////////////////////////////////////////////
 //
 
-void threadFuncFfmpeg(bool verbose, const std::string& cmd) {
+inline std::string buildVideoFile(
+		const std::string& outPath,
+		const std::string& name,
+		const std::string& out_fmt) {
+	return std::filesystem::path(outPath) / (name + "." + out_fmt);
+}
+
+std::string renameVideoFile(
+		const std::string& outVideoFile,
+		const std::string& outPath,
+		const std::string& start_ts,
+		const std::string& out_fmt,
+		const std::string& message) {
+	if( std::filesystem::exists(outVideoFile) ) {
+		std::string stop_ts = getTimeStr();
+		std::string outVideoFile2 = buildVideoFile(outPath, start_ts + "_" + stop_ts, out_fmt);
+		_INFO(message << " Saving video " << outVideoFile2);
+		rename(outVideoFile.c_str(), outVideoFile2.c_str());
+		return outVideoFile2;
+	}
+	return outVideoFile;
+}
+
+
+// specialization/override for default WorkerThread::run
+template<>
+void FfmpegThread ::run() {
+	verbose = getParams().verbose;
 	std::thread::id tid= std::this_thread::get_id();
-	_VERBOSE("threadFuncFfmpeg start [" << tid << "]: " << cmd);
-	exec(verbose, cmd, true);
-	_VERBOSE("threadFuncFfmpeg leave [" << tid << "]: " << cmd);
+	_VERBOSE("FfmpegThread start [" << tid << "]: " << getParams().cmd);
+
+	_INFO(getParams().start_ts << ": <SYSTEMCALL> " << getParams().cmd);
+	//system(cmd);
+
+	// NOTE: in future improve async subprocess execution with reworked exec API.
+	try {
+		exec(verbose,
+			 getParams().cmd,
+			 true,
+			 [this]() { return isTerminated(); }
+		);
+	} catch(std::exception& e) {
+		_ERROR("FfmpegThread unhandled exception: " << e.what());
+	}
+	renameVideoFile(getParams().outVideoFile,
+					getParams().outPath,
+					getParams().start_ts,
+					getParams().outExt,
+					":\tFfmpeg thread terminated.");
+
+	_VERBOSE("FfmpegThread leave [" << tid << "]: " << getParams().cmd);
 }
 
 ////////////////////////////////////////////////////////////////////////////
@@ -74,6 +120,10 @@ VideoCaptureApp::VideoCaptureApp() {
 	audioEnabled = true;
 }
 
+VideoCaptureApp::~VideoCaptureApp() {
+	m_ffmpegExec.shutdown();
+}
+
 void VideoCaptureApp::onCaptureStart() {
 	start_ts = startRecording(vssCur.cx,
 							  vssCur.cy,
@@ -93,7 +143,7 @@ void VideoCaptureApp::onCaptureStop(const std::string& message) {
 		std::string stop_str = getTimeStr();
 		stopRecording(start_ts, opts.outPath);
 		recording = 0;
-		_INFO(stop_str << message);
+		_INFO(stop_str << " " << message);
 	}
 }
 
@@ -174,10 +224,11 @@ std::string VideoCaptureApp::startRecording(int cx, int cy, const std::string& f
 	const FfmpegOpts& opts = cfg.ffm_opts;
 	std::string a_dev2 = a_dev;
 	if( a_dev2.find("-i ")!=0 ) a_dev2 = "-i " + a_dev2;
+	std::string outVideoFile = buildVideoFile(outPath, start_ts + "_", opts.out_fmt);
 	sprintf(
 			ffmpg,
 			"ffmpeg %s %s %s %s %s -framerate %s -video_size %ix%i %s -i %s "
-			"%s %s %s %s %s/%s_.%s 2>&1", // > /dev/null &",
+			"%s %s %s %s %s 2>&1", // > /dev/null &",
 			opts.a_fmt.c_str(),
 			opts.a_nchan.c_str(),
 			opts.a_opt.c_str(),
@@ -192,45 +243,43 @@ std::string VideoCaptureApp::startRecording(int cx, int cy, const std::string& f
 			opts.pix_fmt.c_str(),
 			opts.n_threads.c_str(),
 			opts.a_enc.c_str(),
-			outPath.c_str(),
-			start_ts.c_str(),
-			opts.out_fmt.c_str()
+			outVideoFile.c_str()
 	);
 
-	std::string cmd = ffmpg;
-	_INFO(start_ts << ": <SYSTEMCALL> " << cmd.c_str());
-	//system(cmd);
-	std::thread t(&threadFuncFfmpeg, verbose, cmd);
-	_VERBOSE("started recording thread: " << t.get_id());
-	t.detach(); // make daemon thread
+	FfmpegThread* pt = FfmpegThread::newInstance(FfmpegParams{
+			verbose,
+			ffmpg,
+			opts.out_fmt,
+			outPath,
+			outVideoFile,
+			start_ts
+	});
+
+	m_ffmpegExec.schedule(pt);
 	return start_ts;
 }
 
 void VideoCaptureApp::stopRecording(const std::string& start_ts, const std::string& vpath) {
-	std::string oldname = vpath + "/" + start_ts + "_.mkv";
-	std::string ffmpid;
-	ffmpid = exec(true, "pidof ffmpeg");
+	std::string out_fmt = cfg.ffm_opts.out_fmt;
+	std::string oldname = buildVideoFile(vpath, start_ts + "_", out_fmt);
+	std::string ffmpid = exec(true, "pidof ffmpeg");
 	_INFO("stop record says: " << ffmpid.c_str());
 	while ( ffmpid.length() > 0 ) {
 		_INFO("<> PID of ffmpeg\t===> " << ffmpid.c_str());
 		std::string stop_ts = getTimeStr();
 		std::string killCmd = "kill -9 " + ffmpid;
 		system(killCmd.c_str());
 		//
-		std::string newname = vpath + "/" + start_ts + "_" + stop_ts + ".mkv";
-		_INFO(stop_ts << ":\tKilling " << ffmpid.c_str() << ". Saving video " << newname);
-		rename(oldname.c_str(), newname.c_str());
 		SLEEP_SEC(1.5); // Allow time for ffmpeg to stop
 		ffmpid = exec(true, "pidof ffmpeg");
 	}
 
+	m_ffmpegExec.schedule(nullptr);
+
 	// finally double check file again, as sometime ffmpeg
 	// process killed while unfinished video file exists
-	if( std::filesystem::exists(oldname) ) {
-		std::string stop_ts = getTimeStr();
-		std::string newname2 = vpath + "/" + start_ts + "_" + stop_ts + ".mkv";
-		_INFO(":\tFound still unfinished video file, fixing it. Saving video " << newname2);
-		rename(oldname.c_str(), newname2.c_str());
-	}
+	renameVideoFile(oldname, vpath, start_ts, out_fmt,
+					":\tFound still unfinished video file, fixing it.");
 }
 
+

Capture/videocapture/VideoCapture.h

@@ -2,20 +2,40 @@
 #define CAPTURE_VIDEOCAPTURE_H
 
 #include "CaptureApp.h"
+#include "CaptureThreading.h"
 
 ///////////////////////////////////////////////////////////////////////////
 //
 using namespace reprostim;
 
+// ffmpeg params shared between threads
+// make sure it's thread-safe in usage
+struct FfmpegParams {
+	const bool        verbose;
+	const std::string cmd;
+	const std::string outExt;
+	const std::string outPath;
+	const std::string outVideoFile;
+	const std::string start_ts;
+};
+
+using FfmpegThread = WorkerThread<FfmpegParams>;
+
 class VideoCaptureApp: public CaptureApp {
+private:
+	SingleThreadExecutor<FfmpegThread> m_ffmpegExec;
+
+	std::string startRecording(int cx, int cy, const std::string& frameRate, const std::string& outPath,
+							   const std::string& v_dev, const std::string& a_dev);
+	void stopRecording(const std::string& start_ts, const std::string& vpath);
 public:
 	VideoCaptureApp();
+	~VideoCaptureApp();
+
+	//
 	void onCaptureStart() override;
 	void onCaptureStop(const std::string& message) override;
 	int  parseOpts(AppOpts& opts, int argc, char* argv[]) override;
-	std::string startRecording(int cx, int cy, const std::string& frameRate, const std::string& outPath,
-							   const std::string& v_dev, const std::string& a_dev);
-	void stopRecording(const std::string& start_ts, const std::string& vpath);
 };
 
 #endif //CAPTURE_VIDEOCAPTURE_H