Function task_manager_new

cpworker/src/task.c:321–411

}

static int task_manager_new(task_manager_t *this, Config *config)
{
    this->execution_model = config->execution_model;
    if (this->execution_model == EXECUTION_MODEL_PIPELINE)
    {
        this->ring = spsc_ring_create(1024 * 1024);
        if (!this->ring)
        {
            log_fatal("create spsc_ring_t failed");
            return -1;
        }

        this->alloc = simple_allocator_create(config->pipeline.buffer_size_mb * 1024 * 1024);
        if (!this->alloc)
        {
            log_fatal("create simple_allocator_t failed");
            spsc_ring_destroy(this->ring);
            return -1;
        }
    }
    else
    {
        this->ring = NULL;
        this->alloc = NULL;
    }

    this->tasks = malloc(sizeof(struct TaskList));
    if (!this->tasks)
    {
        log_fatal("malloc failed");
        if (this->ring)
            spsc_ring_destroy(this->ring);
        if (this->alloc)
            simple_allocator_destroy(this->alloc);
        return -1;
    }
    this->errors = malloc(sizeof(struct TaskErrorList));
    if (!this->errors)
    {
        free(this->tasks);
        if (this->ring)
            spsc_ring_destroy(this->ring);
        if (this->alloc)
            simple_allocator_destroy(this->alloc);
        log_fatal("malloc failed");
        return -1;
    }

    this->config = config;
    TAILQ_INIT(this->tasks);
    TAILQ_INIT(this->errors);

    pthread_mutex_init(&this->stats_lock, NULL);

    int num_tasks = config->tasks_cfg->num_tasks;
    log_info("find %d tasks", num_tasks);

    int inited_count = 0;