パイプと socketpair

pipe は単方向のバイトストリームを提供し、socketpair は既に接続されたローカルソケットのペアを作成し、双方向通信に適しています。両方ともファイルディスクリプタを通じて参照とライフサイクルを表現し、正常終了、部分的な転送、エラーを区別するためにプロトコルが必要です。

パイプ：容量とアトミック書き込みは別の概念

pipefd[0] は読み込み側、pipefd[1] は書き込み側です。Linux の pipe2 は作成時に O_CLOEXEC と O_NONBLOCK をアトミックに設定でき、作成後にフラグを設定する際の競合状態を回避できます。

PIPE_BUF は複数の書き込み者が書き込みを交差させないことを保証する境界であり、パイプの総容量ではありません。この境界以内の書き込みはアトミック性が保証されますが、これを超えると他の書き込み者と交差する可能性があります。ノンブロッキングモードでは、PIPE_BUF 以下の書き込みでも空き容量がない場合は EAGAIN が返され、部分的な成功が必ずしも起こるわけではありません。pipe(7)

読み込みにはメッセージ境界がありません：1回の write が1回の read に対応するとは限らず、書き込み自体が他の書き込み者と交差しない場合でも同様です。アプリケーションは固定長、長さプレフィックス、またはエスケープ区切り文字のプロトコルを採用できます。

容量はバッファリングできるデータの量に影響します。一般的な Linux 設定ではマルチページ容量が使用されますが、ページサイズ、ユーザークォータ、カーネル制限によって結果は変わります。F_GETPIPE_SZ でクエリし、F_SETPIPE_SZ で調整を要求し、実際の戻り値を確認してください。バッファを大きくするとバーストが平滑化される可能性がありますが、キューイングレイテンシが増加する可能性もあり、生産者が消費者より速いという不均衡を解消することはできません。

長さプレフィックスには、読み込みバッチを跨ぐ解析状態が必要

以下では、各メッセージの先頭に4バイトのビッグエンディアン符号なし長さを配置し、それに続くペイロードを配置します。空メッセージを許可し、1メッセージあたり最大1024バイトとします。パーサーは読み込まれたバイトチャンクを受け取りますが、チャンク境界はプロトコルには含まれません。例ではヘッダーとペイロードを意図的に分割していますが、それでも2つの完全なメッセージが得られることを検証します。

class Decoder:
    def __init__(self):
        self.pending = bytearray()
        self.expected = None

    def feed(self, chunk):
        self.pending.extend(chunk)
        messages = []
        while True:
            if self.expected is None:
                if len(self.pending) < 4:
                    break
                self.expected = int.from_bytes(self.pending[:4], "big")
                del self.pending[:4]
                if self.expected > 1024:
                    raise ValueError("message too large")
            if len(self.pending) < self.expected:
                break
            messages.append(bytes(self.pending[:self.expected]))
            del self.pending[:self.expected]
            self.expected = None
        return messages

    def finish(self):
        if self.expected is not None or self.pending:
            raise ValueError("truncated message at EOF")

wire = (3).to_bytes(4, "big") + b"abc" + (5).to_bytes(4, "big") + b"hello"
decoder = Decoder()
messages = []
for chunk in [wire[:2], wire[2:6], wire[6:]]:
    messages.extend(decoder.feed(chunk))
assert messages == [b"abc", b"hello"]
decoder.finish()
assert Decoder().feed(bytes(4)) == [b""]
for bad in [(1025).to_bytes(4, "big"), b"\0\0", (3).to_bytes(4, "big") + b"ab"]:
    try:
        decoder = Decoder()
        decoder.feed(bad)
        decoder.finish()
    except ValueError:
        pass
    else:
        raise AssertionError("malformed stream accepted")

解析エラーが発生した場合、この例ではそのストリームを放棄し、次のメッセージヘッダーを推測して再同期しようとしません。アプリケーションが再同期を必要とする場合、プロトコルは明確な回復ルールを設計する必要があります。feed は入力チャンクと現在のバッチの完全なメッセージをメモリに一時保存するため、実際の接続では単回の read サイズと待機中のメッセージキューを制限する必要があります。「1メッセージ1024バイト」だけでプロセス全体のメモリ使用量が制限されるわけではありません。

EOF を受信することは、バイトストリームが終了したことを示すだけです。ペイロードの2バイトが不足している場合、半分のメッセージを成功として扱うのではなく、切り捨てとして報告すべきです。すでに配信された以前の完全なメッセージを元に戻す必要があるかどうかは、上位層のトランザクションプロトコルの問題です。

EOF と SIGPIPE は参照状態に依存する

