From: Dirk Koopman Date: Thu, 16 Apr 2020 20:41:40 +0000 (+0100) Subject: new version of subprocessing X-Git-Url: http://www.dxcluster.org/gitweb/gitweb.cgi?a=commitdiff_plain;ds=sidebyside;h=5fc4f15ca36b6926fac4a8b33b3944cc1f7cef49;p=spider.git new version of subprocessing --- diff --git a/cmd/crontab b/cmd/crontab index ebc0d24c..ac62bd30 100644 --- a/cmd/crontab +++ b/cmd/crontab @@ -5,6 +5,6 @@ # create and edit the one in /spider/local_cmd/crontab # for doing connections and things # -1 0 * * 3 DXUser::export() -5 0 * * * DXDebug::dbgclean() +2 0 * * * DXDebug::dbgclean() +2 30 * * 3 DXUser::export() 0 3 * * * Spot::daily() diff --git a/perl/DXCommandmode.pm b/perl/DXCommandmode.pm index 4212d968..1b08b36e 100644 --- a/perl/DXCommandmode.pm +++ b/perl/DXCommandmode.pm @@ -43,7 +43,7 @@ use JSON; use Time::HiRes qw(gettimeofday tv_interval); use Mojo::IOLoop; -use Mojo::IOLoop::ForkCall; +use Mojo::IOLoop::Subprocess; use Mojo::UserAgent; use strict; @@ -490,7 +490,7 @@ sub send_ans } # -# this is the thing that runs the command, it is done like this for the +# this is the thing that preps for running the command, it is done like this for the # benefit of remote command execution # @@ -1310,18 +1310,26 @@ sub spawn_cmd no strict 'refs'; - my $fc = Mojo::IOLoop::ForkCall->new; - $fc->serializer(\&encode_json); - $fc->deserializer(\&decode_json); + my $fc = Mojo::IOLoop::Subprocess->new; +# $fc->serializer(\&encode_json); +# $fc->deserializer(\&decode_json); $fc->run( - sub {my @args = @_; my @res = $cmdref->(@args); return @res}, - $args, + sub { + my $subpro = shift; + if (isdbg('chan')) { + my $s = "line: $line"; + $s .= ", args: " . join(', ', @$args) if $args && @$args; + } + my @res = $cmdref->(@$args); + return @res; + }, +# $args, sub { my ($fc, $err, @res) = @_; my $dxchan = DXChannel::get($call); return unless $dxchan; - if (defined $err) { + if ($err) { my $s = "DXCommand::spawn_cmd: call $call error $err"; dbg($s) if isdbg('chan'); $dxchan->send($s); diff --git a/perl/DXCron.pm b/perl/DXCron.pm index 757ec61c..e60d36b4 100644 --- a/perl/DXCron.pm +++ b/perl/DXCron.pm @@ -15,6 +15,8 @@ use DXDebug; use IO::File; use DXLog; +use Mojo::IOLoop::Subprocess; + use strict; use vars qw{@crontab @lcrontab @scrontab $mtime $lasttime $lastmin}; @@ -244,13 +246,13 @@ sub spawn { my $line = shift; - my $fc = Mojo::IOLoop::ForkCall->new; + my $fc = Mojo::IOLoop::Subprocess->new(); $fc->run( sub {my @res = `$line`; return @res}, - [], +# [], sub { my ($fc, $err, @res) = @_; - if (defined $err) { + if ($err) { my $s = "DXCron::spawn: error $err"; dbg($s); return; @@ -268,13 +270,13 @@ sub spawn_cmd my $line = shift; dbg("spawn_cmd run: $line") if isdbg('cron'); - my $fc = Mojo::IOLoop::ForkCall->new; + my $fc = Mojo::IOLoop::Subprocess->new(); $fc->run( sub {my @res = DXCommandmode::run_cmd($main::me, $line); return @res}, - [], +# [], sub { my ($fc, $err, @res) = @_; - if (defined $err) { + if ($err) { my $s = "spawn_cmd: error $err"; dbg($s); } @@ -310,5 +312,6 @@ sub run_cmd dbg("cmd out: $_") if isdbg('cron'); } } + 1; __END__ diff --git a/perl/DXProt.pm b/perl/DXProt.pm index 68dd099e..7c1ed1a4 100644 --- a/perl/DXProt.pm +++ b/perl/DXProt.pm @@ -34,6 +34,9 @@ use Route::Node; use Script; use DXProtHandle; +use Time::HiRes qw(gettimeofday tv_interval); +use Mojo::IOLoop::Subprocess; + use strict; use vars qw($pc11_max_age $pc23_max_age $last_pc50 $eph_restime $eph_info_restime $eph_pc34_restime @@ -1133,8 +1136,14 @@ sub process_rcmd if ($ref->{priv}) { # you have to have SOME privilege, the commands have further filtering $self->{remotecmd} = 1; # for the benefit of any command that needs to know my $oldpriv = $self->{priv}; - $self->{priv} = $ref->{priv}; # assume the user's privilege level + $self->{priv} = 1; # set a maximum privilege + + # park homenode and user for any spawned command that run_cmd may do. + $self->{_rcmd_user} = $user; + $self->{_rcmd_fromnode} = $fromnode; my @in = (DXCommandmode::run_cmd($self, $cmd)); + delete $self->{_rcmd_fromnode}; + delete $self->{_rcmd_user}; $self->{priv} = $oldpriv; $self->send_rcmd_reply($main::mycall, $fromnode, $user, @in); delete $self->{remotecmd}; @@ -1154,6 +1163,105 @@ sub process_rcmd } } + +sub send_rcmd_reply +{ + my $self = shift; + my $tonode = shift; + my $fromnode = shift; + my $user = shift; + while (@_) { + my $line = shift; + $line =~ s/\s*$//; + Log('rcmd', 'out', $fromnode, $line); + if ($self->is_clx) { + $self->send(pc85($main::mycall, $fromnode, $user, "$main::mycall:$line")); + } else { + $self->send(pc35($main::mycall, $fromnode, "$main::mycall:$line")); + } + } +} + +# Punt off a long running command into a separate process - this will be caused by an rcmd from outside +# +# This is called from commands to run some potentially long running +# function. The process forks and then runs the function and returns +# the result back to the cmd. +# +# NOTE: this merely forks the current process and then runs the cmd in that (current) context. +# IT DOES NOT START UP SOME NEW PROGRAM AND RELIES ON THE FACT THAT IT IS RUNNING DXSPIDER +# THE CURRENT CONTEXT!! +# +# call: $self->spawn_cmd($original_cmd_line, \, [cb => sub{...}], [prefix => "cmd> "], [progress => 0|1], [args => [...]]); +sub spawn_cmd +{ + my $self = shift; + my $line = shift; + my $cmdref = shift; + my $call = $self->{call}; + my %args = @_; + my @out; + + my $cb = delete $args{cb}; + my $prefix = delete $args{prefix}; + my $progress = delete $args{progress}; + my $args = delete $args{args} || []; + my $t0 = [gettimeofday]; + + # remembered from process_cmd when spawn_cmd was called thru DXCommandmode::run_cmd which was called by process_rcmd + my $fromnode = $self->{_rcmd_fromnode}; + my $user = $self->{_rcmd_user}; + + no strict 'refs'; + + my $fc = Mojo::IOLoop::Subprocess->new; + + # $fc->serializer(\&encode_json); +# $fc->deserializer(\&decode_json); + $fc->run( + sub { + my $subpro = shift; + if (isdbg('chan')) { + my $s = "line: $line"; + $s .= ", args: " . join(', ', @$args) if $args && @$args; + } + + my @res = $cmdref->(@$args); + return @res; + }, +# $args, + sub { + my ($fc, $err, @res) = @_; + my $self = DXChannel::get($call); + return unless $self; + + if ($err) { + my $s = "DXCommand::spawn_cmd: call $call error $err"; + dbg($s) if isdbg('chan'); + if ($fromnode && $user) { + $self->send_rcmd_reply($main::mycall, $fromnode, $user, $s); + } else { + $self->send($s); + } + return; + } + if ($cb) { + # transform output if required + @res = $cb->($self, @res); + } + if (@res) { + if ($fromnode && $user) { + $self->send_rcmd_reply($main::mycall, $fromnode, $user, @res); + } else { + $self->send(@res); + } + } + DXCommandmode::_diffms($call, $line, $t0); + }); + + return @out; +} + sub process_rcmd_reply { my ($self, $tonode, $fromnode, $user, $line) = @_; @@ -1179,23 +1287,7 @@ sub process_rcmd_reply } } -sub send_rcmd_reply -{ - my $self = shift; - my $tonode = shift; - my $fromnode = shift; - my $user = shift; - while (@_) { - my $line = shift; - $line =~ s/\s*$//; - Log('rcmd', 'out', $fromnode, $line); - if ($self->is_clx) { - $self->send(pc85($main::mycall, $fromnode, $user, "$main::mycall:$line")); - } else { - $self->send(pc35($main::mycall, $fromnode, "$main::mycall:$line")); - } - } -} + # add a rcmd request to the rcmd queues sub addrcmd @@ -1690,5 +1782,8 @@ sub clean_pc92_find { } + + + 1; __END__