ref updates: ignore non-update updates

[girocco.git] / taskd / taskd.pl

#!/usr/bin/perl
#
# taskd - Clone repositories on request
#
# taskd is Girocco mirroring servant; it processes requests for clones
# of given URLs received over its socket.
#
# When a request is received, new process is spawned that sets up
# the repository and reports further progress
# to .clonelog within the repository. In case the clone fails,
# .clone_failed is touched and .clone_in_progress is removed.

# Clone protocol:
# Alice sets up repository and touches .cloning
# Alice opens connection to Bob
# Alice sends project name through the connection
# Bob opens the repository and sends error code if there is a problem
# Bob closes connection
# Alice polls .clonelog in case of success.
# If Alice reads "@OVER@" from .clonelog, it stops polling.

# Ref-change protocol:
# Alice opens connection to Bob
# Alice sends ref-change command for each changed ref
# Alice closes connection
# Bob sends out notifications

# Initially based on perlipc example.

use 5.008; # we need safe signals
use strict;
use warnings;

use Getopt::Long;
use Pod::Usage;
use Socket;
use Errno;
use Fcntl;
use POSIX ":sys_wait_h";
use File::Basename;

use lib dirname($0);
use Girocco::Config;
use Girocco::Notify;
use Girocco::Project;
use Girocco::User;
use Girocco::Util qw(noFatalsToBrowser get_git);
BEGIN {noFatalsToBrowser}

# Throttle Classes Defaults
# Note that any same-named classes in @Girocco::Config::throttle_classes
# will override (completely replacing the entire hash) these ones.
my @throttle_defaults = (
        {
                name => "ref-change",
                maxproc => 0,
                maxjobs => 1,
                interval => 1
        },
        {
                name => "clone",
                maxproc => 0,
                maxjobs => 2,
                interval => 5
        },
        {
                name => "snapshot",
                #maxproc => max(5, cpucount + maxjobs), # this is the default
                #maxjobs => max(1, int(cpucount / 4)) , # this is the default
                interval => 5
        }
);

# Options
my $quiet;
my $progress;
my $syslog;
my $stderr;
my $inetd;
my $idle_timeout;
my $abbrev = 8;
my $showff = 1;
my $statusintv = 60;
my $idleintv = 3600;
my $maxspawn = 8;

$| = 1;

my $progname = basename($0);
my $children = 0;
my $idlestart = time;
my $idlestatus = 0;

sub cpucount {
        use Girocco::Util "online_cpus";
        our $online_cpus_result;
        $online_cpus_result = online_cpus unless $online_cpus_result;
        return $online_cpus_result;
}

sub logmsg {
        my $hdr = "[@{[scalar localtime]}] $progname $$: ";
        if (tied *STDOUT) {
                $OStream::only = 2; # STDERR only
                print "$hdr@_\n";
                $OStream::only = 1; # syslog only
                print "@_\n";
                $OStream::only = 0; # back to default
        } else {
                print "$hdr@_\n";
        }
}

sub statmsg {
        return unless $progress;
        my $hdr = "[@{[scalar localtime]}] $progname $$: ";
        if (tied *STDERR) {
                $OStream::only = 2; # STDERR only
                print STDERR "$hdr@_\n";
                $OStream::only = 1; # syslog only
                print STDERR "@_\n";
                $OStream::only = 0; # back to default
        } else {
                print STDERR "$hdr@_\n";
        }
}

sub duration {
        my $secs = shift;
        return $secs unless defined($secs) && $secs >= 0;
        $secs = int($secs);
        my $ans = ($secs % 60) . 's';
        return $ans if $secs < 60;
        $secs = int($secs / 60);
        $ans = ($secs % 60) . 'm' . $ans;
        return $ans if $secs < 60;
        $secs = int($secs / 60);
        $ans = ($secs % 24) . 'h' . $ans;
        return $ans if $secs < 24;
        $secs = int($secs / 24);
        return $secs . 'd' . $ans;
}

sub setnonblock {
        my $fd = shift;
        my $flags = fcntl($fd, F_GETFL, 0);
        defined($flags) or die "fcntl failed: $!";
        fcntl($fd, F_SETFL, $flags | O_NONBLOCK) or die "fcntl failed: $!";
}

sub setblock {
        my $fd = shift;
        my $flags = fcntl($fd, F_GETFL, 0);
        defined($flags) or die "fcntl failed: $!";
        fcntl($fd, F_SETFL, $flags & ~O_NONBLOCK) or die "fcntl failed: $!";
}

package Throttle;

#
## Throttle protocol
##
## 1) Process needing throttle services acquire a control file descriptor
##    a) Either as a result of a fork + exec (the write end of a pipe)
##    b) Or by connecting to the taskd socket (not yet implemented)
##
## 2) The process requesting throttle services will be referred to
##    as the supplicant or just "supp" for short.
##
## 3) The supp first completes any needed setup which may include
##    gathering data it needs to perform the action -- if that fails
##    then there's no need for any throttling.
##
## 4) The supp writes a throttle request to the control descriptor in
##    this format:
##        throttle <pid> <class>\n
##    for example if the supp's pid was 1234 and it was requesting throttle
##    control as a member of the mail class it would write this message:
##        throttle 1234 mail\n
##    Note that if the control descriptor happens to be a pipe rather than a
##    socket, the message should be preceded by another "\n" just be be safe.
##    If the control descriptor is a socket, not a pipe, the message may be
##    preceded by a "\n" but that's not recommended.
##
## 5) For supplicants with a control descriptor that is a pipe
##    (getsockopt(SO_TYPE) returns ENOTSOCK) the (5a) protocol should be used.
##    If the control descriptor is a socket (getsockname succeeds) then
##    protocol (5b) should be used.
##
## 5a) The supp now enters a "pause" loop awaiting either a SIGUSR1, SIGUSR2 or
##    SIGTERM.  It should wake up periodically (SIGALRM works well) and attempt
##    to write a "keepalive\n" message to the control descriptor.  If that
##    fails, the controller has gone away and it may make its own decision
##    whether or not to proceed at that point.  If, on the other hand, it
##    receives a SIGTERM, the process limit for its class has been reached
##    and it should abort without performing its action.  If it receives
##    SIGUSR1, it may proceed without writing anything more to the control
##    descriptor, any MAY even close the control descriptor.  Finally, a
##    SIGUSR2 indicates rejection of the throttle request for some other reason
##    such as unrecognized class name or invalid pid in which case the supp may
##    make its own decision how to proceed.
##
## 5b) The supp now enters a read wait on the socket -- it need accomodate no
##    more than 512 bytes and if a '\n' does not appear within that number of
##    bytes the read should be considered failed.  Otherwise the read should
##    be retried until either a full line has been read or the socket is
##    closed from the other end.  If the lone read is "proceed\n" then it may
##    proceed without reading or writing anything more to the control
##    descriptor, but MUST keep the control descriptor open and not call
##    shutdown on it either.  Any other result (except EINTR or EAGAIN which
##    should be retried) constitutes failure.  If a full line starting with at
##    least one alpha character was read but it was not "proceed" then it
##    should abort without performing its action.  For any other failure it
##    may make its own decision whether or not to proceed as the controller has
##    gone away.
##
## 6) The supp now performs its throttled action.
##
## 7) The supp now closes its control descriptor (if it hasn't already in the
##    case of (5a)) and exits -- in the case of a socket, the other end receives
##    notification that the socket has been closed (read EOF).  In the case of
##    a pipe the other end receives a SIGCHLD (multiple processes have a hold
##    of the other end of the pipe, so it will not reaach EOF by the supp's
##    exit in that case).
#

