AnyEvent-Fork/Fork.pm

=head1 NAME

AnyEvent::ProcessPool - manage pools of perl worker processes, exec'ed or fork'ed

=head1 SYNOPSIS

   use AnyEvent::ProcessPool;

=head1 DESCRIPTION

This module allows you to create single worker processes but also worker
pool that share memory, by forking from the main program, or exec'ing new
perl interpreters from a module.

You create a new processes in a pool by specifying a function to call
with any combination of string values and file handles.

A pool can have initialisation code which is executed before forking. The
initialisation code is only executed once and the resulting process is
cached, to be used as a template.

Pools without such initialisation code don't cache an extra process.

=head1 PROBLEM STATEMENT

There are two ways to implement parallel processing on UNIX like operating
systems - fork and process, and fork+exec and process. They have different
advantages and disadvantages that I describe below, together with how this
module tries to mitigate the disadvantages.

=over 4

=item Forking from a big process can be very slow (a 5GB process needs
0.05s to fork on my 3.6GHz amd64 GNU/Linux box for example). This overhead
is often shared with exec (because you have to fork first), but in some
circumstances (e.g. when vfork is used), fork+exec can be much faster.

This module can help here by telling a small(er) helper process to fork,
or fork+exec instead.

=item Forking usually creates a copy-on-write copy of the parent
process. Memory (for example, modules or data files that have been
will not take additional memory). When exec'ing a new process, modules
and data files might need to be loaded again, at extra cpu and memory
cost. Likewise when forking, all data structures are copied as well - if
the program frees them and replaces them by new data, the child processes
will retain the memory even if it isn't used.

This module allows the main program to do a controlled fork, and allows
modules to exec processes safely at any time. When creating a custom
process pool you can take advantage of data sharing via fork without
risking to share large dynamic data structures that will blow up child
memory usage.

=item Exec'ing a new perl process might be difficult and slow. For
example, it is not easy to find the correct path to the perl interpreter,
and all modules have to be loaded from disk again. Long running processes
might run into problems when perl is upgraded for example.

This module supports creating pre-initialised perl processes to be used
as template, and also tries hard to identify the correct path to the perl
interpreter. With a cooperative main program, exec'ing the interpreter
might not even be necessary.

=item Forking might be impossible when a program is running. For example,
POSIX makes it almost impossible to fork from a multithreaded program and
do anything useful in the child - strictly speaking, if your perl program
uses posix threads (even indirectly via e.g. L<IO::AIO> or L<threads>),
you cannot call fork on the perl level anymore, at all.

This module can safely fork helper processes at any time, by caling
fork+exec in C, in a POSIX-compatible way.

=item Parallel processing with fork might be inconvenient or difficult
to implement. For example, when a program uses an event loop and creates
watchers it becomes very hard to use the event loop from a child
program, as the watchers already exist but are only meaningful in the
parent. Worse, a module might want to use such a system, not knowing
whether another module or the main program also does, leading to problems.

This module only lets the main program create pools by forking (because
only the main program can know when it is still safe to do so) - all other
pools are created by fork+exec, after which such modules can again be
loaded.

=back

=over 4

=cut

package AnyEvent::ProcessPool;

use common::sense;

use Socket ();

use Proc::FastSpawn;
use AnyEvent;
use AnyEvent::ProcessPool::Util;
use AnyEvent::Util ();

BEGIN {
#   require Exporter;
}

=item my $pool = new AnyEvent::ProcessPool key => value...

Create a new process pool. The following named parameters are supported:

=over 4

=back

=cut

# the template process
our $template;

sub _queue {
   my ($pid, $fh) = @_;

   [
      $pid,
      $fh,
      [],
      undef
   ]
}

sub queue_cmd {
   my ($queue, $cmd) = @_;

   push @{ $queue->[2] }, pack "N/a", $cmd;

   $queue->[3] ||= AE::io $queue->[1], 1, sub {
      warn "oopl0 ", scalar @{ $queue->[2] };
      if (ref $queue->[2][0]) {
         warn "oopla2\n";#d#
         AnyEvent::ProcessPool::Util::fd_send fileno $queue->[1], fileno ${ $queue->[2][0] }
            and shift @{ $queue->[2] };
      } else {
         warn "write ", length $queue->[2][0];#d#
         my $len = syswrite $queue->[1], $queue->[2][0]
            or die "AnyEvent::ProcessPool::queue write failure: $!";
         substr $queue->[2][0], 0, $len, "";
         shift @{ $queue->[2] } unless length $queue->[2][0];
      }

      undef $queue->[3] unless @{ $queue->[2] };
      warn "oopl3 ", scalar @{ $queue->[2] };
      warn "oopl4 $queue->[3]\n";#d#
   };
}

