FS/bin/freeside-queued

   1 #!/usr/bin/perl -w
   2
   3 use strict;
   4 use vars qw( $DEBUG $kids $max_kids %kids );
   5 use POSIX qw(:sys_wait_h);
   6 use IO::File;
   7 use FS::UID qw(adminsuidsetup forksuidsetup driver_name dbh myconnect);
   8 use FS::Daemon qw(daemonize1 drop_root logfile daemonize2 sigint sigterm);
   9 use FS::Record qw(qsearch qsearchs);
  10 use FS::queue;
  11 use FS::queue_depend;
  12 use FS::UI::Web;
  13
  14 # no autoloading for non-FS classes...
  15 use Net::SSH 0.07;
  16
  17 $DEBUG = 0;
  18
  19 $max_kids = '10'; #guess it should be a config file...
  20 $kids = 0;
  21
  22 my $user = shift or die &usage;
  23
  24 warn "starting daemonization (forking)\n" if $DEBUG;
  25 #daemonize1('freeside-queued',$user); #to keep pid files unique w/multi installs
  26 daemonize1('freeside-queued');
  27
  28 warn "dropping privledges\n" if $DEBUG;
  29 drop_root();
  30
  31
  32 $ENV{HOME} = (getpwuid($>))[7]; #for ssh
  33
  34 warn "connecting to database\n" if $DEBUG;
  35 $@ = 'not connected';
  36 while ( $@ ) {
  37   eval { adminsuidsetup $user; };
  38   if ( $@ ) {
  39     warn $@;
  40     warn "sleeping for reconnect...\n";
  41     sleep 5;
  42   }
  43 }
  44
  45 logfile( "%%%FREESIDE_LOG%%%/queuelog.". $FS::UID::datasrc );
  46
  47 warn "completing daemonization (detaching))\n" if $DEBUG;
  48 daemonize2();
  49
  50 #--
  51
  52 my $warnkids=0;
  53 while (1) {
  54
  55   &reap_kids;
  56   #prevent runaway forking
  57   if ( $kids >= $max_kids ) {
  58     warn "WARNING: maximum $kids children reached\n" unless $warnkids++;
  59     &reap_kids;
  60     sleep 1; #waiting for signals is cheap
  61     next;
  62   }
  63   $warnkids=0;
  64
  65   unless ( dbh && dbh->ping ) {
  66     warn "WARNING: connection to database lost, reconnecting...\n";
  67
  68     eval { $FS::UID::dbh = myconnect; };
  69
  70     unless ( !$@ && dbh && dbh->ping ) {
  71       warn "WARNING: still no connection to database, sleeping for retry...\n";
  72       sleep 10;
  73       next;
  74     } else {
  75       warn "WARNING: reconnected to database\n";
  76     }
  77   }
  78
  79   #my($job, $ljob);
  80   #{
  81   #  my $oldAutoCommit = $FS::UID::AutoCommit;
  82   #  local $FS::UID::AutoCommit = 0;
  83   $FS::UID::AutoCommit = 0;
  84
  85   #assuming mysql 4.1 w/subqueries now
  86   #my $nodepend = driver_name eq 'mysql'
  87   # ? ''
  88   # : 'AND 0 = ( SELECT COUNT(*) FROM queue_depend'.
  89   #   ' WHERE queue_depend.jobnum = queue.jobnum ) ';
  90   my $nodepend = 'AND 0 = ( SELECT COUNT(*) FROM queue_depend'.
  91                  '           WHERE queue_depend.jobnum = queue.jobnum ) ';
  92
  93   my $job = qsearchs(
  94     'queue',
  95     { 'status' => 'new' },
  96     '',
  97     driver_name eq 'mysql'
  98       ? "$nodepend ORDER BY jobnum LIMIT 1 FOR UPDATE"
  99       : "$nodepend ORDER BY jobnum FOR UPDATE LIMIT 1"
 100   ) or do {
 101     # if $oldAutoCommit {
 102     dbh->commit or do {
 103       warn "WARNING: database error, closing connection: ". dbh->errstr;
 104       undef $FS::UID::dbh;
 105       next;
 106     };
 107     # }
 108     sleep 5; #connecting to db is expensive
 109     next;
 110   };
 111
 112   #assuming mysql 4.1 w/subqueries now
 113   #if ( driver_name eq 'mysql'
 114   #     && qsearch('queue_depend', { 'jobnum' => $job->jobnum } ) ) {
 115   #  dbh->commit or die dbh->errstr; #if $oldAutoCommit;
 116   #  sleep 5; #would be better if mysql could do everything in query above
 117   #  next;
 118   #}
 119
 120   my %hash = $job->hash;
 121   $hash{'status'} = 'locked';
 122   my $ljob = new FS::queue ( \%hash );
 123   my $error = $ljob->replace($job);
 124   if ( $error ) {
 125     warn "WARNING: database error locking job, closing connection: ".
 126          dbh->errstr;
 127     undef $FS::UID::dbh;
 128     next;
 129   }
 130
 131   # if $oldAutoCommit {
 132   dbh->commit or do {
 133     warn "WARNING: database error, closing connection: ". dbh->errstr;
 134     undef $FS::UID::dbh;
 135     next;
 136   };
 137   # }
 138
 139   $FS::UID::AutoCommit = 1;
 140   #}
 141
 142   my @args = $ljob->args;
 143   splice @args, 0, 1, $ljob if $args[0] eq '_JOB';
 144
 145   defined( my $pid = fork ) or do {
 146     warn "WARNING: can't fork: $!\n";
 147     my %hash = $job->hash;
 148     $hash{'status'} = 'failed';
 149     $hash{'statustext'} = "[freeside-queued] can't fork: $!";
 150     my $ljob = new FS::queue ( \%hash );
 151     my $error = $ljob->replace($job);
 152     die $error if $error;
 153     next; #don't increment the kid counter
 154   };
 155
 156   if ( $pid ) {
 157     $kids++;
 158     $kids{$pid} = 1;
 159   } else { #kid time
 160
 161     #get new db handle
 162     $FS::UID::dbh->{InactiveDestroy} = 1;
 163
 164     forksuidsetup($user);
 165
 166     #auto-use classes...
 167     #if ( $ljob->job =~ /(FS::part_export::\w+)::/ ) {
 168     if (    $ljob->job =~ /(FS::part_export::\w+)::/
 169          || $ljob->job =~ /(FS::\w+)::/
 170        )
 171     {
 172       my $class = $1;
 173       eval "use $class;";
 174       if ( $@ ) {
 175         warn "job use $class failed";
 176         my %hash = $ljob->hash;
 177         $hash{'status'} = 'failed';
 178         $hash{'statustext'} = $@;
 179         my $fjob = new FS::queue( \%hash );
 180         my $error = $fjob->replace($ljob);
 181         die $error if $error;
 182         exit; #end-of-kid
 183       };
 184     }
 185
 186     my $eval = "&". $ljob->job. '(@args);';
 187     warn 'running "&'. $ljob->job. '('. join(', ', @args). ")\n" if $DEBUG;
 188     eval $eval; #throw away return value?  suppose so
 189     if ( $@ ) {
 190       warn "job $eval failed";
 191       my %hash = $ljob->hash;
 192       $hash{'status'} = 'failed';
 193       $hash{'statustext'} = $@;
 194       my $fjob = new FS::queue( \%hash );
 195       my $error = $fjob->replace($ljob);
 196       die $error if $error;
 197     } else {
 198       $ljob->delete;
 199     }
 200
 201     exit;
 202     #end-of-kid
 203   }
 204
 205 } continue {
 206   if ( sigterm() ) {
 207     warn "received TERM signal; exiting\n";
 208     exit;
 209   }
 210   if ( sigint() ) {
 211     warn "received INT signal; exiting\n";
 212     exit;
 213   }
 214 }
 215
 216 sub usage {
 217   die "Usage:\n\n  freeside-queued user\n";
 218 }
 219
 220 sub reap_kids {
 221   foreach my $pid ( keys %kids ) {
 222     my $kid = waitpid($pid, WNOHANG);
 223     if ( $kid > 0 ) {
 224       $kids--;
 225       delete $kids{$kid};
 226     }
 227   }
 228 }
 229
 230 =head1 NAME
 231
 232 freeside-queued - Job queue daemon
 233
 234 =head1 SYNOPSIS
 235
 236   freeside-queued user
 237
 238 =head1 DESCRIPTION
 239
 240 Job queue daemon.  Should be running at all times.
 241
 242 user: from the mapsecrets file - see config.html from the base documentation
 243
 244 =head1 VERSION
 245
 246 =head1 BUGS
 247
 248 =head1 SEE ALSO
 249
 250 =cut
 251