From cfa76e762abe22cab74349f8427bdf2e15b34fd6 Mon Sep 17 00:00:00 2001 From: dockfries Date: Thu, 10 Sep 2026 23:46:57 +0800 Subject: [PATCH 1/3] Console: read stdin via poll+read on POSIX so a non-blocking fd 0 can't kill the reader --- Server/Components/Console/console_impl.hpp | 74 ++++++++++++++++++++++ 1 file changed, 74 insertions(+) diff --git a/Server/Components/Console/console_impl.hpp b/Server/Components/Console/console_impl.hpp index 474a03377..9d3522657 100644 --- a/Server/Components/Console/console_impl.hpp +++ b/Server/Components/Console/console_impl.hpp @@ -21,6 +21,12 @@ #include #include +#ifndef WIN32 +#include +#include +#include +#endif + using namespace Impl; class PlayerConsoleData final : public IPlayerConsoleData @@ -234,6 +240,73 @@ class ConsoleComponent final : public IConsoleComponent, public CoreEventHandler } } +#ifndef WIN32 + static void ThreadProc(ThreadProcData* threadData) + { + // Read stdin with a blocking poll()+read() pair instead of + // std::getline(std::wcin). Some edge-case plugins/components can flip fd 0 + // into non-blocking mode, which would make the wide-stream getline() fail + // with EAGAIN and permanently kill this thread. poll() also ignores + // O_NONBLOCK, and being a cancellation point it keeps the destructor's + // immediate pthread_cancel() working. + char buffer[1024]; + String pending; + + while (true) + { + pollfd fd { STDIN_FILENO, POLLIN, 0 }; + if (poll(&fd, 1, -1) < 0) + { + if (errno == EINTR) + { + continue; + } + threadData->isRunning = false; + return; + } + + const ssize_t bytesRead = read(STDIN_FILENO, buffer, sizeof(buffer)); + if (bytesRead < 0) + { + if (errno == EAGAIN || errno == EWOULDBLOCK || errno == EINTR) + { + continue; + } + threadData->isRunning = false; + return; + } + if (bytesRead == 0) + { + // EOF, handled the same way as the old getline() failure. + threadData->isRunning = false; + return; + } + + for (ssize_t i = 0; i < bytesRead; ++i) + { + const char c = buffer[i]; + if (c != '\n' && c != '\r') + { + pending.push_back(c); + continue; + } + + // Normalise CRLF (and a lone CR from a raw-mode terminal) to one line break. + if (c == '\r' && i + 1 < bytesRead && buffer[i + 1] == '\n') + { + ++i; + } + if (!pending.empty()) + { + std::scoped_lock lock(threadData->component->cmdMutex); + threadData->component->cmd = std::move(pending); + threadData->component->newCmd = true; + pending.clear(); + } + } + } + } +#else static void ThreadProc(ThreadProcData* threadData) { std::wstring line; @@ -255,6 +328,7 @@ class ConsoleComponent final : public IConsoleComponent, public CoreEventHandler } } } +#endif ~ConsoleComponent() { From 52b8f2e436791f78244ce9bba1771889d33a2438 Mon Sep 17 00:00:00 2001 From: Amyr Ahmady Date: Sun, 11 Oct 2026 05:48:23 +0330 Subject: [PATCH 2/3] Console: keep POSIX stdin reader responsive and preserve commands --- Server/Components/Console/console_impl.hpp | 144 +++++++++++++++------ 1 file changed, 105 insertions(+), 39 deletions(-) diff --git a/Server/Components/Console/console_impl.hpp b/Server/Components/Console/console_impl.hpp index 9d3522657..4e19d06af 100644 --- a/Server/Components/Console/console_impl.hpp +++ b/Server/Components/Console/console_impl.hpp @@ -16,6 +16,7 @@ #include #include #include +#include #include #include #include @@ -23,7 +24,10 @@ #ifndef WIN32 #include +#include #include +#include +#include #include #endif @@ -62,17 +66,19 @@ class ConsoleComponent final : public IConsoleComponent, public CoreEventHandler { std::atomic_bool valid; ConsoleComponent* component; - std::atomic_bool isRunning; }; ICore* core = nullptr; DefaultEventDispatcher eventDispatcher; std::mutex cmdMutex; - std::atomic_bool newCmd = false; - String cmd; + std::queue pendingCommands; ThreadProcData* threadData; std::thread cinThread; - std::thread::native_handle_type nativeThreadHandle; +#ifndef WIN32 + int inputFd = STDIN_FILENO; + int stopReadFd = -1; + int stopWriteFd = -1; +#endif struct PlayerRconCommandHandler : public SingleNetworkInEventHandler { @@ -157,6 +163,12 @@ class ConsoleComponent final : public IConsoleComponent, public CoreEventHandler } } playerRconCommandHandler; + void enqueueCommand(String command) + { + std::lock_guard lock(cmdMutex); + pendingCommands.emplace(std::move(command)); + } + public: void sendMessage(const ConsoleCommandSenderData& recipient, StringView message) override { @@ -218,10 +230,51 @@ class ConsoleComponent final : public IConsoleComponent, public CoreEventHandler NetCode::Packet::PlayerRconCommand::addEventHandler(*core, &playerRconCommandHandler); - threadData = new ThreadProcData { true, this, true }; +#ifndef WIN32 + int stopPipe[2]; + if (pipe(stopPipe) == -1) + { + core->logLn(LogLevel::Error, "Console: failed to create stdin reader wake pipe (errno %d)", errno); + return; + } + stopReadFd = stopPipe[0]; + stopWriteFd = stopPipe[1]; + if (fcntl(stopReadFd, F_SETFD, FD_CLOEXEC) == -1 + || fcntl(stopWriteFd, F_SETFD, FD_CLOEXEC) == -1) + { + const int error = errno; + close(stopReadFd); + close(stopWriteFd); + stopReadFd = stopWriteFd = -1; + core->logLn(LogLevel::Error, "Console: failed to configure stdin reader wake pipe (errno %d)", error); + return; + } + +#ifdef __linux__ + // Reopen non-regular stdin to get independent file status flags. dup() + // would share O_NONBLOCK with components using fd 0. Do not reopen a + // regular file: that would reset its offset and replay earlier input. + struct stat stdinStat; + if (fstat(STDIN_FILENO, &stdinStat) == 0 && !S_ISREG(stdinStat.st_mode)) + { + int fd = open("/proc/self/fd/0", O_RDONLY | O_NONBLOCK | O_CLOEXEC | O_NOCTTY); + if (fd != -1) + { + inputFd = fd; + } + else + { + core->logLn(LogLevel::Warning, "Console: could not open an independent non-blocking stdin handle (errno %d)", errno); + } + } +#endif +#endif + + threadData = new ThreadProcData { true, this }; cinThread = std::thread(ThreadProc, threadData); - nativeThreadHandle = cinThread.native_handle(); +#ifdef WIN32 cinThread.detach(); +#endif } void onReady() override @@ -243,42 +296,45 @@ class ConsoleComponent final : public IConsoleComponent, public CoreEventHandler #ifndef WIN32 static void ThreadProc(ThreadProcData* threadData) { - // Read stdin with a blocking poll()+read() pair instead of - // std::getline(std::wcin). Some edge-case plugins/components can flip fd 0 - // into non-blocking mode, which would make the wide-stream getline() fail - // with EAGAIN and permanently kill this thread. poll() also ignores - // O_NONBLOCK, and being a cancellation point it keeps the destructor's - // immediate pthread_cancel() working. + ConsoleComponent* component = threadData->component; char buffer[1024]; String pending; - while (true) + while (threadData->valid) { - pollfd fd { STDIN_FILENO, POLLIN, 0 }; - if (poll(&fd, 1, -1) < 0) + pollfd fds[2] { + { component->inputFd, POLLIN, 0 }, + { component->stopReadFd, POLLIN, 0 }, + }; + if (poll(fds, 2, -1) < 0) { if (errno == EINTR) { continue; } - threadData->isRunning = false; + return; + } + if (fds[1].revents || !threadData->valid) + { return; } - const ssize_t bytesRead = read(STDIN_FILENO, buffer, sizeof(buffer)); + const ssize_t bytesRead = read(component->inputFd, buffer, sizeof(buffer)); if (bytesRead < 0) { if (errno == EAGAIN || errno == EWOULDBLOCK || errno == EINTR) { continue; } - threadData->isRunning = false; return; } if (bytesRead == 0) { - // EOF, handled the same way as the old getline() failure. - threadData->isRunning = false; + // getline() also returns the final, unterminated line at EOF. + if (!pending.empty() && threadData->valid) + { + component->enqueueCommand(std::move(pending)); + } return; } @@ -291,16 +347,14 @@ class ConsoleComponent final : public IConsoleComponent, public CoreEventHandler continue; } - // Normalise CRLF (and a lone CR from a raw-mode terminal) to one line break. + // Normalise CRLF (and a lone CR) to one line break. if (c == '\r' && i + 1 < bytesRead && buffer[i + 1] == '\n') { ++i; } - if (!pending.empty()) + if (!pending.empty() && threadData->valid) { - std::scoped_lock lock(threadData->component->cmdMutex); - threadData->component->cmd = std::move(pending); - threadData->component->newCmd = true; + component->enqueueCommand(std::move(pending)); pending.clear(); } } @@ -316,14 +370,10 @@ class ConsoleComponent final : public IConsoleComponent, public CoreEventHandler std::getline(std::wcin, line) && threadData->valid) { - std::scoped_lock lock(threadData->component->cmdMutex); - - threadData->component->cmd = std::wstring_convert, wchar_t>().to_bytes(line); - threadData->component->newCmd = true; + threadData->component->enqueueCommand(std::wstring_convert, wchar_t>().to_bytes(line)); } else { - threadData->isRunning = false; return; } } @@ -342,10 +392,24 @@ class ConsoleComponent final : public IConsoleComponent, public CoreEventHandler cinThread.join(); } #else - if (threadData->isRunning) + const char wake = 0; + ssize_t wakeResult; + do + { + wakeResult = write(stopWriteFd, &wake, sizeof(wake)); + } while (wakeResult == -1 && errno == EINTR); + if (wakeResult == -1 || inputFd == STDIN_FILENO) + { + // The original stdin may still block in read(). + pthread_cancel(cinThread.native_handle()); + } + cinThread.join(); + if (inputFd != STDIN_FILENO) { - pthread_cancel(nativeThreadHandle); + close(inputFd); } + close(stopReadFd); + close(stopWriteFd); #endif delete threadData; @@ -413,15 +477,17 @@ class ConsoleComponent final : public IConsoleComponent, public CoreEventHandler void onTick(Microseconds elapsed, TimePoint now) override { - bool expected = true; - if (newCmd.compare_exchange_weak(expected, false)) + String command; { - cmdMutex.lock(); - String command = cmd; - cmdMutex.unlock(); - - send(command); + std::lock_guard lock(cmdMutex); + if (pendingCommands.empty()) + { + return; + } + command = std::move(pendingCommands.front()); + pendingCommands.pop(); } + send(command); } bool onConsoleText(StringView command, StringView parameters, const ConsoleCommandSenderData& sender) override; From 3b002c98ec658efcea972397dac802bd95dad266 Mon Sep 17 00:00:00 2001 From: dockfries Date: Mon, 19 Oct 2026 05:20:58 +0800 Subject: [PATCH 3/3] Rework console stdin reader lifecycle and input fd ownership --- Server/Components/Console/console_impl.hpp | 453 ++++++++++++++++----- 1 file changed, 341 insertions(+), 112 deletions(-) diff --git a/Server/Components/Console/console_impl.hpp b/Server/Components/Console/console_impl.hpp index 4e19d06af..a57b23c3d 100644 --- a/Server/Components/Console/console_impl.hpp +++ b/Server/Components/Console/console_impl.hpp @@ -12,23 +12,25 @@ #include #include #include -#include -#include -#include +#include +#include #include -#include #include #include #include +#include #include +#include #ifndef WIN32 #include +#include +#include #include #include -#include -#include #include +#else +#include #endif using namespace Impl; @@ -62,6 +64,10 @@ class PlayerConsoleData final : public IPlayerConsoleData class ConsoleComponent final : public IConsoleComponent, public CoreEventHandler, public ConsoleEventHandler, public PlayerConnectEventHandler { private: + // Hard limits to keep a runaway input stream from growing memory without bound. + static constexpr size_t MAX_LINE_LENGTH = 4096; + static constexpr size_t MAX_PENDING_COMMANDS = 4096; + struct ThreadProcData { std::atomic_bool valid; @@ -71,13 +77,22 @@ class ConsoleComponent final : public IConsoleComponent, public CoreEventHandler ICore* core = nullptr; DefaultEventDispatcher eventDispatcher; std::mutex cmdMutex; - std::queue pendingCommands; - ThreadProcData* threadData; + std::vector pendingCmds; + std::vector tickCmds; // reused by onTick to avoid per-tick allocation + std::atomic_bool cmdOverflowWarned { false }; + ThreadProcData* threadData = nullptr; std::thread cinThread; #ifndef WIN32 - int inputFd = STDIN_FILENO; - int stopReadFd = -1; - int stopWriteFd = -1; + // Our own fd for reading input. When stdin is a TTY we reopen the + // terminal so we get a file description with independent file status + // flags (e.g. O_NONBLOCK) and offset, rather than sharing fd 0's. + // Note this is NOT an independent data stream: on a pty both fds still + // read the same queue, which is only safe because we are the sole reader. + int inputFd = -1; + // Self-pipe used to wake the reader thread from poll() during shutdown. + int wakePipe[2] = { -1, -1 }; +#else + HANDLE stdinHandle = INVALID_HANDLE_VALUE; #endif struct PlayerRconCommandHandler : public SingleNetworkInEventHandler @@ -163,12 +178,6 @@ class ConsoleComponent final : public IConsoleComponent, public CoreEventHandler } } playerRconCommandHandler; - void enqueueCommand(String command) - { - std::lock_guard lock(cmdMutex); - pendingCommands.emplace(std::move(command)); - } - public: void sendMessage(const ConsoleCommandSenderData& recipient, StringView message) override { @@ -191,10 +200,11 @@ class ConsoleComponent final : public IConsoleComponent, public CoreEventHandler } break; case ConsoleCommandSender::Custom: - { - recipient.handler->handleConsoleMessage(message); + if (recipient.handler) + { + recipient.handler->handleConsoleMessage(message); + } break; - } default: break; } @@ -211,8 +221,7 @@ class ConsoleComponent final : public IConsoleComponent, public CoreEventHandler } ConsoleComponent() - : threadData(nullptr) - , playerRconCommandHandler(*this) + : playerRconCommandHandler(*this) { } @@ -231,50 +240,80 @@ class ConsoleComponent final : public IConsoleComponent, public CoreEventHandler NetCode::Packet::PlayerRconCommand::addEventHandler(*core, &playerRconCommandHandler); #ifndef WIN32 - int stopPipe[2]; - if (pipe(stopPipe) == -1) + // If stdin is a TTY, reopen it to get a file description with its own + // status flags. Never fcntl() fd 0 directly: its file description is + // shared with the host process and any other readers + // so changing its flags — such as clearing O_NONBLOCK — would + // silently alter their blocking behaviour. + if (isatty(STDIN_FILENO)) { - core->logLn(LogLevel::Error, "Console: failed to create stdin reader wake pipe (errno %d)", errno); - return; - } - stopReadFd = stopPipe[0]; - stopWriteFd = stopPipe[1]; - if (fcntl(stopReadFd, F_SETFD, FD_CLOEXEC) == -1 - || fcntl(stopWriteFd, F_SETFD, FD_CLOEXEC) == -1) - { - const int error = errno; - close(stopReadFd); - close(stopWriteFd); - stopReadFd = stopWriteFd = -1; - core->logLn(LogLevel::Error, "Console: failed to configure stdin reader wake pipe (errno %d)", error); - return; + char path[PATH_MAX]; + if (ttyname_r(STDIN_FILENO, path, sizeof(path)) == 0) + { + inputFd = open(path, O_RDONLY | O_NOCTTY | O_CLOEXEC); + } + if (inputFd == -1) + { + inputFd = open("/dev/tty", O_RDONLY | O_NOCTTY | O_CLOEXEC); + } + if (inputFd == -1) + { + core->logLn(LogLevel::Warning, "Console: could not reopen the terminal, falling back to fd 0 (errno %d)", errno); + } } -#ifdef __linux__ - // Reopen non-regular stdin to get independent file status flags. dup() - // would share O_NONBLOCK with components using fd 0. Do not reopen a - // regular file: that would reset its offset and replay earlier input. - struct stat stdinStat; - if (fstat(STDIN_FILENO, &stdinStat) == 0 && !S_ISREG(stdinStat.st_mode)) + if (pipe(wakePipe) == 0) { - int fd = open("/proc/self/fd/0", O_RDONLY | O_NONBLOCK | O_CLOEXEC | O_NOCTTY); - if (fd != -1) + int flags = fcntl(wakePipe[0], F_GETFL); + if (flags != -1) { - inputFd = fd; + fcntl(wakePipe[0], F_SETFL, flags | O_NONBLOCK); } - else + flags = fcntl(wakePipe[1], F_GETFL); + if (flags != -1) { - core->logLn(LogLevel::Warning, "Console: could not open an independent non-blocking stdin handle (errno %d)", errno); + fcntl(wakePipe[1], F_SETFL, flags | O_NONBLOCK); } + fcntl(wakePipe[0], F_SETFD, FD_CLOEXEC); + fcntl(wakePipe[1], F_SETFD, FD_CLOEXEC); + } + else + { + wakePipe[0] = -1; + wakePipe[1] = -1; + core->logLn(LogLevel::Warning, "Console: could not create wake pipe (errno %d)", errno); + } +#else + stdinHandle = GetStdHandle(STD_INPUT_HANDLE); + if (stdinHandle != INVALID_HANDLE_VALUE) + { + DWORD consoleMode = 0; + if (GetConsoleMode(stdinHandle, &consoleMode)) + { + // Keep line input and echo enabled so the console itself + // performs echo, line editing and IME composition. We only + // receive complete lines from ReadConsoleW. + SetConsoleMode(stdinHandle, consoleMode | ENABLE_LINE_INPUT | ENABLE_ECHO_INPUT | ENABLE_PROCESSED_INPUT); + } + } + else + { + core->logLn(LogLevel::Warning, "Console: GetStdHandle(STD_INPUT_HANDLE) returned an invalid handle; console input will be unavailable"); } -#endif #endif - threadData = new ThreadProcData { true, this }; - cinThread = std::thread(ThreadProc, threadData); -#ifdef WIN32 - cinThread.detach(); -#endif + std::unique_ptr data(new ThreadProcData { true, this }); + threadData = data.get(); + try + { + cinThread = std::thread(ThreadProc, threadData); + data.release(); + } + catch (const std::exception& e) + { + threadData = nullptr; + core->logLn(LogLevel::Error, "Console: could not start input thread: %s", e.what()); + } } void onReady() override @@ -293,49 +332,99 @@ class ConsoleComponent final : public IConsoleComponent, public CoreEventHandler } } + static void submitLine(ThreadProcData* threadData, const char* begin, size_t len) + { + ConsoleComponent* component = threadData->component; + bool warnOverflow = false; + { + std::scoped_lock lock(component->cmdMutex); + if (component->pendingCmds.size() >= MAX_PENDING_COMMANDS) + { + warnOverflow = !component->cmdOverflowWarned.exchange(true); + } + else + { + component->cmdOverflowWarned.store(false); + component->pendingCmds.emplace_back(begin, len); + } + } + // Log outside the lock so a slow/hooked logger can't stall the reader. + if (warnOverflow) + { + component->core->logLn( + LogLevel::Warning, + "Console: pending command queue is full (%zu), dropping input until it drains", + MAX_PENDING_COMMANDS); + } + } + #ifndef WIN32 static void ThreadProc(ThreadProcData* threadData) { ConsoleComponent* component = threadData->component; + const int fd = (component->inputFd != -1) ? component->inputFd : STDIN_FILENO; + const int wakeFd = component->wakePipe[0]; + char buffer[1024]; String pending; + bool eof = false; - while (threadData->valid) + while (threadData->valid && !eof) { - pollfd fds[2] { - { component->inputFd, POLLIN, 0 }, - { component->stopReadFd, POLLIN, 0 }, - }; - if (poll(fds, 2, -1) < 0) + pollfd fds[2]; + nfds_t nfds = 1; + fds[0] = { fd, POLLIN, 0 }; + if (wakeFd != -1) + { + fds[1] = { wakeFd, POLLIN, 0 }; + nfds = 2; + } + + // Timeout is a backstop in case the wake pipe was not created + // or the write side never gets a chance to fire. + const int ret = poll(fds, nfds, 200); + if (ret < 0) { if (errno == EINTR) { continue; } - return; + break; + } + if (ret == 0) + { + continue; } - if (fds[1].revents || !threadData->valid) + if (nfds > 1 && (fds[1].revents & (POLLIN | POLLHUP | POLLERR))) { - return; + // Wake request, or the write end was closed: either way stop. + break; + } + if (fds[0].revents & (POLLERR | POLLNVAL)) + { + break; + } + if (!(fds[0].revents & (POLLIN | POLLHUP))) + { + continue; } - const ssize_t bytesRead = read(component->inputFd, buffer, sizeof(buffer)); + const ssize_t bytesRead = read(fd, buffer, sizeof(buffer)); if (bytesRead < 0) { + // When we fell back to fd 0, its O_NONBLOCK flag may have + // been set by another reader in the same process, so a + // non-blocking read can legitimately return EAGAIN here. if (errno == EAGAIN || errno == EWOULDBLOCK || errno == EINTR) { continue; } - return; + break; } if (bytesRead == 0) { - // getline() also returns the final, unterminated line at EOF. - if (!pending.empty() && threadData->valid) - { - component->enqueueCommand(std::move(pending)); - } - return; + eof = true; + break; } for (ssize_t i = 0; i < bytesRead; ++i) @@ -343,38 +432,165 @@ class ConsoleComponent final : public IConsoleComponent, public CoreEventHandler const char c = buffer[i]; if (c != '\n' && c != '\r') { - pending.push_back(c); + if (pending.size() < MAX_LINE_LENGTH) + { + pending.push_back(c); + } continue; } - - // Normalise CRLF (and a lone CR) to one line break. if (c == '\r' && i + 1 < bytesRead && buffer[i + 1] == '\n') { - ++i; + ++i; // CRLF counts as one break } if (!pending.empty() && threadData->valid) { - component->enqueueCommand(std::move(pending)); + submitLine(threadData, pending.data(), pending.size()); pending.clear(); } } } + + // Flush a final line when stdin is closed (EOF) without a trailing + // newline. Only do this if the component is still alive. + if (eof && !pending.empty() && threadData->valid) + { + submitLine(threadData, pending.data(), pending.size()); + } } #else + static String wideToUtf8(const std::wstring& wide) + { + if (wide.empty()) + { + return String(); + } + const int size = WideCharToMultiByte(CP_UTF8, 0, wide.data(), static_cast(wide.size()), nullptr, 0, nullptr, nullptr); + if (size <= 0) + { + return String(); + } + String utf8(static_cast(size), '\0'); + WideCharToMultiByte(CP_UTF8, 0, wide.data(), static_cast(wide.size()), utf8.data(), size, nullptr, nullptr); + return utf8; + } + static void ThreadProc(ThreadProcData* threadData) { - std::wstring line; - while (true) + ConsoleComponent* component = threadData->component; + HANDLE handle = component->stdinHandle; + + if (handle == INVALID_HANDLE_VALUE) + { + // Nothing we can read from; onLoad already logged this. + return; + } + + DWORD mode = 0; + const bool isConsole = GetConsoleMode(handle, &mode) != 0; + + if (isConsole) { - if ( - std::getline(std::wcin, line) - && threadData->valid) + // Console path: blocking ReadConsoleW. The console performs + // echo, line editing and IME composition; we receive whole + // lines terminated by CR/LF. + std::wstring pending; + wchar_t buf[1024]; + while (threadData->valid) { - threadData->component->enqueueCommand(std::wstring_convert, wchar_t>().to_bytes(line)); + DWORD charsRead = 0; + if (!ReadConsoleW(handle, buf, static_cast(sizeof(buf) / sizeof(buf[0])), &charsRead, nullptr)) + { + // ERROR_OPERATION_ABORTED means CancelSynchronousIo fired. + break; + } + if (charsRead == 0) + { + // Cancellation may surface as TRUE with zero characters. + break; + } + + for (DWORD i = 0; i < charsRead; ++i) + { + const wchar_t ch = buf[i]; + if (ch != L'\r' && ch != L'\n') + { + if (pending.size() < MAX_LINE_LENGTH) + { + pending.push_back(ch); + } + continue; + } + if (ch == L'\r' && i + 1 < charsRead && buf[i + 1] == L'\n') + { + ++i; // CRLF counts as one break + } + if (!pending.empty() && threadData->valid) + { + const String utf8 = wideToUtf8(pending); + if (!utf8.empty()) + { + submitLine(threadData, utf8.data(), utf8.size()); + } + pending.clear(); + } + } } - else + } + else + { + // Redirected stdin: blocking ReadFile, cancelled via CancelSynchronousIo. + // No PeekNamedPipe, no WaitForMultipleObjects. + char buf[1024]; + String pending; + bool eof = false; + while (threadData->valid && !eof) { - return; + DWORD bytesRead = 0; + if (!ReadFile(handle, buf, static_cast(sizeof(buf)), &bytesRead, nullptr)) + { + const DWORD err = GetLastError(); + if (err == ERROR_BROKEN_PIPE || err == ERROR_OPERATION_ABORTED || err == ERROR_HANDLE_EOF) + { + eof = true; + } + else + { + component->core->logLn(LogLevel::Warning, "Console: ReadFile on stdin failed with error %lu", static_cast(err)); + } + break; + } + if (bytesRead == 0) + { + eof = true; + break; + } + + for (DWORD i = 0; i < bytesRead; ++i) + { + const char c = buf[i]; + if (c != '\n' && c != '\r') + { + if (pending.size() < MAX_LINE_LENGTH) + { + pending.push_back(c); + } + continue; + } + if (c == '\r' && i + 1 < bytesRead && buf[i + 1] == '\n') + { + ++i; + } + if (!pending.empty() && threadData->valid) + { + submitLine(threadData, pending.data(), pending.size()); + pending.clear(); + } + } + } + + if (eof && !pending.empty() && threadData->valid) + { + submitLine(threadData, pending.data(), pending.size()); } } } @@ -386,35 +602,50 @@ class ConsoleComponent final : public IConsoleComponent, public CoreEventHandler { threadData->valid = false; -#ifdef WIN32 - if (cinThread.joinable()) +#ifndef WIN32 + // Wake the reader out of poll() so it can observe valid == false. + if (wakePipe[1] != -1) { - cinThread.join(); + const char byte = 1; + while (write(wakePipe[1], &byte, 1) == -1 && errno == EINTR) + { + } } #else - const char wake = 0; - ssize_t wakeResult; - do - { - wakeResult = write(stopWriteFd, &wake, sizeof(wake)); - } while (wakeResult == -1 && errno == EINTR); - if (wakeResult == -1 || inputFd == STDIN_FILENO) + if (cinThread.joinable()) { - // The original stdin may still block in read(). - pthread_cancel(cinThread.native_handle()); + // Unblock ReadConsoleW / ReadFile on the reader thread. + CancelSynchronousIo(reinterpret_cast(cinThread.native_handle())); } - cinThread.join(); - if (inputFd != STDIN_FILENO) +#endif + + if (cinThread.joinable()) { - close(inputFd); + cinThread.join(); } - close(stopReadFd); - close(stopWriteFd); -#endif delete threadData; threadData = nullptr; } + +#ifndef WIN32 + if (inputFd != -1) + { + close(inputFd); + inputFd = -1; + } + if (wakePipe[0] != -1) + { + close(wakePipe[0]); + wakePipe[0] = -1; + } + if (wakePipe[1] != -1) + { + close(wakePipe[1]); + wakePipe[1] = -1; + } +#endif + if (core) { core->getEventDispatcher().removeEventHandler(this); @@ -477,17 +708,15 @@ class ConsoleComponent final : public IConsoleComponent, public CoreEventHandler void onTick(Microseconds elapsed, TimePoint now) override { - String command; { - std::lock_guard lock(cmdMutex); - if (pendingCommands.empty()) - { - return; - } - command = std::move(pendingCommands.front()); - pendingCommands.pop(); + std::scoped_lock lock(cmdMutex); + tickCmds.swap(pendingCmds); + } + for (const String& command : tickCmds) + { + send(command); } - send(command); + tickCmds.clear(); } bool onConsoleText(StringView command, StringView parameters, const ConsoleCommandSenderData& sender) override;