# keys are class names, values are hash refs with these fields:
# 'maxproc' => integer; maximum number of allowed supplicants (the sum of how
#                       many may be queued waiting plus how many may be
#                       concurrently active) with 0 meaning no limit.
# 'maxjobs' => integer; how many supplicants may proceed simultaneously a value
#                       of 0 is unlimited but the number of concurrent
#                       supplicants will always be limited to no more than
#                       the 'maxproc' value (if > 0) no matter what the
#                       'maxjobs' value is.
# 'total'   -> integer; the total number of pids belonging to this clase that
#                       can currently be found in %pid.
# 'active'  -> integer; the number of currently active supplicants which should
#                       be the same as (the number of elements of %pid with a
#                       matching class name) - (number of my class in @queue).
# 'interval' -> integer; minimum number of seconds between 'proceed' responses
#                       or SIGUSR1 signals to members of this class.
# 'lastqueue' -> time;  last time a supplicant was successfully queued.
# 'lastproceed' => time; last time a supplicant was allowed to proceed.
# 'lastthrottle' => time; last time a supplicant was throttled
# 'lastdied'  => time;  last time a supplicant in this class died/exited/etc.
my %classes = ();

# keys are pid numbers, values are array refs with these elements:
# [0] => name of class (key to classes hash)
# [1] => supplicant state (0 => queued, non-zero => time it started running)
# [2] => descriptive text (e.g. project name)
my %pid = ();

# minimum number of seconds between any two proceed responses no matter what
# class.  this takes priority in that it can effectively increase the
# class's 'interval' value by delaying proceed notifications if the minimum
# interval has not yet elapsed.
my $interval = 1;

# fifo of pids awaiting notification as soon as the next $interval elapses
# provided interval and maxjobs requirements are satisfied
# for the class of the pid that will next be triggered.
my @queue = ();

# time of most recent successful call to AddSupplicant
my $lastqueue = 0;

# time of most recent proceed notification
my $lastproceed = 0;

# time of most recent throttle
my $lastthrottle = 0;

# time of most recent removal
my $lastdied = 0;

# lifetime count of how many have been queued
my $totalqueue = 0;

# lifetime count of how many have been allowed to proceed
my $totalproceed = 0;

# lifetime count of how many have been throttled
my $totalthrottle = 0;

# lifetime count of how many have died
# It should always be true that $totalqueued - $totaldied == $curentlyactive
my $totaldied = 0;

# Returns an unordered list of currently registered class names
sub GetClassList {
        return keys(%classes);
}

sub _max {
        return $_[0] if $_[0] >= $_[1];
        return $_[1];
}

sub _getnum {
        my ($min, $val, $default) = @_;
        my $ans;
        if (defined($val) && $val =~ /^[+-]?\d+$/) {
                $ans = 0 + $val;
        } else {
                $ans = &$default;
        }
        return _max($min, $ans);
}

# [0] => name of class to find
# [1] => if true, create class if it doesn't exist, if a hashref then
#        it contains initial values for maxproc, maxjobs and interval.
#        Otherwise maxjobs defaults to max(cpu cores/4, 1), maxprocs
#        defaults to the max(5, number of cpu cores + maxjobs) and interval
#        defaults to 1.
# Returns a hash ref with info about the class on success
sub GetClassInfo {
        my ($classname, $init) = @_;
        defined($classname) && $classname =~ /^[a-zA-Z][a-zA-Z0-9._+-]*$/
                or return;
        $classname = lc($classname);
        my %info;
        if ($classes{$classname}) {
                %info = %{$classes{$classname}};
                return \%info;
        }
        return unless $init;
        my %newclass = ();
        ref($init) eq 'HASH' or $init = {};
        $newclass{'maxjobs'} = _getnum(0, $init->{'maxjobs'}, sub{_max(1, int(::cpucount() / 4))});
        $newclass{'maxproc'} = _getnum(0, $init->{'maxproc'}, sub{_max(5, ::cpucount() + $newclass{'maxjobs'})});
        $newclass{'interval'} = _getnum(0, $init->{'interval'}, sub{1});
        $newclass{'total'} = 0;
        $newclass{'active'} = 0;
        $newclass{'lastqueue'} = 0;
        $newclass{'lastproceed'} = 0;
        $newclass{'lastthrottle'} = 0;
        $newclass{'lastdied'} = 0;
        $classes{$classname} = \%newclass;
        %info = %newclass;
        return \%info;
}

# [0] => pid to look up
# Returns () if not found otherwise ($classname, $timestarted, $description)
# Where $timestarted will be 0 if it's still queued otherwise a time() value
sub GetPidInfo {
        my $pid = shift;
        return () unless exists $pid{$pid};
        return @{$pid{$pid}};
}

# Returns array of pid numbers that are currently running sorted
# by time started (oldest to newest).  Can return an empty array.
sub GetRunningPids {
        return sort({ ${$pid{$a}}[1] <=> ${$pid{$b}}[1] }
                grep({ ${$pid{$_}}[1] } keys(%pid)));
}

