0
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?

@k2-gc

pybind11 をかじる Part3

0
Posted at

pybind11 をかじる Part 3

本記事では pybind11 の基礎的な使い方を学びます．
本記事は 3 部構成です．

はじめに

Part 2 では，ポインタ，所有権，py::capsule，opaque handle を扱いました．

この記事では，もう少し実践寄りの機能を扱います．

  • C++ から Python callback を呼ぶ
  • NumPy 配列を C++ で読む・書き換える
  • C++ 側のメモリを NumPy 配列として返す
  • 重い C++ 処理中に GIL を解放する

目次

Step 6: Python Callback

説明

callback は，処理の一部を呼び出し元から関数として受け取る仕組みです．pybind11 では Python の関数や lambda を C++ の std::function として受け取り，C++ 側の好きなタイミングで呼び出せます．

たとえば，C++ 側でイベントを検出し，その結果だけを Python の関数へ通知できます．C++ 側は「通知先で何をするか」を知らなくてよいため，ログ出力，結果の保存，UI の更新などを Python 側で差し替えられます．

Python callable と std::function の変換を有効にするには，次の header が必要です．

#include <functional>
#include <pybind11/functional.h>

<functional> は std::function の定義，<pybind11/functional.h> は Python callable と std::function の相互変換を提供します．

呼び出しの流れは次のようになります．

Python の関数や lambda
        ↓
pybind11 が std::function に変換
        ↓
C++ の処理中に callback を呼ぶ
        ↓
Python の処理へ戻る

通常どおり Python から C++ 関数を呼び，その関数内で同期的に callback を実行する場合，GIL は保持されたままです．一方，C++ の別スレッドから呼ぶ場合や，py::gil_scoped_release で GIL を解放した後に呼ぶ場合は，callback の直前に py::gil_scoped_acquire で GIL を取得する必要があります．

Python callback の実行に失敗すると，Python の例外は C++ 側で py::error_already_set として扱われます．C++ 側で捕まえなければ，pybind11 が Python の例外へ戻して呼び出し元へ伝えます．捕まえる場合は，処理を中止する，失敗した要素だけ飛ばす，別の値を返す，といった方針を C++ 側で決めます．

コード例

void notify_event(std::function<void(std::string)> cb) {
    cb("Event Triggered!");
}

m.def("notify_event", &notify_event);

Python:

mymodule.notify_event(lambda msg: print(msg))

notify_event() が cb(...) を実行すると，Python 側で渡した lambda が呼ばれます．

複数の値へ callback を適用し，失敗した要素だけを飛ばす場合は，次のように例外を捕まえられます．

try {
    result.push_back(f(x));
} catch (py::error_already_set& e) {
    std::cerr << "Error applying function: " << e.what() << std::endl;
}

この例では例外を表示して処理を続けます．例外を握りつぶす設計になるため，呼び出し元へ失敗を知らせる必要がないかも検討します．

チャレンジ

  • notify_event(cb) で Python callback を 1 回呼ぶ
  • map_list(data, f) で Python callback を各要素に適用する
  • map_list_safe(data, f) で callback の例外を C++ 側で捕まえて，失敗した要素を skip する

Step 8: NumPy 連携

説明

NumPy 配列は，ざっくり言うと次の情報を持っています．

data pointer
dtype
shape
strides

pybind11 の py::array_t<T> を使うと，Python の NumPy 配列を C++ 側で型付きの配列として受け取れます．
この Step では，読み取り，in-place 書き換え，C++ 側メモリを NumPy として返す，という 3 パターンを確認します．

利用するには，次の header を追加します．

#include <pybind11/numpy.h>

py::array_t<T, flags> の T は要素型，flags は受け入れるメモリ配置や変換方法を表します．たとえば，py::array::c_style は C 言語と同じ row-major の連続したメモリ配置を要求します．py::array::forcecast を付けると，要素型や配置が条件と異なる場合に，変換した一時配列を作って受け入れます（公式ドキュメント）．

NumPy 配列は要素データだけでなく，次元数，shape，strides も持ちます．strides は各次元で隣の要素へ移動するためのバイト数です．C++ 側で生ポインタを使う場合は，想定した次元数，dtype，メモリ配置になっているかを確認する必要があります．

コード例

読み取り例:

double sum_array(py::array_t<double, py::array::c_style | py::array::forcecast> arr) {
    auto buf = arr.request();
    if (buf.ndim != 1) {
        throw std::runtime_error("1D array expected");
    }

    auto* ptr = static_cast<double*>(buf.ptr);
    double sum = 0.0;
    for (py::ssize_t i = 0; i < buf.shape[0]; ++i) {
        sum += ptr[i];
    }
    return sum;
}

