AnyEvent-Fork-Remote/Remote.pm

=head1 NAME

AnyEvent::Fork::Remote - remote processes with AnyEvent::Fork interface

THE API IS NOT FINISHED, CONSIDER THIS A BETA RELEASE

=head1 SYNOPSIS

   use AnyEvent;
   use AnyEvent::Fork::Remote;

   my $rpc = AnyEvent::Fork::Remote
      ->new
      ->require ("MyModule")
      ->run ("MyModule::run", my $cv = AE::cv);

   my $fh = $cv->recv;

=head1 DESCRIPTION

Despite what the name of this module might suggest, it doesn't actualyl
create remote processes for you. But it does make it easy to use them,
once you have started them.

This module implements a very similar API as L<AnyEvent::Fork>. In fact,
similar enough to require at most minor modifications to support both
at the same time. For example, it works with L<AnyEvent::Fork::RPC> and
L<AnyEvent::Fork::Pool>.

The documentation for this module will therefore only document the parts
of the API that differ between the two modules.

=head2 SUMMARY OF DIFFERENCES

Here is a short summary of the main differences between L<AnyEvent::Fork>
and this module:

=over 4

=item * C<send_fh> is not implemented and will fail

=item * the child-side C<run> function must read from STDIN and write to STDOUT

=item * C<fork> does not actually fork, but will create a new process

=back

=head1 EXAMPLES

=head1 PARENT PROCESS USAGE

=over 4

=cut

package AnyEvent::Fork::Remote;

use common::sense;

use Errno ();

use AnyEvent ();
use AnyEvent::Util ();

our $VERSION = 0.1;

# xored together must start and and with \n
my $magic0 = "Pdk{6y[_zZ";
my $magic1 = "Z^yZ7~i=oP";

=item my $proc = new_exec AnyEvent::Fork::Remote $path, @args...

Creates a new C<AnyEvent::Fork::Remote> object. Unlike L<AnyEvent::Fork>,
processes are only created when C<run> is called, every other method call
is is simply recorded until then.

Each time a new process is needed, it executes C<$path> with the given
arguments (the first array member must be the program name, as with
the C<exec> function with explicit PROGRAM argument) and both C<STDIN>
and C<STDOUT> connected to a communications socket. No input must be
consumed by the comamnd before F<perl> is started, and no output should be
generated.

The program I<must> invoke F<perl> somehow, with STDIN and STDOUT intact,
without specifying anything to execute (no script file name, no C<-e>
switch etc.).

Here are some examples to give you an idea:

   # just "perl"
   $proc = new_exec AnyEvent::Fork::Remote
      "/usr/bin/perl", "perl";

   # rsh othernode exec perl
   $proc = new_exec AnyEvent::Fork::Remote
      "/usr/bin/rsh", "rsh", "othernode", "exec perl";

   # a complicated ssh command
   $proc = new_exec AnyEvent::Fork::Remote
      "/usr/bin/ssh",
      qw(ssh -q
         -oCheckHostIP=no -oTCPKeepAlive=yes -oStrictHostKeyChecking=no
         -oGlobalKnownHostsFile=/dev/null -oUserKnownHostsFile=/dev/null
         otherhost
         exec perl);

=item my $proc = new AnyEvent::Fork::Remote $create_callback

Basically the same as C<new_exec>, but instead of a hardcoded command
path, it expects a callback which is invoked each time a process needs to
be created.

The C<$create_callback> is called with another callback as argument,
and should call this callback with the file handle that is connected
to a F<perl> process. This callback can be invoked even after the
C<$create_callback> returns.

Example: emulate C<new_exec> using C<new>.

   use AnyEvent::Util;
   use Proc::FastSpawn;

   $proc = new AnyEvent::Fork::Remote sub {
      my $done = shift;

      my ($a, $b) = AnyEvent::Util::portable_socketpair
         or die;

      open my $oldin , "<&0" or die;
      open my $oldout, ">&1" or die;

      open STDIN , "<&" . fileno $b or die;
      open STDOUT, ">&" . fileno $b or die;

      spawn "/usr/bin/rsh", ["rsh", "othernode", "perl"];

      open STDIN , "<&" . fileno $oldin ;
      open STDOUT, ">&" . fileno $oldout;

      $done->($a);
   };

