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