Method CallInternal

src/plugin/python_plugin.cpp:151–243

Run by the rest handling threads. Pushes the task to Python and wait for its finish.

// Run by the rest handling threads. Pushes the task to Python and wait for its finish.
python_plugin::TaskOutput::ErrorCode PythonPluginManagerImpl::CallInternal(
    const std::string& user, const std::string& function, const std::string& input, double timeout,
    bool in_process, bool read_only, std::string& output) {
    // check timeout
    if (timeout <= 0) timeout = (double)3600 * 24 * 365;
    std::unique_ptr<PythonWorkerProcess> proc(nullptr);
    AutoCleanupAction rollback(nullptr);
    {
        // pick a free process, or create a new one
        std::lock_guard<std::mutex> l(_mtx);
        if (_free_processes.empty()) {
            // no free process, create a new one
            LOG_DEBUG() << "Creating a new Python process";
            proc.reset(new PythonWorkerProcess(db_dir_));
        } else {
            proc.swap(_free_processes.front());
            _free_processes.pop_front();
            if (!proc->IsAlive()) proc.reset(new PythonWorkerProcess(db_dir_));
        }
        PythonWorkerProcess* ptr = proc.get();
        _busy_processes.insert(ptr);
        // on failure, proc is killed and should be removed from busy processes
        rollback.Reset([ptr, this]() {
            std::lock_guard<std::mutex> l(_mtx);
            _busy_processes.erase(ptr);
        });
    }

    proc->ClearOutput();
    // write task
    python_plugin::TaskInput task_input;
    task_input.user = user;
    task_input.graph = graph_name_;
    task_input.plugin_dir = plugin_dir_;
    task_input.function = function;
    task_input.input = input;
    task_input.read_only = read_only;
    task_input.WriteToMessageQueue(proc->GetSendMQ());
    // wait for output
    python_plugin::TaskOutput task_output;
    double start_time = fma_common::GetTime();
    python_plugin::TaskOutput::ErrorCode ec;
    while (true) {
        // if master thread is killing this process
        if (TaskTracker::GetInstance().ShouldKillCurrentTask() || proc->ShouldKill()) {
            proc->Kill();
            std::lock_guard<std::mutex> l(_mtx);
            _marked_processes.insert(std::move(proc));
            THROW_CODE(TaskKilled);
        }
        // if process failed, break
        if (!proc->IsAlive()) {
            output = proc->Stderr();
            THROW_CODE(InternalError, "Plugin failed unexpectly. Stderr:\n{}", output);
        }
        bool r = task_output.ReadFromMessageQueue(proc->GetRecvMQ(), 1000);
        if (r) {
            // done with the task