Lines Matching refs:self
3431 my $self = shift;
3433 unlink $self->{'loadavg_file'};
3434 unlink $self->{'swap_activity_file'};
3438 my $self = shift;
3439 return $self->{'string'};
3443 my $self = shift;
3445 return ($self->{'jobs_running'} || "0");
3449 my $self = shift;
3450 $self->{'jobs_running'}++;
3454 my $self = shift;
3455 $self->{'jobs_running'}--;
3459 my $self = shift;
3460 $self->{'maxlength'} = shift;
3464 my $self = shift;
3465 return $self->{'maxlength'};
3469 my $self = shift;
3470 return $self->{'jobs_completed'};
3479 my $self = shift;
3480 return grep { defined $self->{'hostgroups'}{$_} } @_;
3484 my $self = shift;
3485 return keys %{$self->{'hostgroups'}};
3489 my $self = shift;
3490 $self->{'jobs_completed'}++;
3494 my $self = shift;
3495 if(defined $self->{'max_jobs_running'}) {
3496 $Global::max_jobs_running -= $self->{'max_jobs_running'};
3498 $self->{'max_jobs_running'} = shift;
3499 if(defined $self->{'max_jobs_running'}) {
3501 $Global::max_jobs_running += $self->{'max_jobs_running'};
3504 $self->{'orig_max_jobs_running'} ||= $self->{'max_jobs_running'};
3508 my $self = shift;
3509 my $swapping = $self->swap_activity();
3518 my $self = shift;
3521 if(-r $self->{'swap_activity_file'}) {
3522 open(my $swap_fh, "<", $self->{'swap_activity_file'}) || ::die_bug("swap_activity_file-r");
3526 $self->{'swap_activity'} = $1;
3527 ::debug("swap", "New swap_activity: ", $self->{'swap_activity'});
3529 ::debug("swap", "Last update: ", $self->{'last_swap_activity_update'});
3530 if(time - $self->{'last_swap_activity_update'} > 10) {
3532 ::debug("swap", "Older than 10 sec: ", $self->{'swap_activity_file'});
3536 ::debug("swap", "No swap_activity file: ", $self->{'swap_activity_file'});
3537 $self->{'swap_activity'} = undef;
3541 ::debug("swap", "Updating swap_activity file ", $self->{'swap_activity_file'});
3542 $self->{'last_swap_activity_update'} = time;
3547 if($self->{'string'} ne ":") {
3548 $swap_activity = $self->sshcommand() . " " . $self->serverlogin() . " " .
3554 my $file = $self->{'swap_activity_file'};
3559 return $self->{'swap_activity'};
3694 my $self = shift;
3695 if($self->{'last_login_at'} and $self->{'time_to_login'}) {
3700 my $too_fast = (::now() <= $self->{'last_login_at'}
3701 + $self->{'time_to_login'}/5);
3711 my $self = shift;
3712 return $self->{'last_login_at'};
3716 my $self = shift;
3717 $self->{'last_login_at'} = shift;
3721 my $self = shift;
3722 my $loadavg = $self->loadavg();
3724 $loadavg > $self->max_loadavg());
3737 my $self = shift;
3740 if(open(my $load_fh, "<", $self->{'loadavg_file'})) {
3747 $self->{'loadavg'} = $load - 1;
3748 ::debug("load", "New loadavg: ", $self->{'loadavg'});
3752 ::debug("load", "Last update: ", $self->{'last_loadavg_update'});
3753 if(time - $self->{'last_loadavg_update'} > 10) {
3755 ::debug("load", time - $self->{'last_loadavg_update'}, " secs old: ",
3756 $self->{'loadavg_file'});
3760 ::debug("load", "No loadavg file: ", $self->{'loadavg_file'});
3761 $self->{'loadavg'} = undef;
3765 ::debug("load", "Updating loadavg file", $self->{'loadavg_file'}, "\n");
3766 $self->{'last_loadavg_update'} = time;
3770 if($self->{'string'} ne ":") {
3771 $cmd = $self->sshcommand() . " " . $self->serverlogin() . " ";
3777 my $file = $self->{'loadavg_file'};
3781 return $self->{'loadavg'};
3785 my $self = shift;
3796 if(not defined $self->{'max_loadavg'}) {
3797 $self->{'max_loadavg'} =
3798 $self->compute_max_loadavg($opt::load);
3800 ::debug("load", "max_loadavg: ", $self->string(), " ", $self->{'max_loadavg'});
3801 return $self->{'max_loadavg'};
3805 my $self = shift;
3806 $self->{'max_loadavg'} = shift;
3813 my $self = shift;
3821 $self->ncpus() + $j;
3826 $self->ncpus() - $j;
3830 $self->ncpus() * $j / 100;
3839 $load = $self->compute_max_loadavg($opt_load_file);
3856 my $self = shift;
3857 return $self->{'time_to_login'};
3861 my $self = shift;
3862 $self->{'time_to_login'} = shift;
3866 my $self = shift;
3867 if(not defined $self->{'max_jobs_running'}) {
3868 my $nproc = $self->compute_number_of_processes($opt::jobs);
3869 $self->set_max_jobs_running($nproc);
3871 return $self->{'max_jobs_running'};
3875 my $self = shift;
3876 return $self->{'orig_max_jobs_running'};
3883 my $self = shift;
3885 my $wanted_processes = $self->user_requested_processes($opt_P);
3891 $self->processes_available_by_system_limit($wanted_processes);
3903 my $self = shift;
4057 if($self->string() ne ":" and
4060 $self->simultaneous_sshlogin_limit($system_limit);
4069 my $self = shift;
4071 if($self->{'time_to_login'}) {
4078 ::min($self->simultaneous_sshlogin($wanted_processes),
4079 $self->simultaneous_sshlogin($wanted_processes));
4081 my $serverlogin = $self->serverlogin();
4100 my $self = shift;
4102 my $sshcmd = $self->sshcommand();
4103 my $serverlogin = $self->serverlogin();
4116 my $self = shift;
4117 $self->{'ncpus'} = shift;
4124 my $self = shift;
4132 $self->ncpus() + $j;
4137 $self->ncpus() - $j;
4142 $self->ncpus() * $j / 100;
4155 $processes = $self->user_requested_processes($opt_P_file);
4170 my $self = shift;
4171 if(not defined $self->{'ncpus'}) {
4172 my $sshcmd = $self->sshcommand();
4173 my $serverlogin = $self->serverlogin();
4176 $self->{'ncpus'} = no_of_cpus();
4178 $self->{'ncpus'} = no_of_cores();
4191 $self->{'ncpus'} = $ncpu;
4195 $self->{'ncpus'} = 1;
4199 return $self->{'ncpus'};
4627 my $self = shift;
4628 if (not defined $self->{'sshcommand'}) {
4629 $self->sshcommand_of_sshlogin();
4631 return $self->{'sshcommand'};
4635 my $self = shift;
4636 if (not defined $self->{'serverlogin'}) {
4637 $self->sshcommand_of_sshlogin();
4639 return $self->{'serverlogin'};
4651 my $self = shift;
4653 if($self->{'string'} =~ /(.+) (\S+)$/) {
4660 my $control_path = $self->control_path_dir()."/ssh-%r@%h:%p";
4662 $serverlogin = $self->{'string'};
4663 if(not $self->{'control_path'}{$control_path}++) {
4684 $sshcmd = "ssh"; $serverlogin = $self->{'string'};
4687 $self->{'sshcommand'} = $sshcmd;
4688 $self->{'serverlogin'} = $serverlogin;
4694 my $self = shift;
4695 if(not defined $self->{'control_path_dir'}) {
4698 $self->{'control_path_dir'} =
4703 return $self->{'control_path_dir'};
4713 my $self = shift;
4728 my $sshcmd = $self->sshcommand();
4730 my $serverlogin = $self->serverlogin();
4743 my $self = shift;
4768 my $sshcmd = $self->sshcommand();
4769 my $serverlogin = $self->serverlogin();
4819 my $self = shift;
4821 if(@{$self->{'unget'}}) {
4822 my $job = shift @{$self->{'unget'}};
4825 my $commandline = $self->{'commandlinequeue'}->get();
4836 my $self = shift;
4837 unshift @{$self->{'unget'}}, @_;
4841 my $self = shift;
4842 my $empty = (not @{$self->{'unget'}})
4843 && $self->{'commandlinequeue'}->empty();
4849 my $self = shift;
4850 if(not defined $self->{'total_jobs'}) {
4854 while($job = $self->get()) {
4863 while($job = $self->get()) {
4867 $self->unget(@queue);
4868 $self->{'total_jobs'} = $#queue+1;
4870 return $self->{'total_jobs'};
4874 my $self = shift;
4876 return $self->{'commandlinequeue'}->seq();
4880 my $self = shift;
4881 return $self->{'commandlinequeue'}->quote_args();
4914 my $self = shift;
4915 $self->{'commandline'} or ::die_bug("commandline empty");
4916 return $self->{'commandline'}->replaced();
4920 my $self = shift;
4921 return $self->{'commandline'}->seq();
4925 my $self = shift;
4926 return $self->{'commandline'}->slot();
5001 my $self = shift;
5004 my $args_as_dirname = $self->{'commandline'}->args_as_dirname();
5014 $args_as_dirname = $self->{'commandline'}->args_as_dirname();
5031 $self->set_fh(1,"unlink","");
5032 $self->set_fh(2,"unlink","");
5042 $self->set_fh(1,"unlink","");
5043 $self->set_fh(2,"unlink",$errname);
5047 $self->set_fh(1,"unlink",$outname);
5048 $self->set_fh(2,"unlink",$errname);
5057 $self->set_fh(1,"unlink",$outname);
5058 $self->set_fh(2,"unlink",$errname);
5061 $self->set_fh(1,'w',$outfhw);
5062 $self->set_fh(2,'w',$errfhw);
5063 $self->set_fh(1,'name',$outname);
5064 $self->set_fh(2,'name',$errname);
5072 $self->fh($fdno,'name')) || die $?;
5073 $self->set_fh($fdno,'w',$fdw);
5074 $self->set_fh($fdno,'wpid',$wpid);
5077 $self->fh($fdno,'name'),$self->fh($fdno,'unlink')) || die $?;
5078 $self->set_fh($fdno,'r',$fdr);
5079 $self->set_fh($fdno,'rpid',$rpid);
5087 open(my $fdr,"<", $self->fh($fdno,'name')) ||
5088 ::die_bug("fdr: Cannot open ".$self->fh($fdno,'name'));
5089 $self->set_fh($fdno,'r',$fdr);
5091 $Global::debug or unlink $self->fh($fdno,"unlink");
5098 my $fdr = $self->fh($fdno,'r');
5142 my ($self, $fd_no, $key, $fh) = @_;
5143 $self->{'fd'}{$fd_no,$key} = $fh;
5148 my ($self, $fd_no, $key) = @_;
5149 return $self->{'fd'}{$fd_no,$key};
5153 my $self = shift;
5155 my $stdin_fh = $self->fh(0,"w");
5171 my $self = shift;
5173 $self->{'stdin_buffer'} = ($self->virgin() ? $$header_ref : "").substr($$block_ref,0,$endpos);
5175 remove_rec_sep(\$self->{'stdin_buffer'},$recstart,$recend);
5177 $self->{'stdin_buffer_length'} = length $self->{'stdin_buffer'};
5178 $self->{'stdin_buffer_pos'} = 0;
5182 my $self = shift;
5183 return $self->{'stdin_buffer_length'};
5195 my $self = shift;
5200 for my $buf (substr($self->{'stdin_buffer'},$self->{'stdin_buffer_pos'})) {
5201 my $in = $self->fh(0,"w");
5211 } elsif ($self->{'stdin_buffer_pos'}+$rv != $self->{'stdin_buffer_length'}) {
5214 $self->{'stdin_buffer_pos'} += $rv;
5219 $self->set_stdin_buffer(\$a,\$a,"","");
5230 my $self = shift;
5231 return $self->{'virgin'};
5235 my $self = shift;
5236 $self->{'virgin'} = shift;
5240 my $self = shift;
5241 return $self->{'pid'};
5245 my $self = shift;
5246 $self->{'pid'} = shift;
5252 my $self = shift;
5253 return sprintf("%.3f",$self->{'starttime'});
5257 my $self = shift;
5259 $self->{'starttime'} = $starttime;
5265 my $self = shift;
5266 return sprintf("%.3f",int(($self->endtime() - $self->starttime())*1000)/1000);
5273 my $self = shift;
5274 return ($self->{'endtime'} || 0);
5278 my $self = shift;
5280 $self->{'endtime'} = $endtime;
5289 my $self = shift;
5291 return time > $self->{'starttime'} + $delta_time;
5301 my $self = shift;
5303 my @family_pids = $self->family_pids();
5305 $self->set_exitstatus(-1);
5307 ::debug("run", "Kill seq ", $self->seq(), "\n");
5336 my $self = shift;
5337 my $pid = $self->pid();
5364 my $self = shift;
5366 return $self->{'failed'}{$sshlogin};
5373 my $self = shift;
5374 return $self->{'failed'}{$self->sshlogin()};
5379 my $self = shift;
5381 $self->{'failed'}{$sshlogin}++;
5386 my $self = shift;
5387 $self->{'failed'}{$self->sshlogin()}++;
5392 my $self = shift;
5394 delete $self->{'failed'}{$sshlogin};
5399 my $self = shift;
5400 delete $self->{'failed'}{$self->sshlogin()};
5407 my $self = shift;
5409 ::min(map { $self->{'failed'}{$_} } keys %{$self->{'failed'}});
5410 my $number_of_sshlogins_failed_on = scalar keys %{$self->{'failed'}};
5417 my $self = shift;
5419 for (values %{$self->{'failed'}}) {
5450 my $self = shift;
5451 if(not defined $self->{'wrapped'}) {
5452 my $command = $Global::envvar.$self->replaced();
5471 $self->{'commandline'}->replace_placeholders(["cat > \257<\257>; "], 0, 0).
5473 $self->{'commandline'}->replace_placeholders(
5479 $self->{'commandline'}->replace_placeholders(["mkfifo \257<\257>; ("], 0, 0).
5481 $self->{'commandline'}->replace_placeholders([") & _PID=\$!; cat > \257<\257>; ",
5487 $command = $self->sshlogin_wrap($command);
5509 $command = $self->tmux_wrap($command);
5511 $self->{'wrapped'} = $command;
5513 return $self->{'wrapped'};
5517 my $self = shift;
5519 $self->{'sshlogin'} = $sshlogin;
5520 delete $self->{'sshlogin_wrap'}; # If sshlogin is changed the wrap is wrong
5521 delete $self->{'wrapped'};
5525 my $self = shift;
5526 return $self->{'sshlogin'};
5533 my $self = shift;
5535 if(not defined $self->{'sshlogin_wrap'}) {
5536 my $sshlogin = $self->sshlogin();
5543 $self->{'sshlogin_wrap'} = $command;
5546 $pre .= $self->sshtransfer();
5548 $post .= $self->sshreturn();
5550 $post .= $self->sshcleanup();
5584 my $wd = ::shell_quote_file($self->workdir());
5613 $self->{'sshlogin_wrap'} =
5623 return $self->{'sshlogin_wrap'};
5630 my $self = shift;
5632 $self->{'transfersize'} = 0;
5634 for my $record (@{$self->{'commandline'}{'arg_list'}}) {
5640 $self->{'transfersize'} += (stat($arg->orig()))[7];
5649 my $self = shift;
5650 return $self->{'transfersize'};
5656 my $self = shift;
5658 my $sshlogin = $self->sshlogin();
5659 my $workdir = $self->workdir();
5660 for my $file ($self->transfer()) {
5671 my $self = shift;
5672 return $self->{'commandline'}->
5673 replace_placeholders($self->{'commandline'}{'return_files'},0,0);
5680 my $self = shift;
5681 for my $file ($self->return()) {
5683 $self->{'returnsize'} += (stat($file))[7];
5686 return $self->{'returnsize'};
5692 my $self = shift;
5693 my $sshlogin = $self->sshlogin();
5698 for my $file ($self->return()) {
5707 $wd = ::shell_quote_file($self->workdir()."/");
5733 my $self = shift;
5734 my $sshlogin = $self->sshlogin();
5737 my $workdir = $self->workdir();
5740 for my $file ($self->cleanup()) {
5753 my $self = shift;
5755 my @transfer = $self->transfer();
5756 my @return = $self->return();
5766 my $self = shift;
5767 if(not defined $self->{'workdir'}) {
5801 . "-" . $self->seq();
5812 $self->{'workdir'} = ::shell_quote_scalar($workdir);
5814 return $self->{'workdir'};
5942 my $self = shift;
5949 my $visual_command = $self->replaced();
6001 my $self = shift;
6005 if(not $self->exitstatus()) {
6007 $self->reset_failed_here();
6011 $self->add_failed_here();
6012 if($self->total_failed() == $opt::retries) {
6017 $self->set_endtime(undef);
6018 $Global::JobQueue->unget($self);
6019 ::debug("run", "Retry ", $self->seq(), "\n");
6029 my $self = shift;
6030 ::debug("print", ">>joboutput ", $self->replaced(), "\n");
6034 unlink $self->fh(1,"name");
6036 if($opt::pipe and $self->virgin()) {
6039 if($Global::joblog and defined $self->{'exitstatus'}) {
6041 $self->print_joblog();
6051 not $self->{'verbose_printed'}) {
6052 $self->{'verbose_printed'}++;
6054 print STDOUT $self->replaced(),"\n";
6057 print STDOUT $self->wrapped(),"\n";
6069 my $in_fh = $self->fh($fdno,"r");
6076 ::debug("print", "File descriptor $fdno (", $self->fh($fdno,"name"), "):");
6079 close $self->fh($fdno,"w");
6081 if($opt::pipe and $self->virgin()) {
6085 unlink $self->fh($fdno,"name");
6086 unlink $self->fh($fdno,"unlink");
6088 } elsif($fdno == 1 and $self->fh($fdno,"name")) {
6089 print $out_fd $self->fh($fdno,"name"),"\n";
6093 $self->linebuffer_print($fdno,$in_fh,$out_fd);
6096 close $self->fh($fdno,"w");
6100 my $tag = $self->tag();
6138 my $self = shift;
6140 my $partial = \$self->{'partial_line',$fdno};
6142 if(defined $self->{'exitstatus'}) {
6144 close $self->fh($fdno,"w");
6149 my $fdr = $self->fh($fdno,'r');
6172 if($fdno == 2 and not $self->{'printed_first_line',$fdno}++) {
6181 my $tag = $self->tag();
6192 if(defined $self->{'exitstatus'}) {
6196 my $tag = $self->tag();
6202 if($self->fh($fdno,"rpid") and CORE::kill 0, $self->fh($fdno,"rpid")) {
6212 my $self = shift;
6215 $cmd = $self->replaced();
6221 join("\t", $self->seq(), $self->sshlogin()->string(),
6222 $self->starttime(), sprintf("%10.3f",$self->runtime()),
6223 $self->transfersize(), $self->returnsize(),
6224 $self->exitstatus(), $self->exitsignal(), $cmd
6227 $self->set_job_in_joblog();
6231 my $self = shift;
6232 if(not defined $self->{'tag'}) {
6233 $self->{'tag'} = $self->{'commandline'}->
6236 return $self->{'tag'};
6240 my $self = shift;
6241 if(not defined $self->{'hostgroups'}) {
6242 $self->{'hostgroups'} = $self->{'commandline'}->{'arg_list'}[0][0]->{'hostgroups'};
6244 return @{$self->{'hostgroups'}};
6248 my $self = shift;
6249 return $self->{'exitstatus'};
6253 my $self = shift;
6257 $self->{'exitstatus'} = $exitstatus;
6261 $self->{'exitstatus'} ||= $exitstatus;
6266 my $self = shift;
6267 return $self->{'exitsignal'};
6271 my $self = shift;
6273 $self->{'exitsignal'} = $exitsignal;
6352 my $self = shift;
6353 return $self->{'seq'};
6365 my $self = shift;
6366 if(not $self->{'slot'}) {
6371 $self->{'slot'} = shift @Global::slots;
6373 return $self->{'slot'};
6390 my $self = shift;
6403 while (not $self->{'arg_queue'}->empty()) {
6404 $next_arg = $self->{'arg_queue'}->get();
6408 $self->push($next_arg);
6409 if($self->len() >= $max_len) {
6414 if($self->number_of_args() > 1) {
6416 $self->{'arg_queue'}->unget($self->pop());
6421 $self->len(), " >= ",
6424 $self->{'arg_queue'}->arg_number(),
6427 $self->{'arg_queue'}->unget($self->pop());
6432 if(defined $self->{'max_number_of_args'}) {
6433 if($self->number_of_args() >= $self->{'max_number_of_args'}) {
6439 and $self->{'arg_queue'}->empty() and $Global::max_jobs_running) {
6443 if($self->number_of_args() > 1) {
6444 $self->{'max_number_of_args'} =
6445 ::ceil($self->number_of_args()/$Global::max_jobs_running);
6447 $self->{'max_number_of_args'};
6448 $self->{'arg_queue'}->unget($self->pop_all());
6449 while($self->number_of_args() < $self->{'max_number_of_args'}) {
6450 $self->push($self->{'arg_queue'}->get());
6459 my $self = shift;
6461 push @{$self->{'arg_list'}}, $record;
6467 for my $perlexpr (keys %{$self->{'replacecount'}}) {
6469 $self->{'len'}{$perlexpr} += length $arg->replace($perlexpr,$quote_arg,$self);
6483 my $self = shift;
6484 my $record = pop @{$self->{'arg_list'}};
6488 for my $perlexpr (keys %{$self->{'replacecount'}}) {
6489 $self->{'len'}{$perlexpr} -=
6490 length $arg->replace($perlexpr,$quote_arg,$self);
6501 my $self = shift;
6502 my @popped = @{$self->{'arg_list'}};
6503 for my $replacement_string (keys %{$self->{'replacecount'}}) {
6504 $self->{'len'}{$replacement_string} = 0;
6506 $self->{'arg_list'} = [];
6514 my $self = shift;
6542 return $#{$self->{'arg_list'}}+1;
6549 my $self = shift;
6551 my $nrec = scalar @{$self->{'arg_list'}};
6553 $sum = $nrec * (scalar @{$self->{'arg_list'}[0]});
6561 my $self = shift;
6563 map { @$_ } @{$self->{'arg_list'}});
6571 my $self = shift;
6574 for my $rec_ref (@{$self->{'arg_list'}}) {
6626 my $self = shift;
6630 $len += $self->{'len'}{'noncontext'} + @{$self->{'command'}} -1;
6632 my $recargs = $self->number_of_recargs();
6633 if($self->{'context_replace'}) {
6635 $len += $recargs * $self->{'len'}{'context'};
6636 for my $replstring (keys %{$self->{'replacecount'}}) {
6638 $len += $self->{'len'}{$replstring} *
6639 $self->{'replacecount'}{$replstring};
6640 ::debug("length", $replstring, " ", $self->{'len'}{$replstring}, "*",
6641 $self->{'replacecount'}{$replstring}, "\n");
6646 ::debug("length", "Ctxgrp: ", $self->{'len'}{'contextgroups'},
6647 " Groups: ", $self->{'len'}{'noncontextgroups'}, "\n");
6649 $len += ($recargs-1) * ($self->{'len'}{'contextgroups'});
6653 $len += 1*$self->{'len'}{'context'};
6655 for my $replstring (keys %{$self->{'replacecount'}}) {
6658 $len += ($recargs -1 + $self->{'len'}{$replstring}) *
6659 $self->{'replacecount'}{$replstring};
6689 my $self = shift;
6690 if(not defined $self->{'replaced'}) {
6693 $self->{'replaced'} = $self->replace_placeholders($self->{'command'},$Global::quoting,$quote_arg);
6694 my $len = length $self->{'replaced'};
6695 if ($len != $self->len()) {
6696 ::debug("length", $len, " != ", $self->len(), " ", $self->{'replaced'}, "\n");
6698 ::debug("length", $len, " == ", $self->len(), " ", $self->{'replaced'}, "\n");
6701 return $self->{'replaced'};
6712 my $self = shift;
6716 my $context_replace = $self->{'context_replace'};
6759 for my $record (@{$self->{'arg_list'}}) {
6771 my $job = $self;
6788 $arg[$2 > 0 ? $2-1 : $n+$2]->replace($3,$quote_arg,$self)
6810 for my $perlexpr (keys %{$self->{'replacecount'}}) {
6814 {$arg ? $arg->replace($perlexpr,$quote_arg,$self) : ""}eg;
6978 my $self = shift;
6979 if(@{$self->{'unget'}}) {
6980 my $cmd_line = shift @{$self->{'unget'}};
6984 $cmd_line = CommandLine->new($self->seq(),
6985 $self->{'command'},
6986 $self->{'arg_queue'},
6987 $self->{'context_replace'},
6988 $self->{'max_number_of_args'},
6989 $self->{'return_files'},
6990 $self->{'replacecount'},
6991 $self->{'len'},
7008 return $self->get();
7011 $self->set_seq($self->seq()+1);
7017 my $self = shift;
7018 unshift @{$self->{'unget'}}, @_;
7022 my $self = shift;
7023 my $empty = (not @{$self->{'unget'}}) && $self->{'arg_queue'}->empty();
7029 my $self = shift;
7030 return $self->{'seq'};
7034 my $self = shift;
7035 $self->{'seq'} = shift;
7039 my $self = shift;
7041 return $self->{'command'};
7045 my $self = shift;
7046 if(not $self->{'size'}) {
7048 while(not $self->{'arg_queue'}->empty()) {
7049 push @all_lines, CommandLine->new($self->{'command'},
7050 $self->{'arg_queue'},
7051 $self->{'context_replace'},
7052 $self->{'max_number_of_args'});
7054 $self->{'size'} = @all_lines;
7055 $self->unget(@all_lines);
7057 return $self->{'size'};
7169 my $self = shift;
7170 if(@{$self->{'unget'}}) {
7171 $self->{'arg_number'}++;
7172 return shift @{$self->{'unget'}};
7174 my $ret = $self->{'arg_sub_queue'}->get();
7185 my $self = shift;
7187 $self->{'arg_number'} -= @_;
7188 unshift @{$self->{'unget'}}, @_;
7192 my $self = shift;
7193 my $empty = not @{$self->{'unget'}};
7194 $empty &&= $self->{'arg_sub_queue'}->empty();
7200 my $self = shift;
7201 return $self->{'arg_number'};
7221 my $self = shift;
7222 if(@{$self->{'unget'}}) {
7223 return shift @{$self->{'unget'}};
7225 my $unget_ref=$self->{'unget'};
7226 if($self->{'arg_sub_queue'}->empty()) {
7229 my $in_record = $self->{'arg_sub_queue'}->get();
7251 my $self = shift;
7253 unshift @{$self->{'unget'}}, @_;
7257 my $self = shift;
7258 my $empty = (not @{$self->{'unget'}} and $self->{'arg_sub_queue'}->empty());
7286 my $self = shift;
7288 return $self->xapply_get();
7290 return $self->nest_get();
7295 my $self = shift;
7297 unshift @{$self->{'unget'}}, @_;
7301 my $self = shift;
7303 and not @{$self->{'unget'}});
7304 for my $fh (@{$self->{'fhs'}}) {
7312 my $self = shift;
7313 if(@{$self->{'unget'}}) {
7314 return shift @{$self->{'unget'}};
7319 for my $fh (@{$self->{'fhs'}}) {
7323 push @{$self->{'arg_matrix'}{$fh}}, $arg;
7329 push @{$self->{'arg_matrix'}{$fh}}, shift @{$self->{'arg_matrix'}{$fh}};
7331 push @record, @{$self->{'arg_matrix'}{$fh}}[-1];
7342 my $self = shift;
7343 if(@{$self->{'unget'}}) {
7344 return shift @{$self->{'unget'}};
7349 my $no_of_inputsources = $#{$self->{'fhs'}} + 1;
7350 if(not $self->{'arg_matrix'}) {
7356 my $arg = read_arg_from_fh($self->{'fhs'}[$fhno]);
7360 $self->{'arg_matrix'}[$fhno][0] = $arg || Arg->new("");
7361 push @first_arg_set, $self->{'arg_matrix'}[$fhno][0];
7375 my $arg = read_arg_from_fh($self->{'fhs'}[0]);
7382 if(eof($self->{'fhs'}[$fhno])) {
7386 my $arg = read_arg_from_fh($self->{'fhs'}[$fhno]);
7388 my $len = $#{$self->{'arg_matrix'}[$fhno]} + 1;
7389 $self->{'arg_matrix'}[$fhno][$len] = $arg;
7393 push @combarg, [0, $#{$self->{'arg_matrix'}[$fhn]}];
7404 push @a, $self->{'arg_matrix'}[$n][$$c[$n]];
7409 push @{$self->{'unget'}}, @mapped;
7411 return shift @{$self->{'unget'}};
7415 return shift @{$self->{'unget'}};
7525 my $self = shift;
7532 if(not defined $self->{"rpl",0,$perlexpr}) {
7535 $_ = $self->{'orig'};
7537 $_ = trim_of($self->{'orig'});
7555 $self->{"rpl",0,$perlexpr} = $_;
7557 if(not defined $self->{"rpl",$quote,$perlexpr}) {
7558 $self->{"rpl",1,$perlexpr} =
7559 ::shell_quote_scalar($self->{"rpl",0,$perlexpr});
7561 return $self->{"rpl",$quote,$perlexpr};
7565 my $self = shift;
7566 return $self->{'orig'};
7617 my $self = shift;
7618 return $self->{'delta_time'};
7622 my $self = shift;
7623 $self->{'delta_time'} = shift;
7627 my $self = shift;
7628 return $self->{'remedian'};
7637 my $self = shift;
7639 my $i = $self->{'remedian_idx'}++;
7640 my $rref = $self->{'remedian_arr'};
7644 $self->{'remedian'} = (sort @{$rref->[2]})[$#{$rref->[2]}/2];
7650 my $self = shift;
7652 if($self->{'pct'}) {
7653 $self->set_remedian($runtime);
7654 $self->{'delta_time'} = $self->{'pct'} * $self->remedian();
7655 ::debug("run", "Timeout: $self->{'delta_time'}s ");
7661 my $self = shift;
7663 while (@{$self->{'queue'}}) {
7664 my $job = $self->{'queue'}[0];
7668 shift @{$self->{'queue'}};
7669 } elsif($job->timedout($self->{'delta_time'})) {
7672 shift @{$self->{'queue'}};
7682 my $self = shift;
7684 push @{$self->{'queue'}}, $in;
7726 my $self = shift;
7730 $self->atomic_link_if_count_less_than() and last;
7732 my $lockdir = $self->{'lockdir'};
7747 $self->atomic_link_if_count_less_than() and last;
7755 if(not -e $self->{'idfile'}) {
7756 open (my $fh, ">", $self->{'idfile'}) or
7757 ::die_bug("timeout_write_idfile: $self->{'idfile'}");
7760 link $self->{'idfile'}, $self->{'pidfile'};
7764 ::debug("sem", "acquired $self->{'pid'}\n");
7768 my $self = shift;
7769 unlink $self->{'pidfile'};
7770 if($self->nlinks() == 1) {
7772 $self->lock();
7773 if($self->nlinks() == 1) {
7774 unlink $self->{'idfile'};
7775 rmdir $self->{'lockdir'};
7777 $self->unlock();
7779 ::debug("run", "released $self->{'pid'}\n");
7783 my $self = shift;
7785 unlink $self->{'pidfile'};
7786 $self->lock();
7787 my $nlinks = $self->nlinks();
7788 ::debug("sem", $nlinks, "<", $self->{'count'});
7790 unlink $self->{'idfile'};
7791 open (my $fh, ">", $self->{'idfile'}) or
7792 ::die_bug("write_idfile: $self->{'idfile'}");
7796 unlink $self->{'idfile'};
7797 rmdir $self->{'lockdir'};
7799 $self->unlock();
7800 ::debug("sem", "released $self->{'pid'}\n");
7805 my $self = shift;
7807 $self->lock();
7808 ::debug($self->nlinks(), "<", $self->{'count'});
7809 if($self->nlinks() < $self->{'count'}) {
7810 -d $self->{'lockdir'} or mkdir_or_die($self->{'lockdir'});
7811 if(not -e $self->{'idfile'}) {
7812 open (my $fh, ">", $self->{'idfile'}) or
7813 ::die_bug("write_idfile: $self->{'idfile'}");
7816 $retval = link $self->{'idfile'}, $self->{'pidfile'};
7818 $self->unlock();
7825 my $self = shift;
7827 $self->lock();
7828 my $nlinks = $self->nlinks();
7829 ::debug("sem", $nlinks, "<", $self->{'count'});
7830 if($nlinks++ < $self->{'count'}) {
7831 -d $self->{'lockdir'} or mkdir_or_die($self->{'lockdir'});
7832 if(not -e $self->{'idfile'}) {
7833 open (my $fh, ">", $self->{'idfile'}) or
7834 ::die_bug("write_idfile: $self->{'idfile'}");
7837 open (my $fh, ">", $self->{'idfile'}) or
7838 ::die_bug("write_idfile: $self->{'idfile'}");
7841 $retval = link $self->{'idfile'}, $self->{'pidfile'};
7843 $self->unlock();
7849 my $self = shift;
7850 if(-e $self->{'idfile'}) {
7859 my $self = shift;
7865 if(tell($self->{'lockfh'}) == -1) {
7867 open($self->{'lockfh'}, ">", $self->{'lockfile'})
7868 or ::debug("run", "Cannot open $self->{'lockfile'}");
7870 if($self->{'lockfh'}) {
7872 chmod 0666, $self->{'lockfile'}; # assuming you want it a+rw
7873 if(flock($self->{'lockfh'}, LOCK_EX()|LOCK_NB())) {
7907 ::debug("run", "locked $self->{'lockfile'}");
7911 my $self = shift;
7912 unlink $self->{'lockfile'};
7913 close $self->{'lockfh'};