request() は，配列の data pointer，次元数，shape，strides などを py::buffer_info として取得します．ここでは c_style によりデータが連続しているため，buf.ptr を double* に変換して先頭から順番に読めます．

forcecast によって一時配列が作られても，sum_array() は読み取りだけなので問題ありません．一方，C++ 側の変更を呼び出し元の配列へ反映したい場合，一時配列を書き換えても元の配列は変化しません．そのため，in-place 処理では forcecast を安易に付けず，期待する dtype と配置の配列だけを受け取る方が挙動を理解しやすくなります．

in-place 書き換えの例です．

void brighten(py::array_t<uint8_t, py::array::c_style> img) {
    auto r = img.mutable_unchecked<2>();
    for (py::ssize_t i = 0; i < r.shape(0); ++i) {
        for (py::ssize_t j = 0; j < r.shape(1); ++j) {
            int value = r(i, j) * 2;
            r(i, j) = value > 255 ? 255 : static_cast<uint8_t>(value);
        }
    }
}

mutable_unchecked<2>() は，書き込み可能な 2 次元配列として要素へアクセスするための view を返します．view を作る時点で，配列が 2 次元か，書き込み可能かは確認されます．一方，各 r(i, j) について i と j が配列の範囲内かは確認されません．上のコードでは r.shape(0) と r.shape(1) を loop の上限にすることで，有効な添字だけを渡しています．なお，sum_array() の ptr[i] も C++ の生ポインタによるアクセスなので，範囲は loop 側で保証しています．

C++ 側のメモリを NumPy 配列として返す場合は，データの寿命にも注意が必要です．次の例では，std::vector<float> のデータをコピーせず，そのメモリを参照する NumPy 配列を返します．

py::array_t<float> get_cpp_data() {
    auto* data = new std::vector<float>(100);
    for (size_t i = 0; i < data->size(); ++i) {
        (*data)[i] = static_cast<float>(i);
    }

    py::capsule owner(data, [](void* ptr) {
        delete static_cast<std::vector<float>*>(ptr);
    });

    return py::array_t<float>(
        {data->size()},
        {sizeof(float)},
        data->data(),
        owner
    );
}

py::array_t<float> のコンストラクタには，shape，strides，データの先頭アドレス，所有者となる Python オブジェクトを渡しています．ここで strides の sizeof(float) は，隣の要素までのバイト数です．

配列の base となる owner に capsule を渡すことで，NumPy 配列が生きている間は capsule も保持されます．配列への参照がなくなると capsule の destructor が呼ばれ，heap 上の std::vector<float> が解放されます．

ローカル変数として作った std::vector<float> の data() をそのまま返してはいけません．関数を抜けると vector が破棄され，NumPy 配列だけが無効なメモリを参照することになります．コピーせずに返す場合は，この例のように元データの寿命を NumPy 配列へ結び付ける必要があります．

チャレンジ

  • sum_array(arr) で 1D NumPy 配列の合計を返す
  • brighten(img) で uint8 の 2D 配列を in-place で明るくする
  • get_cpp_data() で C++ 側の std::vector<float> を NumPy 配列として返す
  • capsule を使って C++ 側メモリの寿命を管理する

Step 9: GIL / GPU 連携の入口

説明

重い C++ 処理中は GIL を解放できます．

GIL は Python オブジェクトを守るためのロックです．C++ 側だけで完結する重い処理中は GIL を解放すると，他の Python スレッドが動きやすくなります．
一方で，GIL を解放している間は Python オブジェクトを触ってはいけません．

コード例

void long_running_task() {
    py::gil_scoped_release release;
    std::this_thread::sleep_for(std::chrono::seconds(1));
}

Python から複数スレッドで呼ぶと，GIL を握りっぱなしのときより並列に動きやすくなります．

ただし，GIL を解放した状態で Python オブジェクトを触ってはいけません．Python callback を呼ぶ必要があるなら，py::gil_scoped_acquire で取り直します．

GPU バッファや DLPack はこの先の応用です．ポインタ，所有権，寿命，GIL の理解ができてから触る方が安全です．

チャレンジ

  • long_running_task() で GIL を解放する
  • Python の ThreadPoolExecutor から 2 回呼び，合計時間が 1 秒台になることを確認する
  • GIL を解放したまま Python callback を呼ばない理由を説明する
  • GPU バッファを扱う場合に，所有権，CUDA context，stream 同期が必要になる理由を整理する

