Linux pipes tips & tricks

Исходный код, уровень 1, shell

Т. к. лучшая документация — исходный код, обратимся к нему. Bash использует Yacc для парсинга входных команд и возвращает ‘command_connect()’, когда встречает символ ‘|’.
parse.y:

1242 pipeline:   pipeline ‘|’ newline_list pipeline
1243             { $$ = command_connect ($1,  $4,  ‘|’); }
1244     |       pipeline BAR_AND newline_list pipeline
1245             {
1246               /* Make cmd1 |& cmd2 equivalent to cmd1 2>&1 | cmd2 */
1247               COMMAND *tc;
1248               REDIRECTEE rd,  sd;
1249               REDIRECT *r;
1250 
1251               tc = $1->type == cm_simple ? (COMMAND *)$1->value.Simple : $1;
1252               sd.dest = 2;
1253               rd.dest = 1;
1254               r = make_redirection (sd,  r_duplicating_output,  rd,  0);
1255               if (tc->redirects)
1256               {
1257                   register REDIRECT *t;
1258                   for (t = tc->redirects; t->next; t = t->next)
1259                       ;
1260                   t->next = r;
1261               }
1262               else
1263                   tc->redirects = r;
1264 
1265               $$ = command_connect ($1,  $4,  ‘|’);
1266             }
1267     |       command
1268             { $$ = $1; }
1269     ;

Также здесь мы видим обработку пары символов ‘|&’, что эквивалентно перенаправлению как stdout, так и stderr в конвеер. Далее обратимся к command_connect():make_cmd.c:

194 COMMAND *
195 command_connect (com1,  com2,  connector)
196      COMMAND *com1,  *com2;
197      int connector;
198 {
199   CONNECTION *temp;
200 
201   temp = (CONNECTION *)xmalloc (sizeof (CONNECTION));
202   temp->connector = connector;
203   temp->first = com1;
204   temp->second = com2;
205   return (make_command (cm_connection,  (SIMPLE_COM *)temp));
206 }

где connector это символ ‘|’ как int. При выполнении последовательности команд (связанных через ‘&’, ‘|’, ‘;’, и т. д.) вызывается execute_connection():execute_cmd.c:

2325     case ‘|’:
...
2331       exec_result = execute_pipeline (command,  asynchronous,  pipe_in,  pipe_out,  fds_to_close);

PIPE_IN и PIPE_OUT — файловые дескрипторы, содержащие информацию о входном и выходном потоках. Они могут принимать значение NO_PIPE, которое означает, что I/O является stdin/stdout.
execute_pipeline() довольно объемная функция, имплементация которой содержится в execute_cmd.c. Мы рассмотрим наиболее интересные для нас части.
execute_cmd.c:

2112   prev = pipe_in;
2113   cmd = command;
2114
2115   while (cmd && cmd->type == cm_connection &&
2116   cmd->value.Connection && cmd->value.Connection->connector == ‘|’)
2117     {
2118       /* Создание конвеера между двумя командами */
2119       if (pipe (fildes) < 0)
2120       { /* возвращаем ошибку */ }
.......
           /* Выполняем первую команду из конвейера,  используя в качестве
              входных данных prev — вывод предыдущей команды,  а в качестве
              выходных fildes[1] — выходной файловый дескриптор,  полученный
              в результате вызова pipe() */
2178       execute_command_internal (cmd->value.Connection->first,  asynchronous, 
2179         prev,  fildes[1],  fd_bitmap);
2180 
2181       if (prev >= 0)
2182           close (prev);
2183 
2184       prev = fildes[0];    /* Наш вывод становится вводом для следующей команды */
2185       close (fildes[1]);
.......
2190       cmd = cmd->value.Connection->second;  /* “Сдвигаемся” на следующую команду из конвейера */
2191     }

Таким образом, bash обрабатывает символ конвейера путем системного вызова pipe() для каждого встретившегося символа ‘|’ и выполняет каждую команду в отдельном процессе с использованием соответствующих файловых дескрипторов в качестве входного и выходного потоков.

Исходный код, уровень 2, ядро

Обратимся к коду ядра и посмотрим на имплементацию функции pipe(). В статье рассматривается ядро версии 3.10.10 stable.
fs/pipe.c (пропущены незначительные для данной статьи участки кода):

