
use Socket;
use POSIX ":sys_wait_h";

#
# Don't buffer stdout
#

$| = 1; 
chop($DATE = `date`);
print "scheduler: Restart on $DATE\n";

#
# Set up spooling directories
#

$JOB_SPOOL  = "$ENV{SERVER_SPOOL}/JOBS";
$PORT_SPOOL = "$ENV{SERVER_SPOOL}/PORTS";
$PROC  = "$ENV{SERVER_VAR}/proc/stations";
unless (-d "$PROC") {
  system("mkdir $PROC");
}
unless (-d "$PORT_SPOOL") {
  system("mkdir $PORT_SPOOL");
}
unless (-d "$JOB_SPOOL") {
  system("mkdir $JOB_SPOOL");
}

#
# Make sure station_type_port exists.
#

system("touch $ENV{SERVER_VAR}/lib/station_type_port");


#
# This associative array will keep track of how many attempts
# have been made to schedule a particular job.
#

%ATTEMPT = (); 
%OnStation = ();
$MAX = 5;
$QMAX = 15;
$DefaultMachineMax = 4;

#
# Loop forever.
#

$OLDEXEC = -1;
&update_station_list;
&check_abandoned_jobs;
&updateMonitor;

while (1) {

  while (!-f "$JOB_SPOOL/JOBS") {

    if ((keys %PID)||(keys %EXEC)||(keys %QUEUE)) {

#
# Schedule jobs in QUEUE
#

      foreach $key (keys %QUEUE) {
# key represents the solver
       
        $TIMEOUT     = 0;
        $BAD_PARSE   = 0;
        $QFULL       = 0;
        %JobsAllowed = (); #per station
        $JobsAllowed{"localhost"} = 2; 
        #allow a couple of bad jobs to be handled at once
        %CurJobs     = (); #per station
        @process = split /:/, $QUEUE{"$key"};
        $count=0;
        if ($key =~ /^ADMIN:.*$/) {
          $LIBDIR = "$ENV{SERVER_ADMIN_SOLVERS}/$key";
        } else {
          $LIBDIR = "$ENV{SERVER_SOLVERS}/$key";
        }
        if (-e "$LIBDIR") {
          if (-s "$LIBDIR/STATIONS") {
            open(STATS, "$LIBDIR/STATIONS");
            chomp(@STATS = <STATS>);
            close STATS;
            @STATIONS = ();
            foreach $station (@STATS) {
              if($station =~ /(\S+)\s+(\d+)/) {
                $station = $1;
                $JobsAllowed{"$station"} = $2;
              } else {
                $JobsAllowed{"$station"} = 1;
              }
              @STATIONS = (@STATIONS, $station);

# 
# Determine current $key jobs running on $station
#

              &findCurJobs($station, $key);
#              print "station $station allows $JobsAllowed{$station} has $CurJobs{$station} solver jobs and $OnStation{$station} total\n";
            }
# try the stations with most capacity first
            @STATIONS = (sort {($JobsAllowed{"$b"} - $CurJobs{"$b"} - $OnStation{"$b"}) <=> ($JobsAllowed{"$a"} - $CurJobs{"$a"} - $OnStation{"$a"})} @STATIONS) ;
          } else {		 
            $station = "localhost";
            @STATIONS = ($station);
            $stationFile = $station;
            if (-s "$LIBDIR/CONTACT") {
              chomp($CONTACT=`cat $LIBDIR/CONTACT`);
            } else {
              $CONTACT=$ENV{"SERVER_ADMIN"};
            }
            if($key !~ /ADMIN/) {
              &notify_contact("Directory $LIBDIR contained no STATIONS file for $key.\n");
              $BAD_PARSE=1;
            }
            if(-s "$PROC/$stationFile") {
              open(RUNNING, "$PROC/$stationFile");
              chomp($cur_run = <RUNNING>);
              close RUNNING;
              $CurJobs{"$station"} = $cur_run;
            } elsif (-f "$PROC/$stationFile") {
              $CurJobs{"$station"} = 1;
            } else {
              $CurJobs{"$station"} = 0;
            }
          }			
          if (-s "$LIBDIR/CONTACT") {
            chomp($CONTACT=`cat $LIBDIR/CONTACT`);
          } else {
            $CONTACT=$ENV{"SERVER_ADMIN"};
          }
        }
        @KEYSTATIONS = @STATIONS;
        foreach $next (@process) {
# process represents the job number in queue
          $JOBDIR      = "$ENV{SERVER_JOBS}/job.$next";
          @STATIONS = @KEYSTATIONS;
          if($count > $QMAX) {
            if($key eq "ADMIN:BAD") {
              #don't do all of this stuff
            } else {
              if ($count == $QMAX) {
                &notify_contact("There are jobs competing to enter the $key queue.  You may want to check that the currently executing job is legitimate.");
              }
              $QFULL = 1;
              open(TOUCH, ">$JOBDIR/QFULL");
              close(TOUCH);
              $note = "The solver queue for $key has more than $QMAX entries.";
              &notify_user ($note); 
              $QUEUE{"$key"} =~ s/$next://;
              if($QUEUE{"$key"} eq "") {
                delete $QUEUE{"$key"};
              }
              $QUEUE{"ADMIN:BAD"} .= "$next:";
              print "scheduler: $date: job #$next will be culled.  $QMAX jobs in $key queue.\n";
              next;	   
            }
          }
          $ATTEMPT{"$next"}++;
          if ($ATTEMPT{"$next"} > $MAX) {
            $TIMEOUT = 1;
            @STATIONS = ("localhost");
            &findCurJobs("localhost");
#
# Only notify the user if this ADMIN:BAD job will be run
#
            if($CurJobs{"localhost"} < $JobsAllowed{"localhost"}) {
              &notify_user;
            } 
            system("touch $JOBDIR/TIMEOUT");
            chomp($date = `date`);
            print "scheduler: $date: job #$next timed out after $MAX attempts\n";
            } else {
              if ($ATTEMPT{"$next"} > 1) {
                if ($key =~ /\?:\?/) {
                  $QUEUE{"$key"} =~ s/$next://; 
                  if($QUEUE{"$key"} eq "") {
                    delete $QUEUE{"$key"};
                  }
#
# Do not spend time trying repeatedly to
# schedule bad jobs.
#
                } else {
                  chop($date = `date`);
                  print "scheduler: $date: attempt $ATTEMPT{$next} for job #$next\n" ;
                  }  
              }
              if (-f "$JOBDIR/BAD_PARSE") {
                $station = "localhost";
                @STATIONS = ("$station");
                $BAD_PARSE = 1;
                $JobsAllowed{"$station"} = 2;
                $stationFile = $station;
                if(-s "$PROC/$stationFile") {
                  open(RUNNING, "$PROC/$stationFile");
                  chomp($cur_run = <RUNNING>);
                  close RUNNING;
                  $CurJobs{"$station"} = $cur_run;
                } elsif (-f "$PROC/$stationFile") {
                  $CurJobs{"$station"} = 1;
                } else {
                  $CurJobs{"$station"} = 0;
                }
                
              } elsif (-f "$JOBDIR/QFULL") {
                $QFULL = 1;
                @STATIONS = ("localhost");
                &findCurJobs;
                chop($date = `date`);
              }
            }#else - not over MAX attempts
              
            STATION: foreach $STATION (@STATIONS) {
              $MMax = $DefaultMachineMax;
              if(open(MAXIN, "$ENV{SERVER_VAR}/lib/stations/station_max_$STATION")) {
                chomp($MMax = <MAXIN>);
                if($MMax < 0) {
                  $MMax = $DefaultMachineMax;
                }
                close MAXIN;
              } 
              if(($CurJobs{"$STATION"} == $JobsAllowed{"$STATION"}) or ($OnStation{"$STATION"} >= $MMax)) { 
#                print "scheduler: $STATION is running $JobsAllowed{$STATION} $key jobs or $MMax total jobs.\n";
                $ATTEMPT{"$next"}-- if ($ATTEMPT{"$next"} > 0); 
                next STATION;
              }
              &update_station_list if (-f "$PORT_SPOOL/UPDATE");
              unless (($key =~ /^ADMIN:.*$/)||($STATION_LIST =~ /$STATION:$key:(\d+)/)||($TIMEOUT)||($BAD_PARSE)||($QFULL)) {
                chop($date = `date`);
                print "scheduler: $date: no port entry for \"$STATION\"\n";
                unless ($PORT_ERROR{"$STATION$next"}) {
                  $note  = "You need to start a comms-daemon on $STATION\n"; 
                  $note .= "before it can accept jobs from the server.\n"; 
                  &notify_contact($note); 
                  $PORT_ERROR{"$STATION$next"} = 1;
                }
                next STATION;
              }
              $port = $1; 
              unless ($CurJobs{"$STATION"}>=$JobsAllowed{"$STATION"}) {
                unless (($key =~ /^ADMIN:.*$/)||($TIMEOUT)||($BAD_PARSE)||($QFULL)) {
                  chop($id = `echo verify|$^X $ENV{SERVER_BIN}/client.pl -silent -server $STATION:$port -alarm 30`);
                  $id = "\U$id";
                  $compare = "\U$key\@$ENV{SERVER_HOSTNAME}:$ENV{SERVER_PORT}";
                  unless (($id eq $compare) or ($id =~ /^ALL/) ) {
                    chop($date = `date`);
                    print "scheduler: $date: wrong id for $STATION port $port.\n";
                    print "           Expected \"$compare\"\n";
                    print "           got \"$id\"\n";
                    unless ($BAD_ID{"$STATION$next"}) {
                      $note  = "Error connecting to $STATION:$port.\n";
                      $note .= "Please restart your comms-daemon."; 
                      &notify_contact("$note"); 
                      $BAD_ID{"$STATION$next"} = 1;
                    }
                    next STATION; 
                  }
                }#unless TIMEOUT BAD QFULL
                  
                  
# Fork solver.pl only if counter.pl allows
	    

                  unless ($BAD_PARSE||$TIMEOUT||$QFULL) {
                    $counters = `$ENV{SERVER_PERL} $ENV{SERVER_BIN}/counters.pl $key $JOBDIR`;
                    if ($? != 0) {
                      open(HANDLE, ">>$JOBDIR/job.results");
                      print HANDLE "Unable to schedule $key job \#$next\n\n";
                      print HANDLE "$counters\n*************************************************************\n\n";
                      close(HANDLE);
                      system("cat $JOBDIR/job.received >> $JOBDIR/job.results; touch $JOBDIR/LIMIT_REACHED");
                      $STATION = "localhost";
                      @STATIONS = ($STATION);
                      #
                      # if we've gone to all the trouble to check the counts,
                      # we will run this job, no matter how busy localhost is
                      #
                    } 
                  } #unless QFULL, BADPARSE, TIMEOUT
                
#
# Show station as busy
#

                if ($STATION eq "localhost") {
                  $stationFile = $STATION;
                } else {
                  $stationFile = "$STATION---$key";
                }
     
                if($CurJobs{"$STATION"} >= 0) {  #sanity check
                  $newcur = $CurJobs{"$STATION"} + 1;
                } else {
                  $newcur = 1;
                }
                $CurJobs{"$STATION"} = $newcur;
                open(WRITE, ">$PROC/$stationFile");
                print WRITE $newcur;
                close WRITE;		  
     
#
# Fork solver.pl script to run job
#
                
                $PID{"$next"} = fork();
                if ($PID{"$next"} == 0) {
                  exec("$^X $ENV{SERVER_BIN}/job_start.pl $next $STATION $port 2>> $ENV{SERVER_VAR}/logs/jobs");
                }      
                
#
# Push job's stationFile onto EXEC and pop finished jobs
#

                $OnStation{"$STATION"}++;
                $EXEC{"$next"} = "$stationFile"; 
     
	
#
# Pop job from QUEUE. 
#

                $count--; # since we're popping it, no longer counts as in queue 
                $QUEUE{"$key"} =~ s/$next://; 
	
                if ($QUEUE{"$key"} eq "") {
                  
#
# Delete solver queue if empty.
#

                  delete $QUEUE{"$key"}; 
                  
                } else {

#
# Output current queue to all job.outs
#  

                  @CLIENTS = split(/:/, $QUEUE{"$key"}); 
                  foreach $CLIENT (@CLIENTS) {
                    open (HANDLE, ">>$ENV{SERVER_JOBS}/job.$CLIENT/job.out");
                    if(open(IN, "$ENV{SERVER_JOBS}/job.$CLIENT/job.password")) {
                      chomp($passwd = <IN>);
                      close IN;
                    } else {
                      $passwd = "could not be found";
                    }
                    &outputQueues($CLIENT, $passwd);
                    close(HANDLE);
                  }
                }
                chomp($date = `date`);
                print "scheduler: $date: $key job #$next scheduled on $STATION\n";
                open (HANDLE, ">$JOBDIR/job.station");
                print HANDLE $STATION;
                close (HANDLE);	  
                last STATION; # We successfully got a workstation.
              } # end unless curjobs >= jobs allowed on station
            }  # end for each station loop
          $count++;	     
        } # end for each job to process
      } # end for each keys (queues) loop
	    
#
# Wait for any children to avoid "zombies"
#

      foreach $key (keys %PID) {
        if (waitpid($PID{"$key"}, &WNOHANG)) {
          if (-f "$ENV{SERVER_JOBS}/job.$key/DONE") {
	       delete $PID{"$key"};
          }
        }
      }
      
#
# Pop finished jobs and decrement station files
#
      &cleanold;


#
# Update monitor if jobs have finished 
#

      if (!($OLDEXEC==(keys %EXEC))) {
        &updateMonitor;
      }
      
    } # end exists keys loop
    &update_station_list if (-f "$PORT_SPOOL/UPDATE");
    fsleep(1);
  } # end no new jobs loop


#
# Get new jobs from parser, add to QUEUE, and update monitor.
#

  system("rm $JOB_SPOOL/JOBS");
  chop(@LIST = `ls -c -r $JOB_SPOOL/job.* 2>/dev/null`);
  foreach (@LIST) {
    $TYPE = `cat $_`;
    $TYPE =~ s/\s*//g;
    system("rm $_");
    s/.*job\.(\d+)$/$1/;
    $QUEUE{"$TYPE"} .= "$_:";
    $OLDEXEC = -1; # so the monitor gets updated
    if(open(IN, "$ENV{SERVER_JOBS}/job.$_/job.password")) {
      chomp($passwd = <IN>);
      close IN;
    } else {
      $passwd = "could not be found";
    }
    open (HANDLE, ">>$ENV{SERVER_JOBS}/job.$_/job.out");
    &outputQueues($_,$passwd);
    close(HANDLE);
  }
} # end outer loop