sub run_template {
   return if $template;

   my ($fh, $slave) = AnyEvent::Util::portable_socketpair;
   AnyEvent::Util::fh_nonblocking $fh, 1;
   fd_inherit fileno $slave;

   my %env = %ENV;
   $env{PERL5LIB} = join ":", grep !ref, @INC;

   my $pid = spawn
         $^X,
         ["perl", "-MAnyEvent::ProcessPool::Serve", "-e", "AnyEvent::ProcessPool::Serve::me", fileno $slave],
         [map "$_=$env{$_}", keys %env],
      or die "unable to spawn AnyEvent::ProcessPool server: $!";

   close $slave;

   $template = _queue $pid, $fh;

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

   queue_cmd $template, "Iabc";
   push @{ $template->[2] }, \$b;

   use Coro::AnyEvent; Coro::AnyEvent::sleep 1;
}

sub new {
   my $class = shift;

   my $self = bless {
      @_
   }, $class;

   run_template;

   $self
}

=back

=head1 AUTHOR

 Marc Lehmann <schmorp@schmorp.de>
 http://home.schmorp.de/

=cut

1

Revision:	1.1
Committed:	Sun Mar 31 03:21:27 2013 UTC (13 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::ProcessPool - manage pools of perl worker processes, exec'ed or fork'ed
4
5			=head1 SYNOPSIS
6
7			use AnyEvent::ProcessPool;
8
9			=head1 DESCRIPTION
10
11			This module allows you to create single worker processes but also worker
12			pool that share memory, by forking from the main program, or exec'ing new
13			perl interpreters from a module.
14
15			You create a new processes in a pool by specifying a function to call
16			with any combination of string values and file handles.
17
18			A pool can have initialisation code which is executed before forking. The
19			initialisation code is only executed once and the resulting process is
20			cached, to be used as a template.
21
22			Pools without such initialisation code don't cache an extra process.
23
24			=head1 PROBLEM STATEMENT
25
26			There are two ways to implement parallel processing on UNIX like operating
27			systems - fork and process, and fork+exec and process. They have different
28			advantages and disadvantages that I describe below, together with how this
29			module tries to mitigate the disadvantages.
30
31			=over 4
32
33			=item Forking from a big process can be very slow (a 5GB process needs
34			0.05s to fork on my 3.6GHz amd64 GNU/Linux box for example). This overhead
35			is often shared with exec (because you have to fork first), but in some
36			circumstances (e.g. when vfork is used), fork+exec can be much faster.
37
38			This module can help here by telling a small(er) helper process to fork,
39			or fork+exec instead.
40
41			=item Forking usually creates a copy-on-write copy of the parent
42			process. Memory (for example, modules or data files that have been
43			will not take additional memory). When exec'ing a new process, modules
44			and data files might need to be loaded again, at extra cpu and memory
45			cost. Likewise when forking, all data structures are copied as well - if
46			the program frees them and replaces them by new data, the child processes
47			will retain the memory even if it isn't used.
48
49			This module allows the main program to do a controlled fork, and allows
50			modules to exec processes safely at any time. When creating a custom
51			process pool you can take advantage of data sharing via fork without
52			risking to share large dynamic data structures that will blow up child
53			memory usage.
54
55			=item Exec'ing a new perl process might be difficult and slow. For
56			example, it is not easy to find the correct path to the perl interpreter,
57			and all modules have to be loaded from disk again. Long running processes
58			might run into problems when perl is upgraded for example.
59
60			This module supports creating pre-initialised perl processes to be used
61			as template, and also tries hard to identify the correct path to the perl
62			interpreter. With a cooperative main program, exec'ing the interpreter
63			might not even be necessary.
64
65			=item Forking might be impossible when a program is running. For example,
66			POSIX makes it almost impossible to fork from a multithreaded program and
67			do anything useful in the child - strictly speaking, if your perl program
68			uses posix threads (even indirectly via e.g. L<IO::AIO> or L<threads>),
69			you cannot call fork on the perl level anymore, at all.
70
71			This module can safely fork helper processes at any time, by caling
72			fork+exec in C, in a POSIX-compatible way.
73
74			=item Parallel processing with fork might be inconvenient or difficult
75			to implement. For example, when a program uses an event loop and creates
76			watchers it becomes very hard to use the event loop from a child
77			program, as the watchers already exist but are only meaningful in the
78			parent. Worse, a module might want to use such a system, not knowing
79			whether another module or the main program also does, leading to problems.
80
81			This module only lets the main program create pools by forking (because
82			only the main program can know when it is still safe to do so) - all other
83			pools are created by fork+exec, after which such modules can again be
84			loaded.
85
86			=back
87
88			=over 4
89
90			=cut
91
92			package AnyEvent::ProcessPool;
93
94			use common::sense;
95
96			use Socket ();
97
98			use Proc::FastSpawn;
99			use AnyEvent;
100			use AnyEvent::ProcessPool::Util;
101			use AnyEvent::Util ();
102
103			BEGIN {
104			# require Exporter;
105			}
106
107			=item my $pool = new AnyEvent::ProcessPool key => value...
108
109			Create a new process pool. The following named parameters are supported:
110
111			=over 4
112
113			=back
114
115			=cut
116
117			# the template process
118			our $template;
119
120			sub _queue {
121			my ($pid, $fh) = @_;
122
123			[
124			$pid,
125			$fh,
126			[],
127			undef
128			]
129			}
130
131			sub queue_cmd {
132			my ($queue, $cmd) = @_;
133
134			push @{ $queue->[2] }, pack "N/a", $cmd;
135
136			$queue->[3] \|\|= AE::io $queue->[1], 1, sub {
137			warn "oopl0 ", scalar @{ $queue->[2] };
138			if (ref $queue->[2][0]) {
139			warn "oopla2\n";#d#
140			AnyEvent::ProcessPool::Util::fd_send fileno $queue->[1], fileno ${ $queue->[2][0] }
141			and shift @{ $queue->[2] };
142			} else {
143			warn "write ", length $queue->[2][0];#d#
144			my $len = syswrite $queue->[1], $queue->[2][0]
145			or die "AnyEvent::ProcessPool::queue write failure: $!";
146			substr $queue->[2][0], 0, $len, "";
147			shift @{ $queue->[2] } unless length $queue->[2][0];
148			}
149
150			undef $queue->[3] unless @{ $queue->[2] };
151			warn "oopl3 ", scalar @{ $queue->[2] };
152			warn "oopl4 $queue->[3]\n";#d#
153			};
154			}
155
156			sub run_template {
157			return if $template;
158
159			my ($fh, $slave) = AnyEvent::Util::portable_socketpair;
160			AnyEvent::Util::fh_nonblocking $fh, 1;
161			fd_inherit fileno $slave;
162
163			my %env = %ENV;
164			$env{PERL5LIB} = join ":", grep !ref, @INC;
165
166			my $pid = spawn
167			$^X,
168			["perl", "-MAnyEvent::ProcessPool::Serve", "-e", "AnyEvent::ProcessPool::Serve::me", fileno $slave],
169			[map "$_=$env{$_}", keys %env],
170			or die "unable to spawn AnyEvent::ProcessPool server: $!";
171
172			close $slave;
173
174			$template = _queue $pid, $fh;
175
176			my ($a, $b) = AnyEvent::Util::portable_socketpair;
177
178			queue_cmd $template, "Iabc";
179			push @{ $template->[2] }, \$b;
180
181			use Coro::AnyEvent; Coro::AnyEvent::sleep 1;
182			}
183
184			sub new {
185			my $class = shift;
186
187			my $self = bless {
188			@_
189			}, $class;
190
191			run_template;
192
193			$self
194			}
195
196			=back
197
198			=head1 AUTHOR
199
200			Marc Lehmann <schmorp@schmorp.de>
201			http://home.schmorp.de/
202
203			=cut
204
205			1
206