Function application_job_run_attempt

src/daemon/application.c:1055–1131

}

static application_attempt_status_t application_job_run_attempt(cbm_daemon_application_job_t *job,
                                                                const char *marker_path,
                                                                const char *quarantine_path,
                                                                application_attempt_t *attempt) {
    application_attempt_init(attempt);
    cbm_daemon_application_t *application = job->application;
    if (application_job_cancel_requested(job)) {
        return APPLICATION_ATTEMPT_CANCELLED;
    }

    cbm_daemon_application_worker_t worker = NULL;
    application_tmp_lock();
    int start_result = application->worker_ops.start(
        application->worker_ops.context, job->args_json, application->worker_memory_budget_bytes,
        marker_path, quarantine_path, &worker);
    application_tmp_unlock();
    if (start_result != 0 || !worker) {
        return application_job_cancel_requested(job) ? APPLICATION_ATTEMPT_CANCELLED
                                                     : APPLICATION_ATTEMPT_START_FAILED;
    }

    cbm_mutex_lock(&application->mutex);
    job->worker = worker;
    bool cancel_now = job->cancel_requested || application->stopping;
    cbm_mutex_unlock(&application->mutex);
    if (cancel_now) {
        /* The worker thread owns this handle until destroy below. Invoke the
         * external supervisor without the application mutex held. */
        (void)application->worker_ops.cancel(application->worker_ops.context, worker);
    }

    const cbm_index_worker_result_t *borrowed = NULL;
    for (;;) {
        cbm_index_worker_poll_t state =
            application->worker_ops.poll(application->worker_ops.context, worker, &borrowed);
        if (state == CBM_INDEX_WORKER_POLL_TERMINAL) {
            break;
        }
        cbm_mutex_lock(&application->mutex);
        bool cancel_pending =
            (job->cancel_requested || application->stopping) && job->worker == worker;
        cbm_mutex_unlock(&application->mutex);
        if (cancel_pending || state == CBM_INDEX_WORKER_POLL_ERROR) {
            (void)application->worker_ops.cancel(application->worker_ops.context, worker);
        }
        cbm_usleep(APPLICATION_JOB_POLL_US);
    }

    if (borrowed) {
        attempt->result = *borrowed;
        attempt->result.response = borrowed->response ? cbm_strdup(borrowed->response) : NULL;
        attempt->has_result = true;
    }
    const char *worker_log =
        application->worker_ops.log_path(application->worker_ops.context, worker);
    if (worker_log) {
        (void)snprintf(attempt->log_path, sizeof(attempt->log_path), "%s", worker_log);
    }