X-Git-Url: http://git.freeside.biz/gitweb/?a=blobdiff_plain;f=FS%2Fbin%2Ffreeside-queued;h=3a0a9b4e50d76a34b867ebe429f5d612a47bda4a;hb=673b9a458d9138523026963df6fa3b4683e09bae;hp=311fe62f9312349fbe5e83537002d61044e13259;hpb=748983cb2a2cd21a82950c5000088508a416d670;p=freeside.git diff --git a/FS/bin/freeside-queued b/FS/bin/freeside-queued index 311fe62f9..3a0a9b4e5 100644 --- a/FS/bin/freeside-queued +++ b/FS/bin/freeside-queued @@ -1,13 +1,11 @@ #!/usr/bin/perl -w use strict; -use vars qw( $log_file $sigterm $sigint $kids $max_kids ); -use subs qw( _die _logmsg ); -use Fcntl qw(:flock); -use POSIX qw(setsid); -use Date::Format; +use vars qw( $DEBUG $kids $max_kids %kids ); +use POSIX qw(:sys_wait_h); use IO::File; -use FS::UID qw(adminsuidsetup forksuidsetup driver_name dbh); +use FS::UID qw(adminsuidsetup forksuidsetup driver_name dbh myconnect); +use FS::Daemon qw(daemonize1 drop_root logfile daemonize2 sigint sigterm); use FS::Record qw(qsearch qsearchs); use FS::queue; use FS::queue_depend; @@ -18,73 +16,82 @@ use FS::svc_acct; use Net::SSH 0.07; use FS::part_export; +$DEBUG = 0; + $max_kids = '10'; #guess it should be a config file... $kids = 0; my $user = shift or die &usage; -#my $pid_file = "/var/run/freeside-queued.$user.pid"; -my $pid_file = "/var/run/freeside-queued.pid"; - -&daemonize1; - -sub REAPER { my $pid = wait; $SIG{CHLD} = \&REAPER; $kids--; } -$SIG{CHLD} = \&REAPER; - -$sigterm = 0; -$sigint = 0; -$SIG{INT} = sub { warn "SIGINT received; shutting down\n"; $sigint++; }; -$SIG{TERM} = sub { warn "SIGTERM received; shutting down\n"; $sigterm++; }; +warn "starting daemonization (forking)\n" if $DEBUG; +#daemonize1('freeside-queued',$user); #to keep pid files unique w/multi installs +daemonize1('freeside-queued'); -my $freeside_gid = scalar(getgrnam('freeside')) - or die "can't setgid to freeside group\n"; -$) = $freeside_gid; -$( = $freeside_gid; -#if freebsd can't setuid(), presumably it can't setgid() either. grr fleabsd -($(,$)) = ($),$(); -$) = $freeside_gid; +warn "dropping privledges\n" if $DEBUG; +drop_root(); -$> = $FS::UID::freeside_uid; -$< = $FS::UID::freeside_uid; -#freebsd is sofa king broken, won't setuid() -($<,$>) = ($>,$<); -$> = $FS::UID::freeside_uid; $ENV{HOME} = (getpwuid($>))[7]; #for ssh -adminsuidsetup $user; -$log_file = "/usr/local/etc/freeside/queuelog.". $FS::UID::datasrc; +warn "connecting to database\n" if $DEBUG; +$@ = 'not connected'; +while ( $@ ) { + eval { adminsuidsetup $user; }; + if ( $@ ) { + warn $@; + warn "sleeping for reconnect...\n"; + sleep 5; + } +} -&daemonize2; +logfile( "/usr/local/etc/freeside/queuelog.". $FS::UID::datasrc ); -$SIG{__DIE__} = \&_die; -$SIG{__WARN__} = \&_logmsg; +warn "completing daemonization (detaching))\n" if $DEBUG; +daemonize2(); -warn "freeside-queued starting\n"; +#-- my $warnkids=0; while (1) { + &reap_kids; #prevent runaway forking if ( $kids >= $max_kids ) { warn "WARNING: maximum $kids children reached\n" unless $warnkids++; + &reap_kids; sleep 1; #waiting for signals is cheap next; } $warnkids=0; - my $nodepend = driver_name eq 'mysql' - ? '' - : 'AND 0 = ( SELECT COUNT(*) FROM queue_depend'. - ' WHERE queue_depend.jobnum = queue.jobnum ) '; + unless ( dbh && dbh->ping ) { + warn "WARNING: connection to database lost, reconnecting...\n"; + + eval { $FS::UID::dbh = myconnect; }; + + unless ( !$@ && dbh && dbh->ping ) { + warn "WARNING: still no connection to database, sleeping for retry...\n"; + sleep 10; + next; + } else { + warn "WARNING: reconnected to database\n"; + } + } #my($job, $ljob); #{ # my $oldAutoCommit = $FS::UID::AutoCommit; # local $FS::UID::AutoCommit = 0; $FS::UID::AutoCommit = 0; - my $dbh = dbh; - + + #assuming mysql 4.1 w/subqueries now + #my $nodepend = driver_name eq 'mysql' + # ? '' + # : 'AND 0 = ( SELECT COUNT(*) FROM queue_depend'. + # ' WHERE queue_depend.jobnum = queue.jobnum ) '; + my $nodepend = 'AND 0 = ( SELECT COUNT(*) FROM queue_depend'. + ' WHERE queue_depend.jobnum = queue.jobnum ) '; + my $job = qsearchs( 'queue', { 'status' => 'new' }, @@ -93,30 +100,49 @@ while (1) { ? "$nodepend ORDER BY jobnum LIMIT 1 FOR UPDATE" : "$nodepend ORDER BY jobnum FOR UPDATE LIMIT 1" ) or do { - $dbh->commit or die $dbh->errstr; #if $oldAutoCommit; + # if $oldAutoCommit { + dbh->commit or do { + warn "WARNING: database error, closing connection: ". dbh->errstr; + undef $FS::UID::dbh; + next; + }; + # } sleep 5; #connecting to db is expensive next; }; - if ( driver_name eq 'mysql' - && qsearch('queue_depend', { 'jobnum' => $job->jobnum } ) ) { - $dbh->commit or die $dbh->errstr; #if $oldAutoCommit; - sleep 5; #would be better if mysql could do everything in query above - next; - } + #assuming mysql 4.1 w/subqueries now + #if ( driver_name eq 'mysql' + # && qsearch('queue_depend', { 'jobnum' => $job->jobnum } ) ) { + # dbh->commit or die dbh->errstr; #if $oldAutoCommit; + # sleep 5; #would be better if mysql could do everything in query above + # next; + #} my %hash = $job->hash; $hash{'status'} = 'locked'; my $ljob = new FS::queue ( \%hash ); my $error = $ljob->replace($job); - die $error if $error; + if ( $error ) { + warn "WARNING: database error locking job, closing connection: ". + dbh->errstr; + undef $FS::UID::dbh; + next; + } - $dbh->commit or die $dbh->errstr; #if $oldAutoCommit; + # if $oldAutoCommit { + dbh->commit or do { + warn "WARNING: database error, closing connection: ". dbh->errstr; + undef $FS::UID::dbh; + next; + }; + # } $FS::UID::AutoCommit = 1; #} my @args = $ljob->args; + splice @args, 0, 1, $ljob if $args[0] eq '_JOB'; defined( my $pid = fork ) or do { warn "WARNING: can't fork: $!\n"; @@ -131,6 +157,7 @@ while (1) { if ( $pid ) { $kids++; + $kids{$pid} = 1; } else { #kid time #get new db handle @@ -138,8 +165,12 @@ while (1) { forksuidsetup($user); - #auto-use export classes... - if ( $ljob->job =~ /(FS::part_export::\w+)::/ ) { + #auto-use classes... + #if ( $ljob->job =~ /(FS::part_export::\w+)::/ ) { + if ( $ljob->job =~ /(FS::part_export::\w+)::/ + || $ljob->job =~ /(FS::\w+)::/ + ) + { my $class = $1; eval "use $class;"; if ( $@ ) { @@ -155,7 +186,7 @@ while (1) { } my $eval = "&". $ljob->job. '(@args);'; - warn "running $eval"; + warn 'running "&'. $ljob->job. '('. join(', ', @args). ")\n" if $DEBUG; eval $eval; #throw away return value? suppose so if ( $@ ) { warn "job $eval failed"; @@ -174,11 +205,11 @@ while (1) { } } continue { - if ( $sigterm ) { + if ( sigterm() ) { warn "received TERM signal; exiting\n"; exit; } - if ( $sigint ) { + if ( sigint() ) { warn "received INT signal; exiting\n"; exit; } @@ -188,46 +219,14 @@ sub usage { die "Usage:\n\n freeside-queued user\n"; } -sub _die { - my $msg = shift; - unlink $pid_file if -e $pid_file; - _logmsg($msg); -} - -sub _logmsg { - chomp( my $msg = shift ); - my $log = new IO::File ">>$log_file"; - flock($log, LOCK_EX); - seek($log, 0, 2); - print $log "[". time2str("%a %b %e %T %Y",time). "] [$$] $msg\n"; - flock($log, LOCK_UN); - close $log; -} - -sub daemonize1 { - - chdir "/" or die "Can't chdir to /: $!"; - open STDIN, '/dev/null' or die "Can't read /dev/null: $!"; - defined(my $pid = fork) or die "Can't fork: $!"; - if ( $pid ) { - print "freeside-queued started with pid $pid\n"; #logging to $log_file\n"; - exit unless $pid_file; - my $pidfh = new IO::File ">$pid_file" or exit; - print $pidfh "$pid\n"; - exit; +sub reap_kids { + foreach my $pid ( keys %kids ) { + my $kid = waitpid($pid, WNOHANG); + if ( $kid > 0 ) { + $kids--; + delete $kids{$kid}; + } } - #open STDOUT, '>/dev/null' - # or die "Can't write to /dev/null: $!"; - #setsid or die "Can't start a new session: $!"; - #open STDERR, '>&STDOUT' or die "Can't dup stdout: $!"; - -} - -sub daemonize2 { - open STDOUT, '>/dev/null' - or die "Can't write to /dev/null: $!"; - setsid or die "Can't start a new session: $!"; - open STDERR, '>&STDOUT' or die "Can't dup stdout: $!"; } =head1 NAME