#
# This procedure will update <SERVER_VAR>/lib/station_type_port
# Ports have been registered or deleted when <PORT_SPOOL>/UPDATE is
# present.   Each numbered file in <PORT_SPOOL> contains a particular 
# STATION:TYPE:PORT entry.  
#

sub update_station_list {
  unlink("$PORT_SPOOL/UPDATE")or system("rm -f $PORT_SPOOL/UPDATE");
  $station_type_port = `cat $ENV{SERVER_VAR}/lib/station_type_port`;
  @list = glob("$PORT_SPOOL/port_update.*") or 1; 
  foreach $list_item (@list) {
    $content = `cat $list_item`;
    $content =~ s/\s*//g;
    $content =~ /(.+):(.+):(\d+)/;
    $station = $1;
    $type = $2;
    $port = $3;
    unless ($station_type_port =~ s/$station:$type:\d+/$station:$type:$port/) {
      print "scheduler: adding $station:$type:$port\n"; 
      $station_type_port .= "$station:$type:$port\n";
    }  
    unlink("$list_item") or system("rm -f $list_item");
  }
  @list = glob("$PORT_SPOOL/port_delete.*") or 1; 
  foreach $list_item (@list) {
    $content = `cat $list_item`;
    $content =~ s/\s*//g;
    print "scheduler: removing $content\n"; 
    $station_type_port =~ s/($content(:\d+){0,1}\n)//gs or print "scheduler: ERROR: deletion in port file failed.\n";
    unlink("$list_item") or system("rm -f $list_item");
  }
  open(HANDLE3, ">$ENV{SERVER_VAR}/lib/station_type_port");
  print HANDLE3 $station_type_port;
  close(HANDLE3);
  $STATION_LIST = $station_type_port;
}				 

