Linux非阻塞fifo(按需记录)

dro*_*nus 32 linux bash logging fifo

我喜欢按需"记录"节目输出.例如.输出记录到终端,但另一个进程可以随时挂接当前输出.

经典的方式是:

myprogram 2>&1 | tee /tmp/mylog
Run Code Online (Sandbox Code Playgroud)

并按需提供

tail /tmp/mylog
Run Code Online (Sandbox Code Playgroud)

但是,这将创建一个不断增长的日志文件,即使在驱动器空间不足之前不使用它也是如此.所以我的尝试是:

mkfifo /tmp/mylog
myprogram 2>&1 | tee /tmp/mylog
Run Code Online (Sandbox Code Playgroud)

并按需提供

cat /tmp/mylog
Run Code Online (Sandbox Code Playgroud)

现在我可以随时阅读/ tmp/mylog.但是,任何输出都会阻塞程序,直到读取/ tmp/mylog.我喜欢fifo来刷新任何未读回的数据.怎么做?

rac*_*cic 48

受你的问题的启发,我写了一个简单的程序,可以让你这样做:

$ myprogram 2>&1 | ftee /tmp/mylog

它的行为类似tee但是将stdin克隆到stdout和一个命名管道(现在的要求)而没有阻塞.这意味着如果你想以这种方式记录,你可能会丢失你的日志数据,但我想这在你的场景中是可以接受的.诀窍是阻止SIGPIPE信号并忽略写入损坏的fifo时的错误.当然,这个样本可以通过各种方式进行优化,但到目前为止,它的确可以完成这项工作.

/* ftee - clone stdin to stdout and to a named pipe 
(c) racic@stackoverflow
WTFPL Licence */

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/types.h>
#include <sys/stat.h>
#include <fcntl.h>
#include <errno.h>
#include <signal.h>
#include <unistd.h>

int main(int argc, char *argv[])
{
    int readfd, writefd;
    struct stat status;
    char *fifonam;
    char buffer[BUFSIZ];
    ssize_t bytes;

    signal(SIGPIPE, SIG_IGN);

    if(2!=argc)
    {
        printf("Usage:\n someprog 2>&1 | %s FIFO\n FIFO - path to a"
            " named pipe, required argument\n", argv[0]);
        exit(EXIT_FAILURE);
    }
    fifonam = argv[1];

    readfd = open(fifonam, O_RDONLY | O_NONBLOCK);
    if(-1==readfd)
    {
        perror("ftee: readfd: open()");
        exit(EXIT_FAILURE);
    }

    if(-1==fstat(readfd, &status))
    {
        perror("ftee: fstat");
        close(readfd);
        exit(EXIT_FAILURE);
    }

    if(!S_ISFIFO(status.st_mode))
    {
        printf("ftee: %s in not a fifo!\n", fifonam);
        close(readfd);
        exit(EXIT_FAILURE);
    }

    writefd = open(fifonam, O_WRONLY | O_NONBLOCK);
    if(-1==writefd)
    {
        perror("ftee: writefd: open()");
        close(readfd);
        exit(EXIT_FAILURE);
    }

    close(readfd);

    while(1)
    {
        bytes = read(STDIN_FILENO, buffer, sizeof(buffer));
        if (bytes < 0 && errno == EINTR)
            continue;
        if (bytes <= 0)
            break;

        bytes = write(STDOUT_FILENO, buffer, bytes);
        if(-1==bytes)
            perror("ftee: writing to stdout");
        bytes = write(writefd, buffer, bytes);
        if(-1==bytes);//Ignoring the errors
    }
    close(writefd); 
    return(0);
}
Run Code Online (Sandbox Code Playgroud)

您可以使用以下标准命令对其进行编译:

$ gcc ftee.c -o ftee

您可以通过运行eg来快速验证它:

$ ping www.google.com | ftee /tmp/mylog

$ cat /tmp/mylog

另请注意 - 这不是多路复用器.您一次只能处理一个进程$ cat /tmp/mylog.

  • 我认为这对于任何未经考虑的长期运行程序来说非常有用,只要没有问题,它就会产生很多不感兴趣的dbeug输出.例如,考虑用于单一目的的嵌入式设备.如果系统多年无人值守运行,则日志文件不是很有用.也许甚至还有一个只读文件系统来保护嵌入式功能免受fs乱码和断电.所以日志文件没有意义. (5认同)
  • 你想要实现的是非常罕见的,这种方法适用于构建应用程序时.对于大多数情况,`tail -f logfile.log`工作得很好. (2认同)

