Fix Process async reader teardown and reap killed children

- Join the async readers before freeing the process state and pipes.
  - Make the shutdown and reader-done flags atomic.
  - Capture the POSIX reader descriptors before starting the thread.
  - Keep kill() from waiting on readers, so callers holding callback locks don't deadlock.
  - Let destroy() drain an exited child's output, then stop the readers before closing the pipes.
  - Cancel blocked Windows pipe reads with CancelIoEx, for grandchildren that keep the pipe open.
  - Reap killed POSIX children without blocking the caller, and close the Windows process handle.
  - Reset the shutdown and kill state when a killed Process is reused.
  - Ignore a missing read callback instead of throwing std::bad_function_call.
  - Fix the Log::error calls whose arguments didn't match the format string.
  - Add Process regressions for teardown, draining, reuse, reaping, callback locks and grandchildren.
This commit is contained in:
Martín Lucas Golini
2026-10-18 23:32:27 -03:00
parent e14f969dcc
commit cfb545f8f4
3 changed files with 424 additions and 31 deletions
+18 -3
View File
@@ -1,6 +1,7 @@
#ifndef EE_SYSTEM_PROCESS_HPP
#define EE_SYSTEM_PROCESS_HPP
#include <atomic>
#include <eepp/config.hpp>
#include <eepp/system/mutex.hpp>
#include <eepp/system/time.hpp>
@@ -190,14 +191,17 @@ class EE_API Process {
** @return On success true is returned.
**
** If the process to be destroyed had not finished execution, it will be
** terminated (i.e killed). */
** terminated (i.e killed) and reaped. The asynchronous readers are not waited for: they
** stop on their own, and their pipes are released once they did (at the latest when the
** Process is destroyed), so a read callback may still run after kill() returns. */
bool kill();
/** @brief Destroy a previously created process.
** @return On success true is returned.
**
** If the process to be destroyed had not finished execution, it may out live
** the parent process. */
** the parent process. The asynchronous readers are stopped before their pipes are closed:
** they first get up to a second to read what an exited child wrote. */
bool destroy();
/** @brief Returns if the subprocess is currently still alive and executing.
@@ -240,7 +244,9 @@ class EE_API Process {
protected:
void* mProcess{ nullptr };
bool mShuttingDown{ false };
std::atomic<bool> mShuttingDown{ false };
std::atomic<bool> mStdOutDone{ true };
std::atomic<bool> mStdErrDone{ true };
bool mKilled{ false };
bool mIsAsync{ false };
bool mKilling{ false };
@@ -253,6 +259,15 @@ class EE_API Process {
ReadFn mReadStdErrFn;
size_t readAll( std::string& buffer, bool readErr, Time timeout = Time::Zero );
bool terminate();
bool readersStopped() const;
void stopReaders();
/** Stops the readers, then frees the pipes and the process state they use. */
void release();
};
}} // namespace EE::System
+149 -28
View File
@@ -1,5 +1,6 @@
#include <bitset>
#include <cerrno>
#include <chrono>
#include <eepp/core/debug.hpp>
#include <eepp/core/memorymanager.hpp>
#include <eepp/core/string.hpp>
@@ -22,8 +23,13 @@
#include <fcntl.h>
#include <poll.h>
#include <stdio.h>
#include <sys/wait.h>
#elif EE_PLATFORM == EE_PLATFORM_WIN
#include <windows.h>
// subprocess.h keeps winbase.h out (it declares the subset it uses), and with it ioapiset.h.
#ifndef _IO_APISET_H_
extern "C" __declspec( dllimport ) int __stdcall CancelIoEx( void* file, void* overlapped );
#endif
#endif
#define CHUNK_SIZE 4096
@@ -98,18 +104,9 @@ Process::Process( const std::string& command, const std::vector<std::string>& ar
Process::~Process() {
mShuttingDown = true;
if ( mProcess ) {
if ( isAlive() ) {
kill();
} else {
destroy();
}
}
if ( mStdOutThread.joinable() )
mStdOutThread.join();
if ( mStdErrThread.joinable() )
mStdErrThread.join();
eeSAFE_FREE( mProcess );
if ( isAlive() )
kill();
release();
}
bool Process::create( const std::string& command, Uint32 options,
@@ -133,8 +130,17 @@ bool Process::create( const std::string& command, const std::vector<std::string>
Uint32 options,
const std::unordered_map<std::string, std::string>& environment,
const std::string& workingDirectory ) {
if ( mProcess )
return false;
if ( mProcess ) {
if ( !mKilled )
return false;
// A killed process keeps its pipes until its readers stopped.
release();
}
// A reused Process starts over: a killed one would otherwise refuse to kill the new child
// and stop its readers at once.
mShuttingDown = false;
mKilled = false;
mKilling = false;
std::vector<const char*> strings;
std::string rcommand;
mProcess = eeMalloc( sizeof( subprocess_s ) );
@@ -282,19 +288,126 @@ bool Process::kill() {
mKilling = true;
mShuttingDown = true;
subprocess_init_shutdown( PROCESS_PTR );
if ( PROCESS_PTR->alive ) {
int ret = subprocess_terminate( PROCESS_PTR );
if ( !PROCESS_PTR->alive )
return false;
const bool terminated = terminate();
mKilled = true;
// The readers may still be using the pipes. kill() never waits for them: a caller can hold
// a lock that their callbacks take. They stop on their own (the child is gone and they
// observe mShuttingDown), and release() frees the pipes once they did.
if ( readersStopped() )
release();
return terminated;
}
#ifdef EE_PLATFORM_POSIX
static void reapChild( pid_t child ) {
// SIGKILL is delivered asynchronously: wait for it briefly, then leave the wait to a thread
// so that the caller never stalls on a child that is slow to exit.
for ( int attempt = 0; attempt < 10; ++attempt ) {
const pid_t result = waitpid( child, nullptr, WNOHANG );
if ( result == child || ( result < 0 && errno != EINTR ) )
return;
if ( result == 0 )
std::this_thread::sleep_for( std::chrono::milliseconds( 1 ) );
}
try {
std::thread( [child] {
while ( waitpid( child, nullptr, 0 ) < 0 && errno == EINTR ) {
}
} ).detach();
} catch ( const std::system_error& ) {
Log::error( "Process: could not start a thread to reap child %d", child );
}
}
#endif
bool Process::terminate() {
#if EE_PLATFORM == EE_PLATFORM_WIN
// subprocess_terminate() forgets the process handle without closing it, and destroy() only
// closes the stdin handle along with it: hand it back for destroy().
void* handle = PROCESS_PTR->hProcess;
const int result = subprocess_terminate( PROCESS_PTR );
PROCESS_PTR->hProcess = handle;
return 0 == result;
#elif defined( EE_PLATFORM_POSIX )
// subprocess_terminate() forgets the child without waiting for it, which would leave a
// zombie until this process exits.
const pid_t child = PROCESS_PTR->child;
if ( child <= 0 )
return false;
const int result = subprocess_terminate( PROCESS_PTR );
reapChild( child );
return 0 == result;
#else
return 0 == subprocess_terminate( PROCESS_PTR );
#endif
}
bool Process::readersStopped() const {
return ( !mStdOutThread.joinable() || mStdOutDone ) &&
( !mStdErrThread.joinable() || mStdErrDone );
}
void Process::stopReaders() {
mShuttingDown = true;
#if EE_PLATFORM == EE_PLATFORM_WIN
// A reader blocks in a pipe read until the pipe breaks, which a surviving grandchild holding
// the pipe can postpone indefinitely. Cancel its read until it returns: a cancel that lands
// before the reader issued its next read misses it, hence the retries.
if ( mProcess ) {
// _get_osfhandle() returns -1 (INVALID_HANDLE_VALUE) for a bad descriptor.
auto handleOf = []( FILE* file ) -> void* {
if ( !file )
return nullptr;
const intptr_t handle = _get_osfhandle( _fileno( file ) );
return handle == -1 ? nullptr : reinterpret_cast<void*>( handle );
};
void* stdOut = handleOf( PROCESS_PTR->stdout_file );
void* stdErr = handleOf( PROCESS_PTR->stderr_file );
const auto readerIsRunning = []( const std::thread& thread,
const std::atomic<bool>& done ) {
return thread.joinable() && !done && thread.get_id() != std::this_thread::get_id();
};
while ( readerIsRunning( mStdOutThread, mStdOutDone ) ||
readerIsRunning( mStdErrThread, mStdErrDone ) ) {
if ( stdOut )
CancelIoEx( stdOut, nullptr );
if ( stdErr && stdErr != stdOut )
CancelIoEx( stdErr, nullptr );
std::this_thread::sleep_for( std::chrono::milliseconds( 5 ) );
}
}
#endif
// POSIX readers poll with a timeout and observe mShuttingDown.
if ( mStdOutThread.joinable() )
mStdOutThread.join();
if ( mStdErrThread.joinable() )
mStdErrThread.join();
}
void Process::release() {
stopReaders();
if ( mProcess ) {
destroy();
eeSAFE_FREE( mProcess );
mKilled = true;
return 0 == ret;
}
return false;
}
bool Process::destroy() {
if ( mProcess == nullptr )
return true;
if ( mStdOutThread.joinable() || mStdErrThread.joinable() ) {
// The readers use the pipes. Let them drain what an exited child wrote (they stop at the
// end of the pipe), then stop them: a grandchild can keep the pipe open. Readers that
// already ended are joined too, so that startAsyncRead() can start new ones.
if ( !readersStopped() && !isAlive() ) {
const auto until = std::chrono::steady_clock::now() + std::chrono::seconds( 1 );
while ( !readersStopped() && std::chrono::steady_clock::now() < until )
std::this_thread::sleep_for( std::chrono::milliseconds( 1 ) );
}
stopReaders();
}
return 0 == subprocess_destroy( PROCESS_PTR );
}
@@ -407,8 +520,11 @@ size_t Process::readAll( std::string& buffer, bool readErr, Time timeout ) {
void Process::startAsyncRead( ReadFn readStdOut, ReadFn readStdErr ) {
eeASSERT( mProcess != nullptr );
mReadStdOutFn = readStdOut;
mReadStdErrFn = readStdErr;
// The readers call both unconditionally: a missing one would throw std::bad_function_call
// on the reader thread (terminating the program) as soon as the child wrote to that stream.
auto ignore = []( const char*, size_t ) {};
mReadStdOutFn = readStdOut ? std::move( readStdOut ) : ReadFn( ignore );
mReadStdErrFn = readStdErr ? std::move( readStdErr ) : ReadFn( ignore );
Log::info( "Process::startAsyncRead called for command: %s", mCmdLine );
#if EE_PLATFORM == EE_PLATFORM_WIN
void* stdOutFd =
@@ -416,6 +532,7 @@ void Process::startAsyncRead( ReadFn readStdOut, ReadFn readStdErr ) {
void* stdErrFd =
SUBPROCESS_PTR_CAST( void*, _get_osfhandle( _fileno( PROCESS_PTR->stderr_file ) ) );
if ( stdOutFd ) {
mStdOutDone = false;
mStdOutThread = std::thread( [this, stdOutFd]() {
Log::info( "Process::startAsyncRead thread started for command: %s", mCmdLine );
unsigned n;
@@ -431,9 +548,11 @@ void Process::startAsyncRead( ReadFn readStdOut, ReadFn readStdErr ) {
if ( !mShuttingDown )
mReadStdOutFn( buffer.c_str(), static_cast<size_t>( n ) );
}
mStdOutDone = true;
} );
}
if ( stdErrFd && stdErrFd != stdOutFd ) {
mStdErrDone = false;
mStdErrThread = std::thread( [this, stdErrFd]() {
unsigned n;
std::string buffer;
@@ -449,13 +568,16 @@ void Process::startAsyncRead( ReadFn readStdOut, ReadFn readStdErr ) {
if ( !mShuttingDown )
mReadStdErrFn( buffer.c_str(), static_cast<size_t>( n ) );
}
mStdErrDone = true;
} );
}
#elif defined( EE_PLATFORM_POSIX )
mStdOutThread = std::thread( [this] {
// Read on the calling thread: the reader never touches the process state.
const int stdOutFd = PROCESS_PTR->stdout_file ? fileno( PROCESS_PTR->stdout_file ) : 0;
const int stdErrFd = PROCESS_PTR->stderr_file ? fileno( PROCESS_PTR->stderr_file ) : 0;
mStdOutDone = false;
mStdOutThread = std::thread( [this, stdOutFd, stdErrFd] {
Log::info( "Process::startAsyncRead thread started for command: %s", mCmdLine );
auto stdOutFd = fileno( PROCESS_PTR->stdout_file );
auto stdErrFd = PROCESS_PTR->stderr_file ? fileno( PROCESS_PTR->stderr_file ) : 0;
std::vector<pollfd> pollfds;
std::bitset<2> fdIsStdOut;
if ( stdOutFd ) {
@@ -478,8 +600,7 @@ void Process::startAsyncRead( ReadFn readStdOut, ReadFn readStdErr ) {
}
pollfds.back().events = POLLIN;
} else {
Log::error( "Process::startAsyncRead %s stdOutFd PROCESS_PTR->stdout_file: %p fd: %d",
mCmdLine, PROCESS_PTR->stdout_file, stdOutFd );
Log::error( "Process::startAsyncRead %s has no stdout fd", mCmdLine );
}
if ( stdErrFd && stdOutFd != stdErrFd ) {
@@ -501,8 +622,7 @@ void Process::startAsyncRead( ReadFn readStdOut, ReadFn readStdErr ) {
}
pollfds.back().events = POLLIN;
} else if ( !stdErrFd ) {
Log::error( "Process::startAsyncRead %s stdOutFd PROCESS_PTR->stderr_file: %p fd: %d",
PROCESS_PTR->stderr_file, stdErrFd );
Log::error( "Process::startAsyncRead %s has no stderr fd", mCmdLine );
}
std::string buffer;
buffer.resize( mBufferSize );
@@ -567,6 +687,7 @@ void Process::startAsyncRead( ReadFn readStdOut, ReadFn readStdErr ) {
Log::info( "Process::startAsyncRead polling ended for: %s, any open: %s, shutting down: %s",
mCmdLine, anyOpen ? "true" : "false", mShuttingDown ? "true" : "false" );
mStdOutDone = true;
} );
#endif
}
+257
View File
@@ -123,4 +123,261 @@ UTEST( Process, joinRetriesInterruptedWaits ) {
#endif
namespace {
/** Runs script in the platform shell (cmd.exe takes posix's place on Windows). */
bool createShell( Process& process, const char* posix, const char* windows ) {
#if EE_PLATFORM == EE_PLATFORM_WIN
(void)posix;
return process.create( "cmd.exe", std::vector<std::string>{ "/D", "/Q", "/C", windows },
Process::getDefaultOptions() | Process::EnableAsync );
#else
(void)windows;
return process.create( "/bin/sh", std::vector<std::string>{ "-c", posix },
Process::getDefaultOptions() | Process::EnableAsync );
#endif
}
/** A child that writes continuously, so that its reader is always delivering. */
bool createChatty( Process& process ) {
return createShell( process, "while :; do echo chatter; done",
"for /L %i in (0,0,1) do @echo chatter" );
}
} // namespace
UTEST( Process, aMissingReadCallbackIsIgnored ) {
// SSHProcessTransport only reads stdout: a ProxyCommand's warning on stderr used to throw
// std::bad_function_call on the reader thread.
std::mutex mutex;
std::condition_variable changed;
std::string output;
Process process;
ASSERT_TRUE(
createShell( process, "echo warning 1>&2; echo done", "echo warning 1>&2& echo done" ) );
process.startAsyncRead( [&]( const char* bytes, size_t count ) {
std::lock_guard<std::mutex> lock( mutex );
output.append( bytes, count );
changed.notify_all();
} );
std::unique_lock<std::mutex> lock( mutex );
EXPECT_TRUE( changed.wait_for( lock, std::chrono::seconds( 10 ),
[&] { return output.find( "done" ) != std::string::npos; } ) );
EXPECT_EQ( std::string::npos, output.find( "warning" ) );
}
UTEST( Process, destroyingRightAfterStartingTheReadersIsSafe ) {
// The readers used to read the process state when they started and keep using its pipes
// after kill() had freed them: destroy at every point of a reader's life, including before
// it started. Run under ASan/TSan to see the difference.
std::atomic<bool> released{ false };
std::atomic<int> lateCallbacks{ 0 };
for ( int i = 0; i < 100; ++i ) {
std::optional<Process> process( std::in_place );
ASSERT_TRUE( createChatty( *process ) );
process->startAsyncRead( [&]( const char*, size_t ) {
if ( released )
++lateCallbacks;
} );
if ( i % 2 )
process->kill();
process.reset();
released = true;
std::this_thread::yield();
released = false;
}
EXPECT_EQ( 0, lateCallbacks.load() );
}
UTEST( Process, destroyStopsTheReadersBeforeClosingTheirPipes ) {
std::atomic<int> active{ 0 };
std::atomic<int> afterDestroy{ 0 };
std::atomic<bool> destroyed{ false };
std::atomic<bool> delivered{ false };
Process process;
// destroy() leaves a running child alone: write a bounded amount, so that it ends by itself
// (Windows' cmd.exe keeps looping on a closed pipe).
ASSERT_TRUE( createShell( process,
"i=0; while [ $i -lt 5000 ]; do echo chatter; i=$((i+1)); done",
"for /L %i in (1,1,5000) do @echo chatter" ) );
process.startAsyncRead( [&]( const char*, size_t ) {
++active;
if ( destroyed )
++afterDestroy;
delivered = true;
std::this_thread::sleep_for( std::chrono::milliseconds( 5 ) );
--active;
} );
const auto until = std::chrono::steady_clock::now() + std::chrono::seconds( 2 );
while ( !delivered && std::chrono::steady_clock::now() < until )
std::this_thread::sleep_for( std::chrono::milliseconds( 1 ) );
ASSERT_TRUE( delivered.load() );
process.destroy();
destroyed = true;
EXPECT_EQ( 0, active.load() );
std::this_thread::sleep_for( std::chrono::milliseconds( 20 ) );
EXPECT_EQ( 0, afterDestroy.load() );
}
UTEST( Process, destroyAfterJoinKeepsTheLastOutput ) {
// Like the ecode build runner: join(), then destroy() while the reader still drains.
for ( int i = 0; i < 5; ++i ) {
std::mutex mutex;
std::string output;
Process process;
ASSERT_TRUE( createShell( process, "printf first; sleep 0.01; printf last",
"echo first& ping -n 2 127.0.0.1 >nul & echo last" ) );
process.startAsyncRead( [&]( const char* bytes, size_t count ) {
bool first;
{
std::lock_guard<std::mutex> lock( mutex );
first = output.empty();
output.append( bytes, count );
}
// Still delivering "first" when the child exits: "last" is unread when destroy()
// starts.
if ( first )
std::this_thread::sleep_for( std::chrono::milliseconds( 50 ) );
} );
int code = -1;
EXPECT_TRUE( process.join( &code ) );
process.destroy();
std::lock_guard<std::mutex> lock( mutex );
EXPECT_NE( std::string::npos, output.find( "first" ) );
EXPECT_NE( std::string::npos, output.find( "last" ) );
}
}
UTEST( Process, aKilledProcessCanBeReused ) {
Process process;
ASSERT_TRUE( createChatty( process ) );
process.startAsyncRead( []( const char*, size_t ) {} );
EXPECT_TRUE( process.kill() );
std::mutex mutex;
std::condition_variable changed;
std::string output;
ASSERT_TRUE( createShell( process, "echo $$; exec sleep 30",
"echo started& ping -n 30 127.0.0.1 >nul" ) );
EXPECT_FALSE( process.isShuttingDown() );
EXPECT_FALSE( process.killed() );
process.startAsyncRead( [&]( const char* bytes, size_t count ) {
std::lock_guard<std::mutex> lock( mutex );
output.append( bytes, count );
changed.notify_all();
} );
{
std::unique_lock<std::mutex> lock( mutex );
ASSERT_TRUE( changed.wait_for( lock, std::chrono::seconds( 10 ),
[&] { return output.find( '\n' ) != std::string::npos; } ) );
}
ASSERT_TRUE( process.isAlive() );
// The stale flags made this kill() return at once, leaving the child running.
EXPECT_TRUE( process.kill() );
EXPECT_FALSE( process.isAlive() );
#if EE_PLATFORM != EE_PLATFORM_WIN
const pid_t pid = static_cast<pid_t>( std::strtol( output.c_str(), nullptr, 10 ) );
ASSERT_GT( pid, 0 );
const auto until = std::chrono::steady_clock::now() + std::chrono::seconds( 2 );
bool gone = false;
while ( !gone && std::chrono::steady_clock::now() < until ) {
gone = ::kill( pid, 0 ) != 0 && errno == ESRCH;
if ( !gone )
std::this_thread::sleep_for( std::chrono::milliseconds( 5 ) );
}
EXPECT_TRUE( gone );
#endif
}
UTEST( Process, killNeverWaitsForAReadCallback ) {
// Like LSPClientServer's destructor: kill() while holding a lock the read callback takes.
std::mutex mutex;
std::atomic<bool> delivering{ false };
Process process;
ASSERT_TRUE( createChatty( process ) );
process.startAsyncRead( [&]( const char*, size_t ) {
delivering = true;
std::lock_guard<std::mutex> lock( mutex );
} );
std::unique_lock<std::mutex> lock( mutex );
const auto until = std::chrono::steady_clock::now() + std::chrono::seconds( 2 );
while ( !delivering && std::chrono::steady_clock::now() < until )
std::this_thread::sleep_for( std::chrono::milliseconds( 1 ) );
ASSERT_TRUE( delivering.load() );
std::atomic<bool> killed{ false };
std::thread killing( [&] {
process.kill();
killed = true;
} );
const auto deadline = std::chrono::steady_clock::now() + std::chrono::seconds( 2 );
while ( !killed && std::chrono::steady_clock::now() < deadline )
std::this_thread::sleep_for( std::chrono::milliseconds( 1 ) );
EXPECT_TRUE( killed.load() );
lock.unlock();
killing.join();
}
#if EE_PLATFORM != EE_PLATFORM_WIN
UTEST( Process, killedChildrenAreReaped ) {
std::mutex mutex;
std::condition_variable changed;
std::string output;
Process child;
ASSERT_TRUE( child.create( "/bin/sh",
std::vector<std::string>{ "-c", "echo $$; exec sleep 30" },
Process::getDefaultOptions() | Process::EnableAsync ) );
child.startAsyncRead( [&]( const char* bytes, size_t count ) {
std::lock_guard<std::mutex> lock( mutex );
output.append( bytes, count );
changed.notify_all();
} );
pid_t pid;
{
std::unique_lock<std::mutex> lock( mutex );
ASSERT_TRUE( changed.wait_for( lock, std::chrono::seconds( 2 ),
[&] { return output.find( '\n' ) != std::string::npos; } ) );
pid = static_cast<pid_t>( std::strtol( output.c_str(), nullptr, 10 ) );
}
ASSERT_GT( pid, 0 );
EXPECT_TRUE( child.kill() );
// A zombie still answers kill( pid, 0 ); a reaped child is gone.
const auto until = std::chrono::steady_clock::now() + std::chrono::seconds( 2 );
bool gone = false;
while ( !gone && std::chrono::steady_clock::now() < until ) {
gone = ::kill( pid, 0 ) != 0 && errno == ESRCH;
if ( !gone )
std::this_thread::sleep_for( std::chrono::milliseconds( 5 ) );
}
EXPECT_TRUE( gone );
}
#endif
UTEST( Process, teardownDoesNotWaitForAGrandchildHoldingThePipe ) {
// The grandchild keeps the output pipe open after its parent is killed, without writing to
// it: on Windows only cancelling the reader's pending read releases it.
std::mutex mutex;
std::condition_variable changed;
std::string output;
std::optional<Process> process( std::in_place );
ASSERT_TRUE( createShell( process.value(), "sleep 3 & printf ready; wait",
"start /b cmd /D /C \"ping -n 6 127.0.0.1 >nul\"& echo ready& "
"ping -n 30 127.0.0.1 >nul" ) );
process->startAsyncRead( [&]( const char* bytes, size_t count ) {
std::lock_guard<std::mutex> lock( mutex );
output.append( bytes, count );
changed.notify_all();
} );
{
std::unique_lock<std::mutex> lock( mutex );
ASSERT_TRUE( changed.wait_for( lock, std::chrono::seconds( 10 ), [&] {
return output.find( "ready" ) != std::string::npos;
} ) );
}
const auto started = std::chrono::steady_clock::now();
process->kill();
process.reset();
const auto elapsed = std::chrono::duration_cast<std::chrono::milliseconds>(
std::chrono::steady_clock::now() - started );
EXPECT_LT( elapsed.count(), 1000 );
}
#endif