--- trunk/run.pl 2006/07/02 12:42:02 559 +++ trunk/run.pl 2006/08/01 13:59:47 606 @@ -7,20 +7,23 @@ use lib './lib'; use WebPAC::Common 0.02; -use WebPAC::Lookup; -use WebPAC::Input 0.03; +use WebPAC::Lookup 0.03; +use WebPAC::Input 0.07; use WebPAC::Store 0.03; -use WebPAC::Normalize; +use WebPAC::Normalize 0.11; use WebPAC::Output::TT; use WebPAC::Validate; +use WebPAC::Output::MARC; use YAML qw/LoadFile/; use Getopt::Long; use File::Path; use Time::HiRes qw/time/; use File::Slurp; -use MARC::Record 2.0; # need 2.0 for utf-8 encoding see marcpm.sf.net -use MARC::Lint; use Data::Dump qw/dump/; +use Storable qw/dclone/; + +use Proc::Queue size => 1; +use POSIX ":sys_wait_h"; # imports WNOHANG =head1 NAME @@ -81,6 +84,11 @@ Force dump or input and marc record for debugging. +=item --parallel 4 + +Run databases in parallel (aproximatly same as number of processors in +machine if you want to use full load) + =back =cut @@ -98,6 +106,8 @@ my $marc_lint = 1; my $marc_dump = 0; +my $parallel = 0; + GetOptions( "limit=i" => \$limit, "offset=i" => \$offset, @@ -105,13 +115,14 @@ "one=s" => \$only_filter, "only=s" => \$only_filter, "config" => \$config, - "debug" => \$debug, + "debug+" => \$debug, "stats" => \$stats, "validate=s" => \$validate_path, "marc-normalize=s" => \$marc_normalize, "marc-output=s" => \$marc_output, "marc-lint!" => \$marc_lint, "marc-dump!" => \$marc_dump, + "parallel=i" => \$parallel, ); $config = LoadFile($config); @@ -145,13 +156,25 @@ my @links; my $indexer; -my $lint = new MARC::Lint if ($marc_lint); +if ($parallel) { + $log->info("Using $parallel processes for speedup"); + Proc::Queue::size($parallel); +} while (my ($database, $db_config) = each %{ $config->{databases} }) { my ($only_database,$only_input) = split(m#/#, $only_filter) if ($only_filter); next if ($only_database && $database !~ m/$only_database/i); + if ($parallel) { + my $f=fork; + if(defined ($f) and $f==0) { + $log->info("Created processes $$ for speedup"); + } else { + next; + } + } + if ($use_indexer) { my $indexer_config = $config->{$use_indexer} || $log->logdie("can't find '$use_indexer' part in confguration"); $indexer_config->{database} = $database; @@ -224,9 +247,13 @@ die "I know only how to handle input types ", join(",", @supported_inputs), " not '$type'!\n" unless (grep(/$type/, @supported_inputs)); - my $lookup = new WebPAC::Lookup( - lookup_file => $input->{lookup}, - ) if ($input->{lookup}); + my $lookup; + if ($input->{lookup}) { + $lookup = new WebPAC::Lookup( + lookup_file => $input->{lookup}, + ); + delete( $input->{lookup} ); + } my $input_module = $config->{webpac}->{inputs}->{$type}; @@ -236,12 +263,16 @@ my $input_db = new WebPAC::Input( module => $input_module, - code_page => $config->{webpac}->{webpac_encoding}, + encoding => $config->{webpac}->{webpac_encoding}, limit => $limit || $input->{limit}, offset => $offset, - lookup => $lookup, + lookup_coderef => sub { + my $rec = shift || return; + $lookup->add( $rec ); + }, recode => $input->{recode}, stats => $stats, + modify_records => $input->{modify_records}, ); $log->logdie("can't create input using $input_module") unless ($input); @@ -271,13 +302,11 @@ $log->info("Using $normalize_path for normalization..."); - my $marc_fh; - if (my $path = $normalize->{output}) { - open($marc_fh, '>', $path) || - $log->logdie("can't open MARC output $path: $!"); - - $log->info("Creating MARC export file $path", $marc_lint ? ' (with lint)' : '', "\n"); - } + my $marc = new WebPAC::Output::MARC( + path => $normalize->{output}, + lint => $marc_lint, + dump => $marc_dump, + ) if ($normalize->{output}); # reset position in database $input_db->seek(1); @@ -300,11 +329,22 @@ $log->error( "MFN $mfn validation errors:\n", join("\n", @errors) ) if (@errors); } - + my $ds_config = dclone($db_config); + + # default values -> database key + $ds_config->{_} = $database; + + # current mfn + $ds_config->{_mfn} = $mfn; + + # attach current input + $ds_config->{input} = $input; + my $ds = WebPAC::Normalize::data_structure( row => $row, rules => $rules, lookup => $lookup ? $lookup->lookup_hash : undef, + config => $ds_config, marc_encoding => 'utf-8', ); @@ -320,34 +360,20 @@ type => $config->{$use_indexer}->{type}, ) if ($indexer && $ds); - if ($marc_fh) { - my $marc = new MARC::Record; - $marc->encoding( 'utf-8' ); - my @marc_fields = WebPAC::Normalize::_get_marc_fields(); - if (! @marc_fields) { - $log->warn("MARC record $mfn is empty, skipping"); - } else { - $marc->add_fields( @marc_fields ); - - if ($marc_lint) { - $lint->check_record( $marc ); - my $err = join( "\n", $lint->warnings ); - $log->error("MARC lint detected warning on MFN $mfn\n", - "Original imput row: ",dump($row), "\n", - "Normalized MARC row: ",dump(@marc_fields), "\n", - "MARC lint warnings: ",$err,"\n" - ) if ($err); - } - - if ($marc_dump) { - $log->info("MARC record on MFN $mfn\n", - "Original imput row: ",dump($row), "\n", - "Normalized MARC row: ",dump(@marc_fields), "\n", - ); - } + if ($marc) { + my $i = 0; - print $marc_fh $marc->as_usmarc; + while (my $fields = WebPAC::Normalize::_get_marc_fields( fetch_next => 1 ) ) { + $marc->add( + id => $mfn . ( $i ? "/$i" : '' ), + fields => $fields, + leader => WebPAC::Normalize::marc_leader(), + row => $row, + ); + $i++; } + + $log->info("Created $i instances of MFN $mfn\n") if ($i > 1); } $total_rows++; @@ -356,7 +382,7 @@ $log->info("statistics of fields usage:\n", $input_db->stats) if ($stats); # close MARC file - close($marc_fh) if ($marc_fh); + $marc->finish if ($marc); } @@ -389,9 +415,22 @@ } } + # end forked process + if ($parallel) { + $log->info("parallel process $$ finished"); + exit(0); + } + +} + +if ($parallel) { + # wait all children to finish + sleep(1) while wait != -1; + $log->info("all parallel processes finished"); } foreach my $link (@links) { $log->info("adding link $link->{from} -> $link->{to} [$link->{credit}]"); $indexer->add_link( %{ $link } ); } +