FS/bin/freeside-queued

   1 #!/usr/bin/perl -w
   2
   3 use strict;
   4 use vars qw( $DEBUG $kids $max_kids $sleep_time %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 $sleep_time = $conf->config('queued-sleep_time') || 10;
  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   my $nodepend = 'AND NOT EXISTS( SELECT 1 FROM queue_depend'.
  88                  '           WHERE queue_depend.jobnum = queue.jobnum )';
  89
  90   #anything with a priority goes after stuff without one
  91   my $order_by = ' ORDER BY COALESCE(priority,0) ASC, jobnum ASC ';
  92
  93   my $limit = $max_kids - $kids;
  94
  95   $order_by .= ( driver_name eq 'mysql'
  96                    ? " LIMIT $limit FOR UPDATE "
  97                    : " FOR UPDATE LIMIT $limit " );
  98
  99   my @jobs = qsearch({
 100     'table'     => 'queue',
 101     'hashref'   => { 'status' => 'new' },
 102     'extra_sql' => $nodepend,
 103     'order_by'  => $order_by,
 104   });
 105
 106   unless ( @jobs ) {
 107     dbh->commit or do {
 108       warn "WARNING: database error, closing connection: ". dbh->errstr;
 109       undef $FS::UID::dbh;
 110       next;
 111     };
 112     sleep $sleep_time;
 113     next;
 114   }
 115
 116   foreach my $job ( @jobs ) {
 117
 118     my %hash = $job->hash;
 119     $hash{'status'} = 'locked';
 120     my $ljob = new FS::queue ( \%hash );
 121     my $error = $ljob->replace($job);
 122     if ( $error ) {
 123       warn "WARNING: database error locking job, closing connection: ".
 124            dbh->errstr;
 125       undef $FS::UID::dbh;
 126       next;
 127     }
 128
 129     dbh->commit or do {
 130       warn "WARNING: database error, closing connection: ". dbh->errstr;
 131       undef $FS::UID::dbh;
 132       next;
 133     };
 134
 135     $FS::UID::AutoCommit = 1;
 136
 137     my @args = $ljob->args;
 138     splice @args, 0, 1, $ljob if $args[0] eq '_JOB';
 139
 140     defined( my $pid = fork ) or do {
 141       warn "WARNING: can't fork: $!\n";
 142       my %hash = $job->hash;
 143       $hash{'status'} = 'failed';
 144       $hash{'statustext'} = "[freeside-queued] can't fork: $!";
 145       my $ljob = new FS::queue ( \%hash );
 146       my $error = $ljob->replace($job);
 147       die $error if $error;
 148       next; #don't increment the kid counter
 149     };
 150
 151     if ( $pid ) {
 152       $kids++;
 153       $kids{$pid} = 1;
 154     } else { #kid time
 155
 156       #get new db handle
 157       $FS::UID::dbh->{InactiveDestroy} = 1;
 158
 159       forksuidsetup($user);
 160
 161       #auto-use classes...
 162       if (    $ljob->job =~ /(FS::(part_export|cust_main)::\w+)::/
 163            || $ljob->job =~ /(FS::\w+)::/
 164          )
 165       {
 166         my $class = $1;
 167         eval "use $class;";
 168         if ( $@ ) {
 169           warn "job use $class failed";
 170           my %hash = $ljob->hash;
 171           $hash{'status'} = 'failed';
 172           $hash{'statustext'} = $@;
 173           my $fjob = new FS::queue( \%hash );
 174           my $error = $fjob->replace($ljob);
 175           die $error if $error;
 176           exit; #end-of-kid
 177         };
 178       }
 179
 180       my $eval = "&". $ljob->job. '(@args);';
 181       warn 'running "&'. $ljob->job. '('. join(', ', @args). ")\n" if $DEBUG;
 182       eval $eval; #throw away return value?  suppose so
 183       if ( $@ ) {
 184         warn "job $eval failed";
 185         my %hash = $ljob->hash;
 186         $hash{'status'} = 'failed';
 187         $hash{'statustext'} = $@;
 188         my $fjob = new FS::queue( \%hash );
 189         my $error = $fjob->replace($ljob);
 190         die $error if $error;
 191       } else {
 192         $ljob->delete;
 193       }
 194
 195       exit;
 196       #end-of-kid
 197     }
 198
 199   } #foreach my $job
 200
 201 } continue {
 202   if ( sigterm() ) {
 203     warn "received TERM signal; exiting\n";
 204     exit;
 205   }
 206   if ( sigint() ) {
 207     warn "received INT signal; exiting\n";
 208     exit;
 209   }
 210 }
 211
 212 sub usage {
 213   die "Usage:\n\n  freeside-queued user\n";
 214 }
 215
 216 sub reap_kids {
 217   foreach my $pid ( keys %kids ) {
 218     my $kid = waitpid($pid, WNOHANG);
 219     if ( $kid > 0 ) {
 220       $kids--;
 221       delete $kids{$kid};
 222     }
 223   }
 224 }
 225
 226 =head1 NAME
 227
 228 freeside-queued - Job queue daemon
 229
 230 =head1 SYNOPSIS
 231
 232   freeside-queued user
 233
 234 =head1 DESCRIPTION
 235
 236 Job queue daemon.  Should be running at all times.
 237
 238 user: from the mapsecrets file - see config.html from the base documentation
 239
 240 =head1 VERSION
 241
 242 =head1 BUGS
 243
 244 =head1 SEE ALSO
 245
 246 =cut
 247