すべての書き込み側が閉じられ、バッファが読み尽くされると、read は 0 を返します。書き込み時にすべての読み込み側が閉じられている場合、通常 SIGPIPE が発生します。このシグナルが無視または適切に処理された場合、write は EPIPE で失敗します。空のバッファだが書き込み側が残っている状態は、真の EOF とは異なる状態です。

fork、dup、および継承は参照を増やします。O_CLOEXEC は成功した exec の際に該当するディスクリプタを閉じますが、fork の直後に使用しないエンドを閉じてくれるわけではありません。ソケットの shutdown(SHUT_WR) は、特定の方向を明示的にハーフクローズでき、単に fd 参照を減らすこととは異なります。

書き込み側を1つ閉じても、なぜまだ EOF ではないのか？

書き込み側を複製し、元の参照を閉じた後、空の読み込みでも EAGAIN になります。その後、書き込んで最後の書き込み側を閉じると、最初の読み込みではまだバッファ内のデータが返され、次の読み込みで初めて EOF になります。閉じるのは参照であり、読み込むのはバイトです。これらの2つの状態は一緒に判断する必要があります。

EOF を検証する親子プロセスの例

以下は、固定の5バイトペイロードのみを送信し、短書き込みと EINTR を処理し、子プロセスが EOF を読み取った後に総長さを確認する例です。

#define _GNU_SOURCE
#include <errno.h>
#include <fcntl.h>
#include <stddef.h>
#include <sys/wait.h>
#include <unistd.h>

int main(void) {
    int fd[2];
    if (pipe2(fd, O_CLOEXEC) == -1) return 1;
    pid_t child = fork();
    if (child == -1) { close(fd[0]); close(fd[1]); return 1; }
    if (child == 0) {
        close(fd[1]);
        size_t total = 0;
        char buf[32];
        for (;;) {
            ssize_t n = read(fd[0], buf, sizeof buf);
            if (n > 0) total += (size_t)n;
            else if (n == 0) break;
            else if (errno != EINTR) _exit(2);
        }
        close(fd[0]);
        _exit(total == 5 ? 0 : 3);
    }
    close(fd[0]);
    const char payload[] = "hello";
    size_t done = 0;
    int failed = 0;
    while (done < 5) {
        ssize_t n = write(fd[1], payload + done, 5 - done);
        if (n > 0) done += (size_t)n;
        else if (n < 0 && errno == EINTR) continue;
        else { failed = 1; break; }
    }
    close(fd[1]);
    int status;
    while (waitpid(child, &status, 0) == -1) {
        if (errno != EINTR) return 1;
    }
    return failed || !WIFEXITED(status) || WEXITSTATUS(status) != 0;
}

cc -Wall -Wextra pipe_demo.c -o pipe_demo でコンパイルします。実際のプロトコルでは、相手側の早期終了、タイムアウト、SIGPIPE ポリシー、入力サイズの制限を処理する必要があります。ここでの長さの確認はライフサイクルを説明するためのものであり、一般的なメッセージ完全性チェックではありません。

タイムアウトを待たなくても、不要な書き込み側が EOF を発生させないことを検証できる

以下の単一プロセスの例では、書き込み側を複製し、元の書き込み側を閉じた後、空のパイプに対してノンブロッキング読み込みを行います。複製された書き込み側が存在するため、結果は EAGAIN になります。最後の書き込み側を閉じた後、同じ読み込みで初めて 0 が返ります。

#define _GNU_SOURCE
#include <errno.h>
#include <fcntl.h>
#include <stdio.h>
#include <unistd.h>
int main(void) {
    int fd[2];
    if (pipe2(fd, O_NONBLOCK | O_CLOEXEC) == -1) return 1;
    int extra = fcntl(fd[1], F_DUPFD_CLOEXEC, 0);
    if (extra == -1) { close(fd[0]); close(fd[1]); return 1; }
    if (close(fd[1]) == -1) { close(extra); close(fd[0]); return 1; }
    char byte;
    errno = 0;
    ssize_t before = read(fd[0], &byte, 1);
    int saved = errno;
    int failed = !(before == -1 && (saved == EAGAIN || saved == EWOULDBLOCK));
    if (close(extra) == -1) failed = 1;
    if (read(fd[0], &byte, 1) != 0) failed = 1;
    if (close(fd[0]) == -1) failed = 1;
    if (!failed) puts("writer remains: EAGAIN; last writer closed: EOF");
    return failed;
}