# Returns a hash with various about the current state
# 'interval'      => global minimum interval between proceeds
# 'active'        => how many pids are currently queued + how many are running
# 'queue'         => how many pids are currently queued
# 'lastqueue'     => time (epoch seconds) of last queue
# 'lastproceed'   => time (epoch seconds) of last proceed
# 'lastthrottle'  => time (epoch seconds) of last throttle
# 'lastdied'      => time (epoch seconds) of last removal
# 'totalqueue'    => lifetime total number of processes queued
# 'totalproceed'  => lifetime total number of processes proceeded
# 'totalthrottle' => lifetime total number of processes throttled
# 'totaldied'     => lifetime total number of removed processes
sub GetInfo {
        return {
                interval => $interval,
                active => scalar(keys(%pid)) - scalar(@queue),
                queue => scalar(@queue),
                lastqueue => $lastqueue,
                lastproceed => $lastproceed,
                lastthrottle => $lastthrottle,
                lastdied => $lastdied,
                totalqueue => $totalqueue,
                totalproceed => $totalproceed,
                totalthrottle => $totalthrottle,
                totaldied => $totaldied
        };
}

# with no args get the global interval
# with one arg set it, returns previous value if set
sub Interval {
        my $ans = $interval;
        $interval = 0 + $_[0] if defined($_[0]) && $_[0] =~ /^\d+$/;
        return $ans;
}

sub RemoveSupplicant;

# Perform queue service (i.e. send SIGUSR1 to any eligible queued process)
# Returns minimum interval until next proceed is possible
# Returns undef if there's nothing waiting to proceed or
# the 'maxjobs' limits have been reached for all queued items (in which
# case it won't be possible to proceed until one of them exits, hence undef)
# This is called automatially by AddSupplicant and RemoveSupplicant
sub ServiceQueue {
        RETRY:
        return undef unless @queue; # if there's nothing queued, nothing to do
        my $now = time;
        my $min = _max(0, $interval - ($now - $lastproceed));
        my $classmin = undef;
        my $classchecked = 0;
        my %seenclass = ();
        my $classcount = scalar(keys(%classes));
        for (my $i=0; $i <= $#queue && $classchecked < $classcount; ++$i) {
                my $pid = $queue[$i];
                my $procinfo = $pid{$pid};
                if (!$procinfo) {
                        RemoveSupplicant($pid, 1);
                        goto RETRY;
                }
                my $classinfo = $classes{$$procinfo[0]};
                if (!$classinfo) {
                        RemoveSupplicant($pid, 1);
                        goto RETRY;
                }
                if (!$seenclass{$$procinfo[0]}) {
                        $seenclass{$$procinfo[0]} = 1;
                        ++$classchecked;
                        if (!$classinfo->{'maxjobs'} || $classinfo->{'active'} < $classinfo->{'maxjobs'}) {
                                my $cmin = _max(0, $classinfo->{'interval'} - ($now - $classinfo->{'lastproceed'}));
                                if (!$cmin && !$min) {
                                        $now = time;
                                        $$procinfo[1] = $now;
                                        splice(@queue, $i, 1);
                                        ++$totalproceed;
                                        $lastproceed = $now;
                                        $classinfo->{'lastproceed'} = $now;
                                        ++$classinfo->{'active'};
                                        kill("USR1", $pid) or RemoveSupplicant($pid, 1);
                                        goto RETRY;
                                }
                                $classmin = $cmin unless defined($classmin) && $classmin < $cmin;
                        }
                }
        }
        return defined($classmin) ? _max($min, $classmin) : undef;
}

# $1 => pid to add (must not already be in %pids)
# $2 => class name (must exist)
# Returns -1 if no such class or pid already present or invalid
# Returns 0 if added successfully (and possibly already SIGUSR1'd)
# Return 1 if throttled and cannot be added
sub AddSupplicant {
        my ($pid, $classname, $text, $noservice) = @_;
        return -1 unless $pid && $pid =~ /^[1-9][0-9]*$/;
        $pid += 0;
        kill(0, $pid) or return -1;
        my $classinfo = $classes{$classname};
        return -1 unless $classinfo;
        return -1 if $pid{$pid};
        $text = '' unless defined($text);
        my $now = time;
        if ($classinfo->{'maxproc'} && $classinfo->{'total'} >= $classinfo->{'maxproc'}) {
                ++$totalthrottle;
                $lastthrottle = $now;
                $classinfo->{'lastthrottle'} = $now;
                return 1;
        }
        ++$totalqueue;
        $lastqueue = $now;
        $pid{$pid} = [$classname, 0, $text];
        ++$classinfo->{'total'};
        $classinfo->{'lastqueue'} = $now;
        push(@queue, $pid);
        ServiceQueue unless $noservice;
        return 0;
}

# $1 => pid to remove (died, killed, exited normally, doesn't matter)
# Returns 0 if removed
# Returns -1 if unknown pid or other error during removal
sub RemoveSupplicant {
        my ($pid, $noservice) = @_;
        return -1 unless defined($pid) && $pid =~ /^\d+$/;
        $pid += 0;
        my $pidinfo = $pid{$pid};
        $pidinfo or return -1;
        my $now = time;
        $lastdied = $now;
        ++$totaldied;
        delete $pid{$pid};
        if (!$$pidinfo[1]) {
                for (my $i=0; $i<=$#queue; ++$i) {
                        if ($queue[$i] == $pid) {
                                splice(@queue, $i, 1);
                                --$i;
                        }
                }
        }
        my $classinfo = $classes{$$pidinfo[0]};
        ServiceQueue, return -1 unless $classinfo;
        --$classinfo->{'active'} if $$pidinfo[1];
        --$classinfo->{'total'};
        $classinfo->{'lastdied'} = $now;
        ServiceQueue unless $noservice;
        return 0;
}

# Instance Methods

package main;

#
## ---------
## Functions
## ---------
#