Fab*_*ias 12

这是一个(非常)旧的线程,但我最近遇到了类似的问题.事实上,我需要的是将stdin克隆到stdout,并将副本复制到非阻塞的管道.第一个答案中提议的ftee真的有帮助,但是(对于我的用例)太不稳定了.意思是我丢失了数据,如果我及时完成的​​话我可以处理.

我面临的情况是我有一个进程(some_process)聚合一些数据并将其结果每三秒写入stdout.(简化)设置看起来像这样(在实际设置中我使用命名管道):

some_process | ftee >(onlineAnalysis.pl > results) | gzip > raw_data.gz
Run Code Online (Sandbox Code Playgroud)

现在,raw_data.gz必须被压缩并且必须完整.ftee做得很好.但是我在中间使用的管道太慢而无法抓取冲出的数据 - 但它足够快,可以处理所有内容,如果它可以到达它,这是用普通的T恤测试的.然而,如果未命名的管道发生任何事情,普通的T恤会阻塞,并且因为我希望能够按需挂钩,所以不能选择T恤.回到主题:当我在其间放置缓冲区时,它变得更好,导致:

some_process | ftee >(mbuffer -m 32M| onlineAnalysis.pl > results) | gzip > raw_data.gz
Run Code Online (Sandbox Code Playgroud)

但那仍然是我可以处理的数据丢失.所以我继续将之前建议的ftee扩展到缓冲版本(bftee).它仍具有所有相同的属性,但在写入失败时使用(低效?)内部缓冲区.如果缓冲区已满,它仍然会丢失数据,但它对我的情况很有效.一如往常有很大的改进空间,但是当我复制代码时,我想将它分享给可能有用的人.

