/*

    Task Spooler - a task queue system for the unix user

    Copyright (C) 2007-2009  Lluís Batlle i Rossell

    Please find the license in the provided COPYING file.

*/

#include <stdlib.h>

#include <unistd.h>

#include <stdio.h>

#include <string.h>

#include <sys/time.h>

#include <time.h>

#include "main.h"

/* The list will access them */

int busy_slots = 0;

int max_slots = 1;

struct Notify

{

    int socket;

    int jobid;

    struct Notify *next;

};

/* Globals */

static struct Job *firstjob = 0;

static struct Job *first_finished_job = 0;

static int jobids = 0;

/* This is used for dependencies from jobs

 * already out of the queue */

static int last_errorlevel = 0; /* Before the first job, let's consider

                                   a good previous result */

/* We need this to handle well "-d" after a "-nf" run */

static int last_finished_jobid;

static struct Notify *first_notify = 0;

int max_jobs;

static struct Job * get_job(int jobid);

void notify_errorlevel(struct Job *p);

static void send_list_line(int s, const char * str)

{

    struct msg m;

    /* Message */

    m.type = LIST_LINE;

    m.u.size = strlen(str) + 1;

    send_msg(s, &m);

    /* Send the line */

    send_bytes(s, str, m.u.size);

}

static void send_urgent_ok(int s)

{

    struct msg m;

    /* Message */

    m.type = URGENT_OK;

    send_msg(s, &m);

}

static void send_swap_jobs_ok(int s)

{

    struct msg m;

    /* Message */

    m.type = SWAP_JOBS_OK;

    send_msg(s, &m);

}

static struct Job * find_previous_job(const struct Job *final)

{

    struct Job *p;

    /* Show Queued or Running jobs */

    p = firstjob;

    while(p != 0)

    {

        if (p->next == final)

            return p;

        p = p->next;

    }

    return 0;

}

static struct Job * findjob(int jobid)

{

    struct Job *p;

    /* Show Queued or Running jobs */

    p = firstjob;

    while(p != 0)

    {

        if (p->jobid == jobid)

            return p;

        p = p->next;

    }

    return 0;

}

static struct Job * findjob_holding_client()

{

    struct Job *p;

    /* Show Queued or Running jobs */

    p = firstjob;

    while(p != 0)

    {

        if (p->state == HOLDING_CLIENT)

            return p;

        p = p->next;

    }

    return 0;

}

static struct Job * find_finished_job(int jobid)

{

    struct Job *p;

    /* Show Queued or Running jobs */

    p = first_finished_job;

    while(p != 0)

    {

        if (p->jobid == jobid)

            return p;

        p = p->next;

    }

    return 0;

}

static int count_not_finished_jobs()

{

    int count=0;

    struct Job *p;

    /* Show Queued or Running jobs */

    p = firstjob;

    while(p != 0)

    {

        ++count;

        p = p->next;

    }

    return count;

}

static void add_notify_errorlevel_to(struct Job *job, int jobid)

{

    int *p;

    int newsize = (job->notify_errorlevel_to_size + 1)

        * sizeof(int);

    p = (int *) realloc(job->notify_errorlevel_to,

            newsize);

    if (p == 0)

        error("Cannot allocate more memory for notify_errorlist_to for jobid %i,"

                " having already %i elements",

                job->jobid, job->notify_errorlevel_to_size);

    job->notify_errorlevel_to = p;

    job->notify_errorlevel_to_size += 1;

    job->notify_errorlevel_to[job->notify_errorlevel_to_size - 1] = jobid;

}

void s_mark_job_running(int jobid)

{

    struct Job *p;

    p = findjob(jobid);

    if (!p)

        error("Cannot mark the jobid %i RUNNING.", jobid);

    p->state = RUNNING;

}

/* -1 means nothing awaken, otherwise returns the jobid awaken */

int wake_hold_client()