my @reapedpids = ();
my %signame = (
         # http://pubs.opengroup.org/onlinepubs/000095399/utilities/trap.html
         1 => 'SIGHUP',
         2 => 'SIGINT',
         3 => 'SIGQUIT',
         6 => 'SIGABRT',
         9 => 'SIGKILL',
        14 => 'SIGALRM',
        15 => 'SIGTERM',
);
sub REAPER {
        local $!;
        my $child;
        my $waitedpid;
        while (($waitedpid = waitpid(-1, WNOHANG)) > 0) {
                my $code = $? & 0xffff;
                $idlestart = time if !--$children;
                my $codemsg = '';
                if (!($code & 0xff)) {
                        $codemsg = " with exit code ".($code >> 8) if $code;
                } elsif ($code & 0x7f) {
                        my $signum = ($code & 0x7f);
                        $codemsg = " with signal ".
                                ($signame{$signum}?$signame{$signum}:$signum);
                }
                logmsg "reaped $waitedpid$codemsg";
                push(@reapedpids, $waitedpid);
        }
        $SIG{CHLD} = \&REAPER;  # loathe sysV
}

$SIG{CHLD} = \&REAPER;  # Apollo 440

my ($piperead, $pipewrite);
sub spawn {
        my $coderef = shift;

        my $pid = fork;
        if (not defined $pid) {
                logmsg "cannot fork: $!";
                return;
        } elsif ($pid) {
                $idlestart = time if !++$children;
                $idlestatus = 0;
                logmsg "begat $pid";
                return; # I'm the parent
        }

        close(Server) unless fileno(Server) == 0;
        close($piperead);
        $SIG{'CHLD'} = sub {};

        open STDIN, "+<&Client" or die "can't dup client to stdin";
        close(Client);
        exit &$coderef();
}

# returns:
#   < 0: error
#   = 0: proceed
#   > 0: throttled
sub request_throttle {
        use POSIX qw(sigprocmask sigsuspend SIG_SETMASK);
        my $classname = shift;
        my $text = shift;

        Throttle::GetClassInfo($classname)
                or return -1; # no such throttle class

        my $throttled = 0;
        my $proceed = 0;
        my $error = 0;
        my $controldead = 0;
        my $setempty = POSIX::SigSet->new;
        my $setfull = POSIX::SigSet->new;
        $setempty->emptyset();
        $setfull->fillset();
        $SIG{'TERM'} = sub {$throttled = 1};
        $SIG{'USR1'} = sub {$proceed = 1};
        $SIG{'USR2'} = sub {$error = 1};
        $SIG{'PIPE'} = sub {$controldead = 1};
        $SIG{'ALRM'} = sub {};

        # After writing we can expect a SIGTERM, SIGUSR1 or SIGUSR2
        print $pipewrite "\nthrottle $$ $classname $text\n";
        my $old = POSIX::SigSet->new;
        sigprocmask(SIG_SETMASK, $setfull, $old);
        until ($controldead || $throttled || $proceed || $error) {
                alarm(30);
                sigsuspend($setempty);
                alarm(0);
                sigprocmask(SIG_SETMASK, $setempty, $old);
                print $pipewrite "\nkeepalive $$\n";
                sigprocmask(SIG_SETMASK, $setfull, $old);
        }
        sigprocmask(SIG_SETMASK, $setempty, $old);
        $SIG{'TERM'} = "DEFAULT";
        $SIG{'USR1'} = "DEFAULT";
        $SIG{'USR2'} = "DEFAULT";
        $SIG{'ALRM'} = "DEFAULT";
        $SIG{'PIPE'} = "DEFAULT";

        my $result = -1;
        if ($throttled) {
                $result = 1;
        } elsif ($proceed) {
                $result = 0;
        }
        return $result;
}

sub clone {
        my ($name) = @_;
        Girocco::Project::does_exist($name, 1) or die "no such project: $name";
        my $proj = Girocco::Project->load($name);
        $proj or die "failed to load project $name";
        $proj->{clone_in_progress} or die "project $name is not marked for cloning";
        $proj->{clone_logged} and die "project $name is already being cloned";
        request_throttle("clone", $name) <= 0 or die "cloning $name aborted (throttled)";
        statmsg "cloning $name";
        open STDOUT, '>', "$Girocco::Config::reporoot/$name.git/.clonelog" or die "cannot open clonelog: $!";
        open STDERR, ">&STDOUT";
        open STDIN, '<', '/dev/null';
        exec "$Girocco::Config::basedir/taskd/clone.sh", "$name.git" or die "exec failed: $!";
}

sub ref_indicator {
        return ' -> ' unless $showff && defined($_[0]);
        my ($git_dir, $old, $new) = @_;
        return '..' unless defined($old) && defined($new) && $old !~ /^0+$/ && $new !~ /^0+$/ && $old ne $new;
        # In many cases `git merge-base` is slower than this even if using the
        # `--is-ancestor` option available since Git 1.8.0, but it's never faster
        my $ans = get_git("--git-dir=$git_dir", "rev-list", "-n", "1", "^$new^0", "$old^0", "--") ? '...' : '..';
        return wantarray ? ($ans, 1) : $ans;
}

sub ref_change {
        my ($arg) = @_;
        my ($username, $name, $oldrev, $newrev, $ref) = split(/\s+/, $arg);
        $username && $name && $oldrev && $newrev && $ref or return 0;
        $oldrev =~ /^[0-9a-f]{40}$/ && $newrev =~ /^[0-9a-f]{40}$/ && $ref =~ m{^refs/} or return 0;
        $newrev ne $oldrev or return 0;

        Girocco::Project::does_exist($name, 1) or die "no such project: $name";
        my $proj = Girocco::Project->load($name);
        $proj or die "failed to load project $name";
        my $has_notify = $proj->has_notify;
        my $type = $has_notify ? "notify" : "change";

        my $user;
        if ($username && $username !~ /^%.*%$/) {
                Girocco::User::does_exist($username, 1) or die "no such user: $username";
                $user = Girocco::User->load($username);
                $user or die "failed to load user $username";
        } elsif ($username eq "%$name%") {
                $username = "-";
        }

        request_throttle("ref-change", $name) <= 0 or die "ref-change $name aborted (throttled)";
        my $ind = ref_indicator($proj->{path}, $oldrev, $newrev);
        statmsg "ref-$type $username $name ($ref: @{[substr($oldrev,0,$abbrev)]}$ind@{[substr($newrev,0,$abbrev)]})";
        open STDIN, '<', '/dev/null';
        Girocco::Notify::ref_change($proj, $user, $ref, $oldrev, $newrev) if $has_notify;
        return 0;
}

