plsense 0.3.4-1 / usr / bin / plsense-server-work

#!/usr/bin/perl

eval 'exec /usr/bin/perl  -S $0 ${1+"$@"}'
    if 0; # not running under some shell

use strict;
use warnings;
use Getopt::Long qw{:config posix_default no_ignore_case gnu_compat};
use IO::Socket;
use File::Spec;
use File::Basename;
use Time::Piece;
use Try::Tiny;
use POSIX ();
use PlSense::Logger;
use PlSense::Configure;
use PlSense::SocketClient;
use PlSense::Util;
use PlSense::Builtin;

my %opthelp_of = ("-h, --help"      => "Show this message.",
                  "-c, --cachedir"  => "Path of directory caching information for Completion/Help.",
                  "--port1"         => "Port number for listening by main server process. Default is 33333.",
                  "--port2"         => "Port number for listening by work server process. Default is 33334.",
                  "--port3"         => "Port number for listening by resolve server process. Default is 33335.",
                  "--maxtasks"      => "Limit count of task that run on server process.",
                  "--loglevel"      => "Level of logging. Its value is for Log::Handler.",
                  "--logfile"       => "Path of log file.",
                  "--never-give-up" => "Never give up to listen port", );

my %function_of = (status    => \&get_status,
                   pid       => \&get_own_pid,
                   removeall => \&remove_all,
                   build     => \&build,
                   buildr    => \&build_recursive,
                   buildf    => \&build_force,
                   buildrf   => \&build_recursive_force,
                   buildfr   => \&build_recursive_force,
                   open      => \&open_file,
                   current   => \&current_file,
                   ps        => \&get_process_list,
                   queue     => \&get_task_queue,
                   finfind   => \&finish_find,
                   finbuild  => \&finish_build,
                   loglvl    => \&update_loglevel,
                   );

my ($cachedir, $port1, $port2, $port3, $maxtasks, $loglvl, $logfile, $never_give_up);
GetOptions ('help|h'        => sub { show_usage(); exit 0; },
            'cachedir|c=s'  => \$cachedir,
            'port1=i'       => \$port1,
            'port2=i'       => \$port2,
            'port3=i'       => \$port3,
            'maxtasks=i'    => \$maxtasks,
            'loglevel=s'    => \$loglvl,
            'logfile=s'     => \$logfile,
            'never-give-up' => \$never_give_up, );

setup_logger($loglvl, $logfile);
if ( ! $cachedir || ! -d $cachedir ) {
    logger->fatal("Not exist cache directory [$cachedir]");
    exit 1;
}
set_primary_config( cachedir => $cachedir,
                    port1    => $port1,
                    port2    => $port2,
                    port3    => $port3,
                    maxtasks => $maxtasks,
                    loglevel => $loglvl,
                    logfile  => $logfile, );
setup_config() or exit 1;

my $scli = PlSense::SocketClient->new({ retryinterval => 0.2, maxretry => 10 });
my (%taskinfo_of, @taskqueue, $local);
set_builtin( PlSense::Builtin->new() );

my $sock;
my $myport = get_config("port2");
SOCK_OPEN:
until ( $sock = IO::Socket::INET->new( LocalAddr => "localhost",
                                       LocalPort => $myport,
                                       Proto     => "tcp",
                                       Listen    => 1,
                                       ReUse     => 1, ) ) {
    if ( ! $never_give_up ) { last SOCK_OPEN; }
    logger->debug("Wait for releasing port : $myport");
    sleep 5;
}
if ( ! $sock ) {
    logger->fatal("Can't create socket : $!");
    exit 1;
}
if ( ! $sock->listen ) {
    logger->fatal("Can't listening port [$myport] : $!");
    exit 1;
}

set_signal_handler();
if ( ! initialize() ) {
    logger->fatal("Failed initialize");
    exit 1;
}
accept_client();
exit 0;


