File Coverage

lib/Remote/Perl.pm

Criterion	Covered	Total	%
statement	202	205	98.5
branch	71	84	84.5
condition	43	55	78.1
subroutine	25	28	89.2
pod	5	6	83.3
total	346	378	91.5

use v5.36;
package Remote::Perl;

use autodie qw(open);
use List::Util qw(min);
use IO::Select;

use Remote::Perl::Bootstrap    qw(bootstrap_payload wait_for_ready);
use Remote::Perl::ModuleServer;
use Remote::Perl::Protocol   qw(
    PROTOCOL_VERSION
    MSG_HELLO MSG_READY MSG_RUN MSG_DATA MSG_EOF
    MSG_CREDIT MSG_MOD_REQ MSG_MOD_MISSING MSG_RETURN
    MSG_SIGNAL MSG_SIGNAL_ACK
    MSG_ERROR MSG_BYE
    STREAM_CONTROL STREAM_STDIN STREAM_STDOUT STREAM_STDERR
    TMPFILE_NONE TMPFILE_AUTO TMPFILE_LINUX TMPFILE_PERL TMPFILE_NAMED
    FLAGS_WARNINGS
    encode_message encode_hello encode_credit decode_return
    encode_run
);
use Remote::Perl::Transport;

our $VERSION = '0.004';

# -- Tmpfile strategy mapping --------------------------------------------------

my %_TMPFILE_FLAG = (
    off     => TMPFILE_NONE,
    1       => TMPFILE_AUTO,
    auto    => TMPFILE_AUTO,
    linux   => TMPFILE_LINUX,
    perl    => TMPFILE_PERL,
    named   => TMPFILE_NAMED,
);

sub _tmpfile_flag($v) {
    return TMPFILE_NONE unless $v;
    die "Unknown tmpfile strategy '$v'; valid values: auto, linux, perl, named\n"
        unless exists $_TMPFILE_FLAG{$v};
    return $_TMPFILE_FLAG{$v};
}

# -- Constructor ---------------------------------------------------------------

sub new($class, %args) {
    my $self = bless {
        cmd       => $args{cmd}      // die("'cmd' is required\n"),
        window    => $args{window}   // 65_536,
        serve     => $args{serve}    // 0,
        tmpfile   => $args{tmpfile}  // 0,
        data_warn => $args{data_warn} // 0,
        _mod_srv  => ($args{serve}
                        ? Remote::Perl::ModuleServer->new(
                              inc          => $args{inc} // \@INC,
                              serve_filter => $args{serve_filter},
                          )
                        : undef),
        # set by _connect:
        _t      => undef,   # Remote::Perl::Transport
        _parser => undef,   # Remote::Perl::Protocol::Parser
        _done   => 0,
        _ready  => 0,
        # reset before each run:
        _on_stdout      => undef,
        _on_stderr      => undef,
        _returned       => 0,
        _return_code    => undef,
        _return_msg     => undef,
        _stdin_credits  => 0,
        _stdin_eof_sent => 0,
        _no_stdin       => 0,
        _stdin_str      => undef,   # plain-string stdin buffer; undef = use fh or none
    }, $class;
    $self->_connect;
    return $self;
}

# -- Internal: connect and handshake -------------------------------------------

sub _connect($self) {
    my $t = Remote::Perl::Transport->new(cmd => $self->{cmd});
    $t->connect;
    $self->{_t}      = $t;
    $self->{_parser} = Remote::Perl::Protocol::Parser->new;

    $t->write_bytes(bootstrap_payload(serve => $self->{serve}));

    # Scan for the readiness marker; any leftover bytes belong to the protocol.
    my $leftover = wait_for_ready($t->out_fh);
    $self->{_parser}->feed($leftover) if length $leftover;

    $self->_send(MSG_HELLO, STREAM_CONTROL,
        encode_hello(PROTOCOL_VERSION, $self->{window}));

    $self->_pump_until(sub { $self->{_ready} });

    unless ($self->{_ready}) {
        my $err = '';
        while ($t->stderr_ready(0.25)) {
            my $chunk = $t->read_stderr(4096);
            last unless defined $chunk && length $chunk;
            $err .= $chunk;
        }
        $err =~ s/\s+\z//;
        my $detail = length($err) ? ": $err" : '';
        die "remperl: connection failed$detail\n";
    }
}

