子进程只是在打印日志，为什么 wait 一直等不完？

"""Bounded child output; wait without reading versus communicate.

Only launches this same Python file. No shell, arbitrary command, or network.
"""
from pathlib import Path
from tempfile import TemporaryDirectory
import os
import subprocess
import sys
import time

BLOCK_SIZE = 65536
BLOCKS = 16


def child(marker):
    Path(marker).write_text("ready", encoding="ascii")
    for _ in range(BLOCKS):
        sys.stdout.buffer.write(b"O" * BLOCK_SIZE)
        sys.stdout.buffer.flush()
        sys.stderr.buffer.write(b"E" * BLOCK_SIZE)
        sys.stderr.buffer.flush()


def run_child(marker, wait_first):
    kwargs = {}
    if os.name == "nt":
        kwargs["creationflags"] = subprocess.CREATE_NO_WINDOW
    process = subprocess.Popen(
        [sys.executable, "-u", str(Path(__file__).resolve()), "--child", str(marker)],
        stdin=subprocess.DEVNULL, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
        shell=False, **kwargs,
    )
    timed_out = False
    try:
        ready_deadline = time.monotonic() + 5
        while not marker.exists():
            if process.poll() is not None or time.monotonic() >= ready_deadline:
                raise RuntimeError("child did not reach the output stage")
            time.sleep(0.01)
        if wait_first:
            try:
                process.wait(timeout=0.25)
            except subprocess.TimeoutExpired:
                timed_out = True
        # This drains BOTH pipes while waiting for this finite child to exit.
        output, errors = process.communicate(timeout=5)
        return timed_out, output, errors, process.returncode
    finally:
        if process.poll() is None:
            process.kill()
            process.communicate(timeout=5)
        if process.stdout is not None:
            process.stdout.close()
        if process.stderr is not None:
            process.stderr.close()


def main():
    checks = 0

    def check(condition, message):
        nonlocal checks
        if not condition:
            raise AssertionError(message)
        checks += 1

    expected_size = BLOCK_SIZE * BLOCKS
    with TemporaryDirectory(prefix="qinghuan-pipes-") as folder:
        root = Path(folder)
        delayed = run_child(root / "first.ready", wait_first=True)
        direct = run_child(root / "second.ready", wait_first=False)
        check(delayed[0], "wait without reading should time out for this output size")
        check(not direct[0], "the direct path does not use the blocking wait probe")
        for label, (_, output, errors, code) in zip(("rescued", "direct"), (delayed, direct)):
            check(code == 0, f"{label}: child must exit normally after draining")
            check(output == b"O" * expected_size, f"{label}: stdout must be complete")
            check(errors == b"E" * expected_size, f"{label}: stderr must be complete")
            print(f"{label}: stdout={len(output)}; stderr={len(errors)}; exit={code}")
        check(delayed[1:3] == direct[1:3], "both paths must recover identical finite output")
        print("wait_before_drain=timeout; communicate_after_timeout=completed")
    print(f"PASS checks={checks}; children_reaped=True; temporary_files_cleaned=True")


if __name__ == "__main__":
    if len(sys.argv) == 3 and sys.argv[1] == "--child":
        child(sys.argv[2])
    else:
        main()

import java.io.ByteArrayOutputStream;
import java.io.Closeable;
import java.io.IOException;
import java.io.InputStream;
import java.nio.file.Files;
import java.nio.file.Path;
import java.util.Arrays;
import java.util.concurrent.CompletionService;
import java.util.concurrent.ExecutorCompletionService;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;

/** Starts only this class as a child, with finite stdout and stderr. */
public final class SubprocessPipesDemo {
    private static final int BLOCK_SIZE = 65536;
    private static final int BLOCKS = 16;
    private static int checks;

    private record Result(boolean waitTimedOut, byte[] stdout, byte[] stderr, int exit) {}
    private record Captured(String channel, byte[] bytes) {}

    private static void check(boolean condition, String message) {
        if (!condition) throw new AssertionError(message);
        checks++;
    }

    private static byte[] repeated(byte value, int length) {
        byte[] data = new byte[length];
        Arrays.fill(data, value);
        return data;
    }

    private static void child(Path marker) throws IOException {
        Files.writeString(marker, "ready");
        byte[] out = repeated((byte) 'O', BLOCK_SIZE);
        byte[] err = repeated((byte) 'E', BLOCK_SIZE);
        for (int i = 0; i < BLOCKS; i++) {
            System.out.write(out);
            System.out.flush();
            System.err.write(err);
            System.err.flush();
        }
        if (System.out.checkError() || System.err.checkError()) {
            throw new IOException("child output stream failed");
        }
    }

    private static void closeQuietly(Closeable resource) {
        try {
            resource.close();
        } catch (IOException ignored) {
            // Original execution/cleanup failure is more informative.
        }
    }

