I have a server and a client. The server is responsible for forking child processes and interacting with them to get log outputs to send to the client. It also receives input from the client and sends it to the child process for execution. Since the child process does not need a socket, I closed it, but then I couldn’t receive any input from the client anymore. However, I could still send output to the client. As long as I remove the clientStream.close(); line, there is no problem (but I must close it in the child process). I don’t know what’s going wrong?

server code:

class Server
{
public:
    static Server *getInstance()
    {

        static Server pm;
        return ±
    }

    Server(const Server &) = delete;

    Server(Server &&) = delete;

    Server &operator=(const Server &) = delete;

    Server &operator=(Server &&) = delete;

    ~Server()
    {
        shutdown();
        clear();
    }

    void shutdown()
    {
        // close all clients?
        running = false;
        if (acc_->is_open())
        {

            acc_->close();
        }
    }
    void start()
    {
        boost::asio::io_context io_context;

        co_spawn(io_context, acceptClients(), detached);
        io_context.run();
    }

private:
    bool running{false};
    Server()
    {
        clear();
        fs::path dirPath = fs::path(IPC_SOCKET_PATH).parent_path();

        if (!std::filesystem::exists(dirPath))
        {
            try
            {
                fs::create_directories(dirPath);
            }
            catch (const fs::filesystem_error &e)
            {
                std::cerr << "Error creating directory: " << e.what() << std::endl;
            }
        }
    }
    void clear()
    {
        if (std::filesystem::exists(IPC_SOCKET_PATH))
        {
            if (!std::filesystem::remove(IPC_SOCKET_PATH))
            {
                RUNTIME_ABORT("Failed to remove existing socket file");
            }
        }
    }

    awaitable<void> acceptClients()
    {
        try
        {
            auto ex = co_await asio::this_coro::executor;
            // TODO: chmod(ipc_socket.c_str(), S_IRUSR | S_IWUSR | S_IXUSR | S_IRGRP | S_IWGRP | S_IXGRP | S_IROTH |
            // S_IWOTH | S_IXOTH); 测试没有socket文件的情况
            acc_ = std::make_unique<asio::local::stream_protocol::acceptor>(ex, IPC_SOCKET_PATH);
            acc_->listen(5);
            LOG_FN(INFO, "Server listening");
            // set connected
            running = true;
            while (running)
            {
                asio::local::stream_protocol::socket client(ex);
                co_await acc_->async_accept(client, use_awaitable);
                co_spawn(ex, handleClient(std::move(client)), asio::detached);
            }
        }
        catch (const std::exception &e)
        {
            printStacktrace();
            LOG_FN(ERR, "Error accepting client - {}", e.what());
        }
    }

    awaitable<void> onAsyncSpawn(const std::string &appId, asio::local::stream_protocol::socket &client,
                                 bool force = false)
    {
        try
        {
            co_await ProcessManager::getInstance()->asyncSpawn(appId, std::move(client), force);
        }
        catch (std::exception &e)
        {
            LOG_FN(ERR, "[spawn] Failed - {}", e.what());
        }
    }

    awaitable<void> handleClient(asio::local::stream_protocol::socket client)
    {
        LOG_FN(INFO, "Accepted new connection, waiting for appId");
        auto [appId, msgtype] = co_await receiveAppId(client);
        LOG_FN(INFO, "Received appId[{}]", appId);
        switch (msgtype)
        {
        case 2:
            co_await onAsyncSpawn(appId, client);
            break;
        default:
            break;
        }
    }

    awaitable<std::pair<std::string, int>> receiveAppId(asio::local::stream_protocol::socket &client)
    {
        std::string buf;
        co_await async_read_until(client, asio::dynamic_buffer(buf), 'n', asio::deferred);
        Message msg;
        msg.unpack(buf.data(), buf.size());
        auto appId = msg.payload;
        auto type = msg.msgtype;
        while (appId.ends_with('n'))
        {
            appId.pop_back();
        }

        co_return std::make_pair(appId, type);
    }

private:
    char buffer_[65535];
    std::unique_ptr<asio::local::stream_protocol::acceptor> acc_;
};


