################################################################################
#      writer: Dan Tsafrir (dants AT cs DOT huji DOT ac DOT il)
# description: Iterate through a PBS log and convert it to SWF. This module 
#              concentrates all the knowledge about the format of PBS logs.
################################################################################
package ParsePBS;
use warnings;
use strict;
use Data::Dumper;
use Time::Local;
use POSIX;

use lib '.';
use ConversionConfig;
use ConversionLog;



###############################################################################
# THE EXPORTED INTERFACE:
#
# &parse_pbs - parse a PBS log and return and return it in its SWF from.
#              see header comment for further details.
################################################################################
use Exporter;
our @ISA       = qw(Exporter);
our @EXPORT    = qw(&parse_pbs);
our $VERSION   = 1.00;



################################################################################
# internal data structures
################################################################################

#-------------------------------------------------------------------------------
# translation from partition/queue/user/group/executable to (SWF) numbers.
# init the hash from %CONFIG to enforce a particular translation, e.g. 
#
#   $g_anonymize_tbl{group} = {'dteam'=>3};
#
# will result in that 3 would be the SWF identifier of group 'dteam'.
# if there's no anonymization data in %CONFIG, SWF IDs would simply be 
# serial (increment as a new value is encountered).
#-------------------------------------------------------------------------------
my %g_anonymize_tbl = (
		       'uid'        => {},
		       'gid'        => {},
		       'executable' => {},
		       'queue'      => {},
		       'partition'  => {},
		      );

#-------------------------------------------------------------------------------
# for each of the keys in %g_anonymize_tbl, max+1 value which were assigned so
# far to a job.
#-------------------------------------------------------------------------------
my %g_anonymize_next = (
			'uid'        => 1,
			'gid'        => 1,
			'executable' => 1,
			'queue'      => 1,
			'partition'  => 1,
		       );

#-------------------------------------------------------------------------------
# specify properties of PBS memory resources.
#-------------------------------------------------------------------------------
my %g_mem_info = (
		  'pvmem' => {'type'=>'virtual' , 'quantity'=>'per_process'},
		  'vmem'  => {'type'=>'virtual' , 'quantity'=>'per_job'    },
		  'pmem'  => {'type'=>'physical', 'quantity'=>'per_process'},
		  'mem'   => {'type'=>'physical', 'quantity'=>'per_job'    },
		 );

#-------------------------------------------------------------------------------
# specify properties of PBS used-cpu resources.
#-------------------------------------------------------------------------------
my %g_usedcpu_info = ('resources_used.pcput' => 'per_process', 
		      'resources_used.cput'  => 'per_job');