    private static byte[] drainPipe(InputStream stream) throws IOException {
        int expected = BLOCK_SIZE * BLOCKS;
        ByteArrayOutputStream captured = new ByteArrayOutputStream(expected);
        byte[] buffer = new byte[8192];
        int count;
        // Positive-length reads avoid the pipe's FileInputStream.readAllBytes path.
        while ((count = stream.read(buffer, 0, buffer.length)) != -1) {
            if (count <= 0 || captured.size() > expected - count) {
                throw new IOException("unexpected zero read or output beyond the fixed limit");
            }
            captured.write(buffer, 0, count);
        }
        if (captured.size() != expected) {
            throw new IOException("output ended early: " + captured.size() + " of " + expected);
        }
        return captured.toByteArray();
    }

    private static Result runChild(Path marker, boolean waitFirst) throws Exception {
        boolean windows = System.getProperty("os.name").startsWith("Windows");
        String launcher = windows ? "javaw.exe" : "java";
        Path java = Path.of(System.getProperty("java.home"), "bin", launcher);
        if (!Files.isRegularFile(java)) throw new IOException("missing Java launcher: " + java);
        Process process = new ProcessBuilder(java.toString(), "-cp",
                System.getProperty("java.class.path"), SubprocessPipesDemo.class.getName(),
                "--child", marker.toString()).start();
        ExecutorService readers = Executors.newFixedThreadPool(2, task -> {
            Thread thread = new Thread(task, "child-pipe-reader");
            thread.setDaemon(true);
            return thread;
        });
        try {
            process.getOutputStream().close(); // No input is needed by this child.
            long readyDeadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(5);
            while (!Files.exists(marker)) {
                if (!process.isAlive() || System.nanoTime() >= readyDeadline) {
                    throw new IOException("child did not reach the output stage");
                }
                Thread.sleep(10);
            }
            boolean timedOut = waitFirst && !process.waitFor(250, TimeUnit.MILLISECONDS);
            // Drain both pipes and observe whichever reader finishes/fails first.
            CompletionService<Captured> completed = new ExecutorCompletionService<>(readers);
            completed.submit(() -> new Captured("stdout", drainPipe(process.getInputStream())));
            completed.submit(() -> new Captured("stderr", drainPipe(process.getErrorStream())));
            long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(5);
            byte[] stdout = null;
            byte[] stderr = null;
            for (int i = 0; i < 2; i++) {
                Future<Captured> finished = completed.poll(
                        Math.max(0, deadline - System.nanoTime()), TimeUnit.NANOSECONDS);
                if (finished == null) throw new IOException("pipe readers exceeded the guard");
                Captured capture = finished.get(); // Propagate a reader failure immediately.
                if (capture.channel().equals("stdout")) stdout = capture.bytes();
                else stderr = capture.bytes();
            }
            if (!process.waitFor(Math.max(0, deadline - System.nanoTime()), TimeUnit.NANOSECONDS)) {
                throw new IOException("child did not exit after its output was drained");
            }
            return new Result(timedOut, stdout, stderr, process.exitValue());
        } finally {
            try {
                if (process.isAlive()) {
                    process.destroyForcibly();
                    if (!process.waitFor(5, TimeUnit.SECONDS)) {
                        throw new IOException("child did not terminate during cleanup");
                    }
                }
            } finally {
                closeQuietly(process.getOutputStream());
                closeQuietly(process.getInputStream());
                closeQuietly(process.getErrorStream());
                readers.shutdownNow();
                if (!readers.awaitTermination(5, TimeUnit.SECONDS)) {
                    throw new IOException("pipe readers did not terminate");
                }
            }
        }
    }

    public static void main(String[] args) throws Exception {
        if (args.length == 2 && args[0].equals("--child")) {
            child(Path.of(args[1]));
            return;
        }
        Path folder = Files.createTempDirectory("qinghuan-java-pipes-");
        Path first = folder.resolve("first.ready");
        Path second = folder.resolve("second.ready");
        try {
            Result delayed = runChild(first, true);
            Result direct = runChild(second, false);
            check(delayed.waitTimedOut(), "waitFor without reading should time out");
            check(!direct.waitTimedOut(), "direct path does not use the wait-only probe");
            byte[] out = repeated((byte) 'O', BLOCK_SIZE * BLOCKS);
            byte[] err = repeated((byte) 'E', BLOCK_SIZE * BLOCKS);
            Result[] results = {delayed, direct};
            String[] labels = {"rescued", "direct"};
            for (int i = 0; i < results.length; i++) {
                Result result = results[i];
                check(result.exit() == 0, labels[i] + ": normal exit required");
                check(Arrays.equals(result.stdout(), out), labels[i] + ": complete stdout required");
                check(Arrays.equals(result.stderr(), err), labels[i] + ": complete stderr required");
                System.out.println(labels[i] + ": stdout=" + result.stdout().length
                        + "; stderr=" + result.stderr().length + "; exit=" + result.exit());
            }
            check(Arrays.equals(delayed.stdout(), direct.stdout())
                    && Arrays.equals(delayed.stderr(), direct.stderr()), "both paths must agree");
            System.out.println("waitFor_before_drain=timeout; two_readers_after_timeout=completed");
        } finally {
            Files.deleteIfExists(first);
            Files.deleteIfExists(second);
            Files.deleteIfExists(folder);
        }
        System.out.println("PASS checks=" + checks
                + "; children_reaped=true; readers_stopped=true; temporary_files_cleaned=true");
    }
}