Linux non-bloquant fifo (journalisation à la demande))

c'est un (très) vieux fil, mais j'ai rencontré un problème similaire récemment. En fait, ce dont j'avais besoin c'est d'un clonage de stdin à stdout avec une copie à un tuyau qui ne bloque pas. la ftee proposée dans la première réponse a vraiment aidé là, mais était (pour mon cas d'utilisation) trop volatile. Ce qui signifie que j'ai perdu des données que j'aurais pu traiter si je les avais eu à temps.

le scénario auquel j'ai été confronté est que j'ai un processus (some_process) qui regroupe certaines données et écrit son résultats toutes les trois secondes à stdout. La configuration (simplifiée) ressemblait à ceci (dans la configuration réelle j'utilise un tube nommé):

some_process | ftee >(onlineAnalysis.pl > results) | gzip > raw_data.gz

maintenant, raw_data.gz doit être compressé et être complets. ftee fait très bien ce travail. Mais le tuyau que j'utilise au milieu était trop lent pour saisir les données évacuées - mais il était assez rapide pour tout traiter s'il pouvait y arriver, ce qui a été testé avec un tee normal. Cependant, un tee normal bloque si quelque chose arrive à le tuyau sans nom, et comme je veux pouvoir m'accrocher à la demande, tee n'est pas une option. Revenons au sujet: C'est mieux quand je mets un tampon entre les deux, ce qui entraîne:

some_process | ftee >(mbuffer -m 32M| onlineAnalysis.pl > results) | gzip > raw_data.gz

mais je perdais encore des données que j'aurais pu traiter. Je suis donc allé de l'avant et étendu le ftee proposé avant à une version tamponnée (bftee). Il a toujours les mêmes propriétés, mais utilise un (inefficace ?) tampon interne en cas de défaillance d'une écriture. Il perd toujours de données si la mémoire tampon est plein, mais ça marche très bien pour mon affaire. Comme toujours, il y a beaucoup de place pour l'amélioration, mais comme j'ai copié le code d'ici, j'aimerais le partager avec des gens qui pourraient avoir un usage pour elle.

/* 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);
    }

cette version prend un argument de plus (optionnel) qui spécifie le nombre de blocs qui doivent être tamponnés pour la pipe. Mon exemple d'appel ressemble maintenant à ceci:

some_process | bftee >(onlineAnalysis.pl > results) 16384 | gzip > raw_data.gz

résultant en 16384 blocs à tamponner avant les rejets arriver. cela utilise environ 32 Mocts de plus de mémoire, mais... qui s'en soucie ?

bien sûr, dans l'environnement réel, j'utilise un tuyau nommé pour pouvoir fixer et détacher au besoin. Il y a ressemble à ceci:

mkfifo named_pipe
some_process | bftee named_pipe 16384 | gzip > raw_data.gz &
cat named_pipe | onlineAnalysis.pl > results

en outre, le processus réagit sur les signaux comme suit: SIGUSR1 - > print counters to STDERR SIGTERM, SIGINT - > sort d'abord de la boucle principale et envoie le tampon au tuyau, le second termine immédiatement le programme.

peut-être que cela aidera quelqu'un dans le futur... Profiter de

cependant, cela créerait un fichier journal en croissance constante même s'il n'est pas utilisé jusqu'à ce que le lecteur manque d'espace.

pourquoi ne pas tourner périodiquement les bûches? Il y a même un programme pour le faire pour vous logrotate .

il y a aussi un système pour générer des messages log et faire différentes choses avec eux selon le type. Ça s'appelle syslog .

vous pourriez même combiner les deux. Votre le programme génère des messages syslog, configure syslog pour les placer dans un fichier et utilise logrotate pour s'assurer qu'ils ne remplissent pas le disque.

" s'il s'avère que vous écriviez pour un petit système intégré et que la sortie du programme est lourde, il y a une variété de techniques que vous pourriez envisager.

• syslog Distant: envoyer les messages syslog à un serveur syslog sur le réseau.
• utilisez la les niveaux de gravité disponibles dans syslog pour faire différentes choses avec les messages. Par exemple: jeter " INFO "mais se connecter et avancer" ERR " ou plus. Par exemple: pour consoler
• utilisez un gestionnaire de signal dans votre programme pour relire la configuration sur HUP et faire varier la génération de log "sur demande" de cette façon.
• demandez à votre programme d'écouter sur une socket unix et d'y écrire des messages lorsqu'il est ouvert. Vous pourriez même implémenter et la console interactive dans votre programme de cette façon.
• à l'aide d'un fichier de configuration, fournir un contrôle granulaire de la sortie de journalisation.

BusyBox souvent utilisé sur les appareils embarqués peut créer une mémoire tampon ram par

syslogd -C

qui peut être rempli par

logger

et lu par

logread

fonctionne assez bien, mais ne fournit qu'un seul log global.

Si vous pouvez installer l'écran sur le périphérique intégré, alors vous pouvez exécuter 'myprogram", et de le détacher et rattacher à tout moment vous souhaitez voir le journal. Quelque chose comme:

$ screen -t sometitle myprogram
Hit Ctrl+A, then d to detach it.

chaque fois que vous voulez voir la sortie, rattachez-la:

$ screen -DR sometitle
Hit Ctrl-A, then d to detach it again.

de cette façon, vous n'aurez pas à vous soucier de la sortie du programme en utilisant l'espace disque.

il semble que bash <> opérateur de redirection ( 3.6.10 ouverture des descripteurs de fichier pour la lecture et L'écriture ) rend l'écriture au fichier/fifo ouvert avec lui non-blocage. Cela devrait fonctionner:

$ mkfifo /tmp/mylog
$ exec 4<>/tmp/mylog
$ myprogram 2>&1 | tee >&4
$ cat /tmp/mylog # on demend

Solution donnée par gniourf_gniourf sur le canal IRC # bash.

Le problème avec le fifo est que le tout se bloquer lorsque le tuyau buffer se remplit et aucun processus de lecture.

pour l'approche fifo de travail je pense que vous auriez à mettre en œuvre un modèle client-serveur nommé pipe similaire à celui mentionné dans BASH: meilleure architecture pour la lecture à partir de deux flux d'entrée (voir code légèrement modifié ci-dessous, exemple de code 2).

pour une solution de contournement, vous pouvez également utiliser une construction while ... read au lieu de tee ing stdout à un tuyau nommé en mettant en œuvre un mécanisme de comptage à l'intérieur de la boucle while ... read qui écrasera le fichier journal périodiquement par un nombre spécifié de lignes. Cela empêcherait un fichier journal de se développer (exemple de code 1).

# sample code 1

# terminal window 1
rm -f /tmp/mylog
touch /tmp/mylog
while sleep 2; do date '+%Y-%m-%d_%H.%M.%S'; done 2>&1 | while IFS="" read -r line; do 
  lno=$((lno+1))
  #echo $lno
  array[${lno}]="${line}"
  if [[ $lno -eq 10 ]]; then
    lno=$((lno+1))
    array[${lno}]="-------------"
    printf '%s\n' "${array[@]}" > /tmp/mylog
    unset lno array
  fi
  printf '%s\n' "${line}"
done

# terminal window 2
tail -f /tmp/mylog


#------------------------


# sample code 2

# code taken from: 
# https://stackoverflow.com/questions/6702474/bash-best-architecture-for-reading-from-two-input-streams
# terminal window 1

# server
(
rm -f /tmp/to /tmp/from
mkfifo /tmp/to /tmp/from
while true; do 
  while IFS="" read -r -d $'\n' line; do 
    printf '%s\n' "${line}"
  done </tmp/to >/tmp/from &
  bgpid=$!
  exec 3>/tmp/to
  exec 4</tmp/from
  trap "kill -TERM $bgpid; exit" 0 1 2 3 13 15
  wait "$bgpid"
  echo "restarting..."
done
) &
serverpid=$!
#kill -TERM $serverpid

# client
(
exec 3>/tmp/to;
exec 4</tmp/from;
while IFS="" read -r -d $'\n' <&4 line; do
  if [[ "${line:0:1}" == $'7' ]]; then 
    printf 'line from stdin: %s\n' "${line:1}"  > /dev/null
  else       
    printf 'line from fifo: %s\n' "$line"       > /dev/null
  fi
done &
trap "kill -TERM $"'!; exit' 1 2 3 13 15
while IFS="" read -r -d $'\n' line; do
  # can we make it atomic?
  # sleep 0.5
  # dd if=/tmp/to iflag=nonblock of=/dev/null  # flush fifo
  printf '7%s\n' "${line}"
done >&3
) &
# kill -TERM $!


# terminal window 2
# tests
echo hello > /tmp/to
yes 1 | nl > /tmp/to
yes 1 | nl | tee /tmp/to
while sleep 2; do date '+%Y-%m-%d_%H.%M.%S'; done 2>&1 | tee -a /tmp/to


# terminal window 3
cat /tmp/to | head -n 10

si votre processus écrit dans n'importe quel fichier journal et essuie ensuite le fichier et recommence de temps en temps, de sorte qu'il ne devient pas trop grand, ou utilise logrotate .

tail --follow=name --retry my.log

est tout ce dont vous avez besoin. Vous obtiendrez autant de scroll-back que votre terminal.

Rien non standard est nécessaire. Je ne l'ai pas essayé avec de petits fichiers de log mais tous nos logs tournent comme ceci et je n'ai jamais remarqué des lignes qui se desserrent.