共有メモリを用いたストリームの実験

共有メモリを用いたストリームの実験#

共有メモリは UDP などの通信プロトコルに比べて速いと聞くため、 共有メモリを用いた軽いストリームを実装して性能を確認した。

共有メモリを用いたストリーム#

今回、共有メモリを用いて以下のようなリングバッファによるストリームを cpp-shm-stream リポジトリ に用意した。

  • light_stream: アトミック演算による wait free なリングバッファのストリーム

  • blocking_stream: アトミック演算を主に用いつつ、 データの待機処理がビジーループなしで行えるように実装したリングバッファのストリーム

どちらも、リングバッファ用に

  • 書き込み側のインデックス

  • 読み込み側のインデックス

  • バッファのサイズ

  • バッファ

を共有メモリ上に共有する。 2 つのインデックスについてはアトミック演算を行うため、 共有メモリ上でもアトミック演算を行うことができる Boost.Atomic ライブラリ の boost::atomics::ipc_atomic クラスを用いた。

比較の対象とした通信方法#

以下を対象に、通信時間の比較を行った。

  • 共有メモリを用いたストリーム

  • UDP

    • UDP による通信の実装には Asio ライブラリを用いた。

    • IPv4 と IPv6 の UDP の両方を対象とした。

環境#

実験を行った環境は以下の通り。

  • CPU：Intel(R) Core(TM) i7-9750H CPU @ 2.60GHz

  • メモリ：16 GB

  • OS：Ubuntu 22.04

レイテンシの測定#

各通信方法とデータサイズ（32 byte, 1 KB, 32 kB, 1 MB）ごとに 10000 回往復の通信をし、データを送受信するのにかかる時間を箱ひげ図でプロットした。

Hide code cell content

 
import typing

import msgpack
import pandas


def parse_data(filepath: str) -> pandas.DataFrame:
    with open(filepath, mode="rb") as file:
        raw_data = msgpack.load(file)

    df: typing.Optional[pandas.DataFrame] = None

    for measurement in raw_data["measurements"]:
        if measurement["measurer_name"] != "Processing Time":
            continue

        protocol = measurement["case_name"]
        data_size = measurement["params"]["size"]
        time_list = measurement["durations"]["values"][0]

        cur_df = pandas.DataFrame(
            {
                "protocol": [protocol] * len(time_list),
                "data_size": [data_size] * len(time_list),
                "time": time_list,
            }
        )
        if df is None:
            df = cur_df
        else:
            df = pandas.concat([df, cur_df])

    assert df is not None

    return df

Hide code cell source

 
import pandas
import plotly.express as px
import plotly.graph_objects as go

bench_results = parse_data("shm_stream_result_20230414/ping_pong.data")

fig = px.box(
    bench_results,
    x="data_size",
    y="time",
    color="protocol",
    log_y=True,
    points=False,
    labels = {
        "data_size": "データサイズ [byte]",
        "time": "レイテンシ [sec]",
        "protocol": "通信方法",
    },
    title="レイテンシの測定結果",
)
fig.show(renderer="notebook_connected")

---------------------------------------------------------------------------
ExtraData                                 Traceback (most recent call last)
Cell In[2], line 5
      2 import plotly.express as px
      3 import plotly.graph_objects as go
----> 5 bench_results = parse_data("shm_stream_result_20230414/ping_pong.data")
      7 fig = px.box(
      8     bench_results,
      9     x="data_size",
   (...)     19     title="レイテンシの測定結果",
     20 )
     21 fig.show(renderer="notebook_connected")

Cell In[1], line 9, in parse_data(filepath)
      7 def parse_data(filepath: str) -> pandas.DataFrame:
      8     with open(filepath, mode="rb") as file:
----> 9         raw_data = msgpack.load(file)
     11     df: typing.Optional[pandas.DataFrame] = None
     13     for measurement in raw_data["measurements"]:

File /builds/MusicScience37/til/.venv/lib/python3.13/site-packages/msgpack/__init__.py:47, in unpack(stream, **kwargs)
     40 """
     41 Unpack an object from `stream`.
     42 
     43 Raises `ExtraData` when `stream` contains extra bytes.
     44 See :class:`Unpacker` for options.
     45 """
     46 data = stream.read()
---> 47 return unpackb(data, **kwargs)

File msgpack/_unpacker.pyx:201, in msgpack._cmsgpack.unpackb()

ExtraData: unpack(b) received extra data.

送信時間の測定#

各通信方法とデータサイズ（32 byte, 1 KB, 32 kB, 1 MB）ごとに 10000 回データを送信し、データの送信処理にかかる時間を箱ひげ図でプロットした。

Hide code cell source

 
bench_results = parse_data("shm_stream_result_20230414/send_messages.data")

fig = px.box(
    bench_results,
    x="data_size",
    y="time",
    color="protocol",
    log_y=True,
    points=False,
    labels = {
        "data_size": "データサイズ [byte]",
        "time": "送信時間 [sec]",
        "protocol": "通信方法",
    },
    title="送信時間の測定結果",
)
fig.show(renderer="notebook_connected")

比較#

  • 共有メモリを用いた通信はどちらも UDP より速くなっている。

  • IPv4 と IPv6 の UDP の両方を試したが、 IPv4 と IPv6 では特に差が見受けられなかった。

  • 共有メモリを用いた通信 2 種類では、 アトミック演算のみを使用した light_stream の方が速い。 light_stream ではデータの待機処理にビジーループを用いているため比較的負荷はかかるが、 代わりに速度は速くなった。

ソースコードなど#

実験は cpp-shm-stream リポジトリ のコミット fa8c023d51fcf9ac9f398a11c7e548fc0539255a 上で Release ビルドを行い、 以下のコマンドを実行することで行った。

  • ./build/Release/bin/bench_send_messages --msgpack send_messages.data --samples 10000

  • python3 ./tests/bench/ping_pong/bench.py build/Release/

Contents