{

    struct Job *p;

    p = findjob_holding_client();

    if (p)

    {

        p->state = QUEUED;

        return p->jobid;

    }

    return -1;

}

const char * jstate2string(enum Jobstate s)

{

    const char * jobstate;

    switch(s)

    {

        case QUEUED:

            jobstate = "queued";

            break;

        case RUNNING:

            jobstate = "running";

            break;

        case FINISHED:

            jobstate = "finished";

            break;

        case SKIPPED:

            jobstate = "skipped";

            break;

        case HOLDING_CLIENT:

            jobstate = "skipped";

            break;

    }

    return jobstate;

}

void s_list(int s)

{

    struct Job *p;

    char *buffer;

    /* Times:   0.00/0.00/0.00 - 4+4+4+2 = 14*/ 

    buffer = joblist_headers();

    send_list_line(s,buffer);

    free(buffer);

    /* Show Queued or Running jobs */

    p = firstjob;

    while(p != 0)

    {

        if (p->state != HOLDING_CLIENT)

        {

            buffer = joblist_line(p);

            send_list_line(s,buffer);

            free(buffer);

        }

        p = p->next;

    }

    p = first_finished_job;

    /* Show Finished jobs */

    while(p != 0)

    {

        buffer = joblist_line(p);

        send_list_line(s,buffer);

        free(buffer);

        p = p->next;

    }

}

static struct Job * newjobptr()

{

    struct Job *p;

    if (firstjob == 0)

    {

        firstjob = (struct Job *) malloc(sizeof(*firstjob));

        firstjob->next = 0;

        firstjob->output_filename = 0;

        firstjob->command = 0;

        return firstjob;

    }

    p = firstjob;

    while(p->next != 0)

        p = p->next;

    p->next = (struct Job *) malloc(sizeof(*p));

    p->next->next = 0;

    p->next->output_filename = 0;

    p->next->command = 0;

    return p->next;

}

/* Returns -1 if no last job id found */

static int find_last_jobid_in_queue(int neglect_jobid)

{

    struct Job *p;

    int last_jobid = -1;

    p = firstjob;

    while(p != 0)

    {

        if (p->jobid != neglect_jobid &&

            p->jobid > last_jobid)

            last_jobid = p->jobid;

        p = p->next;

    }

    return last_jobid;

}

/* Returns -1 if no last job id found */

static int find_last_stored_jobid_finished()

{

    struct Job *p;

    int last_jobid = -1;

    p = first_finished_job;

    while(p != 0)

    {

        if (p->jobid > last_jobid)

            last_jobid = p->jobid;

        p = p->next;

    }

    return last_jobid;

}

/* Returns job id or -1 on error */

int s_newjob(int s, struct msg *m)