=cut

sub new_exec {
   my ($class, $program, @argv) = @_;

   require AnyEvent::Util;
   require Proc::FastSpawn;

   $class->new (sub {
      my $done = shift;

      my ($a, $b) = AnyEvent::Util::portable_socketpair ()
         or die;

      open my $oldin , "<&0" or die;
      open my $oldout, ">&1" or die;

      open STDIN , "<&" . fileno $b or die;
      open STDOUT, ">&" . fileno $b or die;

      Proc::FastSpawn::spawn ($program, \@argv);

      open STDIN , "<&" . fileno $oldin ;
      open STDOUT, ">&" . fileno $oldout;

      $done->($a);
   })
}

sub new {
   my ($class, $create) = @_;

   bless [
      $create,
      "",
      [],
   ], $class
}

=item $new_proc = $proc->fork

Quite the same as the same method of L<AnyEvent::Fork>, except that it
simply clones the object without creating an actual process.

=cut

sub fork {
   my $self = shift;

   bless [
      $self->[0],
      $self->[1],
      [@{ $self->[2] }],
   ], ref $self
}

=item undef = $proc->pid

The C<pid> method always returns C<undef> and only exists for
compatibility with L<AnyEvent::Fork>.

=cut

sub pid {
   undef
}

=item $proc = $proc->send_fh (...)

Not supported and always croaks.

=cut

sub send_fh {
   require Carp;
   Carp::croak ("send_fh is not supported on AnyEvent::Fork::Remote objects");
}

=item $proc = $proc->eval ($perlcode, @args)

Quite the same as the same method of L<AnyEvent::Fork>.

=cut

# quote a binary string as a perl scalar
sub sq($) {
   my $s = shift;

   $s =~ /'/
      or return "'$s'";

   $s =~ s/(\x10+)/\x10.'$1'.q\x10/g;
   "q\x10$s\x10"
}

# quote a list of strings
sub aq(@) {
   "(" . (join ",", map sq $_, @_) . ")"
}

sub eval {
   my ($self, $perlcode, @args) = @_;

   $self->[1] .= '{ local @_ = ' . (aq @args) . "; $perlcode }\n";
}

=item $proc = $proc->require ($module, ...)

Quite the same as the same method of L<AnyEvent::Fork>.

=cut

sub require {
   my ($self, @modules) = @_;

   s%::%/%g for @modules;
   $self->eval ('require "$_.pm" for @_', @modules);

   $self
}

=item $proc = $proc->send_arg ($string, ...)

Quite the same as the same method of L<AnyEvent::Fork>.

=cut

sub send_arg {
   my ($self, @arg) = @_;

   push @{ $self->[2] }, @arg;

   $self
}

=item $proc->run ($func, $cb->($fh))

Very similar to the run method of L<AnyEvent::Fork>.

On the parent side, the API is identical. On the child side, the
"communications socket" is in fact just C<*STDIN>, and typically can only
be read from.

If the run function wants to read data that is written to C<$fh> in the
parent, then it should read from STDIN. If the run function wants to
provide data that can later be read from C<$fh>, then it should write them
to STDOUT.

You can write a run function that works with both L<AnyEvent::Fork> and
this module by checking C<fileno $fh> in on the passed callback in the run
function:

   sub run {
      my ($rfh, ...) = @_;
      my $wfh = fileno $rfh ? $rfh : *STDOUT;

      # now use $rfh for reading and $wfh for writing
   }

=cut

sub run {
   my ($self, $func, $cb) = @_;

   my $code = 'BEGIN {' . $self->[1] . '}'
            . 'syswrite STDOUT, ' . (sq $magic0) . '^' . (sq $magic1) . ';'
            . $func . (aq @{ $self->[2] }) . ';'
            . "\n__END__\n";

   $self->[0](sub {
      my $fh = shift
         or die "AnyEvent::Fork::Remote: create callback failed";


   });
}

