主页 / 问题 / 1038896
Accepted
Jens
Asked:2020-10-26 21:15:57 +0000 UTC2020-10-26 21:15:57 +0000 UTC

通过 Boost.Python 并行执行 Python 函数

  • 772

数据部分：

static Market market;

// boost::asio::thread_pool thread_pool;

static std::unordered_map < std::string, std::unordered_map < std::string,
    std::pair < Market::path_t, std::string > > > data; // future

Python设置：

static Python_Initializer python_initializer;

static boost::python::object global = boost::python::import("__main__").attr("__dict__");

boost::python::exec("from analysis.src.main import run", global, global);

static boost::python::object module = global["run"];

在Python_Initializer被称为Py_Initialize。static这里和数据部分需要避免重新构造对象，因为 此代码放置在从外部循环调用的库的 DLL 函数中。该函数run是线程安全的。

主要执行：

for (const auto & asset : market.assets())
{
    for (const auto & scale : market.scales())
    {
        auto file = market.get(asset, scale, 250);

        //std::packaged_task < std::string() > task([asset, scale, file, &module]()
        //{
        //    return boost::python::extract < std::string > (
        //      module(asset.c_str(), scale.c_str(), file.string().c_str()));
        //});

        //data[asset][scale] = std::make_pair(std::move(file),
        //  boost::asio::post(thread_pool, std::move(task)));

        std::string message = boost::python::extract < std::string >(
            module(asset.c_str(), scale.c_str(), file.string().c_str()));

        data[asset][scale] = std::make_pair(std::move(file), std::move(message));
    }
}

在当前的串行执行中，一切正常。但是，如果添加thread_pool，将后者包裹在std::string中data，std::future并将主执行中的代码替换为当前注释掉的代码，则会发生错误。它的内容没有说什么具体的，错误是在Boost.Python级别的，如果你跟踪它，那么在设置某种回调函数时读/写指针有问题。

问题：如何配置Boost.Python，让Boost的并行执行机制和标准库可以并行执行一些线程安全的Python函数？

c++

1 个回答

  1. Best Answer

    一般的想法是为 python 解释器使用第二个线程。

    #include <Python.h>

#include <boost/python/exec.hpp>
#include <iostream>
#include <thread>
#include <sys/wait.h>

#include <boost/python/extract.hpp>
#include <boost/python/import.hpp>
#include <boost/python/object.hpp>

void python (std::string fork, int m, int n) {
  const char *prog = "def ack(m, n):\n"
                     "  if m == 0:\n"
                     "    return n + 1\n"
                     "  elif n == 0:\n"
                     "    return ack(m - 1, 1)\n"
                     "  else:\n"
                     "    return ack(m - 1, ack(m, n - 1))";

  Py_Initialize();
  try {
    std::cout << fork << std::endl;

    boost::python::object mainModule = boost::python::import("__main__");
    boost::python::object mainNamespace = mainModule.attr("__dict__");

    std::stringstream commandstream;
    commandstream << "ack(" << m << "," << n << ")";
    std::string command = commandstream.str();
    boost::python::exec(prog, mainNamespace, mainNamespace);
    int val = boost::python::extract<int>(boost::python::eval(command.c_str(), mainNamespace, mainNamespace));
    std::cout << fork << " result: " << val << std::endl;
  } catch (boost::python::error_already_set const &e) {
    PyErr_Print();
  }
}

int main (int argc, char const *argv[]) {
  pid_t pid = fork();
  if (pid == 0) {
    python("f1", 3, 4);
  } else if (pid > 0) {
    python("f2", 3, 3);

    int status;
    waitpid(pid, &status, 0);
  } else {
    std::cout << "Fork failed." << std::endl;
  }

  return 0;
}

    相关链接：：

    • 1

相关问题