#-------------------------------------------------------------------------------
# this is encapsulates the instructions of how to parse SWF fields from a PBS 
# record (see explanation about PBS records in the header documentation of 
# parse_pbs_line). the keys of the hash are from %SWF_COL2INDEX (that is, all
# the fields of the SWF).
#
# pbsfields
#   are the field(s) from the PBS record from which the information should be 
#   extracted. if there's more than one field, then usually the conversion will
#   1st try to use the 1st field, if this isn't found in the PBS record then 
#   the 2nd field will be used etc.
#
# action
#   a ref2func that gets ($swf_field, $swf_record, $r, $pbsfields), and returns
#   the converted value, to be used in the SWF (-1 if conversion failed).
#    $swf_field - key in %g_parse_instructions 
#   $swf_record - ref2arr, the SWF record converted so far (order is important)
#            $r - ref2hash, the job's record as returned by parse_pbs_line().
#                 this is usually the E record, but for jobs that were canceled
#                 before they managed to start running this is the 1st Q record
#    $pbsfields - ref2arr, associated with $swf_field as explained above.
#      $pbsline - the actual PBS line being converted. this is only needed for
#                 when problems are encountered. to avoid creating a copy of
#                 the line for each and every converted field, the $pbsline is
#                 the last parameter and the convert_* functions do not name it.
#                 rather, they use $_[-1] instead.
#
# mandatory
#   if the 'action' failed to convert the corresponding value, the entire
#   record will not be used.
#-------------------------------------------------------------------------------
my %g_parse_instructions =
    (
     'job' => 
     {
      # 'pbsID' isn't a field-name in the original PBS record. rather, it is
      # set by parse_pbs_line() and is associated with the job PBS ID.
      'pbsfields' => ['pbsID'],
      'action'    => \&convert_string,
      'mandatory' => 1,
     },
     'submit' => 
     {
      # by 10.12.5 of PBSPro-5.4 administrator guide, 'ctime' is the 
      # "Time in seconds [since the epoch] when a job was created (first 
      # submitted)."
      'pbsfields' => ['ctime'],
      'action'    => \&convert_int,
      'mandatory' => 1,
     },
     'wait' => 
     {
      # computed by subtracting the job's 'ctime' from its 'start' time.
      # by 10.12.5 of PBSPro-5.4 administrator guide, 'start' is the 
      # "Time in seconds [since the epoch] when the job execution started".
      'pbsfields' => [],	# not converting fields directly
      'action'    => \&convert_waittime,
      'mandatory' => 0,
     },
     'runtime' => 
     {
      # by 11.20.1 of PBSPro-5.4 administrator guide (associated with the
      # --wall option): "Use the walltime attribute rather than wall time
      # calculated by subtracting the job start time from end time. The 
      # walltime resource attribute does _not_ accumulate when a job is 
      # suspended for any reason".
      #
      # HOWEVER, some records have their 'resources_used.walltime' missing
      # from their record (in LPC these are the 352 jobs with Exit_status=-4;
      # that is, that died on signal). in this case, the runtime is computed
      # by subtracting the job's 'ctime' from its 'etime'.
      'pbsfields' => ['resources_used.walltime'],
      'action'    => \&convert_runtime,
      'mandatory' => 0,
     },
     'proc_used' => 
     {
      # - 'resources_used' is the most appropriate "list" to obtain this data
      # - 'ncpus' is the most appropriate field, as by 4.8 of PBSPro-5.4 user
      #   guide we have: "'ncpus' is the number of CPUs (processors) 
      #   required by a job."
      # - if we don't have 'ncpus', let's try the 'nodect', as by 6.7.1 of 
      #   PBSPro-5.4 administrative guide:
      #   "It is possible to set limits on queues (and the Server) as to how 
      #   many nodes a job can request. The 'nodes' resource itself is a text
      #   string and difficult to limit. Thus, 'nodect' (node count) is set 
      #   by the Server to the integer number of nodes desired by the user 
      #   as declared in the nodes resource specification. That declaration 
      #   is parsed and the resulting total number of nodes is set in 
      #   nodect."
      # - if we don't have 'nodect': though the value associated with 'nodes'
      #   might be a complex string, it may also a simple number (as is
      #   the case for LPC), so we try to use 'nodes' as is.
      # - if we don't have ncpus/nodect/nodes in the 'resources_used' list 
      #   (as is the case in LPC) rhe _requested_ size will serve as the 
      #   allocated size (extracted from the 'Resource_List'), by using the
      #   fields described above.
      'pbsfields' => ['resources_used.ncpus' , 
		      'resources_used.nodect',
		      'resources_used.nodes' ,
		      'Resource_List.ncpus'  ,
		      'Resource_List.nodect' ,
		      'Resource_List.nodes'  ],
      'action'    => \&convert_int,
      'mandatory' => 0,
     },
     'cpu_used' => 
     {
      # by 4.8 of PBSPro-5.4 user guide, 
      # 'pcput' is the "per_process maximum CPU time (i.e. for any single
      #    process in the job" [e.g. pcput=3600].
      # 'cput' is the "Maximum aggregated CPU time required by all 
      #    processes in job" [e.g. cput=5:00:00].
      # we need a specialized function to handle the aggregation.
      'pbsfields' => ['resources_used.pcput', 'resources_used.cput'],
      'action'    => \&convert_usedcpu,
      'mandatory' => 0,
     },
     'mem_used' => 
     {
      # by 4.8 of PBSPro-5.4 user guide, 
      # 'pvmem' is the "per_process maximum amount of virtual memory (i.e. 
      #    for any single process in the job)." [e.g. pvmem=200mb].
      # 'vmem' is the "Maximum, aggregate amount of virtual memory used by
      #    all concurrent processes in the job." 
      # 'pmem' is the "per_process maximum amount of physical memory (i.e.
      #    for any single process of the job)."
      # 'mem' is the "Maximum amount of physical memory (RAM) required by 
      #    job."
      'pbsfields' => ['resources_used.pvmem', 'resources_used.vmem',
		      'resources_used.pmem' , 'resources_used.mem' ],
      'action'    => \&convert_mem,
      'mandatory' => 0,
     },
     'proc_req' => 
     {
      # see 'proc_used'. now we use only the 'Resource_List' list 
      # and drop the 'resources_used' list.
      'pbsfields' => ['Resource_List.ncpus'  ,
		      'Resource_List.nodect' ,
		      'Resource_List.nodes'  ],
      'action'    => \&convert_int,
      'mandatory' => 0,
     },
     'cpu_req' => 
     {
      # see 'runtime'
      'pbsfields' => ['Resource_List.walltime'],
      'action'    => \&convert_pbstime,
      'mandatory' => 0,
     },
     'mem_req' => 
     {
      # see 'mem_used' (simply replacing the 'resources_used' list with
      # the 'Resource_List' list).
      'pbsfields' => ['Resource_List.pvmem', 'Resource_List.vmem',
		      'Resource_List.pmem' , 'Resource_List.mem' ],
      'action'    => \&convert_mem,
      'mandatory' => 0,
     },
     'status' => 
     {
      # by 10.12.5 of PBSPro-5.4 administrator guide,
      # "The exit status of the job. If the value is less than 10000 
      # (decimal) it is the exit value of the top level process of the job, 
      # typically the shell. If the value is greater than 10000, the top 
      # process exited on a signal whose number is given by subtracting 
      # 10000 from the exit value."
      'pbsfields' => ['Exit_status'],
      'action'    => \&convert_status,
      'mandatory' => 0,
     },
     'uid' => 
     {
      # by 10.12.5 of PBSPro-5.4 administrator guide,
      # "user=username is "The user name under which the job executed."
      'pbsfields' => ['user'],
      'action'    => \&convert_user,
      'mandatory' => 0,
     },
     'gid' => 
     {
      # by 10.12.5 of PBSPro-5.4 administrator guide,
      # group=groupname is "The group name under which the job executed."
      'pbsfields' => ['group'],
      'action'    => \&convert_group,
      'mandatory' => 0,
     },
     'executable' => 
     {
      # by 10.12.5 of PBSPro-5.4 administrator guide,
      # jobname=job_name is "The name of the job."
      'pbsfields' => ['jobname'],
      'action'    => \&convert_executable,
      'mandatory' => 0,
     },
     'queue' => 
     {
      # by 10.12.5 of PBSPro-5.4 administrator guide,
      # queue=queue_name is "The name of the queue in which the job 
      #   executed."
      'pbsfields' => ['queue'],
      'action'    => \&convert_queue,
      'mandatory' => 0,
     },
     'partition' => 
     {
      # the partition is determined according to the PBS job ID 
      # (in the format number.server; the server determines the partition)
      # the 'partition' field is not a "real" PBS field. rather, it is 
      # assigned by parse_pbs_line().
      'pbsfields' => ['partition'],
      'action'    => \&convert_partition,
      'mandatory' => 0,
     },
     'prev_jobs' => 
     {
      'pbsfields' => [],
      'action'    => \&convert_invalid, # return -1
      'mandatory' => 0,
     },
     'think_time' => 
     {
      'pbsfields' => [],
      'action'    => \&convert_invalid, # return -1
      'mandatory' => 0,
     },
    );


################################################################################
################################################################################
# simple utility functions: 
# translating PBS time/memory string to a number (sec/kb respectively).
################################################################################
################################################################################