pipe-refs.c として保存し、cc -Wall -Wextra pipe-refs.c -o pipe-refs でコンパイルして実行します。この例ではデータを書き込んでいないため、最後の閉鎖の直後に EOF が確認できます。すでにデータがある場合は、まずそれを読み尽くす必要があります。F_DUPFD_CLOEXEC は新しいディスクリプタにも FD_CLOEXEC フラグを設定し、すべての複製インターフェースが自動的に FD_CLOEXEC を継承すると誤解しないようにします。F_DUPFD_CLOEXEC

socketpair：タイプが境界を決定する

int sv[2];
int rc = socketpair(AF_UNIX, SOCK_STREAM | SOCK_CLOEXEC, 0, sv);
/* rc を確認；sv[0]、sv[1] はどちらも読み書き可能で、終了時にそれぞれ close する。 */

これは <sys/socket.h> が必要な API の断片です。socketpair の両端は作成時に互いに接続されており、bind、listen、accept を行う必要はありません。socketpair(2)

Linux の AF_UNIX データグラムは UDP の配信セマンティクスとは異なり、「信頼性なし」として記述することはできません。SOCK_SEQPACKET も Linux 5.3 で登場したわけではありません。信頼性はインターフェース範囲内の転送セマンティクスを示すものであり、受信プロセスが読み取った後にビジネス処理が必ず完了することを保証するものではありません。クラッシュリカバリには上位層プロトコルが必要です。unix(7)

AF_UNIX は IP/TCP パスを回避しますが、独自のプロトコル実装、バッファ、バックプレッシャーがあります。「プロトコルスタックを通らない、フロー制御がない」という記述は正確ではありません。レイテンシとスループットは、メッセージサイズ、並発性、コピーパス、スケジューリングに基づいて測定する必要があります。

クレデンシャルとファイルディスクリプタの伝送

SCM_RIGHTS は開かれたファイル記述子への参照を伝送し、受信プロセスは独自の fd 番号を取得しますが、送信者の整数をそのまま利用可能なハンドルに変換するわけではありません。共有されるファイルオフセットと状態も、その開かれたファイル記述子に関連する可能性があり、所有権と閉鎖責任を理解する必要があります。

SO_PEERCRED は接続相手のクレデンシャルを照会するために使用でき、SCM_CREDENTIALS はメッセージの補助データメカニズムです。受信ポリシーとカーネルの検証はインターフェースの規定に従います。カーネルが提供する身份信息は、アプリケーションの認可ルールと組み合わせて使用し、UID が見えただけで無条件に操作を信頼してはいけません。

recvmsg では、MSG_CTRUNC、制御バッファのサイズ、受信した fd の数をチェックする必要があります。メッセージを拒否する場合、受信したが使用しない fd を閉じる必要があります。Linux の MSG_CMSG_CLOEXEC は、受信したディスクリプタが exec を跨いで意図せず漏洩するのを防ぎます。unix(7) 補助データ

splice：ユーザーバッファの往復を減らすものの、絶対的なコピーなしを保証しない

splice は少なくとも一端がパイプであることを要求し、サポートされている fd とパイプの間でデータを転送できます。特定のパスではページ参照を移動し、ユーザーバッファにコピーしてからカーネルに書き戻すのを回避できますが、実際にコピーされるかどうかはエンドポイントとカーネルの実装に依存し、物理データが1回もコピーされないことを一律に保証することはできません。splice(2)

1回の splice でも短転送、ブロック、または EAGAIN の返却が発生する可能性があります。ファイルからパイプを経てソケットへのループでは、パイプに入力済みだがまだ送信されていないバイトを記録し、バックプレッシャーを先に処理して、繰り返し読み込みや末尾のドロップを避けてください。sendfile と関連する最適化用途がありますが、sendfile のすべての実装詳細を特定の2回の splice 呼び出しに固定することはできません。

eventfd と選択

eventfd はカーネルが管理する64ビットカウンターインターフェースです：通常の読み込みは累積値を返し、ゼロにリセットします。EFD_SEMAPHORE モードでは、読み込むたびに1ずつ減算します。通知や作業カウントに適しており、任意のバイトメッセージを転送するものではなく、単方向のエンドポイントのみを介して転送されるパイプでもありません。eventfd(2)

単方向ストリームには pipe を選択；双方向バイトプロトコルには socketpair STREAM を選択；レコード境界が必要な場合は SEQPACKET を検討；fd や身份信息を転送するには AF_UNIX 補助データを使用；通知のみが必要な場合は eventfd を検討します。大容量共有データのライフサイクルと同期については、共有メモリとセマフォ を参照してください。

キーワード：pipe, PIPE_BUF, EOF, SIGPIPE, socketpair, AF_UNIX, SOCK_SEQPACKET, SCM_RIGHTS, credentials, splice, eventfd, backpressure