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 resreve 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 del _any_
   create_port create_port_on
   create_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->()

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) = split /#/, shift, 2;

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

   my $cb = shift;

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

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

=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",
      names => [$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 del $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 there mere existance alone, and need to
be destroyed explicitly.

=cut

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

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

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

   delete $AnyEvent::MP::Base::PORT{$_}
      for @{ $self->{names} };
}

=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.20
Committed:	Mon Aug 3 22:05:55 2009 UTC (14 years, 9 months ago) by root
Branch:	MAIN
Changes since 1.19:	+5 -2 lines
Log Message:	* empty log message *
#	Content
1	=head1 NAME
2
3	AnyEvent::MP - multi-processing/message-passing framework
4
5	=head1 SYNOPSIS
6
7	use AnyEvent::MP;
8
9	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	=head1 DESCRIPTION
26
27	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	At the moment, this module family is severly brokena nd underdocumented,
33	so do not use. This was uploaded mainly to resreve the CPAN namespace -
34	stay tuned!
35
36	=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	=item port id - C<noderef#portname>
47
48	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
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	(connected to a master node only). Only when they epxlicitly "become
59	public" can you send them messages from unrelated other nodes.
60
61	=item noderef - C<host:port,host:port...>, C<id@noderef>, C<id>
62
63	A noderef is a string that either uniquely identifies a given node (for
64	private and hidden nodes), or contains a recipe on how to reach a given
65	node (for public nodes).
66
67	=back
68
69	=head1 VARIABLES/FUNCTIONS
70
71	=over 4
72
73	=cut
74
75	package AnyEvent::MP;
76
77	use AnyEvent::MP::Base;
78
79	use common::sense;
80
81	use Carp ();
82
83	use AE ();
84
85	use base "Exporter";
86
87	our $VERSION = '0.02';
88	our @EXPORT = qw(
89	NODE $NODE $PORT snd rcv mon del _any_
90	create_port create_port_on
91	create_miniport
92	become_slave become_public
93	);
94
95	=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	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
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	=item $guard = mon $portid, $cb->()
124
125	Monitor the given port and call the given callback when the port is
126	destroyed or connection to it's node is lost.
127
128	#TODO
129
130	=cut
131
132	sub mon {
133	my ($noderef, $port) = split /#/, shift, 2;
134
135	my $node = AnyEvent::MP::Base::add_node $noderef;
136
137	my $cb = shift;
138
139	$node->monitor ($port, $cb);
140
141	defined wantarray
142	and AnyEvent::Util::guard { $node->unmonitor ($port, $cb) }
143	}
144
145	=item $local_port = create_port
146
147	Create a new local port object. See the next section for allowed methods.
148
149	=cut
150
151	sub create_port {
152	my $id = "$AnyEvent::MP::Base::UNIQ." . $AnyEvent::MP::Base::ID++;
153
154	my $self = bless {
155	id => "$NODE#$id",
156	names => [$id],
157	}, "AnyEvent::MP::Port";
158
159	$AnyEvent::MP::Base::PORT{$id} = sub {
160	unshift @_, $self;
161
162	for (@{ $self->{rc0}{$_[1]} }) {
163	$_ && &{$_->[0]}
164	&& undef $_;
165	}
166
167	for (@{ $self->{rcv}{$_[1]} }) {
168	$_ && [@_[1 .. @{$_->[1]}]] ~~ $_->[1]
169	&& &{$_->[0]}
170	&& undef $_;
171	}
172
173	for (@{ $self->{any} }) {
174	$_ && [@_[0 .. $#{$_->[1]}]] ~~ $_->[1]
175	&& &{$_->[0]}
176	&& undef $_;
177	}
178	};
179
180	$self
181	}
182
183	=item $portid = miniport { my @msg = @_; $finished }
184
185	Creates a "mini port", that is, a very lightweight port without any
186	pattern matching behind it, and returns its ID.
187
188	The block will be called for every message received on the port. When the
189	callback returns a true value its job is considered "done" and the port
190	will be destroyed. Otherwise it will stay alive.
191
192	The message will be passed as-is, no extra argument (i.e. no port id) will
193	be passed to the callback.
194
195	If you need the local port id in the callback, this works nicely:
196
197	my $port; $port = miniport {
198	snd $otherport, reply => $port;
199	};
200
201	=cut
202
203	sub miniport(&) {
204	my $cb = shift;
205	my $id = "$AnyEvent::MP::Base::UNIQ." . $AnyEvent::MP::Base::ID++;
206
207	$AnyEvent::MP::Base::PORT{$id} = sub {
208	&$cb
209	and del $id;
210	};
211
212	"$NODE#$id"
213	}
214
215	package AnyEvent::MP::Port;
216
217	=back
218
219	=head1 METHODS FOR PORT OBJECTS
220
221	=over 4
222
223	=item "$port"
224
225	A port object stringifies to its port ID, so can be used directly for
226	C<snd> operations.
227
228	=cut
229
230	use overload
231	'""' => sub { $_[0]{id} },
232	fallback => 1;
233
234	sub TO_JSON { $_[0]{id} }
235
236	=item $port->rcv (type => $callback->($port, @msg))
237
238	=item $port->rcv ($smartmatch => $callback->($port, @msg))
239
240	=item $port->rcv ([$smartmatch...] => $callback->($port, @msg))
241
242	Register a callback on the given port.
243
244	The callback has to return a true value when its work is done, after
245	which is will be removed, or a false value in which case it will stay
246	registered.
247
248	If the match is an array reference, then it will be matched against the
249	first elements of the message, otherwise only the first element is being
250	matched.
251
252	Any element in the match that is specified as C<_any_> (a function
253	exported by this module) matches any single element of the message.
254
255	While not required, it is highly recommended that the first matching
256	element is a string identifying the message. The one-string-only match is
257	also the most efficient match (by far).
258
259	=cut
260
261	sub rcv($@) {
262	my ($self, $match, $cb) = @_;
263
264	if (!ref $match) {
265	push @{ $self->{rc0}{$match} }, [$cb];
266	} elsif (("ARRAY" eq ref $match && !ref $match->[0])) {
267	my ($type, @match) = @$match;
268	@match
269	? push @{ $self->{rcv}{$match->[0]} }, [$cb, \@match]
270	: push @{ $self->{rc0}{$match->[0]} }, [$cb];
271	} else {
272	push @{ $self->{any} }, [$cb, $match];
273	}
274	}
275
276	=item $port->register ($name)
277
278	Registers the given port under the well known name C<$name>. If the name
279	already exists it is replaced.
280
281	A port can only be registered under one well known name.
282
283	=cut
284
285	sub register {
286	my ($self, $name) = @_;
287
288	$self->{wkname} = $name;
289	$AnyEvent::MP::Base::WKP{$name} = "$self";
290	}
291
292	=item $port->destroy
293
294	Explicitly destroy/remove/nuke/vaporise the port.
295
296	Ports are normally kept alive by there mere existance alone, and need to
297	be destroyed explicitly.
298
299	=cut
300
301	sub destroy {
302	my ($self) = @_;
303
304	AnyEvent::MP::Base::del $self->{id};
305
306	delete $AnyEvent::MP::Base::WKP{ $self->{wkname} };
307
308	delete $AnyEvent::MP::Base::PORT{$_}
309	for @{ $self->{names} };
310	}
311
312	=back
313
314	=head1 FUNCTIONS FOR NODES
315
316	=over 4
317
318	=item mon $noderef, $callback->($noderef, $status, $)
319
320	Monitors the given noderef.
321
322	=item become_public endpoint...
323
324	Tells the node to become a public node, i.e. reachable from other nodes.
325
326	If no arguments are given, or the first argument is C<undef>, then
327	AnyEvent::MP tries to bind on port C<4040> on all IP addresses that the
328	local nodename resolves to.
329
330	Otherwise the first argument must be an array-reference with transport
331	endpoints ("ip:port", "hostname:port") or port numbers (in which case the
332	local nodename is used as hostname). The endpoints are all resolved and
333	will become the node reference.
334
335	=cut
336
337	=back
338
339	=head1 NODE MESSAGES
340
341	Nodes understand the following messages sent to them. Many of them take
342	arguments called C<@reply>, which will simply be used to compose a reply
343	message - C<$reply[0]> is the port to reply to, C<$reply[1]> the type and
344	the remaining arguments are simply the message data.
345
346	=over 4
347
348	=cut
349
350	=item wkp => $name, @reply
351
352	Replies with the port ID of the specified well-known port, or C<undef>.
353
354	=item devnull => ...
355
356	Generic data sink/CPU heat conversion.
357
358	=item relay => $port, @msg
359
360	Simply forwards the message to the given port.
361
362	=item eval => $string[ @reply]
363
364	Evaluates the given string. If C<@reply> is given, then a message of the
365	form C<@reply, $@, @evalres> is sent.
366
367	Example: crash another node.
368
369	snd $othernode, eval => "exit";
370
371	=item time => @reply
372
373	Replies the the current node time to C<@reply>.
374
375	Example: tell the current node to send the current time to C<$myport> in a
376	C<timereply> message.
377
378	snd $NODE, time => $myport, timereply => 1, 2;
379	# => snd $myport, timereply => 1, 2, <time>
380
381	=back
382
383	=head1 SEE ALSO
384
385	L<AnyEvent>.
386
387	=head1 AUTHOR
388
389	Marc Lehmann <schmorp@schmorp.de>
390	http://home.schmorp.de/
391
392	=cut
393
394	1
395