C++ call to PythonThreaded call

[x3lif]

Hello everyone,

I've want to extend an application in order let the customer create it's own extensions through Python. I choose to use PyBind11 for this.

The code is working as long the python interpreter is called from the same thread, thus creating a bottleneck really impacting the performance of the application. So i'm actually trying to reduce this penalty by enqueing the tasks in multiple threads on python side.

here is the following illustration of the architecture i want to achieve :

This seems to work fine when the python function called through pybind11 doing simple stuff (allocating np.array, print strings, ...) , but not anymore after any calls like a REST call (using requests module or httpx).

Using the parallel stack debug view from my favourite IDE, i can observe that the process is stucked into the method take_gil from CPython.

As far i understand, my attempt is different from everything i could have read on the internet around using the py::gil_scoped_release because i'm not calling my C++ from Python, and pybind11 needs to get the GIL in order to wrap informations to python.

Before diving into code details, my questions are the following :

• Can this work ?
• If no : If no what are the options ?
• How to find what is locking the GIL from inside python ?
• ...

Here is a minimal example of the freezing code, the python code, and the C++ that requires only boost and pybind as dependecies. Note that the function call_callback is defined by C++ code. When i uncomment call_callback and comment the other lines into the worker_thread function the GIL doesn't behave as a deadlock

AsyncCustomTasks.py

import configure_venv_modules
import random
import threading

#import httpx
import numpy as np
import requests

def enqueueTask(message: str, cmdId: int) -> int:
    def worker_thread():
        print(f"---------------------- Python : Worker thread for {message} and cmdId {cmdId}")
        #url = 'http://dskgre1073l:9001/v1/models/resnet50/metadata'
        url = 'https://www.example.org/'
        print(f"---------------------- Python : ready to call httpx on {url}")
        #response = httpx.get(url)
        
        mat = np.zeros((512,512), np.float32)
        print(f"---------------------- Python : Successfully created np.array")
        
        response = requests.get(url)
        print(f"---------------------- {response.text}")   # This line never appears on std::out
        
        # print(f"---------------------- Python : try to call httpx.get")
        # response = httpx.get(url)
        #response = httpx.post(url, json=data)
        #print(f"---------------------- {response}")
        
        call_callback(cmdId)



    print(f"---------------------- Let's thread the processing of {message}")
    wt = threading.Thread(target=worker_thread)
    wt.start()
    worker_thread.wt = wt
    print(f"---------------------- {message} threaded")

and here is the standalone cpp file :

#include <iostream>
#include <sstream>
#include <filesystem>
#include <thread>
#include <mutex>
#include <queue>
#include <condition_variable>
#include <memory>
#include <pybind11/functional.h>
#include <pybind11/pybind11.h>
#include <pybind11/embed.h>
#include <chrono>
#include <thread>
#include <future>
#include <map>

namespace py = pybind11;

std::filesystem::path _python_script;
std::string _python_function_name;

struct PythonTask {
    PythonTask(std::string message) : _message(message) {}
    std::string _message;
    std::promise<int> _res;

    void set_result(int res)
    {
        std::cout << "C++ : set result for " << _message << " res = " << res << std::endl;
        _res.set_value(res);
    };
};


struct TaskQueue {
    std::map<int, std::shared_ptr<PythonTask>> cmdQueue;
    std::queue<int> unfinishedTasks;
    std::atomic<int> _cmdId = 0;
    std::mutex mtx;
    std::condition_variable cv;

    TaskQueue() {}

    int addTask(std::string message) {
        std::unique_lock<std::mutex> lock(mtx);
        int cmdId = ++_cmdId;
        cmdQueue[cmdId] = std::make_shared<PythonTask>(message);
        unfinishedTasks.push(cmdId);
        lock.unlock();
        cv.notify_one();
        return cmdId;
    }

    int WaitTask() {
        std::unique_lock<std::mutex> lock(mtx);
        cv.wait(lock, [this] { return !unfinishedTasks.empty(); });
        int cmdId = unfinishedTasks.front();
        unfinishedTasks.pop();
        lock.unlock();
        return cmdId;
    }

    std::shared_ptr<PythonTask> getTask(int cmdId) {
        std::unique_lock<std::mutex> lock(mtx);
        return cmdQueue[cmdId];
    }

    static void callback(int cmdId) {
        std::cout << "C++ : received callback for " << cmdId << std::endl;
    }
};

static TaskQueue adcCmdQueue;

void adc_callback_static(int cmdId)
{
    adcCmdQueue.callback(cmdId);
}

void call_python_function(std::string message, int cmdId)
{
    try
    {
        py::dict globals = py::globals();
        auto module_name = _python_script.stem().string();
        auto module_path = _python_script.parent_path().string();

        std::stringstream ss;
        ss << "import sys" << std::endl;
        ss << "module_path = r\'" << _python_script.parent_path().string() << "\'" << std::endl;
        ss << "if module_path not in sys.path:" << std::endl;
        ss << "    sys.path.append(module_path) " << std::endl;
        ss << "module_path = r\'" << _python_script.string() << "\'" << std::endl;
        ss << "if module_path not in sys.path:" << std::endl;
        ss << "    sys.path.append(module_path) " << std::endl;

        auto command = ss.str();
        pybind11::exec(command, globals, globals);

        // import module and call function
        pybind11::module pyModule = pybind11::module::import(module_name.c_str());
        if (!pybind11::hasattr(pyModule, "call_callback"))
        {
            pyModule.def("call_callback", &adc_callback_static, "c++ callback function");
        }
        auto result = pyModule.attr("enqueueTask")(message, cmdId);

        std::cout << "C++ : python function called successfully." << std::endl;
    }
    catch (const py::error_already_set& e)
    {
        std::cerr << "!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!" << std::endl;
        std::cerr << "C++ : Caught the follwing error : " << std::endl;
        std::cerr << "C++ : "<< e.what() << std::endl;
    }
    catch (...)
    {
        std::cerr << "!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!" << std::endl;
        std::cerr << "unknown exception caught" << std::endl;
    }
}

void thread_queue_execution() {
    py::scoped_interpreter guard{};
    while (true) {
        int cmdId = adcCmdQueue.WaitTask();
        std::shared_ptr<PythonTask> inOut = adcCmdQueue.getTask(cmdId);
        std::cout << "C++ : let's process " << inOut->_message << std::endl;
        call_python_function(inOut->_message, cmdId);
    }
}

void enqueTaskToQueue(std::string message) {
    int cmdId = adcCmdQueue.addTask(message);
}

int main(int argc, const char* argv[])
{
    namespace fs = std::filesystem;
    _python_script = fs::path(R"(./AsyncCustomTasks.py)");
    _python_function_name = "enqueueTask";

    try
    {
        //std::vector<std::string> messages{ "m1" , "m2", "m3" };
        std::vector<std::string> messages{ "m1" };

        std::thread t(thread_queue_execution);

        std::vector<std::thread> componentInspectionThreads;
        for (const auto& message : messages) {
            componentInspectionThreads.emplace_back(enqueTaskToQueue, message);
        }

        while (true) {
            std::this_thread::sleep_for(std::chrono::seconds(11));
            std::cout << "C++ : waiting ..." << std::endl;
        }

        std::cout << "C++ : Execution complete." << std::endl;
    }
    catch (const std::exception& e)
    {
        std::cerr << "C++ : Caught exception : " << e.what() << std::endl;
        return -1;
    }
    catch (...)
    {
        std::cerr << "C++ : An exception has been caught"<<std::endl;
        return -1;
    }
}

Thank you for giving me any clue, tips or advices