☰

组合来自多个文件/管道的输入而不破坏线路或阻塞？

tag-icon • tag-icon utilities pipe
组合来自多个文件/管道的输入而不破坏线路或阻塞？

是否有一种工具可以从多个文件或管道获取输入并将其写入标准输出，而不会阻塞读取，从而使各个输入行完好无损？我基本上想将一堆输入复用到一个输出上，而不破坏线路。

$ combine file1 <(prog2) ... > nice-output.txt
  1. 我不关心输出的顺序
  2. 只要某些输入有数据，它就不应该阻塞
  3. 它应该是高效的（即，我可能会否决你的 Perl 一句话；）

答案1

你应该能够multitail很容易地做到这一点。

答案2

如果进程在一次write调用中写入行，这需要进程使用行缓冲（如果它们的标准输出不是终端，通常会关闭），您可以将它们全部指向一个管道。

{ { sleep .1; echo one; sleep .1; echo two; } &
  { echo hello; sleep .15; echo world; };
  wait; } | cat

如果进程仅在写入终端时执行行缓冲，则简单的方法是使用script.它有点笨拙：它只能写入文件。

script -q -c '
    { { sleep .1; echo one; sleep .1; echo two; } &
      { echo hello; sleep .15; echo world; };
      wait; }'
tail -n +2 typescript

如果程序写入长行或只是不使用行缓冲，则此方法将不起作用。您将需要一个收集器程序，该程序可以分别读取和缓冲来自每个输入的行，并在行结尾处执行同步。没有具有此功能的标准实用程序。我第二迦勒的建议multitail。

下面是一个 Python 脚本，它读取多个命令生成的行并将它们吐出到标准输出上，而不会打断一行。我还没有对其进行太多测试，所以请用户注意。我根本没有对它进行基准测试。

#!/usr/bin/env python
import Queue, itertools, os, subprocess, sys, threading
# Queue of (producer_id, line). line==None indicates the end of a producer.
lq = Queue.Queue()

# Line producer
def run_task(i, cmd):
    p = subprocess.Popen(cmd, stdout=subprocess.PIPE, shell=True)
    line = p.stdout.readline()
    while line <> "":
        lq.put((i, line))
        line = p.stdout.readline()
    lq.put((i, None))

# Start a producer for each command passed as an argument
for i in range(1,len(sys.argv)):
    threading.Thread(target=run_task, args=(i, sys.argv[i])).start()
sources = len(sys.argv) - 1
# Consumer: print lines as they come in, until no producer is left.
while sources > 0:
    (k, line) = lq.get()
    if line == None: sources -= 1
    else: sys.stdout.write(str(k) + ":" + line)

使用示例：

./collect.py 'sleep 1; ls /; sleep 1; ls /' \
             '/bin/echo -n foo; sleep 1; /bin/echo -n bar; sleep 1; /bin/echo qux'

答案3

是的，多尾似乎与作为终端子集的“窗口”的概念有关；我无法让它作为管道组件发挥良好作用。

所以看起来我们必须自己做这件事指关节开裂

/* Copyright © 2015 [email protected]
** Use/modify as you see fit but leave this attribution.
** If you change the interface and want to distribute the
** result please change the binary name too! */
#include <err.h>
#include <errno.h>
#include <string.h>
#include <stdlib.h>
#include <unistd.h>
#include <fcntl.h>

#include <sys/types.h>
#include <sys/stat.h>
#include <sys/select.h>

/* typedefs are for pussies */
struct {
    char *filename; /* for clarity of errors */
    char *data;
    long len;
    long cap;
} saved[FD_SETSIZE] = {0};

void
ewriten(int fd, char *buf, int n)
{
    int done = 0, c;
    while (done < n) {
        if ((c=write(fd, buf + done, n - done)) <= 0 && errno != EINTR) {
            err(1, "write");
        }
        done += c;
    }
}

int
empty(fd_set *fdset, int maxfd)
{
    int i;
    for (i=0; i <= maxfd; i++) {
        if (FD_ISSET(i, fdset)) return 0;
    }
    return 1;
}