#-------------------------------------------------------------------------------
# translate PBS time string to seconds. by PBS user guide section 4.8
# a time format is:  
#                 [[hours:]minutes:]seconds[.milliseconds]
#-------------------------------------------------------------------------------
sub time2sec($) {
    
    my $t = shift;
    
    if( $t =~ m/^(\d+):(\d+):(\d+)$/ ) {
	# the common case: HH:MM:SS
	return $1*3600 + $2*60 + $3;
    }

    else {
	# the generalization: [[hours:]minutes:]seconds[.milliseconds]

	my $ms = 0;
	if( $t =~ m/([^.]+)[.](\d+)$/ ) {
	    $ms = $2 / 1000.0;
	    $t  = $1;
	}

	my $sec  = 0;
	my @secs = (1, 60, 3600);
	my @toks = split(/:/, $t);
	my $n    = scalar(@toks);

	($n <= 3 && $n >= 1) or die "bad time: '$t'";

	for(my $i=0; $i<$n; $i++) {
	    my $val =  $toks[$n-($i+1)];
	    $val    =~ m/^\d+$/ or die "bad time: '$t'";
	    $sec   +=  ( $val * $secs[$i] );
	}

	return $sec + $ms;
    }
}

#-------------------------------------------------------------------------------
# translate PBS memory string to KB. by PBS user guide section 4.8 a memory
# may be given like so:
#
#   integer[suffix]
#
# where:
#
#   suffix meaning
#   ------ --------------------------------------------
#   none   bytes
#   b|w    bytes or words
#   kb|kw  klio bytes or words (1024)
#   mb|mw  mega bytes or words (1024^2 = 1,048,576)
#   gb|gw  giga bytes or words (1024^3 = 1,073,741,824)
#-------------------------------------------------------------------------------
my $g_kb = 1024;
my %g_mem_suffix2bytes = (''   => 1                ,
			  'b'  => 1                , 'w'  => 1,
			  'kb' => $g_kb            , 'kw' => $g_kb,
			  'mb' => $g_kb*$g_kb      , 'mw' => $g_kb*$g_kb,
			  'gb' => $g_kb*$g_kb*$g_kb, 'gw' => $g_kb*$g_kb*$g_kb);
sub mem2kb($) {
    my $s = shift;
    my ($n, $suffix) = ($s =~ m/^(\d+)(.*)$/);
    ( defined($n) && defined($suffix) && exists $g_mem_suffix2bytes{$suffix})
	or die "mem2kb bug: '$s'";
    my $bytes = $n * $g_mem_suffix2bytes{$suffix};
    return $bytes / $g_kb;
}



################################################################################
################################################################################
# convert_*: the 'action's specified in %g_parse_instructions
################################################################################
################################################################################

#-------------------------------------------------------------------------------
# convert a string value (no conversion, just return the string itself)
#-------------------------------------------------------------------------------
sub convert_string($$$$$) {

    my ($swf_field, $swf_record, $r, $pbsfields) = @_; # don't name last param

    foreach my $s (@$pbsfields) {
	return( $r->{$s} )
	    if exists $r->{$s};
    }

    return -1;
}


#-------------------------------------------------------------------------------
# convert an int value.
#-------------------------------------------------------------------------------
sub convert_int($$$$$) {

    my ($swf_field, $swf_record, $r, $pbsfields) = @_; # don't name last param
    my $chosen = "-1";
    my $value  = -1;

    foreach my $f (@$pbsfields) {
	if( (exists $r->{$f}) && ($r->{$f} =~ m/^\d+$/) ) {
	    $chosen = $f;
	    $value  = $r->{$f};
	    last;
	}
    }

    ConversionLog::log_field($swf_field, $chosen);
    return $value;
}


#-------------------------------------------------------------------------------
# convert a pbs time value.
#-------------------------------------------------------------------------------
sub convert_pbstime($$$$$) {

    my ($swf_field, $swf_record, $r, $pbsfields) = @_; # don't name last param
    my $chosen = "-1";
    my $value  = -1;

    foreach my $f (@$pbsfields) {
	if( (exists $r->{$f}) && (($value = time2sec($r->{$f})) != -1) ) {
	    $chosen = $f;
	    last;
	}
    }

    ConversionLog::log_field($swf_field, $chosen);
    return $value;
}

#-------------------------------------------------------------------------------
# convert wait-time.
#-------------------------------------------------------------------------------
sub convert_runtime($$$$$) {

    my ($swf_field, $swf_record, $r, $pbsfields) = @_; # don't name last param

    # we usually have the direct resource (resources_used.walltime) and 
    # should simply convert a pbstime to seconds
    my $f  = $pbsfields->[0];
    my $rt = exists $r->{$f} ? time2sec( $r->{$f} ) : -1;
    if( $rt != -1 ) {
	ConversionLog::log_field($swf_field, $f);
	return $rt;
    }
	

    # however, jobs that were killed on signals appear to lack the direct
    # resource (at least in LPC), so let's do it using ctime/etime:
    if( exists $r->{ctime} && exists $r->{etime} ) {
	ConversionLog::log_field($swf_field, "etime-ctime");
	return $r->{etime} - $r->{ctime};
    }
    else {
	ConversionLog::log_field($swf_field, "-1");
	return -1;
    }
}


#-------------------------------------------------------------------------------
# convert wait-time.
#-------------------------------------------------------------------------------
sub convert_waittime($$$$$) {

    my ($swf_field, $swf_record, $r, $pbsfields) = @_; # don't name last param

    if( exists $r->{ctime} && exists $r->{start} ) {
	ConversionLog::log_field($swf_field, "start-ctime");
	return $r->{start} - $r->{ctime};
    }
    else {
	ConversionLog::log_field($swf_field, "-1");
	return -1;
    }
}


