如何检查标准输出是否已关闭 - 不向其写入数据？

Question

我编写了一个程序，它读取数据、过滤和处理数据并将其写入标准输出。如果标准输出通过管道传输到另一个进程，并且管道进程终止，我得到 SIGPIPEd，这很好，因为程序终止，并且管道及时结束。然而，根据过滤器参数，可能有几十秒没有一次写入，并且在此期间不会有 SIGPIPE，尽管下游进程早已完成。我怎样才能检测到这一点，而无需实际向标准输出写入内容？目前，管道只是挂起，直到我的程序因自然原因终止。我试着写一个零长度切片if _, err := os.Stdout.Write([]byte{}); err != nil但不幸的是，这不会导致错误。NB 理想情况下，无论平台如何，这都应该有效，但如果它仅适用于 Linux，那已经是一种改进。

Answer 1

这在 Go 中没有回答，但你可能会找到一种使用它的方法。如果您可以将 Poll(2) 应用于管道的写入端，您将在它变得不可写时收到通知。如何将其集成到您的 Go 代码中取决于您的程序；希望它可能有用：

#include <errno.h>

#include <poll.h>

#include <signal.h>

#include <stdio.h>

#include <stdlib.h>

#include <string.h>

#include <unistd.h>

void sp(int sno) {

    write(2, "sigpipe!\n", 9);

    _exit(1);

}

int waitfd(int fd) {

    int n;

    struct pollfd p;

    p.fd = fd;

    p.events = POLLOUT | POLLRDBAND;

    /* RDBAND is for what looks like a bug in illumos fifovnops.c */

    p.revents = 0;

    if ((n=poll(&p, 1, -1)) == 1) {

        if (p.revents & POLLOUT) {

            return fd;

        }

        if (p.revents & (POLLERR|POLLHUP)) {

            return -1;

        }

    }

    fprintf(stderr, "poll=%d (%d:%s), r=%#x\n",

        n, errno, strerror(errno), p.revents);

    return -1;

}

int main() {

    int count = 0;

    char c;

    signal(SIGPIPE, sp);

    while (read(0, &c, 1) > 0) {

        int w;

        while ((w=waitfd(1)) != -1 &&

            write(1, &c, 1) != 1) {

        }

        if (w == -1) {

            break;

        }

        count++;

    }

    fprintf(stderr, "wrote %d\n", count);

    return 0;

}

在 linux 中，您可以将这个程序运行为：./a.out < /dev/zero | sleep 1，它会打印出类似的内容：wrote 61441。您可以将其更改为休眠 3 秒，它会打印相同的内容。这是一个很好的证据，它已经填满了管道，正在等待空间。睡眠永远不会从管道中读取，因此当它的时间到时，它会关闭读取端，这会用 POLLERR 事件唤醒 poll(2)。

如果将轮询事件更改为不包括 POLLOUT，您将获得更简单的程序：

#include <errno.h>

#include <fcntl.h>

#include <poll.h>

#include <signal.h>

#include <stdio.h>

#include <stdlib.h>

#include <string.h>

#include <unistd.h>

int waitfd(int fd) {

    int n;

    struct pollfd p;

    p.fd = fd;

    p.events = POLLRDBAND;

    p.revents = 0;

    if ((n=poll(&p, 1, -1)) == 1) {

        if (p.revents & (POLLERR|POLLHUP)) {

            return -1;

        }

    }

    fprintf(stderr, "poll=%d (%d:%s), r=%#x\n",

        n, errno, strerror(errno), p.revents);

    return -1;

}

int main() {

    if (waitfd(1) == -1) {

        fprintf(stderr, "Got an error!\n");

    }

    return 0;

}

“出错了！” 表示管道已关闭。我不知道这是多么可移植，因为 poll(2) 文档有点粗略。如果没有 POLLRDBAND（所以 events 为 0），这适用于 Linux，但不适用于 UNIX（至少 Solaris 和 macos）。再一次，文档没用，但是内核源代码回答了很多问题:)

这个例子，使用线程，可以直接映射到go：

#include <pthread.h>

#include <errno.h>

#include <poll.h>

#include <signal.h>

#include <stdio.h>

#include <stdint.h>

#include <stdlib.h>

#include <string.h>

#include <unistd.h>

int Events = POLLRDBAND;

void sp(int sno) {

    char buf[64];

    write(2, buf, snprintf(buf, sizeof buf, "%d: sig%s(%d)\n", getpid(), sys_siglist[sno], sno));

    _exit(1);

}

int waitfd(int fd) {

    int n;

    struct pollfd p;

    p.fd = fd;

    p.events = Events;

    /* RDBAND is for what looks like a bug in illumos fifovnops.c */

    p.revents = 0;

    if ((n=poll(&p, 1, -1)) == 1) {

        if (p.revents & (POLLERR|POLLHUP)) {

            return -1;

        }

        return fd;

    }

    return -1;

}

void *waitpipe(void *t) {

    int x = (int)(intptr_t)t; /*gcc braindead*/

    waitfd(x);

    kill(getpid(), SIGUSR1);

    return NULL;

}

int main(int ac) {

    pthread_t killer;

    int count = 0;

    char c;

    

    Events |= (ac > 1) ? POLLOUT : 0;

    signal(SIGPIPE, sp);

    signal(SIGUSR1, sp);

    pthread_create(&killer, 0, waitpipe, (int *)1);

    while (read(0, &c, 1) > 0) {

        write(1, &c, 1);

        count++;

    }

    fprintf(stderr, "wrote %d\n", count);

    return 0;

}

请注意，它会在 poll 上驻留一个线程，并生成一个 SIGUSR1。这是运行它：

mcloud:pipe $ ./spthr < /dev/zero | hexdump -n80

0000000 0000 0000 0000 0000 0000 0000 0000 0000

*

0000050

185965: sigUser defined signal 1(10)

mcloud:pipe $ ./spthr < /dev/zero | sleep 1

185969: sigUser defined signal 1(10)

mcloud:pipe $ ./spthr | sleep 1

185972: sigUser defined signal 1(10)

mcloud:pipe $ ./spthr < /dev/zero | hexdump -n800000

0000000 0000 0000 0000 0000 0000 0000 0000 0000

*

00c3500

185976: sigBroken pipe(13)

在第一个命令中，hexdump 在 80 字节后退出，轮询基本上是在与 read+write 循环竞争，因此它可能生成了一个 sigpipe 或 sigusr1。

后两个演示了 sleep 将导致 sigusr1 （轮询返回异常事件），无论管道读取器退出时管道的写入端是否已满。

第四，使用 hexdump 读取大量数据，远远超过管道容量，这更确定性地导致了 sigpipe。

您可以生成更准确地对其建模的测试程序，但关键是一旦管道关闭，程序就会收到通知；不必等到下一次写入。

Answer 2

这不是问题的真正解决方案 - 即，检测管道中的进程是否在没有写入的情况下终止 - 但这是一个解决方法，在 Daniel Farrell 的评论中建议：（定义和）使用将被忽略的心跳信号下游。

由于此解决方法不透明，因此如果您不控制所涉及的所有进程，则可能无法实现。

这是一个使用 NUL 字节作为基于文本数据的心跳信号的示例：

my-cmd | head -1 | tr -d '\000' > file

my-cmd 将在不活动时发送 NUL 字节以获得及时的 EPIPE / SIGPIPE。

请注意，tr一旦达到其目的，使用 就再次剥离心跳 - 否则它们最终会出现在file.