/* bftee - clone stdin to stdout and to a buffered, non-blocking pipe 
    (c) racic@stackoverflow
    (c) fabraxias@stackoverflow
    WTFPL Licence */

    #include <stdio.h>
    #include <stdlib.h>
    #include <string.h>
    #include <sys/types.h>
    #include <sys/stat.h>
    #include <fcntl.h>
    #include <errno.h>
    #include <signal.h>
    #include <unistd.h>

    // the number of sBuffers that are being held at a maximum
    #define BUFFER_SIZE 4096
    #define BLOCK_SIZE 2048

    typedef struct {
      char data[BLOCK_SIZE];
      int bytes;
    } sBuffer;

    typedef struct {
      sBuffer *data;  //array of buffers
      int bufferSize; // number of buffer in data
      int start;      // index of the current start buffer
      int end;        // index of the current end buffer
      int active;     // number of active buffer (currently in use)
      int maxUse;     // maximum number of buffers ever used
      int drops;      // number of discarded buffer due to overflow
      int sWrites;    // number of buffer written to stdout
      int pWrites;    // number of buffers written to pipe
    } sQueue;

    void InitQueue(sQueue*, int);              // initialized the Queue
    void PushToQueue(sQueue*, sBuffer*, int);  // pushes a buffer into Queue at the end 
    sBuffer *RetrieveFromQueue(sQueue*);       // returns the first entry of the buffer and removes it or NULL is buffer is empty
    sBuffer *PeakAtQueue(sQueue*);             // returns the first entry of the buffer but does not remove it. Returns NULL on an empty buffer
    void ShrinkInQueue(sQueue *queue, int);    // shrinks the first entry of the buffer by n-bytes. Buffer is removed if it is empty
    void DelFromQueue(sQueue *queue);          // removes the first entry of the queue

    static void sigUSR1(int);                  // signal handled for SUGUSR1 - used for stats output to stderr
    static void sigINT(int);                   // signla handler for SIGKILL/SIGTERM - allows for a graceful stop ?

    sQueue queue;                              // Buffer storing the overflow
    volatile int quit;                         // for quiting the main loop

    int main(int argc, char *argv[])
    {   
        int readfd, writefd;
        struct stat status;
        char *fifonam;
        sBuffer buffer;
        ssize_t bytes;
        int bufferSize = BUFFER_SIZE;

        signal(SIGPIPE, SIG_IGN);
        signal(SIGUSR1, sigUSR1);
        signal(SIGTERM, sigINT);
        signal(SIGINT,  sigINT);

        /** Handle commandline args and open the pipe for non blocking writing **/

        if(argc < 2 || argc > 3)
        {   
            printf("Usage:\n someprog 2>&1 | %s FIFO [BufferSize]\n"
                   "FIFO - path to a named pipe, required argument\n"
                   "BufferSize - temporary Internal buffer size in case write to FIFO fails\n", argv[0]);
            exit(EXIT_FAILURE);
        }

        fifonam = argv[1];
        if (argc == 3) {
          bufferSize = atoi(argv[2]);
          if (bufferSize == 0) bufferSize = BUFFER_SIZE;
        }

        readfd = open(fifonam, O_RDONLY | O_NONBLOCK);
        if(-1==readfd)
        {   
            perror("bftee: readfd: open()");
            exit(EXIT_FAILURE);
        }

        if(-1==fstat(readfd, &status))
        {
            perror("bftee: fstat");
            close(readfd);
            exit(EXIT_FAILURE);
        }

        if(!S_ISFIFO(status.st_mode))
        {
            printf("bftee: %s in not a fifo!\n", fifonam);
            close(readfd);
            exit(EXIT_FAILURE);
        }

        writefd = open(fifonam, O_WRONLY | O_NONBLOCK);
        if(-1==writefd)
        {
            perror("bftee: writefd: open()");
            close(readfd);
            exit(EXIT_FAILURE);
        }

        close(readfd);


        InitQueue(&queue, bufferSize);
        quit = 0;

        while(!quit)
        {
            // read from STDIN
            bytes = read(STDIN_FILENO, buffer.data, sizeof(buffer.data));

            // if read failed due to interrupt, then retry, otherwise STDIN has closed and we should stop reading
            if (bytes < 0 && errno == EINTR) continue;
            if (bytes <= 0) break;

            // save the number if read bytes in the current buffer to be processed
            buffer.bytes = bytes;

            // this is a blocking write. As long as buffer is smaller than 4096 Bytes, the write is atomic to a pipe in Linux
            // thus, this cannot be interrupted. however, to be save this should handle the error cases of partial or interrupted write none the less.
            bytes = write(STDOUT_FILENO, buffer.data, buffer.bytes);
            queue.sWrites++;

            if(-1==bytes) {
                perror("ftee: writing to stdout");
                break;
            }

            sBuffer *tmpBuffer = NULL;

            // if the queue is empty (tmpBuffer gets set to NULL) the this does nothing - otherwise it tries to write
            // the buffered data to the pipe. This continues until the Buffer is empty or the write fails.
            // NOTE: bytes cannot be -1  (that would have failed just before) when the loop is entered. 
            while ((bytes != -1) && (tmpBuffer = PeakAtQueue(&queue)) != NULL) {
               // write the oldest buffer to the pipe
               bytes = write(writefd, tmpBuffer->data, tmpBuffer->bytes);

               // the  written bytes are equal to the buffer size, the write is successful - remove the buffer and continue
               if (bytes == tmpBuffer->bytes) {
                 DelFromQueue(&queue);
                 queue.pWrites++;
               } else if (bytes > 0) {
                 // on a positive bytes value there was a partial write. we shrink the current buffer
                 //  and handle this as a write failure
                 ShrinkInQueue(&queue, bytes);
                 bytes = -1;
               }
            }
            // There are several cases here:
            // 1.) The Queue is empty -> bytes is still set from the write to STDOUT. in this case, we try to write the read data directly to the pipe
            // 2.) The Queue was not empty but is now -> bytes is set from the last write (which was successful) and is bigger 0. also try to write the data
            // 3.) The Queue was not empty and still is not -> there was a write error before (even partial), and bytes is -1. Thus this line is skipped.
            if (bytes != -1) bytes = write(writefd, buffer.data, buffer.bytes);

            // again, there are several cases what can happen here
            // 1.) the write before was successful -> in this case bytes is equal to buffer.bytes and nothing happens
            // 2.) the write just before is partial or failed all together - bytes is either -1 or smaller than buffer.bytes -> add the remaining data to the queue
            // 3.) the write before did not happen as the buffer flush already had an error. In this case bytes is -1 -> add the remaining data to the queue
            if (bytes != buffer.bytes)
              PushToQueue(&queue, &buffer, bytes);
            else 
              queue.pWrites++;
        }

        // once we are done with STDIN, try to flush the buffer to the named pipe
        if (queue.active > 0) {
           //set output buffer to block - here we wait until we can write everything to the named pipe
           // --> this does not seem to work - just in case there is a busy loop that waits for buffer flush aswell. 
           int saved_flags = fcntl(writefd, F_GETFL);
           int new_flags = saved_flags & ~O_NONBLOCK;
           int res = fcntl(writefd, F_SETFL, new_flags);

           sBuffer *tmpBuffer = NULL;
           //TODO: this does not handle partial writes yet
           while ((tmpBuffer = PeakAtQueue(&queue)) != NULL) {
             int bytes = write(writefd, tmpBuffer->data, tmpBuffer->bytes);
             if (bytes != -1) DelFromQueue(&queue);
           }
        }

        close(writefd);

    }


    /** init a given Queue **/
    void InitQueue (sQueue *queue, int bufferSize) {
      queue->data = calloc(bufferSize, sizeof(sBuffer));
      queue->bufferSize = bufferSize;
      queue->start = 0;
      queue->end = 0;
      queue->active = 0;
      queue->maxUse = 0;
      queue->drops = 0;
      queue->sWrites = 0;
      queue->pWrites = 0;
    }

    /** push a buffer into the Queue**/
    void PushToQueue(sQueue *queue, sBuffer *p, int offset)
    {

        if (offset < 0) offset = 0;      // offset cannot be smaller than 0 - if that is the case, we were given an error code. Set it to 0 instead
        if (offset == p->bytes) return;  // in this case there are 0 bytes to add to the queue. Nothing to write

        // this should never happen - offset cannot be bigger than the buffer itself. Panic action
        if (offset > p->bytes) {perror("got more bytes to buffer than we read\n"); exit(EXIT_FAILURE);}

        // debug output on a partial write. TODO: remove this line
        // if (offset > 0 ) fprintf(stderr, "partial write to buffer\n");

        // copy the data from the buffer into the queue and remember its size
        memcpy(queue->data[queue->end].data, p->data + offset , p->bytes-offset);
        queue->data[queue->end].bytes = p->bytes - offset;

        // move the buffer forward
        queue->end = (queue->end + 1) % queue->bufferSize;

        // there is still space in the buffer
        if (queue->active < queue->bufferSize)
        {
            queue->active++;
            if (queue->active > queue->maxUse) queue->maxUse = queue->active;
        } else {
            // Overwriting the oldest. Move start to next-oldest
            queue->start = (queue->start + 1) % queue->bufferSize;
            queue->drops++;
        }
    }

    /** return the oldest entry in the Queue and remove it or return NULL in case the Queue is empty **/
    sBuffer *RetrieveFromQueue(sQueue *queue)
    {
        if (!queue->active) { return NULL; }

        queue->start = (queue->start + 1) % queue->bufferSize;
        queue->active--;
        return &(queue->data[queue->start]);
    }

    /** return the oldest entry in the Queue or NULL if the Queue is empty. Does not remove the entry **/
    sBuffer *PeakAtQueue(sQueue *queue)
    {
        if (!queue->active) { return NULL; }
        return &(queue->data[queue->start]);
    }

    /*** Shrinks the oldest entry i the Queue by bytes. Removes the entry if buffer of the oldest entry runs empty*/
    void ShrinkInQueue(sQueue *queue, int bytes) {

      // cannot remove negative amount of bytes - this is an error case. Ignore it
      if (bytes <= 0) return;

      // remove the entry if the offset is equal to the buffer size
      if (queue->data[queue->start].bytes == bytes) {
        DelFromQueue(queue);
        return;
      };

      // this is a partial delete
      if (queue->data[queue->start].bytes > bytes) {
        //shift the memory by the offset
        memmove(queue->data[queue->start].data, queue->data[queue->start].data + bytes, queue->data[queue->start].bytes - bytes);
        queue->data[queue->start].bytes = queue->data[queue->start].bytes - bytes;
        return;
      }

      // panic is the are to remove more than we have the buffer
      if (queue->data[queue->start].bytes < bytes) {
        perror("we wrote more than we had - this should never happen\n");
        exit(EXIT_FAILURE);
        return;
      }
    }

    /** delete the oldest entry from the queue. Do nothing if the Queue is empty **/
    void DelFromQueue(sQueue *queue)
    {
        if (queue->active > 0) {
          queue->start = (queue->start + 1) % queue->bufferSize;
          queue->active--;
        }
    }

    /** Stats output on SIGUSR1 **/
    static void sigUSR1(int signo) {
      fprintf(stderr, "Buffer use: %i (%i/%i), STDOUT: %i PIPE: %i:%i\n", queue.active, queue.maxUse, queue.bufferSize, queue.sWrites, queue.pWrites, queue.drops);
    }

    /** handle signal for terminating **/
    static void sigINT(int signo) {
      quit++;
      if (quit > 1) exit(EXIT_FAILURE);
    }
