gc: fix yet another typo
[girocco/readme.git] / taskd / taskd.pl
blobdbe2f44a75746e90d218291922ad900cae1f711a
1 #!/usr/bin/perl
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.
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.
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
31 use strict;
32 use warnings;
34 use Getopt::Long;
35 use Pod::Usage;
36 use Socket;
37 use Errno;
38 use Fcntl;
39 use POSIX qw(:sys_wait_h :fcntl_h);
40 use File::Basename;
41 use File::Spec ();
42 use Cwd qw(realpath);
44 use lib "__BASEDIR__";
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;
53 use constant SOCKFDENV => "GIROCCO_TASKD_SOCKET_FD";
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 = (
60 name => "ref-change",
61 maxproc => 0,
62 maxjobs => 1,
63 interval => 2
66 name => "clone",
67 maxproc => 0,
68 maxjobs => 2,
69 interval => 5
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
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;
95 $| = 1;
97 my $progname = basename($0);
98 my $children = 0;
99 my $idlestart = time;
100 my $idlestatus = 0;
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;
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";
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";
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;
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);
160 sub setnoncloexec {
161 my $fd = shift;
162 fcntl($fd, F_SETFD, 0) or die "fcntl failed: $!";
165 sub setcloexec {
166 my $fd = shift;
167 fcntl($fd, F_SETFD, FD_CLOEXEC) or die "fcntl failed: $!";
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: $!";
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: $!";
184 package Throttle;
187 ## Throttle protocol
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)
193 ## 2) The process requesting throttle services will be referred to
194 ## as the supplicant or just "supp" for short.
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.
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.
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.
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.
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.
243 ## 6) The supp now performs its throttled action.
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).
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 class 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 = ();
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 = ();
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;
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 = ();
292 # time of most recent successful call to AddSupplicant
293 my $lastqueue = 0;
295 # time of most recent proceed notification
296 my $lastproceed = 0;
298 # time of most recent throttle
299 my $lastthrottle = 0;
301 # time of most recent removal
302 my $lastdied = 0;
304 # lifetime count of how many have been queued
305 my $totalqueue = 0;
307 # lifetime count of how many have been allowed to proceed
308 my $totalproceed = 0;
310 # lifetime count of how many have been throttled
311 my $totalthrottle = 0;
313 # lifetime count of how many have died
314 # It should always be true that $totalqueued - $totaldied == $curentlyactive
315 my $totaldied = 0;
317 # Returns an unordered list of currently registered class names
318 sub GetClassList {
319 return keys(%classes);
322 sub _max {
323 return $_[0] if $_[0] >= $_[1];
324 return $_[1];
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;
335 return _max($min, $ans);
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;
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;
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}};
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)));
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
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;
424 sub RemoveSupplicant;
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;
448 my $classinfo = $classes{$$procinfo[0]};
449 if (!$classinfo) {
450 RemoveSupplicant($pid, 1);
451 goto RETRY;
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;
469 $classmin = $cmin unless defined($classmin) && $classmin < $cmin;
473 return defined($classmin) ? _max($min, $classmin) : undef;
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;
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;
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;
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;
537 # Instance Methods
539 package main;
542 ## ---------
543 ## Functions
544 ## ---------
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',
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);
573 logmsg "reaped $waitedpid$codemsg";
574 push(@reapedpids, $waitedpid);
576 $SIG{CHLD} = \&REAPER; # loathe sysV
579 $SIG{CHLD} = \&REAPER; # Apollo 440
581 my ($piperead, $pipewrite);
582 sub spawn {
583 my $coderef = shift;
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
596 close(Server) unless fileno(Server) == 0;
597 close($piperead);
598 $SIG{'CHLD'} = sub {};
600 open STDIN, "+<&Client" or die "can't dup client to stdin";
601 close(Client);
602 exit &$coderef();
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;
614 Throttle::GetClassInfo($classname)
615 or return -1; # no such throttle class
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 {};
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);
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";
650 my $result = -1;
651 if ($throttled) {
652 $result = 1;
653 } elsif ($proceed) {
654 $result = 0;
656 return $result;
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);
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 my $devnullfd = POSIX::open(File::Spec->devnull, O_RDWR);
679 defined($devnullfd) && $devnullfd >= 0 or die "cannot open /dev/null: $!";
680 POSIX::dup2($devnullfd, 0) or
681 die "cannot dup2 STDIN_FILENO: $!";
682 POSIX::close($devnullfd);
683 my $duperr;
684 open $duperr, '>&2' or
685 die "cannot dup STDERR_FILENO: $!";
686 my $clonelogfd = POSIX::open("$Girocco::Config::reporoot/$name.git/.clonelog", O_WRONLY|O_TRUNC|O_CREAT, 0664);
687 defined($clonelogfd) && $clonelogfd >= 0 or die "cannot open clonelog for writing: $!";
688 POSIX::dup2($clonelogfd, 1) or
689 die "cannot dup2 STDOUT_FILENO: $!";
690 POSIX::dup2($clonelogfd, 2) or
691 POSIX::dup2(fileno($duperr), 2), die "cannot dup2 STDERR_FILENO: $!";
692 POSIX::close($clonelogfd);
693 exec "$Girocco::Config::basedir/taskd/clone.sh", "$name.git" or
694 POSIX::dup2(fileno($duperr), 2), die "exec failed: $!";
697 sub ref_indicator {
698 return ' -> ' unless $showff && defined($_[0]);
699 my ($git_dir, $old, $new) = @_;
700 return '..' unless defined($old) && defined($new) && $old !~ /^0+$/ && $new !~ /^0+$/ && $old ne $new;
701 # In many cases `git merge-base` is slower than this even if using the
702 # `--is-ancestor` option available since Git 1.8.0, but it's never faster
703 my $ans = get_git("--git-dir=$git_dir", "rev-list", "-n", "1", "^$new^0", "$old^0", "--") ? '...' : '..';
704 return wantarray ? ($ans, 1) : $ans;
707 sub ref_change {
708 my ($arg) = @_;
709 my ($username, $name, $oldrev, $newrev, $ref) = split(/\s+/, $arg);
710 $username && $name && $oldrev && $newrev && $ref or return 0;
711 $oldrev =~ /^[0-9a-f]{40}$/ && $newrev =~ /^[0-9a-f]{40}$/ && $ref =~ m{^refs/} or return 0;
712 $newrev ne $oldrev or return 0;
714 Girocco::Project::does_exist($name, 1) or die "no such project: $name";
715 my $proj = Girocco::Project->load($name);
716 $proj or die "failed to load project $name";
717 my $has_notify = $proj->has_notify;
718 my $type = $has_notify ? "notify" : "change";
720 my $user;
721 if ($username && $username !~ /^%.*%$/) {
722 Girocco::User::does_exist($username, 1) or die "no such user: $username";
723 $user = Girocco::User->load($username);
724 $user or die "failed to load user $username";
725 } elsif ($username eq "%$name%") {
726 $username = "-";
729 request_throttle("ref-change", $name) <= 0 or die "ref-change $name aborted (throttled)";
730 my $ind = ref_indicator($proj->{path}, $oldrev, $newrev);
731 statmsg "ref-$type $username $name ($ref: @{[substr($oldrev,0,$abbrev)]}$ind@{[substr($newrev,0,$abbrev)]})";
732 open STDIN, '<', File::Spec->devnull;
733 Girocco::Notify::ref_changes($proj, $user, [$oldrev, $newrev, $ref]) if $has_notify;
734 return 0;
737 sub ref_changes {
738 my ($arg) = @_;
739 my ($username, $name) = split(/\s+/, $arg);
740 $username && $name or return 0;
742 Girocco::Project::does_exist($name, 1) or die "no such project: $name";
743 my $proj = Girocco::Project->load($name);
744 $proj or die "failed to load project $name";
745 my $has_notify = $proj->has_notify;
746 my $type = $has_notify ? "notify" : "change";
748 my $user;
749 if ($username && $username !~ /^%.*%$/) {
750 Girocco::User::does_exist($username, 1) or die "no such user: $username";
751 $user = Girocco::User->load($username);
752 $user or die "failed to load user $username";
753 } elsif ($username eq "%$name%") {
754 $username = "-";
757 my @changes = ();
758 my %oldheads = ();
759 my %deletedheads = ();
760 while (my $change = <STDIN>) {
761 my ($oldrev, $newrev, $ref) = split(/\s+/, $change);
762 $oldrev ne "done" or last;
763 $oldrev =~ /^[0-9a-f]{40}$/ && $newrev =~ /^[0-9a-f]{40}$/ && $ref =~ m{^refs/} or next;
764 if ($ref =~ m{^refs/heads/.}) {
765 if ($oldrev =~ /^0{40}$/) {
766 delete $oldheads{$ref};
767 $deletedheads{$ref} = 1;
768 } elsif ($newrev ne $oldrev || (!exists($oldheads{$ref}) && !$deletedheads{$ref})) {
769 $oldheads{$ref} = $oldrev;
772 $newrev ne $oldrev or next;
773 push(@changes, [$oldrev, $newrev, $ref]);
775 return 0 unless @changes;
776 open STDIN, '<', File::Spec->devnull;
777 request_throttle("ref-change", $name) <= 0 or die "ref-changes $name aborted (throttled)";
778 my $statproc = sub {
779 my ($old, $new, $ref, $ran_mail_sh) = @_;
780 my ($ind, $ran_git) = ref_indicator($proj->{path}, $old, $new);
781 statmsg "ref-$type $username $name ($ref: @{[substr($old,0,$abbrev)]}$ind@{[substr($new,0,$abbrev)]})";
782 if ($ran_mail_sh) {
783 sleep 2;
784 } elsif ($ran_git) {
785 sleep 1;
788 if ($has_notify) {
789 Girocco::Notify::ref_changes($proj, $user, $statproc, \%oldheads, @changes);
790 } else {
791 &$statproc(@$_) foreach @changes;
793 return 0;
796 sub throttle {
797 my ($arg) = @_;
798 my ($pid, $classname, $text) = split(/\s+/, $arg);
799 $pid =~ /^\d+/ or return 0; # invalid pid
800 $pid += 0;
801 $pid > 0 or return 0; # invalid pid
802 kill(0, $pid) || $!{EPERM} or return 0; # no such process
803 Throttle::GetClassInfo($classname) or return 0; # no such throttle class
804 defined($text) && $text ne '' or return 0; # no text no service
806 my $throttled = 0;
807 my $proceed = 0;
808 my $error = 0;
809 my $controldead = 0;
810 my $suppdead = 0;
811 my ($waker, $wakew);
812 pipe($waker, $wakew) or die "pipe failed: $!";
813 select((select($wakew),$|=1)[0]);
814 setnonblock($wakew);
815 $SIG{'TERM'} = sub {$throttled = 1; syswrite($wakew, '!')};
816 $SIG{'USR1'} = sub {$proceed = 1; syswrite($wakew, '!')};
817 $SIG{'USR2'} = sub {$error = 1; syswrite($wakew, '!')};
818 $SIG{'PIPE'} = sub {$controldead = 1; syswrite($wakew, '!')};
819 select((select(STDIN),$|=1)[0]);
821 logmsg "throttle $pid $classname $text request";
822 # After writing we can expect a SIGTERM or SIGUSR1
823 print $pipewrite "\nthrottle $$ $classname $text\n";
825 # NOTE: the only way to detect the socket close is to read all the
826 # data until EOF is reached -- recv can be used to peek.
827 my $v = '';
828 vec($v, fileno(STDIN), 1) = 1;
829 vec($v, fileno($waker), 1) = 1;
830 setnonblock(\*STDIN);
831 setnonblock($waker);
832 until ($controldead || $throttled || $proceed || $error || $suppdead) {
833 my ($r, $e);
834 select($r=$v, undef, $e=$v, 30);
835 my ($bytes, $discard);
836 do {$bytes = sysread($waker, $discard, 512)} while (defined($bytes) && $bytes > 0);
837 do {$bytes = sysread(STDIN, $discard, 4096)} while (defined($bytes) && $bytes > 0);
838 $suppdead = 1 unless !defined($bytes) && $!{EAGAIN};
839 print $pipewrite "\nkeepalive $$\n";
841 setblock(\*STDIN);
843 if ($throttled && !$suppdead) {
844 print STDIN "throttled\n";
845 logmsg "throttle $pid $classname $text throttled";
846 } elsif ($proceed && !$suppdead) {
847 print STDIN "proceed\n";
848 logmsg "throttle $pid $classname $text proceed";
849 $SIG{'TERM'} = 'DEFAULT';
850 # Stay alive until the child dies which we detect by EOF on STDIN
851 setnonblock(\*STDIN);
852 until ($controldead || $suppdead) {
853 my ($r, $e);
854 select($r=$v, undef, $e=$v, 30);
855 my ($bytes, $discard);
856 do {$bytes = sysread($waker, $discard, 512)} while (defined($bytes) && $bytes > 0);
857 do {$bytes = sysread(STDIN, $discard, 512)} while (defined($bytes) && $bytes > 0);
858 $suppdead = 1 unless !defined($bytes) && $!{EAGAIN};
859 print $pipewrite "\nkeepalive $$\n";
861 setblock(\*STDIN);
862 } else {
863 my $prefix = '';
864 $prefix = "control" if $controldead && !$suppdead;
865 logmsg "throttle $pid $classname $text ${prefix}died";
867 exit 0;
870 sub process_pipe_msg {
871 my ($act, $pid, $cls, $text) = split(/\s+/, $_[0]);
872 if ($act eq "throttle") {
873 $pid =~ /^\d+$/ or return 0;
874 $pid += 0;
875 $pid > 0 or return 0; # invalid pid
876 kill(0, $pid) or return 0; # invalid pid
877 defined($cls) && $cls ne "" or kill('USR2', $pid), return 0;
878 defined($text) && $text ne "" or kill('USR2', $pid), return 0;
879 Throttle::GetClassInfo($cls) or kill('USR2', $pid), return 0;
880 # the AddSupplicant call could send SIGUSR1 before it returns
881 my $result = Throttle::AddSupplicant($pid, $cls, $text);
882 kill('USR2', $pid), return 0 if $result < 0;
883 kill('TERM', $pid), return 0 if $result > 0;
884 # $pid was added to class $cls and will receive SIGUSR1 when
885 # it's time for it to proceed
886 return 0;
887 } elsif ($act eq "keepalive") {
888 # nothing to do although we could verify pid is valid and
889 # still in %Throttle::pids and send a SIGUSR2 if not, but
890 # really keepalive should just be ignored.
891 return 0;
893 print STDERR "discarding unknown pipe message \"$_[0]\"\n";
894 return 0;
898 ## -------
899 ## OStream
900 ## -------
903 package OStream;
905 # Set to 1 for only syslog output (if enabled by mode)
906 # Set to 2 for only stderr output (if enabled by mode)
907 our $only = 0; # This is a hack
909 use Carp 'croak';
910 use Sys::Syslog qw(:DEFAULT :macros);
912 sub writeall {
913 my ($fd, $data) = @_;
914 my $offset = 0;
915 my $remaining = length($data);
916 while ($remaining) {
917 my $bytes = POSIX::write(
918 $fd,
919 substr($data, $offset, $remaining),
920 $remaining);
921 next if !defined($bytes) && $!{EINTR};
922 croak "POSIX::write failed: $!" unless defined $bytes;
923 croak "POSIX::write wrote 0 bytes" unless $bytes;
924 $remaining -= $bytes;
925 $offset += $bytes;
929 sub dumpline {
930 use POSIX qw(STDERR_FILENO);
931 my ($self, $line) = @_;
932 $only = 0 unless defined($only);
933 writeall(STDERR_FILENO, $line) if $self->{'stderr'} && $only != 1;
934 substr($line, -1, 1) = '' if substr($line, -1, 1) eq "\n";
935 return unless length($line);
936 syslog(LOG_NOTICE, "%s", $line) if $self->{'syslog'} && $only != 2;
939 sub TIEHANDLE {
940 my $class = shift || 'OStream';
941 my $mode = shift;
942 my $syslogname = shift;
943 my $syslogfacility = shift;
944 defined($syslogfacility) or $syslogfacility = LOG_USER;
945 my $self = {};
946 $self->{'syslog'} = $mode > 0;
947 $self->{'stderr'} = $mode <= 0 || $mode > 1;
948 $self->{'lastline'} = '';
949 if ($self->{'syslog'}) {
950 # Some Sys::Syslog have a stupid default setlogsock order
951 eval {Sys::Syslog::setlogsock("native"); 1;} or
952 eval {Sys::Syslog::setlogsock("unix");};
953 openlog($syslogname, "ndelay,pid", $syslogfacility)
954 or croak "Sys::Syslog::openlog failed: $!";
956 return bless $self, $class;
959 sub BINMODE {return 1}
960 sub FILENO {return undef}
961 sub EOF {return 0}
962 sub CLOSE {return 1}
964 sub PRINTF {
965 my $self = shift;
966 my $template = shift;
967 return $self->PRINT(sprintf $template, @_);
970 sub PRINT {
971 my $self = shift;
972 my $data = join('', $self->{'lastline'}, @_);
973 my $pos = 0;
974 while ((my $idx = index($data, "\n", $pos)) >= 0) {
975 ++$idx;
976 my $line = substr($data, $pos, $idx - $pos);
977 substr($data, $pos, $idx - $pos) = '';
978 $pos = $idx;
979 $self->dumpline($line);
981 $self->{'lastline'} = $data;
982 return 1;
985 sub DESTROY {
986 my $self = shift;
987 $self->dumpline($self->{'lastline'})
988 if length($self->{'lastline'});
989 closelog;
992 sub WRITE {
993 my $self = shift;
994 my ($scalar, $length, $offset) = @_;
995 $scalar = '' if !defined($scalar);
996 $length = length($scalar) if !defined($length);
997 croak "OStream::WRITE invalid length $length"
998 if $length < 0;
999 $offset = 0 if !defined($offset);
1000 $offset += length($scalar) if $offset < 0;
1001 croak "OStream::WRITE invalid write offset"
1002 if $offset < 0 || $offset > $length;
1003 my $max = length($scalar) - $offset;
1004 $length = $max if $length > $max;
1005 $self->PRINT(substr($scalar, $offset, $length));
1006 return $length;
1010 ## ----
1011 ## main
1012 ## ----
1015 package main;
1017 # returns pid of process that will schedule jobd.pl restart on success
1018 # returns 0 if fork or other system call failed with error in $!
1019 # returns undef if jobd.pl does not currently appear to be running (no lockfile)
1020 sub schedule_jobd_restart {
1021 use POSIX qw(_exit setpgid dup2 :fcntl_h);
1022 my $devnull = File::Spec->devnull;
1023 my $newpg = shift;
1024 my $jdlf = "/tmp/jobd-$Girocco::Config::tmpsuffix.lock";
1025 return undef unless -f $jdlf;
1026 my $oldsigchld = $SIG{'CHLD'};
1027 defined($oldsigchld) or $oldsigchld = sub {};
1028 my ($read, $write, $read2, $write2);
1029 pipe($read, $write) or return 0;
1030 select((select($write),$|=1)[0]);
1031 if (!pipe($read2, $write2)) {
1032 local $!;
1033 close $write;
1034 close $read;
1035 return 0;
1037 select((select($write2),$|=1)[0]);
1038 $SIG{'CHLD'} = sub {};
1039 my $retries = 3;
1040 my $child;
1041 while (!defined($child) && $retries--) {
1042 $child = fork;
1043 sleep 1 unless defined($child) || !$retries;
1045 if (!defined($child)) {
1046 local $!;
1047 close $write2;
1048 close $read2;
1049 close $write;
1050 close $read;
1051 $SIG{'CHLD'} = $oldsigchld;
1052 return 0;
1054 # double fork the child
1055 if (!$child) {
1056 close $read2;
1057 my $retries2 = 3;
1058 my $child2;
1059 while (!defined($child2) && $retries2--) {
1060 $child2 = fork;
1061 sleep 1 unless defined($child2) || !$retries2;
1063 if (!defined($child2)) {
1064 my $ec = 0 + $!;
1065 $ec = 255 unless $ec;
1066 print $write2 ":$ec";
1067 close $write2;
1068 _exit 127;
1070 if ($child2) {
1071 # pass new child pid up to parent and exit
1072 print $write2 $child2;
1073 close $write2;
1074 _exit 0;
1075 } else {
1076 # this is the grandchild
1077 close $write2;
1079 } else {
1080 close $write2;
1081 my $result = <$read2>;
1082 close $read2;
1083 chomp $result if defined($result);
1084 if (!defined($result) || $result !~ /^:?\d+$/) {
1085 # something's wrong with the child -- kill it
1086 kill(9, $child) && waitpid($child, 0);
1087 my $oldsigpipe = $SIG{'PIPE'};
1088 # make sure the grandchild, if any,
1089 # doesn't run the success proc
1090 $SIG{'PIPE'} = sub {};
1091 print $write 1;
1092 close $write;
1093 close $read;
1094 $SIG{'PIPE'} = defined($oldsigpipe) ?
1095 $oldsigpipe : 'DEFAULT';
1096 $! = 255;
1097 $SIG{'CHLD'} = $oldsigchld;
1098 return 0;
1100 if ($result =~ /^:(\d+)$/) {
1101 # fork failed in child, there is no grandchild
1102 my $ec = $1;
1103 waitpid($child, 0);
1104 close $write;
1105 close $read;
1106 $! = $ec;
1107 $SIG{'CHLD'} = $oldsigchld;
1108 return 0;
1110 # reap the child and set $child to grandchild's pid
1111 waitpid($child, 0);
1112 $child = $result;
1114 if (!$child) {
1115 # grandchild that actually initiates the jobd.pl restart
1116 close $write;
1117 my $wait = 5;
1118 my $ufd = POSIX::open($devnull, O_RDWR);
1119 if (defined($ufd)) {
1120 dup2($ufd, 0) unless $ufd == 0;
1121 dup2($ufd, 1) unless $ufd == 1;
1122 dup2($ufd, 2) unless $ufd == 2;
1123 POSIX::close($ufd) unless $ufd == 0 || $ufd == 1 || $ufd == 2;
1125 chdir "/";
1126 if ($newpg) {
1127 my $makepg = sub {
1128 my $result = setpgid(0, 0);
1129 if (!defined($result)) {
1130 --$wait;
1131 sleep 1;
1133 $result;
1135 my $result = &$makepg;
1136 defined($result) or $result = &$makepg;
1137 defined($result) or $result = &$makepg;
1138 defined($result) or $result = &$makepg;
1140 sleep $wait;
1141 my $result = <$read>;
1142 close $read;
1143 chomp $result if defined($result);
1144 if (!defined($result) || $result eq 0) {
1145 open JDLF, '+<', $jdlf or _exit(1);
1146 select((select(JDLF),$|=1)[0]);
1147 print JDLF "restart\n";
1148 truncate JDLF, tell(JDLF);
1149 close JDLF;
1151 _exit(0);
1153 close $write;
1154 close $read;
1155 $SIG{'CHLD'} = $oldsigchld;
1156 return $child;
1159 sub cancel_jobd_restart {
1160 my $restarter = shift;
1161 return unless defined($restarter) && $restarter != 0;
1162 return -1 unless kill(0, $restarter);
1163 kill(9, $restarter) or die "failed to kill jobd restarter process (pid $restarter): $!\n";
1164 # we must not waitpid because $restarter was doubly forked and will
1165 # NOT send us a SIGCHLD when it terminates
1166 return $restarter;
1169 my $reexec = Girocco::ExecUtil->new;
1170 my $realpath0 = realpath($0);
1171 chdir "/";
1172 close(DATA) if fileno(DATA);
1173 my $sfac;
1174 Getopt::Long::Configure('bundling');
1175 my ($stiv, $idiv);
1176 my $parse_res = GetOptions(
1177 'help|?|h' => sub {
1178 pod2usage(-verbose => 2, -exitval => 0, -input => $realpath0)},
1179 'quiet|q' => \$quiet,
1180 'no-quiet' => sub {$quiet = 0},
1181 'progress|P' => \$progress,
1182 'inetd|i' => sub {$inetd = 1; $syslog = 1; $quiet = 1;},
1183 'idle-timeout|t=i' => \$idle_timeout,
1184 'daemon' => sub {$daemon = 1; $syslog = 1; $quiet = 1;},
1185 'max-lifetime=i' => \$max_lifetime,
1186 'syslog|s:s' => \$sfac,
1187 'no-syslog' => sub {$syslog = 0; $sfac = undef;},
1188 'stderr' => \$stderr,
1189 'abbrev=i' => \$abbrev,
1190 'show-fast-forward-info' => \$showff,
1191 'no-show-fast-forward-info' => sub {$showff = 0},
1192 'same-pid' => \$same_pid,
1193 'no-same-pid' => sub {$same_pid = 0},
1194 'status-interval=i' => \$stiv,
1195 'idle-status-interval=i' => \$idiv,
1196 ) || pod2usage(-exitval => 2, -input => $realpath0);
1197 $same_pid = !$daemon unless defined($same_pid);
1198 $syslog = 1 if defined($sfac);
1199 $progress = 1 unless $quiet;
1200 $abbrev = 128 unless $abbrev > 0;
1201 pod2usage(-msg => "--inetd and --daemon are incompatible") if ($inetd && $daemon);
1202 if (defined($idle_timeout)) {
1203 die "--idle-timeout must be a whole number\n" unless $idle_timeout =~ /^\d+$/;
1204 die "--idle-timeout may not be used without --inetd\n" unless $inetd;
1206 if (defined($max_lifetime)) {
1207 die "--max-lifetime must be a whole number\n" unless $max_lifetime =~ /^\d+$/;
1208 $max_lifetime += 0;
1210 defined($max_lifetime) or $max_lifetime = 604800; # 1 week
1211 if (defined($stiv)) {
1212 die "--status-interval must be a whole number\n" unless $stiv =~ /^\d+$/;
1213 $statusintv = $stiv * 60;
1215 if (defined($idiv)) {
1216 die "--idle-status-interval must be a whole number\n" unless $idiv =~ /^\d+$/;
1217 $idleintv = $idiv * 60;
1220 open STDIN, '<'.File::Spec->devnull or die "could not redirect STDIN to /dev/null\n" unless $inetd;
1221 open STDOUT, '>&STDERR' if $inetd;
1222 if ($syslog) {
1223 use Sys::Syslog qw();
1224 my $mode = 1;
1225 ++$mode if $stderr;
1226 $sfac = "user" unless defined($sfac) && $sfac ne "";
1227 my $ofac = $sfac;
1228 $sfac = uc($sfac);
1229 $sfac = 'LOG_'.$sfac unless $sfac =~ /^LOG_/;
1230 my $facility;
1231 my %badfac = map({("LOG_$_" => 1)}
1232 (qw(PID CONS ODELAY NDELAY NOWAIT PERROR FACMASK NFACILITIES PRIMASK LFMT)));
1233 eval "\$facility = Sys::Syslog::$sfac; 1" or die "invalid syslog facility: $ofac\n";
1234 die "invalid syslog facility: $ofac\n"
1235 if ($facility & ~0xf8) || ($facility >> 3) > 23 || $badfac{$sfac};
1236 tie *STDERR, 'OStream', $mode, $progname, $facility or die "tie failed";
1238 if ($quiet) {
1239 open STDOUT, '>', File::Spec->devnull;
1240 } elsif ($inetd) {
1241 *STDOUT = *STDERR;
1244 my ($NAME, $INO);
1246 my $restart_file = $Girocco::Config::chroot.'/etc/taskd.restart';
1247 my $restart_active = 1;
1248 my $resumefd = $ENV{(SOCKFDENV)};
1249 delete $ENV{(SOCKFDENV)};
1250 if (defined($resumefd)) {{
1251 unless ($resumefd =~ /^(\d+)(?::(-?\d+))?$/) {
1252 warn "ignoring invalid ".SOCKFDENV." environment value (\"$resumefd\") -- bad format\n";
1253 $resumefd = undef;
1254 last;
1256 my $resumeino;
1257 ($resumefd, $resumeino) = ($1, $2);
1258 $resumefd += 0;
1259 unless (isfdopen($resumefd)) {
1260 warn "ignoring invalid ".SOCKFDENV." environment value -- fd \"$resumefd\" not open\n";
1261 $resumefd = undef;
1262 last;
1264 unless ($inetd) {
1265 unless (defined($resumeino)) {
1266 warn "ignoring invalid ".SOCKFDENV." environment value (\"$resumefd\") -- missing inode\n";
1267 POSIX::close($resumefd);
1268 $resumefd = undef;
1269 last;
1271 $resumeino += 0;
1272 my $sockloc = $Girocco::Config::chroot.'/etc/taskd.socket';
1273 my $slinode = (stat($sockloc))[1];
1274 unless (defined($slinode) && -S _) {
1275 warn "ignoring ".SOCKFDENV." environment value; socket file does not exist: $sockloc\n";
1276 POSIX::close($resumefd);
1277 $resumefd = undef;
1278 last;
1280 open Test, "<&$resumefd" or die "open: $!";
1281 my $sockname = getsockname Test;
1282 my $sockpath;
1283 $sockpath = unpack_sockaddr_un $sockname if $sockname && sockaddr_family($sockname) == AF_UNIX;
1284 close Test;
1285 if (!defined($resumeino) || !defined($sockpath) || $resumeino != $slinode || realpath($sockloc) ne realpath($sockpath)) {
1286 warn "ignoring ".SOCKFDENV." environment value; does not match socket file: $sockloc\n";
1287 POSIX::close($resumefd);
1288 $resumefd = undef;
1290 $INO = $resumeino;
1293 if ($inetd || defined($resumefd)) {
1294 my $fdopen = defined($resumefd) ? $resumefd : 0;
1295 open Server, "<&=$fdopen" or die "open: $!";
1296 setcloexec(\*Server) if $fdopen > $^F;
1297 my $sockname = getsockname Server;
1298 die "getsockname: $!" unless $sockname;
1299 die "socket already connected! must be 'wait' socket\n" if getpeername Server;
1300 die "getpeername: $!" unless $!{ENOTCONN};
1301 my $st = getsockopt Server, SOL_SOCKET, SO_TYPE;
1302 die "getsockopt(SOL_SOCKET, SO_TYPE): $!" unless $st;
1303 my $socktype = unpack('i', $st);
1304 die "stream socket required\n" unless defined $socktype && $socktype == SOCK_STREAM;
1305 die "AF_UNIX socket required\n" unless sockaddr_family($sockname) == AF_UNIX;
1306 $NAME = unpack_sockaddr_un $sockname;
1307 my $expected = $Girocco::Config::chroot.'/etc/taskd.socket';
1308 if (realpath($NAME) ne realpath($expected)) {
1309 $restart_active = 0;
1310 warn "listening on \"$NAME\" but expected \"$expected\", restart file disabled\n";
1312 my $mode = (stat($NAME))[2];
1313 die "stat: $!" unless $mode;
1314 $mode &= 07777;
1315 if (($mode & 0660) != 0660) {
1316 chmod(($mode|0660), $NAME) == 1 or die "chmod ug+rw \"$NAME\" failed: $!";
1318 } else {
1319 $NAME = $Girocco::Config::chroot.'/etc/taskd.socket';
1320 my $uaddr = sockaddr_un($NAME);
1322 socket(Server, PF_UNIX, SOCK_STREAM, 0) or die "socket failed: $!";
1323 die "already exists but not a socket: $NAME\n" if -e $NAME && ! -S _;
1324 if (-e _) {
1325 # Do not unlink another instance's active listen socket!
1326 socket(my $sfd, PF_UNIX, SOCK_STREAM, 0) or die "socket failed: $!";
1327 connect($sfd, $uaddr) || $!{EPROTOTYPE} and
1328 die "Live socket '$NAME' exists. Please make sure no other instance of taskd is running.\n";
1329 close($sfd);
1330 unlink($NAME);
1332 bind(Server, $uaddr) or die "bind failed: $!";
1333 listen(Server, SOMAXCONN) or die "listen failed: $!";
1334 chmod 0666, $NAME or die "chmod failed: $!";
1335 $INO = (stat($NAME))[1] or die "stat failed: $!";
1338 foreach my $throttle (@Girocco::Config::throttle_classes, @throttle_defaults) {
1339 my $classname = $throttle->{"name"};
1340 $classname or next;
1341 Throttle::GetClassInfo($classname, $throttle);
1344 sub _min {
1345 return $_[0] <= $_[1] ? $_[0] : $_[1];
1348 pipe($piperead, $pipewrite) or die "pipe failed: $!";
1349 setnonblock($piperead);
1350 select((select($pipewrite), $|=1)[0]);
1351 my $pipebuff = '';
1352 my $fdset_both = '';
1353 vec($fdset_both, fileno($piperead), 1) = 1;
1354 my $fdset_pipe = $fdset_both;
1355 vec($fdset_both, fileno(Server), 1) = 1;
1356 my $penalty = 0;
1357 my $t = time;
1358 my $penaltytime = $t;
1359 my $nextwakeup = $t + 60;
1360 my $nextstatus = undef;
1361 $nextstatus = $t + $statusintv if $statusintv;
1362 if ($restart_active) {
1363 unless (unlink($restart_file) || $!{ENOENT}) {
1364 $restart_active = 0;
1365 statmsg "restart file disabled could not unlink \"$restart_file\": $!";
1368 daemon(1, 1) or die "failed to daemonize: $!\n" if $daemon;
1369 my $starttime = time;
1370 my $endtime = $max_lifetime ? $starttime + $max_lifetime : 0;
1371 statmsg "listening on $NAME";
1372 while (1) {
1373 my ($rout, $eout, $nfound);
1374 do {
1375 my $wait;
1376 my $now = time;
1377 my $adjustpenalty = sub {
1378 if ($penaltytime < $now) {
1379 my $credit = $now - $penaltytime;
1380 $penalty = $penalty > $credit ? $penalty - $credit : 0;
1381 $penaltytime = $now;
1384 if (defined($nextstatus) && $now >= $nextstatus) {
1385 unless ($idlestatus && !$children && (!$idleintv || $now - $idlestatus < $idleintv)) {
1386 my $statmsg = "STATUS: $children active";
1387 my @running = ();
1388 if ($children) {
1389 my @stats = ();
1390 my $cnt = 0;
1391 foreach my $cls (sort(Throttle::GetClassList())) {
1392 my $inf = Throttle::GetClassInfo($cls);
1393 if ($inf->{'total'}) {
1394 $cnt += $inf->{'total'};
1395 push(@stats, substr(lc($cls),0,1)."=".
1396 $inf->{'total'}.'/'.$inf->{'active'});
1399 push(@stats, "?=".($children-$cnt)) if @stats && $cnt < $children;
1400 $statmsg .= " (".join(" ",@stats).")" if @stats;
1401 foreach (Throttle::GetRunningPids()) {
1402 my ($cls, $ts, $desc) = Throttle::GetPidInfo($_);
1403 next unless $ts;
1404 push(@running, "[${cls}::$desc] ".duration($now-$ts));
1407 my $idlesecs;
1408 $statmsg .= ", idle " . duration($idlesecs)
1409 if !$children && ($idlesecs = $now - $idlestart) >= 2;
1410 statmsg $statmsg;
1411 statmsg "STATUS: currently running: ".join(", ", @running)
1412 if @running;
1413 $idlestatus = $now if !$children;
1415 $nextstatus += $statusintv while $nextstatus <= $now;
1417 $nextwakeup += 60, $now = time while ($wait = $nextwakeup - $now) <= 0;
1418 $wait = _min($wait, (Throttle::ServiceQueue()||60));
1419 &$adjustpenalty; # this prevents ignoring accept when we shouldn't
1420 my $fdset;
1421 if ($penalty <= $maxspawn) {
1422 $fdset = $fdset_both;
1423 } else {
1424 $fdset = $fdset_pipe;
1425 $wait = $penalty - $maxspawn if $wait > $penalty - $maxspawn;
1427 $nfound = select($rout=$fdset, undef, $eout=$fdset, $wait);
1428 logmsg("select failed: $!"), exit(1) unless $nfound >= 0 || $!{EINTR} || $!{EAGAIN};
1429 my $reaped;
1430 Throttle::RemoveSupplicant($reaped) while ($reaped = shift(@reapedpids));
1431 $now = time;
1432 &$adjustpenalty; # this prevents banking credits for elapsed time
1433 if (!$children && !$nfound && $restart_active && (($endtime && $now >= $endtime) || -e $restart_file)) {
1434 statmsg "RESTART: restart requested; max lifetime ($max_lifetime) exceeded" if $endtime && $now >= $endtime;
1435 $SIG{CHLD} = sub {};
1436 my $restarter = schedule_jobd_restart($inetd);
1437 if (defined($restarter) && !$restarter) {
1438 statmsg "RESTART: restart requested; retrying failed scheduling of jobd restart: $!";
1439 sleep 2; # *cough*
1440 $restarter = schedule_jobd_restart;
1441 if (!defined($restarter)) {
1442 statmsg "RESTART: restart requested; reschedule skipped jobd no longer running";
1443 } elsif (defined($restarter) && !$restarter) {
1444 statmsg "RESTART: restart requested; retry of jobd restart scheduling failed, skipping jobd restart: $!";
1445 $restarter = undef;
1448 if ($inetd) {
1449 statmsg "RESTART: restart requested; now exiting for inetd restart";
1450 statmsg "RESTART: restart requested; jobd restart scheduled in 5 seconds" if $restarter;
1451 sleep 2; # *cough*
1452 exit 0;
1453 } else {
1454 statmsg "RESTART: restart requested; now restarting";
1455 statmsg "RESTART: restart requested; jobd restart scheduled in 5 seconds" if $restarter;
1456 setnoncloexec(\*Server);
1457 $reexec->setenv(SOCKFDENV, fileno(Server).":$INO");
1458 $reexec->reexec($same_pid);
1459 setcloexec(\*Server) if fileno(Server) > $^F;
1460 statmsg "RESTART: continuing after failed restart: $!";
1461 chdir "/";
1462 cancel_jobd_restart($restarter) if $restarter;
1463 statmsg "RESTART: scheduled jobd restart has been cancelled" if $restarter;
1464 $SIG{CHLD} = \&REAPER;
1467 if ($idle_timeout && !$children && !$nfound && $now - $idlestart >= $idle_timeout) {
1468 statmsg "idle timeout (@{[duration($idle_timeout)]}) exceeded now exiting";
1469 exit 0;
1471 } while $nfound < 1;
1472 my $reout = $rout | $eout;
1473 if (vec($reout, fileno($piperead), 1)) {{
1474 my $nloff = -1;
1476 my $bytes;
1477 do {$bytes = sysread($piperead, $pipebuff, 512, length($pipebuff))}
1478 while (!defined($bytes) && $!{EINTR});
1479 last if !defined($bytes) && $!{EAGAIN};
1480 die "sysread failed: $!" unless defined $bytes;
1481 # since we always keep a copy of $pipewrite open EOF is fatal
1482 die "sysread returned EOF on pipe read" unless $bytes;
1483 $nloff = index($pipebuff, "\n", 0);
1484 if ($nloff < 0 && length($pipebuff) >= 512) {
1485 $pipebuff = '';
1486 print STDERR "discarding 512 bytes of control pipe data with no \\n found\n";
1488 redo unless $nloff >= 0;
1490 last unless $nloff >= 0;
1491 do {
1492 my $msg = substr($pipebuff, 0, $nloff);
1493 substr($pipebuff, 0, $nloff + 1) = '';
1494 $nloff = index($pipebuff, "\n", 0);
1495 process_pipe_msg($msg) if length($msg);
1496 } while $nloff >= 0;
1497 redo;
1499 next unless vec($reout, fileno(Server), 1);
1500 unless (accept(Client, Server)) {
1501 logmsg "accept failed: $!" unless $!{EINTR};
1502 next;
1504 logmsg "connection on $NAME";
1505 ++$penalty;
1506 spawn sub {
1507 my $inp = <STDIN>;
1508 $inp = <STDIN> if defined($inp) && $inp eq "\n";
1509 chomp $inp if defined($inp);
1510 # ignore empty and "nop" connects
1511 defined($inp) && $inp ne "" && $inp ne "nop" or exit 0;
1512 my ($cmd, $arg) = $inp =~ /^([a-zA-Z][a-zA-Z0-9._+-]*)(?:\s+(.*))?$/;
1513 defined($arg) or $arg = '';
1514 if ($cmd eq 'ref-changes') {
1515 ref_changes($arg);
1516 } elsif ($cmd eq 'clone') {
1517 clone($arg);
1518 } elsif ($cmd eq 'ref-change') {
1519 statmsg "processing obsolete ref-change message (please switch to ref-changes)";
1520 ref_change($arg);
1521 } elsif ($cmd eq 'throttle') {
1522 throttle($arg);
1523 } else {
1524 statmsg "ignoring unknown command: $cmd";
1525 exit 3;
1528 close Client;
1532 ## -------------
1533 ## Documentation
1534 ## -------------
1537 __END__
1539 =head1 NAME
1541 taskd.pl - Perform Girocco service tasks
1543 =head1 SYNOPSIS
1545 taskd.pl [options]
1547 Options:
1548 -h | --help detailed instructions
1549 -q | --quiet run quietly
1550 --no-quiet do not run quietly
1551 -P | --progress show occasional status updates
1552 -i | --inetd run as inetd unix stream wait service
1553 implies --quiet --syslog
1554 -t SECONDS | --idle-timeout=SECONDS how long to wait idle before exiting
1555 requires --inetd
1556 --daemon become a background daemon
1557 implies --quiet --syslog
1558 --max-lifetime=SECONDS how long before graceful restart
1559 default is 1 week, 0 disables
1560 -s | --syslog[=facility] send messages to syslog instead of
1561 stderr but see --stderr
1562 enabled by --inetd
1563 --no-syslog do not send message to syslog
1564 --stderr always send messages to stderr too
1565 --abbrev=n abbreviate hashes to n (default is 8)
1566 --show-fast-forward-info show fast-forward info (default is on)
1567 --no-show-fast-forward-info disable showing fast-forward info
1568 --same-pid keep same pid during graceful restart
1569 --no-same-pid do not keep same pid on graceful rstrt
1570 --status-interval=MINUTES status update interval (default 1)
1571 --idle-status-interval=IDLEMINUTES idle status interval (default 60)
1573 =head1 OPTIONS
1575 =over 8
1577 =item B<--help>
1579 Print the full description of taskd.pl's options.
1581 =item B<--quiet>
1583 Suppress non-error messages, e.g. for use when running this task as an inetd
1584 service. Enabled by default by --inetd.
1586 =item B<--no-quiet>
1588 Enable non-error messages. When running in --inetd mode these messages are
1589 sent to STDERR instead of STDOUT.
1591 =item B<--progress>
1593 Show information about the current status of the task operation occasionally.
1594 This is automatically enabled if --quiet is not given.
1596 =item B<--inetd>
1598 Run as an inetd wait service. File descriptor 0 must be an unconnected unix
1599 stream socket ready to have accept called on it. To be useful, the unix socket
1600 should be located at "$Girocco::Config::chroot/etc/taskd.socket". A warning
1601 will be issued if the socket is not in the expected location. Socket file
1602 permissions will be adjusted if necessary and if they cannot be taskd.pl will
1603 die. The --inetd option also enables the --quiet and --syslog options but
1604 --no-quiet and --no-syslog may be used to alter that.
1606 The correct specification for the inetd socket is a "unix" protocol "stream"
1607 socket in "wait" mode with user and group writable permissions (0660). An
1608 attempt will be made to alter the socket's file mode if needed and if that
1609 cannot be accomplished taskd.pl will die.
1611 Although most inetd stream services run in nowait mode, taskd.pl MUST be run
1612 in wait mode and will die if the passed in socket is already connected.
1614 Note that while *BSD's inetd happily supports unix sockets (and so does
1615 Darwin's launchd), neither xinetd nor GNU's inetd supports unix sockets.
1616 However, systemd does seem to.
1618 =item B<--idle-timeout=SECONDS>
1620 Only permitted when running in --inetd mode. After SECONDS of inactivity
1621 (i.e. all outstanding tasks have completed and no new requests have come in)
1622 exit normally. The default is no timeout at all (a SECONDS value of 0).
1623 Note that it may actually take up to SECONDS+60 for the idle exit to occur.
1625 =item B<--daemon>
1627 Fork and become a background daemon. Implies B<--syslog> and B<--quiet> (which
1628 can be altered by subsequent B<--no-syslog> and/or B<--no-quiet> options).
1629 Also implies B<--no-same-pid>, but since graceful restarts work by re-exec'ing
1630 taskd.pl with all of its original arguments, using B<--same-pid> won't really
1631 be effective with B<--daemon> since although it will cause the graceful restart
1632 exec to happen from the same pid, when the B<--daemon> option is subsequently
1633 processed it will end up in a new pid anyway.
1635 =item B<--max-lifetime=SECONDS>
1637 After taskd has been running for SECONDS of realtime, it will behave as though
1638 a graceful restart has been requested. A graceful restart takes place the
1639 next time taskd becomes idle (which may require up to 60 seconds to notice).
1640 If jobd is running when a graceful restart occurs, then jabd will also receive
1641 a graceful restart request at that time. The default value is 1 week (604800),
1642 set to 0 to disable.
1644 =item B<--syslog[=facility]>
1646 Normally error output is sent to STDERR. With this option it's sent to
1647 syslog instead. Note that when running in --inetd mode non-error output is
1648 also affected by this option as it's sent to STDERR in that case. If
1649 not specified, the default for facility is LOG_USER. Facility names are
1650 case-insensitive and the leading 'LOG_' is optional. Messages are logged
1651 with the LOG_NOTICE priority.
1653 =item B<--no-syslog>
1655 Send error message output to STDERR but not syslog.
1657 =item B<--stderr>
1659 Always send error message output to STDERR. If --syslog is in effect then
1660 a copy will also be sent to syslog. In --inetd mode this applies to non-error
1661 messages as well.
1663 =item B<--abbrev=n>
1665 Abbreviate displayed hash values to only the first n hexadecimal characters.
1666 The default is 8 characters. Set to 0 for no abbreviation at all.
1668 =item B<--show-fast-forward-info>
1670 Instead of showing ' -> ' in ref-change/ref-notify update messages, show either
1671 '..' for a fast-forward, creation or deletion or '...' for non-fast-forward.
1672 This requires running an extra git command for each ref update that is not a
1673 creation or deletion in order to determine whether or not it's a fast forward.
1675 =item B<--no-show-fast-forward-info>
1677 Disable showing of fast-forward information for ref-change/ref-notify update
1678 messages. Instead just show a ' -> ' indicator.
1680 =item B<--same-pid>
1682 When performing a graceful restart, perform the graceful restart exec from
1683 the same pid rather than switching to a new one. This is implied when
1684 I<--daemon> is I<NOT> used.
1686 =item B<--no-same-pid>
1688 When performing a graceful restart, perform the graceful restart exec after
1689 switching to a new pid. This is implied when I<--daemon> I<IS> used.
1691 =item B<--status-interval=MINUTES>
1693 If progress is enabled (with --progress or by default if no --inetd or --quiet)
1694 status updates are shown at each MINUTES interval. Setting the interval to 0
1695 disables them entirely even with --progress.
1697 =item B<--idle-status-interval=IDLEMINUTES>
1699 Two consecutive "idle" status updates with no intervening activity will not be
1700 shown unless IDLEMINUTES have elapsed between them. The default is 60 minutes.
1701 Setting the interval to 0 prevents any consecutive idle updates (with no
1702 activity between them) from appearing at all.
1704 =back
1706 =head1 DESCRIPTION
1708 taskd.pl is Girocco's service request servant; it listens for service requests
1709 such as new clone requests and ref update notifications and spawns a task to
1710 perform the requested action.
1712 =cut