mirror of
git://scm.dxcluster.org/scm/spider
synced 2024-09-21 07:47:10 +00:00
new version of subprocessing
This commit is contained in:
parent
a5f343afa1
commit
5fc4f15ca3
@ -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()
|
||||
|
@ -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);
|
||||
|
@ -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__
|
||||
|
131
perl/DXProt.pm
131
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, \<function>, [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__
|
||||
|
Loading…
Reference in New Issue
Block a user