sub ref_changes {
        my ($arg) = @_;
        my ($username, $name) = split(/\s+/, $arg);
        $username && $name or return 0;

        Girocco::Project::does_exist($name, 1) or die "no such project: $name";
        my $proj = Girocco::Project->load($name);
        $proj or die "failed to load project $name";
        my $has_notify = $proj->has_notify;
        my $type = $has_notify ? "notify" : "change";

        my $user;
        if ($username && $username !~ /^%.*%$/) {
                Girocco::User::does_exist($username, 1) or die "no such user: $username";
                $user = Girocco::User->load($username);
                $user or die "failed to load user $username";
        } elsif ($username eq "%$name%") {
                $username = "-";
        }

        my @changes = ();
        while (my $change = <STDIN>) {
                my ($oldrev, $newrev, $ref) = split(/\s+/, $change);
                $oldrev =~ /^[0-9a-f]{40}$/ && $newrev =~ /^[0-9a-f]{40}$/ && $ref =~ m{^refs/} or next;
                $newrev ne $oldrev or next;
                push(@changes, [$oldrev, $newrev, $ref]);
        }
        return 0 unless @changes;
        open STDIN, '<', '/dev/null';
        request_throttle("ref-change", $name) <= 0 or die "ref-changes $name aborted (throttled)";
        foreach my $change (@changes) {
                my ($oldrev, $newrev, $ref) = @$change;
                my ($ind, $rangit) = ref_indicator($proj->{path}, $oldrev, $newrev);
                statmsg "ref-$type $username $name ($ref: @{[substr($oldrev,0,$abbrev)]}$ind@{[substr($newrev,0,$abbrev)]})";
                Girocco::Notify::ref_change($proj, $user, $ref, $oldrev, $newrev) if $has_notify;
                sleep 1 if $has_notify || $rangit;
        }
        return 0;
}

sub throttle {
        my ($arg) = @_;
        my ($pid, $classname, $text) = split(/\s+/, $arg);
        $pid =~ /^\d+/ or return 0; # invalid pid
        $pid += 0;
        $pid > 0 or return 0; # invalid pid
        kill(0, $pid) || $!{EPERM} or return 0; # no such process
        Throttle::GetClassInfo($classname) or return 0; # no such throttle class
        defined($text) && $text ne '' or return 0; # no text no service

        my $throttled = 0;
        my $proceed = 0;
        my $error = 0;
        my $controldead = 0;
        my $suppdead = 0;
        my ($waker, $wakew);
        pipe($waker, $wakew) or die "pipe failed: $!";
        select((select($wakew),$|=1)[0]);
        setnonblock($wakew);
        $SIG{'TERM'} = sub {$throttled = 1; syswrite($wakew, '!')};
        $SIG{'USR1'} = sub {$proceed = 1; syswrite($wakew, '!')};
        $SIG{'USR2'} = sub {$error = 1; syswrite($wakew, '!')};
        $SIG{'PIPE'} = sub {$controldead = 1; syswrite($wakew, '!')};
        select((select(STDIN),$|=1)[0]);

        logmsg "throttle $pid $classname $text request";
        # After writing we can expect a SIGTERM or SIGUSR1
        print $pipewrite "\nthrottle $$ $classname $text\n";

        # NOTE: the only way to detect the socket close is to read all the
        #       data until EOF is reached -- recv can be used to peek.
        my $v = '';
        vec($v, fileno(STDIN), 1) = 1;
        vec($v, fileno($waker), 1) = 1;
        setnonblock(\*STDIN);
        setnonblock($waker);
        until ($controldead || $throttled || $proceed || $error || $suppdead) {
                my ($r, $e);
                select($r=$v, undef, $e=$v, 30);
                my ($bytes, $discard);
                do {$bytes = sysread($waker, $discard, 512)} while (defined($bytes) && $bytes > 0);
                do {$bytes = sysread(STDIN, $discard, 4096)} while (defined($bytes) && $bytes > 0);
                $suppdead = 1 unless !defined($bytes) && $!{EAGAIN};
                print $pipewrite "\nkeepalive $$\n";
        }
        setblock(\*STDIN);

        if ($throttled && !$suppdead) {
                print STDIN "throttled\n";
                logmsg "throttle $pid $classname $text throttled";
        } elsif ($proceed && !$suppdead) {
                print STDIN "proceed\n";
                logmsg "throttle $pid $classname $text proceed";
                $SIG{'TERM'} = 'DEFAULT';
                # Stay alive until the child dies which we detect by EOF on STDIN
                setnonblock(\*STDIN);
                until ($controldead || $suppdead) {
                        my ($r, $e);
                        select($r=$v, undef, $e=$v, 30);
                        my ($bytes, $discard);
                        do {$bytes = sysread($waker, $discard, 512)} while (defined($bytes) && $bytes > 0);
                        do {$bytes = sysread(STDIN, $discard, 512)} while (defined($bytes) && $bytes > 0);
                        $suppdead = 1 unless !defined($bytes) && $!{EAGAIN};
                        print $pipewrite "\nkeepalive $$\n";
                }
                setblock(\*STDIN);
        } else {
                my $prefix = '';
                $prefix = "control" if $controldead && !$suppdead;
                logmsg "throttle $pid $classname $text ${prefix}died";
        }
        exit 0;
}

sub process_pipe_msg {
        my ($act, $pid, $cls, $text) = split(/\s+/, $_[0]);
        if ($act eq "throttle") {
                $pid =~ /^\d+$/ or return 0;
                $pid += 0;
                $pid > 0 or return 0; # invalid pid
                kill(0, $pid) or return 0; # invalid pid
                defined($cls) && $cls ne "" or kill('USR2', $pid), return 0;
                defined($text) && $text ne "" or kill('USR2', $pid), return 0;
                Throttle::GetClassInfo($cls) or kill('USR2', $pid), return 0;
                # the AddSupplicant call could send SIGUSR1 before it returns
                my $result = Throttle::AddSupplicant($pid, $cls, $text);
                kill('USR2', $pid), return 0 if $result < 0;
                kill('TERM', $pid), return 0 if $result > 0;
                # $pid was added to class $cls and will receive SIGUSR1 when
                # it's time for it to proceed
                return 0;
        } elsif ($act eq "keepalive") {
                # nothing to do although we could verify pid is valid and
                # still in %Throttle::pids and send a SIGUSR2 if not, but
                # really keepalive should just be ignored.
                return 0;
        }
        print STDERR "discarding unknown pipe message \"$_[0]\"\n";
        return 0;
}

#
## -------
## OStream
## -------
#

package OStream;

