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