在 Linux 或 Mac 上的 C 中对 stdin/stdout 的非阻塞读/写

Question

我有两个程序通过命名管道进行通信（在 Mac 上），但命名管道的缓冲区大小太小。程序 1 在读取管道 2 之前将 50K 字节写入管道 1。命名管道为 8K（在我的系统上），因此程序 1 阻塞，直到数据被消耗。程序 2 从管道 1 读取 20K 字节，然后将 20K 字节写入管道 2。 Pipe2 不能容纳 20K 所以程序 2 现在阻塞。只有在程序 1 进行读取时才会释放它。但是程序1被阻塞等待程序2。死锁

我想我可以通过创建一个垫圈程序来解决这个问题，该程序读取标准输入非阻塞并写入标准输出非阻塞，将数据临时存储在一个大缓冲区中。我使用 cat data 测试了程序 | ./垫片 0 | ./gasket 1 > out，期望 out 是数据的副本。然而，虽然垫片的第一次调用按预期工作，但第二个程序中的读取在所有数据被消耗之前返回 0，并且在后续调用中永远不会返回 0 以外的任何内容。

我在 MAC 和 Linux 上都尝试了下面的代码。两者的行为相同。我添加了日志记录，以便我可以看到第二次调用垫片的 fread 开始没有数据，即使它没有读取第一次调用写入的所有数据。

#include <stdio.h>
#include <fcntl.h>
#include <time.h>
#include <stdlib.h>
#include <unistd.h>

#define BUFFER_SIZE 100000
char buffer[BUFFER_SIZE];
int elements=0;
int main(int argc, char **argv)
{
  int total_read=0, total_write=0;
  FILE *logfile=fopen(argv[1],"w");

  int flags = fcntl(fileno(stdin), F_GETFL, 0);
  fcntl(fileno(stdin), F_SETFL, flags | O_NONBLOCK);
  flags = fcntl(fileno(stdout), F_GETFL, 0);
  fcntl(fileno(stdout), F_SETFL, flags | O_NONBLOCK);

  while (1) {
    int num_read=0;
    if (elements < (BUFFER_SIZE-1024)) { // space in buffer
      num_read = fread(&buffer[elements], sizeof(char), 1024, stdin);
      elements += num_read;
      total_read += num_read;
      fprintf(logfile,"read %d (%d) elements \n",num_read, total_read); fflush(logfile);
    }
    if (elements > 0) { // something in buffer that we can write
      int num_written = fwrite(&buffer[0],sizeof(char),elements, stdout); fflush(stdout);
      total_write += num_written;
      fprintf(logfile,"wrote %d (%d) elements \n",num_written, total_write); fflush(logfile);
      if (num_written > 0) { // copy data to top of buffer
        for (int i=0; i<(elements-num_written); i++) {
          buffer[i] = buffer[i+num_written];
        }
        elements -= num_written;
      }
    }
  }
}

我想我可以让垫片成为多线程，并在一个线程中使用阻塞读取并在另一个线程中使用阻塞写入，但我想了解为什么非阻塞 IO 似乎对我不利。

谢谢！

Answer 1

我对任何 IPC 项目的一般解决方案是使客户端和服务器非阻塞 I/O。为此，需要在写入和读取时对数据进行排队，以处理操作系统无法读取/写入或只能读取/写入部分消息的情况。

下面的代码可能看起来有点矫枉过正，但如果你让它工作，你可以在你的职业生涯中使用它，无论是命名管道、套接字、网络，你都可以命名。

在伪代码中：

typedef struct {
  const char* pcData, * pcToFree; // pcData may no longer point to malloc'd region
  int   iToSend;
} DataToSend_T;

queue of DataToSend_T qdts;

// Caller will use malloc() to allocate storage, and create the message in
// that buffer.  MyWrite() will free it now, or WritableCB() will free it
// later.  Either way, the app must NOT free it, and must not even refer to
// it again.

MyWrite( const char* pcData, int iToSend ) {
  iSent = 0;

  // Normally the OS will tell select() if the socket is writable, but if were hugely
  // compute-bound, then it won't have a chance to.  So let's call WritableCB() to
  // send anything in our queue that is now sendable.  We have to send the data in
  // order, of course, so can't send the new data until the entire queue is done.
  WritableCB();

  if ( qdts has no entries ) {
     iSent = write( pcData, iToSend );
      // TODO: check error
      // Did we send it all?  We're done.
      if ( iSent == iToSend ) {
          free( pcData );
          return;
      }
  }

  // OK, either 1) we had stuff queued already meaning we can't send, or 2)
  // we tried to send but couldn't send it all.
  add to queue qdts the DataToSend ( pcData + iSent, pcData, iToSend - iSent );
}



WritableCB() {
  while ( qdts has entries ) {
      DataToSend_T* pdts = qdts head;
      int iSent = write( pdts->cData, pdts->iToSend );
      // TODO: check error
      if ( iSent == pdts->iToSend ) {
          free( pdts->pcToFree );
          pop the front node off qdts
      else {
          pdts->pcData  += iSent;
          pdts->iToSend -= iSent;
          return;   
      }
  }
}



// Off-subject but I like a TINY buffer as an original value, that will always
// exercise the "buffer growth" code for almost all usage, so we're sure it works.
// If the initial buffer size is like 1M, and almost never grows, then the grow code
// may be buggy and we won't know until there's a crash years later.

int iBufSize = 1, iEnd = 0;  iEnd is the first byte NOT in a message
char* pcBuf = malloc( iBufSize );

ReadableCB() {
  // Keep reading the socket until there's no more data.  Grow buffer if necessary.
  while (1) {
      int iRead = read( pcBuf + iEnd, iBufSize - iEnd);
      // TODO: check error
      iEnd += iRead;

      // If we read less than we had space for, then read returned because this is
      // all the available data, not because the buffer was too small.
      if ( iRead < iBufSize - iEnd )
          break;

      // Otherwise, double the buffer and try reading some more.
      iBufSize *= 2;
      pcBuf = realloc( pcBuf, iBufSize );
  }

  iStart = 0;
  while (1) {
      if ( pcBuf[ iStart ] until iEnd-1 is less than a message ) {
          // If our partial message isn't at the front of the buffer move it there.
          if ( iStart ) {
              memmove( pcBuf, pcBuf + iStart, iEnd - iStart );
              iEnd -= iStart;
          }
          return;
      }
      // process a message, and advance iStart by the size of that message.
  }
}



main() {
  // Do your initial processing, and call MyWrite() to send and/or queue data.

  while (1) {
       select() // see man page
       if ( the file handle is readable )
           ReadableCB();
       if ( the file handle is writable )
           WritableCB();
       if ( the file handle is in error )
           // handle it;
       if ( application is finished )
           exit( EXIT_SUCCESS );
  }
}