Switched from sync transcoder to async transcoder

2020-12-12 14:43:33 +01:00
parent 0586e93655
commit eea27b86f0
34 changed files with 959 additions and 504 deletions
@@ -0,0 +1,201 @@
+#include "ChildProcess.hpp"
+
+#include <cstring>
+#include <cerrno>
+#include <fcntl.h>
+#include <stdexcept>
+#include <sys/types.h>
+#include <sys/wait.h>
+#include <signal.h>
+#include <unistd.h>
+
+#include <algorithm>
+#include <iostream>
+#include <mutex>
+
+#include <boost/asio/read.hpp>
+#include <boost/asio/buffer.hpp>
+
+#include "utils/Exception.hpp"
+#include "utils/Logger.hpp"
+
+namespace
+{
+	class SystemException : public ChildProcessException
+	{
+		public:
+			SystemException(int err, const std::string& errMsg)
+				: ChildProcessException {errMsg + ": " + strerror(err)}
+			{}
+	};
+}
+
+ChildProcess::ChildProcess(boost::asio::io_context& ioContext, const std::filesystem::path& path, const Args& args)
+: _ioContext {ioContext}
+, _childStdout {_ioContext}
+
+{
+	// make sure only one thread is executing this part of code
+	static std::mutex mutex;
+	std::unique_lock<std::mutex> lock {mutex};
+
+	int pipe[2];
+
+	int res {pipe2(pipe, O_NONBLOCK | O_CLOEXEC)};
+	if (res < 0)
+		throw SystemException {errno, "pipe2 failed!"};
+
+	{
+		const std::size_t pipeSize {65536*8};
+
+		if (fcntl(pipe[0], F_SETPIPE_SZ, pipeSize) == -1)
+			throw SystemException {errno, "fcntl failed!"};
+
+		if (fcntl(pipe[1], F_SETPIPE_SZ, pipeSize) == -1)
+			throw SystemException {errno, "fcntl failed!"};
+	}
+
+	res = fork();
+	if (res == -1)
+		throw SystemException {errno, "fork failed!"};
+
+	if (res == 0) // CHILD
+	{
+		close(pipe[0]);
+		close(STDIN_FILENO);
+		close(STDERR_FILENO);
+
+		// Replace stdout with pipe write
+		if (dup2(pipe[1], STDOUT_FILENO) == -1)
+			exit(-1);
+
+		std::vector<const char*> execArgs;
+		std::transform(std::cbegin(args), std::cend(args), std::back_inserter(execArgs), [](const std::string& arg) { return arg.c_str(); });
+		execArgs.push_back(nullptr);
+
+		res = execv(path.string().c_str(), (char *const*)&execArgs[0]);
+		if (res == -1)
+			exit(-1);
+	}
+	else // PARENT
+	{
+		close(pipe[1]);
+		_childStdout.assign(pipe[0]);
+		_childPID = res;
+	}
+}
+
+ChildProcess::~ChildProcess()
+{
+	if (!_waited)
+	{
+		close(_childStdout.native_handle());
+		kill();
+		wait(true);
+	}
+}
+
+void
+ChildProcess::drain()
+{
+	char buf[128];
+
+	while (boost::asio::read(_childStdout, boost::asio::buffer(buf)) > 0)
+		LMS_LOG(CHILDPROCESS, DEBUG) << "drained some bytes" << std::endl;
+}
+
+void
+ChildProcess::kill()
+{
+	::kill(_childPID, SIGKILL);
+}
+
+bool
+ChildProcess::wait(bool block)
+{
+	int wstatus {};
+
+	pid_t pid {waitpid(_childPID, &wstatus, block ? 0 : WNOHANG)};
+
+	if (pid == -1)
+		throw SystemException {errno, "waitpid failed!"};
+	else if (pid == 0)
+		return false;
+
+	if (WIFEXITED(wstatus))
+		_exitCode = WEXITSTATUS(wstatus);
+
+	_waited = true;
+	return true;
+}
+
+
+void
+ChildProcess::asyncRead(std::byte* data, std::size_t bufferSize, ReadCallback callback) 
+{
+	LMS_LOG(CHILDPROCESS, DEBUG) << "ASYNC READ";
+
+	boost::asio::async_read(_childStdout, boost::asio::buffer(data, bufferSize),
+		[this, callback {std::move(callback)}](const boost::system::error_code& error, std::size_t bytesTransferred)
+		{
+			LMS_LOG(CHILDPROCESS, DEBUG) << "ASYNC READ CB - error = '" << error.message() << "', bytesTransferred = " << bytesTransferred;
+
+			if (error)
+			{
+				{
+					boost::system::error_code closeError;
+					_childStdout.close(closeError);
+				}
+
+				if (error == boost::asio::error::operation_aborted)
+				{
+					return;
+				}
+				if (error == boost::asio::error::eof)
+				{
+					callback(ReadResult::EndOfFile, bytesTransferred);
+					return;
+				}
+				else
+				{
+					callback(ReadResult::Error, bytesTransferred);
+					return;
+				}
+			}
+
+			callback(ReadResult::Success, bytesTransferred);
+		});
+}
+
+void
+ChildProcess::asyncWaitForData(WaitCallback cb)
+{
+	LMS_LOG(CHILDPROCESS, DEBUG) << "Async wait requested";
+
+	_childStdout.async_wait(boost::asio::posix::stream_descriptor::wait_read,
+			[cb {std::move(cb)}](const boost::system::error_code& ec)
+	{
+		LMS_LOG(CHILDPROCESS, DEBUG) << "Wait CB, error = " << ec.message();
+		if (!ec)
+			cb();
+	});
+}
+
+std::size_t
+ChildProcess::readSome(std::byte* data, std::size_t bufferSize)
+{
+	boost::system::error_code ec;
+	const std::size_t res {_childStdout.read_some(boost::asio::buffer(data, bufferSize), ec)};
+	LMS_LOG(CHILDPROCESS, DEBUG) << "read some " << res << " bytes, ec = " << ec.message();
+	if (ec)
+		_childStdout.close(ec);
+
+	return res;
+}
+
+bool
+ChildProcess::finished()
+{
+	return !_childStdout.is_open();
+}
+
@@ -0,0 +1,37 @@
+#pragma once
+
+#include <sys/types.h>
+#include <unistd.h>
+
+#include <filesystem>
+
+#include <boost/asio/io_context.hpp>
+#include <boost/asio/posix/stream_descriptor.hpp>
+
+#include "utils/IChildProcess.hpp"
+
+
+class ChildProcess : public IChildProcess
+{
+	public:
+		~ChildProcess();
+		ChildProcess(boost::asio::io_context& ioContext, const std::filesystem::path& path, const Args& args);
+
+	private:
+		void		asyncRead(std::byte* data, std::size_t bufferSize, ReadCallback callback) override;
+		void		asyncWaitForData(WaitCallback cb) override;
+		std::size_t	readSome(std::byte* data, std::size_t bufferSize) override;
+		bool		finished() override;
+
+		void	kill();
+		void	drain();
+		bool	wait(bool block); // return true if waited
+
+		using FileDescriptor = boost::asio::posix::stream_descriptor;
+
+		boost::asio::io_context&	_ioContext;
+		FileDescriptor				_childStdout;
+		::pid_t						_childPID {};
+		bool						_waited {};
+		std::optional<int>			_exitCode;
+};
@@ -0,0 +1,54 @@
+
+#include "ChildProcessManager.hpp"
+
+#include "utils/Logger.hpp"
+
+#include "ChildProcess.hpp"
+
+
+std::unique_ptr<IChildProcessManager>
+createChildProcessManager()
+{
+	return std::make_unique<ChildProcessManager>();
+}
+
+ChildProcessManager::ChildProcessManager()
+: _work {boost::asio::make_work_guard(_ioContext)}
+{
+	start();
+}
+
+ChildProcessManager::~ChildProcessManager()
+{
+	stop();
+}
+
+void
+ChildProcessManager::start()
+{
+	LMS_LOG(CHILDPROCESS, INFO) << "Starting child process manager...";
+
+	_thread = std::make_unique<std::thread>([&]()
+	{
+		_ioContext.run();
+	});
+
+	LMS_LOG(CHILDPROCESS, INFO) << "Child process manager started!";
+}
+
+void
+ChildProcessManager::stop()
+{
+	LMS_LOG(CHILDPROCESS, INFO) << "Stopping child process manager";
+	_work.reset();
+	_thread->join();
+	LMS_LOG(CHILDPROCESS, INFO) << "Stopped child process manager";
+}
+
+std::unique_ptr<IChildProcess>
+ChildProcessManager::spawnChildProcess(const std::filesystem::path& path, const IChildProcess::Args& args)
+{
+	return std::make_unique<ChildProcess>(_ioContext, path, args);
+}
+
+
@@ -0,0 +1,34 @@
+#pragma once
+
+#include <memory>
+#include <thread>
+
+#include <boost/asio/io_context.hpp>
+#include <boost/asio/executor_work_guard.hpp>
+
+#include "utils/IChildProcessManager.hpp"
+
+class ChildProcessManager : public IChildProcessManager
+{
+	public:
+
+		ChildProcessManager();
+		~ChildProcessManager();
+
+		ChildProcessManager(const ChildProcessManager&) = delete;
+		ChildProcessManager(ChildProcessManager&&) = delete;
+		ChildProcessManager& operator=(const ChildProcessManager&) = delete;
+		ChildProcessManager& operator=(ChildProcessManager&&) = delete;
+
+	private:
+		std::unique_ptr<IChildProcess> spawnChildProcess(const std::filesystem::path& path, const IChildProcess::Args& args) override;
+
+		void start();
+		void stop();
+
+		boost::asio::io_context	_ioContext;
+		std::unique_ptr<std::thread> _thread;
+		boost::asio::executor_work_guard<boost::asio::io_context::executor_type> _work;
+};
+
+
@@ -36,7 +36,7 @@ FileResourceHandler::FileResourceHandler(const std::filesystem::path& path)
 }


