AnyEvent-MP/MP.pm

=head1 NAME

AnyEvent::MP - multi-processing/message-passing framework

=head1 SYNOPSIS

   use AnyEvent::MP;

   NODE    # returns this node identifier
   $NODE   # contains this node identifier

   snd $port, type => data...;

   rcv $port, smartmatch => $cb->($port, @msg);

   # examples:
   rcv $port2, ping => sub { snd $_[0], "pong"; 0 };
   rcv $port1, pong => sub { warn "pong received\n" };
   snd $port2, ping => $port1;

   # more, smarter, matches (_any_ is exported by this module)
   rcv $port, [child_died => $pid] => sub { ...
   rcv $port, [_any_, _any_, 3] => sub { .. $_[2] is 3

=head1 DESCRIPTION

This module (-family) implements a simple message passing framework.

Despite its simplicity, you can securely message other processes running
on the same or other hosts.

At the moment, this module family is severly brokena nd underdocumented,
so do not use. This was uploaded mainly to reserve the CPAN namespace -
stay tuned!

=head1 CONCEPTS

=over 4

=item port

A port is something you can send messages to with the C<snd> function, and
you can register C<rcv> handlers with. All C<rcv> handlers will receive
messages they match, messages will not be queued.

=item port id - C<noderef#portname>

A port id is always the noderef, a hash-mark (C<#>) as separator, followed
by a port name (a printable string of unspecified format).

=item node

A node is a single process containing at least one port - the node
port. You can send messages to node ports to let them create new ports,
among other things.

Initially, nodes are either private (single-process only) or hidden
(connected to a master node only). Only when they epxlicitly "become
public" can you send them messages from unrelated other nodes.

=item noderef - C<host:port,host:port...>, C<id@noderef>, C<id>

A noderef is a string that either uniquely identifies a given node (for
private and hidden nodes), or contains a recipe on how to reach a given
node (for public nodes).

=back

=head1 VARIABLES/FUNCTIONS

=over 4

=cut

package AnyEvent::MP;

use AnyEvent::MP::Base;

use common::sense;

use Carp ();

use AE ();

use base "Exporter";

our $VERSION = '0.02';
our @EXPORT = qw(
   NODE $NODE $PORT snd rcv mon kil _any_
   create_port create_port_on
   miniport
   become_slave become_public
);

=item NODE / $NODE

The C<NODE ()> function and the C<$NODE> variable contain the noderef of
the local node. The value is initialised by a call to C<become_public> or
C<become_slave>, after which all local port identifiers become invalid.

=item snd $portid, type => @data

=item snd $portid, @msg

Send the given message to the given port ID, which can identify either
a local or a remote port, and can be either a string or soemthignt hat
stringifies a sa port ID (such as a port object :).

While the message can be about anything, it is highly recommended to use a
string as first element (a portid, or some word that indicates a request
type etc.).

The message data effectively becomes read-only after a call to this
function: modifying any argument is not allowed and can cause many
problems.

The type of data you can transfer depends on the transport protocol: when
JSON is used, then only strings, numbers and arrays and hashes consisting
of those are allowed (no objects). When Storable is used, then anything
that Storable can serialise and deserialise is allowed, and for the local
node, anything can be passed.

=item $guard = mon $portid, $cb->()

=item $guard = mon $portid, $otherport

=item $guard = mon $portid, $otherport, @msg

Monitor the given port and call the given callback when the port is
destroyed or connection to it's node is lost.

#TODO

=cut

sub mon {
   my ($noderef, $port, $cb) = ((split /#/, shift, 2), shift);

   my $node = AnyEvent::MP::Base::add_node $noderef;

   #TODO: ports must not be references
   if (!ref $cb or "AnyEvent::MP::Port" eq ref $cb) {
      if (@_) {
         # send a kill info message
         my (@msg) = ($cb, @_);
         $cb = sub { snd @msg, @_ };
      } else {
         # simply kill other port
         my $port = $cb;
         $cb = sub { kil $port, @_ };
      }
   }

   $node->monitor ($port, $cb);

   defined wantarray
      and AnyEvent::Util::guard { $node->unmonitor ($port, $cb) }
}

=item $guard = mon_guard $port, $ref, $ref...

Monitors the given C<$port> and keeps the passed references. When the port
is killed, the references will be freed.

Optionally returns a guard that will stop the monitoring.

This function is useful when you create e.g. timers or other watchers and
want to free them when the port gets killed:

  $port->rcv (start => sub {
     my $timer; $timer = mon_guard $port, AE::timer 1, 1, sub {
        undef $timer if 0.9 < rand;
     });
  });

=cut

sub mon_guard {
   my ($port, @refs) = @_;

   mon $port, sub { 0 && @refs }
}

=item $local_port = create_port

Create a new local port object. See the next section for allowed methods.

=cut

sub create_port {
   my $id = "$AnyEvent::MP::Base::UNIQ." . $AnyEvent::MP::Base::ID++;

   my $self = bless {
      id    => "$NODE#$id",
   }, "AnyEvent::MP::Port";

   $AnyEvent::MP::Base::PORT{$id} = sub {
      unshift @_, $self;

      for (@{ $self->{rc0}{$_[1]} }) {
         $_ && &{$_->[0]}
            && undef $_;
      }

      for (@{ $self->{rcv}{$_[1]} }) {
         $_ && [@_[1 .. @{$_->[1]}]] ~~ $_->[1]
            && &{$_->[0]}
            && undef $_;
      }

      for (@{ $self->{any} }) {
         $_ && [@_[0 .. $#{$_->[1]}]] ~~ $_->[1]
            && &{$_->[0]}
            && undef $_;
      }
   };

   $self
}

=item $portid = miniport { my @msg = @_; $finished }

Creates a "mini port", that is, a very lightweight port without any
pattern matching behind it, and returns its ID.

The block will be called for every message received on the port. When the
callback returns a true value its job is considered "done" and the port
will be destroyed. Otherwise it will stay alive.

The message will be passed as-is, no extra argument (i.e. no port id) will
be passed to the callback.

If you need the local port id in the callback, this works nicely:

   my $port; $port = miniport {
      snd $otherport, reply => $port;
   };

=cut

sub miniport(&) {
   my $cb = shift;
   my $id = "$AnyEvent::MP::Base::UNIQ." . $AnyEvent::MP::Base::ID++;

   $AnyEvent::MP::Base::PORT{$id} = sub {
      &$cb
         and kil $id;
   };

   "$NODE#$id"
}

package AnyEvent::MP::Port;

=back

=head1 METHODS FOR PORT OBJECTS

=over 4

=item "$port"

A port object stringifies to its port ID, so can be used directly for
C<snd> operations.

=cut

use overload
   '""'     => sub { $_[0]{id} },
   fallback => 1;

sub TO_JSON { $_[0]{id} }

=item $port->rcv (type => $callback->($port, @msg))

=item $port->rcv ($smartmatch => $callback->($port, @msg))

=item $port->rcv ([$smartmatch...] => $callback->($port, @msg))

Register a callback on the given port.

The callback has to return a true value when its work is done, after
which is will be removed, or a false value in which case it will stay
registered.

If the match is an array reference, then it will be matched against the
first elements of the message, otherwise only the first element is being
matched.

Any element in the match that is specified as C<_any_> (a function
exported by this module) matches any single element of the message.

While not required, it is highly recommended that the first matching
element is a string identifying the message. The one-string-only match is
also the most efficient match (by far).

=cut

sub rcv($@) {
   my ($self, $match, $cb) = @_;

   if (!ref $match) {
      push @{ $self->{rc0}{$match}      }, [$cb];
   } elsif (("ARRAY" eq ref $match && !ref $match->[0])) {
      my ($type, @match) = @$match;
      @match
         ? push @{ $self->{rcv}{$match->[0]} }, [$cb, \@match]
         : push @{ $self->{rc0}{$match->[0]} }, [$cb];
   } else {
      push @{ $self->{any}              }, [$cb, $match];
   }
}

=item $port->register ($name)

Registers the given port under the well known name C<$name>. If the name
already exists it is replaced.

A port can only be registered under one well known name.

=cut

sub register {
   my ($self, $name) = @_;

   $self->{wkname} = $name;
   $AnyEvent::MP::Base::WKP{$name} = "$self";
}

=item $port->destroy

Explicitly destroy/remove/nuke/vaporise the port.

Ports are normally kept alive by their mere existance alone, and need to
be destroyed explicitly.

=cut

sub destroy {
   my ($self) = @_;

   delete $AnyEvent::MP::Base::WKP{ $self->{wkname} };

   AnyEvent::MP::Base::kil $self->{id};
}

=back

=head1 FUNCTIONS FOR NODES

=over 4

=item mon $noderef, $callback->($noderef, $status, $)

Monitors the given noderef.

=item become_public endpoint...

Tells the node to become a public node, i.e. reachable from other nodes.

If no arguments are given, or the first argument is C<undef>, then
AnyEvent::MP tries to bind on port C<4040> on all IP addresses that the
local nodename resolves to.

Otherwise the first argument must be an array-reference with transport
endpoints ("ip:port", "hostname:port") or port numbers (in which case the
local nodename is used as hostname). The endpoints are all resolved and
will become the node reference.

=cut

=back

=head1 NODE MESSAGES

Nodes understand the following messages sent to them. Many of them take
arguments called C<@reply>, which will simply be used to compose a reply
message - C<$reply[0]> is the port to reply to, C<$reply[1]> the type and
the remaining arguments are simply the message data.

=over 4

=cut

=item wkp => $name, @reply

Replies with the port ID of the specified well-known port, or C<undef>.

=item devnull => ...

Generic data sink/CPU heat conversion.

=item relay => $port, @msg

Simply forwards the message to the given port.

=item eval => $string[ @reply]

Evaluates the given string. If C<@reply> is given, then a message of the
form C<@reply, $@, @evalres> is sent.

Example: crash another node.

   snd $othernode, eval => "exit";

=item time => @reply

Replies the the current node time to C<@reply>.

Example: tell the current node to send the current time to C<$myport> in a
C<timereply> message.

   snd $NODE, time => $myport, timereply => 1, 2;
   # => snd $myport, timereply => 1, 2, <time>

=back

=head1 SEE ALSO

L<AnyEvent>.

=head1 AUTHOR

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

=cut

1

Revision:	1.21
Committed:	Tue Aug 4 14:10:51 2009 UTC (14 years, 9 months ago) by root
Branch:	MAIN
Changes since 1.20:	+47 -12 lines
Log Message:	* empty log message *
#	User	Rev	Content
1	root	1.1	=head1 NAME
2
3			AnyEvent::MP - multi-processing/message-passing framework
4
5			=head1 SYNOPSIS
6
7			use AnyEvent::MP;
8
9	root	1.2	NODE # returns this node identifier
10			$NODE # contains this node identifier
11
12			snd $port, type => data...;
13
14			rcv $port, smartmatch => $cb->($port, @msg);
15
16			# examples:
17			rcv $port2, ping => sub { snd $_[0], "pong"; 0 };
18			rcv $port1, pong => sub { warn "pong received\n" };
19			snd $port2, ping => $port1;
20
21			# more, smarter, matches (_any_ is exported by this module)
22			rcv $port, [child_died => $pid] => sub { ...
23			rcv $port, [_any_, _any_, 3] => sub { .. $_[2] is 3
24
25	root	1.1	=head1 DESCRIPTION
26
27	root	1.2	This module (-family) implements a simple message passing framework.
28
29			Despite its simplicity, you can securely message other processes running
30			on the same or other hosts.
31
32	root	1.6	At the moment, this module family is severly brokena nd underdocumented,
33	root	1.21	so do not use. This was uploaded mainly to reserve the CPAN namespace -
34	root	1.6	stay tuned!
35
36	root	1.2	=head1 CONCEPTS
37
38			=over 4
39
40			=item port
41
42			A port is something you can send messages to with the C<snd> function, and
43			you can register C<rcv> handlers with. All C<rcv> handlers will receive
44			messages they match, messages will not be queued.
45
46	root	1.3	=item port id - C<noderef#portname>
47	root	1.2
48	root	1.3	A port id is always the noderef, a hash-mark (C<#>) as separator, followed
49			by a port name (a printable string of unspecified format).
50	root	1.2
51			=item node
52
53			A node is a single process containing at least one port - the node
54			port. You can send messages to node ports to let them create new ports,
55			among other things.
56
57			Initially, nodes are either private (single-process only) or hidden
58	root	1.3	(connected to a master node only). Only when they epxlicitly "become
59			public" can you send them messages from unrelated other nodes.
60	root	1.2
61	root	1.5	=item noderef - C<host:port,host:port...>, C<id@noderef>, C<id>
62	root	1.2
63	root	1.3	A noderef is a string that either uniquely identifies a given node (for
64	root	1.2	private and hidden nodes), or contains a recipe on how to reach a given
65			node (for public nodes).
66
67			=back
68
69	root	1.3	=head1 VARIABLES/FUNCTIONS
70	root	1.2
71			=over 4
72
73	root	1.1	=cut
74
75			package AnyEvent::MP;
76
77	root	1.8	use AnyEvent::MP::Base;
78	root	1.2
79	root	1.1	use common::sense;
80
81	root	1.2	use Carp ();
82
83	root	1.1	use AE ();
84
85	root	1.2	use base "Exporter";
86
87	root	1.9	our $VERSION = '0.02';
88	root	1.8	our @EXPORT = qw(
89	root	1.21	NODE $NODE $PORT snd rcv mon kil _any_
90	root	1.8	create_port create_port_on
91	root	1.21	miniport
92	root	1.8	become_slave become_public
93			);
94	root	1.2
95	root	1.3	=item NODE / $NODE
96
97			The C<NODE ()> function and the C<$NODE> variable contain the noderef of
98			the local node. The value is initialised by a call to C<become_public> or
99			C<become_slave>, after which all local port identifiers become invalid.
100
101			=item snd $portid, type => @data
102
103			=item snd $portid, @msg
104
105	root	1.8	Send the given message to the given port ID, which can identify either
106			a local or a remote port, and can be either a string or soemthignt hat
107			stringifies a sa port ID (such as a port object :).
108
109			While the message can be about anything, it is highly recommended to use a
110			string as first element (a portid, or some word that indicates a request
111			type etc.).
112	root	1.3
113			The message data effectively becomes read-only after a call to this
114			function: modifying any argument is not allowed and can cause many
115			problems.
116
117			The type of data you can transfer depends on the transport protocol: when
118			JSON is used, then only strings, numbers and arrays and hashes consisting
119			of those are allowed (no objects). When Storable is used, then anything
120			that Storable can serialise and deserialise is allowed, and for the local
121			node, anything can be passed.
122
123	root	1.20	=item $guard = mon $portid, $cb->()
124	root	1.18
125	root	1.21	=item $guard = mon $portid, $otherport
126
127			=item $guard = mon $portid, $otherport, @msg
128
129	root	1.20	Monitor the given port and call the given callback when the port is
130			destroyed or connection to it's node is lost.
131
132			#TODO
133	root	1.18
134			=cut
135
136			sub mon {
137	root	1.21	my ($noderef, $port, $cb) = ((split /#/, shift, 2), shift);
138	root	1.18
139			my $node = AnyEvent::MP::Base::add_node $noderef;
140
141	root	1.21	#TODO: ports must not be references
142			if (!ref $cb or "AnyEvent::MP::Port" eq ref $cb) {
143			if (@_) {
144			# send a kill info message
145			my (@msg) = ($cb, @_);
146			$cb = sub { snd @msg, @_ };
147			} else {
148			# simply kill other port
149			my $port = $cb;
150			$cb = sub { kil $port, @_ };
151			}
152			}
153	root	1.18
154			$node->monitor ($port, $cb);
155
156			defined wantarray
157			and AnyEvent::Util::guard { $node->unmonitor ($port, $cb) }
158			}
159
160	root	1.21	=item $guard = mon_guard $port, $ref, $ref...
161
162			Monitors the given C<$port> and keeps the passed references. When the port
163			is killed, the references will be freed.
164
165			Optionally returns a guard that will stop the monitoring.
166
167			This function is useful when you create e.g. timers or other watchers and
168			want to free them when the port gets killed:
169
170			$port->rcv (start => sub {
171			my $timer; $timer = mon_guard $port, AE::timer 1, 1, sub {
172			undef $timer if 0.9 < rand;
173			});
174			});
175
176			=cut
177
178			sub mon_guard {
179			my ($port, @refs) = @_;
180
181			mon $port, sub { 0 && @refs }
182			}
183
184	root	1.8	=item $local_port = create_port
185
186			Create a new local port object. See the next section for allowed methods.
187
188	root	1.3	=cut
189
190	root	1.8	sub create_port {
191	root	1.16	my $id = "$AnyEvent::MP::Base::UNIQ." . $AnyEvent::MP::Base::ID++;
192	root	1.8
193			my $self = bless {
194			id => "$NODE#$id",
195			}, "AnyEvent::MP::Port";
196
197			$AnyEvent::MP::Base::PORT{$id} = sub {
198			unshift @_, $self;
199
200			for (@{ $self->{rc0}{$_[1]} }) {
201			$_ && &{$_->[0]}
202			&& undef $_;
203			}
204
205			for (@{ $self->{rcv}{$_[1]} }) {
206			$_ && [@_[1 .. @{$_->[1]}]] ~~ $_->[1]
207			&& &{$_->[0]}
208			&& undef $_;
209			}
210	root	1.2
211	root	1.8	for (@{ $self->{any} }) {
212			$_ && [@_[0 .. $#{$_->[1]}]] ~~ $_->[1]
213			&& &{$_->[0]}
214			&& undef $_;
215			}
216			};
217	root	1.2
218	root	1.8	$self
219	root	1.3	}
220
221	root	1.15	=item $portid = miniport { my @msg = @_; $finished }
222	root	1.10
223	root	1.15	Creates a "mini port", that is, a very lightweight port without any
224			pattern matching behind it, and returns its ID.
225
226			The block will be called for every message received on the port. When the
227			callback returns a true value its job is considered "done" and the port
228			will be destroyed. Otherwise it will stay alive.
229
230	root	1.17	The message will be passed as-is, no extra argument (i.e. no port id) will
231	root	1.15	be passed to the callback.
232
233			If you need the local port id in the callback, this works nicely:
234
235			my $port; $port = miniport {
236			snd $otherport, reply => $port;
237			};
238	root	1.10
239			=cut
240
241	root	1.15	sub miniport(&) {
242	root	1.10	my $cb = shift;
243	root	1.16	my $id = "$AnyEvent::MP::Base::UNIQ." . $AnyEvent::MP::Base::ID++;
244	root	1.10
245			$AnyEvent::MP::Base::PORT{$id} = sub {
246			&$cb
247	root	1.21	and kil $id;
248	root	1.10	};
249
250			"$NODE#$id"
251			}
252
253	root	1.8	package AnyEvent::MP::Port;
254
255			=back
256
257			=head1 METHODS FOR PORT OBJECTS
258
259			=over 4
260
261			=item "$port"
262
263			A port object stringifies to its port ID, so can be used directly for
264			C<snd> operations.
265
266			=cut
267
268			use overload
269			'""' => sub { $_[0]{id} },
270			fallback => 1;
271
272	root	1.18	sub TO_JSON { $_[0]{id} }
273
274	root	1.8	=item $port->rcv (type => $callback->($port, @msg))
275	root	1.3
276	root	1.8	=item $port->rcv ($smartmatch => $callback->($port, @msg))
277	root	1.3
278	root	1.8	=item $port->rcv ([$smartmatch...] => $callback->($port, @msg))
279	root	1.3
280	root	1.8	Register a callback on the given port.
281	root	1.3
282			The callback has to return a true value when its work is done, after
283			which is will be removed, or a false value in which case it will stay
284			registered.
285
286			If the match is an array reference, then it will be matched against the
287			first elements of the message, otherwise only the first element is being
288			matched.
289
290			Any element in the match that is specified as C<_any_> (a function
291			exported by this module) matches any single element of the message.
292
293			While not required, it is highly recommended that the first matching
294			element is a string identifying the message. The one-string-only match is
295			also the most efficient match (by far).
296
297			=cut
298
299			sub rcv($@) {
300	root	1.8	my ($self, $match, $cb) = @_;
301	root	1.3
302			if (!ref $match) {
303	root	1.8	push @{ $self->{rc0}{$match} }, [$cb];
304	root	1.3	} elsif (("ARRAY" eq ref $match && !ref $match->[0])) {
305			my ($type, @match) = @$match;
306			@match
307	root	1.8	? push @{ $self->{rcv}{$match->[0]} }, [$cb, \@match]
308			: push @{ $self->{rc0}{$match->[0]} }, [$cb];
309	root	1.3	} else {
310	root	1.8	push @{ $self->{any} }, [$cb, $match];
311	root	1.3	}
312	root	1.2	}
313
314	root	1.8	=item $port->register ($name)
315	root	1.2
316	root	1.8	Registers the given port under the well known name C<$name>. If the name
317			already exists it is replaced.
318	root	1.2
319	root	1.8	A port can only be registered under one well known name.
320	root	1.3
321	root	1.8	=cut
322	root	1.3
323	root	1.8	sub register {
324			my ($self, $name) = @_;
325	root	1.3
326	root	1.8	$self->{wkname} = $name;
327			$AnyEvent::MP::Base::WKP{$name} = "$self";
328	root	1.1	}
329
330	root	1.8	=item $port->destroy
331	root	1.2
332	root	1.8	Explicitly destroy/remove/nuke/vaporise the port.
333	root	1.2
334	root	1.21	Ports are normally kept alive by their mere existance alone, and need to
335	root	1.8	be destroyed explicitly.
336	root	1.2
337	root	1.8	=cut
338	root	1.1
339	root	1.8	sub destroy {
340			my ($self) = @_;
341	root	1.1
342	root	1.8	delete $AnyEvent::MP::Base::WKP{ $self->{wkname} };
343	root	1.2
344	root	1.21	AnyEvent::MP::Base::kil $self->{id};
345	root	1.2	}
346
347	root	1.8	=back
348
349			=head1 FUNCTIONS FOR NODES
350
351			=over 4
352	root	1.2
353	root	1.8	=item mon $noderef, $callback->($noderef, $status, $)
354	root	1.2
355	root	1.8	Monitors the given noderef.
356	root	1.2
357	root	1.8	=item become_public endpoint...
358
359			Tells the node to become a public node, i.e. reachable from other nodes.
360
361			If no arguments are given, or the first argument is C<undef>, then
362			AnyEvent::MP tries to bind on port C<4040> on all IP addresses that the
363			local nodename resolves to.
364
365			Otherwise the first argument must be an array-reference with transport
366			endpoints ("ip:port", "hostname:port") or port numbers (in which case the
367			local nodename is used as hostname). The endpoints are all resolved and
368			will become the node reference.
369	root	1.2
370	root	1.8	=cut
371	root	1.1
372	root	1.4	=back
373
374			=head1 NODE MESSAGES
375
376	root	1.5	Nodes understand the following messages sent to them. Many of them take
377			arguments called C<@reply>, which will simply be used to compose a reply
378			message - C<$reply[0]> is the port to reply to, C<$reply[1]> the type and
379			the remaining arguments are simply the message data.
380	root	1.4
381			=over 4
382
383			=cut
384
385	root	1.8	=item wkp => $name, @reply
386	root	1.3
387	root	1.8	Replies with the port ID of the specified well-known port, or C<undef>.
388	root	1.3
389	root	1.7	=item devnull => ...
390
391			Generic data sink/CPU heat conversion.
392
393	root	1.4	=item relay => $port, @msg
394
395			Simply forwards the message to the given port.
396
397			=item eval => $string[ @reply]
398
399			Evaluates the given string. If C<@reply> is given, then a message of the
400	root	1.5	form C<@reply, $@, @evalres> is sent.
401
402			Example: crash another node.
403
404			snd $othernode, eval => "exit";
405	root	1.4
406			=item time => @reply
407
408			Replies the the current node time to C<@reply>.
409
410	root	1.5	Example: tell the current node to send the current time to C<$myport> in a
411			C<timereply> message.
412
413			snd $NODE, time => $myport, timereply => 1, 2;
414			# => snd $myport, timereply => 1, 2, <time>
415
416	root	1.2	=back
417
418	root	1.1	=head1 SEE ALSO
419
420			L<AnyEvent>.
421
422			=head1 AUTHOR
423
424			Marc Lehmann <schmorp@schmorp.de>
425			http://home.schmorp.de/
426
427			=cut
428
429			1
430