{

    struct Job *p;

    int res;

    p = newjobptr();

    p->jobid = jobids++;

    if (count_not_finished_jobs() < max_jobs)

        p->state = QUEUED;

    else

        p->state = HOLDING_CLIENT;

    p->store_output = m->u.newjob.store_output;

    p->should_keep_finished = m->u.newjob.should_keep_finished;

    p->notify_errorlevel_to = 0;

    p->notify_errorlevel_to_size = 0;

    p->do_depend = m->u.newjob.do_depend;

    p->depend_on = -1; /* By default. May be overriden in the next conditions */

    if (m->u.newjob.do_depend == 1)

    {

        /* Depend on the last queued job. */

        /* As we already have 'p' in the queue,

         * neglect it during the find_last_jobid_in_queue() */

        if (m->u.newjob.depend_on == -1)

        {

            p->depend_on = find_last_jobid_in_queue(p->jobid);

            /* We don't trust the last jobid in the queue (running or queued)

             * if it's not the last added job. In that case, let

             * the next control flow handle it as if it could not

             * do_depend on any still queued job. */

            if (last_finished_jobid > p->depend_on)

                p->depend_on = -1;

            /* If it's queued still without result, let it know

             * its result to p when it finishes. */

            if (p->depend_on != -1)

            {

                struct Job *depended_job;

                depended_job = findjob(p->depend_on);

                if (depended_job != 0)

                    add_notify_errorlevel_to(depended_job, p->jobid);

                else

                    warning("The jobid %i is queued to do_depend on the jobid %i"

                        " suddenly non existant in the queue", p->jobid,

                        p->depend_on);

            }

            else /* Otherwise take the finished job, or the last_errorlevel */

            {

                if (m->u.newjob.depend_on == -1)

                {

                    int ljobid = find_last_stored_jobid_finished();

                    /* If we have a newer result stored, use it */

                    if (last_finished_jobid < ljobid)

                    {

                        struct Job *parent;

                        parent = find_finished_job(ljobid);

                        if (!parent)

                            error("jobid %i suddenly disappeared from the finished list",

                                ljobid);

                        p->dependency_errorlevel = parent->result.errorlevel;

                    }

                    else

                        p->dependency_errorlevel = last_errorlevel;

                }

            }

        }

        else

        {

            /* The user decided what's the job this new job depends on */

            struct Job *depended_job;

            p->depend_on = m->u.newjob.depend_on;

            depended_job = findjob(p->depend_on);

            if (depended_job != 0)

                add_notify_errorlevel_to(depended_job, p->jobid);

            else

            {

                struct Job *parent;

                parent = find_finished_job(p->depend_on);

                if (parent)

                {

                    p->dependency_errorlevel = parent->result.errorlevel;

                }

                else

                {

                    /* We consider as if the job not found

                       didn't finish well */

                    p->dependency_errorlevel = -1;

                }

            }

        }

    }

    pinfo_init(&p->info);

    pinfo_set_enqueue_time(&p->info);

    /* load the command */

    p->command = malloc(m->u.newjob.command_size);

    if (p->command == 0)

        error("Cannot allocate memory in s_newjob command_size (%i)",

                m->u.newjob.command_size);

    res = recv_bytes(s, p->command, m->u.newjob.command_size);

    if (res == -1)

        error("wrong bytes received");

    /* load the label */

    p->label = 0;

    if (m->u.newjob.label_size > 0)

    {

        char *ptr;

        ptr = (char *) malloc(m->u.newjob.label_size);

        if (ptr == 0)

            error("Cannot allocate memory in s_newjob env_size(%i)",

                    m->u.newjob.env_size);

        res = recv_bytes(s, ptr, m->u.newjob.label_size);

        if (res == -1)

            error("wrong bytes received");

        p->label = ptr;

    }

    /* load the info */

    if (m->u.newjob.env_size > 0)

    {

        char *ptr;

        ptr = (char *) malloc(m->u.newjob.env_size);

        if (ptr == 0)

            error("Cannot allocate memory in s_newjob env_size(%i)",

                    m->u.newjob.env_size);

        res = recv_bytes(s, ptr, m->u.newjob.env_size);

        if (res == -1)

            error("wrong bytes received");

        pinfo_addinfo(&p->info, m->u.newjob.env_size+100,

                "Environment:\n%s", ptr);

        free(ptr);

    }

    return p->jobid;

}

/* This assumes the jobid exists */

void s_removejob(int jobid)

{

    struct Job *p;

    struct Job *newnext;

    if (firstjob->jobid == jobid)

    {

        struct Job *newfirst;

        /* First job is to be removed */

        newfirst = firstjob->next;

        free(firstjob->command);

        free(firstjob->output_filename);

        pinfo_free(&firstjob->info);

        free(firstjob->label);

        free(firstjob);

        firstjob = newfirst;

        return;

    }

    p = firstjob;

    /* Not first job */

    while (p->next != 0)

    {

        if (p->next->jobid == jobid)

            break;

        p = p->next;

    }

    if (p->next == 0)

        error("Job to be removed not found. jobid=%i", jobid);

    newnext = p->next->next;