# Set to 1 for only syslog output (if enabled by mode)
# Set to 2 for only stderr output (if enabled by mode)
our $only = 0; # This is a hack

use Carp 'croak';
use Sys::Syslog qw(:DEFAULT :macros);

sub writeall {
        use POSIX qw();
        use Errno;
        my ($fd, $data) = @_;
        my $offset = 0;
        my $remaining = length($data);
        while ($remaining) {
                my $bytes = POSIX::write(
                        $fd,
                        substr($data, $offset, $remaining),
                        $remaining);
                next if !defined($bytes) && $!{EINTR};
                croak "POSIX::write failed: $!" unless defined $bytes;
                croak "POSIX::write wrote 0 bytes" unless $bytes;
                $remaining -= $bytes;
                $offset += $bytes;
        }
}

sub dumpline {
        use POSIX qw(STDERR_FILENO);
        my ($self, $line) = @_;
        $only = 0 unless defined($only);
        writeall(STDERR_FILENO, $line) if $self->{'stderr'} && $only != 1;
        substr($line, -1, 1) = '' if substr($line, -1, 1) eq "\n";
        return unless length($line);
        syslog(LOG_NOTICE, "%s", $line) if $self->{'syslog'} && $only != 2;
}

sub TIEHANDLE {
        my $class = shift || 'OStream';
        my $mode = shift;
        my $syslogname = shift;
        my $syslogfacility = shift;
        defined($syslogfacility) or $syslogfacility = LOG_USER;
        my $self = {};
        $self->{'syslog'} = $mode > 0;
        $self->{'stderr'} = $mode <= 0 || $mode > 1;
        $self->{'lastline'} = '';
        if ($self->{'syslog'}) {
                # Some Sys::Syslog have a stupid default setlogsock order
                eval {Sys::Syslog::setlogsock("native"); 1;} or
                eval {Sys::Syslog::setlogsock("unix");};
                openlog($syslogname, "ndelay,pid", $syslogfacility)
                        or croak "Sys::Syslog::openlog failed: $!";
        }
        return bless $self, $class;
}

sub BINMODE {return 1}
sub FILENO {return undef}
sub EOF {return 0}
sub CLOSE {return 1}

sub PRINTF {
        my $self = shift;
        my $template = shift;
        return $self->PRINT(sprintf $template, @_);
}

sub PRINT {
        my $self = shift;
        my $data = join('', $self->{'lastline'}, @_);
        my $pos = 0;
        while ((my $idx = index($data, "\n", $pos)) >= 0) {
                ++$idx;
                my $line = substr($data, $pos, $idx - $pos);
                substr($data, $pos, $idx - $pos) = '';
                $pos = $idx;
                $self->dumpline($line);
        }
        $self->{'lastline'} = $data;
        return 1;
}

sub DESTROY {
        my $self = shift;
        $self->dumpline($self->{'lastline'})
                if length($self->{'lastline'});
        closelog;
}

sub WRITE {
        my $self = shift;
        my ($scalar, $length, $offset) = @_;
        $scalar = '' if !defined($scalar);
        $length = length($scalar) if !defined($length);
        croak "OStream::WRITE invalid length $length"
                if $length < 0;
        $offset = 0 if !defined($offset);
        $offset += length($scalar) if $offset < 0;
        croak "OStream::WRITE invalid write offset"
                if $offset < 0 || $offset > $length;
        my $max = length($scalar) - $offset;
        $length = $max if $length > $max;
        $self->PRINT(substr($scalar, $offset, $length));
        return $length;
}

#
## ----
## main
## ----
#

package main;

close(DATA) if fileno(DATA);
my $sfac;
Getopt::Long::Configure('bundling');
my ($stiv, $idiv);
my $parse_res = GetOptions(
        'help|?|h' => sub {pod2usage(-verbose => 2, -exitval => 0)},
        'quiet|q' => \$quiet,
        'no-quiet' => sub {$quiet = 0},
        'progress|P' => \$progress,
        'inetd|i' => sub {$inetd = 1; $syslog = 1; $quiet = 1;},
        'idle-timeout|t=i' => \$idle_timeout,
        'syslog|s:s' => \$sfac,
        'no-syslog' => sub {$syslog = 0; $sfac = undef;},
        'stderr' => \$stderr,
        'abbrev=i' => \$abbrev,
        'show-fast-forward-info' => \$showff,
        'no-show-fast-forward-info' => sub {$showff = 0},
        'status-interval=i' => \$stiv,
        'idle-status-interval=i' => \$idiv,
) || pod2usage(2);
$syslog = 1 if defined($sfac);
$progress = 1 unless $quiet;
$abbrev = 128 unless $abbrev > 0;
if (defined($idle_timeout)) {
        die "--idle-timeout must be a whole number" unless $idle_timeout =~ /^\d+$/;
        die "--idle-timeout may not be used without --inetd" unless $inetd;
}
if (defined($stiv)) {
        die "--status-interval must be a whole number" unless $stiv =~ /^\d+$/;
        $statusintv = $stiv * 60;
}
if (defined($idiv)) {
        die "--idle-status-interval must be a whole number" unless $idiv =~ /^\d+$/;
        $idleintv = $idiv * 60;
}

open STDOUT, '>&STDERR' if $inetd;
if ($syslog) {
        use Sys::Syslog qw();
        my $mode = 1;
        ++$mode if $stderr;
        $sfac = "user" unless defined($sfac) && $sfac ne "";
        my $ofac = $sfac;
        $sfac = uc($sfac);
        $sfac = 'LOG_'.$sfac unless $sfac =~ /^LOG_/;
        my $facility;
        my %badfac = map({("LOG_$_" => 1)}
                (qw(PID CONS ODELAY NDELAY NOWAIT PERROR FACMASK NFACILITIES PRIMASK LFMT)));
        eval "\$facility = Sys::Syslog::$sfac; 1" or die "invalid syslog facility: $ofac";
        die "invalid syslog facility: $ofac"
                if ($facility & ~0xf8) || ($facility >> 3) > 23 || $badfac{$sfac};
        tie *STDERR, 'OStream', $mode, $progname, $facility or die "tie failed";
}
if ($quiet) {
        open STDOUT, '>', '/dev/null';
} elsif ($inetd) {
        *STDOUT = *STDERR;
}

my $NAME;