Run Code Online (Sandbox Code Playgroud)

此版本需要一个(可选)参数,该参数指定要为管道缓冲的块数.我的示例调用现在看起来像这样:

some_process | bftee >(onlineAnalysis.pl > results) 16384 | gzip > raw_data.gz
Run Code Online (Sandbox Code Playgroud)

导致在丢弃发生之前缓冲16384个块.这使用了大约32 MB的内存,但......谁在乎呢?

当然,在真实环境中我使用命名管道,以便我可以根据需要进行附加和分离.看起来像这样:

mkfifo named_pipe
some_process | bftee named_pipe 16384 | gzip > raw_data.gz &
cat named_pipe | onlineAnalysis.pl > results
Run Code Online (Sandbox Code Playgroud)

此外,该过程对信号作出如下反应:SIGUSR1 - >将计数器打印到STDERR SIGTERM,SIGINT - >首先退出主循环并将缓冲区刷新到管道,第二个立即终止程序.

也许这可以帮助将来的某个人......享受


Mat*_*ttH 8

但是,这将创建一个不断增长的日志文件,即使在驱动器空间不足之前不使用它也是如此.

为什么不定期旋转日志?甚至还有一个程序可以帮到你logrotate.

还有一个系统,用于生成日志消息并根据类型对它们执行不同的操作.它被称为syslog.

