#!/usr/bin/perl
use strict;
my($app_svnid) = '$HeadURL$ $LastChangedRevision$';
#  Allow bare words, so &ade_err_error() calls look nicer.
no strict 'subs';
use lib substr(`ade-config ade_lib_prefix`,0,-1) . '/include';
use ADE;
use lib substr(`js-config js_lib_prefix`,0,-1) . '/include';
use JS;
#  see perlipc(1) for explanation of why this is needed
use POSIX ":sys_wait_h";
use Data::Dumper;
no warnings 'experimental::smartmatch';
use IPC::SysV qw(IPC_PRIVATE IPC_RMID IPC_CREAT S_IRUSR S_IWUSR);
#use Storable;

my(%jsd_defined_errors) = (
    jsd_err_misc         => { fmt => "%s" },
);

#  Options
my(%opt_resources);
my($opt_schedule_flag);
my($dbinit_file) = substr(`js-config js_lib_prefix`,0,-1) . '/include/jsd.sql';

#  Other globals
my(%exit_code_of_recently_reaped_pids, $js_shell);
my($sqlite_max_int) = 9223372036854775807;

sub jsd
{
    my($errstack_ref) = @_;
    my($rc, $db_file, $pid, $dbh);
    my($sql_statement);
    my($resource_id, $i);
    my($quit_flag);
    my(@possible_job_ids, @command_ids);
    my($possible_jobs_count);
    my($lock_file, %message, $name);
    my($server_handle, $server_name, $server_address);

    #  Set ADE options
    &ade_err_registerdefderrs(\%jsd_defined_errors);

    #  Defaults for options 
    #  (identical line in jsd, jss, jsq).
    $opt_schedule_flag = 1;

    #  Register options
    if (($rc=&ade_opt_register($errstack_ref, "r:s", "resources:s,suspend", "main::jsd_opt_handler_%s")) != $ade_err_ok) {
        return($rc);
    }

    #  Register handler functions
    if (($rc=&ade_msg_register($errstack_ref, \&jsd_usage, \&jsd_version, \&jsd_listpaths)) != $ade_err_ok) {
        return($rc);
    } 

    #  Process options
    if (($rc=&ade_opt_process($errstack_ref)) != $ade_err_ok) {
        return($rc);
    } 

    #  Process arguments
    &ade_msg_usage($errstack_ref, \&app_usage, 1) if ($#ARGV+1 != 0);

    #  Sanity checks and derivations
    &JS::get_server_name($errstack_ref, \$server_name);
    $lock_file = "$ade_tmp_dir/jsd-$server_name.lock";
    $db_file = "$ade_tmp_dir/jsd-$server_name.sqlite";
    if (scalar(keys(%opt_resources)) == 0) {
        &ade_err_error($errstack_ref, jsd_err_misc, "no resources specified");
        return($ade_err_fail);
    }
    foreach $resource_id (keys(%opt_resources)) {
       if ($resource_id !~ /^[a-z][a-z0-9_]*$/) {
           &ade_err_error($errstack_ref, jsd_err_misc, "$resource_id: invalid resource name (hint: it should be word-like)");
           return($ade_err_fail);
       }
       &ade_err_debug($errstack_ref, 10, "jsd: resource_id=$resource_id, quantity_total=$opt_resources{$resource_id}{quantity_total}, affinous=$opt_resources{$resource_id}{affinous}");
       if (!defined($opt_resources{$resource_id}{quantity_total}) or $opt_resources{$resource_id}{quantity_total} !~ /^(?:0|[1-9][0-9]*)$/) {
           &ade_err_error($errstack_ref, jsd_err_misc, (defined($opt_resources{$resource_id}{quantity_total}) ? $opt_resources{$resource_id}{quantity_total} : "") . ": invalid quantity (hint: should be an integer)");
           return($ade_err_fail);
       }
       if (!defined($opt_resources{$resource_id}{affinous})) {
           $opt_resources{$resource_id}{affinous} = 'false';
       } elsif ($opt_resources{$resource_id}{affinous} ~~ ['t', 'true', '1', 'yes', 'y']) {
           $opt_resources{$resource_id}{affinous} = 'true';
       } elsif ($opt_resources{$resource_id}{affinous} ~~ ['f', 'false', '0', 'no', 'n']) {
           $opt_resources{$resource_id}{affinous} = 'false';
       } else {
           &ade_err_error($errstack_ref, jsd_err_misc, "$opt_resources{$resource_id}{affinous}: invalid affinousness (hint: should be empty or a logical)");
           return($ade_err_fail);
       }
       if ($opt_resources{$resource_id}{quantity_total} eq '0') {
           #  Don't allow infinite resources (indicated by quanity zero) to be affinous because
           #  the list of resource pool sets will be so big (i.e. 2^$sqlite_max_int). If we used
           #  a value smaller than $sqlite_max_int to be infinity, then perhaps we could allow it.
           if ($opt_resources{$resource_id}{affinous} eq 'true') {
               &ade_err_error($errstack_ref, jsd_err_misc, "$opt_resources{$resource_id}: infinite resources cannot be affinous");
               return($ade_err_fail);
           }
           &ade_err_debug($errstack_ref, 10, "jsd: resource_id=$resource_id, quantity is zero; making infinite ...");
           $opt_resources{$resource_id}{quantity_total} = sprintf("%d", $sqlite_max_int);
       }
    }
    #  Set $js_shell
    if (defined($ENV{'JS_SHELL'}) and (-x $ENV{'JS_SHELL'})) {
        $js_shell = $ENV{'JS_SHELL'};
    } elsif (defined($ENV{'SHELL'}) and (-x $ENV{'SHELL'})) {
        $js_shell = $ENV{'SHELL'};
    } else {
        $js_shell = '/bin/sh';
    }

    #  Lock
    if (($rc=&ade_lck_lock_simple($errstack_ref, $lock_file)) != $ade_err_ok) {
        return($rc);
    }

    #  Define and initialise database
    if (-f $db_file) {
        &ade_err_warning($errstack_ref, jsd_err_misc, "$db_file: removing old database ...");
        unlink($db_file);
    }
    &ade_tmp_registerfile($errstack_ref, $db_file);
    &ade_err_debug($errstack_ref, 10, "jsd: db_file=$db_file");
    #  Pass one arg to system() in order that it uses the shell, which is needed to use redirection, 
    #  which is needed to get sqlite not to output stuff about loading ~/.sqliterc.
    if (system("sqlite3 $db_file \".read $dbinit_file\" < /dev/null") != 0) {
        &ade_err_error($errstack_ref, jsd_err_misc, "failed to initialise database");
        return($ade_err_fail);
    }
    #  Now connect to to the database for real work.
    if (!defined($dbh = DBI->connect("dbi:SQLite:dbname=$db_file","",""))) {
        &ade_err_error($errstack_ref, jsd_err_misc, $DBI::errstr);
        return($ade_err_fail);
    }
    $dbh->do("PRAGMA foreign_keys = ON");
    $dbh->sqlite_busy_timeout(10000);

    #  Populate resources table
    &ade_err_debug($errstack_ref, 10, "jsd: populating resources table ...");
    foreach $resource_id (keys %opt_resources) {
        $sql_statement = "INSERT INTO resources (resource_id, quantity_total, affinous) VALUES ('$resource_id', $opt_resources{$resource_id}{quantity_total}, '$opt_resources{$resource_id}{affinous}');";
        if (($rc=&ade_sql_execute($errstack_ref, $dbh, $sql_statement)) != $ade_err_ok) {
            return($rc);
        }
    }
    undef %opt_resources;			#  let's be clear; this is no longer needed.
    #  Expand resources into resource pools. We assign arbitrary priorities to the pools. This
    #  will be used to help select which of the a resource pool sets that satisfies a job's
    #  requirements will be preferred.
    #
    #  Note that we use i+1 instead of i as the resource_pool_priority. If the pools have
    #  priority:
    #
    #      pool#0: 0
    #      pool#1: 1
    #
    #  then pool *set* #3 (consisting of pool#0 *and* pool#1) has the same priority as pool set #2.
    #  More specifically, if a job wants only one CPU, then pool #5 - *which consists of two CPUs*
    #  is equally attractive to pool #4, which consists of one CPU. Which of these sqlite chooses
    #  to show is not certain. For this reason we make the priority i+1.
    #  
    $sql_statement = "INSERT INTO resource_pools (resource_id, resource_pool_id, resource_pool_priority, quantity_total)
                          SELECT resource_id, i AS resource_pool_id, i+1 AS resource_pool_priority, 1 AS quantity_total
                              FROM resources, i1000
                              WHERE affinous = 'true' and i < quantity_total
                          UNION SELECT resource_id, 0 AS resource_pool_id, 0 as resource_pool_priority, quantity_total
                              FROM resources
                              WHERE affinous = 'false';";
    if (($rc=&ade_sql_execute($errstack_ref, $dbh, $sql_statement)) != $ade_err_ok) {
        return($rc);
    }

    #  Initialise the UNIX socket server 
    &JS::init_server($errstack_ref, $server_name, \$server_handle);
    
    #  Main loop
    $SIG{CHLD} = \&sighandler_chld;		#  help reap children 
    $quit_flag = 0;
    while (1) {
        &ade_err_debug($errstack_ref, 5, "jsd: top of loop");

        #  Top priority: release job slots occupied by completed jobs!
        if (($rc=&process_recently_reaped_pids($errstack_ref, $dbh)) != $ade_err_ok) {
            return($rc);
        }

        #  Next priority start as many new jobs as possible but always going back 
        #  and checking if any slots are releasable.
        if ($opt_schedule_flag) {
            &ade_err_debug($errstack_ref, 5, "jsd: checking for runnable jobs ...");
            if (($rc=&get_job_ids($errstack_ref, $dbh, "possible", \@possible_job_ids)) != $ade_err_ok) {
                return($rc);
            }
            $possible_jobs_count = $#possible_job_ids+1;
            &ade_err_debug($errstack_ref, 5, "jsd: $possible_jobs_count jobs are runnable ...");
            if ($possible_jobs_count > 0) {
                &ade_err_debug($errstack_ref, 5, "jsd: starting job $possible_job_ids[0] ...");
                if (($rc=&start_job_id($errstack_ref, $dbh, $possible_job_ids[0])) != $ade_err_ok) {
                    return($rc);
                }
                next;
            }
        }

        #  Lowest priority: deal with incoming messages.
        &ade_err_debug($errstack_ref, 5, "jsd: waiting for a message ...");
        #if (($rc=&read_message($errstack_ref, $js_msgqueue_id, \$sender_id, $js_msgqueue_type_jsd, \%message)) != $ade_err_ok) {
        #    return($rc);
        ##  Return early if read_message() got interrupted.
        #} elsif (not %message) {
        #    &ade_err_debug($errstack_ref, 5, "jsd: read_message() returned an empty message; we assume it was interrupted");
        #} else {
        #    &ade_err_debug($errstack_ref, 5, "jsd: read_message() returned a message so now calling process_message() to process it ...");
        #    if (($rc=&process_message($errstack_ref, $dbh, $js_msgqueue_id, $sender_id, \%message, \$quit_flag, \$opt_schedule_flag)) != $ade_err_ok) {
        #        return($rc);
        #    }
        #    #  If the call to process_message() resulting in quit_flag being set then break out of loop.
        #    if ($quit_flag) {
        #        &ade_err_debug($errstack_ref, 5, "jsd: process_message() set quit_flag; breaking out of main loop ...");
        #        last; 
        #    }
        #}
        if (($rc=&JS::read_message_str_and_send_reply_str($errstack_ref, $server_handle, \&process_message, [ $dbh, \$quit_flag, \$opt_schedule_flag ])) != $ade_err_ok) {
            return($rc);
        }
        #  If the call to process_message() resulting in quit_flag being set then break out of loop.
        if ($quit_flag) {
            &ade_err_debug($errstack_ref, 5, "jsd: process_message() set quit_flag; breaking out of main loop ...");
            last; 
        }
    }
    $SIG{CHLD} = 'DEFAULT';

    #  Tidy up
    sleep(1);   #  give jsc time to read the reply we've sent to its instruction to quit.
    #msgctl($js_msgqueue_id, IPC_RMID, 0);
    $dbh->disconnect;
    unlink $db_file;
    &ade_tmp_deregisterfile($errstack_ref, $db_file);
    if (($rc=&ade_lck_unlock_simple($errstack_ref, $lock_file)) != $ade_err_ok) {
        return($rc);
    }
    return($ade_err_ok);
}

sub jsd_opt_handler_r
{
    my($errstack_ref, $resource_specs_string) = @_;
    my($resource_spec, $resource_id, $quantity_total, $affinous);

    foreach $resource_spec (split(/,/, $resource_specs_string)) {
        ($resource_id, $quantity_total, $affinous) = split(/:/, $resource_spec);
        $opt_resources{$resource_id}{quantity_total} = $quantity_total;
        $opt_resources{$resource_id}{affinous} = $affinous;
    }

    return $ade_err_ok;
} 

sub jsd_opt_handler_resources
{
    return(&jsd_opt_handler_r(@_));
}

sub jsd_opt_handler_suspend
{
    my($errstack_ref) = @_;

    $opt_schedule_flag = 0;

    return $ade_err_ok;
}

sub jsd_version
{
    my($errstack_ref, $version_text_ref) = @_;

    return(&ade_smf_extractversionfromsvnstring($errstack_ref, $app_svnid, $version_text_ref));
}

sub jsd_listpaths
{
    my($errstack_ref, $pathlist_text_ref) = @_;
    my($rc);

    #${$pathlist_text_ref} = "";

    return($ade_err_ok);
}

sub jsd_usage
{
    my($errstack_ref, $usage_text_ref, $passno) = @_;

    if ($passno == 1) {
        ${$usage_text_ref} = "\n";
    } elsif ($passno == 2) {
        ${$usage_text_ref} = "                     --suspend               start with scheduler suspended\n" .
                             "         -r ...    | --resources=...         set system resources\n";
    }

    return($ade_err_ok);
}

sub process_recently_reaped_pids
{
    my($errstack_ref, $dbh) = @_;
    my($pid, $rc, $job_id, $sql_statement);

    foreach $pid (keys %exit_code_of_recently_reaped_pids) {
        &ade_err_debug($errstack_ref, 5, "process_recently_reaped_pids: processing recently reaped pid $pid ...");
        #  Translate exited PID to job number.
        if (($rc=&get_job_id_by_pid($errstack_ref, $dbh, $pid, \$job_id)) != $ade_err_ok) {
            return($rc);
        }

        $sql_statement = "BEGIN EXCLUSIVE TRANSACTION;";
        if (($rc=&ade_sql_execute($errstack_ref, $dbh, $sql_statement)) != $ade_err_ok) {
            return($rc);
        }

        # Release the running demands (i.e. delete all entries related to this job from the running_job_resource_pools table).  
        &ade_err_debug($errstack_ref, 10, "process_recently_reaped_pids: $job_id: deleting running resources allocations ...");
        $sql_statement = "delete from running_job_resource_pools where job_id = $job_id;";
        if (($rc=&ade_sql_execute($errstack_ref, $dbh, $sql_statement)) != $ade_err_ok) {
            return($rc);
        }
    
        #  Finally change the state of the job.
        &ade_err_debug($errstack_ref, 10, "process_recently_reaped_pids: $job_id: changing job state to completed ...");
        if (($rc=&set_job_state_pid_exit_code_by_job_id($errstack_ref, $dbh, $job_id, "completed", 0, $exit_code_of_recently_reaped_pids{$pid})) != $ade_err_ok) {
            return($rc);
        }
    
        #  And if we managed to do all of that then we can commit the transaction.
        &ade_err_debug($errstack_ref, 10, "process_recently_reaped_pids: $job_id: committing transaction ...");
        $sql_statement = "END TRANSACTION;";
        if (($rc=&ade_sql_execute($errstack_ref, $dbh, $sql_statement)) != $ade_err_ok) {
            return($rc);
        }
    
        &ade_err_debug($errstack_ref, 10, "process_recently_reaped_pids: $job_id: marking job as completed ...");
        delete $exit_code_of_recently_reaped_pids{$pid};
    }

    return($ade_err_ok);
}

sub start_job_id
{
    my($errstack_ref, $dbh, $job_id) = @_;
    my($pid, $zencoded_arg_list, $user, $rc, $i, $current_resource_value, $demanded_resource_value, @select_results, $resource);
    my($sql_statement, @wanted_resources, @available_resources);
    my($resource_id, $quantity_wanted, $affinous, $tuple_ref);
    my($resource_id2, $resource_pool_id, $quantity_available, $tuple_ref2, $quantity_allocated_so_far, $quantity_taking);
    my($taking_resource, @taking_resources, @new_envs, $new_env_ref, @arg_list);

    #  Get the job's command amd user.
    $sql_statement = "SELECT command, [user] FROM jobs WHERE job_id == '$job_id';";
    if (($rc=&ade_sql_select($errstack_ref, $dbh, \@select_results, $sql_statement)) != $ade_err_ok) {
        return($rc);
    }
    $zencoded_arg_list = ${$select_results[0]}[0];
    $user = ${$select_results[0]}[1];
    &ade_err_debug($errstack_ref, 10, "start_job_id: $job_id: zencoded_arg_list is \"$zencoded_arg_list\", user is \"$user\"");

    #  Determine which resources we'll use
    $sql_statement = "select pending_job_resources.resource_id, pending_job_resources.quantity_wanted, resources.affinous from pending_job_resources, resources where pending_job_resources.resource_id=resources.resource_id and pending_job_resources.job_id = $job_id";
    if (($rc=&ade_sql_select($errstack_ref, $dbh, \@wanted_resources, $sql_statement)) != $ade_err_ok) {
        return($rc);
    }
    $sql_statement = "select resource_id, resource_pool_id, quantity_available from pending_job_preferred_resource_pool_quantities_available where job_id = $job_id";
    if (($rc=&ade_sql_select($errstack_ref, $dbh, \@available_resources, $sql_statement)) != $ade_err_ok) {
        return($rc);
    }
    @taking_resources = ();
    @new_envs = ();
    foreach $tuple_ref (@wanted_resources) {
        $quantity_allocated_so_far = 0;
        ($resource_id, $quantity_wanted, $affinous) = @{$tuple_ref};
        &ade_err_debug($errstack_ref, 10, "start_job_id: resource_id=$resource_id, quantity_wanted=$quantity_wanted, affinous=$affinous");
        foreach $tuple_ref2 (@available_resources) {
            ($resource_id2, $resource_pool_id, $quantity_available) = @{$tuple_ref2};
            next if ($resource_id2 ne $resource_id);
            &ade_err_debug($errstack_ref, 10, "start_job_id: resource_id2=$resource_id2, resource_pool_id=$resource_pool_id, quantity_available=$quantity_available");
            $quantity_taking = ($quantity_available<$quantity_wanted-$quantity_allocated_so_far)?$quantity_available:$quantity_wanted-$quantity_allocated_so_far;
            &ade_err_debug($errstack_ref, 10, "start_job_id: would take $quantity_taking");
            push @taking_resources, { resource_id=>$resource_id, resource_pool_id=>$resource_pool_id, quantity_taking=>$quantity_taking };
            $quantity_allocated_so_far += $quantity_taking;
        }
        if ($affinous eq 'true') {
            #  Since this code is executed inside the outer loop, the first time @taking_resources will contain resources of only 1 type, but on subsequent
            #  loops it will contain also others.
            push @new_envs, { var=>sprintf("JS_%s_AFFINITY", uc($resource_id)), val=>join(',', map { ($_->{resource_id} eq $resource_id) ? $_->{resource_pool_id} : () } @taking_resources) };
        } else {
            push @new_envs, { var=>sprintf("JS_%s_LIMIT", uc($resource_id)), val=>$quantity_wanted };
        }
    }
    push @new_envs, { var=>'JS_JOB_ID', val=>$job_id };
    push @new_envs, { var=>'JS_USER', val=>$user };

    #  Additional environment variables.
    

    #  Start the job
    &ade_err_debug($errstack_ref, 10, "start_job_id: $job_id: forking ...");
    if (($pid=fork) < 0) {
        &ade_err_error($errstack_ref, jsd_err_misc, "fork: failed");
        return($ade_err_fail);
    } elsif ($pid == 0) {
        &ade_err_debug($errstack_ref, 10, "start_job_id: $job_id: child setting up environment and execing command ...");
        foreach $new_env_ref (@new_envs) {
            $ENV{$new_env_ref->{var}} = $new_env_ref->{val};
        }
        &zdecode($errstack_ref, $zencoded_arg_list, \@arg_list);
        exec $js_shell, '-c', @arg_list;
        &ade_err_error($errstack_ref, jsd_err_misc, "failed to exec");
        return($ade_err_fail);
    }
    &ade_err_debug($errstack_ref, 5, "start_job_id: forked pid $pid");

    #  Take the resources
    foreach $taking_resource (@taking_resources) {
        $sql_statement = "insert into running_job_resource_pools values ($job_id, '$taking_resource->{resource_id}', $taking_resource->{resource_pool_id}, $taking_resource->{quantity_taking});";
        if (($rc=&ade_sql_execute($errstack_ref, $dbh, $sql_statement)) != $ade_err_ok) {
            return($rc);
        }
    }
    # Release the pending demands (i.e. delete all entries related to this job from the pending_job_resources table).  
    $sql_statement = "delete from pending_job_resources where job_id = $job_id;";
    if (($rc=&ade_sql_execute($errstack_ref, $dbh, $sql_statement)) != $ade_err_ok) {
        return($rc);
    }

    #  Finally change the state of the job.
    if (($rc=&set_job_state_pid_exit_code_by_job_id($errstack_ref, $dbh, $job_id, "running", $pid, 0)) != $ade_err_ok) {
        return($rc);
    }

    return($ade_err_ok);
}
        
sub sighandler_chld
{
    my($sig) = @_;
    my($pid, @errstack_local, $rc, $job_id);

    @errstack_local = ();

    #  Reap and note exit codes. The main loop will update the database table with this info.
    while (($pid = waitpid(-1,WNOHANG)) > 0) {
        $exit_code_of_recently_reaped_pids{$pid} = $?;
        &ade_err_debug(\@errstack_local, 5, "======> sighandler_chld: noting $pid exited with $exit_code_of_recently_reaped_pids{$pid} ...");
    }
}

sub set_job_state_pid_exit_code_by_job_id
{
    my($errstack_ref, $dbh, $job_id, $job_state, $pid, $exit_code) = @_;
    my($rc, $sql_statement);

    &ade_err_debug($errstack_ref, 10, "set_job_state_pid_exit_code_by_job_id: sof (job_id=$job_id, job_state=$job_state, pid=$pid)");
    $sql_statement = "UPDATE jobs SET job_state='$job_state', pid=$pid, exit_code=$exit_code WHERE job_id == '$job_id';";
    if (($rc=&ade_sql_execute($errstack_ref, $dbh, $sql_statement)) != $ade_err_ok) {
        return($rc);
    }
    
    return($ade_err_ok);
}

sub get_job_id_by_pid
{
    my($errstack_ref, $dbh, $pid, $job_id_ref) = @_;
    my($rc, @select_results, $sql_statement);

    $sql_statement = "SELECT job_id FROM jobs WHERE pid == '$pid';";
    if (($rc=&ade_sql_select($errstack_ref, $dbh, \@select_results, $sql_statement)) != $ade_err_ok) {
        return($rc);
    }

    #  Extract the first column out of each row, assemble an array out of that
    #  and write it where to caller told us to write it.
    ${$job_id_ref} = ${$select_results[0]}[0];

    return($ade_err_ok);
}

sub get_current_resource_value_by_name
{
    my($errstack_ref, $dbh, $resource_id, $resource_value_ref) = @_;
    my(@select_results, $rc, $sql_statement);

    $sql_statement = "SELECT current FROM resources WHERE resource_id == '$resource_id';";
    if (($rc=&ade_sql_select($errstack_ref, $dbh, \@select_results, $sql_statement)) != $ade_err_ok) {
        return($rc);
    }

    ${$resource_value_ref} = ${$select_results[0]}[0];
    return($ade_err_ok);
}

sub get_demanded_resource_value_by_job_id
{
    my($errstack_ref, $dbh, $resource_id, $job_id, $demanded_resource_value_ref) = @_;
    my(@select_results, $rc, $sql_statement);

    &ade_err_debug($errstack_ref, 10, "get_demanded_resource_value_by_job_id: sof (resource_id=$resource_id, job_id=$job_id)");

    $sql_statement = "SELECT $resource_id FROM jobs WHERE job_id == '$job_id';";
    if (($rc=&ade_sql_select($errstack_ref, $dbh, \@select_results, $sql_statement)) != $ade_err_ok) {
        return($rc);
    }

    ${$demanded_resource_value_ref} = ${$select_results[0]}[0];
    return($ade_err_ok);
}

sub dump_tables
{
    my($errstack_ref, $level, $dbh) = @_;
    my($table, $sql_command, $sth, @row, $select_result, @select_results, $rc, $sql_statement);

    foreach $table (qw(resources jobs)) {
        &ade_err_debug($errstack_ref, $level, "dump_tables: name=$table");
        undef @select_results;
        $sql_statement = "SELECT * FROM $table;";
        if (($rc=&ade_sql_select($errstack_ref, $dbh, \@select_results, $sql_statement)) != $ade_err_ok) {
            return($rc);
        }
        foreach $select_result (@select_results) {
           &ade_err_debug($errstack_ref, $level, sprintf("dump_tables: " . scalar("%-10s" x ($#{$select_result}+1)), @{$select_result}));
        }
    }
}

sub get_job_ids
{
    my($errstack_ref, $dbh, $job_state, $job_ids_arrayref) = @_;
    my($rc, @select_results, $sql_statement);

    if ($job_state eq "pending" || $job_state eq "running" || $job_state eq "completed") {
        $sql_statement = "SELECT job_id FROM jobs WHERE job_state == '$job_state';";
        if (($rc=&ade_sql_select($errstack_ref, $dbh, \@select_results, $sql_statement)) != $ade_err_ok) {
            return($rc);
        }
        #  Extract the first column out of each row, assemble an array out of that
        #  and write it where to caller told us to write it.
        @{$job_ids_arrayref} = map { ${$_}[0] } @select_results;

    } elsif ($job_state eq "possible") {
        #  find the IDs of jobs whose count of *met* resource requirements are *equal*
        #  to the number of resource types on the system. (E.g. a job with *unmet* RAM
        #  requirement and *met* cores requirement is not eligble to run ('cos 1!=2).

        #sqlite> select * from next_runnable_job;
        #job_id    
        #----------
        #1         

        $sql_statement = "SELECT job_id FROM runnable_jobs;";
        if (($rc=&ade_sql_select($errstack_ref, $dbh, \@select_results, $sql_statement)) != $ade_err_ok) {
            return($rc);
        }
        #  Extract the first column out of each row, assemble an array out of that
        #  and write it where to caller told us to write it.
        @{$job_ids_arrayref} = map { ${$_}[0] } @select_results;

    } else {
       &ade_err_internal($errstack_ref, "get_job_ids: $job_state: invalid job_state");
    }
 
    return($ade_err_ok);
}

sub process_message
{
    my($errstack_ref, $callback_args_ref, $message_hash_ref, $reply_hash_ref) = @_;
    my($dbh, $quit_flag_ref, $schedule_flag_ref, $rc, $show_queue_output, %wanted_resource_counts);

    $dbh               = ${$callback_args_ref}[0];
    $quit_flag_ref     = ${$callback_args_ref}[1];
    $schedule_flag_ref = ${$callback_args_ref}[2];

    &ade_err_debug($errstack_ref, 10, sprintf("process_message: message=[%s]", join(',', map { sprintf("%s=>%s", $_, ${$message_hash_ref}{$_}) } keys(%{$message_hash_ref}))));
    &ade_err_debug($errstack_ref, 10, sprintf("process_message: message has keys [ %s ]", join(',',keys(%{$message_hash_ref}))));

    if (!defined(${$message_hash_ref}{'command'})) {
        ${$reply_hash_ref}{rc} = 1;
        ${$reply_hash_ref}{text} = "message has no command";
        
    } elsif (${$message_hash_ref}{'command'} eq 'quit') {
        ${$quit_flag_ref} = 1;
        ${$reply_hash_ref}{rc} = 0;

    } elsif (${$message_hash_ref}{'command'} eq 'suspend') {
        ${$schedule_flag_ref} = 0;
        ${$reply_hash_ref}{rc} = 0;

    } elsif (${$message_hash_ref}{'command'} eq 'resume') {
        ${$schedule_flag_ref} = 1;
        ${$reply_hash_ref}{rc} = 0;

    } elsif (${$message_hash_ref}{'command'} eq 'ping') {
        ${$reply_hash_ref}{text} = "pong";
        ${$reply_hash_ref}{rc} = 0;

    } elsif (${$message_hash_ref}{'command'} eq 'show-queue') {
        &ade_err_debug($errstack_ref, 10, "process_message: calling show_queue() and saving output to \$show_queue_output ...");
        if (($rc=&show_queue($errstack_ref, $dbh, \$show_queue_output)) != $ade_err_ok) {
            return($rc);
        }
        ${$reply_hash_ref}{text} = $show_queue_output;
        ${$reply_hash_ref}{rc} = 0;

    } elsif (${$message_hash_ref}{'command'} eq 'submit-job') {
        if (!defined(${$message_hash_ref}{'zencoded-arg-list'})) {
            ${$reply_hash_ref}{rc} = 1;
            ${$reply_hash_ref}{text} = "submit-job command has no zencoded-arg-list";
        } elsif (!defined(${$message_hash_ref}{'resource-list'})) {
            ${$reply_hash_ref}{rc} = 1;
            ${$reply_hash_ref}{text} = "submit-job command has no resource-list";
        } else {
            &ade_err_debug($errstack_ref, 10, sprintf("process_message: zencoded argument list is ${$message_hash_ref}{'zencoded-arg-list'}"));
            &ade_err_debug($errstack_ref, 10, "process_message: chopping up wanted resource list ...");
            my($wanted_resource_name, $wanted_resource, $wanted_resource_count);
            foreach $wanted_resource (split(',',${$message_hash_ref}{'resource-list'})) {
                ($wanted_resource_name, $wanted_resource_count) = split(':', $wanted_resource);
                $wanted_resource_counts{$wanted_resource_name} = $wanted_resource_count;
            }
            &ade_err_debug($errstack_ref, 10, sprintf("process_message: wanted resource list is [%s]", join(',', map { sprintf("%s=>%s", $_, $wanted_resource_counts{$_}) } keys(%wanted_resource_counts))));
            if (($rc=&submit_job($errstack_ref, $dbh, ${$message_hash_ref}{'zencoded-arg-list'}, \%wanted_resource_counts, $reply_hash_ref)) != $ade_err_ok) {
                return($rc);
            }
        }

    } else {
        ${$reply_hash_ref}{rc} = 1;
        ${$reply_hash_ref}{text} = "message has unknown command";
    }
 
    return($ade_err_ok);
}

sub zdecode
{
    my($errstack_ref, $zencoded_str, $arg_list_ref) = @_;
    my($arg);

    @{$arg_list_ref} = ();
    foreach $arg (split(' ', $zencoded_str)) {
       &ade_err_debug($errstack_ref, 10, "zdecode: decoding \"$arg\" ...");
       $arg =~ s/Z25/\%/g;
       $arg =~ s/Z5C/\\/g;
       $arg =~ s/Z20/ /g;
       $arg =~ s/Z27/\'/g;
       $arg =~ s/Z22/\"/g;
       $arg =~ s/Z5A/Z/g;
       &ade_err_debug($errstack_ref, 10, "zdecode: decoded to \"$arg\"");
       push @{$arg_list_ref}, $arg;
    }
}

sub submit_job
{
    my($errstack_ref, $dbh, $zencoded_arg_list, $wanted_resource_counts_ref, $reply_hash_ref) = @_;
    my($command, @arguments, $rc, $sql_statement, @select_results, $job_id, $resource_id, $i);

    &ade_err_debug($errstack_ref, 10, "submit_job: zencoded_arg_list=$zencoded_arg_list");

    #  Get the list of resources jsd manages.
    $sql_statement = "SELECT resource_id FROM resources;";
    if (($rc=&ade_sql_select($errstack_ref, $dbh, \@select_results, $sql_statement)) != $ade_err_ok) {
        return($rc);
    }
    #  If that differs from what's been passed then error. This can be done more succintly
    #  with the Array::Diff module but introduces that extra dependency. Instead we check
    #  the count and then, item by item, the sorted lists.
    if ($#select_results+1 != scalar(keys(%{$wanted_resource_counts_ref}))) {
        ${$reply_hash_ref}{rc} = 1;
        ${$reply_hash_ref}{text} = "resource count mismatch";
        return($ade_err_ok);
    }
    for ($i=0; $i<$#select_results+1; $i++) {
        if ((sort(map { $select_results[$_][0] } (0..$#select_results)))[$i] ne (sort(keys(%{$wanted_resource_counts_ref})))[$i]) {
            ${$reply_hash_ref}{rc} = 1;
            ${$reply_hash_ref}{text} = "resource list mismatch";
            return($ade_err_ok);
        }
    }

    #  Let the database decide the job id but do the whole thing in one transaction
    $sql_statement = "begin transaction;";
    if (($rc=&ade_sql_execute($errstack_ref, $dbh, $sql_statement)) != $ade_err_ok) {
        return($rc);
    }
    ##  Escape single quotes in $command else the quoting in the next line doesn't work.
    #$command =~ s/'/''/g;

    $sql_statement = sprintf("INSERT INTO jobs (command, [user]) VALUES('$zencoded_arg_list', '%s');", getpwuid($<));
    if (($rc=&ade_sql_execute($errstack_ref, $dbh, $sql_statement)) != $ade_err_ok) {
        $sql_statement = "rollback transaction;";
        &ade_sql_execute($errstack_ref, $dbh, $sql_statement);
        return($rc);
    }
    $job_id = $dbh->last_insert_id("","","","");
    &ade_err_debug($errstack_ref, 10, "submit_job: job_id=$job_id");

    foreach $resource_id (keys(%{$wanted_resource_counts_ref})) {
    &ade_err_debug($errstack_ref, 10, "submit_job: resource_id=$resource_id");
        $sql_statement = "INSERT INTO pending_job_resources VALUES ($job_id, '$resource_id', ${$wanted_resource_counts_ref}{$resource_id});";
        if (($rc=&ade_sql_execute($errstack_ref, $dbh, $sql_statement)) != $ade_err_ok) {
            $sql_statement = "rollback transaction;";
            &ade_sql_execute($errstack_ref, $dbh, $sql_statement);
            return($rc);
        }
    }

    $sql_statement = "end transaction;";
    if (($rc=&ade_sql_execute($errstack_ref, $dbh, $sql_statement)) != $ade_err_ok) {
        return($rc);
    }

    ${$reply_hash_ref}{rc} = 0;
    ${$reply_hash_ref}{text} = "job submitted with job id $job_id";
    return($ade_err_ok);
}

sub show_queue
{
    my($errstack_ref, $dbh, $reply_ref) = @_;
    my($fh, $sql_statement, $pending_jobs_count, $running_jobs_count, $completed_jobs_count, $select_result, @select_results, $rc);

    my($opt_show_all_flag) = 0;  #  move to function parameters later

    open($fh, '>', $reply_ref) || die;

    $sql_statement = "SELECT count(*) from jobs where job_state=='pending';";
    if (($rc=&ade_sql_select($errstack_ref, $dbh, \@select_results, $sql_statement)) != $ade_err_ok) {
        return($rc);
    }
    $pending_jobs_count = ${$select_results[0]}[0];
    $sql_statement = "SELECT count(*) from jobs where job_state=='running';";
    if (($rc=&ade_sql_select($errstack_ref, $dbh, \@select_results, $sql_statement)) != $ade_err_ok) {
        return($rc);
    }
    $running_jobs_count = ${$select_results[0]}[0];
    $sql_statement = "SELECT count(*) from jobs where job_state=='completed';";
    if (($rc=&ade_sql_select($errstack_ref, $dbh, \@select_results, $sql_statement)) != $ade_err_ok) {
        return($rc);
    }
    $completed_jobs_count = ${$select_results[0]}[0];
    if ($opt_show_all_flag) {
        print $fh "jobs ($running_jobs_count running, $pending_jobs_count pending, $completed_jobs_count completed)\n----\n\n";
    } else {
        print $fh "jobs ($running_jobs_count running, $pending_jobs_count pending)\n----\n\n";
    }
    printf $fh "%-4s %9s %5s %3s %s\n%s\n","id","state","pid","rc","command","-"x78;
    if ($opt_show_all_flag) {
        $sql_statement = "SELECT job_id,job_state,pid,exit_code>>8,command FROM jobs order by job_state,job_id;";
    } else {
        $sql_statement = "SELECT job_id,job_state,pid,exit_code>>8,command FROM jobs where job_state!='completed' order by job_state,job_id;";
    }
    if (($rc=&ade_sql_select($errstack_ref, $dbh, \@select_results, $sql_statement)) != $ade_err_ok) {
        return($rc);
    }
    foreach $select_result (@select_results) {
        printf $fh "%-4d %9s %5d %3d %s\n", @{$select_result};
    }
    print $fh "\n";

    print $fh "resources\n---------\n\n";
    printf $fh "%-8s %8s %4s\n%s\n","id","max","curr","-"x78;
    $sql_statement = "select resources.resource_id, quantity_total, sum(coalesce(quantity_used,0)) as quantity_used from resources left outer join running_job_resource_pools on resources.resource_id = running_job_resource_pools.resource_id group by resources.resource_id order by resources.resource_id;";
    if (($rc=&ade_sql_select($errstack_ref, $dbh, \@select_results, $sql_statement)) != $ade_err_ok) {
        return($rc);
    }
    foreach $select_result (@select_results) {
        printf $fh "%-8s %8d %4d\n", @{$select_result};
    }
    print $fh "\n";

    close($fh);

    return($ade_err_ok);
}

&ade_gep_main(\&jsd);
