#!/usr/bin/perl -w
#
# Copyright (c) 2006, 2007 Michael Schroeder, Novell Inc.
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License version 2 as
# published by the Free Software Foundation.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program (see the file COPYING); if not, write to the
# Free Software Foundation, Inc.,
# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA
#
################################################################
#
# The Job Dispatcher
#

BEGIN {
  my ($wd) = $0 =~ m-(.*)/- ;
  $wd ||= '.';
  # FIXME: currently the bs_dispatcher makes assumptions on being in a
  # properly set up working dir, e.g. with subdirs 'worker' and
  # 'build'.  Either that is cleaned up or this stays in, for the sake
  # of startproc and others being able to start a bs_srcserver without
  # knowing that it has to be started in the right directory....

  chdir "$wd";
  unshift @INC,  "build";
}

use POSIX;
use Data::Dumper;
use Digest::MD5 ();
use List::Util;
use Fcntl qw(:DEFAULT :flock);
use XML::Structured ':bytes';

use BSConfig;
use BSRPC;
use BSUtil;
use BSXML;

use strict;

my $user = $BSConfig::bsuser;
my $group = $BSConfig::bsgroup;

!defined($user) || defined($user = (getpwnam($user))[2]) || die("unknown user\n");
!defined($group) || defined($group = (getgrnam($group))[2]) || die("unknown group\n");
if (defined $group) {
  ($(, $)) = ($group, $group);
  die "setgid: $!\n" if ($) != $group);
}
if (defined $user) {
  ($<, $>) = ($user, $user); 
  die "setuid: $!\n" if ($> != $user); 
}

my $port = 5252;        #'RR'
$port = $1 if $BSConfig::reposerver =~ /:(\d+)$/;

my %cando = (
  'i586'   => ['i586'],
  'i686'   => ['i586', 'i686'],
  'x86_64' => ['x86_64', 'i586', 'i686'],
  'ppc'    => ['ppc'],
  'ppc64'  => ['ppc64', 'ppc'],
  'sh4'    => ['sh4'],
);

my $rundir = "$BSConfig::bsdir/run";
my $workersdir = "$BSConfig::bsdir/workers";
my $jobsdir = "$BSConfig::bsdir/jobs";
my $eventdir = "$BSConfig::bsdir/events";

my $reporoot = "$BSConfig::bsdir/build";

sub getcodemd5 {
  my ($dir, $cache) = @_;
  my $md5 = '';
  my %new;
  my $doclean;
  my @files = grep {!/^\./} ls($dir);
  my @bfiles = grep {!/^\./} ls("$dir/Build");
  my %bfiles = map {$_ => 1} @bfiles;
  @files = sort(@files, @bfiles);
  $cache ||= {};
  for my $file (@files) {
    my $f = $bfiles{$file} ? "$dir/Build/$file" : "$dir/$file";
    next unless -f $f;
    my @s = stat _;
    my $id = "$s[9]/$s[7]/$s[1]";
    $new{$id} = 1; 
    if ($cache->{$id}) {
      $md5 .= "$cache->{$id}  $file\n";
      next;
    }    
    $cache->{$id} = Digest::MD5::md5_hex(readstr($f));
    $md5 .= "$cache->{$id}  $file\n";
    $doclean = 1; 
  }
  if ($doclean) {
    for (keys %$cache) {
      delete $cache->{$_} unless $new{$_};
    }    
  }
  return Digest::MD5::md5_hex($md5);
}

my $workerdircache = {};
my $builddircache = {};