sub show_usage {
    my $optstr = "";
    OPTHELP:
    foreach my $key ( sort keys %opthelp_of ) {
        $optstr .= sprintf("  %-25s %s\n", $key, $opthelp_of{$key});
    }

    print <<"EOF";
Run PlSense Work Server.
Work Server manages task that find/build module.

Usage:
  plsense-server-work [Option]

Option:
$optstr
EOF
    return;
}

sub set_signal_handler {
    logger->debug("Start set signal handler");
    my $set_stop = POSIX::SigSet->new( &POSIX::SIGINT, &POSIX::SIGTERM );
    my $act_stop = POSIX::SigAction->new(
        sub {
            logger->notice("Receive SIGINT/SIGTERM");
            kill_all_task();
            exit 0;
        }, $set_stop, &POSIX::SA_NODEFER);

    my $set_restart = POSIX::SigSet->new( &POSIX::SIGUSR1, &POSIX::SIGHUP );
    my $act_restart = POSIX::SigAction->new(
        sub {
            logger->notice("Receive SIGUSR1/SIGHUP");
            kill_all_task();
            exec $0, get_common_options(), "--maxtasks", get_config("maxtasks"), "--never-give-up";
            exit 0;
        }, $set_restart, &POSIX::SA_NODEFER);

    POSIX::sigprocmask( &POSIX::SIG_UNBLOCK, $set_stop );
    POSIX::sigprocmask( &POSIX::SIG_UNBLOCK, $set_restart );

    POSIX::sigaction( &POSIX::SIGINT,  $act_stop );
    POSIX::sigaction( &POSIX::SIGTERM, $act_stop );
    POSIX::sigaction( &POSIX::SIGUSR1, $act_restart );
    POSIX::sigaction( &POSIX::SIGHUP,  $act_restart );
}

sub initialize {
    logger->debug("Start initialize");
    setup_config() or return;
    %taskinfo_of = ();
    @taskqueue = ();
    $local = 0;
    add_find_installed_task();
    run_task();
    builtin->reset;
    builtin->setup_without_reload();
    builtin->build;
    $scli->request_main_server("builtin", { ignore_error => 1,
                                            retryinterval => 0.5,
                                            maxretry => 200 }) or return;
    return 1;
}

sub accept_client {
    logger->info("Starting work server");

    ACCEPT_CLIENT:
    while ( my $cli = $sock->accept ) {
        logger->debug("Waiting client ...");

        my $line = $cli->getline || "";
        chomp $line;
        logger->info("Receive request : $line");
        my $cmdnm = $line =~ s{ ^ \s* ([a-z]+) }{}xms ? $1 : "";
        if ( $cmdnm eq "quit" ) {
            $cli->close;
            next ACCEPT_CLIENT;
        }
        elsif ( $cmdnm eq "stop" ) {
            $cli->close;
            last ACCEPT_CLIENT;
        }
        elsif ( exists $function_of{$cmdnm} ) {
            try {
                my $fnc = $function_of{$cmdnm};
                my $ret = &$fnc($line) || "";
                $cli->print($ret);
                $cli->flush;
                run_task();
            }
            catch {
                my $e = shift;
                logger->error("Failed do $cmdnm : $e");
            };
        }
        else {
            logger->error("Unknown command [$cmdnm]");
        }
        $cli->close;

    }
    $sock->close;

    logger->info("Stopping work server");
}

