cannot pickle 'PyCapsule' object error, when using pybind11 function and dask

huangapple
go评论75阅读模式
英文:

cannot pickle 'PyCapsule' object error, when using pybind11 function and dask

问题

以下是您提供的代码的中文翻译部分：

setup.py 部分：

  1. import sys
  2. from pybind11 import get_cmake_dir
  3. from pybind11.setup_helpers import Pybind11Extension, build_ext
  4. from setuptools import setup
  6. __version__ = "0.0.1"
  8. ext_modules = [
  9.     Pybind11Extension("python_example",
  10.         ["src/main.cpp"],
  11.         define_macros=[('VERSION_INFO', __version__)],
  12.     ),
  13. ]
  15. setup(
  16.     name="python_example",
  17.     version=__version__,
  18.     author="Sylvain Corlay",
  19.     author_email="sylvain.corlay@gmail.com",
  20.     url="https://github.com/pybind/python_example",
  21.     description="A test project using pybind11",
  22.     long_description="",
  23.     ext_modules=ext_modules,
  24.     extras_require={"test": "pytest"},
  25.     cmdclass={"build_ext": build_ext},
  26.     zip_safe=False,
  27.     python_requires=">=3.7",
  28. )

src/main.cpp 部分：

  1. #include <pybind11/pybind11.h>
  3. #define STRINGIFY(x) #x
  4. #define MACRO_STRINGIFY(x) STRINGIFY(x)
  6. int add(int i, int j) {
  7.     return i + j;
  8. }
  10. namespace py = pybind11;
  12. PYBIND11_MODULE(python_example, m) {
  13.     m.doc() = R"pbdoc(
  14.         Pybind11 example plugin
  15.         -----------------------
  17.         .. currentmodule:: python_example
  19.         .. autosummary::
  20.         :toctree: _generate
  22.         add
  23.         subtract
  24.     )pbdoc";
  26.     m.def("add", &add, R"pbdoc(
  27.         Add two numbers
  29.         Some other explanation about the add function.
  30.     )pbdoc");
  32.     m.def("subtract", [](int i, int j) { return i - j; }, R"pbdoc(
  33.         Subtract two numbers
  35.         Some other explanation about the subtract function.
  36.     )pbdoc";
  38. #ifdef VERSION_INFO
  39.     m.attr("__version__") = MACRO_STRINGIFY(VERSION_INFO);
  40. #else
  41.     m.attr("__version__") = "dev";
  42. #endif
  43. }

example.py 部分：

  1. import numpy as np
  2. import pandas as pd
  3. import dask.dataframe as dd
  4. from dask.diagnostics import ProgressBar
  5. from python_example import add
  7. def python_add(i: int, j: int) -> int:
  8.     return i + j
  10. def add_column_values_python(row: pd.Series) -> pd.Series:
  11.     row['sum'] = python_add(row['i'], row['j'])
  13. def add_column_values(row: pd.Series) -> pd.Series:
  14.     row['sum'] = add(int(row['i']), int(row['j']))
  16. def main():
  17.     dataframe = pd.read_csv('./example.csv', index_col=[])
  18.     dataframe['sum'] = np.nan
  20.     with ProgressBar():
  21.         d_dataframe = dd.from_pandas(dataframe, npartitions=16)
  22.         dataframe = d_dataframe.map_partitions(
  23.             lambda df: df.apply(add_column_values_python, axis=1)).compute(scheduler='processes')
  25.     with ProgressBar():
  26.         d_dataframe = dd.from_pandas(dataframe, npartitions=16)
  27.         dataframe = d_dataframe.map_partitions(
  28.             lambda df: df.apply(add_column_values, axis=1), meta=pd.Series(dtype='float64')).compute(scheduler='processes')
  30. if __name__ == '__main__':
  31.     main()

以上是代码的翻译，如果您有关于代码运行时出现的错误的问题，可以提供更多信息，以便我提供帮助。

英文:

See the following example extracted from the pybind11 help (https://github.com/pybind/python_example):

The setup.py is:

  1. import sys
  3. # Available at setup time due to pyproject.toml
  4. from pybind11 import get_cmake_dir
  5. from pybind11.setup_helpers import Pybind11Extension, build_ext
  6. from setuptools import setup
  8. __version__ = &quot;0.0.1&quot;
  10. # The main interface is through Pybind11Extension.
  11. # * You can add cxx_std=11/14/17, and then build_ext can be removed.
  12. # * You can set include_pybind11=false to add the include directory yourself,
  13. #   say from a submodule.
  14. #
  15. # Note:
  16. #   Sort input source files if you glob sources to ensure bit-for-bit
  17. #   reproducible builds (https://github.com/pybind/python_example/pull/53)
  19. ext_modules = [
  20.     Pybind11Extension(&quot;python_example&quot;,
  21.         [&quot;src/main.cpp&quot;],
  22.         # Example: passing in the version to the compiled code
  23.         define_macros = [(&#39;VERSION_INFO&#39;, __version__)],
  24.         ),
  25. ]
  27. setup(
  28.     name=&quot;python_example&quot;,
  29.     version=__version__,
  30.     author=&quot;Sylvain Corlay&quot;,
  31.     author_email=&quot;sylvain.corlay@gmail.com&quot;,
  32.     url=&quot;https://github.com/pybind/python_example&quot;,
  33.     description=&quot;A test project using pybind11&quot;,
  34.     long_description=&quot;&quot;,
  35.     ext_modules=ext_modules,
  36.     extras_require={&quot;test&quot;: &quot;pytest&quot;},
  37.     # Currently, build_ext only provides an optional &quot;highest supported C++
  38.     # level&quot; feature, but in the future it may provide more features.
  39.     cmdclass={&quot;build_ext&quot;: build_ext},
  40.     zip_safe=False,
  41.     python_requires=&quot;&gt;=3.7&quot;,
  42. )

The Cpp part is (src/main.cpp):

  1. #include &lt;pybind11/pybind11.h&gt;
  3. #define STRINGIFY(x) #x
  4. #define MACRO_STRINGIFY(x) STRINGIFY(x)
  6. int add(int i, int j) {
  7.     return i + j;
  8. }
  10. namespace py = pybind11;
  12. PYBIND11_MODULE(python_example, m) {
  13.     m.doc() = R&quot;pbdoc(
  14.         Pybind11 example plugin
  15.         -----------------------
  17.         .. currentmodule:: python_example
  19.         .. autosummary::
  20.         :toctree: _generate
  22.         add
  23.         subtract
  24.     )pbdoc&quot;;
  26.     m.def(&quot;add&quot;, &amp;add, R&quot;pbdoc(
  27.         Add two numbers
  29.         Some other explanation about the add function.
  30.     )pbdoc&quot;);
  32.     m.def(&quot;subtract&quot;, [](int i, int j) { return i - j; }, R&quot;pbdoc(
  33.         Subtract two numbers
  35.         Some other explanation about the subtract function.
  36.     )pbdoc&quot;);
  38. #ifdef VERSION_INFO
  39.     m.attr(&quot;__version__&quot;) = MACRO_STRINGIFY(VERSION_INFO);
  40. #else
  41.     m.attr(&quot;__version__&quot;) = &quot;dev&quot;;
  42. #endif
  43. }

And the python code that I want to run is this (example.py):

  1. import numpy as np
  2. import pandas as pd
  3. import dask.dataframe as dd
  4. from dask.diagnostics import ProgressBar
  5. from python_example import add
  8. def python_add(i: int, j: int) -&gt; int:
  9.     return i + j
  12. def add_column_values_python(row: pd.Series) -&gt; pd.Series:
  13.     row[&#39;sum&#39;] = python_add(row[&#39;i&#39;], row[&#39;j&#39;])
  16. def add_column_values(row: pd.Series) -&gt; pd.Series:
  17.     row[&#39;sum&#39;] = add(int(row[&#39;i&#39;]), int(row[&#39;j&#39;]))
  20. def main():
  21.     dataframe = pd.read_csv(&#39;./example.csv&#39;, index_col=[])
  22.     dataframe[&#39;sum&#39;] = np.nan
  24.     with ProgressBar():
  25.         d_dataframe = dd.from_pandas(dataframe, npartitions=16)
  26.         dataframe = d_dataframe.map_partitions(
  27.             lambda df: df.apply(add_column_values_python, axis=1)).compute(scheduler=&#39;processes&#39;)
  29.     with ProgressBar():
  30.         d_dataframe = dd.from_pandas(dataframe, npartitions=16)
  31.         dataframe = d_dataframe.map_partitions(
  32.             lambda df: df.apply(add_column_values, axis=1), meta=pd.Series(dtype=&#39;float64&#39;)).compute(scheduler=&#39;processes&#39;)
  35. if __name__ == &#39;__main__&#39;:
  36.     main()

And the example.csv file looks like this:

  1. i,j
  2. 1,2
  3. 3,4
  4. 5,6
  5. 7,8
  6. 9,10

But when I run this code I get the following error when using the C++ add version:

  1. [########################################] | 100% Completed | 1.24 ss
  2. [                                        ] | 0% Completed | 104.05 ms
  3. Traceback (most recent call last):
  4.   File &quot;/Users/user/local/src/python_example/example.py&quot;, line 38, in &lt;module&gt;
  5.     main()
  6.   File &quot;/Users/user/local/src/python_example/example.py&quot;, line 33, in main
  7.     dataframe = d_dataframe.map_partitions(
  8.   File &quot;/Users/user/local/src/python_example/.venv/lib/python3.9/site-packages/dask/base.py&quot;, line 314, in compute
  9.     (result,) = compute(self, traverse=False, **kwargs)
  10.   File &quot;/Users/user/local/src/python_example/.venv/lib/python3.9/site-packages/dask/base.py&quot;, line 599, in compute
  11.     results = schedule(dsk, keys, **kwargs)
  12.   File &quot;/Users/user/local/src/python_example/.venv/lib/python3.9/site-packages/dask/multiprocessing.py&quot;, line 233, in get
  13.     result = get_async(
  14.   File &quot;/Users/user/local/src/python_example/.venv/lib/python3.9/site-packages/dask/local.py&quot;, line 499, in get_async
  15.     fire_tasks(chunksize)
  16.   File &quot;/Users/user/local/src/python_example/.venv/lib/python3.9/site-packages/dask/local.py&quot;, line 481, in fire_tasks
  17.     dumps((dsk[key], data)),
  18.   File &quot;/Users/user/local/src/python_example/.venv/lib/python3.9/site-packages/cloudpickle/cloudpickle_fast.py&quot;, line 73, in dumps
  19.     cp.dump(obj)
  20.   File &quot;/Users/user/local/src/python_example/.venv/lib/python3.9/site-packages/cloudpickle/cloudpickle_fast.py&quot;, line 632, in dump
  21.     return Pickler.dump(self, obj)
  22. TypeError: cannot pickle &#39;PyCapsule&#39; object

Is there a way to solve that, maybe by defining something in the C++ module definition?

Note that this example is only to illustrate the problem.

答案1

得分: 2

由于mdurant的帮助，我终于让它正常工作了，以下是更新后的代码：

现在的main.cpp看起来像这样：

  1. #include <pybind11/pybind11.h>
  3. #define STRINGIFY(x) #x
  4. #define MACRO_STRINGIFY(x) STRINGIFY(x)
  6. int add(int i, int j) {
  7.     return i + j;
  8. }
  10. class Add {
  11. public:
  12.     Add() {};
  13.     int add(int i, int j) {
  14.         return i + j;
  15.     }
  16. };
  18. namespace py = pybind11;
  20. PYBIND11_MODULE(python_example, m) {
  21.     m.doc() = R"pbdoc(
  22.         Pybind11 example plugin
  23.         -----------------------
  25.         .. currentmodule:: python_example
  27.         .. autosummary::
  28.         :toctree: _generate
  30.         add
  31.         subtract
  32.     )pbdoc";
  34.     m.def("add", &add, R"pbdoc(
  35.         Add two numbers
  37.         Some other explanation about the add function.
  38.     )pbdoc");
  40.     m.def("subtract", [](int i, int j) { return i - j; }, R"pbdoc(
  41.         Subtract two numbers
  43.         Some other explanation about the subtract function.
  44.     )pbdoc");
  46. #ifdef VERSION_INFO
  47.     m.attr("__version__") = MACRO_STRINGIFY(VERSION_INFO);
  48. #else
  49.     m.attr("__version__") = "dev";
  50. #endif
  52.     py::class_<Add>(m, "Add")
  53.         .def(py::init<>())
  54.         .def("__call__", &Add::add)
  55.         .def("__getstate__", [](const Add &p) {
  56.             /* 返回完全编码对象状态的元组 */
  57.             return py::make_tuple();
  58.         })
  59.         .def("__setstate__", [](Add &p, py::tuple t) {
  60.             if (t.size() != 0)
  61.                 throw std::runtime_error("无效状态!");
  63.             /* 调用就地构造函数。请注意，即使对象只有一个平凡的默认构造函数，这也是必要的 */
  64.             new (&p) Add();
  66.             /* 分配任何其他状态 */
  67.         });
  68. }

而example.py文件如下：

  1. import numpy as np
  2. import pandas as pd
  3. import dask.dataframe as dd
  4. from dask.diagnostics import ProgressBar
  5. from python_example import Add
  7. def python_add(i: int, j: int) -> int:
  8.     return i + j
  10. def add_column_values_python(row: pd.Series) -> pd.Series:
  11.     row['sum'] = python_add(row['i'], row['j'])
  12.     return row
  14. def add_column_values(row: pd.Series) -> pd.Series:
  15.     row['sum'] = Add()(int(row['i']), int(row['j']))
  16.     return row
  18. def add_column_values_import(row: pd.Series) -> pd.Series:
  19.     from python_example import add
  20.     row['sum'] = add(int(row['i']), int(row['j']))
  21.     return row
  23. def main():
  24.     dataframe = pd.read_csv('./example.csv', index_col=[])
  25.     dataframe['sum'] = np.nan
  27.     with ProgressBar():
  28.         d_dataframe = dd.from_pandas(dataframe, npartitions=16)
  29.         dataframe = d_dataframe.map_partitions(
  30.             lambda df: df.apply(add_column_values_python, axis=1)).compute(scheduler='processes')
  32.     with ProgressBar():
  33.         d_dataframe = dd.from_pandas(dataframe, npartitions=16)
  34.         dataframe = d_dataframe.map_partitions(
  35.             lambda df: df.apply(add_column_values, axis=1)).compute(scheduler='processes')
  37.     with ProgressBar():
  38.         d_dataframe = dd.from_pandas(dataframe, npartitions=16)
  39.         dataframe = d_dataframe.map_partitions(
  40.             lambda df: df.apply(add_column_values_import, axis=1)).compute(scheduler='processes')
  42. if __name__ == '__main__':
  43.     main()

这个想法是将函数放入一个类中，然后定义__getstate__和__setstate__python魔术函数，或者在函数内导入。

了解更多信息：https://pybind11-jagerman.readthedocs.io/en/stable/advanced.html

英文:

Thanks to mdurant I finally got it working, here is the updated code:

Now the main.cpp look like this:

  1. #include &lt;pybind11/pybind11.h&gt;
  3. #define STRINGIFY(x) #x
  4. #define MACRO_STRINGIFY(x) STRINGIFY(x)
  6. int add(int i, int j) {
  7.     return i + j;
  8. }
  10. class Add {
  11. public:
  12.     Add() {};
  13.     int add(int i, int j) {
  14.         return i + j;
  15.     }
  16. };
  18. namespace py = pybind11;
  20. PYBIND11_MODULE(python_example, m) {
  21.     m.doc() = R&quot;pbdoc(
  22.         Pybind11 example plugin
  23.         -----------------------
  25.         .. currentmodule:: python_example
  27.         .. autosummary::
  28.         :toctree: _generate
  30.         add
  31.         subtract
  32.     )pbdoc&quot;;
  34.     m.def(&quot;add&quot;, &amp;add, R&quot;pbdoc(
  35.         Add two numbers
  37.         Some other explanation about the add function.
  38.     )pbdoc&quot;);
  40.     m.def(&quot;subtract&quot;, [](int i, int j) { return i - j; }, R&quot;pbdoc(
  41.         Subtract two numbers
  43.         Some other explanation about the subtract function.
  44.     )pbdoc&quot;);
  46. #ifdef VERSION_INFO
  47.     m.attr(&quot;__version__&quot;) = MACRO_STRINGIFY(VERSION_INFO);
  48. #else
  49.     m.attr(&quot;__version__&quot;) = &quot;dev&quot;;
  50. #endif
  52.     py::class_&lt;Add&gt;(m, &quot;Add&quot;)
  53.         .def(py::init&lt;&gt;())
  54.         .def(&quot;__call__&quot;, &amp;Add::add)
  55.         .def(&quot;__getstate__&quot;, [](const Add &amp;p) {
  56.             /* Return a tuple that fully encodes the state of the object */
  57.             return py::make_tuple();
  58.         })
  59.         .def(&quot;__setstate__&quot;, [](Add &amp;p, py::tuple t) {
  60.             if (t.size() != 0)
  61.                 throw std::runtime_error(&quot;Invalid state!&quot;);
  63.             /* Invoke the in-place constructor. Note that this is needed even
  64.             when the object just has a trivial default constructor */
  65.             new (&amp;p) Add();
  67.             /* Assign any additional state */
  68.         });
  69. }

And the example.py file looks like this:

  1. import numpy as np
  2. import pandas as pd
  3. import dask.dataframe as dd
  4. from dask.diagnostics import ProgressBar
  5. from python_example import Add
  8. def python_add(i: int, j: int) -&gt; int:
  9.     return i + j
  12. def add_column_values_python(row: pd.Series) -&gt; pd.Series:
  13.     row[&#39;sum&#39;] = python_add(row[&#39;i&#39;], row[&#39;j&#39;])
  14.     return row
  17. def add_column_values(row: pd.Series) -&gt; pd.Series:
  18.     row[&#39;sum&#39;] = Add()(int(row[&#39;i&#39;]), int(row[&#39;j&#39;]))
  19.     return row
  21. def add_column_values_import(row: pd.Series) -&gt; pd.Series:
  22.     from python_example import add
  23.     row[&#39;sum&#39;] = add(int(row[&#39;i&#39;]), int(row[&#39;j&#39;]))
  24.     return row
  27. def main():
  28.     dataframe = pd.read_csv(&#39;./example.csv&#39;, index_col=[])
  29.     dataframe[&#39;sum&#39;] = np.nan
  31.     with ProgressBar():
  32.         d_dataframe = dd.from_pandas(dataframe, npartitions=16)
  33.         dataframe = d_dataframe.map_partitions(
  34.             lambda df: df.apply(add_column_values_python, axis=1)).compute(scheduler=&#39;processes&#39;)
  36.     with ProgressBar():
  37.         d_dataframe = dd.from_pandas(dataframe, npartitions=16)
  38.         dataframe = d_dataframe.map_partitions(
  39.             lambda df: df.apply(add_column_values, axis=1)).compute(scheduler=&#39;processes&#39;)
  41.     with ProgressBar():
  42.         d_dataframe = dd.from_pandas(dataframe, npartitions=16)
  43.         dataframe = d_dataframe.map_partitions(
  44.             lambda df: df.apply(add_column_values_import, axis=1)).compute(scheduler=&#39;processes&#39;)
  47. if __name__ == &#39;__main__&#39;:
  48.     main()

The idea is to put the functions inside a class and then define the __getstate__ and __setstate__ python magic functions or the import inside the function.

For more information:
https://pybind11-jagerman.readthedocs.io/en/stable/advanced.html

答案2

得分: 1

General

如果你想在Python中将一个对象从一个进程传递到另一个进程（使用Dask或不使用Dask），你需要一种序列化的方式。默认的方法是使用"pickle"。C库中的对象基本上是基于动态指针的东西，pickle不知道该如何处理它们。你可以通过提供getstate/setstate或reduce双下划线方法来为你的C对象实现pickle协议。

另外，Dask有一个序列化层，你可以为特定类注册特定的序列化和反序列化函数，但这仅适用于分布式调度器，不适用于多进程（前者在各方面都更好，没有好的理由使用多进程）。

Specific

还有一些更简单的选项：

  • 使用线程调度器，这样就不需要序列化（C代码应该释放GIL并获得完全的并行性）。
  • 我认为只有add函数是问题所在，可能只需将你的导入移到add_column_values函数中，这样每个工作进程都会得到自己的副本，而不是从闭包中传递它。
英文:

General

If you want to pass an object from one process to another in python (with or without dask), you need a way to serialise it. The default method for this is "pickle". Objects in C libraries are fundamentally dynamic pointer-based things and pickle doesn't know what to do with them. You can implement the pickle protocol for your C object by providing getstate/setstate or reduce dunder methods.

Alternatively, dask has a layer of serialisation where you can register specific ser/de functions for specific classes, but that is only with the distributed scheduler, not multiprocessing (the former is better in every way, there is no good reason you should be using multiprocessing).

Specific

A couple of simpler options:

  • use the threading scheduler, so that no serialisation is needed (C code ought to release the GIL and get full parallelism)
  • I think it's only the add function that is the problem, it's probably enough to move your import into the add_column_values function, so that each worker gets its own copy instead of passing it from the closure.

huangapple
  • 本文由 发表于 2023年2月14日 21:36:37
  • 转载请务必保留本文链接：https://go.coder-hub.com/75448632.html
  • dask
  • pandas
  • pybind11
匿名

发表评论

匿名网友

:?: :razz: :sad: :evil: :!: :smile: :oops: :grin: :eek: :shock: :???: :cool: :lol: :mad: :twisted: :roll: :wink: :idea: :arrow: :neutral: :cry: :mrgreen:

确定