3

私のプロジェクトでは、pybind11を使用して C++ コードを Python にバインドします。最近、私は非常に大きなデータ セット (70 GB 以上) を処理する必要があり、データを 1 つのデータからstd::deque複数std::dequeのデータ セットに分割する必要が生じました。私のデータセットは非常に大きいため、分割によるメモリ オーバーヘッドはそれほど大きくないと予想されます。したがって、私は 1 ポップ - 1 プッシュ戦略を採用しました。これにより、一般的に、私の要件が確実に満たされるはずです。

それはすべて理論上です。実際には、私のプロセスは殺されました。そのため、過去2日間苦労し、最終的に問題を示す最小限の例を思いつきました。

一般に、最小限の例ではdeque(~11GB) に大量のデータを作成し、それを Python に返し、再度 を呼び出しC++て要素を移動します。そのような単純な。移動部分はエグゼキュータで行います。

興味深いことに、エグゼキュータを使用しない場合、メモリ使用量は期待どおりであり、ulimit による仮想メモリの制限が課されている場合でも、プログラムはこれらの制限を実際に尊重し、クラッシュしません。

test.py

from test import _test
import asyncio
import concurrent

async def test_main(loop, executor):
    numbers = _test.generate()
    # moved_numbers = _test.move(numbers) # This works!
    moved_numbers = await loop.run_in_executor(executor, _test.move, numbers) # This doesn't!

if __name__ == '__main__':
    loop = asyncio.get_event_loop()
    executor = concurrent.futures.ThreadPoolExecutor(1)

    task = loop.create_task(test_main(loop, executor))
    loop.run_until_complete(task)

    executor.shutdown()
    loop.close()

test.cpp

#include <deque>
#include <iostream>
#include <pybind11/pybind11.h>
#include <pybind11/stl.h>

namespace py = pybind11;

PYBIND11_MAKE_OPAQUE(std::deque<uint64_t>);
PYBIND11_DECLARE_HOLDER_TYPE(T, std::shared_ptr<T>);

template<class T>
void py_bind_opaque_deque(py::module& m, const char* type_name) {
    py::class_<std::deque<T>, std::shared_ptr<std::deque<T>>>(m, type_name)
    .def(py::init<>())
    .def(py::init<size_t, T>());
}

PYBIND11_PLUGIN(_test) {
    namespace py = pybind11;
    pybind11::module m("_test");
    py_bind_opaque_deque<uint64_t>(m, "NumbersDequeue");

    // Generate ~11Gb of data.
    m.def("generate", []() {
        std::deque<uint64_t> numbers;
        for (uint64_t i = 0; i < 1500 * 1000000; ++i) {
            numbers.push_back(i);
        }
        return numbers;
    });

    // Move data from one dequeue to another.
    m.def("move", [](std::deque<uint64_t>& numbers) {
        std::deque<uint64_t> numbers_moved;

        while (!numbers.empty()) {
            numbers_moved.push_back(std::move(numbers.back()));
            numbers.pop_back();
        }
        std::cout << "Done!\n";
        return numbers_moved;
    });

    return m.ptr();
}

test/__init__.py

import warnings
warnings.simplefilter("default")

コンパイル:

g++ -std=c++14 -O2 -march=native -fPIC -Iextern/pybind11 `python3.5-config --includes` `python3.5-config --ldflags` `python3.5-config --libs` -shared -o test/_test.so test.cpp

所見:

  • 移動部分がエグゼキュータによって行われないため、 を呼び出すだけで、すべてが期待どおりに機能し、moved_numbers = _test.move(numbers)htop によって示されるメモリ使用量は約のままです。11Gb
  • エグゼキュータで部分の移動を行うと、プログラムが 2 倍のメモリを消費してクラッシュします。

  • 仮想メモリの制限 (~15Gb) が導入されると、すべてが正常に機能します。これはおそらく最も興味深い部分です。

    ulimit -Sv 15000000 && python3.5 test.py>> Done!.

  • 制限を増やすと、プログラムがクラッシュします (150Gb > 私の RAM)。

    ulimit -Sv 150000000 && python3.5 test.py>>[1] 2573 killed python3.5 test.py

  • deque メソッドの使用はshrink_to_fit役に立ちません (また、そうすべきではありません)

使用ソフトウェア

Ubuntu 14.04
gcc version 5.4.1 20160904 (Ubuntu 5.4.1-2ubuntu1~14.04)
Python 3.5.2
pybind11 latest release - v1.8.1

ノート

この例は、単に問題を示すために作成されたものであることに注意してください。問題が発生するためには 、asyncioとの使用が必要です。pybind

何が起こっているのかについてのアイデアは大歓迎です。

4

1 に答える 1

1

この問題は、データが 1 つのスレッドで作成され、別のスレッドで割り当て解除されることが原因であることが判明しました。これは、glibc の malloc アリーナが原因です(参照については、これを参照してください)。次のようにすると、うまく実証できます。

executor1 = concurrent.futures.ThreadPoolExecutor(1)
executor2 = concurrent.futures.ThreadPoolExecutor(1)

numbers = await loop.run_in_executor(executor1, _test.generate)
moved_numbers = await loop.run_in_executor(executor2, _test.move, numbers)

_test.generateこれは、 andによって割り当てられたメモリの 2 倍を必要とします。

executor = concurrent.futures.ThreadPoolExecutor(1)

numbers = await loop.run_in_executor(executor, _test.generate)
moved_numbers = await loop.run_in_executor(executor, _test.move, numbers)

傷つかない。

この問題は、要素をあるコンテナから別のコンテナに移動しないようにコードを書き直すか (私の場合)、またはexport MALLOC_ARENA_MAX=1malloc アリーナの数を 1 に制限する環境変数を設定することで解決できます。 (複数のアリーナを持つには十分な理由があります)。

于 2016-10-16T09:41:51.113 に答える