Simplified interface + fixed segfault when eof is reached. ref #287

This commit is contained in:
emeric
2022-12-30 21:01:17 +01:00
parent 040973209e
commit c6fe63f3b2
6 changed files with 33 additions and 66 deletions
@@ -46,6 +46,8 @@ namespace Av
{
if (_estimatedContentLength)
LMS_LOG(TRANSCODE, DEBUG) << "Estimated content length = " << *_estimatedContentLength;
else
LMS_LOG(TRANSCODE, DEBUG) << "Not using estimated content length";
}
Wt::Http::ResponseContinuation*
@@ -58,8 +60,8 @@ namespace Av
if (_bytesReadyCount > 0)
{
response.out().write(reinterpret_cast<const char *>(&_buffer[0]), _bytesReadyCount);
_bytesReadyCount = 0;
_totalServedByteCount += _bytesReadyCount;
_bytesReadyCount = 0;
}
if (!_transcoder.finished())
-13
View File
@@ -178,19 +178,6 @@ Transcoder::start()
}
}
void
Transcoder::asyncWaitForData(WaitCallback cb)
{
assert(_childProcess);
LOG(DEBUG) << "Want to wait for data";
_childProcess->asyncWaitForData([cb = std::move(cb)]
{
cb();
});
}
void
Transcoder::asyncRead(std::byte* buffer, std::size_t bufferSize, ReadCallback readCallback)
{
-3
View File
@@ -40,9 +40,6 @@ namespace Av
Transcoder(Transcoder&&) = delete;
Transcoder& operator=(Transcoder&&) = delete;
using WaitCallback = std::function<void()>;
void asyncWaitForData(WaitCallback cb);
// non blocking calls
using ReadCallback = std::function<void(std::size_t nbReadBytes)>;
void asyncRead(std::byte* buffer, std::size_t bufferSize, ReadCallback);
+28 -43
View File
@@ -44,7 +44,7 @@ namespace
{
public:
SystemException(int err, const std::string& errMsg)
: ChildProcessException {errMsg + ": " + strerror(err)}
: ChildProcessException {errMsg + ": " + ::strerror(err)}
{}
SystemException(boost::system::error_code ec, const std::string& errMsg)
@@ -117,42 +117,36 @@ ChildProcess::ChildProcess(boost::asio::io_context& ioContext, const std::filesy
ChildProcess::~ChildProcess()
{
if (!_waited)
LMS_LOG(CHILDPROCESS, DEBUG) << "Closing child process...";
{
LMS_LOG(CHILDPROCESS, DEBUG) << "Closing child process...";
{
boost::system::error_code closeError;
_childStdout.close(closeError);
if (closeError)
LMS_LOG(CHILDPROCESS, ERROR) << "Closed failed: " << closeError.message();
}
kill();
wait(true);
boost::system::error_code closeError;
_childStdout.close(closeError);
if (closeError)
LMS_LOG(CHILDPROCESS, ERROR) << "Closed failed: " << closeError.message();
}
}
void
ChildProcess::drain()
{
char buf[128];
if (!_finished)
kill();
while (boost::asio::read(_childStdout, boost::asio::buffer(buf)) > 0)
LMS_LOG(CHILDPROCESS, DEBUG) << "drained some bytes" << std::endl;
wait(true);
}
void
ChildProcess::kill()
{
// process may already have finished
LMS_LOG(CHILDPROCESS, DEBUG) << "Killing child process...";
::kill(_childPID, SIGKILL);
if (::kill(_childPID, SIGKILL) == -1)
LMS_LOG(CHILDPROCESS, DEBUG) << "Kill failed: " << ::strerror(errno);
}
bool
ChildProcess::wait(bool block)
{
int wstatus {};
assert(!_waited);
pid_t pid {waitpid(_childPID, &wstatus, block ? 0 : WNOHANG)};
int wstatus {};
const pid_t pid {waitpid(_childPID, &wstatus, block ? 0 : WNOHANG)};
if (pid == -1)
throw SystemException {errno, "waitpid failed!"};
@@ -160,18 +154,22 @@ ChildProcess::wait(bool block)
return false;
if (WIFEXITED(wstatus))
{
_exitCode = WEXITSTATUS(wstatus);
LMS_LOG(CHILDPROCESS, DEBUG) << "Exit code = " << *_exitCode;
}
_waited = true;
return true;
}
void
ChildProcess::asyncRead(std::byte* data, std::size_t bufferSize, ReadCallback callback)
{
assert(!finished());
LMS_LOG(CHILDPROCESS, DEBUG) << "Async read, bufferSize = " << bufferSize;
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)
{
@@ -180,33 +178,20 @@ ChildProcess::asyncRead(std::byte* data, std::size_t bufferSize, ReadCallback ca
ReadResult readResult {ReadResult::Success};
if (error)
{
_finished = true;
if (error == boost::asio::error::eof)
readResult = ReadResult::EndOfFile;
else
if (error != boost::asio::error::eof)
{
// forbidden to read any captured param here as the ChildProcess instance may already have been killed
return;
}
readResult = ReadResult::EndOfFile;
_finished = true;
}
callback(readResult, bytesTransferred);
});
}
void
ChildProcess::asyncWaitForData(WaitCallback cb)
{
LMS_LOG(CHILDPROCESS, DEBUG) << "Async wait requested";
assert(!finished());
_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)
{
@@ -220,7 +205,7 @@ ChildProcess::readSome(std::byte* data, std::size_t bufferSize)
}
bool
ChildProcess::finished()
ChildProcess::finished() const
{
return _finished;
}
+1 -3
View File
@@ -40,12 +40,10 @@ class ChildProcess : public IChildProcess
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;
bool finished() const override;
void kill();
void drain();
bool wait(bool block); // return true if waited
using FileDescriptor = boost::asio::posix::stream_descriptor;
@@ -48,9 +48,7 @@ class IChildProcess
using ReadCallback = std::function<void(ReadResult, std::size_t)>;
virtual void asyncRead(std::byte* data, std::size_t bufferSize, ReadCallback callback) = 0;
using WaitCallback = std::function<void(void)>;
virtual void asyncWaitForData(WaitCallback cb) = 0;
virtual std::size_t readSome(std::byte* data, std::size_t bufferSize) = 0;
virtual bool finished() = 0;
virtual bool finished() const = 0;
};