3 # taskd - Clone repositories on request
5 # taskd is Girocco mirroring servant; it processes requests for clones
6 # of given URLs received over its socket.
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.
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.
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
28 # Initially based on perlipc example.
30 use 5.008; # we need safe signals
39 use POSIX
":sys_wait_h";
47 use Girocco
::Util
qw(noFatalsToBrowser get_git);
48 BEGIN {noFatalsToBrowser
}
49 use Girocco
::ExecUtil
;
51 # Throttle Classes Defaults
52 # Note that any same-named classes in @Girocco::Config::throttle_classes
53 # will override (completely replacing the entire hash) these ones.
54 my @throttle_defaults = (
69 #maxproc => max(5, cpucount + maxjobs), # this is the default
70 #maxjobs => max(1, int(cpucount / 4)) , # this is the default
90 my $progname = basename
($0);
96 use Girocco
::Util
"online_cpus";
97 our $online_cpus_result;
98 $online_cpus_result = online_cpus
unless $online_cpus_result;
99 return $online_cpus_result;
103 my $hdr = "[@{[scalar localtime]}] $progname $$: ";
105 $OStream::only
= 2; # STDERR only
107 $OStream::only
= 1; # syslog only
109 $OStream::only
= 0; # back to default
116 return unless $progress;
117 my $hdr = "[@{[scalar localtime]}] $progname $$: ";
119 $OStream::only
= 2; # STDERR only
120 print STDERR
"$hdr@_\n";
121 $OStream::only
= 1; # syslog only
123 $OStream::only
= 0; # back to default
125 print STDERR
"$hdr@_\n";
131 return $secs unless defined($secs) && $secs >= 0;
133 my $ans = ($secs % 60) . 's';
134 return $ans if $secs < 60;
135 $secs = int($secs / 60);
136 $ans = ($secs % 60) . 'm' . $ans;
137 return $ans if $secs < 60;
138 $secs = int($secs / 60);
139 $ans = ($secs % 24) . 'h' . $ans;
140 return $ans if $secs < 24;
141 $secs = int($secs / 24);
142 return $secs . 'd' . $ans;
147 my $flags = fcntl($fd, F_GETFL
, 0);
148 defined($flags) or die "fcntl failed: $!";
149 fcntl($fd, F_SETFL
, $flags | O_NONBLOCK
) or die "fcntl failed: $!";
154 my $flags = fcntl($fd, F_GETFL
, 0);
155 defined($flags) or die "fcntl failed: $!";
156 fcntl($fd, F_SETFL
, $flags & ~O_NONBLOCK
) or die "fcntl failed: $!";
164 ## 1) Process needing throttle services acquire a control file descriptor
165 ## a) Either as a result of a fork + exec (the write end of a pipe)
166 ## b) Or by connecting to the taskd socket (not yet implemented)
168 ## 2) The process requesting throttle services will be referred to
169 ## as the supplicant or just "supp" for short.
171 ## 3) The supp first completes any needed setup which may include
172 ## gathering data it needs to perform the action -- if that fails
173 ## then there's no need for any throttling.
175 ## 4) The supp writes a throttle request to the control descriptor in
177 ## throttle <pid> <class>\n
178 ## for example if the supp's pid was 1234 and it was requesting throttle
179 ## control as a member of the mail class it would write this message:
180 ## throttle 1234 mail\n
181 ## Note that if the control descriptor happens to be a pipe rather than a
182 ## socket, the message should be preceded by another "\n" just be be safe.
183 ## If the control descriptor is a socket, not a pipe, the message may be
184 ## preceded by a "\n" but that's not recommended.
186 ## 5) For supplicants with a control descriptor that is a pipe
187 ## (getsockopt(SO_TYPE) returns ENOTSOCK) the (5a) protocol should be used.
188 ## If the control descriptor is a socket (getsockname succeeds) then
189 ## protocol (5b) should be used.
191 ## 5a) The supp now enters a "pause" loop awaiting either a SIGUSR1, SIGUSR2 or
192 ## SIGTERM. It should wake up periodically (SIGALRM works well) and attempt
193 ## to write a "keepalive\n" message to the control descriptor. If that
194 ## fails, the controller has gone away and it may make its own decision
195 ## whether or not to proceed at that point. If, on the other hand, it
196 ## receives a SIGTERM, the process limit for its class has been reached
197 ## and it should abort without performing its action. If it receives
198 ## SIGUSR1, it may proceed without writing anything more to the control
199 ## descriptor, any MAY even close the control descriptor. Finally, a
200 ## SIGUSR2 indicates rejection of the throttle request for some other reason
201 ## such as unrecognized class name or invalid pid in which case the supp may
202 ## make its own decision how to proceed.
204 ## 5b) The supp now enters a read wait on the socket -- it need accomodate no
205 ## more than 512 bytes and if a '\n' does not appear within that number of
206 ## bytes the read should be considered failed. Otherwise the read should
207 ## be retried until either a full line has been read or the socket is
208 ## closed from the other end. If the lone read is "proceed\n" then it may
209 ## proceed without reading or writing anything more to the control
210 ## descriptor, but MUST keep the control descriptor open and not call
211 ## shutdown on it either. Any other result (except EINTR or EAGAIN which
212 ## should be retried) constitutes failure. If a full line starting with at
213 ## least one alpha character was read but it was not "proceed" then it
214 ## should abort without performing its action. For any other failure it
215 ## may make its own decision whether or not to proceed as the controller has
218 ## 6) The supp now performs its throttled action.
220 ## 7) The supp now closes its control descriptor (if it hasn't already in the
221 ## case of (5a)) and exits -- in the case of a socket, the other end receives
222 ## notification that the socket has been closed (read EOF). In the case of
223 ## a pipe the other end receives a SIGCHLD (multiple processes have a hold
224 ## of the other end of the pipe, so it will not reaach EOF by the supp's
225 ## exit in that case).
228 # keys are class names, values are hash refs with these fields:
229 # 'maxproc' => integer; maximum number of allowed supplicants (the sum of how
230 # many may be queued waiting plus how many may be
231 # concurrently active) with 0 meaning no limit.
232 # 'maxjobs' => integer; how many supplicants may proceed simultaneously a value
233 # of 0 is unlimited but the number of concurrent
234 # supplicants will always be limited to no more than
235 # the 'maxproc' value (if > 0) no matter what the
236 # 'maxjobs' value is.
237 # 'total' -> integer; the total number of pids belonging to this clase that
238 # can currently be found in %pid.
239 # 'active' -> integer; the number of currently active supplicants which should
240 # be the same as (the number of elements of %pid with a
241 # matching class name) - (number of my class in @queue).
242 # 'interval' -> integer; minimum number of seconds between 'proceed' responses
243 # or SIGUSR1 signals to members of this class.
244 # 'lastqueue' -> time; last time a supplicant was successfully queued.
245 # 'lastproceed' => time; last time a supplicant was allowed to proceed.
246 # 'lastthrottle' => time; last time a supplicant was throttled
247 # 'lastdied' => time; last time a supplicant in this class died/exited/etc.
250 # keys are pid numbers, values are array refs with these elements:
251 # [0] => name of class (key to classes hash)
252 # [1] => supplicant state (0 => queued, non-zero => time it started running)
253 # [2] => descriptive text (e.g. project name)
256 # minimum number of seconds between any two proceed responses no matter what
257 # class. this takes priority in that it can effectively increase the
258 # class's 'interval' value by delaying proceed notifications if the minimum
259 # interval has not yet elapsed.
262 # fifo of pids awaiting notification as soon as the next $interval elapses
263 # provided interval and maxjobs requirements are satisfied
264 # for the class of the pid that will next be triggered.
267 # time of most recent successful call to AddSupplicant
270 # time of most recent proceed notification
273 # time of most recent throttle
274 my $lastthrottle = 0;
276 # time of most recent removal
279 # lifetime count of how many have been queued
282 # lifetime count of how many have been allowed to proceed
283 my $totalproceed = 0;
285 # lifetime count of how many have been throttled
286 my $totalthrottle = 0;
288 # lifetime count of how many have died
289 # It should always be true that $totalqueued - $totaldied == $curentlyactive
292 # Returns an unordered list of currently registered class names
294 return keys(%classes);
298 return $_[0] if $_[0] >= $_[1];
303 my ($min, $val, $default) = @_;
305 if (defined($val) && $val =~ /^[+-]?\d+$/) {
310 return _max
($min, $ans);
313 # [0] => name of class to find
314 # [1] => if true, create class if it doesn't exist, if a hashref then
315 # it contains initial values for maxproc, maxjobs and interval.
316 # Otherwise maxjobs defaults to max(cpu cores/4, 1), maxprocs
317 # defaults to the max(5, number of cpu cores + maxjobs) and interval
319 # Returns a hash ref with info about the class on success
321 my ($classname, $init) = @_;
322 defined($classname) && $classname =~ /^[a-zA-Z][a-zA-Z0-9._+-]*$/
324 $classname = lc($classname);
326 if ($classes{$classname}) {
327 %info = %{$classes{$classname}};
332 ref($init) eq 'HASH' or $init = {};
333 $newclass{'maxjobs'} = _getnum
(0, $init->{'maxjobs'}, sub{_max
(1, int(::cpucount
() / 4))});
334 $newclass{'maxproc'} = _getnum
(0, $init->{'maxproc'}, sub{_max
(5, ::cpucount
() + $newclass{'maxjobs'})});
335 $newclass{'interval'} = _getnum
(0, $init->{'interval'}, sub{1});
336 $newclass{'total'} = 0;
337 $newclass{'active'} = 0;
338 $newclass{'lastqueue'} = 0;
339 $newclass{'lastproceed'} = 0;
340 $newclass{'lastthrottle'} = 0;
341 $newclass{'lastdied'} = 0;
342 $classes{$classname} = \
%newclass;
347 # [0] => pid to look up
348 # Returns () if not found otherwise ($classname, $timestarted, $description)
349 # Where $timestarted will be 0 if it's still queued otherwise a time() value
352 return () unless exists $pid{$pid};
353 return @
{$pid{$pid}};
356 # Returns array of pid numbers that are currently running sorted
357 # by time started (oldest to newest). Can return an empty array.
359 return sort({ ${$pid{$a}}[1] <=> ${$pid{$b}}[1] }
360 grep({ ${$pid{$_}}[1] } keys(%pid)));
363 # Returns a hash with various about the current state
364 # 'interval' => global minimum interval between proceeds
365 # 'active' => how many pids are currently queued + how many are running
366 # 'queue' => how many pids are currently queued
367 # 'lastqueue' => time (epoch seconds) of last queue
368 # 'lastproceed' => time (epoch seconds) of last proceed
369 # 'lastthrottle' => time (epoch seconds) of last throttle
370 # 'lastdied' => time (epoch seconds) of last removal
371 # 'totalqueue' => lifetime total number of processes queued
372 # 'totalproceed' => lifetime total number of processes proceeded
373 # 'totalthrottle' => lifetime total number of processes throttled
374 # 'totaldied' => lifetime total number of removed processes
377 interval
=> $interval,
378 active
=> scalar(keys(%pid)) - scalar(@queue),
379 queue
=> scalar(@queue),
380 lastqueue
=> $lastqueue,
381 lastproceed
=> $lastproceed,
382 lastthrottle
=> $lastthrottle,
383 lastdied
=> $lastdied,
384 totalqueue
=> $totalqueue,
385 totalproceed
=> $totalproceed,
386 totalthrottle
=> $totalthrottle,
387 totaldied
=> $totaldied
391 # with no args get the global interval
392 # with one arg set it, returns previous value if set
395 $interval = 0 + $_[0] if defined($_[0]) && $_[0] =~ /^\d+$/;
399 sub RemoveSupplicant
;
401 # Perform queue service (i.e. send SIGUSR1 to any eligible queued process)
402 # Returns minimum interval until next proceed is possible
403 # Returns undef if there's nothing waiting to proceed or
404 # the 'maxjobs' limits have been reached for all queued items (in which
405 # case it won't be possible to proceed until one of them exits, hence undef)
406 # This is called automatially by AddSupplicant and RemoveSupplicant
409 return undef unless @queue; # if there's nothing queued, nothing to do
411 my $min = _max
(0, $interval - ($now - $lastproceed));
412 my $classmin = undef;
413 my $classchecked = 0;
415 my $classcount = scalar(keys(%classes));
416 for (my $i=0; $i <= $#queue && $classchecked < $classcount; ++$i) {
417 my $pid = $queue[$i];
418 my $procinfo = $pid{$pid};
420 RemoveSupplicant
($pid, 1);
423 my $classinfo = $classes{$$procinfo[0]};
425 RemoveSupplicant
($pid, 1);
428 if (!$seenclass{$$procinfo[0]}) {
429 $seenclass{$$procinfo[0]} = 1;
431 if (!$classinfo->{'maxjobs'} || $classinfo->{'active'} < $classinfo->{'maxjobs'}) {
432 my $cmin = _max
(0, $classinfo->{'interval'} - ($now - $classinfo->{'lastproceed'}));
433 if (!$cmin && !$min) {
435 $$procinfo[1] = $now;
436 splice(@queue, $i, 1);
439 $classinfo->{'lastproceed'} = $now;
440 ++$classinfo->{'active'};
441 kill("USR1", $pid) or RemoveSupplicant
($pid, 1);
444 $classmin = $cmin unless defined($classmin) && $classmin < $cmin;
448 return defined($classmin) ? _max
($min, $classmin) : undef;
451 # $1 => pid to add (must not already be in %pids)
452 # $2 => class name (must exist)
453 # Returns -1 if no such class or pid already present or invalid
454 # Returns 0 if added successfully (and possibly already SIGUSR1'd)
455 # Return 1 if throttled and cannot be added
457 my ($pid, $classname, $text, $noservice) = @_;
458 return -1 unless $pid && $pid =~ /^[1-9][0-9]*$/;
460 kill(0, $pid) or return -1;
461 my $classinfo = $classes{$classname};
462 return -1 unless $classinfo;
463 return -1 if $pid{$pid};
464 $text = '' unless defined($text);
466 if ($classinfo->{'maxproc'} && $classinfo->{'total'} >= $classinfo->{'maxproc'}) {
468 $lastthrottle = $now;
469 $classinfo->{'lastthrottle'} = $now;
474 $pid{$pid} = [$classname, 0, $text];
475 ++$classinfo->{'total'};
476 $classinfo->{'lastqueue'} = $now;
478 ServiceQueue
unless $noservice;
482 # $1 => pid to remove (died, killed, exited normally, doesn't matter)
483 # Returns 0 if removed
484 # Returns -1 if unknown pid or other error during removal
485 sub RemoveSupplicant
{
486 my ($pid, $noservice) = @_;
487 return -1 unless defined($pid) && $pid =~ /^\d+$/;
489 my $pidinfo = $pid{$pid};
490 $pidinfo or return -1;
496 for (my $i=0; $i<=$#queue; ++$i) {
497 if ($queue[$i] == $pid) {
498 splice(@queue, $i, 1);
503 my $classinfo = $classes{$$pidinfo[0]};
504 ServiceQueue
, return -1 unless $classinfo;
505 --$classinfo->{'active'} if $$pidinfo[1];
506 --$classinfo->{'total'};
507 $classinfo->{'lastdied'} = $now;
508 ServiceQueue
unless $noservice;
524 # http://pubs.opengroup.org/onlinepubs/000095399/utilities/trap.html
537 while (($waitedpid = waitpid(-1, WNOHANG
)) > 0) {
538 my $code = $?
& 0xffff;
539 $idlestart = time if !--$children;
541 if (!($code & 0xff)) {
542 $codemsg = " with exit code ".($code >> 8) if $code;
543 } elsif ($code & 0x7f) {
544 my $signum = ($code & 0x7f);
545 $codemsg = " with signal ".
546 ($signame{$signum}?
$signame{$signum}:$signum);
548 logmsg
"reaped $waitedpid$codemsg";
549 push(@reapedpids, $waitedpid);
551 $SIG{CHLD
} = \
&REAPER
; # loathe sysV
554 $SIG{CHLD
} = \
&REAPER
; # Apollo 440
556 my ($piperead, $pipewrite);
561 if (not defined $pid) {
562 logmsg
"cannot fork: $!";
565 $idlestart = time if !++$children;
568 return; # I'm the parent
571 close(Server
) unless fileno(Server
) == 0;
573 $SIG{'CHLD'} = sub {};
575 open STDIN
, "+<&Client" or die "can't dup client to stdin";
584 sub request_throttle
{
585 use POSIX
qw(sigprocmask sigsuspend SIG_SETMASK);
586 my $classname = shift;
589 Throttle
::GetClassInfo
($classname)
590 or return -1; # no such throttle class
596 my $setempty = POSIX
::SigSet
->new;
597 my $setfull = POSIX
::SigSet
->new;
598 $setempty->emptyset();
600 $SIG{'TERM'} = sub {$throttled = 1};
601 $SIG{'USR1'} = sub {$proceed = 1};
602 $SIG{'USR2'} = sub {$error = 1};
603 $SIG{'PIPE'} = sub {$controldead = 1};
604 $SIG{'ALRM'} = sub {};
606 # After writing we can expect a SIGTERM, SIGUSR1 or SIGUSR2
607 print $pipewrite "\nthrottle $$ $classname $text\n";
608 my $old = POSIX
::SigSet
->new;
609 sigprocmask
(SIG_SETMASK
, $setfull, $old);
610 until ($controldead || $throttled || $proceed || $error) {
612 sigsuspend
($setempty);
614 sigprocmask
(SIG_SETMASK
, $setempty, $old);
615 print $pipewrite "\nkeepalive $$\n";
616 sigprocmask
(SIG_SETMASK
, $setfull, $old);
618 sigprocmask
(SIG_SETMASK
, $setempty, $old);
619 $SIG{'TERM'} = "DEFAULT";
620 $SIG{'USR1'} = "DEFAULT";
621 $SIG{'USR2'} = "DEFAULT";
622 $SIG{'ALRM'} = "DEFAULT";
623 $SIG{'PIPE'} = "DEFAULT";
636 Girocco
::Project
::does_exist
($name, 1) or die "no such project: $name";
638 eval {$proj = Girocco
::Project
->load($name)};
639 if (!$proj && Girocco
::Project
::does_exist
($name, 1)) {
640 # If the .clone_in_progress file exists, but the .clonelog does not
641 # and neither does the .clone_failed, be helpful and touch the
642 # .clone_failed file so that the mirror can be restarted
643 my $projdir = $Girocco::Config
::reporoot
."/$name.git";
644 if (-d
"$projdir" && -f
"$projdir/.clone_in_progress" && ! -f
"$projdir/.clonelog" && ! -f
"$projdir/.clone_failed") {
645 open X
, '>', "$projdir/.clone_failed" and close(X
);
648 $proj or die "failed to load project $name";
649 $proj->{clone_in_progress
} or die "project $name is not marked for cloning";
650 $proj->{clone_logged
} and die "project $name is already being cloned";
651 request_throttle
("clone", $name) <= 0 or die "cloning $name aborted (throttled)";
652 statmsg
"cloning $name";
653 open STDOUT
, '>', "$Girocco::Config::reporoot/$name.git/.clonelog" or die "cannot open clonelog: $!";
654 open STDERR
, ">&STDOUT";
655 open STDIN
, '<', '/dev/null';
656 exec "$Girocco::Config::basedir/taskd/clone.sh", "$name.git" or die "exec failed: $!";
660 return ' -> ' unless $showff && defined($_[0]);
661 my ($git_dir, $old, $new) = @_;
662 return '..' unless defined($old) && defined($new) && $old !~ /^0+$/ && $new !~ /^0+$/ && $old ne $new;
663 # In many cases `git merge-base` is slower than this even if using the
664 # `--is-ancestor` option available since Git 1.8.0, but it's never faster
665 my $ans = get_git
("--git-dir=$git_dir", "rev-list", "-n", "1", "^$new^0", "$old^0", "--") ?
'...' : '..';
666 return wantarray ?
($ans, 1) : $ans;
671 my ($username, $name, $oldrev, $newrev, $ref) = split(/\s+/, $arg);
672 $username && $name && $oldrev && $newrev && $ref or return 0;
673 $oldrev =~ /^[0-9a-f]{40}$/ && $newrev =~ /^[0-9a-f]{40}$/ && $ref =~ m{^refs/} or return 0;
674 $newrev ne $oldrev or return 0;
676 Girocco
::Project
::does_exist
($name, 1) or die "no such project: $name";
677 my $proj = Girocco
::Project
->load($name);
678 $proj or die "failed to load project $name";
679 my $has_notify = $proj->has_notify;
680 my $type = $has_notify ?
"notify" : "change";
683 if ($username && $username !~ /^%.*%$/) {
684 Girocco
::User
::does_exist
($username, 1) or die "no such user: $username";
685 $user = Girocco
::User
->load($username);
686 $user or die "failed to load user $username";
687 } elsif ($username eq "%$name%") {
691 request_throttle
("ref-change", $name) <= 0 or die "ref-change $name aborted (throttled)";
692 my $ind = ref_indicator
($proj->{path
}, $oldrev, $newrev);
693 statmsg
"ref-$type $username $name ($ref: @{[substr($oldrev,0,$abbrev)]}$ind@{[substr($newrev,0,$abbrev)]})";
694 open STDIN
, '<', '/dev/null';
695 Girocco
::Notify
::ref_change
($proj, $user, $ref, $oldrev, $newrev) if $has_notify;
701 my ($username, $name) = split(/\s+/, $arg);
702 $username && $name or return 0;
704 Girocco
::Project
::does_exist
($name, 1) or die "no such project: $name";
705 my $proj = Girocco
::Project
->load($name);
706 $proj or die "failed to load project $name";
707 my $has_notify = $proj->has_notify;
708 my $type = $has_notify ?
"notify" : "change";
711 if ($username && $username !~ /^%.*%$/) {
712 Girocco
::User
::does_exist
($username, 1) or die "no such user: $username";
713 $user = Girocco
::User
->load($username);
714 $user or die "failed to load user $username";
715 } elsif ($username eq "%$name%") {
721 my %deletedheads = ();
722 while (my $change = <STDIN
>) {
723 my ($oldrev, $newrev, $ref) = split(/\s+/, $change);
724 $oldrev =~ /^[0-9a-f]{40}$/ && $newrev =~ /^[0-9a-f]{40}$/ && $ref =~ m{^refs/} or next;
725 if ($ref =~ m{^refs/heads/.}) {
726 if ($oldrev =~ /^0{40}$/) {
727 delete $oldheads{$ref};
728 $deletedheads{$ref} = 1;
729 } elsif ($newrev ne $oldrev || (!exists($oldheads{$ref}) && !$deletedheads{$ref})) {
730 $oldheads{$ref} = $oldrev;
733 $newrev ne $oldrev or next;
734 push(@changes, [$oldrev, $newrev, $ref]);
736 return 0 unless @changes;
737 open STDIN
, '<', '/dev/null';
738 request_throttle
("ref-change", $name) <= 0 or die "ref-changes $name aborted (throttled)";
740 my ($old, $new, $ref, $ran_mail_sh) = @_;
741 my ($ind, $ran_git) = ref_indicator
($proj->{path
}, $old, $new);
742 statmsg
"ref-$type $username $name ($ref: @{[substr($old,0,$abbrev)]}$ind@{[substr($new,0,$abbrev)]})";
743 sleep 1 if $ran_mail_sh || $ran_git;
746 Girocco
::Notify
::ref_changes
($proj, $user, $statproc, \
%oldheads, @changes);
748 &$statproc(@
$_) foreach @changes;
755 my ($pid, $classname, $text) = split(/\s+/, $arg);
756 $pid =~ /^\d+/ or return 0; # invalid pid
758 $pid > 0 or return 0; # invalid pid
759 kill(0, $pid) || $!{EPERM
} or return 0; # no such process
760 Throttle
::GetClassInfo
($classname) or return 0; # no such throttle class
761 defined($text) && $text ne '' or return 0; # no text no service
769 pipe($waker, $wakew) or die "pipe failed: $!";
770 select((select($wakew),$|=1)[0]);
772 $SIG{'TERM'} = sub {$throttled = 1; syswrite($wakew, '!')};
773 $SIG{'USR1'} = sub {$proceed = 1; syswrite($wakew, '!')};
774 $SIG{'USR2'} = sub {$error = 1; syswrite($wakew, '!')};
775 $SIG{'PIPE'} = sub {$controldead = 1; syswrite($wakew, '!')};
776 select((select(STDIN
),$|=1)[0]);
778 logmsg
"throttle $pid $classname $text request";
779 # After writing we can expect a SIGTERM or SIGUSR1
780 print $pipewrite "\nthrottle $$ $classname $text\n";
782 # NOTE: the only way to detect the socket close is to read all the
783 # data until EOF is reached -- recv can be used to peek.
785 vec($v, fileno(STDIN
), 1) = 1;
786 vec($v, fileno($waker), 1) = 1;
787 setnonblock
(\
*STDIN
);
789 until ($controldead || $throttled || $proceed || $error || $suppdead) {
791 select($r=$v, undef, $e=$v, 30);
792 my ($bytes, $discard);
793 do {$bytes = sysread($waker, $discard, 512)} while (defined($bytes) && $bytes > 0);
794 do {$bytes = sysread(STDIN
, $discard, 4096)} while (defined($bytes) && $bytes > 0);
795 $suppdead = 1 unless !defined($bytes) && $!{EAGAIN
};
796 print $pipewrite "\nkeepalive $$\n";
800 if ($throttled && !$suppdead) {
801 print STDIN
"throttled\n";
802 logmsg
"throttle $pid $classname $text throttled";
803 } elsif ($proceed && !$suppdead) {
804 print STDIN
"proceed\n";
805 logmsg
"throttle $pid $classname $text proceed";
806 $SIG{'TERM'} = 'DEFAULT';
807 # Stay alive until the child dies which we detect by EOF on STDIN
808 setnonblock
(\
*STDIN
);
809 until ($controldead || $suppdead) {
811 select($r=$v, undef, $e=$v, 30);
812 my ($bytes, $discard);
813 do {$bytes = sysread($waker, $discard, 512)} while (defined($bytes) && $bytes > 0);
814 do {$bytes = sysread(STDIN
, $discard, 512)} while (defined($bytes) && $bytes > 0);
815 $suppdead = 1 unless !defined($bytes) && $!{EAGAIN
};
816 print $pipewrite "\nkeepalive $$\n";
821 $prefix = "control" if $controldead && !$suppdead;
822 logmsg
"throttle $pid $classname $text ${prefix}died";
827 sub process_pipe_msg
{
828 my ($act, $pid, $cls, $text) = split(/\s+/, $_[0]);
829 if ($act eq "throttle") {
830 $pid =~ /^\d+$/ or return 0;
832 $pid > 0 or return 0; # invalid pid
833 kill(0, $pid) or return 0; # invalid pid
834 defined($cls) && $cls ne "" or kill('USR2', $pid), return 0;
835 defined($text) && $text ne "" or kill('USR2', $pid), return 0;
836 Throttle
::GetClassInfo
($cls) or kill('USR2', $pid), return 0;
837 # the AddSupplicant call could send SIGUSR1 before it returns
838 my $result = Throttle
::AddSupplicant
($pid, $cls, $text);
839 kill('USR2', $pid), return 0 if $result < 0;
840 kill('TERM', $pid), return 0 if $result > 0;
841 # $pid was added to class $cls and will receive SIGUSR1 when
842 # it's time for it to proceed
844 } elsif ($act eq "keepalive") {
845 # nothing to do although we could verify pid is valid and
846 # still in %Throttle::pids and send a SIGUSR2 if not, but
847 # really keepalive should just be ignored.
850 print STDERR
"discarding unknown pipe message \"$_[0]\"\n";
862 # Set to 1 for only syslog output (if enabled by mode)
863 # Set to 2 for only stderr output (if enabled by mode)
864 our $only = 0; # This is a hack
867 use Sys
::Syslog
qw(:DEFAULT :macros);
872 my ($fd, $data) = @_;
874 my $remaining = length($data);
876 my $bytes = POSIX
::write(
878 substr($data, $offset, $remaining),
880 next if !defined($bytes) && $!{EINTR
};
881 croak
"POSIX::write failed: $!" unless defined $bytes;
882 croak
"POSIX::write wrote 0 bytes" unless $bytes;
883 $remaining -= $bytes;
889 use POSIX
qw(STDERR_FILENO);
890 my ($self, $line) = @_;
891 $only = 0 unless defined($only);
892 writeall
(STDERR_FILENO
, $line) if $self->{'stderr'} && $only != 1;
893 substr($line, -1, 1) = '' if substr($line, -1, 1) eq "\n";
894 return unless length($line);
895 syslog
(LOG_NOTICE
, "%s", $line) if $self->{'syslog'} && $only != 2;
899 my $class = shift || 'OStream';
901 my $syslogname = shift;
902 my $syslogfacility = shift;
903 defined($syslogfacility) or $syslogfacility = LOG_USER
;
905 $self->{'syslog'} = $mode > 0;
906 $self->{'stderr'} = $mode <= 0 || $mode > 1;
907 $self->{'lastline'} = '';
908 if ($self->{'syslog'}) {
909 # Some Sys::Syslog have a stupid default setlogsock order
910 eval {Sys
::Syslog
::setlogsock
("native"); 1;} or
911 eval {Sys
::Syslog
::setlogsock
("unix");};
912 openlog
($syslogname, "ndelay,pid", $syslogfacility)
913 or croak
"Sys::Syslog::openlog failed: $!";
915 return bless $self, $class;
918 sub BINMODE
{return 1}
919 sub FILENO
{return undef}
925 my $template = shift;
926 return $self->PRINT(sprintf $template, @_);
931 my $data = join('', $self->{'lastline'}, @_);
933 while ((my $idx = index($data, "\n", $pos)) >= 0) {
935 my $line = substr($data, $pos, $idx - $pos);
936 substr($data, $pos, $idx - $pos) = '';
938 $self->dumpline($line);
940 $self->{'lastline'} = $data;
946 $self->dumpline($self->{'lastline'})
947 if length($self->{'lastline'});
953 my ($scalar, $length, $offset) = @_;
954 $scalar = '' if !defined($scalar);
955 $length = length($scalar) if !defined($length);
956 croak
"OStream::WRITE invalid length $length"
958 $offset = 0 if !defined($offset);
959 $offset += length($scalar) if $offset < 0;
960 croak
"OStream::WRITE invalid write offset"
961 if $offset < 0 || $offset > $length;
962 my $max = length($scalar) - $offset;
963 $length = $max if $length > $max;
964 $self->PRINT(substr($scalar, $offset, $length));
976 my $reexec = Girocco
::ExecUtil
->new;
978 close(DATA
) if fileno(DATA
);
980 Getopt
::Long
::Configure
('bundling');
982 my $parse_res = GetOptions
(
983 'help|?|h' => sub {pod2usage
(-verbose
=> 2, -exitval
=> 0)},
984 'quiet|q' => \
$quiet,
985 'no-quiet' => sub {$quiet = 0},
986 'progress|P' => \
$progress,
987 'inetd|i' => sub {$inetd = 1; $syslog = 1; $quiet = 1;},
988 'idle-timeout|t=i' => \
$idle_timeout,
989 'syslog|s:s' => \
$sfac,
990 'no-syslog' => sub {$syslog = 0; $sfac = undef;},
991 'stderr' => \
$stderr,
992 'abbrev=i' => \
$abbrev,
993 'show-fast-forward-info' => \
$showff,
994 'no-show-fast-forward-info' => sub {$showff = 0},
995 'status-interval=i' => \
$stiv,
996 'idle-status-interval=i' => \
$idiv,
998 $syslog = 1 if defined($sfac);
999 $progress = 1 unless $quiet;
1000 $abbrev = 128 unless $abbrev > 0;
1001 if (defined($idle_timeout)) {
1002 die "--idle-timeout must be a whole number" unless $idle_timeout =~ /^\d+$/;
1003 die "--idle-timeout may not be used without --inetd" unless $inetd;
1005 if (defined($stiv)) {
1006 die "--status-interval must be a whole number" unless $stiv =~ /^\d+$/;
1007 $statusintv = $stiv * 60;
1009 if (defined($idiv)) {
1010 die "--idle-status-interval must be a whole number" unless $idiv =~ /^\d+$/;
1011 $idleintv = $idiv * 60;
1014 open STDOUT
, '>&STDERR' if $inetd;
1016 use Sys
::Syslog
qw();
1019 $sfac = "user" unless defined($sfac) && $sfac ne "";
1022 $sfac = 'LOG_'.$sfac unless $sfac =~ /^LOG_/;
1024 my %badfac = map({("LOG_$_" => 1)}
1025 (qw(PID CONS ODELAY NDELAY NOWAIT PERROR FACMASK NFACILITIES PRIMASK LFMT)));
1026 eval "\$facility = Sys::Syslog::$sfac; 1" or die "invalid syslog facility: $ofac";
1027 die "invalid syslog facility: $ofac"
1028 if ($facility & ~0xf8) || ($facility >> 3) > 23 || $badfac{$sfac};
1029 tie
*STDERR
, 'OStream', $mode, $progname, $facility or die "tie failed";
1032 open STDOUT
, '>', '/dev/null';
1039 my $restart_file = $Girocco::Config
::chroot.'/etc/taskd.restart';
1040 my $restart_active = 1;
1042 open Server
, '<&=0' or die "open: $!";
1043 my $sockname = getsockname Server
;
1044 die "getsockname: $!" unless $sockname;
1045 die "socket already connected! must be 'wait' socket" if getpeername Server
;
1046 die "getpeername: $!" unless $!{ENOTCONN
};
1047 my $st = getsockopt Server
, SOL_SOCKET
, SO_TYPE
;
1048 die "getsockopt(SOL_SOCKET, SO_TYPE): $!" unless $st;
1049 my $socktype = unpack('i', $st);
1050 die "stream socket required" unless defined $socktype && $socktype == SOCK_STREAM
;
1051 die "AF_UNIX socket required" unless sockaddr_family
($sockname) == AF_UNIX
;
1052 $NAME = unpack_sockaddr_un
$sockname;
1053 my $expected = $Girocco::Config
::chroot.'/etc/taskd.socket';
1054 if ($NAME ne $expected) {
1055 $restart_active = 0;
1056 warn "listening on \"$NAME\" but expected \"$expected\", restart file disabled";
1058 my $mode = (stat($NAME))[2];
1059 die "stat: $!" unless $mode;
1061 if (($mode & 0660) != 0660) {
1062 chmod(($mode|0660), $NAME) == 1 or die "chmod ug+rw \"$NAME\" failed: $!";
1065 $NAME = $Girocco::Config
::chroot.'/etc/taskd.socket';
1066 my $uaddr = sockaddr_un
($NAME);
1068 socket(Server
, PF_UNIX
, SOCK_STREAM
, 0) or die "socket failed: $!";
1070 bind(Server
, $uaddr) or die "bind failed: $!";
1071 listen(Server
, SOMAXCONN
) or die "listen failed: $!";
1072 chmod 0666, $NAME or die "chmod failed: $!";
1075 foreach my $throttle (@Girocco::Config
::throttle_classes
, @throttle_defaults) {
1076 my $classname = $throttle->{"name"};
1078 Throttle
::GetClassInfo
($classname, $throttle);
1082 return $_[0] <= $_[1] ?
$_[0] : $_[1];
1085 pipe($piperead, $pipewrite) or die "pipe failed: $!";
1086 setnonblock
($piperead);
1087 select((select($pipewrite), $|=1)[0]);
1089 my $fdset_both = '';
1090 vec($fdset_both, fileno($piperead), 1) = 1;
1091 my $fdset_pipe = $fdset_both;
1092 vec($fdset_both, fileno(Server
), 1) = 1;
1095 my $penaltytime = $t;
1096 my $nextwakeup = $t + 60;
1097 my $nextstatus = undef;
1098 $nextstatus = $t + $statusintv if $statusintv;
1099 if ($restart_active) {
1100 unless (unlink($restart_file) || $!{ENOENT
}) {
1101 $restart_active = 0;
1102 statmsg
"restart file disabled could not unlink \"$restart_file\": $!";
1105 statmsg
"listening on $NAME";
1107 my ($rout, $eout, $nfound);
1111 my $adjustpenalty = sub {
1112 if ($penaltytime < $now) {
1113 my $credit = $now - $penaltytime;
1114 $penalty = $penalty > $credit ?
$penalty - $credit : 0;
1115 $penaltytime = $now;
1118 if (defined($nextstatus) && $now >= $nextstatus) {
1119 unless ($idlestatus && !$children && (!$idleintv || $now - $idlestatus < $idleintv)) {
1120 my $statmsg = "STATUS: $children active";
1125 foreach my $cls (sort(Throttle
::GetClassList
())) {
1126 my $inf = Throttle
::GetClassInfo
($cls);
1127 if ($inf->{'total'}) {
1128 $cnt += $inf->{'total'};
1129 push(@stats, substr(lc($cls),0,1)."=".
1130 $inf->{'total'}.'/'.$inf->{'active'});
1133 push(@stats, "?=".($children-$cnt)) if @stats && $cnt < $children;
1134 $statmsg .= " (".join(" ",@stats).")" if @stats;
1135 foreach (Throttle
::GetRunningPids
()) {
1136 my ($cls, $ts, $desc) = Throttle
::GetPidInfo
($_);
1138 push(@running, "[${cls}::$desc] ".duration
($now-$ts));
1142 $statmsg .= ", idle " . duration
($idlesecs)
1143 if !$children && ($idlesecs = $now - $idlestart) >= 2;
1145 statmsg
"STATUS: currently running: ".join(", ", @running)
1147 $idlestatus = $now if !$children;
1149 $nextstatus += $statusintv while $nextstatus <= $now;
1151 $nextwakeup += 60, $now = time while ($wait = $nextwakeup - $now) <= 0;
1152 $wait = _min
($wait, (Throttle
::ServiceQueue
()||60));
1153 &$adjustpenalty; # this prevents ignoring accept when we shouldn't
1155 if ($penalty <= $maxspawn) {
1156 $fdset = $fdset_both;
1158 $fdset = $fdset_pipe;
1159 $wait = $penalty - $maxspawn if $wait > $penalty - $maxspawn;
1161 $nfound = select($rout=$fdset, undef, $eout=$fdset, $wait);
1162 logmsg
("select failed: $!"), exit(1) unless $nfound >= 0 || $!{EINTR
} || $!{EAGAIN
};
1164 Throttle
::RemoveSupplicant
($reaped) while ($reaped = shift(@reapedpids));
1166 &$adjustpenalty; # this prevents banking credits for elapsed time
1167 if (!$children && !$nfound && $restart_active && -e
$restart_file) {
1169 statmsg
"RESTART: restart requested; now exiting for inetd restart";
1172 statmsg
"RESTART: restart requested; now restarting";
1174 statmsg
"RESTART: continuing after failed restart: $!";
1178 if ($idle_timeout && !$children && !$nfound && $now - $idlestart >= $idle_timeout) {
1179 statmsg
"idle timeout (@{[duration($idle_timeout)]}) exceeded now exiting";
1182 } while $nfound < 1;
1183 my $reout = $rout | $eout;
1184 if (vec($reout, fileno($piperead), 1)) {{
1188 do {$bytes = sysread($piperead, $pipebuff, 512, length($pipebuff))}
1189 while (!defined($bytes) && $!{EINTR
});
1190 last if !defined($bytes) && $!{EAGAIN
};
1191 die "sysread failed: $!" unless defined $bytes;
1192 # since we always keep a copy of $pipewrite open EOF is fatal
1193 die "sysread returned EOF on pipe read" unless $bytes;
1194 $nloff = index($pipebuff, "\n", 0);
1195 if ($nloff < 0 && length($pipebuff) >= 512) {
1197 print STDERR
"discarding 512 bytes of control pipe data with no \\n found\n";
1199 redo unless $nloff >= 0;
1201 last unless $nloff >= 0;
1203 my $msg = substr($pipebuff, 0, $nloff);
1204 substr($pipebuff, 0, $nloff + 1) = '';
1205 $nloff = index($pipebuff, "\n", 0);
1206 process_pipe_msg
($msg) if length($msg);
1207 } while $nloff >= 0;
1210 next unless vec($reout, fileno(Server
), 1);
1211 unless (accept(Client
, Server
)) {
1212 logmsg
"accept failed: $!" unless $!{EINTR
};
1215 logmsg
"connection on $NAME";
1219 $inp = <STDIN
> if defined($inp) && $inp eq "\n";
1220 chomp $inp if defined($inp);
1221 # ignore empty and "nop" connects
1222 defined($inp) && $inp ne "" && $inp ne "nop" or exit 0;
1223 my ($cmd, $arg) = $inp =~ /^([a-zA-Z][a-zA-Z0-9._+-]*)(?:\s+(.*))?$/;
1224 defined($arg) or $arg = '';
1225 if ($cmd eq 'ref-changes') {
1227 } elsif ($cmd eq 'clone') {
1229 } elsif ($cmd eq 'ref-change') {
1231 } elsif ($cmd eq 'throttle') {
1234 statmsg
"ignoring unknown command: $cmd";
1251 taskd.pl - Perform Girocco service tasks
1258 -h | --help detailed instructions
1259 -q | --quiet run quietly
1260 --no-quiet do not run quietly
1261 -P | --progress show occasional status updates
1262 -i | --inetd run as inetd unix stream wait service
1263 implies --quiet --syslog
1264 -t SECONDS | --idle-timeout=SECONDS how long to wait idle before exiting
1266 -s | --syslog[=facility] send messages to syslog instead of
1267 stderr but see --stderr
1269 --no-syslog do not send message to syslog
1270 --stderr always send messages to stderr too
1271 --abbrev=n abbreviate hashes to n (default is 8)
1272 --show-fast-forward-info show fast-forward info (default is on)
1273 --no-show-fast-forward-info disable showing fast-forward info
1274 --status-interval=MINUTES status update interval (default 1)
1275 --idle-status-interval=IDLEMINUTES idle status interval (default 60)
1283 Print the full description of taskd.pl's options.
1287 Suppress non-error messages, e.g. for use when running this task as an inetd
1288 service. Enabled by default by --inetd.
1292 Enable non-error messages. When running in --inetd mode these messages are
1293 sent to STDERR instead of STDOUT.
1297 Show information about the current status of the task operation occasionally.
1298 This is automatically enabled if --quiet is not given.
1302 Run as an inetd wait service. File descriptor 0 must be an unconnected unix
1303 stream socket ready to have accept called on it. To be useful, the unix socket
1304 should be located at "$Girocco::Config::chroot/etc/taskd.socket". A warning
1305 will be issued if the socket is not in the expected location. Socket file
1306 permissions will be adjusted if necessary and if they cannot be taskd.pl will
1307 die. The --inetd option also enables the --quiet and --syslog options but
1308 --no-quiet and --no-syslog may be used to alter that.
1310 The correct specification for the inetd socket is a "unix" protocol "stream"
1311 socket in "wait" mode with user and group writable permissions (0660). An
1312 attempt will be made to alter the socket's file mode if needed and if that
1313 cannot be accomplished taskd.pl will die.
1315 Although most inetd stream services run in nowait mode, taskd.pl MUST be run
1316 in wait mode and will die if the passed in socket is already connected.
1318 Note that while *BSD's inetd happily supports unix sockets (and so does
1319 Darwin's launchd), neither xinetd nor GNU's inetd supports unix sockets.
1320 However, systemd does seem to.
1322 =item B<--idle-timeout=SECONDS>
1324 Only permitted when running in --inetd mode. After SECONDS of inactivity
1325 (i.e. all outstanding tasks have completed and no new requests have come in)
1326 exit normally. The default is no timeout at all (a SECONDS value of 0).
1327 Note that it may actually take up to SECONDS+60 for the idle exit to occur.
1329 =item B<--syslog[=facility]>
1331 Normally error output is sent to STDERR. With this option it's sent to
1332 syslog instead. Note that when running in --inetd mode non-error output is
1333 also affected by this option as it's sent to STDERR in that case. If
1334 not specified, the default for facility is LOG_USER. Facility names are
1335 case-insensitive and the leading 'LOG_' is optional. Messages are logged
1336 with the LOG_NOTICE priority.
1338 =item B<--no-syslog>
1340 Send error message output to STDERR but not syslog.
1344 Always send error message output to STDERR. If --syslog is in effect then
1345 a copy will also be sent to syslog. In --inetd mode this applies to non-error
1350 Abbreviate displayed hash values to only the first n hexadecimal characters.
1351 The default is 8 characters. Set to 0 for no abbreviation at all.
1353 =item B<--show-fast-forward-info>
1355 Instead of showing ' -> ' in ref-change/ref-notify update messages, show either
1356 '..' for a fast-forward, creation or deletion or '...' for non-fast-forward.
1357 This requires running an extra git command for each ref update that is not a
1358 creation or deletion in order to determine whether or not it's a fast forward.
1360 =item B<--no-show-fast-forward-info>
1362 Disable showing of fast-forward information for ref-change/ref-notify update
1363 messages. Instead just show a ' -> ' indicator.
1365 =item B<--status-interval=MINUTES>
1367 If progress is enabled (with --progress or by default if no --inetd or --quiet)
1368 status updates are shown at each MINUTES interval. Setting the interval to 0
1369 disables them entirely even with --progress.
1371 =item B<--idle-status-interval=IDLEMINUTES>
1373 Two consecutive "idle" status updates with no intervening activity will not be
1374 shown unless IDLEMINUTES have elapsed between them. The default is 60 minutes.
1375 Setting the interval to 0 prevents any consecutive idle updates (with no
1376 activity between them) from appearing at all.
1382 taskd.pl is Girocco's service request servant; it listens for service requests
1383 such as new clone requests and ref update notifications and spawns a task to
1384 perform the requested action.