taskd/taskd.pl

   1 #!/usr/bin/perl
   2 #
   3 # taskd - Clone repositories on request
   4 #
   5 # taskd is Girocco mirroring servant; it processes requests for clones
   6 # of given URLs received over its socket.
   7 #
   8 # When a request is received, new process is spawned that sets up
   9 # the repository and reports further progress
  10 # to .clonelog within the repository. In case the clone fails,
  11 # .clone_failed is touched and .clone_in_progress is removed.
  12
  13 # Clone protocol:
  14 # Alice sets up repository and touches .cloning
  15 # Alice opens connection to Bob
  16 # Alice sends project name through the connection
  17 # Bob opens the repository and sends error code if there is a problem
  18 # Bob closes connection
  19 # Alice polls .clonelog in case of success.
  20 # If Alice reads "@OVER@" from .clonelog, it stops polling.
  21
  22 # Ref-change protocol:
  23 # Alice opens connection to Bob
  24 # Alice sends ref-change command for each changed ref
  25 # Alice closes connection
  26 # Bob sends out notifications
  27
  28 # Initially based on perlipc example.
  29
  30 use 5.008; # we need safe signals
  31 use strict;
  32 use warnings;
  33
  34 use Getopt::Long;
  35 use Pod::Usage;
  36 use Socket;
  37 use Errno;
  38 use Fcntl;
  39 use POSIX ":sys_wait_h";
  40 use File::Basename;
  41
  42 use lib dirname($0);
  43 use Girocco::Config;
  44 use Girocco::Notify;
  45 use Girocco::Project;
  46 use Girocco::User;
  47 use Girocco::Util qw(noFatalsToBrowser get_git);
  48 BEGIN {noFatalsToBrowser}
  49
  50 # Throttle Classes Defaults
  51 # Note that any same-named classes in @Girocco::Config::throttle_classes
  52 # will override (completely replacing the entire hash) these ones.
  53 my @throttle_defaults = (
  54         {
  55                 name => "ref-change",
  56                 maxproc => 0,
  57                 maxjobs => 1,
  58                 interval => 1
  59         },
  60         {
  61                 name => "clone",
  62                 maxproc => 0,
  63                 maxjobs => 2,
  64                 interval => 5
  65         },
  66         {
  67                 name => "snapshot",
  68                 #maxproc => max(5, cpucount + maxjobs), # this is the default
  69                 #maxjobs => max(1, int(cpucount / 4)) , # this is the default
  70                 interval => 5
  71         }
  72 );
  73
  74 # Options
  75 my $quiet;
  76 my $progress;
  77 my $syslog;
  78 my $stderr;
  79 my $inetd;
  80 my $idle_timeout;
  81 my $abbrev = 8;
  82 my $showff = 1;
  83 my $statusintv = 60;
  84 my $idleintv = 3600;
  85 my $maxspawn = 8;
  86
  87 $| = 1;
  88
  89 my $progname = basename($0);
  90 my $children = 0;
  91 my $idlestart = time;
  92 my $idlestatus = 0;
  93
  94 sub cpucount {
  95         use Girocco::Util "online_cpus";
  96         our $online_cpus_result;
  97         $online_cpus_result = online_cpus unless $online_cpus_result;
  98         return $online_cpus_result;
  99 }
 100
 101 sub logmsg {
 102         my $hdr = "[@{[scalar localtime]}] $progname $$: ";
 103         if (tied *STDOUT) {
 104                 $OStream::only = 2; # STDERR only
 105                 print "$hdr@_\n";
 106                 $OStream::only = 1; # syslog only
 107                 print "@_\n";
 108                 $OStream::only = 0; # back to default
 109         } else {
 110                 print "$hdr@_\n";
 111         }
 112 }
 113
 114 sub statmsg {
 115         return unless $progress;
 116         my $hdr = "[@{[scalar localtime]}] $progname $$: ";
 117         if (tied *STDERR) {
 118                 $OStream::only = 2; # STDERR only
 119                 print STDERR "$hdr@_\n";
 120                 $OStream::only = 1; # syslog only
 121                 print STDERR "@_\n";
 122                 $OStream::only = 0; # back to default
 123         } else {
 124                 print STDERR "$hdr@_\n";
 125         }
 126 }
 127
 128 sub duration {
 129         my $secs = shift;
 130         return $secs unless defined($secs) && $secs >= 0;
 131         $secs = int($secs);
 132         my $ans = ($secs % 60) . 's';
 133         return $ans if $secs < 60;
 134         $secs = int($secs / 60);
 135         $ans = ($secs % 60) . 'm' . $ans;
 136         return $ans if $secs < 60;
 137         $secs = int($secs / 60);
 138         $ans = ($secs % 24) . 'h' . $ans;
 139         return $ans if $secs < 24;
 140         $secs = int($secs / 24);
 141         return $secs . 'd' . $ans;
 142 }
 143
 144 sub setnonblock {
 145         my $fd = shift;
 146         my $flags = fcntl($fd, F_GETFL, 0);
 147         defined($flags) or die "fcntl failed: $!";
 148         fcntl($fd, F_SETFL, $flags | O_NONBLOCK) or die "fcntl failed: $!";
 149 }
 150
 151 sub setblock {
 152         my $fd = shift;
 153         my $flags = fcntl($fd, F_GETFL, 0);
 154         defined($flags) or die "fcntl failed: $!";
 155         fcntl($fd, F_SETFL, $flags & ~O_NONBLOCK) or die "fcntl failed: $!";
 156 }
 157
 158 package Throttle;
 159
 160 #
 161 ## Throttle protocol
 162 ##
 163 ## 1) Process needing throttle services acquire a control file descriptor
 164 ##    a) Either as a result of a fork + exec (the write end of a pipe)
 165 ##    b) Or by connecting to the taskd socket (not yet implemented)
 166 ##
 167 ## 2) The process requesting throttle services will be referred to
 168 ##    as the supplicant or just "supp" for short.
 169 ##
 170 ## 3) The supp first completes any needed setup which may include
 171 ##    gathering data it needs to perform the action -- if that fails
 172 ##    then there's no need for any throttling.
 173 ##
 174 ## 4) The supp writes a throttle request to the control descriptor in
 175 ##    this format:
 176 ##        throttle <pid> <class>\n
 177 ##    for example if the supp's pid was 1234 and it was requesting throttle
 178 ##    control as a member of the mail class it would write this message:
 179 ##        throttle 1234 mail\n
 180 ##    Note that if the control descriptor happens to be a pipe rather than a
 181 ##    socket, the message should be preceded by another "\n" just be be safe.
 182 ##    If the control descriptor is a socket, not a pipe, the message may be
 183 ##    preceded by a "\n" but that's not recommended.
 184 ##
 185 ## 5) For supplicants with a control descriptor that is a pipe
 186 ##    (getsockopt(SO_TYPE) returns ENOTSOCK) the (5a) protocol should be used.
 187 ##    If the control descriptor is a socket (getsockname succeeds) then
 188 ##    protocol (5b) should be used.
 189 ##
 190 ## 5a) The supp now enters a "pause" loop awaiting either a SIGUSR1, SIGUSR2 or
 191 ##    SIGTERM.  It should wake up periodically (SIGALRM works well) and attempt
 192 ##    to write a "keepalive\n" message to the control descriptor.  If that
 193 ##    fails, the controller has gone away and it may make its own decision
 194 ##    whether or not to proceed at that point.  If, on the other hand, it
 195 ##    receives a SIGTERM, the process limit for its class has been reached
 196 ##    and it should abort without performing its action.  If it receives
 197 ##    SIGUSR1, it may proceed without writing anything more to the control
 198 ##    descriptor, any MAY even close the control descriptor.  Finally, a
 199 ##    SIGUSR2 indicates rejection of the throttle request for some other reason
 200 ##    such as unrecognized class name or invalid pid in which case the supp may
 201 ##    make its own decision how to proceed.
 202 ##
 203 ## 5b) The supp now enters a read wait on the socket -- it need accomodate no
 204 ##    more than 512 bytes and if a '\n' does not appear within that number of
 205 ##    bytes the read should be considered failed.  Otherwise the read should
 206 ##    be retried until either a full line has been read or the socket is
 207 ##    closed from the other end.  If the lone read is "proceed\n" then it may
 208 ##    proceed without reading or writing anything more to the control
 209 ##    descriptor, but MUST keep the control descriptor open and not call
 210 ##    shutdown on it either.  Any other result (except EINTR or EAGAIN which
 211 ##    should be retried) constitutes failure.  If a full line starting with at
 212 ##    least one alpha character was read but it was not "proceed" then it
 213 ##    should abort without performing its action.  For any other failure it
 214 ##    may make its own decision whether or not to proceed as the controller has
 215 ##    gone away.
 216 ##
 217 ## 6) The supp now performs its throttled action.
 218 ##
 219 ## 7) The supp now closes its control descriptor (if it hasn't already in the
 220 ##    case of (5a)) and exits -- in the case of a socket, the other end receives
 221 ##    notification that the socket has been closed (read EOF).  In the case of
 222 ##    a pipe the other end receives a SIGCHLD (multiple processes have a hold
 223 ##    of the other end of the pipe, so it will not reaach EOF by the supp's
 224 ##    exit in that case).
 225 #
 226
 227 # keys are class names, values are hash refs with these fields:
 228 # 'maxproc' => integer; maximum number of allowed supplicants (the sum of how
 229 #                       many may be queued waiting plus how many may be
 230 #                       concurrently active) with 0 meaning no limit.
 231 # 'maxjobs' => integer; how many supplicants may proceed simultaneously a value
 232 #                       of 0 is unlimited but the number of concurrent
 233 #                       supplicants will always be limited to no more than
 234 #                       the 'maxproc' value (if > 0) no matter what the
 235 #                       'maxjobs' value is.
 236 # 'total'   -> integer; the total number of pids belonging to this clase that
 237 #                       can currently be found in %pid.
 238 # 'active'  -> integer; the number of currently active supplicants which should
 239 #                       be the same as (the number of elements of %pid with a
 240 #                       matching class name) - (number of my class in @queue).
 241 # 'interval' -> integer; minimum number of seconds between 'proceed' responses
 242 #                       or SIGUSR1 signals to members of this class.
 243 # 'lastqueue' -> time;  last time a supplicant was successfully queued.
 244 # 'lastproceed' => time; last time a supplicant was allowed to proceed.
 245 # 'lastthrottle' => time; last time a supplicant was throttled
 246 # 'lastdied'  => time;  last time a supplicant in this class died/exited/etc.
 247 my %classes = ();
 248
 249 # keys are pid numbers, values are array refs with these elements:
 250 # [0] => name of class (key to classes hash)
 251 # [1] => supplicant state (0 => queued, non-zero => time it started running)
 252 # [2] => descriptive text (e.g. project name)
 253 my %pid = ();
 254
 255 # minimum number of seconds between any two proceed responses no matter what
 256 # class.  this takes priority in that it can effectively increase the
 257 # class's 'interval' value by delaying proceed notifications if the minimum
 258 # interval has not yet elapsed.
 259 my $interval = 1;
 260
 261 # fifo of pids awaiting notification as soon as the next $interval elapses
 262 # provided interval and maxjobs requirements are satisfied
 263 # for the class of the pid that will next be triggered.
 264 my @queue = ();
 265
 266 # time of most recent successful call to AddSupplicant
 267 my $lastqueue = 0;
 268
 269 # time of most recent proceed notification
 270 my $lastproceed = 0;
 271
 272 # time of most recent throttle
 273 my $lastthrottle = 0;
 274
 275 # time of most recent removal
 276 my $lastdied = 0;
 277
 278 # lifetime count of how many have been queued
 279 my $totalqueue = 0;
 280
 281 # lifetime count of how many have been allowed to proceed
 282 my $totalproceed = 0;
 283
 284 # lifetime count of how many have been throttled
 285 my $totalthrottle = 0;
 286
 287 # lifetime count of how many have died
 288 # It should always be true that $totalqueued - $totaldied == $curentlyactive
 289 my $totaldied = 0;
 290
 291 # Returns an unordered list of currently registered class names
 292 sub GetClassList {
 293         return keys(%classes);
 294 }
 295
 296 sub _max {
 297         return $_[0] if $_[0] >= $_[1];
 298         return $_[1];
 299 }
 300
 301 sub _getnum {
 302         my ($min, $val, $default) = @_;
 303         my $ans;
 304         if (defined($val) && $val =~ /^[+-]?\d+$/) {
 305                 $ans = 0 + $val;
 306         } else {
 307                 $ans = &$default;
 308         }
 309         return _max($min, $ans);
 310 }
 311
 312 # [0] => name of class to find
 313 # [1] => if true, create class if it doesn't exist, if a hashref then
 314 #        it contains initial values for maxproc, maxjobs and interval.
 315 #        Otherwise maxjobs defaults to max(cpu cores/4, 1), maxprocs
 316 #        defaults to the max(5, number of cpu cores + maxjobs) and interval
 317 #        defaults to 1.
 318 # Returns a hash ref with info about the class on success
 319 sub GetClassInfo {
 320         my ($classname, $init) = @_;
 321         defined($classname) && $classname =~ /^[a-zA-Z][a-zA-Z0-9._+-]*$/
 322                 or return;
 323         $classname = lc($classname);
 324         my %info;
 325         if ($classes{$classname}) {
 326                 %info = %{$classes{$classname}};
 327                 return \%info;
 328         }
 329         return unless $init;
 330         my %newclass = ();
 331         ref($init) eq 'HASH' or $init = {};
 332         $newclass{'maxjobs'} = _getnum(0, $init->{'maxjobs'}, sub{_max(1, int(::cpucount() / 4))});
 333         $newclass{'maxproc'} = _getnum(0, $init->{'maxproc'}, sub{_max(5, ::cpucount() + $newclass{'maxjobs'})});
 334         $newclass{'interval'} = _getnum(0, $init->{'interval'}, sub{1});
 335         $newclass{'total'} = 0;
 336         $newclass{'active'} = 0;
 337         $newclass{'lastqueue'} = 0;
 338         $newclass{'lastproceed'} = 0;
 339         $newclass{'lastthrottle'} = 0;
 340         $newclass{'lastdied'} = 0;
 341         $classes{$classname} = \%newclass;
 342         %info = %newclass;
 343         return \%info;
 344 }
 345
 346 # [0] => pid to look up
 347 # Returns () if not found otherwise ($classname, $timestarted, $description)
 348 # Where $timestarted will be 0 if it's still queued otherwise a time() value
 349 sub GetPidInfo {
 350         my $pid = shift;
 351         return () unless exists $pid{$pid};
 352         return @{$pid{$pid}};
 353 }
 354
 355 # Returns array of pid numbers that are currently running sorted
 356 # by time started (oldest to newest).  Can return an empty array.
 357 sub GetRunningPids {
 358         return sort({ ${$pid{$a}}[1] <=> ${$pid{$b}}[1] }
 359                 grep({ ${$pid{$_}}[1] } keys(%pid)));
 360 }
 361
 362 # Returns a hash with various about the current state
 363 # 'interval'      => global minimum interval between proceeds
 364 # 'active'        => how many pids are currently queued + how many are running
 365 # 'queue'         => how many pids are currently queued
 366 # 'lastqueue'     => time (epoch seconds) of last queue
 367 # 'lastproceed'   => time (epoch seconds) of last proceed
 368 # 'lastthrottle'  => time (epoch seconds) of last throttle
 369 # 'lastdied'      => time (epoch seconds) of last removal
 370 # 'totalqueue'    => lifetime total number of processes queued
 371 # 'totalproceed'  => lifetime total number of processes proceeded
 372 # 'totalthrottle' => lifetime total number of processes throttled
 373 # 'totaldied'     => lifetime total number of removed processes
 374 sub GetInfo {
 375         return {
 376                 interval => $interval,
 377                 active => scalar(keys(%pid)) - scalar(@queue),
 378                 queue => scalar(@queue),
 379                 lastqueue => $lastqueue,
 380                 lastproceed => $lastproceed,
 381                 lastthrottle => $lastthrottle,
 382                 lastdied => $lastdied,
 383                 totalqueue => $totalqueue,
 384                 totalproceed => $totalproceed,
 385                 totalthrottle => $totalthrottle,
 386                 totaldied => $totaldied
 387         };
 388 }
 389
 390 # with no args get the global interval
 391 # with one arg set it, returns previous value if set
 392 sub Interval {
 393         my $ans = $interval;
 394         $interval = 0 + $_[0] if defined($_[0]) && $_[0] =~ /^\d+$/;
 395         return $ans;
 396 }
 397
 398 sub RemoveSupplicant;
 399
 400 # Perform queue service (i.e. send SIGUSR1 to any eligible queued process)
 401 # Returns minimum interval until next proceed is possible
 402 # Returns undef if there's nothing waiting to proceed or
 403 # the 'maxjobs' limits have been reached for all queued items (in which
 404 # case it won't be possible to proceed until one of them exits, hence undef)
 405 # This is called automatially by AddSupplicant and RemoveSupplicant
 406 sub ServiceQueue {
 407         RETRY:
 408         return undef unless @queue; # if there's nothing queued, nothing to do
 409         my $now = time;
 410         my $min = _max(0, $interval - ($now - $lastproceed));
 411         my $classmin = undef;
 412         my $classchecked = 0;
 413         my %seenclass = ();
 414         my $classcount = scalar(keys(%classes));
 415         for (my $i=0; $i <= $#queue && $classchecked < $classcount; ++$i) {
 416                 my $pid = $queue[$i];
 417                 my $procinfo = $pid{$pid};
 418                 if (!$procinfo) {
 419                         RemoveSupplicant($pid, 1);
 420                         goto RETRY;
 421                 }
 422                 my $classinfo = $classes{$$procinfo[0]};
 423                 if (!$classinfo) {
 424                         RemoveSupplicant($pid, 1);
 425                         goto RETRY;
 426                 }
 427                 if (!$seenclass{$$procinfo[0]}) {
 428                         $seenclass{$$procinfo[0]} = 1;
 429                         ++$classchecked;
 430                         if (!$classinfo->{'maxjobs'} || $classinfo->{'active'} < $classinfo->{'maxjobs'}) {
 431                                 my $cmin = _max(0, $classinfo->{'interval'} - ($now - $classinfo->{'lastproceed'}));
 432                                 if (!$cmin && !$min) {
 433                                         $now = time;
 434                                         $$procinfo[1] = $now;
 435                                         splice(@queue, $i, 1);
 436                                         ++$totalproceed;
 437                                         $lastproceed = $now;
 438                                         $classinfo->{'lastproceed'} = $now;
 439                                         ++$classinfo->{'active'};
 440                                         kill("USR1", $pid) or RemoveSupplicant($pid, 1);
 441                                         goto RETRY;
 442                                 }
 443                                 $classmin = $cmin unless defined($classmin) && $classmin < $cmin;
 444                         }
 445                 }
 446         }
 447         return defined($classmin) ? _max($min, $classmin) : undef;
 448 }
 449
 450 # $1 => pid to add (must not already be in %pids)
 451 # $2 => class name (must exist)
 452 # Returns -1 if no such class or pid already present or invalid
 453 # Returns 0 if added successfully (and possibly already SIGUSR1'd)
 454 # Return 1 if throttled and cannot be added
 455 sub AddSupplicant {
 456         my ($pid, $classname, $text, $noservice) = @_;
 457         return -1 unless $pid && $pid =~ /^[1-9][0-9]*$/;
 458         $pid += 0;
 459         kill(0, $pid) or return -1;
 460         my $classinfo = $classes{$classname};
 461         return -1 unless $classinfo;
 462         return -1 if $pid{$pid};
 463         $text = '' unless defined($text);
 464         my $now = time;
 465         if ($classinfo->{'maxproc'} && $classinfo->{'total'} >= $classinfo->{'maxproc'}) {
 466                 ++$totalthrottle;
 467                 $lastthrottle = $now;
 468                 $classinfo->{'lastthrottle'} = $now;
 469                 return 1;
 470         }
 471         ++$totalqueue;
 472         $lastqueue = $now;
 473         $pid{$pid} = [$classname, 0, $text];
 474         ++$classinfo->{'total'};
 475         $classinfo->{'lastqueue'} = $now;
 476         push(@queue, $pid);
 477         ServiceQueue unless $noservice;
 478         return 0;
 479 }
 480
 481 # $1 => pid to remove (died, killed, exited normally, doesn't matter)
 482 # Returns 0 if removed
 483 # Returns -1 if unknown pid or other error during removal
 484 sub RemoveSupplicant {
 485         my ($pid, $noservice) = @_;
 486         return -1 unless defined($pid) && $pid =~ /^\d+$/;
 487         $pid += 0;
 488         my $pidinfo = $pid{$pid};
 489         $pidinfo or return -1;
 490         my $now = time;
 491         $lastdied = $now;
 492         ++$totaldied;
 493         delete $pid{$pid};
 494         if (!$$pidinfo[1]) {
 495                 for (my $i=0; $i<=$#queue; ++$i) {
 496                         if ($queue[$i] == $pid) {
 497                                 splice(@queue, $i, 1);
 498                                 --$i;
 499                         }
 500                 }
 501         }
 502         my $classinfo = $classes{$$pidinfo[0]};
 503         ServiceQueue, return -1 unless $classinfo;
 504         --$classinfo->{'active'} if $$pidinfo[1];
 505         --$classinfo->{'total'};
 506         $classinfo->{'lastdied'} = $now;
 507         ServiceQueue unless $noservice;
 508         return 0;
 509 }
 510
 511 # Instance Methods
 512
 513 package main;
 514
 515 #
 516 ## ---------
 517 ## Functions
 518 ## ---------
 519 #
 520
 521 my @reapedpids = ();
 522 my %signame = (
 523          # http://pubs.opengroup.org/onlinepubs/000095399/utilities/trap.html
 524          1 => 'SIGHUP',
 525          2 => 'SIGINT',
 526          3 => 'SIGQUIT',
 527          6 => 'SIGABRT',
 528          9 => 'SIGKILL',
 529         14 => 'SIGALRM',
 530         15 => 'SIGTERM',
 531 );
 532 sub REAPER {
 533         local $!;
 534         my $child;
 535         my $waitedpid;
 536         while (($waitedpid = waitpid(-1, WNOHANG)) > 0) {
 537                 my $code = $? & 0xffff;
 538                 $idlestart = time if !--$children;
 539                 my $codemsg = '';
 540                 if (!($code & 0xff)) {
 541                         $codemsg = " with exit code ".($code >> 8) if $code;
 542                 } elsif ($code & 0x7f) {
 543                         my $signum = ($code & 0x7f);
 544                         $codemsg = " with signal ".
 545                                 ($signame{$signum}?$signame{$signum}:$signum);
 546                 }
 547                 logmsg "reaped $waitedpid$codemsg";
 548                 push(@reapedpids, $waitedpid);
 549         }
 550         $SIG{CHLD} = \&REAPER;  # loathe sysV
 551 }
 552
 553 $SIG{CHLD} = \&REAPER;  # Apollo 440
 554
 555 my ($piperead, $pipewrite);
 556 sub spawn {
 557         my $coderef = shift;
 558
 559         my $pid = fork;
 560         if (not defined $pid) {
 561                 logmsg "cannot fork: $!";
 562                 return;
 563         } elsif ($pid) {
 564                 $idlestart = time if !++$children;
 565                 $idlestatus = 0;
 566                 logmsg "begat $pid";
 567                 return; # I'm the parent
 568         }
 569
 570         close(Server) unless fileno(Server) == 0;
 571         close($piperead);
 572         $SIG{'CHLD'} = sub {};
 573
 574         open STDIN, "+<&Client" or die "can't dup client to stdin";
 575         close(Client);
 576         exit &$coderef();
 577 }
 578
 579 # returns:
 580 #   < 0: error
 581 #   = 0: proceed
 582 #   > 0: throttled
 583 sub request_throttle {
 584         use POSIX qw(sigprocmask sigsuspend SIG_SETMASK);
 585         my $classname = shift;
 586         my $text = shift;
 587
 588         Throttle::GetClassInfo($classname)
 589                 or return -1; # no such throttle class
 590
 591         my $throttled = 0;
 592         my $proceed = 0;
 593         my $error = 0;
 594         my $controldead = 0;
 595         my $setempty = POSIX::SigSet->new;
 596         my $setfull = POSIX::SigSet->new;
 597         $setempty->emptyset();
 598         $setfull->fillset();
 599         $SIG{'TERM'} = sub {$throttled = 1};
 600         $SIG{'USR1'} = sub {$proceed = 1};
 601         $SIG{'USR2'} = sub {$error = 1};
 602         $SIG{'PIPE'} = sub {$controldead = 1};
 603         $SIG{'ALRM'} = sub {};
 604
 605         # After writing we can expect a SIGTERM, SIGUSR1 or SIGUSR2
 606         print $pipewrite "\nthrottle $$ $classname $text\n";
 607         my $old = POSIX::SigSet->new;
 608         sigprocmask(SIG_SETMASK, $setfull, $old);
 609         until ($controldead || $throttled || $proceed || $error) {
 610                 alarm(30);
 611                 sigsuspend($setempty);
 612                 alarm(0);
 613                 sigprocmask(SIG_SETMASK, $setempty, $old);
 614                 print $pipewrite "\nkeepalive $$\n";
 615                 sigprocmask(SIG_SETMASK, $setfull, $old);
 616         }
 617         sigprocmask(SIG_SETMASK, $setempty, $old);
 618         $SIG{'TERM'} = "DEFAULT";
 619         $SIG{'USR1'} = "DEFAULT";
 620         $SIG{'USR2'} = "DEFAULT";
 621         $SIG{'ALRM'} = "DEFAULT";
 622         $SIG{'PIPE'} = "DEFAULT";
 623
 624         my $result = -1;
 625         if ($throttled) {
 626                 $result = 1;
 627         } elsif ($proceed) {
 628                 $result = 0;
 629         }
 630         return $result;
 631 }
 632
 633 sub clone {
 634         my ($name) = @_;
 635         Girocco::Project::does_exist($name, 1) or die "no such project: $name";
 636         my $proj;
 637         eval {$proj = Girocco::Project->load($name)};
 638         if (!$proj && Girocco::Project::does_exist($name, 1)) {
 639                 # If the .clone_in_progress file exists, but the .clonelog does not
 640                 # and neither does the .clone_failed, be helpful and touch the
 641                 # .clone_failed file so that the mirror can be restarted
 642                 my $projdir = $Girocco::Config::reporoot."/$name.git";
 643                 if (-d "$projdir" && -f "$projdir/.clone_in_progress" && ! -f "$projdir/.clonelog" && ! -f "$projdir/.clone_failed") {
 644                         open X, '>', "$projdir/.clone_failed" and close(X);
 645                 }
 646         }
 647         $proj or die "failed to load project $name";
 648         $proj->{clone_in_progress} or die "project $name is not marked for cloning";
 649         $proj->{clone_logged} and die "project $name is already being cloned";
 650         request_throttle("clone", $name) <= 0 or die "cloning $name aborted (throttled)";
 651         statmsg "cloning $name";
 652         open STDOUT, '>', "$Girocco::Config::reporoot/$name.git/.clonelog" or die "cannot open clonelog: $!";
 653         open STDERR, ">&STDOUT";
 654         open STDIN, '<', '/dev/null';
 655         exec "$Girocco::Config::basedir/taskd/clone.sh", "$name.git" or die "exec failed: $!";
 656 }
 657
 658 sub ref_indicator {
 659         return ' -> ' unless $showff && defined($_[0]);
 660         my ($git_dir, $old, $new) = @_;
 661         return '..' unless defined($old) && defined($new) && $old !~ /^0+$/ && $new !~ /^0+$/ && $old ne $new;
 662         # In many cases `git merge-base` is slower than this even if using the
 663         # `--is-ancestor` option available since Git 1.8.0, but it's never faster
 664         my $ans = get_git("--git-dir=$git_dir", "rev-list", "-n", "1", "^$new^0", "$old^0", "--") ? '...' : '..';
 665         return wantarray ? ($ans, 1) : $ans;
 666 }
 667
 668 sub ref_change {
 669         my ($arg) = @_;
 670         my ($username, $name, $oldrev, $newrev, $ref) = split(/\s+/, $arg);
 671         $username && $name && $oldrev && $newrev && $ref or return 0;
 672         $oldrev =~ /^[0-9a-f]{40}$/ && $newrev =~ /^[0-9a-f]{40}$/ && $ref =~ m{^refs/} or return 0;
 673         $newrev ne $oldrev or return 0;
 674
 675         Girocco::Project::does_exist($name, 1) or die "no such project: $name";
 676         my $proj = Girocco::Project->load($name);
 677         $proj or die "failed to load project $name";
 678         my $has_notify = $proj->has_notify;
 679         my $type = $has_notify ? "notify" : "change";
 680
 681         my $user;
 682         if ($username && $username !~ /^%.*%$/) {
 683                 Girocco::User::does_exist($username, 1) or die "no such user: $username";
 684                 $user = Girocco::User->load($username);
 685                 $user or die "failed to load user $username";
 686         } elsif ($username eq "%$name%") {
 687                 $username = "-";
 688         }
 689
 690         request_throttle("ref-change", $name) <= 0 or die "ref-change $name aborted (throttled)";
 691         my $ind = ref_indicator($proj->{path}, $oldrev, $newrev);
 692         statmsg "ref-$type $username $name ($ref: @{[substr($oldrev,0,$abbrev)]}$ind@{[substr($newrev,0,$abbrev)]})";
 693         open STDIN, '<', '/dev/null';
 694         Girocco::Notify::ref_change($proj, $user, $ref, $oldrev, $newrev) if $has_notify;
 695         return 0;
 696 }
 697
 698 sub ref_changes {
 699         my ($arg) = @_;
 700         my ($username, $name) = split(/\s+/, $arg);
 701         $username && $name or return 0;
 702
 703         Girocco::Project::does_exist($name, 1) or die "no such project: $name";
 704         my $proj = Girocco::Project->load($name);
 705         $proj or die "failed to load project $name";
 706         my $has_notify = $proj->has_notify;
 707         my $type = $has_notify ? "notify" : "change";
 708
 709         my $user;
 710         if ($username && $username !~ /^%.*%$/) {
 711                 Girocco::User::does_exist($username, 1) or die "no such user: $username";
 712                 $user = Girocco::User->load($username);
 713                 $user or die "failed to load user $username";
 714         } elsif ($username eq "%$name%") {
 715                 $username = "-";
 716         }
 717
 718         my @changes = ();
 719         my %oldheads = ();
 720         my %deletedheads = ();
 721         while (my $change = <STDIN>) {
 722                 my ($oldrev, $newrev, $ref) = split(/\s+/, $change);
 723                 $oldrev =~ /^[0-9a-f]{40}$/ && $newrev =~ /^[0-9a-f]{40}$/ && $ref =~ m{^refs/} or next;
 724                 if ($ref =~ m{^refs/heads/.}) {
 725                         if ($oldrev =~ /^0{40}$/) {
 726                                 delete $oldheads{$ref};
 727                                 $deletedheads{$ref} = 1;
 728                         } elsif ($newrev ne $oldrev || (!exists($oldheads{$ref}) && !$deletedheads{$ref})) {
 729                                 $oldheads{$ref} = $oldrev;
 730                         }
 731                 }
 732                 $newrev ne $oldrev or next;
 733                 push(@changes, [$oldrev, $newrev, $ref]);
 734         }
 735         return 0 unless @changes;
 736         open STDIN, '<', '/dev/null';
 737         request_throttle("ref-change", $name) <= 0 or die "ref-changes $name aborted (throttled)";
 738         my $statproc = sub {
 739                 my ($old, $new, $ref, $ran_mail_sh) = @_;
 740                 my ($ind, $ran_git) = ref_indicator($proj->{path}, $old, $new);
 741                 statmsg "ref-$type $username $name ($ref: @{[substr($old,0,$abbrev)]}$ind@{[substr($new,0,$abbrev)]})";
 742                 sleep 1 if $ran_mail_sh || $ran_git;
 743         };
 744         if ($has_notify) {
 745                 Girocco::Notify::ref_changes($proj, $user, $statproc, \%oldheads, @changes);
 746         } else {
 747                 &$statproc(@$_) foreach @changes;
 748         }
 749         return 0;
 750 }
 751
 752 sub throttle {
 753         my ($arg) = @_;
 754         my ($pid, $classname, $text) = split(/\s+/, $arg);
 755         $pid =~ /^\d+/ or return 0; # invalid pid
 756         $pid += 0;
 757         $pid > 0 or return 0; # invalid pid
 758         kill(0, $pid) || $!{EPERM} or return 0; # no such process
 759         Throttle::GetClassInfo($classname) or return 0; # no such throttle class
 760         defined($text) && $text ne '' or return 0; # no text no service
 761
 762         my $throttled = 0;
 763         my $proceed = 0;
 764         my $error = 0;
 765         my $controldead = 0;
 766         my $suppdead = 0;
 767         my ($waker, $wakew);
 768         pipe($waker, $wakew) or die "pipe failed: $!";
 769         select((select($wakew),$|=1)[0]);
 770         setnonblock($wakew);
 771         $SIG{'TERM'} = sub {$throttled = 1; syswrite($wakew, '!')};
 772         $SIG{'USR1'} = sub {$proceed = 1; syswrite($wakew, '!')};
 773         $SIG{'USR2'} = sub {$error = 1; syswrite($wakew, '!')};
 774         $SIG{'PIPE'} = sub {$controldead = 1; syswrite($wakew, '!')};
 775         select((select(STDIN),$|=1)[0]);
 776
 777         logmsg "throttle $pid $classname $text request";
 778         # After writing we can expect a SIGTERM or SIGUSR1
 779         print $pipewrite "\nthrottle $$ $classname $text\n";
 780
 781         # NOTE: the only way to detect the socket close is to read all the
 782         #       data until EOF is reached -- recv can be used to peek.
 783         my $v = '';
 784         vec($v, fileno(STDIN), 1) = 1;
 785         vec($v, fileno($waker), 1) = 1;
 786         setnonblock(\*STDIN);
 787         setnonblock($waker);
 788         until ($controldead || $throttled || $proceed || $error || $suppdead) {
 789                 my ($r, $e);
 790                 select($r=$v, undef, $e=$v, 30);
 791                 my ($bytes, $discard);
 792                 do {$bytes = sysread($waker, $discard, 512)} while (defined($bytes) && $bytes > 0);
 793                 do {$bytes = sysread(STDIN, $discard, 4096)} while (defined($bytes) && $bytes > 0);
 794                 $suppdead = 1 unless !defined($bytes) && $!{EAGAIN};
 795                 print $pipewrite "\nkeepalive $$\n";
 796         }
 797         setblock(\*STDIN);
 798
 799         if ($throttled && !$suppdead) {
 800                 print STDIN "throttled\n";
 801                 logmsg "throttle $pid $classname $text throttled";
 802         } elsif ($proceed && !$suppdead) {
 803                 print STDIN "proceed\n";
 804                 logmsg "throttle $pid $classname $text proceed";
 805                 $SIG{'TERM'} = 'DEFAULT';
 806                 # Stay alive until the child dies which we detect by EOF on STDIN
 807                 setnonblock(\*STDIN);
 808                 until ($controldead || $suppdead) {
 809                         my ($r, $e);
 810                         select($r=$v, undef, $e=$v, 30);
 811                         my ($bytes, $discard);
 812                         do {$bytes = sysread($waker, $discard, 512)} while (defined($bytes) && $bytes > 0);
 813                         do {$bytes = sysread(STDIN, $discard, 512)} while (defined($bytes) && $bytes > 0);
 814                         $suppdead = 1 unless !defined($bytes) && $!{EAGAIN};
 815                         print $pipewrite "\nkeepalive $$\n";
 816                 }
 817                 setblock(\*STDIN);
 818         } else {
 819                 my $prefix = '';
 820                 $prefix = "control" if $controldead && !$suppdead;
 821                 logmsg "throttle $pid $classname $text ${prefix}died";
 822         }
 823         exit 0;
 824 }
 825
 826 sub process_pipe_msg {
 827         my ($act, $pid, $cls, $text) = split(/\s+/, $_[0]);
 828         if ($act eq "throttle") {
 829                 $pid =~ /^\d+$/ or return 0;
 830                 $pid += 0;
 831                 $pid > 0 or return 0; # invalid pid
 832                 kill(0, $pid) or return 0; # invalid pid
 833                 defined($cls) && $cls ne "" or kill('USR2', $pid), return 0;
 834                 defined($text) && $text ne "" or kill('USR2', $pid), return 0;
 835                 Throttle::GetClassInfo($cls) or kill('USR2', $pid), return 0;
 836                 # the AddSupplicant call could send SIGUSR1 before it returns
 837                 my $result = Throttle::AddSupplicant($pid, $cls, $text);
 838                 kill('USR2', $pid), return 0 if $result < 0;
 839                 kill('TERM', $pid), return 0 if $result > 0;
 840                 # $pid was added to class $cls and will receive SIGUSR1 when
 841                 # it's time for it to proceed
 842                 return 0;
 843         } elsif ($act eq "keepalive") {
 844                 # nothing to do although we could verify pid is valid and
 845                 # still in %Throttle::pids and send a SIGUSR2 if not, but
 846                 # really keepalive should just be ignored.
 847                 return 0;
 848         }
 849         print STDERR "discarding unknown pipe message \"$_[0]\"\n";
 850         return 0;
 851 }
 852
 853 #
 854 ## -------
 855 ## OStream
 856 ## -------
 857 #
 858
 859 package OStream;
 860
 861 # Set to 1 for only syslog output (if enabled by mode)
 862 # Set to 2 for only stderr output (if enabled by mode)
 863 our $only = 0; # This is a hack
 864
 865 use Carp 'croak';
 866 use Sys::Syslog qw(:DEFAULT :macros);
 867
 868 sub writeall {
 869         use POSIX qw();
 870         use Errno;
 871         my ($fd, $data) = @_;
 872         my $offset = 0;
 873         my $remaining = length($data);
 874         while ($remaining) {
 875                 my $bytes = POSIX::write(
 876                         $fd,
 877                         substr($data, $offset, $remaining),
 878                         $remaining);
 879                 next if !defined($bytes) && $!{EINTR};
 880                 croak "POSIX::write failed: $!" unless defined $bytes;
 881                 croak "POSIX::write wrote 0 bytes" unless $bytes;
 882                 $remaining -= $bytes;
 883                 $offset += $bytes;
 884         }
 885 }
 886
 887 sub dumpline {
 888         use POSIX qw(STDERR_FILENO);
 889         my ($self, $line) = @_;
 890         $only = 0 unless defined($only);
 891         writeall(STDERR_FILENO, $line) if $self->{'stderr'} && $only != 1;
 892         substr($line, -1, 1) = '' if substr($line, -1, 1) eq "\n";
 893         return unless length($line);
 894         syslog(LOG_NOTICE, "%s", $line) if $self->{'syslog'} && $only != 2;
 895 }
 896
 897 sub TIEHANDLE {
 898         my $class = shift || 'OStream';
 899         my $mode = shift;
 900         my $syslogname = shift;
 901         my $syslogfacility = shift;
 902         defined($syslogfacility) or $syslogfacility = LOG_USER;
 903         my $self = {};
 904         $self->{'syslog'} = $mode > 0;
 905         $self->{'stderr'} = $mode <= 0 || $mode > 1;
 906         $self->{'lastline'} = '';
 907         if ($self->{'syslog'}) {
 908                 # Some Sys::Syslog have a stupid default setlogsock order
 909                 eval {Sys::Syslog::setlogsock("native"); 1;} or
 910                 eval {Sys::Syslog::setlogsock("unix");};
 911                 openlog($syslogname, "ndelay,pid", $syslogfacility)
 912                         or croak "Sys::Syslog::openlog failed: $!";
 913         }
 914         return bless $self, $class;
 915 }
 916
 917 sub BINMODE {return 1}
 918 sub FILENO {return undef}
 919 sub EOF {return 0}
 920 sub CLOSE {return 1}
 921
 922 sub PRINTF {
 923         my $self = shift;
 924         my $template = shift;
 925         return $self->PRINT(sprintf $template, @_);
 926 }
 927
 928 sub PRINT {
 929         my $self = shift;
 930         my $data = join('', $self->{'lastline'}, @_);
 931         my $pos = 0;
 932         while ((my $idx = index($data, "\n", $pos)) >= 0) {
 933                 ++$idx;
 934                 my $line = substr($data, $pos, $idx - $pos);
 935                 substr($data, $pos, $idx - $pos) = '';
 936                 $pos = $idx;
 937                 $self->dumpline($line);
 938         }
 939         $self->{'lastline'} = $data;
 940         return 1;
 941 }
 942
 943 sub DESTROY {
 944         my $self = shift;
 945         $self->dumpline($self->{'lastline'})
 946                 if length($self->{'lastline'});
 947         closelog;
 948 }
 949
 950 sub WRITE {
 951         my $self = shift;
 952         my ($scalar, $length, $offset) = @_;
 953         $scalar = '' if !defined($scalar);
 954         $length = length($scalar) if !defined($length);
 955         croak "OStream::WRITE invalid length $length"
 956                 if $length < 0;
 957         $offset = 0 if !defined($offset);
 958         $offset += length($scalar) if $offset < 0;
 959         croak "OStream::WRITE invalid write offset"
 960                 if $offset < 0 || $offset > $length;
 961         my $max = length($scalar) - $offset;
 962         $length = $max if $length > $max;
 963         $self->PRINT(substr($scalar, $offset, $length));
 964         return $length;
 965 }
 966
 967 #
 968 ## ----
 969 ## main
 970 ## ----
 971 #
 972
 973 package main;
 974
 975 chdir "/";
 976 close(DATA) if fileno(DATA);
 977 my $sfac;
 978 Getopt::Long::Configure('bundling');
 979 my ($stiv, $idiv);
 980 my $parse_res = GetOptions(
 981         'help|?|h' => sub {pod2usage(-verbose => 2, -exitval => 0)},
 982         'quiet|q' => \$quiet,
 983         'no-quiet' => sub {$quiet = 0},
 984         'progress|P' => \$progress,
 985         'inetd|i' => sub {$inetd = 1; $syslog = 1; $quiet = 1;},
 986         'idle-timeout|t=i' => \$idle_timeout,
 987         'syslog|s:s' => \$sfac,
 988         'no-syslog' => sub {$syslog = 0; $sfac = undef;},
 989         'stderr' => \$stderr,
 990         'abbrev=i' => \$abbrev,
 991         'show-fast-forward-info' => \$showff,
 992         'no-show-fast-forward-info' => sub {$showff = 0},
 993         'status-interval=i' => \$stiv,
 994         'idle-status-interval=i' => \$idiv,
 995 ) || pod2usage(2);
 996 $syslog = 1 if defined($sfac);
 997 $progress = 1 unless $quiet;
 998 $abbrev = 128 unless $abbrev > 0;
 999 if (defined($idle_timeout)) {
1000         die "--idle-timeout must be a whole number" unless $idle_timeout =~ /^\d+$/;
1001         die "--idle-timeout may not be used without --inetd" unless $inetd;
1002 }
1003 if (defined($stiv)) {
1004         die "--status-interval must be a whole number" unless $stiv =~ /^\d+$/;
1005         $statusintv = $stiv * 60;
1006 }
1007 if (defined($idiv)) {
1008         die "--idle-status-interval must be a whole number" unless $idiv =~ /^\d+$/;
1009         $idleintv = $idiv * 60;
1010 }
1011
1012 open STDOUT, '>&STDERR' if $inetd;
1013 if ($syslog) {
1014         use Sys::Syslog qw();
1015         my $mode = 1;
1016         ++$mode if $stderr;
1017         $sfac = "user" unless defined($sfac) && $sfac ne "";
1018         my $ofac = $sfac;
1019         $sfac = uc($sfac);
1020         $sfac = 'LOG_'.$sfac unless $sfac =~ /^LOG_/;
1021         my $facility;
1022         my %badfac = map({("LOG_$_" => 1)}
1023                 (qw(PID CONS ODELAY NDELAY NOWAIT PERROR FACMASK NFACILITIES PRIMASK LFMT)));
1024         eval "\$facility = Sys::Syslog::$sfac; 1" or die "invalid syslog facility: $ofac";
1025         die "invalid syslog facility: $ofac"
1026                 if ($facility & ~0xf8) || ($facility >> 3) > 23 || $badfac{$sfac};
1027         tie *STDERR, 'OStream', $mode, $progname, $facility or die "tie failed";
1028 }
1029 if ($quiet) {
1030         open STDOUT, '>', '/dev/null';
1031 } elsif ($inetd) {
1032         *STDOUT = *STDERR;
1033 }
1034
1035 my $NAME;
1036
1037 if ($inetd) {
1038         open Server, '<&=0' or die "open: $!";
1039         my $sockname = getsockname Server;
1040         die "getsockname: $!" unless $sockname;
1041         die "socket already connected! must be 'wait' socket" if getpeername Server;
1042         die "getpeername: $!" unless $!{ENOTCONN};
1043         my $st = getsockopt Server, SOL_SOCKET, SO_TYPE;
1044         die "getsockopt(SOL_SOCKET, SO_TYPE): $!" unless $st;
1045         my $socktype = unpack('i', $st);
1046         die "stream socket required" unless defined $socktype && $socktype == SOCK_STREAM;
1047         die "AF_UNIX socket required" unless sockaddr_family($sockname) == AF_UNIX;
1048         $NAME = unpack_sockaddr_un $sockname;
1049         my $expected = $Girocco::Config::chroot.'/etc/taskd.socket';
1050         warn "listening on \"$NAME\" but expected \"$expected\"" unless $NAME eq $expected;
1051         my $mode = (stat($NAME))[2];
1052         die "stat: $!" unless $mode;
1053         $mode &= 07777;
1054         if (($mode & 0660) != 0660) {
1055                 chmod(($mode|0660), $NAME) == 1 or die "chmod ug+rw \"$NAME\" failed: $!";
1056         }
1057 } else {
1058         $NAME = $Girocco::Config::chroot.'/etc/taskd.socket';
1059         my $uaddr = sockaddr_un($NAME);
1060
1061         socket(Server, PF_UNIX, SOCK_STREAM, 0) or die "socket failed: $!";
1062         unlink($NAME);
1063         bind(Server, $uaddr) or die "bind failed: $!";
1064         listen(Server, SOMAXCONN) or die "listen failed: $!";
1065         chmod 0666, $NAME or die "chmod failed: $!";
1066 }
1067
1068 foreach my $throttle (@Girocco::Config::throttle_classes, @throttle_defaults) {
1069         my $classname = $throttle->{"name"};
1070         $classname or next;
1071         Throttle::GetClassInfo($classname, $throttle);
1072 }
1073
1074 sub _min {
1075         return $_[0] <= $_[1] ? $_[0] : $_[1];
1076 }
1077
1078 pipe($piperead, $pipewrite) or die "pipe failed: $!";
1079 setnonblock($piperead);
1080 select((select($pipewrite), $|=1)[0]);
1081 my $pipebuff = '';
1082 my $fdset_both = '';
1083 vec($fdset_both, fileno($piperead), 1) = 1;
1084 my $fdset_pipe = $fdset_both;
1085 vec($fdset_both, fileno(Server), 1) = 1;
1086 my $penalty = 0;
1087 my $t = time;
1088 my $penaltytime = $t;
1089 my $nextwakeup = $t + 60;
1090 my $nextstatus = undef;
1091 $nextstatus = $t + $statusintv if $statusintv;
1092 statmsg "listening on $NAME";
1093 while (1) {
1094         my ($rout, $eout, $nfound);
1095         do {
1096                 my $wait;
1097                 my $now = time;
1098                 my $adjustpenalty = sub {
1099                         if ($penaltytime < $now) {
1100                                 my $credit = $now - $penaltytime;
1101                                 $penalty = $penalty > $credit ? $penalty - $credit : 0;
1102                                 $penaltytime = $now;
1103                         }
1104                 };
1105                 if (defined($nextstatus) && $now >= $nextstatus) {
1106                         unless ($idlestatus && !$children && (!$idleintv || $now - $idlestatus < $idleintv)) {
1107                                 my $statmsg = "STATUS: $children active";
1108                                 my @running = ();
1109                                 if ($children) {
1110                                         my @stats = ();
1111                                         my $cnt = 0;
1112                                         foreach my $cls (sort(Throttle::GetClassList())) {
1113                                                 my $inf = Throttle::GetClassInfo($cls);
1114                                                 if ($inf->{'total'}) {
1115                                                         $cnt += $inf->{'total'};
1116                                                         push(@stats, substr(lc($cls),0,1)."=".
1117                                                                 $inf->{'total'}.'/'.$inf->{'active'});
1118                                                 }
1119                                         }
1120                                         push(@stats, "?=".($children-$cnt)) if @stats && $cnt < $children;
1121                                         $statmsg .= " (".join(" ",@stats).")" if @stats;
1122                                         foreach (Throttle::GetRunningPids()) {
1123                                                 my ($cls, $ts, $desc) = Throttle::GetPidInfo($_);
1124                                                 next unless $ts;
1125                                                 push(@running, "[${cls}::$desc] ".duration($now-$ts));
1126                                         }
1127                                 }
1128                                 my $idlesecs;
1129                                 $statmsg .= ", idle " . duration($idlesecs)
1130                                         if !$children && ($idlesecs = $now - $idlestart) >= 2;
1131                                 statmsg $statmsg;
1132                                 statmsg "STATUS: currently running: ".join(", ", @running)
1133                                         if @running;
1134                                 $idlestatus = $now if !$children;
1135                         }
1136                         $nextstatus += $statusintv while $nextstatus <= $now;
1137                 }
1138                 $nextwakeup += 60, $now = time while ($wait = $nextwakeup - $now) <= 0;
1139                 $wait = _min($wait, (Throttle::ServiceQueue()||60));
1140                 &$adjustpenalty; # this prevents ignoring accept when we shouldn't
1141                 my $fdset;
1142                 if ($penalty <= $maxspawn) {
1143                         $fdset = $fdset_both;
1144                 } else {
1145                         $fdset = $fdset_pipe;
1146                         $wait = $penalty - $maxspawn if $wait > $penalty - $maxspawn;
1147                 }
1148                 $nfound = select($rout=$fdset, undef, $eout=$fdset, $wait);
1149                 logmsg("select failed: $!"), exit(1) unless $nfound >= 0 || $!{EINTR} || $!{EAGAIN};
1150                 my $reaped;
1151                 Throttle::RemoveSupplicant($reaped) while ($reaped = shift(@reapedpids));
1152                 $now = time;
1153                 &$adjustpenalty; # this prevents banking credits for elapsed time
1154                 if ($idle_timeout && !$children && !$nfound && $now - $idlestart >= $idle_timeout) {
1155                         statmsg "idle timeout (@{[duration($idle_timeout)]}) exceeded now exiting";
1156                         exit 0;
1157                 }
1158         } while $nfound < 1;
1159         my $reout = $rout | $eout;
1160         if (vec($reout, fileno($piperead), 1)) {{
1161                 my $nloff = -1;
1162                 {
1163                         my $bytes;
1164                         do {$bytes = sysread($piperead, $pipebuff, 512, length($pipebuff))}
1165                                 while (!defined($bytes) && $!{EINTR});
1166                         last if !defined($bytes) && $!{EAGAIN};
1167                         die "sysread failed: $!" unless defined $bytes;
1168                         # since we always keep a copy of $pipewrite open EOF is fatal
1169                         die "sysread returned EOF on pipe read" unless $bytes;
1170                         $nloff = index($pipebuff, "\n", 0);
1171                         if ($nloff < 0 && length($pipebuff) >= 512) {
1172                                 $pipebuff = '';
1173                                 print STDERR "discarding 512 bytes of control pipe data with no \\n found\n";
1174                         }
1175                         redo unless $nloff >= 0;
1176                 }
1177                 last unless $nloff >= 0;
1178                 do {
1179                         my $msg = substr($pipebuff, 0, $nloff);
1180                         substr($pipebuff, 0, $nloff + 1) = '';
1181                         $nloff = index($pipebuff, "\n", 0);
1182                         process_pipe_msg($msg) if length($msg);
1183                 } while $nloff >= 0;
1184                 redo;
1185         }}
1186         next unless vec($reout, fileno(Server), 1);
1187         unless (accept(Client, Server)) {
1188                 logmsg "accept failed: $!" unless $!{EINTR};
1189                 next;
1190         }
1191         logmsg "connection on $NAME";
1192         ++$penalty;
1193         spawn sub {
1194                 my $inp = <STDIN>;
1195                 $inp = <STDIN> if defined($inp) && $inp eq "\n";
1196                 chomp $inp if defined($inp);
1197                 $inp or exit 0; # ignore empty connects
1198                 my ($cmd, $arg) = $inp =~ /^([a-zA-Z][a-zA-Z0-9._+-]*)(?:\s+(.*))?$/;
1199                 defined($arg) or $arg = '';
1200                 if ($cmd eq 'ref-changes') {
1201                         ref_changes($arg);
1202                 } elsif ($cmd eq 'clone') {
1203                         clone($arg);
1204                 } elsif ($cmd eq 'ref-change') {
1205                         ref_change($arg);
1206                 } elsif ($cmd eq 'throttle') {
1207                         throttle($arg);
1208                 } else {
1209                         die "ignoring unknown command: $cmd\n";
1210                 }
1211         };
1212         close Client;
1213 }
1214
1215 #
1216 ## -------------
1217 ## Documentation
1218 ## -------------
1219 #
1220
1221 __END__
1222
1223 =head1 NAME
1224
1225 taskd.pl - Perform Girocco service tasks
1226
1227 =head1 SYNOPSIS
1228
1229 taskd.pl [options]
1230
1231  Options:
1232    -h | --help                           detailed instructions
1233    -q | --quiet                          run quietly
1234    --no-quiet                            do not run quietly
1235    -P | --progress                       show occasional status updates
1236    -i | --inetd                          run as inetd unix stream wait service
1237                                          implies --quiet --syslog
1238    -t SECONDS | --idle-timeout=SECONDS   how long to wait idle before exiting
1239                                          requires --inetd
1240    -s | --syslog[=facility]              send messages to syslog instead of
1241                                          stderr but see --stderr
1242                                          enabled by --inetd
1243    --no-syslog                           do not send message to syslog
1244    --stderr                              always send messages to stderr too
1245    --abbrev=n                            abbreviate hashes to n (default is 8)
1246    --show-fast-forward-info              show fast-forward info (default is on)
1247    --no-show-fast-forward-info           disable showing fast-forward info
1248    --status-interval=MINUTES             status update interval (default 1)
1249    --idle-status-interval=IDLEMINUTES    idle status interval (default 60)
1250
1251 =head1 OPTIONS
1252
1253 =over 8
1254
1255 =item B<--help>
1256
1257 Print the full description of taskd.pl's options.
1258
1259 =item B<--quiet>
1260
1261 Suppress non-error messages, e.g. for use when running this task as an inetd
1262 service.  Enabled by default by --inetd.
1263
1264 =item B<--no-quiet>
1265
1266 Enable non-error messages.  When running in --inetd mode these messages are
1267 sent to STDERR instead of STDOUT.
1268
1269 =item B<--progress>
1270
1271 Show information about the current status of the task operation occasionally.
1272 This is automatically enabled if --quiet is not given.
1273
1274 =item B<--inetd>
1275
1276 Run as an inetd wait service.  File descriptor 0 must be an unconnected unix
1277 stream socket ready to have accept called on it.  To be useful, the unix socket
1278 should be located at "$Girocco::Config::chroot/etc/taskd.socket".  A warning
1279 will be issued if the socket is not in the expected location.  Socket file
1280 permissions will be adjusted if necessary and if they cannot be taskd.pl will
1281 die.  The --inetd option also enables the --quiet and --syslog options but
1282 --no-quiet and --no-syslog may be used to alter that.
1283
1284 The correct specification for the inetd socket is a "unix" protocol "stream"
1285 socket in "wait" mode with user and group writable permissions (0660).  An
1286 attempt will be made to alter the socket's file mode if needed and if that
1287 cannot be accomplished taskd.pl will die.
1288
1289 Although most inetd stream services run in nowait mode, taskd.pl MUST be run
1290 in wait mode and will die if the passed in socket is already connected.
1291
1292 Note that while *BSD's inetd happily supports unix sockets (and so does
1293 Darwin's launchd), neither xinetd nor GNU's inetd supports unix sockets.
1294 However, systemd does seem to.
1295
1296 =item B<--idle-timeout=SECONDS>
1297
1298 Only permitted when running in --inetd mode.  After SECONDS of inactivity
1299 (i.e. all outstanding tasks have completed and no new requests have come in)
1300 exit normally.  The default is no timeout at all (a SECONDS value of 0).
1301 Note that it may actually take up to SECONDS+60 for the idle exit to occur.
1302
1303 =item B<--syslog[=facility]>
1304
1305 Normally error output is sent to STDERR.  With this option it's sent to
1306 syslog instead.  Note that when running in --inetd mode non-error output is
1307 also affected by this option as it's sent to STDERR in that case.  If
1308 not specified, the default for facility is LOG_USER.  Facility names are
1309 case-insensitive and the leading 'LOG_' is optional.  Messages are logged
1310 with the LOG_NOTICE priority.
1311
1312 =item B<--no-syslog>
1313
1314 Send error message output to STDERR but not syslog.
1315
1316 =item B<--stderr>
1317
1318 Always send error message output to STDERR.  If --syslog is in effect then
1319 a copy will also be sent to syslog.  In --inetd mode this applies to non-error
1320 messages as well.
1321
1322 =item B<--abbrev=n>
1323
1324 Abbreviate displayed hash values to only the first n hexadecimal characters.
1325 The default is 8 characters.  Set to 0 for no abbreviation at all.
1326
1327 =item B<--show-fast-forward-info>
1328
1329 Instead of showing ' -> ' in ref-change/ref-notify update messages, show either
1330 '..' for a fast-forward, creation or deletion or '...' for non-fast-forward.
1331 This requires running an extra git command for each ref update that is not a
1332 creation or deletion in order to determine whether or not it's a fast forward.
1333
1334 =item B<--no-show-fast-forward-info>
1335
1336 Disable showing of fast-forward information for ref-change/ref-notify update
1337 messages.  Instead just show a ' -> ' indicator.
1338
1339 =item B<--status-interval=MINUTES>
1340
1341 If progress is enabled (with --progress or by default if no --inetd or --quiet)
1342 status updates are shown at each MINUTES interval.  Setting the interval to 0
1343 disables them entirely even with --progress.
1344
1345 =item B<--idle-status-interval=IDLEMINUTES>
1346
1347 Two consecutive "idle" status updates with no intervening activity will not be
1348 shown unless IDLEMINUTES have elapsed between them.  The default is 60 minutes.
1349 Setting the interval to 0 prevents any consecutive idle updates (with no
1350 activity between them) from appearing at all.
1351
1352 =back
1353
1354 =head1 DESCRIPTION
1355
1356 taskd.pl is Girocco's service request servant; it listens for service requests
1357 such as new clone requests and ref update notifications and spawns a task to
1358 perform the requested action.
1359
1360 =cut