sub notify_contact {

  chomp($DATE = `date`);
  print "scheduler: $DATE: Notifying $CONTACT\n";  
  open(MAIL, "|sendmail -oi -t") or die("Denied access to sendmail program\n");
  print MAIL <<END_SOLVER_RESULTS;
To: $CONTACT
From: $ENV{SERVER_NAME} Server <$ENV{SERVER_EMAIL}>
Subject: ERROR WITH $key

**************************************
A message from the scheduler regarding 
$key job #$next:
**************************************

$_[0]
END_SOLVER_RESULTS
  close(MAIL);                                              
}			       

sub notify_user {
  my $preamble = "";
  if($_[0]) {
    $preamble = $_[0];
  }
  
  open (USER, ">> $JOBDIR/job.results");
  print USER <<END_SOLVER_RESULTS;
  
$preamble
We were unable to schedule job #$next. The $key 
administrator has been notified. Please try back later.

$ENV{SERVER_SIGNATURE}
*************************************************************

END_SOLVER_RESULTS

  open(REC, "$JOBDIR/job.received");
  while(<REC>) {
    print USER;
  }
  close REC;
  close (USER);
}				 
				 
sub fsleep {			
  select(undef, undef, undef, $_[0]);
}

sub outputQueues {
  my $passwd = $_[1];
  print HANDLE "<CLEAR_SCREEN>\n";
  print HANDLE "Scheduling:\n\n";
  print HANDLE "  This job #$_[0], password $passwd.\n";
  print HANDLE "  If you wish to terminate your job or dequeue it,\n";
  print HANDLE "  submit this number and password to the Kill Job solver.\n\n";  
  print HANDLE "  Solver Queues:\n\n";
  if (keys %QUEUE) {		
    foreach $key (keys %QUEUE) {
      print HANDLE "    $key: ";
      print HANDLE "$QUEUE{$key}\n";
      print HANDLE "\n";
    }
  } else {
    print HANDLE "    No jobs waiting.\n";
  }
  print HANDLE "\n  Jobs Executing:\n\n";
  if (keys %EXEC) {
    foreach $exec (keys %EXEC) {
      print HANDLE "    job#$exec executing on $EXEC{$exec}.\n";
    }
  } else {
    print HANDLE "    No jobs executing.\n";
  }
}