#-------------------------------------------------------------------------------
# convert a pbs cputime value (per_job|per_process)
#-------------------------------------------------------------------------------
sub convert_usedcpu($$$$$) {

    my ($swf_field, $swf_record, $r, $pbsfields) = @_; # don't name last param
    my $chosen     = "-1";
    my $value      = -1;
    my $found_info = 0;
    my $cpuform    = $CONFIG{spec}{$swf_field}{quantity};

    foreach my $f (@$pbsfields) {

	if( (exists $r->{$f}) && (($value = time2sec($r->{$f})) != -1) ) {

	    if( $value != 0 && ($cpuform ne $g_usedcpu_info{$f}) ) {
		# e.g. we're using per_process information, but $f is per_job
		# (or visa versa). we need a conversion between the two and
		# we must have the job's $size to do so.
		my $size = $swf_record->[ $SWF_COL2INDEX0{proc_used} ];
		if( $size < 1 ) {
		    $found_info = 1;
		    next;
		}

		if( $cpuform eq 'per_process' ) {
		    # converting per_job to per_process
		    $g_usedcpu_info{$f} eq 'per_job' or die "bug";
		    $value = int( 0.4999 + $value / $size );
		}
		else {
		    # converting per_process to per_job
		    $g_usedcpu_info{$f} eq 'per_process' or die "bug";
		    $value = * $size;
		}
	    }

	    $chosen = $f;
	    last;
	}
    }

    if( $chosen eq "-1" && $found_info ) {
	# there is information, but it is unusable :(
	# recall that the last param passed to this function ($_[-1]) is the
	# actual PBS line
	ConversionLog::report_problem
	      ('CONVERT_CPUUSED',$_[-1], "using $cpuform");
    }

    ConversionLog::log_field($swf_field, $chosen);
    return $value;
}


#-------------------------------------------------------------------------------
# convert a PBS memory value (per_job|per_process, virtual|physical)
#-------------------------------------------------------------------------------
sub convert_mem($$$$$) {

    my ($swf_field, $swf_record, $r, $pbsfields) = @_; # don't name last param
    my $chosen       = "-1";
    my $value        = -1;
    my $found_info   = 0;     # 1 = type probelm; 2 = quantity problem
    my $swf_size_col = ($swf_field eq 'mem_used') ? 'proc_used' : 'proc_req';
    my $spec         = $CONFIG{spec};

    foreach my $f (@$pbsfields) {

	next if( ! exists $r->{$f} );
	next if( ($value = mem2kb($r->{$f})) == -1 );

	# e.g. resources_used.pvmem (used) or Resource_List.pvmem (requested)
	my ($list, $resource) = split(/\./, $f); 
	($list && $resource) or die "bug";

	if( $spec->{$swf_field}{type} ne $g_mem_info{$resource}{type} ) {
	    # even though there's a value associated with $resource in $r
	    # we can't use it (can't mix virtual/physical data in same field)
	    $found_info = 1;
	    next;
	}

	if( $value != 0 &&
	    $spec->{$swf_field}{quantity} ne $g_mem_info{$resource }{quantity}){
	    # value associated with $resource is different than 0 and has the 
	    # type we've previously used in this column (physical|virtual), 
	    # but the quantity isn't the same (e.g. we've previously used 
	    # 'per_process', but $resource is 'per_job').
	    my $size = $swf_record->[ $SWF_COL2INDEX0{$swf_size_col} ];

	    if( $size < 1 ) { # must have size to perform the translation
		$found_info = 2;
		next;
	    }

	    if( $spec->{$swf_field}{quantity} eq 'per_process' )  {
		# from per_job to per_process
		($g_mem_info{$resource}{quantity} eq 'per_job') or die "bug";
		$value = int( 0.49999 + $value / $size );
	    } else {
		# from per_process to per_job
		$g_mem_info{$resource}{quantity} eq 'per_process' or die "bug";
		$value *= $size;
	    }
	}

	$chosen = $f;
	last;
    }

    if( $chosen eq "-1" && ($found_info != 0) ) {
	my $problem;
	my $info;
	if( $found_info == 1 ) { # type problem (virtual vs. physical)
	    $problem = ($swf_field eq 'mem_used')
		? 'TYPE_MEMUSED' : 'TYPE_MEMREQ';
	    $info = "using $spec->{$swf_field}{type}";
	} else {		 # quantity problem (per process vs. per-job)
	    $problem = ($swf_field eq 'mem_used')
		? 'CONVERT_MEMUSED' : 'CONVERT_MEMREQ';
	    $info = "using $spec->{$swf_field}{quantity}";
	}
	# recall that the last param passed to this function ($_[-1]) is the
	# actual PBS line
	ConversionLog::report_problem($problem, $_[-1], $info);
    }

    ConversionLog::log_field($swf_field, $chosen);
    return $value;
}


#-------------------------------------------------------------------------------
# convert a pbs time value.
#-------------------------------------------------------------------------------
sub convert_status($$$$$) {

    my ($swf_field, $swf_record, $r, $pbsfields) = @_; # don't name last param
    my $chosen = $pbsfields->[0];
    my $value  = $r->{$chosen};

    if( $r->{has_cancelation_record} == 1 ) {
	# a D/A record appeared before $r => SWF cancelation (5)
	$value = 5;		
    }
    elsif( defined($value) && ($value =~ m/^-?\d+$/) ) {
	# no cancelation, and $chosen exists in $r
	$value = ($value==0) ? 1 : 0;
    }
    else {
	$chosen = "-1";
	$value  = -1;
    }

    ConversionLog::log_field($swf_field, $chosen);
    return $value;
}


#-------------------------------------------------------------------------------
# used by covert_{uid,gid,executed,queue,partition}, that is all functions that
# must convert a string to a number in a consistent manner. 
# in addition to the usual convert_* arguments, the associated $hash (ref) is
# also passed (one of the hash refs held by %g_anonymize_tbl).
#-------------------------------------------------------------------------------
sub anonymize {

    my ($swf_field, $swf_record, $r, $pbsfields, $hash) = @_;
    my $chosen = $pbsfields->[0];
    my $str    = $r->{$chosen};
    my $value;

    if( defined($str) ) {	# there's a value associated with the PBSfield
	if( exists $hash->{$str} ) {
	    $value = $hash->{$str};
	}
	else {
	    $value = $hash->{$str} = $g_anonymize_next{$swf_field}++;
	}
    }
    else {			# can't convert
	$chosen = "-1";
	$value  = -1;
    }

    ConversionLog::log_field($swf_field, $chosen);
    return $value;
}


