4 use vars qw( $DEBUG $kids $max_kids $sleep_time %kids );
5 use POSIX qw(:sys_wait_h);
8 use FS::UID qw(adminsuidsetup forksuidsetup driver_name dbh myconnect);
9 use FS::Daemon qw(daemonize1 drop_root logfile daemonize2 sigint sigterm);
11 use FS::Record qw(qsearch);
16 use FS::Cron::expire_user_pref qw( expire_user_pref );
18 # no autoloading for non-FS classes...
25 &untaint_argv; #what it sounds like (eww)
27 getopts('sn', \%opt );
29 my $user = shift or die &usage;
31 warn "starting daemonization (forking)\n" if $DEBUG;
32 #daemonize1('freeside-queued',$user); #to keep pid files unique w/multi installs
33 daemonize1('freeside-queued');
35 warn "dropping privledges\n" if $DEBUG;
38 $ENV{HOME} = (getpwuid($>))[7]; #for ssh
40 warn "connecting to database\n" if $DEBUG;
41 #$@ = 'not connected';
43 # eval { adminsuidsetup $user; };
46 # warn "sleeping for reconnect...\n";
52 my $log = FS::Log->new('queue');
53 logfile( "%%%FREESIDE_LOG%%%/queuelog.". $FS::UID::datasrc );
55 warn "completing daemonization (detaching))\n" if $DEBUG;
60 my $conf = new FS::Conf;
61 $max_kids = $conf->config('queued-max_kids') || 10;
62 $sleep_time = $conf->config('queued-sleep_time') || 10;
68 #prevent runaway forking
69 if ( $kids >= $max_kids ) {
70 warn "WARNING: maximum $kids children reached\n" unless $warnkids++;
72 expire_user_pref() unless $warnkids % 10;
73 sleep 1; #waiting for signals is cheap
78 unless ( dbh && dbh->ping ) {
79 warn "WARNING: connection to database lost, reconnecting...\n";
81 eval { $FS::UID::dbh = myconnect; };
83 unless ( !$@ && dbh && dbh->ping ) {
84 warn "WARNING: still no connection to database, sleeping for retry...\n";
88 warn "WARNING: reconnected to database\n";
94 # my $oldAutoCommit = $FS::UID::AutoCommit;
95 # local $FS::UID::AutoCommit = 0;
96 $FS::UID::AutoCommit = 0;
98 my $nodepend = 'AND NOT EXISTS( SELECT 1 FROM queue_depend'.
99 ' WHERE queue_depend.jobnum = queue.jobnum )';
101 #anything with a priority goes after stuff without one
102 my $order_by = ' ORDER BY COALESCE(priority,0) ASC, jobnum ASC ';
104 my $limit = $max_kids - $kids;
106 $order_by .= ( driver_name eq 'mysql'
107 ? " LIMIT $limit FOR UPDATE "
108 : " FOR UPDATE LIMIT $limit " );
110 my $hashref = { 'status' => 'new' };
112 $hashref->{'secure'} = 'Y';
113 } elsif ( $opt{'n'} ) {
114 $hashref->{'secure'} = '';
117 #qsearch dies when the db goes away
121 'hashref' => $hashref,
122 'extra_sql' => $nodepend,
123 'order_by' => $order_by,
127 warn "WARNING: error searching for jobs, closing connection: $@";
134 warn "WARNING: database error, closing connection: ". dbh->errstr;
143 foreach my $job ( @jobs ) {
145 my $start_date = time;
147 $log->debug('locking queue job', object => $job);
149 my %hash = $job->hash;
150 $hash{'status'} = 'locked';
151 my $ljob = new FS::queue ( \%hash );
152 my $error = $ljob->replace($job);
154 warn "WARNING: database error locking job, closing connection: ".
161 warn "WARNING: database error, closing connection: ". dbh->errstr;
166 $FS::UID::AutoCommit = 1;
168 my @args = eval { $ljob->args; };
170 warn "WARNING: error retrieving job arguments, closing connection: $@";
174 splice @args, 0, 1, $ljob if $args[0] eq '_JOB';
176 defined( my $pid = fork ) or do {
177 warn "WARNING: can't fork: $!\n";
178 my %hash = $job->hash;
179 $hash{'status'} = 'failed';
180 $hash{'statustext'} = "[freeside-queued] can't fork: $!";
181 my $ljob = new FS::queue ( \%hash );
182 my $error = $ljob->replace($job);
183 die $error if $error; #XXX still dying if we can't fork AND we can't connect to the db
184 next; #don't increment the kid counter
193 $FS::UID::dbh->{InactiveDestroy} = 1;
195 forksuidsetup($user);
197 dbh->{'private_profile'} = {} if UNIVERSAL::can(dbh, 'sprintProfile');
200 if ( $ljob->job =~ /(FS::(part_export|cust_main|cust_pkg|part_pkg|Cron)::\w+)::/
201 || $ljob->job =~ /(FS::\w+)::/
207 warn "job use $class failed";
208 my %hash = $ljob->hash;
209 $hash{'status'} = 'failed';
210 $hash{'statustext'} = $@;
211 my $fjob = new FS::queue( \%hash );
212 my $error = $fjob->replace($ljob);
213 die $error if $error;
218 my $eval = "&". $ljob->job. '(@args);';
219 # don't put @args in the log, may expose passwords
220 $log->info('starting job ('.$ljob->job.')');
221 warn 'running "&'. $ljob->job. '('. join(', ', @args). ")\n" if $DEBUG;
222 local $FS::UID::AutoCommit = 0; # so that we can clean up failures
223 eval $eval; #throw away return value? suppose so
226 my %hash = $ljob->hash;
227 $hash{'statustext'} = $@;
228 if ( $hash{'statustext'} =~ /\/misc\/queued_report/ ) { #use return?
229 $hash{'status'} = 'done';
231 $hash{'status'} = 'failed';
232 warn "job $eval failed";
234 my $fjob = new FS::queue( \%hash );
235 my $error = $fjob->replace($ljob);
236 die $error if $error;
237 dbh->commit; # for the status change only
240 dbh->commit; # for the job itself
243 if ( $ljob->job eq 'FS::cust_main::queued_bill' ) {
244 my $queue_stat = new FS::queue_stat {
245 'jobnum' => $ljob->jobnum,
247 'custnum' => $ljob->custnum,
248 'insert_date' => $ljob->_date,
249 'start_date' => $start_date,
252 my $error = $queue_stat->insert;
253 die $error if $error;
254 dbh->commit; #for the stat
257 if ( UNIVERSAL::can(dbh, 'sprintProfile') ) {
258 open(PROFILE,">%%%FREESIDE_LOG%%%/queueprofile.$$.".time)
259 or die "can't open profile file: $!";
260 print PROFILE dbh->sprintProfile();
261 close PROFILE or die "can't close profile file: $!";
272 warn "received TERM signal; exiting\n";
276 warn "received INT signal; exiting\n";
282 foreach $_ ( $[ .. $#ARGV ) { #untaint @ARGV
283 #$ARGV[$_] =~ /^([\w\-\/]*)$/ || die "Illegal arguement \"$ARGV[$_]\"";
285 $ARGV[$_] =~ /^(.*)$/ || die "Illegal arguement \"$ARGV[$_]\"";
291 die "Usage:\n\n freeside-queued user\n";
295 foreach my $pid ( keys %kids ) {
296 my $kid = waitpid($pid, WNOHANG);
306 freeside-queued - Job queue daemon
310 freeside-queued [ -s | -n ] user
314 Job queue daemon. Should be running at all times.
316 -s: "secure" jobs only (queued billing jobs)
318 -n: non-"secure" jobs only (other jobs)
320 user: Typically "fs_queue"