sub updateMonitor {		
  open(MONITOR, ">$ENV{SERVER_LOGS}/queues");
  print MONITOR "Job Queues:\n\n";
  if (keys %QUEUE) {
    foreach $key (keys %QUEUE) {
      print MONITOR "$key:\n\n";
      print MONITOR "$QUEUE{$key}\n";
      print MONITOR "\n";
    } 
  } else {
    print MONITOR "No jobs in queue.\n";
  }
  print MONITOR "\nJob Execution:\n\n";
  foreach $exec (keys %EXEC) {
    print MONITOR "job#$exec executing on $EXEC{$exec}.\n";
  }
  $OLDEXEC = keys %EXEC;
  unless ($OLDEXEC) {
    print MONITOR "No jobs executing.\n";
    system("rm -f $PROC/*");  #redundant check
  }
  close(MONITOR);
}			   


sub check_abandoned_jobs {
#
# Do not lose track of jobs that are already running or in the queue.
# If you do want to lose these jobs, delete the queues file as
# well as the $PROC station files.
#

  local ($f, $e, @csf, @cse, @csfp, $fp, @f);
    
  $queued = 0;	     
  $running = 0;
  $solver = "";
  if(open(TEXT, "$ENV{SERVER_LOGS}/queues")) {
    
    while(<TEXT>) {
      s/^\s+//;
      s/\s+$//;
      if(/^$/) {
        next;
      }
      if(/Job Queues:/) {
        $queued = 1;
      } elsif ($queued) {
        if (/^(\d+):/) {
          while(/^(\d+):/) {
            $solverQ = $QUEUE{"$solver"};
            $QUEUE{"$solver"} = "$solverQ"."$1:";
            $_ =~ s/^(\d+)://;
          }	     
        } elsif(/^(\S+:\S+):/) {
          $solver = $1;
          print "scheduler queue for $solver\n";
        }	     
      } 
      if(/Job Execution:/) {
        $queued = 0;
        $running = 1;
      } elsif ($running) {
        print "scheduler.pl: finding running jobs\n";
        if(/^job#(\d+) executing on (.+)\./) {
           if(-f "$PROC/$2") {
             $EXEC{"$1"} = $2;
             if(/executing on (.+)---(.+?)\./) {
               $OnStation{"$1"}++;
             }
           }
         }
      }
    }
    close TEXT;
#
# Here, check that all of the files under $PROC
# had corresponding elements in the queues file.
#

    @csfp = <$PROC/*>;
    foreach $fp (@csfp) {
      @f = split '/', $fp;
      @csf = (@csf, $f[$#f]);
    }
    @cse = values(%EXEC);
    foreach $f (@csf) {
      $f =~ s/\s//g;
      $num_running = 1;
      if(open(RUNNING, "$PROC/$f")) {
        chomp($num_running = <RUNNING>);
        close RUNNING;
        unless ($num_running =~ /\d+/) {
          $num_running = 1;
        }
      }
      $match = 0;
      foreach $e (@cse) {
        $e =~ s/\s//g;
        if($e =~/^$f$/) {
          $match++;
        }
      }
      if ($match==0) {
        print "scheduler.pl: removing file $f from $PROC\n";
        unlink "$PROC/$f";
      } elsif ($match != $num_running) {
        if(open(RUNNING, ">$PROC/$f")) {
          print RUNNING $match;
          close RUNNING;
        }
      }
    }
  } else {
    print "scheduler.pl: no queues file of old jobs\n";
    system("rm -f $PROC/*");
  }
}			       

sub cleanold {
#
# See if any jobs have finished
#

  foreach $old (keys %EXEC) {
    if (-f "$ENV{SERVER_JOBS}/job.$old/DONE") { 
      $stationFile = $EXEC{"$old"};
      $station = $stationFile;
      $station =~ s/---(.*:.*)//; 
      if(-s "$PROC/$stationFile") {
        open(RUNNING, "$PROC/$stationFile");
        chomp($cur_run = <RUNNING>);
        close RUNNING;
        $CurJobs{"$station"} = $cur_run - 1;
      } elsif (-f "$PROC/$stationFile") {
        $CurJobs{"$station"} = 0;
      } else {
        print "scheduler Error, job $old in EXEC but has no station file\n";
      }
      $newcur = $CurJobs{"$station"};
      if($CurJobs{"$station"}<=0) {
        system("rm -f $PROC/$EXEC{$old}");
      } else {
        open(WRITE, ">$PROC/$EXEC{$old}");
        print WRITE $newcur;
        close WRITE;		  
      }
      if($OnStation{"$station"} > 0) {
        $OnStation{"$station"}--;
      }
      delete $EXEC{"$old"};
    } else {
      #
      # make sure solver.pl is still running
      #
#	   if(exists($PID{"$old"})) {
#	       unless(kill 0 => $PID{"$old"}) {
#		   print "solver.pl for $old died\n";
#		   $PID{"$old"} = fork();
#		   if ($PID{"$old"} == 0) {
#		       $stationFile = $EXEC{"$old"};
#		       $station = $stationFile;
#		       $station =~ s/-(.*)//; 
#		       $key = $1;
#		       if($STATION_LIST =~ /$station:$key:(\d+)/) {
#			   $port = $1;
#			   exec("$^X $ENV{SERVER_BIN}/solver.pl -recon $old $station $port 2>&1");
#		       } else {	
#			   open(TOUCH, ">$ENV{SERVER_JOBS}/job.$old/DONE");
#			   close TOUCH;	 
#		       }	 
#		   }		 
#	       }		 
#	   }			 
    }			 
  }			       
}



sub findCurJobs {

  my ($station, $key);
  if(defined($_[0])) {
    $station = $_[0];
  } 
  if(defined($_[1])) {
    $key = $_[1];
  }

# 
# Determine current $key jobs running on $station
#

  if ($station eq "localhost") {
    $stationFile = $station; 
  } else {
    $stationFile = "$station---$key";
  }	      
  if(-s "$PROC/$stationFile") {
    open(RUNNING, "$PROC/$stationFile");
    chomp($cur_run = <RUNNING>);
    close RUNNING;
    $CurJobs{"$station"} =  $cur_run;
  } elsif (-f "$PROC/$stationFile") {
    $CurJobs{"$station"} = 1;
  } else {
    $CurJobs{"$station"} = 0;
  }
  
}