sub run_task {
    logger->notice("Start run task");

    my $now = localtime;
    my $taskcount = 0;
    CHK_TIMEOUT:
    foreach my $key ( keys %taskinfo_of ) {
        my $task = $taskinfo_of{$key};
        if ( $now > $task->{limit} ) {
            logger->notice("Timeout task : $key");
            my $pid = $task->{pid};
            if ( $pid && kill(0, $pid) ) {
                kill 'INT', $pid;
            }
            elsif ( $key =~ m{ \A build \s+ (.+) \z }xms ) {
                my $mdl_or_file = $1;
                logger->notice("Request reload module of zombie task : $key");
                $scli->request_main_server("built $mdl_or_file");
            }
            delete $taskinfo_of{$key};
        }
        else {
            $taskcount++;
        }
    }

    # do not run task while doing find task
    my @findtasks = grep { $_ =~ m{ \A find \s+ }xms } keys %taskinfo_of;
    if ( $#findtasks >= 0 ) {
        logger->notice("Exit run task cause find task is running");
        return;
    }

    my $limit = $now + 60 * 10;
    RUN_TASK:
    while ( $taskcount < get_config("maxtasks") ) {
        my $nexttask = shift @taskqueue or last RUN_TASK;
        my $taskkey = $nexttask->{key};
        logger->debug("Next task : $taskkey");
        if ( is_running($taskkey) ) { next RUN_TASK; }
        my $cmd = $nexttask->{cmd};
        logger->info("Run task : $cmd");
        system "$cmd &";
        # TODO: get pid of task
        $taskinfo_of{$taskkey} = { pid => undef, limit => $limit };
        $taskcount++;
        if ( $taskkey =~ m{ \A find \s+ }xms ) { last RUN_TASK; }
    }
    logger->debug("Finish run task");
}

sub add_task {
    my ($taskkey, $cmdstr) = @_;
    if ( ! $taskkey || ! $cmdstr ) { return; }
    QUEUE:
    foreach my $task ( @taskqueue ) {
        if ( $taskkey eq $task->{key} ) { return; }
    }
    push @taskqueue, { key => $taskkey, cmd => $cmdstr };
}

sub finish_task {
    my $taskkey = shift || "";
    if ( ! $taskkey ) { return; }
    logger->info("Finished task : $taskkey");
    delete $taskinfo_of{$taskkey};
}

sub is_running {
    my $taskkey = shift || "";
    exists $taskinfo_of{$taskkey};
}

sub kill_all_task {
    KILL_TASK:
    foreach my $key ( keys %taskinfo_of ) {
        my $task = $taskinfo_of{$key};
        my $pid = $task->{pid};
        if ( ! $pid ) { next KILL_TASK; }
        logger->notice("Kill task : $key");
        if ( kill(0, $pid) ) { kill 'INT', $pid; }
        delete $taskinfo_of{$key};
    }
}

sub get_worker_common_option {
    my $is_project = shift || 0;
    my $ret = get_common_option_string();
    $ret .= " --confpath '".( get_config("confpath") || "" )."'";
    if ( $is_project ) { $ret .= " --project"; }
    return $ret;
}

sub add_find_task {
    my ($global, $tasknm, @rootdirs) = @_;
    logger->notice("Add task find module of $tasknm");
    my $cmdstr = "plsense-worker-find".get_worker_common_option($global ? 0 : 1);
    $cmdstr .= " --tasknm '$tasknm'";
    ROOTDIR:
    foreach my $dir ( @rootdirs ) {
        $cmdstr .= " --rootdir '$dir'";
    }
    add_task("find $tasknm", $cmdstr);
    return;
}

sub add_find_installed_task {
    my @dirs;
    my $perl = get_config("perl");
    LIBPATH:
    foreach my $dir ( qx{ $perl -e 'pop \@INC; print join("\\n", \@INC);' } ) {
        chomp $dir;
        if ( ! -d $dir ) { next LIBPATH; }
        push @dirs, $dir;
    }
    add_find_task(1, "installed", @dirs);
}

sub update_location {
    my $filepath = shift || "";
    if ( ! -f $filepath ) { return; }

    $scli->request_main_server("onfile ".$filepath);

    my $oldconfpath = get_config("confpath") || "";
    setup_config($filepath) or return;
    my $newconfpath = get_config("confpath") || "";
    if ( $oldconfpath eq $newconfpath ) { return; }

    if ( builtin->setup() ) {
        builtin->build;
        $scli->request_main_server("builtin");
    }

    my $newlocal = get_config("local");
    if ( $local || $newlocal ) { add_find_installed_task(); }
    $local = $newlocal;

    my $newlibpath = get_config("lib-path") || "";
    if ( -d $newlibpath ) { add_find_task(0, get_config("name"), $newlibpath); }

    return 1;
}

