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 = Girocco::Project->load($name);
 637         $proj or die "failed to load project $name";
 638         $proj->{clone_in_progress} or die "project $name is not marked for cloning";
 639         $proj->{clone_logged} and die "project $name is already being cloned";
 640         request_throttle("clone", $name) <= 0 or die "cloning $name aborted (throttled)";
 641         statmsg "cloning $name";
 642         open STDOUT, '>', "$Girocco::Config::reporoot/$name.git/.clonelog" or die "cannot open clonelog: $!";
 643         open STDERR, ">&STDOUT";
 644         open STDIN, '<', '/dev/null';
 645         exec "$Girocco::Config::basedir/taskd/clone.sh", "$name.git" or die "exec failed: $!";
 646 }
 647
 648 sub ref_indicator {
 649         return ' -> ' unless $showff && defined($_[0]);
 650         my ($git_dir, $old, $new) = @_;
 651         return '..' unless defined($old) && defined($new) && $old !~ /^0+$/ && $new !~ /^0+$/ && $old ne $new;
 652         # In many cases `git merge-base` is slower than this even if using the
 653         # `--is-ancestor` option available since Git 1.8.0, but it's never faster
 654         my $ans = get_git("--git-dir=$git_dir", "rev-list", "-n", "1", "^$new^0", "$old^0", "--") ? '...' : '..';
 655         return wantarray ? ($ans, 1) : $ans;
 656 }
 657
 658 sub ref_change {
 659         my ($arg) = @_;
 660         my ($username, $name, $oldrev, $newrev, $ref) = split(/\s+/, $arg);
 661         $username && $name && $oldrev && $newrev && $ref or return 0;
 662         $oldrev =~ /^[0-9a-f]{40}$/ && $newrev =~ /^[0-9a-f]{40}$/ && $ref =~ m{^refs/} or return 0;
 663         $newrev ne $oldrev or return 0;
 664
 665         Girocco::Project::does_exist($name, 1) or die "no such project: $name";
 666         my $proj = Girocco::Project->load($name);
 667         $proj or die "failed to load project $name";
 668         my $has_notify = $proj->has_notify;
 669         my $type = $has_notify ? "notify" : "change";
 670
 671         my $user;
 672         if ($username && $username !~ /^%.*%$/) {
 673                 Girocco::User::does_exist($username, 1) or die "no such user: $username";
 674                 $user = Girocco::User->load($username);
 675                 $user or die "failed to load user $username";
 676         } elsif ($username eq "%$name%") {
 677                 $username = "-";
 678         }
 679
 680         request_throttle("ref-change", $name) <= 0 or die "ref-change $name aborted (throttled)";
 681         my $ind = ref_indicator($proj->{path}, $oldrev, $newrev);
 682         statmsg "ref-$type $username $name ($ref: @{[substr($oldrev,0,$abbrev)]}$ind@{[substr($newrev,0,$abbrev)]})";
 683         open STDIN, '<', '/dev/null';
 684         Girocco::Notify::ref_change($proj, $user, $ref, $oldrev, $newrev) if $has_notify;
 685         return 0;
 686 }
 687
 688 sub ref_changes {
 689         my ($arg) = @_;
 690         my ($username, $name) = split(/\s+/, $arg);
 691         $username && $name or return 0;
 692
 693         Girocco::Project::does_exist($name, 1) or die "no such project: $name";
 694         my $proj = Girocco::Project->load($name);
 695         $proj or die "failed to load project $name";
 696         my $has_notify = $proj->has_notify;
 697         my $type = $has_notify ? "notify" : "change";
 698
 699         my $user;
 700         if ($username && $username !~ /^%.*%$/) {
 701                 Girocco::User::does_exist($username, 1) or die "no such user: $username";
 702                 $user = Girocco::User->load($username);
 703                 $user or die "failed to load user $username";
 704         } elsif ($username eq "%$name%") {
 705                 $username = "-";
 706         }
 707
 708         my @changes = ();
 709         my %oldheads = ();
 710         my %deletedheads = ();
 711         while (my $change = <STDIN>) {
 712                 my ($oldrev, $newrev, $ref) = split(/\s+/, $change);
 713                 $oldrev =~ /^[0-9a-f]{40}$/ && $newrev =~ /^[0-9a-f]{40}$/ && $ref =~ m{^refs/} or next;
 714                 if ($ref =~ m{^refs/heads/.}) {
 715                         if ($oldrev =~ /^0{40}$/) {
 716                                 delete $oldheads{$ref};
 717                                 $deletedheads{$ref} = 1;
 718                         } elsif ($newrev ne $oldrev || (!exists($oldheads{$ref}) && !$deletedheads{$ref})) {
 719                                 $oldheads{$ref} = $oldrev;
 720                         }
 721                 }
 722                 $newrev ne $oldrev or next;
 723                 push(@changes, [$oldrev, $newrev, $ref]);
 724         }
 725         return 0 unless @changes;
 726         open STDIN, '<', '/dev/null';
 727         request_throttle("ref-change", $name) <= 0 or die "ref-changes $name aborted (throttled)";
 728         my $statproc = sub {
 729                 my ($old, $new, $ref, $ran_mail_sh) = @_;
 730                 my ($ind, $ran_git) = ref_indicator($proj->{path}, $old, $new);
 731                 statmsg "ref-$type $username $name ($ref: @{[substr($old,0,$abbrev)]}$ind@{[substr($new,0,$abbrev)]})";
 732                 sleep 1 if $ran_mail_sh || $ran_git;
 733         };
 734         if ($has_notify) {
 735                 Girocco::Notify::ref_changes($proj, $user, $statproc, \%oldheads, @changes);
 736         } else {
 737                 &$statproc(@$_) foreach @changes;
 738         }
 739         return 0;
 740 }
 741
 742 sub throttle {
 743         my ($arg) = @_;
 744         my ($pid, $classname, $text) = split(/\s+/, $arg);
 745         $pid =~ /^\d+/ or return 0; # invalid pid
 746         $pid += 0;
 747         $pid > 0 or return 0; # invalid pid
 748         kill(0, $pid) || $!{EPERM} or return 0; # no such process
 749         Throttle::GetClassInfo($classname) or return 0; # no such throttle class
 750         defined($text) && $text ne '' or return 0; # no text no service
 751
 752         my $throttled = 0;
 753         my $proceed = 0;
 754         my $error = 0;
 755         my $controldead = 0;
 756         my $suppdead = 0;
 757         my ($waker, $wakew);
 758         pipe($waker, $wakew) or die "pipe failed: $!";
 759         select((select($wakew),$|=1)[0]);
 760         setnonblock($wakew);
 761         $SIG{'TERM'} = sub {$throttled = 1; syswrite($wakew, '!')};
 762         $SIG{'USR1'} = sub {$proceed = 1; syswrite($wakew, '!')};
 763         $SIG{'USR2'} = sub {$error = 1; syswrite($wakew, '!')};
 764         $SIG{'PIPE'} = sub {$controldead = 1; syswrite($wakew, '!')};
 765         select((select(STDIN),$|=1)[0]);
 766
 767         logmsg "throttle $pid $classname $text request";
 768         # After writing we can expect a SIGTERM or SIGUSR1
 769         print $pipewrite "\nthrottle $$ $classname $text\n";
 770
 771         # NOTE: the only way to detect the socket close is to read all the
 772         #       data until EOF is reached -- recv can be used to peek.
 773         my $v = '';
 774         vec($v, fileno(STDIN), 1) = 1;
 775         vec($v, fileno($waker), 1) = 1;
 776         setnonblock(\*STDIN);
 777         setnonblock($waker);
 778         until ($controldead || $throttled || $proceed || $error || $suppdead) {
 779                 my ($r, $e);
 780                 select($r=$v, undef, $e=$v, 30);
 781                 my ($bytes, $discard);
 782                 do {$bytes = sysread($waker, $discard, 512)} while (defined($bytes) && $bytes > 0);
 783                 do {$bytes = sysread(STDIN, $discard, 4096)} while (defined($bytes) && $bytes > 0);
 784                 $suppdead = 1 unless !defined($bytes) && $!{EAGAIN};
 785                 print $pipewrite "\nkeepalive $$\n";
 786         }
 787         setblock(\*STDIN);
 788
 789         if ($throttled && !$suppdead) {
 790                 print STDIN "throttled\n";
 791                 logmsg "throttle $pid $classname $text throttled";
 792         } elsif ($proceed && !$suppdead) {
 793                 print STDIN "proceed\n";
 794                 logmsg "throttle $pid $classname $text proceed";
 795                 $SIG{'TERM'} = 'DEFAULT';
 796                 # Stay alive until the child dies which we detect by EOF on STDIN
 797                 setnonblock(\*STDIN);
 798                 until ($controldead || $suppdead) {
 799                         my ($r, $e);
 800                         select($r=$v, undef, $e=$v, 30);
 801                         my ($bytes, $discard);
 802                         do {$bytes = sysread($waker, $discard, 512)} while (defined($bytes) && $bytes > 0);
 803                         do {$bytes = sysread(STDIN, $discard, 512)} while (defined($bytes) && $bytes > 0);
 804                         $suppdead = 1 unless !defined($bytes) && $!{EAGAIN};
 805                         print $pipewrite "\nkeepalive $$\n";
 806                 }
 807                 setblock(\*STDIN);
 808         } else {
 809                 my $prefix = '';
 810                 $prefix = "control" if $controldead && !$suppdead;
 811                 logmsg "throttle $pid $classname $text ${prefix}died";
 812         }
 813         exit 0;
 814 }
 815
 816 sub process_pipe_msg {
 817         my ($act, $pid, $cls, $text) = split(/\s+/, $_[0]);
 818         if ($act eq "throttle") {
 819                 $pid =~ /^\d+$/ or return 0;
 820                 $pid += 0;
 821                 $pid > 0 or return 0; # invalid pid
 822                 kill(0, $pid) or return 0; # invalid pid
 823                 defined($cls) && $cls ne "" or kill('USR2', $pid), return 0;
 824                 defined($text) && $text ne "" or kill('USR2', $pid), return 0;
 825                 Throttle::GetClassInfo($cls) or kill('USR2', $pid), return 0;
 826                 # the AddSupplicant call could send SIGUSR1 before it returns
 827                 my $result = Throttle::AddSupplicant($pid, $cls, $text);
 828                 kill('USR2', $pid), return 0 if $result < 0;
 829                 kill('TERM', $pid), return 0 if $result > 0;
 830                 # $pid was added to class $cls and will receive SIGUSR1 when
 831                 # it's time for it to proceed
 832                 return 0;
 833         } elsif ($act eq "keepalive") {
 834                 # nothing to do although we could verify pid is valid and
 835                 # still in %Throttle::pids and send a SIGUSR2 if not, but
 836                 # really keepalive should just be ignored.
 837                 return 0;
 838         }
 839         print STDERR "discarding unknown pipe message \"$_[0]\"\n";
 840         return 0;
 841 }
 842
 843 #
 844 ## -------
 845 ## OStream
 846 ## -------
 847 #
 848
 849 package OStream;
 850
 851 # Set to 1 for only syslog output (if enabled by mode)
 852 # Set to 2 for only stderr output (if enabled by mode)
 853 our $only = 0; # This is a hack
 854
 855 use Carp 'croak';
 856 use Sys::Syslog qw(:DEFAULT :macros);
 857
 858 sub writeall {
 859         use POSIX qw();
 860         use Errno;
 861         my ($fd, $data) = @_;
 862         my $offset = 0;
 863         my $remaining = length($data);
 864         while ($remaining) {
 865                 my $bytes = POSIX::write(
 866                         $fd,
 867                         substr($data, $offset, $remaining),
 868                         $remaining);
 869                 next if !defined($bytes) && $!{EINTR};
 870                 croak "POSIX::write failed: $!" unless defined $bytes;
 871                 croak "POSIX::write wrote 0 bytes" unless $bytes;
 872                 $remaining -= $bytes;
 873                 $offset += $bytes;
 874         }
 875 }
 876
 877 sub dumpline {
 878         use POSIX qw(STDERR_FILENO);
 879         my ($self, $line) = @_;
 880         $only = 0 unless defined($only);
 881         writeall(STDERR_FILENO, $line) if $self->{'stderr'} && $only != 1;
 882         substr($line, -1, 1) = '' if substr($line, -1, 1) eq "\n";
 883         return unless length($line);
 884         syslog(LOG_NOTICE, "%s", $line) if $self->{'syslog'} && $only != 2;
 885 }
 886
 887 sub TIEHANDLE {
 888         my $class = shift || 'OStream';
 889         my $mode = shift;
 890         my $syslogname = shift;
 891         my $syslogfacility = shift;
 892         defined($syslogfacility) or $syslogfacility = LOG_USER;
 893         my $self = {};
 894         $self->{'syslog'} = $mode > 0;
 895         $self->{'stderr'} = $mode <= 0 || $mode > 1;
 896         $self->{'lastline'} = '';
 897         if ($self->{'syslog'}) {
 898                 # Some Sys::Syslog have a stupid default setlogsock order
 899                 eval {Sys::Syslog::setlogsock("native"); 1;} or
 900                 eval {Sys::Syslog::setlogsock("unix");};
 901                 openlog($syslogname, "ndelay,pid", $syslogfacility)
 902                         or croak "Sys::Syslog::openlog failed: $!";
 903         }
 904         return bless $self, $class;
 905 }
 906
 907 sub BINMODE {return 1}
 908 sub FILENO {return undef}
 909 sub EOF {return 0}
 910 sub CLOSE {return 1}
 911
 912 sub PRINTF {
 913         my $self = shift;
 914         my $template = shift;
 915         return $self->PRINT(sprintf $template, @_);
 916 }
 917
 918 sub PRINT {
 919         my $self = shift;
 920         my $data = join('', $self->{'lastline'}, @_);
 921         my $pos = 0;
 922         while ((my $idx = index($data, "\n", $pos)) >= 0) {
 923                 ++$idx;
 924                 my $line = substr($data, $pos, $idx - $pos);
 925                 substr($data, $pos, $idx - $pos) = '';
 926                 $pos = $idx;
 927                 $self->dumpline($line);
 928         }
 929         $self->{'lastline'} = $data;
 930         return 1;
 931 }
 932
 933 sub DESTROY {
 934         my $self = shift;
 935         $self->dumpline($self->{'lastline'})
 936                 if length($self->{'lastline'});
 937         closelog;
 938 }
 939
 940 sub WRITE {
 941         my $self = shift;
 942         my ($scalar, $length, $offset) = @_;
 943         $scalar = '' if !defined($scalar);
 944         $length = length($scalar) if !defined($length);
 945         croak "OStream::WRITE invalid length $length"
 946                 if $length < 0;
 947         $offset = 0 if !defined($offset);
 948         $offset += length($scalar) if $offset < 0;
 949         croak "OStream::WRITE invalid write offset"
 950                 if $offset < 0 || $offset > $length;
 951         my $max = length($scalar) - $offset;
 952         $length = $max if $length > $max;
 953         $self->PRINT(substr($scalar, $offset, $length));
 954         return $length;
 955 }
 956
 957 #
 958 ## ----
 959 ## main
 960 ## ----
 961 #
 962
 963 package main;
 964
 965 close(DATA) if fileno(DATA);
 966 my $sfac;
 967 Getopt::Long::Configure('bundling');
 968 my ($stiv, $idiv);
 969 my $parse_res = GetOptions(
 970         'help|?|h' => sub {pod2usage(-verbose => 2, -exitval => 0)},
 971         'quiet|q' => \$quiet,
 972         'no-quiet' => sub {$quiet = 0},
 973         'progress|P' => \$progress,
 974         'inetd|i' => sub {$inetd = 1; $syslog = 1; $quiet = 1;},
 975         'idle-timeout|t=i' => \$idle_timeout,
 976         'syslog|s:s' => \$sfac,
 977         'no-syslog' => sub {$syslog = 0; $sfac = undef;},
 978         'stderr' => \$stderr,
 979         'abbrev=i' => \$abbrev,
 980         'show-fast-forward-info' => \$showff,
 981         'no-show-fast-forward-info' => sub {$showff = 0},
 982         'status-interval=i' => \$stiv,
 983         'idle-status-interval=i' => \$idiv,
 984 ) || pod2usage(2);
 985 $syslog = 1 if defined($sfac);
 986 $progress = 1 unless $quiet;
 987 $abbrev = 128 unless $abbrev > 0;
 988 if (defined($idle_timeout)) {
 989         die "--idle-timeout must be a whole number" unless $idle_timeout =~ /^\d+$/;
 990         die "--idle-timeout may not be used without --inetd" unless $inetd;
 991 }
 992 if (defined($stiv)) {
 993         die "--status-interval must be a whole number" unless $stiv =~ /^\d+$/;
 994         $statusintv = $stiv * 60;
 995 }
 996 if (defined($idiv)) {
 997         die "--idle-status-interval must be a whole number" unless $idiv =~ /^\d+$/;
 998         $idleintv = $idiv * 60;
 999 }
