mmap＋POSIXセマフォによる高速なプロセス間通信 - mp::ipc_vector

同じホストの異なるプロセスの間で、高速な通信を行いたいときがあります：

fork() したプロセスと通信したい場合や、「アプリケーションたくさん → localhostのクライアントデーモン１つ → サーバー」という形で接続したい*1 *2 場合などなど。
あるいは共有リソースを管理するプロセスが必要なとき、具体的には memcached が localhost で動いるイメージ。

こんなときに高速なプロセス間通信がしたい。スレッドなみに高速だと嬉しい。

そこで、POSIXセマフォを使って排他制御をしつつ、mmap(2) を使って共有したメモリの上でデータをやりとりすることで、高速にプロセス間で通信を行えるようにするライブラリを作ってみました。

さっそくベンチマーク：

	時間	スループット換算
UNIXドメインソケット	4.44 sec	25 MB/s
ipc_vector（sleepなし）	たぶん3.76 secくらい	30.7 MB/s
ipc_vector（sleepあり）	0.72 sec	160 MB/s

環境は Linux vcore.local 2.6.22.9-vcore16 #1 SMP Sun Oct 14 22:13:32 JST 2007 x86_64 AMD Athlon(tm) 64 X2 Dual Core Processor 5000+ GNU/Linux

UNIXドメインソケットは、片方のプロセスでひたすらwriteし、もう片方でひたすらreadしたものです。writeが終了するまでの時間を計測しています。実際にはreadしたバイト列は一つ一つのメッセージ単位に切り出さないといけないため、もう少しだけ遅くなると思います。

ipc_vector（sleepなし）は、mmap(2) + POSIX セマフォのプロセス間通信で、受け取り（read）側はひたすら受け取るだけで他は何もしないもの。結果が非常に不安定で、1秒で終わることもあれば5秒以上かかることもありました。これはLinuxのPOSIXの実装が

まずCPUのアトミック命令を使う（ユーザー空間で完結する）
競合が発生しているようならfutex(2)（システムコールで遅い）を呼ぶ

という実装になっているので、プロセススケジューラの具合によって競合が多く発生し、スループットが大きく変わったのだろうと思われます。

ipc_vector（sleepあり）は、受け取り側で多少は重い処理もするだろう、ということで、たまに usleep(1) を挟んでいます。

ベンチマークに使ったプログラム一式：ipc-vector-benchmark.tar.gz

ipc_vectorの実装ですが、基本的には sem_wait(3) と sem_post(3) で排他制御しつつ、mmap(2) に MAP_SHARED フラグを渡して共有したメモリ空間でデータをやりとりしているだけです。

POSIX セマフォはシステム全体で共有される「名前付きセマフォ」ではなく、共有されたメモリ領域に置かれる「名前なしセマフォ」を使っています（SEM_OVERVIEW(7)）*3

また mmap しておいた領域が足りなくなったら、ftruncate(2) してファイルを大きくし、再度 mmap(2) をやり直しています。可変長mmap ;- )

ipc_vectorのソースコードは mpio ライブラリの中に含まれていて、CodeRepos://lang/c/mpio にあります。

ipc-server.cc

#include <iostream>
#include <sys/time.h>
#include <mp/ipc_vector.h>

int main(void)
{
    mp::ipc_vector::server vec("test.map");
    size_t n = 0;
    while(true) {
        mp::ipc_vector::buffer buf;
        vec.receive(buf);
        for(mp::ipc_vector::buffer::iterator it(buf.begin()), it_end(buf.end());
                it != it_end; ++it) {
            ++n;
            //std::cout.write((*it).ptr, (*it).size) << std::endl;
        }
#ifdef USE_USLEEP
        usleep(1);
#endif
    }
}

ipc-client.cc

#include <iostream>
#include <mp/ipc_vector.h>

#include "timer.h"

int main(void)
{
    mp::ipc_vector::client vec("test.map");

    const char* tmp =
        "abcdefghijklmnopqrstuvwxyz"
        "abcdefghijklmnopqrstuvwxyz"
        "abcdefghijklmnopqrstuvwxyz"
        "abcdefghijklmnopqrstuvwxyz"
        "abcdefghijklmnopqrstuvwxyz"
        "abcdefghijklmnopqrstuvwxyz"
        ;

    reset_timer();

    size_t n = 0;
    for(size_t j=0; j < 10000; ++j) {
        for(size_t i=1; i < strlen(tmp); ++i) {
            vec.push(tmp, i);
            n += i;
        }
    }

    show_timer(n);

    std::cout << n << std::endl;
}