Method execute

engine/src/execution_graph/executor.cpp:217–255

}

void executor::execute(){
    while(shutdown == 0 && exception_holder.empty()){
        //consider using get_all and calling in a loop.
        auto cur_task = this->task_queue.pop_or_wait();
        pool.push([cur_task{std::move(cur_task)}, this](int thread_id){
            std::size_t memory_needed = cur_task->task_memory_needed();

            // Here we want to wait until we make sure we have enough memory to operate, or if there are no tasks currently running, then we want to go ahead and run
            std::unique_lock<std::mutex> lock(memory_safety_mutex);
            memory_safety_cv.wait(lock, [this, memory_needed] { 
                if (memory_needed < (processing_memory_limit - resource->get_memory_used())){
                    return true;
                } else if (active_tasks_counter.load() == 0){
                    std::shared_ptr<spdlog::logger> logger = spdlog::get("batch_logger");
                    if (logger){
                        logger->warn("|||{info}|||||",
                                "info"_a="WARNING: launching task even though over limit, because there are no tasks running. Memory used: {}"_format(std::to_string(resource->get_memory_used())));
                    }
                    return true;
                } else {
                    return false;
                }
                });

            active_tasks_counter++;

            try {
                cur_task->run(this->streams[thread_id], this);
            } catch(...) {
                std::unique_lock<std::mutex> lock(exception_holder_mutex);
                exception_holder.push(std::current_exception());
                cur_task->fail();
            }

            active_tasks_counter--;
            memory_safety_cv.notify_all();
        });
    }
}

std::exception_ptr executor::last_exception(){
    std::unique_lock<std::mutex> lock(exception_holder_mutex);