# -- Internal: wire I/O --------------------------------------------------------

sub _send($self, $type, $stream, $body = '') {
    $self->{_t}->write_bytes(encode_message($type, $stream, $body));
}

# -- Internal: event loop ------------------------------------------------------

# Drive the event loop until $pred->() returns true or the connection closes.
# $stdin_fh: optional filehandle whose data is forwarded to STREAM_STDIN.
#   Plain-string stdin is handled via _stdin_str / _drain_stdin_str; $stdin_fh
#   is always undef in that case (strings have no real fd for select).
sub _pump_until($self, $pred, $stdin_fh = undef) {
    my $t      = $self->{_t};
    my $out_fd = fileno($t->out_fh);
    my $in_fd  = defined($stdin_fh) ? (fileno($stdin_fh) // -1) : -1;

    until ($pred->() || $self->{_done}) {
        my $sel = IO::Select->new($t->out_fh);
        if ($in_fd >= 0 && !$self->{_stdin_eof_sent}
                        && $self->{_stdin_credits} > 0) {
            $sel->add($stdin_fh);
        }

        for my $fh ($sel->can_read(1)) {
            if (fileno($fh) == $out_fd) {
                my $data = $t->read_bytes(65_536);
                unless (defined $data) { $self->{_done} = 1; last }
                $self->_dispatch($_) for $self->{_parser}->feed($data);
            }
            elsif (fileno($fh) == $in_fd) {
                $self->_forward_stdin($stdin_fh);
            }
        }
    }
}

sub _dispatch($self, $msg) {
    my ($type, $stream, $body) = @{$msg}{qw(type stream body)};

    if    ($type == MSG_READY) { $self->{_ready} = 1 }

    elsif ($type == MSG_DATA) {
        if ($stream == STREAM_STDOUT) {
            ($self->{_on_stdout} // sub {})->($body);
            $self->_send(MSG_CREDIT, STREAM_STDOUT, encode_credit(length($body)));
        }
        elsif ($stream == STREAM_STDERR) {
            ($self->{_on_stderr} // sub {})->($body);
            $self->_send(MSG_CREDIT, STREAM_STDERR, encode_credit(length($body)));
        }
    }

    elsif ($type == MSG_EOF) { }   # EOF on stdout/stderr; RETURN signals end-of-run

    elsif ($type == MSG_CREDIT) {
        if ($stream == STREAM_STDIN) {
            if ($self->{_no_stdin} && !$self->{_stdin_eof_sent}) {
                $self->_send(MSG_EOF, STREAM_STDIN);
                $self->{_stdin_eof_sent} = 1;
            }
            else {
                $self->{_stdin_credits} += unpack('N', $body);
                $self->_drain_stdin_str if defined $self->{_stdin_str};
            }
        }
    }

    elsif ($type == MSG_MOD_REQ) {
        die "remperl: serve => 1 but no ModuleServer configured\n"
            if $self->{serve} && !$self->{_mod_srv};
        my $source = ($self->{serve} && $self->{_mod_srv}) ? $self->{_mod_srv}->find($body) : undef;
        if (defined $source) {
            my $chunk = 65_536;
            for (my $off = 0; $off < length($source); $off += $chunk) {
                $self->_send(MSG_DATA, $stream, substr($source, $off, $chunk));
            }
            $self->_send(MSG_EOF, $stream);
        }
        else {
            $self->_send(MSG_MOD_MISSING, $stream);
        }
    }

    elsif ($type == MSG_SIGNAL_ACK) { }   # reserved for future unresponsive-remote detection

    elsif ($type == MSG_RETURN) {
        ($self->{_return_code}, $self->{_return_msg}) = decode_return($body);
        $self->{_returned} = 1;
    }

    elsif ($type == MSG_ERROR) { die "remperl: remote error: $body\n" }

    elsif ($type == MSG_BYE)   { $self->{_done} = 1 }
}

sub _drain_stdin_str($self) {
    while ($self->{_stdin_credits} > 0 && length($self->{_stdin_str})) {
        my $take = min(65_536, $self->{_stdin_credits}, length($self->{_stdin_str}));
        $self->_send(MSG_DATA, STREAM_STDIN, substr($self->{_stdin_str}, 0, $take, ''));
        $self->{_stdin_credits} -= $take;
    }
    if (!length($self->{_stdin_str}) && !$self->{_stdin_eof_sent}) {
        $self->_send(MSG_EOF, STREAM_STDIN);
        $self->{_stdin_eof_sent} = 1;
    }
}

sub _forward_stdin($self, $stdin_fh) {
    my $n = min(65_536, $self->{_stdin_credits});
    my $data;
    my $bytes = sysread($stdin_fh, $data, $n);
    if (!defined $bytes) {
        die "sysread stdin: $!\n";
    }
    elsif ($bytes == 0) {
        $self->_send(MSG_EOF, STREAM_STDIN);
        $self->{_stdin_eof_sent} = 1;
    }
    else {
        $self->_send(MSG_DATA, STREAM_STDIN, $data);
        $self->{_stdin_credits} -= $bytes;
    }
}

# -- Public API ----------------------------------------------------------------

# Run Perl source code on the remote.  Returns the exit code (integer).
# In list context also returns the error message (empty string on success).
#
# Options:
#   on_stdout => sub($chunk) { ... }   called for each stdout DATA chunk
#   on_stderr => sub($chunk) { ... }   called for each stderr DATA chunk
#   stdin     => $fh_or_str            filehandle or plain string to use as
#                                      remote STDIN (omit or pass undef for none)
#   args      => \@args                (optional) set remote @ARGV before running
sub run_code($self, $source, %opts) {
    $self->{_on_stdout}      = $opts{on_stdout};
    $self->{_on_stderr}      = $opts{on_stderr};
    $self->{_returned}       = 0;
    $self->{_return_code}    = undef;
    $self->{_return_msg}     = '';
    $self->{_stdin_credits}  = 0;
    $self->{_stdin_eof_sent} = 0;

    my $tmpfile  = exists $opts{tmpfile} ? $opts{tmpfile} : $self->{tmpfile};
    my $warnings = $opts{warnings} // 0;
    my $flags    = _tmpfile_flag($tmpfile) | ($warnings ? FLAGS_WARNINGS : 0);

    if (!$flags && $self->{data_warn} && $source =~ /^__DATA__(?:\r?\n|$)/m) {
        warn "remperl: script contains __DATA__ but --tmpfile is not set;"
           . " use --tmpfile for __DATA__ support\n";
    }

    my $stdin = $opts{stdin};
    my $stdin_fh;
    if (!defined $stdin) {
        $self->{_no_stdin}  = 1;
        $self->{_stdin_str} = undef;
    }
    elsif (!ref($stdin)) {
        # Plain string: buffer it; credits drive delivery in _dispatch.
        $self->{_no_stdin}  = 0;
        $self->{_stdin_str} = $stdin;
    }
    else {
        $self->{_no_stdin}  = 0;
        $self->{_stdin_str} = undef;
        $stdin_fh           = $stdin;
    }

    my @argv = $opts{args} ? @{$opts{args}} : ();
    $self->_send(MSG_RUN, STREAM_CONTROL, encode_run($flags, $source, @argv));
    $self->_pump_until(sub { $self->{_returned} }, $stdin_fh);

    return wantarray
        ? ($self->{_return_code}, $self->{_return_msg})
        : $self->{_return_code};
}

# Read a local file and run its contents as Perl source on the remote.
# Accepts the same options as run_code, including tmpfile.
sub run_file($self, $path, %opts) {
    open(my $fh, '<', $path);
    local $/;
    my $source = <$fh>;
    return $self->run_code($source, %opts);
}

# Send BYE and wait for the remote to echo it back, then close the transport.
sub disconnect($self) {
    return if $self->{_done};
    eval { $self->_send(MSG_BYE, STREAM_CONTROL) };
    $self->_pump_until(sub { $self->{_done} }) unless $self->{_done};
    $self->{_t}->disconnect;
}

sub pid($self) { $self->{_t}->pid }

# Send a signal to the remote script by name (e.g. 'INT', 'TERM').
# The remote relay process delivers it to the executor and replies with
# MSG_SIGNAL_ACK, which can later be used to detect unresponsive remotes.
sub send_signal($self, $signame) {
    $self->_send(MSG_SIGNAL, STREAM_CONTROL, $signame);
}

sub DESTROY($self) {
    eval { $self->disconnect } unless $self->{_done};
}

1;

__END__