#-------------------------------------------------------------------------------
# convert a pbs user/group/executable/queue/partition.
# no need to pass the last received param which is the $pbsline
#-------------------------------------------------------------------------------
my $atbl = \%g_anonymize_tbl;
sub convert_user      ($$$$$) { return anonymize(@_[0..3],$atbl->{uid}       );}
sub convert_group     ($$$$$) { return anonymize(@_[0..3],$atbl->{gid}       );}
sub convert_executable($$$$$) { return anonymize(@_[0..3],$atbl->{executable});}
sub convert_queue     ($$$$$) { return anonymize(@_[0..3],$atbl->{queue}     );}
sub convert_partition ($$$$$) { return anonymize(@_[0..3],$atbl->{partition} );}


#-------------------------------------------------------------------------------
# return -1 on any input
#-------------------------------------------------------------------------------
sub convert_invalid($$$$$) { return -1; }


#-------------------------------------------------------------------------------
# initialize %g_anonymize_tbl according to user prefs (%CONFIG).
#-------------------------------------------------------------------------------
sub init_anonymize_tbl() {

    while( my ($swf_field, $tbl) = each %{$CONFIG{anonymize}} ) {
	while( my ($pbsval, $swfval) = each %$tbl ) {
	    $g_anonymize_tbl {$swf_field}{$pbsval} = $swfval;
	    $g_anonymize_next{$swf_field}          = $swfval+1
		if( $swfval >= $g_anonymize_next{$swf_field} );
	}
    }
}



################################################################################
################################################################################
# parsing the PBS file
################################################################################
################################################################################


