ノンブロッキングパイプで別のサーバーに転送するプロキシを設計する

Question

トラフィックを複製するプロキシを作成しました。私はすべての入力を受け取り、すべての要求を処理する必要があるレプリカサーバーにネットワークトラフィックを複製しようとしています。ただし、メインサーバー上の応答のみがクライアントに表示されます。 https://gist.github.com/nipunarora/679d49e81086b5a75195ec35ced646de

テストが小さく、データとトランザクションのために動作しているようですが、私はを使用する際は、次のエラーを取得しているようだ：

Thread 1. Take input from client forward it to a pipe in non-blocking way, and to the server 
Thread 2. Read from server and send to client 
Thread 3. Read from pipe and forward to replica server 
Thread 4. Read from replica server and drop 

コードこの要旨で提供されていますが、次のようにハイレベルのワークフローがありますiperfのより大きなデータセット：

Buffer overflow? : Resource temporarily unavailable 

問題が起因されるコード内の特定の部分：

void forward_data_asynch(int source_sock, int destination_sock) { 
    char buffer[BUF_SIZE]; 
    int n; 

    //put in error condition for -1, currently the socket is shutdown 
    while ((n = recv(source_sock, buffer, BUF_SIZE, 0)) > 0)// read data from input socket 
    { 
     send(destination_sock, buffer, n, 0); // send data to output socket 
     if(write(pfds[1],buffer,n) < 0)//send data to pipe 
     { 
      //fprintf(stats_file,"buffer_overflow \n"); 
      //printf("format string" ,a0,a1); 
      //int_timeofday(); 
      perror("Buffer overflow? "); 
     } 
     //DEBUG_PRINT("Data sent to pipe %s \n", buffer); 
    } 

    shutdown(destination_sock, SHUT_RDWR); // stop other processes from using socket 
    close(destination_sock); 

    shutdown(source_sock, SHUT_RDWR); // stop other processes from using socket 
    close(source_sock); 
} 

次のように読み取り処理は次のとおりです。

void forward_data_pipe(int destination_sock) { 

    char buffer[BUF_SIZE]; 
    int n; 
    sleep(10); 
    //put in error condition for -1, currently the socket is shutdown 
    while ((n = read(pfds[0], buffer, BUF_SIZE)) > 0)// read data from pipe socket 
    { 
     //sleep(1); 
     //DEBUG_PRINT("Data received in pipe %s \n", buffer); 
     send(destination_sock, buffer, n, 0); // send data to output socket 
    } 

    shutdown(destination_sock, SHUT_RDWR); // stop other processes from using socket 
    close(destination_sock); 
} 

注意してください、次のように、パイプが定義されています

/** Make file descriptor non blocking */ 
int setNonblocking(int fd) 
{ 
    int flags; 

    /* If they have O_NONBLOCK, use the Posix way to do it */ 
#if defined(O_NONBLOCK) 
    /* Fixme: O_NONBLOCK is defined but broken on SunOS 4.1.x and AIX 3.2.5. */ 
    if (-1 == (flags = fcntl(fd, F_GETFL, 0))) 
    flags = 0; 
    return fcntl(fd, F_SETFL, flags | O_NONBLOCK); 
#else 
    /* Otherwise, use the old way of doing it */ 
    flags = 1; 
    return ioctl(fd, FIOBIO, &flags); 
#endif 
} 

は、誰もが、エラーの理由何ができるか固定に助けてもらえますか？

Answer 1

あなたの場合の問題は、非ブロッキングモードに設定されているソケットにデータが非常に速く送信されることです。いくつかのオプションがあります：

1. データが失われる可能性があることを受け入れます。メインサーバーでの処理を遅延させたくない場合は、これが唯一のオプションです。
2. ソケットを非ブロッキングモードに設定しないでください。デフォルトモードであるblockingは、データを失わないようにしたい場合にアプリケーションに適しているようです。しかし、これはまた、システムが遅くなる可能性があることを意味します。
3. ソケットに十分なバッファスペースがあるまで待機するには、poll（）、select（）、kqueue（）、epoll（）、/ dev/pollなどを使用します。しかし、これを使用するときには、それでもブロックする必要がある場合は、最初にソケットを非ブロックモードに設定する必要があります。これはまた、システムの減速につながる。