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