1000
1001 open STDOUT, '>&STDERR' if $inetd;
1002 if ($syslog) {
1003         use Sys::Syslog qw();
1004         my $mode = 1;
1005         ++$mode if $stderr;
1006         $sfac = "user" unless defined($sfac) && $sfac ne "";
1007         my $ofac = $sfac;
1008         $sfac = uc($sfac);
1009         $sfac = 'LOG_'.$sfac unless $sfac =~ /^LOG_/;
1010         my $facility;
1011         my %badfac = map({("LOG_$_" => 1)}
1012                 (qw(PID CONS ODELAY NDELAY NOWAIT PERROR FACMASK NFACILITIES PRIMASK LFMT)));
1013         eval "\$facility = Sys::Syslog::$sfac; 1" or die "invalid syslog facility: $ofac";
1014         die "invalid syslog facility: $ofac"
1015                 if ($facility & ~0xf8) || ($facility >> 3) > 23 || $badfac{$sfac};
1016         tie *STDERR, 'OStream', $mode, $progname, $facility or die "tie failed";
1017 }
1018 if ($quiet) {
1019         open STDOUT, '>', '/dev/null';
1020 } elsif ($inetd) {
1021         *STDOUT = *STDERR;
1022 }
1023
1024 my $NAME;
1025
1026 if ($inetd) {
1027         open Server, '<&=0' or die "open: $!";
1028         my $sockname = getsockname Server;
1029         die "getsockname: $!" unless $sockname;
1030         die "socket already connected! must be 'wait' socket" if getpeername Server;
1031         die "getpeername: $!" unless $!{ENOTCONN};
1032         my $st = getsockopt Server, SOL_SOCKET, SO_TYPE;
1033         die "getsockopt(SOL_SOCKET, SO_TYPE): $!" unless $st;
1034         my $socktype = unpack('i', $st);
1035         die "stream socket required" unless defined $socktype && $socktype == SOCK_STREAM;
1036         die "AF_UNIX socket required" unless sockaddr_family($sockname) == AF_UNIX;
1037         $NAME = unpack_sockaddr_un $sockname;
1038         my $expected = $Girocco::Config::chroot.'/etc/taskd.socket';
1039         warn "listening on \"$NAME\" but expected \"$expected\"" unless $NAME eq $expected;
1040         my $mode = (stat($NAME))[2];
1041         die "stat: $!" unless $mode;
1042         $mode &= 07777;
1043         if (($mode & 0660) != 0660) {
1044                 chmod(($mode|0660), $NAME) == 1 or die "chmod ug+rw \"$NAME\" failed: $!";
1045         }
1046 } else {
1047         $NAME = $Girocco::Config::chroot.'/etc/taskd.socket';
1048         my $uaddr = sockaddr_un($NAME);
1049
1050         socket(Server, PF_UNIX, SOCK_STREAM, 0) or die "socket failed: $!";
1051         unlink($NAME);
1052         bind(Server, $uaddr) or die "bind failed: $!";
1053         listen(Server, SOMAXCONN) or die "listen failed: $!";
1054         chmod 0666, $NAME or die "chmod failed: $!";
1055 }
1056
1057 foreach my $throttle (@Girocco::Config::throttle_classes, @throttle_defaults) {
1058         my $classname = $throttle->{"name"};
1059         $classname or next;
1060         Throttle::GetClassInfo($classname, $throttle);
1061 }
1062
1063 sub _min {
1064         return $_[0] <= $_[1] ? $_[0] : $_[1];
1065 }
1066
1067 pipe($piperead, $pipewrite) or die "pipe failed: $!";
1068 setnonblock($piperead);
1069 select((select($pipewrite), $|=1)[0]);
1070 my $pipebuff = '';
1071 my $fdset_both = '';
1072 vec($fdset_both, fileno($piperead), 1) = 1;
1073 my $fdset_pipe = $fdset_both;
1074 vec($fdset_both, fileno(Server), 1) = 1;
1075 my $penalty = 0;
1076 my $t = time;
1077 my $penaltytime = $t;
1078 my $nextwakeup = $t + 60;
1079 my $nextstatus = undef;
1080 $nextstatus = $t + $statusintv if $statusintv;
1081 statmsg "listening on $NAME";
1082 while (1) {
1083         my ($rout, $eout, $nfound);
1084         do {
1085                 my $wait;
1086                 my $now = time;
1087                 my $adjustpenalty = sub {
1088                         if ($penaltytime < $now) {
1089                                 my $credit = $now - $penaltytime;
1090                                 $penalty = $penalty > $credit ? $penalty - $credit : 0;
1091                                 $penaltytime = $now;
1092                         }
1093                 };
1094                 if (defined($nextstatus) && $now >= $nextstatus) {
1095                         unless ($idlestatus && !$children && (!$idleintv || $now - $idlestatus < $idleintv)) {
1096                                 my $statmsg = "STATUS: $children active";
1097                                 my @running = ();
1098                                 if ($children) {
1099                                         my @stats = ();
1100                                         my $cnt = 0;
1101                                         foreach my $cls (sort(Throttle::GetClassList())) {
1102                                                 my $inf = Throttle::GetClassInfo($cls);
1103                                                 if ($inf->{'total'}) {
1104                                                         $cnt += $inf->{'total'};
1105                                                         push(@stats, substr(lc($cls),0,1)."=".
1106                                                                 $inf->{'total'}.'/'.$inf->{'active'});
1107                                                 }
1108                                         }
1109                                         push(@stats, "?=".($children-$cnt)) if @stats && $cnt < $children;
1110                                         $statmsg .= " (".join(" ",@stats).")" if @stats;
1111                                         foreach (Throttle::GetRunningPids()) {
1112                                                 my ($cls, $ts, $desc) = Throttle::GetPidInfo($_);
1113                                                 next unless $ts;
1114                                                 push(@running, "[${cls}::$desc] ".duration($now-$ts));
1115                                         }
1116                                 }
1117                                 my $idlesecs;
1118                                 $statmsg .= ", idle " . duration($idlesecs)
1119                                         if !$children && ($idlesecs = $now - $idlestart) >= 2;
1120                                 statmsg $statmsg;
1121                                 statmsg "STATUS: currently running: ".join(", ", @running)
1122                                         if @running;
1123                                 $idlestatus = $now if !$children;
1124                         }
1125                         $nextstatus += $statusintv while $nextstatus <= $now;
1126                 }
1127                 $nextwakeup += 60, $now = time while ($wait = $nextwakeup - $now) <= 0;
1128                 $wait = _min($wait, (Throttle::ServiceQueue()||60));
1129                 &$adjustpenalty; # this prevents ignoring accept when we shouldn't
1130                 my $fdset;
1131                 if ($penalty <= $maxspawn) {
1132                         $fdset = $fdset_both;
1133                 } else {
1134                         $fdset = $fdset_pipe;
1135                         $wait = $penalty - $maxspawn if $wait > $penalty - $maxspawn;
1136                 }
1137                 $nfound = select($rout=$fdset, undef, $eout=$fdset, $wait);
1138                 logmsg("select failed: $!"), exit(1) unless $nfound >= 0 || $!{EINTR} || $!{EAGAIN};
1139                 my $reaped;
1140                 Throttle::RemoveSupplicant($reaped) while ($reaped = shift(@reapedpids));
1141                 $now = time;
1142                 &$adjustpenalty; # this prevents banking credits for elapsed time
1143                 if ($idle_timeout && !$children && !$nfound && $now - $idlestart >= $idle_timeout) {
1144                         statmsg "idle timeout (@{[duration($idle_timeout)]}) exceeded now exiting";
1145                         exit 0;
1146                 }
1147         } while $nfound < 1;
1148         my $reout = $rout | $eout;
1149         if (vec($reout, fileno($piperead), 1)) {{
1150                 my $nloff = -1;
1151                 {
1152                         my $bytes;
1153                         do {$bytes = sysread($piperead, $pipebuff, 512, length($pipebuff))}
1154                                 while (!defined($bytes) && $!{EINTR});
1155                         last if !defined($bytes) && $!{EAGAIN};
1156                         die "sysread failed: $!" unless defined $bytes;
1157                         # since we always keep a copy of $pipewrite open EOF is fatal
1158                         die "sysread returned EOF on pipe read" unless $bytes;
1159                         $nloff = index($pipebuff, "\n", 0);
1160                         if ($nloff < 0 && length($pipebuff) >= 512) {
1161                                 $pipebuff = '';
1162                                 print STDERR "discarding 512 bytes of control pipe data with no \\n found\n";
1163                         }
1164                         redo unless $nloff >= 0;
1165                 }
1166                 last unless $nloff >= 0;
1167                 do {
1168                         my $msg = substr($pipebuff, 0, $nloff);
1169                         substr($pipebuff, 0, $nloff + 1) = '';
1170                         $nloff = index($pipebuff, "\n", 0);
1171                         process_pipe_msg($msg) if length($msg);
1172                 } while $nloff >= 0;
1173                 redo;
1174         }}
1175         next unless vec($reout, fileno(Server), 1);
1176         unless (accept(Client, Server)) {
1177                 logmsg "accept failed: $!" unless $!{EINTR};
1178                 next;
1179         }
1180         logmsg "connection on $NAME";
1181         ++$penalty;
1182         spawn sub {
1183                 my $inp = <STDIN>;
1184                 $inp = <STDIN> if defined($inp) && $inp eq "\n";
1185                 chomp $inp if defined($inp);
1186                 $inp or exit 0; # ignore empty connects
1187                 my ($cmd, $arg) = $inp =~ /^([a-zA-Z][a-zA-Z0-9._+-]*)(?:\s+(.*))?$/;
1188                 defined($arg) or $arg = '';
1189                 if ($cmd eq 'ref-changes') {
1190                         ref_changes($arg);
1191                 } elsif ($cmd eq 'clone') {
1192                         clone($arg);
1193                 } elsif ($cmd eq 'ref-change') {
1194                         ref_change($arg);
1195                 } elsif ($cmd eq 'throttle') {
1196                         throttle($arg);
1197                 } else {
1198                         die "ignoring unknown command: $cmd\n";
1199                 }
1200         };
1201         close Client;
1202 }
1203
1204 #
1205 ## -------------
1206 ## Documentation
1207 ## -------------
1208 #
1209
1210 __END__
1211
1212 =head1 NAME
1213
1214 taskd.pl - Perform Girocco service tasks
1215
1216 =head1 SYNOPSIS
1217
1218 taskd.pl [options]
1219
1220  Options:
1221    -h | --help                           detailed instructions
1222    -q | --quiet                          run quietly
1223    --no-quiet                            do not run quietly
1224    -P | --progress                       show occasional status updates
1225    -i | --inetd                          run as inetd unix stream wait service
1226                                          implies --quiet --syslog
1227    -t SECONDS | --idle-timeout=SECONDS   how long to wait idle before exiting
1228                                          requires --inetd
1229    -s | --syslog[=facility]              send messages to syslog instead of
1230                                          stderr but see --stderr
1231                                          enabled by --inetd
1232    --no-syslog                           do not send message to syslog
1233    --stderr                              always send messages to stderr too
1234    --abbrev=n                            abbreviate hashes to n (default is 8)
1235    --show-fast-forward-info              show fast-forward info (default is on)
1236    --no-show-fast-forward-info           disable showing fast-forward info
1237    --status-interval=MINUTES             status update interval (default 1)
1238    --idle-status-interval=IDLEMINUTES    idle status interval (default 60)
1239
1240 =head1 OPTIONS
1241
1242 =over 8
1243
1244 =item B<--help>
1245
1246 Print the full description of taskd.pl's options.
1247
1248 =item B<--quiet>
1249
1250 Suppress non-error messages, e.g. for use when running this task as an inetd
1251 service.  Enabled by default by --inetd.
1252
1253 =item B<--no-quiet>
1254
1255 Enable non-error messages.  When running in --inetd mode these messages are
1256 sent to STDERR instead of STDOUT.
1257
1258 =item B<--progress>
1259
1260 Show information about the current status of the task operation occasionally.
1261 This is automatically enabled if --quiet is not given.
1262
1263 =item B<--inetd>
1264
1265 Run as an inetd wait service.  File descriptor 0 must be an unconnected unix
1266 stream socket ready to have accept called on it.  To be useful, the unix socket
1267 should be located at "$Girocco::Config::chroot/etc/taskd.socket".  A warning
1268 will be issued if the socket is not in the expected location.  Socket file
1269 permissions will be adjusted if necessary and if they cannot be taskd.pl will
1270 die.  The --inetd option also enables the --quiet and --syslog options but
1271 --no-quiet and --no-syslog may be used to alter that.
1272
1273 The correct specification for the inetd socket is a "unix" protocol "stream"
1274 socket in "wait" mode with user and group writable permissions (0660).  An
1275 attempt will be made to alter the socket's file mode if needed and if that
1276 cannot be accomplished taskd.pl will die.
1277
1278 Although most inetd stream services run in nowait mode, taskd.pl MUST be run
1279 in wait mode and will die if the passed in socket is already connected.
1280
1281 Note that while *BSD's inetd happily supports unix sockets (and so does
1282 Darwin's launchd), neither xinetd nor GNU's inetd supports unix sockets.
1283 However, systemd does seem to.
1284
1285 =item B<--idle-timeout=SECONDS>
1286
1287 Only permitted when running in --inetd mode.  After SECONDS of inactivity
1288 (i.e. all outstanding tasks have completed and no new requests have come in)
1289 exit normally.  The default is no timeout at all (a SECONDS value of 0).
1290 Note that it may actually take up to SECONDS+60 for the idle exit to occur.
1291
1292 =item B<--syslog[=facility]>
1293
1294 Normally error output is sent to STDERR.  With this option it's sent to
1295 syslog instead.  Note that when running in --inetd mode non-error output is
1296 also affected by this option as it's sent to STDERR in that case.  If
1297 not specified, the default for facility is LOG_USER.  Facility names are
1298 case-insensitive and the leading 'LOG_' is optional.  Messages are logged
1299 with the LOG_NOTICE priority.
1300
1301 =item B<--no-syslog>
1302
1303 Send error message output to STDERR but not syslog.
1304
1305 =item B<--stderr>
1306
1307 Always send error message output to STDERR.  If --syslog is in effect then
1308 a copy will also be sent to syslog.  In --inetd mode this applies to non-error
1309 messages as well.
1310
1311 =item B<--abbrev=n>
1312
1313 Abbreviate displayed hash values to only the first n hexadecimal characters.
1314 The default is 8 characters.  Set to 0 for no abbreviation at all.
1315
1316 =item B<--show-fast-forward-info>
1317
1318 Instead of showing ' -> ' in ref-change/ref-notify update messages, show either
1319 '..' for a fast-forward, creation or deletion or '...' for non-fast-forward.
1320 This requires running an extra git command for each ref update that is not a
1321 creation or deletion in order to determine whether or not it's a fast forward.
1322
1323 =item B<--no-show-fast-forward-info>
1324
1325 Disable showing of fast-forward information for ref-change/ref-notify update
1326 messages.  Instead just show a ' -> ' indicator.
1327
1328 =item B<--status-interval=MINUTES>
1329
1330 If progress is enabled (with --progress or by default if no --inetd or --quiet)
1331 status updates are shown at each MINUTES interval.  Setting the interval to 0
1332 disables them entirely even with --progress.
1333
1334 =item B<--idle-status-interval=IDLEMINUTES>
1335
1336 Two consecutive "idle" status updates with no intervening activity will not be
1337 shown unless IDLEMINUTES have elapsed between them.  The default is 60 minutes.
1338 Setting the interval to 0 prevents any consecutive idle updates (with no
1339 activity between them) from appearing at all.
1340
1341 =back
1342
1343 =head1 DESCRIPTION
1344
1345 taskd.pl is Girocco's service request servant; it listens for service requests
1346 such as new clone requests and ref update notifications and spawns a task to
1347 perform the requested action.
1348
1349 =cut