awaitable<void> ProcessV2::asyncSpawn(const spawnParam &param, asio::local::stream_protocol::socket clientStream)
{
    setState(ProcessState::STARGING);

    int pipe_stdout[2]; // 用于子进程的标准输出
    int pipe_stdin[2];  // 用于子进程的标准输入
    if (pipe(pipe_stdout) == -1 || pipe(pipe_stdin) == -1)
    {
        LOG_FN(ERR, "Failed to create pipes for process communication");
        co_return;
    }

    try
    {
        session_ = std::make_unique<Session<asio::local::stream_protocol::socket>>(std::move(clientStream), appId_,
                                                                                   pid_, "NovRunSession");
        pid_t pid = fork();
        if (pid == 0)
        {
            session_->disconnect();
            signal(SIGPIPE, SIG_IGN); // 忽略 SIGPIPE, 避免托管进程受影响

            close(pipe_stdout[0]);               // 关闭读端
            dup2(pipe_stdout[1], STDOUT_FILENO); // 将标准输出重定向到管道的写端
            dup2(pipe_stdout[1], STDERR_FILENO); // 将标准输出重定向到管道的写端
            close(pipe_stdout[1]);               // 关闭写端

            close(pipe_stdin[1]);              // 关闭写端
            dup2(pipe_stdin[0], STDIN_FILENO); // 将标准输入重定向到管道的读端
            close(pipe_stdin[0]);              // 关闭读端

            setsid(); // 创建新会话，避免子进程受父进程影响

            // 配置子进程的用户和工作目录
            if (setuid(getpwnam("tech")->pw_uid) == -1)
            {
                std::cerr << "Failed to set user ID to tech" << std::endl;
                exit(1);
            }
            if (chdir("/data/run_white/") != 0)
            {
                std::cerr << "Failed to change directory to /data/run_white/" << std::endl;
                exit(1);
            }

            execl("/data/prod/white_smain/build/build/builds/broker88/white_broker88", "white_broker88", NULL);
            std::cerr << "Failed to execute command: " << strerror(errno) << std::endl;
            exit(EXIT_FAILURE);
        }
        else
        {
            LOG_FN(INFO, "session is_open[{}]", session_->get().is_open());
            // 父进程
            pid_ = pid;
            std::cout << "Child process PID: " << pid_ << ", appId: " << appId_ << std::endl;
            setState(ProcessState::RUNNING);

            close(pipe_stdout[1]); // 关闭子进程写端
            close(pipe_stdin[0]);  // 关闭子进程读端

            auto ex = co_await asio::this_coro::executor;

            outputStream_ = std::make_unique<Session<asio::posix::stream_descriptor>>(
                boost::asio::posix::stream_descriptor(ex, pipe_stdout[0]), appId_, pid_, "OutputStream");
            inputStream_ = std::make_unique<Session<asio::posix::stream_descriptor>>(
                boost::asio::posix::stream_descriptor(ex, pipe_stdin[1]), appId_, pid_, "InputStream");

            co_spawn(ex, handleOutput(), asio::detached);
            co_spawn(ex, handleInput(), asio::detached);

            co_return;
        }
    }
    catch (const boost::system::system_error &se)
    {
        LOG_FN(ERR, "Failed to run process: {}", se.code().message());
    }
    catch (std::exception &e)
    {
        printStacktrace();
        API_FATAL("Failed to spawn - {}", e.what());
    }
}

awaitable<void> ProcessV2::handleOutput()
{
    try
    {
        char buffer[65535];
        while (running() && session_->connected())
        {
            size_t n = 0;
            try
            {
                n = co_await outputStream_->get().async_read_some(asio::buffer(buffer), use_awaitable);
            }
            catch (const boost::system::system_error &e)
            {
                outputStream_->handleException(e);
            }
            if (n == 0)
            {
                break;
            }

            auto output = std::string(buffer, n);
            output = remove_trailing_newline(output);
            if (output.empty())
            {
                continue;
            }
            char msgBuf[2046];
            Message msg;
            msg.msgtype = 1;
            msg.payload = output;
            auto totalLen = msg.pack(msgBuf, n);
            try
            {
                co_await asio::async_write(session_->get(), asio::buffer(msgBuf, totalLen), use_awaitable);
            }
            catch (const boost::system::system_error &e)
            {
                session_->handleException(e);
            }
        }
    }
    catch (const std::exception &e)
    {
        LOG_FN(ERR, "Error in handle_output: {}", e.what());
    }

    LOG_FN(INFO, "[{}] appId[{}] pid[{}] coroutine closed", __FUNCTION__, appId_, pid_);
}