/*
    Максимальный размер буфера конвейера для непривилегированного пользователя.
    Может быть выставлен рутом в файле /proc/sys/fs/pipe-max-size
  */
  35 unsigned int pipe_max_size = 1048576; 
   /*
     Минимальный размер буфера конвеера,  согласно рекомендации POSIX
     равен размеру одной страницы памяти,  т.е. 4Кб
    */
  40 unsigned int pipe_min_size = PAGE_SIZE;

 869 int create_pipe_files(struct file **res,  int flags)
 870 {
 871     int err;
 872     struct inode *inode = get_pipe_inode();
 873     struct file *f;
 874     struct path path;
 875     static struct qstr name = {. name = “” };
         /* Выделяем dentry в dcache */
 881     path.dentry = d_alloc_pseudo(pipe_mnt->mnt_sb,  &name);
         /* Выделяем и инициализируем структуру file. Обратите внимание
            на FMODE_WRITE,  а также на флаг O_WRONLY,  т.е. эта структура
            только для записи и будет использоваться как выходной поток
            в конвеере. К флагу O_NONBLOCK мы еще вернемся. */
 889     f = alloc_file(&path,  FMODE_WRITE,  &pipefifo_fops);
 893     f->f_flags = O_WRONLY | (flags & (O_NONBLOCK | O_DIRECT));
         /* Аналогично выделяем и инициализируем структуру file для чтения
            (см. FMODE_READ и флаг O_RDONLY) */
 896     res[0] = alloc_file(&path,  FMODE_READ,  &pipefifo_fops);
 902     res[0]->f_flags = O_RDONLY | (flags & O_NONBLOCK);
 903     res[1] = f;
 904     return 0;
 917 }
 918
 919 static int __do_pipe_flags(int *fd,  struct file **files,  int flags)
 920 {
 921     int error;
 922     int fdw,  fdr;
         /* Создаем структуры file для файловых дескрипторов конвеера
            (см. функцию выше) */ 
 927     error = create_pipe_files(files,  flags);
         /* Выбираем свободные файловые дескрипторы */
 931     fdr = get_unused_fd_flags(flags);
 936     fdw = get_unused_fd_flags(flags);
 941     audit_fd_pair(fdr,  fdw);
 942     fd[0] = fdr;
 943     fd[1] = fdw;
 944     return 0;
 952 }
     /* Непосредственно имплементация функций
        int pipe2(int pipefd[2],  int flags)... */
 969 SYSCALL_DEFINE2(pipe2,  int __user *,  fildes,  int,  flags)
 970 {
 971     struct file *files[2];
 972     int fd[2];
         /* Создаем структуры для ввода/вывода и ищем свободные дескрипторы */
 975     __do_pipe_flags(fd,  files,  flags);
         /* Копируем файловые дескрипторы из kernel space в user space */
 977     copy_to_user(fildes,  fd,  sizeof(fd));
         /* Назначаем файловые дескрипторы указателям на структуры */
 984     fd_install(fd[0],  files[0]);
 985     fd_install(fd[1],  files[1]);
 989 }
     /* ...и int pipe(int pipefd[2]),  которая по сути является
        оболочкой для вызова pipe2 с дефолтными флагами; */
 991 SYSCALL_DEFINE1(pipe,  int __user *,  fildes)
 992 {
 993     return sys_pipe2(fildes,  0);
 994 }

Если вы обратили внимание, в коде идет проверка на флаг O_NONBLOCK. Его можно выставить используя операцию F_SETFL в fcntl. Он отвечает за переход в режим без блокировки I/O потоков в конвеере. В этом режиме вместо блокировки процесс чтения/записи в поток будет завершаться с errno кодом EAGAIN.

Максимальный размер блока данных, который будет записан в конвейер, равен одной странице памяти (4Кб) для архитектуры arm:
arch/arm/include/asm/limits.h:

 8 #define PIPE_BUF    PAGE_SIZE

Для ядер >= 2.6.35 можно изменить размер буфера конвейера:

fcntl(fd,  F_SETPIPE_SZ,  <size>)

Максимально допустимый размер буфера, как мы видели выше, указан в файле /proc/sys/fs/pipe-max-size.