你甚至可以将两者结合起来.让您的程序生成syslog消息,配置syslog以将它们放在一个文件中并使用logrotate来确保它们不会填满磁盘.


如果事实证明你是在为一个小型嵌入式系统编写而且程序的输出很重,那么你可以考虑各种各样的技术.

  • 远程syslog:将syslog消息发送到网络上的syslog服务器.
  • 使用syslog中可用的严重性级别对消息执行不同的操作.例如,丢弃"INFO"但记录并转发"ERR"或更高.例如控制台
  • 在程序中使用信号处理程序重新读取HUP上的配置,并以这种方式"按需"改变日志生成.
  • 让你的程序在unix套接字上监听并在打开时将消息写下来.您甚至可以通过这种方式在程序中实现和交互式控制台.
  • 使用配置文件,提供对日志记录输出的精细控制.


dro*_*nus 6

经常在嵌入式设备上使用的BusyBox可以创建ram缓冲日志

syslogd -C
Run Code Online (Sandbox Code Playgroud)

这可以填补

logger
Run Code Online (Sandbox Code Playgroud)

并阅读

logread
Run Code Online (Sandbox Code Playgroud)

工作得很好,但只提供一个全局日志.


hol*_*eek 5

如果您可以在嵌入式设备上安装屏幕,那么您可以在其中运行“myprogram”并将其分离,并在您想要查看日志的任何时候重新附加它。就像是:

$ screen -t sometitle myprogram
Hit Ctrl+A, then d to detach it.
Run Code Online (Sandbox Code Playgroud)

每当您想查看输出时,请重新附加它:

$ screen -DR sometitle
Hit Ctrl-A, then d to detach it again.
Run Code Online (Sandbox Code Playgroud)

这样您就不必担心使用磁盘空间的程序输出。


Pio*_*ost 5

似乎bash <>重定向运算符(3.6.10打开文件描述符以进行读写)使写入到文件/ fifo的文件打开而不会阻塞。这应该工作:

$ mkfifo /tmp/mylog
$ exec 4<>/tmp/mylog
$ myprogram 2>&1 | tee >&4
$ cat /tmp/mylog # on demend
Run Code Online (Sandbox Code Playgroud)

gniourf_gniourf在#bash IRC频道上给出的解决方案。


wal*_*lly 5

日志记录可以定向到 UDP 套接字。由于 UDP 是无连接的,它不会阻止发送程序。当然,如果接收方或网络跟不上,日志就会丢失。

myprogram 2>&1 | socat - udp-datagram:localhost:3333
Run Code Online (Sandbox Code Playgroud)

然后当你想观察日志时:

socat udp-recv:3333 -
Run Code Online (Sandbox Code Playgroud)

还有一些其他很酷的好处,比如能够同时附加多个侦听器或广播到多个设备。