-void
+Wt::Http::ResponseContinuation*
 FileResourceHandler::processRequest(const Wt::Http::Request& request, Wt::Http::Response& response)
 {
 	::uint64_t startByte {_offset};
@@ -49,7 +49,7 @@ FileResourceHandler::processRequest(const Wt::Http::Request& request, Wt::Http::
 			LMS_LOG(UTILS, ERROR) << "Cannot open file stream for '" << _path.string() << "'";
 			response.setStatus(404);
 			_isFinished = true;
-			return;
+			return {};
 		}
 		else
 		{
@@ -72,7 +72,7 @@ FileResourceHandler::processRequest(const Wt::Http::Request& request, Wt::Http::

 			LMS_LOG(UTILS, DEBUG) << "Range not satisfiable";
 			_isFinished = true;
-			return;
+			return {};
 		}

 		if (ranges.size() == 1)
@@ -102,7 +102,7 @@ FileResourceHandler::processRequest(const Wt::Http::Request& request, Wt::Http::
 	{
 		LMS_LOG(UTILS, ERROR) << "Cannot reopen file stream for '" << _path.string() << "'";
 		_isFinished = true;
-		return;
+		return {};
 	}

 	ifs.seekg(static_cast<std::istream::pos_type>(startByte));
@@ -123,20 +123,16 @@ FileResourceHandler::processRequest(const Wt::Http::Request& request, Wt::Http::
 	if (ifs.good() && actualPieceSize < restSize)
 	{
 		_offset = startByte + actualPieceSize;
-
 		LMS_LOG(UTILS, DEBUG) << "Job not complete! Next chunk offset = " << _offset;
-	}
-	else
-	{
-		_isFinished = true;
-		LMS_LOG(UTILS, DEBUG) << "Job complete!";
-	}
-}

-bool
-FileResourceHandler::isFinished() const
-{
-	return _isFinished;
+		return response.createContinuation();
+	}
+
+	_isFinished = true;
+	LMS_LOG(UTILS, DEBUG) << "Job complete!";
+
+	return {};
 }


+
@@ -29,10 +29,9 @@ class FileResourceHandler final : public IResourceHandler

 	private:

-		void processRequest(const Wt::Http::Request& request, Wt::Http::Response& response) override;
-		bool isFinished() const override;
+		Wt::Http::ResponseContinuation* processRequest(const Wt::Http::Request& request, Wt::Http::Response& response) override;

-		static constexpr std::size_t _chunkSize {262144};
+		static constexpr std::size_t _chunkSize {65536};

 		std::filesystem::path	_path;
 		::uint64_t		_beyondLastByte {};
@@ -24,20 +24,21 @@ const char* getModuleName(Module mod)
 	switch (mod)
 	{
 		case Module::API_SUBSONIC:	return "API_SUBSONIC";
-		case Module::AUTH:		return "AUTH";
-		case Module::AV:		return "AV";
-		case Module::COVER:		return "COVER";
-		case Module::DB:		return "DB";
+		case Module::AUTH:			return "AUTH";
+		case Module::AV:			return "AV";
+		case Module::CHILDPROCESS:	return "CHILDPROC";
+		case Module::COVER:			return "COVER";
+		case Module::DB:			return "DB";
 		case Module::DBUPDATER:		return "DB UPDATER";
 		case Module::FEATURE:		return "FEATURE";
-		case Module::MAIN:		return "MAIN";
+		case Module::MAIN:			return "MAIN";
 		case Module::METADATA:		return "METADATA";
 		case Module::REMOTE:		return "REMOTE";
 		case Module::SERVICE:		return "SERVICE";
 		case Module::RECOMMENDATION:	return "RECOMMENDATION";
 		case Module::TRANSCODE:		return "TRANSCODE";
-		case Module::UI:		return "UI";
-		case Module::UTILS:		return "UTILS";
+		case Module::UI:			return "UI";
+		case Module::UTILS:			return "UTILS";
 	}
 	return "";
 }