void ProcessV2::onSignal(const Message &msg)
{
    if (msg.msgtype != 3)
        return;
    int signal = std::stoi(msg.payload);
    std::cout << "kill: " << pid_ << std::endl;
    kill(pid_, signal);
}

awaitable<std::size_t> ProcessV2::readFromClient(char *buffer, size_t bufferSize)
{
    try
    {
        if (session_->get().is_open())
        {
            LOG_FN(INFO, "session_ is opened");
        }
        co_return co_await session_->get().async_read_some(asio::buffer(buffer, bufferSize), use_awaitable);
    }
    catch (const boost::system::system_error &e)
    {
        session_->handleException(e);
        co_return 0;
    }
}

awaitable<void> ProcessV2::handleInput()
{
    try
    {
        char buffer[2046] = {};

        while (running() && session_->connected())
        {
            LOG_FN(INFO, "input 1");
            // read data
            std::size_t bytes_read = co_await readFromClient(buffer, sizeof(buffer));
            LOG_FN(INFO, "received[{}]", bytes_read);
        }
    }
    catch (const std::exception &e)
    {
        LOG_FN(ERR, "appId[()] pid[{}] Err in handleInput: {}", appId_, pid_, e.what());
    }

    LOG_FN(INFO, "[{}] appId[{}] pid[{}] coroutine closed", __FUNCTION__, appId_, pid_);
}

client code:

namespace test
{

class testRun
{
public:
    testRun(asio::io_context &io_context)
        : io_context_(io_context)
        , socket_(io_context)
    {
    }

    awaitable<void> run(bool force = false)
    {
        try
        {
            co_await connect();

            co_await spawn(force);
            co_spawn(io_context_, output(), asio::detached);
            std::thread inputThread([this]() { this->input(); });
            inputThread.detach();
        }
        catch (const std::exception &e)
        {
            LOG_FN(ERR, "[test_run] Failed - {}", e.what());
        }
    }

    void stop()
    {
        if (running_)
        {
            running_ = false;
        }
        if (socket_.is_open())
        {
            socket_.close();
            LOG_FN(INFO, "[test_run] Exiting...");
        }
    }

    void handleSignal(int signal) { co_spawn(io_context_, sendSignalToServer(signal), boost::asio::detached); }

    awaitable<void> sendSignalToServer(int signal)
    {
        Message msg;
        msg.msgtype = 3;
        msg.payload = std::to_string(signal);
        char buf[1024];
        auto size = msg.pack(buf, msg.payload.size());

        co_await async_write(socket_, buffer(buf, size), use_awaitable);
        // LOG_FN(INFO, "[test_run] Sent signal [{}] to server", signal);
    }


private:
    asio::io_context &io_context_;
    local::stream_protocol::socket socket_;
    hammer::tcp::StreamBuffer recvBuffer_;
    bool running_{true};
    std::string appId_{"999999"};


    awaitable<void> connect()
    {
        if (!std::filesystem::exists(IPC_SOCKET_PATH))
        {
            throw std::runtime_error(fmt::format("socket path not exist: {}", IPC_SOCKET_PATH));
        }
        local::stream_protocol::endpoint endpoint(IPC_SOCKET_PATH);
        co_await socket_.async_connect(endpoint, use_awaitable);
        LOG_FN(INFO, "[test_run] Connected to testd");
    }

    awaitable<void> spawn(bool force = false)
    {
        Message msg;
        msg.msgtype = force ? 6 : 2;
        msg.payload = appId_ + "n";
        char buf[1024];
        auto size = msg.pack(buf, msg.payload.size());

        co_await async_write(socket_, buffer(buf, size), use_awaitable);
        LOG_FN(INFO, "[test_run] spawn appId: {}", appId_);
    }