my $x = new_exec AnyEvent::Fork::Remote "/usr/bin/rsh", "rsh", "rain", "exec perl";#d#
$x->require ("Carp", "Storable");#d#
$x->send_arg (1, 2, 3);#d#
$x->eval ('sub run { die }');#d#
$x->run ("run", sub {
   });


=item my $proc = new_from_stdio $fh

Creates an C<AnyEvent::Fork::Remote> object from a file handle. This file
handle must be connected to both STDIN and STDOUT of a F<perl> process.

This form might be more convenient than C<new> or C<new_exec> when
creating an C<AnyEvent::Fork::Remote> object, but the resulting object
does not support C<fork>.

#TODO: really implement?

=back

=head1 SEE ALSO

L<AnyEvent::Fork>, the same as this module, for local processes.

L<AnyEvent::Fork::RPC>, to talk to the created processes.

L<AnyEvent::Fork::Pool>, to manage whole pools of processes.

=head1 AUTHOR AND CONTACT INFORMATION

 Marc Lehmann <schmorp@schmorp.de>
 http://software.schmorp.de/pkg/AnyEvent-Fork-Remote

=cut

1

Revision:	1.1
Committed:	Sat Apr 27 01:44:55 2013 UTC (11 years, 5 months ago) by root
Branch:	MAIN
Log Message:	* empty log message *
#	User	Rev	Content
1	root	1.1	=head1 NAME
2
3			AnyEvent::Fork::Remote - remote processes with AnyEvent::Fork interface
4
5			THE API IS NOT FINISHED, CONSIDER THIS A BETA RELEASE
6
7			=head1 SYNOPSIS
8
9			use AnyEvent;
10			use AnyEvent::Fork::Remote;
11
12			my $rpc = AnyEvent::Fork::Remote
13			->new
14			->require ("MyModule")
15			->run ("MyModule::run", my $cv = AE::cv);
16
17			my $fh = $cv->recv;
18
19			=head1 DESCRIPTION
20
21			Despite what the name of this module might suggest, it doesn't actualyl
22			create remote processes for you. But it does make it easy to use them,
23			once you have started them.
24
25			This module implements a very similar API as L<AnyEvent::Fork>. In fact,
26			similar enough to require at most minor modifications to support both
27			at the same time. For example, it works with L<AnyEvent::Fork::RPC> and
28			L<AnyEvent::Fork::Pool>.
29
30			The documentation for this module will therefore only document the parts
31			of the API that differ between the two modules.
32
33			=head2 SUMMARY OF DIFFERENCES
34
35			Here is a short summary of the main differences between L<AnyEvent::Fork>
36			and this module:
37
38			=over 4
39
40			=item * C<send_fh> is not implemented and will fail
41
42			=item * the child-side C<run> function must read from STDIN and write to STDOUT
43
44			=item * C<fork> does not actually fork, but will create a new process
45
46			=back
47
48			=head1 EXAMPLES
49
50			=head1 PARENT PROCESS USAGE
51
52			=over 4
53
54			=cut
55
56			package AnyEvent::Fork::Remote;
57
58			use common::sense;
59
60			use Errno ();
61
62			use AnyEvent ();
63			use AnyEvent::Util ();
64
65			our $VERSION = 0.1;
66
67			# xored together must start and and with \n
68			my $magic0 = "Pdk{6y[_zZ";
69			my $magic1 = "Z^yZ7~i=oP";
70
71			=item my $proc = new_exec AnyEvent::Fork::Remote $path, @args...
72
73			Creates a new C<AnyEvent::Fork::Remote> object. Unlike L<AnyEvent::Fork>,
74			processes are only created when C<run> is called, every other method call
75			is is simply recorded until then.
76
77			Each time a new process is needed, it executes C<$path> with the given
78			arguments (the first array member must be the program name, as with
79			the C<exec> function with explicit PROGRAM argument) and both C<STDIN>
80			and C<STDOUT> connected to a communications socket. No input must be
81			consumed by the comamnd before F<perl> is started, and no output should be
82			generated.
83
84			The program I<must> invoke F<perl> somehow, with STDIN and STDOUT intact,
85			without specifying anything to execute (no script file name, no C<-e>
86			switch etc.).
87
88			Here are some examples to give you an idea:
89
90			# just "perl"
91			$proc = new_exec AnyEvent::Fork::Remote
92			"/usr/bin/perl", "perl";
93
94			# rsh othernode exec perl
95			$proc = new_exec AnyEvent::Fork::Remote
96			"/usr/bin/rsh", "rsh", "othernode", "exec perl";
97
98			# a complicated ssh command
99			$proc = new_exec AnyEvent::Fork::Remote
100			"/usr/bin/ssh",
101			qw(ssh -q
102			-oCheckHostIP=no -oTCPKeepAlive=yes -oStrictHostKeyChecking=no
103			-oGlobalKnownHostsFile=/dev/null -oUserKnownHostsFile=/dev/null
104			otherhost
105			exec perl);
106
107			=item my $proc = new AnyEvent::Fork::Remote $create_callback
108
109			Basically the same as C<new_exec>, but instead of a hardcoded command
110			path, it expects a callback which is invoked each time a process needs to
111			be created.
112
113			The C<$create_callback> is called with another callback as argument,
114			and should call this callback with the file handle that is connected
115			to a F<perl> process. This callback can be invoked even after the
116			C<$create_callback> returns.
117
118			Example: emulate C<new_exec> using C<new>.
119
120			use AnyEvent::Util;
121			use Proc::FastSpawn;
122
123			$proc = new AnyEvent::Fork::Remote sub {
124			my $done = shift;
125
126			my ($a, $b) = AnyEvent::Util::portable_socketpair
127			or die;
128
129			open my $oldin , "<&0" or die;
130			open my $oldout, ">&1" or die;
131
132			open STDIN , "<&" . fileno $b or die;
133			open STDOUT, ">&" . fileno $b or die;
134
135			spawn "/usr/bin/rsh", ["rsh", "othernode", "perl"];
136
137			open STDIN , "<&" . fileno $oldin ;
138			open STDOUT, ">&" . fileno $oldout;
139
140			$done->($a);
141			};
142
143			=cut
144
145			sub new_exec {
146			my ($class, $program, @argv) = @_;
147
148			require AnyEvent::Util;
149			require Proc::FastSpawn;
150
151			$class->new (sub {
152			my $done = shift;
153
154			my ($a, $b) = AnyEvent::Util::portable_socketpair ()
155			or die;
156
157			open my $oldin , "<&0" or die;
158			open my $oldout, ">&1" or die;
159
160			open STDIN , "<&" . fileno $b or die;
161			open STDOUT, ">&" . fileno $b or die;
162
163			Proc::FastSpawn::spawn ($program, \@argv);
164
165			open STDIN , "<&" . fileno $oldin ;
166			open STDOUT, ">&" . fileno $oldout;
167
168			$done->($a);
169			})
170			}
171
172			sub new {
173			my ($class, $create) = @_;
174
175			bless [
176			$create,
177			"",
178			[],
179			], $class
180			}
181
182			=item $new_proc = $proc->fork
183
184			Quite the same as the same method of L<AnyEvent::Fork>, except that it
185			simply clones the object without creating an actual process.
186
187			=cut
188
189			sub fork {
190			my $self = shift;
191
192			bless [
193			$self->[0],
194			$self->[1],
195			[@{ $self->[2] }],
196			], ref $self
197			}
198
199			=item undef = $proc->pid
200
201			The C<pid> method always returns C<undef> and only exists for
202			compatibility with L<AnyEvent::Fork>.
203
204			=cut
205
206			sub pid {
207			undef
208			}
209
210			=item $proc = $proc->send_fh (...)
211
212			Not supported and always croaks.
213
214			=cut
215
216			sub send_fh {
217			require Carp;
218			Carp::croak ("send_fh is not supported on AnyEvent::Fork::Remote objects");
219			}
220
221			=item $proc = $proc->eval ($perlcode, @args)
222
223			Quite the same as the same method of L<AnyEvent::Fork>.
224
225			=cut
226
227			# quote a binary string as a perl scalar
228			sub sq($) {
229			my $s = shift;
230
231			$s =~ /'/
232			or return "'$s'";
233
234			$s =~ s/(\x10+)/\x10.'$1'.q\x10/g;
235			"q\x10$s\x10"
236			}
237
238			# quote a list of strings
239			sub aq(@) {
240			"(" . (join ",", map sq $_, @_) . ")"
241			}
242
243			sub eval {
244			my ($self, $perlcode, @args) = @_;
245
246			$self->[1] .= '{ local @_ = ' . (aq @args) . "; $perlcode }\n";
247			}
248
249			=item $proc = $proc->require ($module, ...)
250
251			Quite the same as the same method of L<AnyEvent::Fork>.
252
253			=cut
254
255			sub require {
256			my ($self, @modules) = @_;
257
258			s%::%/%g for @modules;
259			$self->eval ('require "$_.pm" for @_', @modules);
260
261			$self
262			}
263
264			=item $proc = $proc->send_arg ($string, ...)
265
266			Quite the same as the same method of L<AnyEvent::Fork>.
267
268			=cut
269
270			sub send_arg {
271			my ($self, @arg) = @_;
272
273			push @{ $self->[2] }, @arg;
274
275			$self
276			}
277
278			=item $proc->run ($func, $cb->($fh))
279
280			Very similar to the run method of L<AnyEvent::Fork>.
281
282			On the parent side, the API is identical. On the child side, the
283			"communications socket" is in fact just C<*STDIN>, and typically can only
284			be read from.
285
286			If the run function wants to read data that is written to C<$fh> in the
287			parent, then it should read from STDIN. If the run function wants to
288			provide data that can later be read from C<$fh>, then it should write them
289			to STDOUT.
290
291			You can write a run function that works with both L<AnyEvent::Fork> and
292			this module by checking C<fileno $fh> in on the passed callback in the run
293			function:
294
295			sub run {
296			my ($rfh, ...) = @_;
297			my $wfh = fileno $rfh ? $rfh : *STDOUT;
298
299			# now use $rfh for reading and $wfh for writing
300			}
301
302			=cut
303
304			sub run {
305			my ($self, $func, $cb) = @_;
306
307			my $code = 'BEGIN {' . $self->[1] . '}'
308			. 'syswrite STDOUT, ' . (sq $magic0) . '^' . (sq $magic1) . ';'
309			. $func . (aq @{ $self->[2] }) . ';'
310			. "\n__END__\n";
311
312			$self->[0](sub {
313			my $fh = shift
314			or die "AnyEvent::Fork::Remote: create callback failed";
315
316
317
318			});
319			}
320
321			my $x = new_exec AnyEvent::Fork::Remote "/usr/bin/rsh", "rsh", "rain", "exec perl";#d#
322			$x->require ("Carp", "Storable");#d#
323			$x->send_arg (1, 2, 3);#d#
324			$x->eval ('sub run { die }');#d#
325			$x->run ("run", sub {
326			});
327
328
329			=item my $proc = new_from_stdio $fh
330
331			Creates an C<AnyEvent::Fork::Remote> object from a file handle. This file
332			handle must be connected to both STDIN and STDOUT of a F<perl> process.
333
334			This form might be more convenient than C<new> or C<new_exec> when
335			creating an C<AnyEvent::Fork::Remote> object, but the resulting object
336			does not support C<fork>.
337
338			#TODO: really implement?
339
340			=back
341
342			=head1 SEE ALSO
343
344			L<AnyEvent::Fork>, the same as this module, for local processes.
345
346			L<AnyEvent::Fork::RPC>, to talk to the created processes.
347
348			L<AnyEvent::Fork::Pool>, to manage whole pools of processes.
349
350			=head1 AUTHOR AND CONTACT INFORMATION
351
352			Marc Lehmann <schmorp@schmorp.de>
353			http://software.schmorp.de/pkg/AnyEvent-Fork-Remote
354
355			=cut
356
357			1
358