pytest で確認する

python -m pytest -q

Appendix: pytest の例

test/conftest.py 
import pytest


@pytest.fixture(scope="session")
def mymodule():
    return pytest.importorskip("mymodule")


@pytest.fixture
def require_attr():
    def _require_attr(obj, name):
        if not hasattr(obj, name):
            pytest.skip(f"{name} is not implemented yet")
        return getattr(obj, name)

    return _require_attr
test/test_step6.py 
def test_notify_event_calls_python_callback(mymodule, require_attr):
    notify_event = require_attr(mymodule, "notify_event")
    messages = []

    notify_event(messages.append)

    assert messages == ["Event Triggered!"]


def test_map_list_applies_python_callback(mymodule, require_attr):
    map_list = require_attr(mymodule, "map_list")

    assert map_list([1, 2, 3], lambda x: x * x) == [1, 4, 9]


def test_map_list_safe_skips_callback_errors(mymodule, require_attr, capfd):
    map_list_safe = require_attr(mymodule, "map_list_safe")

    def callback(x):
        if x == 2:
            raise ValueError("boom")
        return x * 10

    assert map_list_safe([1, 2, 3], callback) == [10, 30]

    captured = capfd.readouterr()
    output = captured.out + captured.err
    assert "Error applying function" in output
    assert "ValueError: boom" in output
test/test_step8.py 
import pytest


np = pytest.importorskip("numpy")


def test_sum_array_sums_1d_float_array(mymodule, require_attr):
    sum_array = require_attr(mymodule, "sum_array")

    data = np.array([1.5, 2.5, 3.0], dtype=np.float64)

    assert sum_array(data) == pytest.approx(7.0)


def test_brighten_updates_image_in_place_with_saturation(mymodule, require_attr):
    brighten = require_attr(mymodule, "brighten")

    image = np.array([[100, 130], [0, 255]], dtype=np.uint8)
    brighten(image)

    expected = np.array([[200, 255], [0, 255]], dtype=np.uint8)
    np.testing.assert_array_equal(image, expected)


def test_get_cpp_data_returns_float_vector_view(mymodule, require_attr):
    get_cpp_data = require_attr(mymodule, "get_cpp_data")

    data = get_cpp_data()

    assert isinstance(data, np.ndarray)
    assert data.shape == (100,)
    assert data.dtype == np.float32
test/test_step9.py 
import concurrent.futures
import time


def test_long_running_task_releases_gil(mymodule, require_attr):
    long_running_task = require_attr(mymodule, "long_running_task")

    start = time.perf_counter()
    with concurrent.futures.ThreadPoolExecutor(max_workers=2) as executor:
        list(executor.map(lambda _: long_running_task(), range(2)))
    elapsed = time.perf_counter() - start

    assert elapsed < 1.8


def test_fake_gpu_buffer_exposes_pointer_address(mymodule, require_attr):
    FakeGpuBuffer = require_attr(mymodule, "FakeGpuBuffer")
    get_fake_gpu_address = require_attr(mymodule, "get_fake_gpu_address")

    buffer = FakeGpuBuffer(1024)
    address = get_fake_gpu_address(buffer)

    assert isinstance(address, int)
    assert address != 0


def test_fake_gpu_buffer_can_make_capsule(mymodule, require_attr):
    FakeGpuBuffer = require_attr(mymodule, "FakeGpuBuffer")
    make_fake_gpu_capsule = require_attr(mymodule, "make_fake_gpu_capsule")

    buffer = FakeGpuBuffer(1024)
    capsule = make_fake_gpu_capsule(buffer)

    assert type(capsule).__name__ == "PyCapsule"


def test_call_python_after_work_reacquires_gil(mymodule, require_attr):
    call_python_after_work = require_attr(mymodule, "call_python_after_work")
    calls = []

    call_python_after_work(lambda: calls.append("called"))

    assert calls == ["called"]

まとめ

Part 3 では，C++ と Python を実用的につなぐための callback，NumPy，GIL を扱いました．

ここまで来ると，C++ と Python をかなり柔軟に組み合わせられるようになると思いますが，まだまだ色々な機能があります．ぜひ pybind11 のドキュメントを参照し，さらに勉強を進めてもらえたらと思います．
本記事が学習の一助になれば幸いです．

0
0
0

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
Sign upLogin
0
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?