    awaitable<void> output()
    {
        while (running_)
        {
            try
            {

                char buffer[65535] = {};
                std::size_t n = co_await socket_.async_read_some(asio::buffer(buffer), use_awaitable);
                if (n == 0)
                {
                    break;
                }

                recvBuffer_.feed(buffer, n);
                while (true)
                {
                    // find magic num
                    auto str = recvBuffer_.find('01');
                    if (str.empty())
                    {
                        break;
                    }

                    // read head
                    auto headerBuf = recvBuffer_.acquire(HDR_LEN);
                    if (headerBuf.empty())
                    {
                        break;
                    }
                    Message msg;
                    msg.unpackHeader(headerBuf.data(), headerBuf.size());
                    auto totalLen = msg.len + HDR_LEN;

                    // read payload
                    auto data = recvBuffer_.acquire(totalLen);
                    if (data.empty())
                    {
                        break;
                    }

                    msg.unpack(data.data(), data.size());
                    std::cout << msg.payload << std::endl;

                    recvBuffer_.consume(totalLen);
                }

                recvBuffer_.commit();
            }
            catch (const boost::system::system_error &e)
            {
                if (e.code() == boost::asio::error::broken_pipe || e.code() == boost::asio::error::eof ||
                    e.code() == boost::asio::error::operation_aborted)
                {
                    LOG_FN(INFO, "[test_run] disconnected in output: {}", e.code().message());
                }
                else
                {
                    LOG_FN(ERR, "[test_run] unexpected error: {}", e.code().message());
                }
                break;
            }
            catch (std::exception &e)
            {
                LOG_FN(ERR, "[test_run] Error in output: {}", e.what());
                break;
            }
        }
        stop();
    }

    void input()
    {
        std::string input;
        while (std::getline(std::cin, input) && running_)
        {
            try
            {
                std::cout << "socket is open:" << socket_.is_open() << std::endl;
                if (input == "test_run exit")
                {
                    stop();
                    break;
                }
                if (input.empty())
                {
                    continue;
                }

                input += "n";
                char msgBuf[1024];
                Message msg;
                msg.msgtype = 4;
                msg.payload = input;
                auto totalLen = msg.pack(msgBuf, input.size());
                auto size = write(socket_, asio::buffer(msgBuf, totalLen));
                LOG_FN(INFO, "send size[{}]", size);
            }
            catch (const boost::system::system_error &e)
            {
                if (e.code() == boost::asio::error::broken_pipe || e.code() == boost::asio::error::eof ||
                    e.code() == boost::asio::error::operation_aborted)
                {
                    LOG_FN(INFO, "[test_run] disconnect: {}", e.code().message());
                }
                else
                {
                    LOG_FN(ERR, "[test_run] unexpected error: {}", e.code().message());
                }
                break;
            }
            catch (std::exception &e)
            {
                LOG_FN(ERR, "[test_run] Error in input: {}", e.what());
                break;
            }
        }
        stop();
    }
};
}

void setSignalHandlers(testRun &testRun)
{
    hammer::common::setSignalHandler(SIGINT, [&](auto ctx) { testRun.handleSignal(SIGINT); });
    struct sigaction act = {};
    act.sa_flags = SA_RESTART | SA_SIGINFO;
    act.sa_sigaction = white_logger::signalHandler;
    for (auto signum : {SIGTERM, SIGABRT, SIGFPE, SIGILL, SIGSEGV})
    {
        int ret = sigaction(signum, &act, nullptr);
        if (ret != 0)
        {
            throw std::runtime_error(fmt::format("::sigaction::{}", signum).c_str());
        }
    }
}

int main(int argc, char **argv)
{
    white_logger::WhiteLoggerGuard guard("");
    auto logger = white_logger::registerLogger("testd");
    logger->addConsoleHandler();
    LOGGER = std::move(logger);

    CLI::App app{"A Trading Binary Starter"};
    bool force = false;
    // force lanuch is not supported for now
    // app.add_flag("-f", force, "This option forces a takeover when the app is starting and is being taken over by
    // another session");
    CLI11_PARSE(app, argc, argv);

    try
    {
        asio::io_context io_context;

        testRun testRun(io_context);

        setSignalHandlers(testRun);

        co_spawn(io_context, testRun.run(force), asio::detached);

        io_context.run();
    }
    catch (const std::exception &e)
    {
        LOG_FN(ERR, "[test_run] Failed - {}", e.what());
        return 1;
    }

    return 0;
}