sub assignjob {
  my ($job, $idlename, $arch) = @_;
  local *F;

  print "assignjob $arch/$job -> $idlename\n";
  my $jobstatus = {
    'code' => 'dispatching',
  };
  if (!BSUtil::lockcreatexml(\*F, "$jobsdir/$arch/.dispatch.$$", "$jobsdir/$arch/$job:status", $jobstatus, $BSXML::jobstatus)) {
    print "job lock failed!\n";
    return undef;
  }

  # got the lock, re-check if job is still there
  if (! -e "$jobsdir/$arch/$job") {
    unlink("$jobsdir/$arch/$job:status");
    close F;
    print "job disappered!\n";
    return undef;
  }

  # get the worker data
  my $worker = readxml("$workersdir/idle/$idlename", $BSXML::worker, 1);
  if (!$worker) {
    unlink("$jobsdir/$arch/$job:status");
    close F;
    print "worker is gone!\n";
    return undef;
  }

  # ready for building, send info to worker!
  unlink("$workersdir/idle/$idlename");
  my $infoxml = readstr("$jobsdir/$arch/$job");
  my $jobid = Digest::MD5::md5_hex($infoxml);
  my $info = XMLin($BSXML::buildinfo, $infoxml);
  my $prp = $info->{'path'}->[0];
  my $workercode = getcodemd5('worker', $workerdircache);
  my $buildcode = getcodemd5('build', $builddircache);
  eval {
    BSRPC::rpc({
      'uri'     => "http://$worker->{'ip'}:$worker->{'port'}/build",
      'timeout' => 10,
      'request' => "PUT",
      'headers' => [ "Content-Type: text/xml" ],
      'data'    => $infoxml,
    }, undef, "port=$port", "workercode=$workercode", "buildcode=$buildcode");
  };
  if ($@) {
    print "rpc error: $@";
    unlink("$jobsdir/$arch/$job:status");
    close F;
    return undef;
  }
  $jobstatus->{'code'} = 'building';
  $jobstatus->{'uri'} = "http://$worker->{'ip'}:$worker->{'port'}";
  $jobstatus->{'workerid'} = $worker->{'workerid'} if defined $worker->{'workerid'};
  $jobstatus->{'starttime'} = time();
  $jobstatus->{'hostarch'} = $worker->{'hostarch'};
  $jobstatus->{'jobid'} = $jobid;

  # put worker into building list
  $worker->{'job'} = $job;
  $worker->{'arch'} = $arch;
  mkdir_p("$workersdir/building");
  writexml("$workersdir/building/.$idlename", "$workersdir/building/$idlename", $worker, $BSXML::worker);

  # write new status and release lock
  writexml("$jobsdir/$arch/.$job:status", "$jobsdir/$arch/$job:status", $jobstatus, $BSXML::jobstatus);
  close F;
  return 1;
}

sub sendeventtosrcserver {
  my ($ev) = @_;
  my @args;
  for ('type', 'project', 'package', 'repository', 'arch', 'job') {
    push @args, "$_=$ev->{$_}" if defined $ev->{$_};
  }
  my $param = {
    'uri' => "$BSConfig::srcserver/event",
    'timeout' => 10,
  };
  BSRPC::rpc($param, undef, @args);
}

$| = 1;
print "starting build service dispatcher\n";

# get lock
mkdir_p($rundir);
open(RUNLOCK, '>>', "$rundir/bs_dispatch.lock") || die("$rundir/bs_dispatch.lock: $!\n");
flock(RUNLOCK, LOCK_EX | LOCK_NB) || die("dispatcher is already running!\n");
utime undef, undef, "$rundir/bs_dispatch.lock";

while (1) {
  my $assigned = 0;
  my @idle = grep {!/^\./} ls("$workersdir/idle");
  my %idlearch;
  for my $idle (@idle) {
    my $harch = (split(':', $idle, 2))[0];
    for (@{$cando{$harch} || []}) {
      push @{$idlearch{$_}}, $idle;
    }
  }
  my @archs = List::Util::shuffle(keys %idlearch);
  for my $arch (@archs) {
    my @b = grep {!/^\./} ls("$jobsdir/$arch");
    my %locked = map {$_ => 1} grep {/:status$/} @b;
    @b = grep {!/:(?:dir|status|new)$/} @b;
    @b = grep {!$locked{"$_:status"}} @b;

    #poor man's project priority
    my @bn = grep {!/^Ports:DebianBased:Auto/} @b;
    @bn = grep {!/^home:rguenther:playground/} @bn;
    @b = @bn if @bn;

    #@bn = grep {/^openSUSE:Factory/} @bn;
    #@b = @bn if @bn;

    for my $job (List::Util::shuffle(@b)) {
      for my $idle (map {$_} List::Util::shuffle(@{$idlearch{$arch} || []})) {
	next unless -e "$jobsdir/$arch/$job";
	if (!assignjob($job, $idle, $arch)) {
	  my $harch = (split(':', $idle, 2))[0];
	  for (@{$cando{$harch} || []}) {
	    $idlearch{$_} = [ grep {$_ ne $idle} @{$idlearch{$_}} ];
	  }
	  next;
	}
	my $harch = (split(':', $idle, 2))[0];
	for (@{$cando{$harch} || []}) {
	  $idlearch{$_} = [ grep {$_ ne $idle} @{$idlearch{$_}} ];
	}
	$assigned++;
	last;
      }
    }
  }
  for my $evname (ls("$eventdir/repository")) {
    next if $evname =~ /^\./;
    my $ev = readxml("$eventdir/repository/$evname", $BSXML::event, 1);
    next unless $ev;
    eval {
      sendeventtosrcserver($ev);
    };
    if ($@) {
      warn($@);
    } else {
      unlink("$eventdir/repository/$evname");
    }
  }
  sleep(1) unless $assigned;
  printf("assigned $assigned jobs\n") if $assigned;
}