if ($inetd) {
        open Server, '<&=0' or die "open: $!";
        my $sockname = getsockname Server;
        die "getsockname: $!" unless $sockname;
        die "socket already connected! must be 'wait' socket" if getpeername Server;
        die "getpeername: $!" unless $!{ENOTCONN};
        my $st = getsockopt Server, SOL_SOCKET, SO_TYPE;
        die "getsockopt(SOL_SOCKET, SO_TYPE): $!" unless $st;
        my $socktype = unpack('i', $st);
        die "stream socket required" unless defined $socktype && $socktype == SOCK_STREAM;
        die "AF_UNIX socket required" unless sockaddr_family($sockname) == AF_UNIX;
        $NAME = unpack_sockaddr_un $sockname;
        my $expected = $Girocco::Config::chroot.'/etc/taskd.socket';
        warn "listening on \"$NAME\" but expected \"$expected\"" unless $NAME eq $expected;
        my $mode = (stat($NAME))[2];
        die "stat: $!" unless $mode;
        $mode &= 07777;
        if (($mode & 0660) != 0660) {
                chmod(($mode|0660), $NAME) == 1 or die "chmod ug+rw \"$NAME\" failed: $!";
        }
} else {
        $NAME = $Girocco::Config::chroot.'/etc/taskd.socket';
        my $uaddr = sockaddr_un($NAME);

        socket(Server, PF_UNIX, SOCK_STREAM, 0) or die "socket failed: $!";
        unlink($NAME);
        bind(Server, $uaddr) or die "bind failed: $!";
        listen(Server, SOMAXCONN) or die "listen failed: $!";
        chmod 0666, $NAME or die "chmod failed: $!";
}

foreach my $throttle (@Girocco::Config::throttle_classes, @throttle_defaults) {
        my $classname = $throttle->{"name"};
        $classname or next;
        Throttle::GetClassInfo($classname, $throttle);
}

sub _min {
        return $_[0] <= $_[1] ? $_[0] : $_[1];
}

pipe($piperead, $pipewrite) or die "pipe failed: $!";
setnonblock($piperead);
select((select($pipewrite), $|=1)[0]);
my $pipebuff = '';
my $fdset_both = '';
vec($fdset_both, fileno($piperead), 1) = 1;
my $fdset_pipe = $fdset_both;
vec($fdset_both, fileno(Server), 1) = 1;
my $penalty = 0;
my $t = time;
my $penaltytime = $t;
my $nextwakeup = $t + 60;
my $nextstatus = undef;
$nextstatus = $t + $statusintv if $statusintv;
statmsg "listening on $NAME";
while (1) {
        my ($rout, $eout, $nfound);
        do {
                my $wait;
                my $now = time;
                my $adjustpenalty = sub {
                        if ($penaltytime < $now) {
                                my $credit = $now - $penaltytime;
                                $penalty = $penalty > $credit ? $penalty - $credit : 0;
                                $penaltytime = $now;
                        }
                };
                if (defined($nextstatus) && $now >= $nextstatus) {
                        unless ($idlestatus && !$children && (!$idleintv || $now - $idlestatus < $idleintv)) {
                                my $statmsg = "STATUS: $children active";
                                my @running = ();
                                if ($children) {
                                        my @stats = ();
                                        my $cnt = 0;
                                        foreach my $cls (sort(Throttle::GetClassList())) {
                                                my $inf = Throttle::GetClassInfo($cls);
                                                if ($inf->{'total'}) {
                                                        $cnt += $inf->{'total'};
                                                        push(@stats, substr(lc($cls),0,1)."=".
                                                                $inf->{'total'}.'/'.$inf->{'active'});
                                                }
                                        }
                                        push(@stats, "?=".($children-$cnt)) if @stats && $cnt < $children;
                                        $statmsg .= " (".join(" ",@stats).")" if @stats;
                                        foreach (Throttle::GetRunningPids()) {
                                                my ($cls, $ts, $desc) = Throttle::GetPidInfo($_);
                                                next unless $ts;
                                                push(@running, "[${cls}::$desc] ".duration($now-$ts));
                                        }
                                }
                                my $idlesecs;
                                $statmsg .= ", idle " . duration($idlesecs)
                                        if !$children && ($idlesecs = $now - $idlestart) >= 2;
                                statmsg $statmsg;
                                statmsg "STATUS: currently running: ".join(", ", @running)
                                        if @running;
                                $idlestatus = $now if !$children;
                        }
                        $nextstatus += $statusintv while $nextstatus <= $now;
                }
                $nextwakeup += 60, $now = time while ($wait = $nextwakeup - $now) <= 0;
                $wait = _min($wait, (Throttle::ServiceQueue()||60));
                &$adjustpenalty; # this prevents ignoring accept when we shouldn't
                my $fdset;
                if ($penalty <= $maxspawn) {
                        $fdset = $fdset_both;
                } else {
                        $fdset = $fdset_pipe;
                        $wait = $penalty - $maxspawn if $wait > $penalty - $maxspawn;
                }
                $nfound = select($rout=$fdset, undef, $eout=$fdset, $wait);
                logmsg("select failed: $!"), exit(1) unless $nfound >= 0 || $!{EINTR} || $!{EAGAIN};
                my $reaped;
                Throttle::RemoveSupplicant($reaped) while ($reaped = shift(@reapedpids));
                $now = time;
                &$adjustpenalty; # this prevents banking credits for elapsed time
                if ($idle_timeout && !$children && !$nfound && $now - $idlestart >= $idle_timeout) {
                        statmsg "idle timeout (@{[duration($idle_timeout)]}) exceeded now exiting";
                        exit 0;
                }
        } while $nfound < 1;
        my $reout = $rout | $eout;
        if (vec($reout, fileno($piperead), 1)) {{
                my $nloff = -1;
                {
                        my $bytes;
                        do {$bytes = sysread($piperead, $pipebuff, 512, length($pipebuff))}
                                while (!defined($bytes) && $!{EINTR});
                        last if !defined($bytes) && $!{EAGAIN};
                        die "sysread failed: $!" unless defined $bytes;
                        # since we always keep a copy of $pipewrite open EOF is fatal
                        die "sysread returned EOF on pipe read" unless $bytes;
                        $nloff = index($pipebuff, "\n", 0);
                        if ($nloff < 0 && length($pipebuff) >= 512) {
                                $pipebuff = '';
                                print STDERR "discarding 512 bytes of control pipe data with no \\n found\n";
                        }
                        redo unless $nloff >= 0;
                }
                last unless $nloff >= 0;
                do {
                        my $msg = substr($pipebuff, 0, $nloff);
                        substr($pipebuff, 0, $nloff + 1) = '';
                        $nloff = index($pipebuff, "\n", 0);
                        process_pipe_msg($msg) if length($msg);
                } while $nloff >= 0;
                redo;
        }}
        next unless vec($reout, fileno(Server), 1);
        unless (accept(Client, Server)) {
                logmsg "accept failed: $!" unless $!{EINTR};
                next;
        }
        logmsg "connection on $NAME";
        ++$penalty;
        spawn sub {
                my $inp = <STDIN>;
                $inp = <STDIN> if defined($inp) && $inp eq "\n";
                chomp $inp if defined($inp);
                $inp or exit 0; # ignore empty connects
                my ($cmd, $arg) = $inp =~ /^([a-zA-Z][a-zA-Z0-9._+-]*)(?:\s+(.*))?$/;
                defined($arg) or $arg = '';
                if ($cmd eq 'ref-changes') {
                        ref_changes($arg);
                } elsif ($cmd eq 'clone') {
                        clone($arg);
                } elsif ($cmd eq 'ref-change') {
                        ref_change($arg);
                } elsif ($cmd eq 'throttle') {
                        throttle($arg);
                } else {
                        die "ignoring unknown command: $cmd\n";
                }
        };
        close Client;
}