sub build_sentinel {
    my $mdl_or_files = shift || "";
    my $recursive = shift || 0;
    my $force = shift || 0;
    my @mdl_or_files = split m{ \| }xms, $mdl_or_files;
    ENTRY:
    foreach my $mdl_or_file ( @mdl_or_files ) {
        $mdl_or_file =~ s{ ^\s+ }{}xms;
        $mdl_or_file =~ s{ \s+$ }{}xms;
        if ( ! $mdl_or_file ) { next ENTRY; }
        my $target = -f $mdl_or_file ? File::Spec->rel2abs($mdl_or_file) : $mdl_or_file;
        my $taskkey = "build ".$target;
        if ( is_running($taskkey) ) {
            logger->info("Quit build '$mdl_or_file'. It's now building or build already");
            next ENTRY;
        }
        logger->notice("Add task build '$mdl_or_file'. recursive[$recursive]");
        my $cmdstr = "plsense-worker-build".get_worker_common_option();
        $cmdstr .= " --target '$target'";
        if ( $recursive ) { $cmdstr .= " --recursive"; }
        if ( $force ) { $cmdstr .= " --force"; }
        add_task($taskkey, $cmdstr);
    }
    return;
}



sub get_status {
    return "Running\n";
}

sub get_own_pid {
    return $$."\n";
}

sub remove_all {
    kill_all_task();
    builtin->remove;
    return "Done\n";
}

sub build {
    my $mdl_or_files = shift || "";
    build_sentinel($mdl_or_files, 0);
}

sub build_recursive {
    my $mdl_or_files = shift || "";
    build_sentinel($mdl_or_files, 1);
}

sub build_force {
    my $mdl_or_files = shift || "";
    build_sentinel($mdl_or_files, 0, 1);
}

sub build_recursive_force {
    my $mdl_or_files = shift || "";
    build_sentinel($mdl_or_files, 1, 1);
}

sub open_file {
    my $filepath = shift || "";
    $filepath =~ s{ ^\s+ }{}xms;
    $filepath =~ s{ \s+$ }{}xms;
    if ( ! -f $filepath ) {
        logger->error("Not exist file[$filepath]");
        return "Failed\n";
    }
    update_location($filepath);
    build_sentinel($filepath, 1, 0);
    return "Done\n";
}

sub current_file {
    my $filepath = shift || "";
    $filepath =~ s{ ^\s+ }{}xms;
    $filepath =~ s{ \s+$ }{}xms;
    if ( ! -f $filepath ) {
        logger->error("Not exist file[$filepath]");
        return "Failed\n";
    }
    update_location($filepath);
    return "Done\n";
}

sub get_process_list {
    return join("\n", sort keys %taskinfo_of)."\n";
}

sub get_task_queue {
    my @ret;
    QUEUE:
    for my $i ( 0..$#taskqueue ) {
        push @ret, $taskqueue[$i]->{key};
    }
    return join("\n", @ret)."\n";
}

sub finish_find {
    my $tasknm = shift || "";
    $tasknm =~ s{ ^\s+ }{}xms;
    $tasknm =~ s{ \s+$ }{}xms;
    if ( ! $tasknm ) { return; }
    finish_task("find $tasknm");
    return;
}

sub finish_build {
    my $mdl_or_file = shift || "";
    $mdl_or_file =~ s{ ^\s+ }{}xms;
    $mdl_or_file =~ s{ \s+$ }{}xms;
    if ( ! $mdl_or_file ) { return; }
    my $taskkey = "build ".$mdl_or_file;
    finish_task($taskkey);
    return;
}

sub update_loglevel {
    my $loglvl = shift || "";
    $loglvl =~ s{ ^\s+ }{}xms;
    $loglvl =~ s{ \s+$ }{}xms;
    update_logger_level($loglvl);
    set_primary_config(loglevel => $loglvl);
    return;
}