ViewVC Help
View File | Revision Log | Show Annotations | Download File
/cvs/AnyEvent-Fork/Fork.pm
(Generate patch)

Comparing AnyEvent-Fork/Fork.pm (file contents):
Revision 1.5 by root, Wed Apr 3 08:29:21 2013 UTC vs.
Revision 1.10 by root, Thu Apr 4 06:09:15 2013 UTC

1=head1 NAME 1=head1 NAME
2 2
3AnyEvent::Fork - everything you wanted to use fork() for, but couldn't 3AnyEvent::Fork - everything you wanted to use fork() for, but couldn't
4 4
5ATTENTION, this is a very early release, and very untested. Consider it a
6technology preview.
7
5=head1 SYNOPSIS 8=head1 SYNOPSIS
6 9
7 use AnyEvent::Fork; 10 use AnyEvent::Fork;
11
12 ##################################################################
13 # create a single new process, tell it to run your worker function
14
15 AnyEvent::Fork
16 ->new
17 ->require ("MyModule")
18 ->run ("MyModule::worker, sub {
19 my ($master_filehandle) = @_;
20
21 # now $master_filehandle is connected to the
22 # $slave_filehandle in the new process.
23 });
24
25 # MyModule::worker might look like this
26 sub MyModule::worker {
27 my ($slave_filehandle) = @_;
28
29 # now $slave_filehandle is connected to the $master_filehandle
30 # in the original prorcess. have fun!
31 }
32
33 ##################################################################
34 # create a pool of server processes all accepting on the same socket
35
36 # create listener socket
37 my $listener = ...;
38
39 # create a pool template, initialise it and give it the socket
40 my $pool = AnyEvent::Fork
41 ->new
42 ->require ("Some::Stuff", "My::Server")
43 ->send_fh ($listener);
44
45 # now create 10 identical workers
46 for my $id (1..10) {
47 $pool
48 ->fork
49 ->send_arg ($id)
50 ->run ("My::Server::run");
51 }
52
53 # now do other things - maybe use the filehandle provided by run
54 # to wait for the processes to die. or whatever.
55
56 # My::Server::run might look like this
57 sub My::Server::run {
58 my ($slave, $listener, $id) = @_;
59
60 close $slave; # we do not use the socket, so close it to save resources
61
62 # we could go ballistic and use e.g. AnyEvent here, or IO::AIO,
63 # or anything we usually couldn't do in a process forked normally.
64 while (my $socket = $listener->accept) {
65 # do sth. with new socket
66 }
67 }
8 68
9=head1 DESCRIPTION 69=head1 DESCRIPTION
10 70
11This module allows you to create new processes, without actually forking 71This module allows you to create new processes, without actually forking
12them from your current process (avoiding the problems of forking), but 72them from your current process (avoiding the problems of forking), but
… …
111time, and the memory is not shared with anything else. 171time, and the memory is not shared with anything else.
112 172
113This is ideal for when you only need one extra process of a kind, with the 173This is ideal for when you only need one extra process of a kind, with the
114option of starting and stipping it on demand. 174option of starting and stipping it on demand.
115 175
176Example:
177
178 AnyEvent::Fork
179 ->new
180 ->require ("Some::Module")
181 ->run ("Some::Module::run", sub {
182 my ($fork_fh) = @_;
183 });
184
116=item fork a new template process, load code, then fork processes off of 185=item fork a new template process, load code, then fork processes off of
117it and run the code 186it and run the code
118 187
119When you need to have a bunch of processes that all execute the same (or 188When you need to have a bunch of processes that all execute the same (or
120very similar) tasks, then a good way is to create a new template process 189very similar) tasks, then a good way is to create a new template process
… …
128The disadvantage of this approach is that you need to create a template 197The disadvantage of this approach is that you need to create a template
129process for the sole purpose of forking new processes from it, but if you 198process for the sole purpose of forking new processes from it, but if you
130only need a fixed number of proceses you can create them, and then destroy 199only need a fixed number of proceses you can create them, and then destroy
131the template process. 200the template process.
132 201
202Example:
203
204 my $template = AnyEvent::Fork->new->require ("Some::Module");
205
206 for (1..10) {
207 $template->fork->run ("Some::Module::run", sub {
208 my ($fork_fh) = @_;
209 });
210 }
211
212 # at this point, you can keep $template around to fork new processes
213 # later, or you can destroy it, which causes it to vanish.
214
133=item execute a new perl interpreter, load some code, run it 215=item execute a new perl interpreter, load some code, run it
134 216
135This is relatively slow, and doesn't allow you to share memory between 217This is relatively slow, and doesn't allow you to share memory between
136multiple processes. 218multiple processes.
137 219
138The only advantage is that you don't have to have a template process 220The only advantage is that you don't have to have a template process
139hanging around all the time to fork off some new processes, which might be 221hanging around all the time to fork off some new processes, which might be
140an advantage when there are long time spans where no extra processes are 222an advantage when there are long time spans where no extra processes are
141needed. 223needed.
142 224
225Example:
226
227 AnyEvent::Fork
228 ->new_exec
229 ->require ("Some::Module")
230 ->run ("Some::Module::run", sub {
231 my ($fork_fh) = @_;
232 });
233
143=back 234=back
144 235
145=head1 FUNCTIONS 236=head1 FUNCTIONS
146 237
147=over 4 238=over 4
… …
177our $TEMPLATE; 268our $TEMPLATE;
178 269
179sub _cmd { 270sub _cmd {
180 my $self = shift; 271 my $self = shift;
181 272
273 #TODO: maybe append the packet to any existing string command already in the queue
274
182 # ideally, we would want to use "a (w/a)*" as format string, but perl versions 275 # ideally, we would want to use "a (w/a)*" as format string, but perl versions
183 # from at least 5.8.9 to 5.16.3 are all buggy and can't unpack it. 276 # from at least 5.8.9 to 5.16.3 are all buggy and can't unpack it.
184 push @{ $self->[2] }, pack "N/a", pack "(w/a)*", @_; 277 push @{ $self->[2] }, pack "N/a", pack "(w/a)*", @_;
185 278
186 $self->[3] ||= AE::io $self->[1], 1, sub { 279 $self->[3] ||= AE::io $self->[1], 1, sub {
280 # send the next "thing" in the queue - either a reference to an fh,
281 # or a plain string.
282
187 if (ref $self->[2][0]) { 283 if (ref $self->[2][0]) {
284 # send fh
188 AnyEvent::Fork::Util::fd_send fileno $self->[1], fileno ${ $self->[2][0] } 285 AnyEvent::Fork::Util::fd_send fileno $self->[1], fileno ${ $self->[2][0] }
189 and shift @{ $self->[2] }; 286 and shift @{ $self->[2] };
190 287
191 } else { 288 } else {
289 # send string
192 my $len = syswrite $self->[1], $self->[2][0] 290 my $len = syswrite $self->[1], $self->[2][0]
193 or do { undef $self->[3]; die "AnyEvent::Fork: command write failure: $!" }; 291 or do { undef $self->[3]; die "AnyEvent::Fork: command write failure: $!" };
194 292
195 substr $self->[2][0], 0, $len, ""; 293 substr $self->[2][0], 0, $len, "";
196 shift @{ $self->[2] } unless length $self->[2][0]; 294 shift @{ $self->[2] } unless length $self->[2][0];
197 } 295 }
198 296
199 unless (@{ $self->[2] }) { 297 unless (@{ $self->[2] }) {
200 undef $self->[3]; 298 undef $self->[3];
299 # invoke run callback
201 $self->[0]->($self->[1]) if $self->[0]; 300 $self->[0]->($self->[1]) if $self->[0];
202 } 301 }
203 }; 302 };
204} 303}
205 304
206sub _new { 305sub _new {
207 my ($self, $fh) = @_; 306 my ($self, $fh) = @_;
307
308 AnyEvent::Util::fh_nonblocking $fh, 1;
208 309
209 $self = bless [ 310 $self = bless [
210 undef, # run callback 311 undef, # run callback
211 $fh, 312 $fh,
212 [], # write queue - strings or fd's 313 [], # write queue - strings or fd's
213 undef, # AE watcher 314 undef, # AE watcher
214 ], $self; 315 ], $self;
215 316
216# my ($a, $b) = AnyEvent::Util::portable_socketpair;
217
218# queue_cmd $template, "Iabc";
219# push @{ $template->[2] }, \$b;
220
221# use Coro::AnyEvent; Coro::AnyEvent::sleep 1;
222# undef $b;
223# die "x" . <$a>;
224
225 $self 317 $self
318}
319
320# fork template from current process, used by AnyEvent::Fork::Early/Template
321sub _new_fork {
322 my ($fh, $slave) = AnyEvent::Util::portable_socketpair;
323 my $parent = $$;
324
325 my $pid = fork;
326
327 if ($pid eq 0) {
328 require AnyEvent::Fork::Serve;
329 $AnyEvent::Fork::Serve::OWNER = $parent;
330 close $fh;
331 $0 = "$_[1] of $parent";
332 AnyEvent::Fork::Serve::serve ($slave);
333 AnyEvent::Fork::Util::_exit 0;
334 } elsif (!$pid) {
335 die "AnyEvent::Fork::Early/Template: unable to fork template process: $!";
336 }
337
338 AnyEvent::Fork->_new ($fh)
226} 339}
227 340
228=item my $proc = new AnyEvent::Fork 341=item my $proc = new AnyEvent::Fork
229 342
230Create a new "empty" perl interpreter process and returns its process 343Create a new "empty" perl interpreter process and returns its process
231object for further manipulation. 344object for further manipulation.
232 345
233The new process is forked from a template process that is kept around 346The new process is forked from a template process that is kept around
234for this purpose. When it doesn't exist yet, it is created by a call to 347for this purpose. When it doesn't exist yet, it is created by a call to
235C<new_exec> and kept around for future calls. 348C<new_exec> and kept around for future calls.
349
350When the process object is destroyed, it will release the file handle
351that connects it with the new process. When the new process has not yet
352called C<run>, then the process will exit. Otherwise, what happens depends
353entirely on the code that is executed.
236 354
237=cut 355=cut
238 356
239sub new { 357sub new {
240 my $class = shift; 358 my $class = shift;
… …
260 378
261 my ($fh, $slave) = AnyEvent::Util::portable_socketpair; 379 my ($fh, $slave) = AnyEvent::Util::portable_socketpair;
262 380
263 $self->send_fh ($slave); 381 $self->send_fh ($slave);
264 $self->_cmd ("f"); 382 $self->_cmd ("f");
265
266 AnyEvent::Util::fh_nonblocking $fh, 1;
267 383
268 AnyEvent::Fork->_new ($fh) 384 AnyEvent::Fork->_new ($fh)
269} 385}
270 386
271=item my $proc = new_exec AnyEvent::Fork 387=item my $proc = new_exec AnyEvent::Fork
… …
309 } 425 }
310 426
311 require Proc::FastSpawn; 427 require Proc::FastSpawn;
312 428
313 my ($fh, $slave) = AnyEvent::Util::portable_socketpair; 429 my ($fh, $slave) = AnyEvent::Util::portable_socketpair;
314 AnyEvent::Util::fh_nonblocking $fh, 1;
315 Proc::FastSpawn::fd_inherit (fileno $slave); 430 Proc::FastSpawn::fd_inherit (fileno $slave);
431
432 # new fh's should always be set cloexec (due to $^F),
433 # but hey, not on win32, so we always clear the inherit flag.
434 Proc::FastSpawn::fd_inherit (fileno $fh, 0);
316 435
317 # quick. also doesn't work in win32. of course. what did you expect 436 # quick. also doesn't work in win32. of course. what did you expect
318 #local $ENV{PERL5LIB} = join ":", grep !ref, @INC; 437 #local $ENV{PERL5LIB} = join ":", grep !ref, @INC;
319 my %env = %ENV; 438 my %env = %ENV;
320 $env{PERL5LIB} = join ":", grep !ref, @INC; 439 $env{PERL5LIB} = join +(AnyEvent::Fork::Util::WIN32 ? ";" : ":"), grep !ref, @INC;
321 440
322 Proc::FastSpawn::spawn ( 441 Proc::FastSpawn::spawn (
323 $perl, 442 $perl,
324 ["perl", "-MAnyEvent::Fork::Serve", "-e", "AnyEvent::Fork::Serve::me", fileno $slave], 443 ["perl", "-MAnyEvent::Fork::Serve", "-e", "AnyEvent::Fork::Serve::me", fileno $slave, $$],
325 [map "$_=$env{$_}", keys %env], 444 [map "$_=$env{$_}", keys %env],
326 ) or die "unable to spawn AnyEvent::Fork server: $!"; 445 ) or die "unable to spawn AnyEvent::Fork server: $!";
327 446
328 $self->_new ($fh) 447 $self->_new ($fh)
329} 448}
330 449
450=item $proc = $proc->eval ($perlcode, @args)
451
452Evaluates the given C<$perlcode> as ... perl code, while setting C<@_> to
453the strings specified by C<@args>.
454
455This call is meant to do any custom initialisation that might be required
456(for example, the C<require> method uses it). It's not supposed to be used
457to completely take over the process, use C<run> for that.
458
459The code will usually be executed after this call returns, and there is no
460way to pass anything back to the calling process. Any evaluation errors
461will be reported to stderr and cause the process to exit.
462
463Returns the process object for easy chaining of method calls.
464
465=cut
466
467sub eval {
468 my ($self, $code, @args) = @_;
469
470 $self->_cmd (e => $code, @args);
471
472 $self
473}
474
331=item $proc = $proc->require ($module, ...) 475=item $proc = $proc->require ($module, ...)
332 476
333Tries to load the given modules into the process 477Tries to load the given module(s) into the process
334 478
335Returns the process object for easy chaining of method calls. 479Returns the process object for easy chaining of method calls.
480
481=cut
482
483sub require {
484 my ($self, @modules) = @_;
485
486 s%::%/%g for @modules;
487 $self->eval ('require "$_.pm" for @_', @modules);
488
489 $self
490}
336 491
337=item $proc = $proc->send_fh ($handle, ...) 492=item $proc = $proc->send_fh ($handle, ...)
338 493
339Send one or more file handles (I<not> file descriptors) to the process, 494Send one or more file handles (I<not> file descriptors) to the process,
340to prepare a call to C<run>. 495to prepare a call to C<run>.
… …
344accomplished by simply not storing the file handles anywhere after passing 499accomplished by simply not storing the file handles anywhere after passing
345them to this method. 500them to this method.
346 501
347Returns the process object for easy chaining of method calls. 502Returns the process object for easy chaining of method calls.
348 503
504Example: pass an fh to a process, and release it without closing. it will
505be closed automatically when it is no longer used.
506
507 $proc->send_fh ($my_fh);
508 undef $my_fh; # free the reference if you want, but DO NOT CLOSE IT
509
349=cut 510=cut
350 511
351sub send_fh { 512sub send_fh {
352 my ($self, @fh) = @_; 513 my ($self, @fh) = @_;
353 514
… …
394to save on kernel memory. 555to save on kernel memory.
395 556
396The socket is non-blocking in the parent, and blocking in the newly 557The socket is non-blocking in the parent, and blocking in the newly
397created process. The close-on-exec flag is set on both. Even if not used 558created process. The close-on-exec flag is set on both. Even if not used
398otherwise, the socket can be a good indicator for the existance of the 559otherwise, the socket can be a good indicator for the existance of the
399process - if the othe rprocess exits, you get a readable event on it, 560process - if the other process exits, you get a readable event on it,
400because exiting the process closes the socket (if it didn't create any 561because exiting the process closes the socket (if it didn't create any
401children using fork). 562children using fork).
402 563
564Example: create a template for a process pool, pass a few strings, some
565file handles, then fork, pass one more string, and run some code.
566
567 my $pool = AnyEvent::Fork
568 ->new
569 ->send_arg ("str1", "str2")
570 ->send_fh ($fh1, $fh2);
571
572 for (1..2) {
573 $pool
574 ->fork
575 ->send_arg ("str3")
576 ->run ("Some::function", sub {
577 my ($fh) = @_;
578
579 # fh is nonblocking, but we trust that the OS can accept these
580 # extra 3 octets anyway.
581 syswrite $fh, "hi #$_\n";
582
583 # $fh is being closed here, as we don't store it anywhere
584 });
585 }
586
587 # Some::function might look like this - all parameters passed before fork
588 # and after will be passed, in order, after the communications socket.
589 sub Some::function {
590 my ($fh, $str1, $str2, $fh1, $fh2, $str3) = @_;
591
592 print scalar <$fh>; # prints "hi 1\n" and "hi 2\n"
593 }
594
403=cut 595=cut
404 596
405sub run { 597sub run {
406 my ($self, $func, $cb) = @_; 598 my ($self, $func, $cb) = @_;
407 599
408 $self->[0] = $cb; 600 $self->[0] = $cb;
409 $self->_cmd ("r", $func); 601 $self->_cmd (r => $func);
410} 602}
411 603
412=back 604=back
605
606=head1 PORTABILITY NOTES
607
608Native win32 perls are somewhat supported (AnyEvent::Fork::Early is a nop,
609and ::Template is not going to work), and it cost a lot of blood and sweat
610to make it so, mostly due to the bloody broken perl that nobody seems to
611care about. The fork emulation is a bad joke - I have yet to see something
612useful that you cna do with it without running into memory corruption
613issues or other braindamage. Hrrrr.
614
615Cygwin perl is not supported at the moment, as it should implement fd
616passing, but doesn't, and rolling my own is hard, as cygwin doesn't
617support enough functionality to do it.
413 618
414=head1 AUTHOR 619=head1 AUTHOR
415 620
416 Marc Lehmann <schmorp@schmorp.de> 621 Marc Lehmann <schmorp@schmorp.de>
417 http://home.schmorp.de/ 622 http://home.schmorp.de/

Diff Legend

– Removed lines
+ Added lines
< Changed lines
> Changed lines