#
## -------------
## Documentation
## -------------
#

__END__

=head1 NAME

taskd.pl - Perform Girocco service tasks

=head1 SYNOPSIS

taskd.pl [options]

 Options:
   -h | --help                           detailed instructions
   -q | --quiet                          run quietly
   --no-quiet                            do not run quietly
   -P | --progress                       show occasional status updates
   -i | --inetd                          run as inetd unix stream wait service
                                         implies --quiet --syslog
   -t SECONDS | --idle-timeout=SECONDS   how long to wait idle before exiting
                                         requires --inetd
   -s | --syslog[=facility]              send messages to syslog instead of
                                         stderr but see --stderr
                                         enabled by --inetd
   --no-syslog                           do not send message to syslog
   --stderr                              always send messages to stderr too
   --abbrev=n                            abbreviate hashes to n (default is 8)
   --show-fast-forward-info              show fast-forward info (default is on)
   --no-show-fast-forward-info           disable showing fast-forward info
   --status-interval=MINUTES             status update interval (default 1)
   --idle-status-interval=IDLEMINUTES    idle status interval (default 60)

=head1 OPTIONS

=over 8

=item B<--help>

Print the full description of taskd.pl's options.

=item B<--quiet>

Suppress non-error messages, e.g. for use when running this task as an inetd
service.  Enabled by default by --inetd.

=item B<--no-quiet>

Enable non-error messages.  When running in --inetd mode these messages are
sent to STDERR instead of STDOUT.

=item B<--progress>

Show information about the current status of the task operation occasionally.
This is automatically enabled if --quiet is not given.

=item B<--inetd>

Run as an inetd wait service.  File descriptor 0 must be an unconnected unix
stream socket ready to have accept called on it.  To be useful, the unix socket
should be located at "$Girocco::Config::chroot/etc/taskd.socket".  A warning
will be issued if the socket is not in the expected location.  Socket file
permissions will be adjusted if necessary and if they cannot be taskd.pl will
die.  The --inetd option also enables the --quiet and --syslog options but
--no-quiet and --no-syslog may be used to alter that.

The correct specification for the inetd socket is a "unix" protocol "stream"
socket in "wait" mode with user and group writable permissions (0660).  An
attempt will be made to alter the socket's file mode if needed and if that
cannot be accomplished taskd.pl will die.

Although most inetd stream services run in nowait mode, taskd.pl MUST be run
in wait mode and will die if the passed in socket is already connected.

Note that while *BSD's inetd happily supports unix sockets (and so does
Darwin's launchd), neither xinetd nor GNU's inetd supports unix sockets.
However, systemd does seem to.

=item B<--idle-timeout=SECONDS>

Only permitted when running in --inetd mode.  After SECONDS of inactivity
(i.e. all outstanding tasks have completed and no new requests have come in)
exit normally.  The default is no timeout at all (a SECONDS value of 0).
Note that it may actually take up to SECONDS+60 for the idle exit to occur.

=item B<--syslog[=facility]>

Normally error output is sent to STDERR.  With this option it's sent to
syslog instead.  Note that when running in --inetd mode non-error output is
also affected by this option as it's sent to STDERR in that case.  If
not specified, the default for facility is LOG_USER.  Facility names are
case-insensitive and the leading 'LOG_' is optional.  Messages are logged
with the LOG_NOTICE priority.

=item B<--no-syslog>

Send error message output to STDERR but not syslog.

=item B<--stderr>

Always send error message output to STDERR.  If --syslog is in effect then
a copy will also be sent to syslog.  In --inetd mode this applies to non-error
messages as well.

=item B<--abbrev=n>

Abbreviate displayed hash values to only the first n hexadecimal characters.
The default is 8 characters.  Set to 0 for no abbreviation at all.

=item B<--show-fast-forward-info>

Instead of showing ' -> ' in ref-change/ref-notify update messages, show either
'..' for a fast-forward, creation or deletion or '...' for non-fast-forward.
This requires running an extra git command for each ref update that is not a
creation or deletion in order to determine whether or not it's a fast forward.

=item B<--no-show-fast-forward-info>

Disable showing of fast-forward information for ref-change/ref-notify update
messages.  Instead just show a ' -> ' indicator.

=item B<--status-interval=MINUTES>

If progress is enabled (with --progress or by default if no --inetd or --quiet)
status updates are shown at each MINUTES interval.  Setting the interval to 0
disables them entirely even with --progress.

=item B<--idle-status-interval=IDLEMINUTES>

Two consecutive "idle" status updates with no intervening activity will not be
shown unless IDLEMINUTES have elapsed between them.  The default is 60 minutes.
Setting the interval to 0 prevents any consecutive idle updates (with no
activity between them) from appearing at all.

=back

=head1 DESCRIPTION

taskd.pl is Girocco's service request servant; it listens for service requests
such as new clone requests and ref update notifications and spawns a task to
perform the requested action.

=cut