void
combine(fd_set *fdset, int maxfd)
{
    char buf[4096], *cp;
    fd_set ready;
    int n, i, fd, left;
    while (!empty(fdset, maxfd)) {
        ready = *fdset;
        /* timeouts are for pussies */
        if (select(maxfd + 1, &ready, NULL, NULL, NULL) == -1) err(1, "select");
        for (fd=0; fd <= maxfd; fd++) {
            if (!FD_ISSET(fd, &ready)) continue;

            switch (n=read(fd, &buf, sizeof(buf))) {
            case -1:
                if (errno == EINTR)
                    break; /* ignore interrupts; we'll re-read next iteration */
                if (saved[fd].filename) err(1, "read: %s", saved[fd].filename);
                err(1, "read: %d", fd);
            case 0:
                if (saved[fd].len > 0) {
                    /* someone forgot their newline at EOF... */
                    ewriten(1, saved[fd].data, saved[fd].len);
                    saved[fd].data[0] = '\n'; /* put it back for them */
                    ewriten(1, saved[fd].data, 1);
                }
                free(saved[fd].data);
                FD_CLR(fd, fdset);
                break;
            default:
                for (cp=buf + n - 1; cp >= buf && *cp != '\n'; cp--); /* find last newline */
                left = n - (cp - buf + 1);
                if (cp >= buf) {
                    /* we found one! first dump any saved data from the last read */
                    if (saved[fd].len > 0) {
                        ewriten(1, saved[fd].data, saved[fd].len);
                        saved[fd].len = 0;
                    }
                    ewriten(1, buf, cp - buf + 1);
                }
                if (left > 0) {
                    /* now save any leftover data for later */
                    int need = saved[fd].len + left;
                    if (saved[fd].cap < need &&
                       (saved[fd].data=realloc(saved[fd].data, need)) == NULL) {
                        errx(1, "realloc: failed on %d bytes", need);
                        /* it was good enough for quake... */
                    }
                    saved[fd].cap = need;
                    memcpy(saved[fd].data + saved[fd].len, buf + n - 1 - left, left);
                    saved[fd].len += left;
                }
            }
        }
    }
}

void
addfd(int fd, fd_set *fdset, int *maxfd)
{
    FD_SET(fd, fdset);
    if (*maxfd < fd) {
        *maxfd = fd;
    }
}

int
main(int argc, char **argv)
{
    fd_set fdset;
    char **arg = argv + 1;
    char *cp;
    struct stat st;
    int fd, maxfd = -1;
    FD_ZERO(&fdset);
    while (*arg != NULL) {
        /* getopt is for pussies */
        if (strncmp("-u", *arg, 2) == 0) {
            *arg += 2;
            if (**arg == '\0' && *++arg == NULL ) errx(1, "-u requires argument (comma separated FD list)");
            /* reentrancy is for pussies */
            for (cp=strtok(*arg, ","); cp != NULL; cp=strtok(NULL, ",")) {
                fd = atoi(cp);
                if (fstat(fd, &st) != 0) err(1, "%d", fd);
                addfd(fd, &fdset, &maxfd);
            }
            arg++;
        } else if (strcmp("-", *arg) == 0) {
            if (fstat(0, &st) != 0) err(1, "stdin", fd);
            addfd(0, &fdset, &maxfd);
            saved[0].filename = "stdin";
            arg++;
        } else if (strcmp("--", *arg) == 0) {
            arg++;
            break;
        } else if (**arg == '-') {
            errx(1, "unrecognized argument %s", *arg);
        } else {
            break; /* treat as filename */
        }
    }
    /* remaining args are filenames */
    for (; *arg != NULL; arg++) {
        /* stdio is for pussies */
        if ((fd=open(*arg, O_RDONLY)) == -1) err(1, "open: %s", *arg);
        addfd(fd, &fdset, &maxfd);
        saved[fd].filename = *arg;
    }
    combine(&fdset, maxfd);
    return 0;
}

啊啊感觉很好。

（注意：它在大约两组输入上进行了测试。错误可能存在也可能不存在）

相关内容