#-------------------------------------------------------------------------------
# FORMAT OF A PBS RECORD
# ~~~~~~~~~~~~~~~~~~~~~~
#
# - Explained in:
#   1) PBSProAG.pdf administrator guide:
#      - pp 195 sec 10.12.4 (The Accounting Log)
#      - pp 221 (explanation of diff between 'walltime' and 'end - start'
#   2) 
#
# - The PBS record structure is: 
#
#   ####################################################
#   #                                                  #
#   #   date_time;record_type;id_string;message_text   #
#   #                                                  #
#   ####################################################
#
# - date time:
#   Their format is: mm/dd/yyyy hh:mm:ss
#
# - id_string:
#   Either the job identifier (job ID) or a PBS reservation identifier 
#   (seems irrelevant to SWF). In the LPC log, this may look like:
#
#        468.clrce01.in2p3.fr
#
#   where 468 is the job-ID on the cluster partition served by clrce01.in2p3.fr
#
# - message_text:
#   The content depends on the record_type. The message text format is blank 
#   separated keyword=value fields.
#
# - record_type:
#   May be one of the following:
#   
#   A = Job was aborted by server.
#   B = Reservation related (beginning of reservation period)
#   C = Job was check-pointed and held
#   D = Job was deleted by request; msg contains requestor=user@host
#   E = Job ended (terminated execution); msg contains all data needed for SWF
#   F = Reservation related (resources reservation period finished)
#   K = Reservation related (scheduler/server requested removal of reservation)
#   Q = Job entered a queue; record for each move; msg contains queue=name
#   R = Job was rerun
#   S = Job execution started
#
# - Relevant record types:
#     Q = indicate arrival
#   A/D = pressived as cancelations
#     E = indicate termination and contains all data used in the SWF
#
# - Relevant data from 'E' records
#   ...
#
#
# RETURN
# ~~~~~~
#   ($h, $problem, $info)
#   Upon failure, $h=undef, $problem is a key in %g_problems and $info further
#   describes the problem (or is undef).
#   Upon success, $h{$pbs_resource} = $value, and $problem = $info = undef.
#-------------------------------------------------------------------------------
sub parse_pbs_line ($) {

    my $line = shift;
    my %r    = ();

    # 1- record fromat is: date time;record_type;id_string;message_text
    return (undef, 'PARSE_FORMAT', undef)
	if $line !~ m/^([^;]+);([^;]+);([^;]+);([^;]+)$/ ;
    my $datetime = $1;
    my $rtype    = $2;
    my $id       = $3;
    my $msg      = $4;

    # 2- date & time format is: mm/dd/yyyy hh:mm:ss
    #    compute the associated epoch.
    #    note: we're using timelocal() because this script has set the
    #    appropriate time zone in the environment variable $ENV{TZ}
    return (undef, 'PARSE_DATE', "bad date/time: $datetime")
	if $datetime !~ m<^(\d{2}/\d{2}/\d{4})\s+(\d{2}:\d{2}:\d{2})$> ;
    $r{datetime}            = $datetime;
    $r{line_date}           = $1;
    $r{line_time}           = $2;
    my ($mon, $mday, $year) = split(/\//,$r{line_date});
    my ($hr , $min , $sec ) = split(/:/ ,$r{line_time});
    $r{line_epoch} = timelocal($sec,$min,$hr,$mday,$mon-1,$year);

    # 3- identifier format is: job_number.host
    return (undef, 'PARSE_PBSID', "bad id: $id")
	if $id !~ m/^(\d+)[.](\S+)$/ ;
    $r{pbsID}     = $id;
    $r{job}       = $1;
    $r{partition} = $2;


    # 4- record type is
    return (undef, 'PARSE_RECTYPE', "unknown type: $rtype")
	if $rtype !~ m/^[ABCDEFKQRS]$/ ;
    $r{record_type} = $rtype;

    # 5- msg format is key=value separated by white space
    my @pairs = split(/\s+/, $msg);
    foreach my $pair (@pairs) {
	return (undef, 'PARSE_PAIR', "bad token: $pair")
	    if $pair !~ m/^([^=]+)=([^=]+)$/ ;
	$r{$1} = $2;
    }


    #print "'$r{job}' '$r{partition}'\n";
    return (\%r, undef, undef);
}


#-------------------------------------------------------------------------------
# translate to SWF a job that was actually started. such jobs are always 
# associated with an E recored that usually contains most of the information 
# required when performing the translation.
#
# get:
#         $e is a ref2hash as returned by parse_pbs_line() of the E record 
#            associated with the job. 
#  $canceled indicates whether the job was canceled while it was running (1) 
#            or terminated normally (0).
#   $pbsline the pbs line from which $e was constructed.
#
# return $swf_rec
#   which is a ref2arr representing one line in the SWF (expect from the 
#   job ID which is still in PBS format), or undef if some mandatory field 
#   could not be extracted from $e.
#-------------------------------------------------------------------------------
sub pbs2swf_started($$$)
{
    my ($e, $canceled, $pbsline) = @_;

    $e->{has_cancelation_record} = $canceled;
    my @swf_record = map(-1, @SWF_COLS);	 # init all fields to -1s

    foreach my $swf_field (@SWF_COLS) {

	# convert
	my $p = $g_parse_instructions{$swf_field};
	my $v = $p->{action} ( $swf_field, 
			       \@swf_record, 
			       $e, 
			       $p->{pbsfields},
			       $pbsline );
	
	# if needed, use default value (if the default is 'overwrite', this 
	# will be done after the entire PBS log is parsed, so we can know
	# what we've missed...)
	if( $v eq "-1" ) { # need to use 'eq' because jobID is still a string
	    my $d = $CONFIG{defaults}{$swf_field};
	    $v = $d->{value}
		if( defined($d) && $d->{jobs} ne 'canceled');
	}

	# make sure mandatory fields appear
	if( $v eq "-1" && $p->{mandatory} ) {
	    ConversionLog::report_problem
		  ('MANDATORY', $pbsline, "swf_field=$swf_field");
	    return undef;
	}

	my $idx           = $SWF_COL2INDEX0{$swf_field};
	$swf_record[$idx] = $v;
    }

    ConversionLog::inc_njobs_started();
    return \@swf_record;
}


#-------------------------------------------------------------------------------
# translate to SWF a job that was canceled before it was started. in this 
# case there's no E or S records and so the information that can be retrieved 
# is very limited: status, arrival time, wait time, queue, and partition.
#
# get: 
# two ref2hash as returned by parse_pbs(). the 1st is of the 1st Q record
# associated with the job (indicates arrival), and the 2nd is of the 1st
# cancelation record D or A (indicates cancelation time).
#
# return $swf_rec
#   which is a ref2arr representing one line in the SWF (expect from the 
#   job ID which is still in PBS format), or undef if the arrival time
#   is after the cancelation time and the problem is not due to daylight 
#   saving time. 
#-------------------------------------------------------------------------------
sub pbs2swf_not_started($$$)
{
    my ($q, $d, $pbsline) = @_;

    my $arrived  = $q->{line_epoch};
    my $canceled = $d->{line_epoch};

    # since the time format used in PBS logs doesn't contain information about
    # daylight savings, it may happen that $canceled will be smaller than
    # $arrived by up to one hour
    if( $arrived > $canceled ) {
	my $diff = $arrived - $canceled;
	if( $diff > 3600 ) {
	    ConversionLog::report_problem('CANCEL_TIME', $pbsline, 
	        "arrived[$arrived] - canceled[$canceled] = $diff > 3600");
	    return undef;
	}
	$canceled += 3600;
    }

    my $i   = \%SWF_COL2INDEX0;		     # we need a shorter name
    my @j   = map(-1, @SWF_COLS);	     # init all fields to -1s
    my $ano = \%g_anonymize_tbl;	     # for short

    $j[ $$i{job}       ] = $q->{pbsID};
    $j[ $$i{submit}    ] = $arrived;
    $j[ $$i{wait}      ] = $canceled - $arrived;
    $j[ $$i{proc_used} ] = 0;
    $j[ $$i{status}    ] = 5;  	             # 5 = canceled_status
    $j[ $$i{queue}     ] = 
       convert_queue    ('queue'    ,undef,$q,['queue']    ,$ano->{queue}    );
    $j[ $$i{partition} ] = 
       convert_partition('partition',undef,$q,['partition'],$ano->{partition});

    # apply defaults
    my $defaults = $CONFIG{defaults};
    foreach my $swf_field (keys %{$defaults}) {

	my $d    = $defaults->{$swf_field};
	next if( ! defined($d) || ($d->{jobs} eq 'started') );
	my $idx  = $$i{$swf_field};

	$j[$idx] = $d->{value}
	    if( $j[$idx]==-1 || $d->{overwrite} );
    }

    ConversionLog::inc_njobs_canceled();
    return \@j;
}


#-------------------------------------------------------------------------------
# comapre pbsid string [serial.partition, e.g. 36350.clrglop195.in2p3.fr]
# based on the partition, and then, based on the serial.
#-------------------------------------------------------------------------------
sub cmp_pbsid_partition_serial {
    my ($s1, $p1) = ( $a =~ m/^(\d+)\.(.*)$/ );
    my ($s2, $p2) = ( $b =~ m/^(\d+)\.(.*)$/ );
    my $c         = $p1 cmp $p2;
    return ($c != 0) ? $c : ($s1 <=> $s2);
}


#-------------------------------------------------------------------------------
# compare ref2arr-s representing jobs (content of @$swf after initial parse) 
# according to arrival / partition / serial of pbsid 
#-------------------------------------------------------------------------------
my $i_sbm = $SWF_COL2INDEX0{submit};
my $i_job = $SWF_COL2INDEX0{job};
my $i_prt = $SWF_COL2INDEX0{partition};
sub cmp_swfjobs {
    
    return $a->[$i_sbm] <=> $b->[$i_sbm]
	if( $a->[$i_sbm] != $b->[$i_sbm] );

    return $a->[$i_prt] <=> $b->[$i_prt]
	if( $a->[$i_prt] != $b->[$i_prt] );

    # in the case both submission time and partition are the same, use 
    # the serial embedded in the PBSID.

    my ($s1) = ( $a->[$i_job] =~ m/^(\d+)\./ );
    my ($s2) = ( $b->[$i_job] =~ m/^(\d+)\./ );

    return $s1 <=> $s2
	if( $s1 != $s2 );

    die "bug: two jobs with identical: arrival / partition / serial of pbsid\n",
	"a[$a]: @$a\n", "b[$b]: @$b";
}


#-------------------------------------------------------------------------------
# the $swf is almost in standard format what's left is:
# 1) need to convert the 'job', which is a PBS string, to an SWF ID
# 2) need to sort jobs by arrival. 
# 3) need to normalize arrival time such that first job will arrive at 0
#
# return: $epoch_start - the epoch time of the first arrival.
#-------------------------------------------------------------------------------
sub finalize_swf($) {

    my $swf         = shift;
    @$swf           = sort cmp_swfjobs @$swf;
    my $epoch_start = $swf->[0][$i_sbm];
    my $j           = 0;
    my $debug       = $CONFIG{other}{debug};

    foreach my $job (@$swf) {
	$job->[$i_sbm] -= $epoch_start;	# normalize submit time
	push @$job, $job->[$i_job] 	# save PBS-ID for debug
	    if $debug;
	$job->[$i_job]  = ++$j;		# convert PBS-ID to SWF-ID
    }

    return $epoch_start;
}


#-------------------------------------------------------------------------------
# return ($swf, $anonymize_tbls)
# @$swf: 
#   holds ref2arr-s such that each arr represents one SWF job (fields ordered is
#   determined by %SWF_COL2INDEX, that is, the first column is the job-id etc).
#-------------------------------------------------------------------------------
sub parse_pbs()
{
    # BACKGROUND:
    # iterate through the PBS log, line by line:
    #
    #               change
    #               queue
    #                 Q
    #               +->--+
    #               |    |
    #              /|\  \|/
    #               |    |
    #     Q    +----+-----------+  S   +-------------+    E     +----------+
    # -------->| wait queue [1] |----->| running [2] |--------->| end [5]  |
    #  arrival +------+---------+      +------+------+          +--+-------+
    #    [0]          |                       |                    |
    #                \|/                     \|/                  /|\
    #               A | D                   A | D                  |
    #                 |                       |                    |
    #           +-----+--------+      +-------+------+    E        |
    #           | canceled [3] |      | canceled [4] |----->-------+
    #           +---+----+-----+      +---+----+-----+
    #               |    |                |    |
    #              \|/  /|\              \|/  /|\
    #               |    |                |    |
    #               +->--+                +->--+ 
    #                A/D                    A/D 
    #
    # several records are associated with a job (within a partition) throughout 
    # its lifetime:
    # - the first Q record indicates the job's arrival.
    # - this record may be followed by additional Q records indicating the
    #   job moves between queues (each record specifies the destination queue
    #   which apparently may be identical to the source).
    # - the job can than be canceled (A or D record) in which case it is
    #   removed from the system [3]. there can be more than one cancelation
    #   record for the same job (e.g. presumably one for every time the user 
    #   clicked on the 'delete' GUI button).
    # - if the job is not canceled, at some point, it is started (S record).
    # - then, it can either terminate normally or be canceled. in both cases
    #   an E record marks the job "leaving" the system [5], but in the latter 
    #   case, A/D record(s) [4] will mark the cancelation.
    #   
    # VERIFICATION OF THE ABOVE FLOW CHART:
    # only jobs that follow the above flow chart are included in the SWF file.
    # the rest of the records will be reported to the user as unusable.
    # here's the records sequence of all the jobs in the LPC log, with an
    # explanation of why it may happen that some violate the flow chart:
    #
    # ========|======|=========================|================================
    #   RECORD|  JOBS|   JOB PBS ID FOR EXAMPLE| DESCRIPTION
    #     FLOW|      |                         |
    # ========|======|=========================|================================
    #      QSE|223838|16443.clrglop195.in2p3.fr| "normal" jobs [5]
    #    QQQSE|     1|10610.clrce02.in2p3.fr   |
    # --------|------|-------------------------|--------------------------------
    #     QSDE|  2361|43671.clrce01.in2p3.fr   | started jobs that were canceled
    # QSD...DE|     7|33836.clrce01.in2p3.fr   | (reached [5] through [4])
    # --------|------|-------------------------|--------------------------------
    #       QD| 14362|33885.clrglop195.in2p3.fr| job that were canceled before
    #      QDD|     4|43569.clrce02.in2p3.fr   | they were started
    #   QA...A|     7|49921.clrce02.in2p3.fr   |
    # --------|------|-------------------------|--------------------------------
    #        Q|    25|36406.clrglop195.in2p3.fr| jobs for which there's no E
    #       QS|    10|49927.clrce02.in2p3.fr   | record becuase it's located
    #      QSD|    16|78876.clrce01.in2p3.fr   | after the log's "end"
    # --------|------|-------------------------|--------------------------------
    #   QSEQSE|  2106|  835.clrglop195.in2p3.fr| job ID wraparound; 
    #  QSEQSDE|    14| 2202.clrglop195.in2p3.fr| see explanation below
    #     QSEE|     1|  290.clrglop195.in2p3.fr| 
    # --------|------|-------------------------|--------------------------------
    #     QSRD|     2| 5393.clrce02.in2p3.fr   | unknown why there's no E
    # ========|======|=========================|================================
    #
    # THE WRAPAROUND:
    # The ID wraparound (job IDs are reused) occurred within LPC at 26 Nov 2004,
    # when the clrglop195.in2p3.fr machine was upgraded. 
    # so the QSEQSE and QSEQSDE records are simply a result of a job ID reuse. 
    # the QSEE is more subtle: 
    # there's a "hole" in the LPC log such that jobs 1-429 on the 
    # clrglop195.in2p3.fr partition are missing. the only record that survived 
    # is the E of (the second occurrence of) job 290.
    #
    # PARSING:
    # if there's an E record [5], we take all the information from there. 
    # otherwise (the job was canceled before it began to run) we must infer all
    # the available information from the Q and D (or A) records, which is not a
    # lot.
    # due to possible wraparounds, once an E record is encountered [5],
    # or a D/A record of state [3], we treat the next Q occurrence with the same
    # job ID as valid: it is simply considered as a reuse of
    # an ID for a new job.


    # the stats matching the flow chart above
    my $STATE_WAITING          = 1;
    my $STATE_RUNNING          = 2;
    my $STATE_CANCELED         = 3;
    my $STATE_CANCELED_RUNNING = 4;
    my $STATE_ENDED            = 5;


    # dast
    my @swf;			# the returned value
    my %states;			# from $pbsID to $STATE_*
    my %records;		# $records{$pbsID}{$rtype} = $r (parse_pbs_line)


    # 1- initialize anonymization tables
    init_anonymize_tbl();


    # 2- parse PBS log
    while(<>) {

	next if m/^\s*$/;
	next if $_ !~ m/;(Q|S|E|D|A);/;

 	my ($r, $problem, $problem_info) = parse_pbs_line($_);
	if( ! defined($r) ) {
	    ConversionLog::report_problem($problem, $_, $problem_info);
	    next;
	}

	my $pbsID = $r->{pbsID};
	my $rtype = $r->{record_type};
	my $state = $states{$pbsID};

	# [0] arrival
	if( ! defined($state) ) {

	    if( $rtype eq 'Q' ) { 	
		# arrival[0] => waiting[1] 
		$states {$pbsID}    = $STATE_WAITING;
		$records{$pbsID}{Q} = $r;
	    }
	    else {			
		# flow chart violation
		ConversionLog::report_problem('FLOW_ARRIVAL', $_,"type=$rtype");
	    }
	}


	# [1] waiting
	elsif( $state == $STATE_WAITING ) {

	    if( $rtype eq 'S' ) { 	
		# waiting[1] => running[2]
		$states{$pbsID} = $STATE_RUNNING;
		delete $records{$pbsID}{Q}; # no need to save as we'll use E
	    }
	    elsif( $rtype eq 'D' || $rtype eq 'A' ) {
		# waiting[1] => canceled[3]
		$states{$pbsID}= $STATE_CANCELED;
		my $j          = pbs2swf_not_started($records{$pbsID}{Q},$r,$_);
		push( @swf, $j ) if( defined($j) ); 
		delete $records{$pbsID}{Q};
	    }
	    elsif( $rtype eq 'Q' ) {
		# waiting[1] still (queue transition)
	    }
	    else {
		# flow chart violation
		ConversionLog::report_problem('FLOW_WAIT', $_, "type=$rtype");
	    }
	}


	# [2] running
	elsif( $state == $STATE_RUNNING ) {

	    if( $rtype eq 'E' ) {
		# running[2] => ended[5]
		my $j = pbs2swf_started($r,0,$_);
		push( @swf, $j ) if( defined($j) );
		delete $states{$pbsID};
	    }
	    elsif( $rtype eq 'D' || $rtype eq 'A' ) {
		# running[1] => canceled[3]
		$states{$pbsID} = $STATE_CANCELED_RUNNING;
	    }
	    else {
		# flow chart violation
		ConversionLog::report_problem('FLOW_RUN', $_, "type=$rtype");
	    }
	}


	# [3] canceled (before the job started)
	elsif( $state == $STATE_CANCELED ) {

	    if( $rtype eq 'D' || $rtype eq 'A' ) { 
		# ignoring, user clicked 'delete' more than once, or server
		# repeated an 'abort' command more than once.
	    }
	    elsif( $rtype eq 'Q' ) {
		# this can only happen if there was a wraparound, that is, a
		# job id is being used twice for some reason. in any case,
		# the new job is now waiting for schedule...
		ConversionLog::report_problem('WRAPAROUND', $_, "type=$rtype");
		$states {$pbsID}    = $STATE_WAITING;
		$records{$pbsID}{Q} = $r;
	    }
	    else {			
		# flow chart violation
		ConversionLog::report_problem
		      ('FLOW_CANCEL_BEFORE_RUN',$_,"type=$rtype");
	    }
	}


	# [4] canceled running (cancelation after job has been started)
	elsif( $state == $STATE_CANCELED_RUNNING ) {

	    if( $rtype eq 'E' ) { 	
		# canceled[4] => ended[5]
		my $j = pbs2swf_started($r,1,$_);
		push( @swf, $j ) if( defined($j) );
		delete $states{$pbsID};
	    }
	    elsif( $rtype eq 'D' || $rtype eq 'A' ) { 
		# ignoring, user clicked 'delete' more than once, or server
		# repeated an 'abort' command more than once.
	    }
	    else {			
		# flow chart violation
		ConversionLog::report_problem
		      ('FLOW_CANCEL_AFTER_RUN',$_, "type=$rtype");
	    }
	}
    }


    # 3- there may be jobs with an unknown end...
    my %state2str = 
	(
	 $STATE_WAITING          => 'WAITING',
	 $STATE_RUNNING          => 'RUNNING',
	 $STATE_CANCELED         => 'CANCELED',
	 $STATE_CANCELED_RUNNING => 'CANCELED_RUNNING',
	 $STATE_ENDED            => 'ENDED'
	);
    foreach my $pbsid (sort cmp_pbsid_partition_serial keys(%states)) {
	my $state = $states{$pbsid};
	my $inf   = $state2str{$state};
	next if $state == $STATE_CANCELED; # this is allowed; see flow-char

	ConversionLog::report_problem
	      ('FLOW_UNFINISHED', "$pbsid ", "last state = $inf");
    }
    
    # 4- still need to convert the 'job', which is a PBS string, to an SWF ID
    #    and to normalize arrival time
    ( scalar(@swf) > 0 ) 
	or die "didn't find finished PBS jobs in input logs.\n";
    my $start = finalize_swf(\@swf);


    return (\@swf, \%g_anonymize_tbl, $start);
}


################################################################################
#                                   EOF
################################################################################
