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 _any_
   create_port create_port_on
   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 $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 = create_miniport { }

Creates a "mini port", that is, a port without much #TODO

=cut

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

   $AnyEvent::MP::Base::PORT{$id} = sub {
      &$cb
         and delete $AnyEvent::MP::Base::PORT{$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;

=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) = @_;

   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.10
Committed:	Sun Aug 2 18:05:43 2009 UTC (14 years, 9 months ago) by root
Branch:	MAIN
Changes since 1.9:	+18 -0 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 _any_
90	create_port create_port_on
91	become_slave become_public
92	);
93
94	=item NODE / $NODE
95
96	The C<NODE ()> function and the C<$NODE> variable contain the noderef of
97	the local node. The value is initialised by a call to C<become_public> or
98	C<become_slave>, after which all local port identifiers become invalid.
99
100	=item snd $portid, type => @data
101
102	=item snd $portid, @msg
103
104	Send the given message to the given port ID, which can identify either
105	a local or a remote port, and can be either a string or soemthignt hat
106	stringifies a sa port ID (such as a port object :).
107
108	While the message can be about anything, it is highly recommended to use a
109	string as first element (a portid, or some word that indicates a request
110	type etc.).
111
112	The message data effectively becomes read-only after a call to this
113	function: modifying any argument is not allowed and can cause many
114	problems.
115
116	The type of data you can transfer depends on the transport protocol: when
117	JSON is used, then only strings, numbers and arrays and hashes consisting
118	of those are allowed (no objects). When Storable is used, then anything
119	that Storable can serialise and deserialise is allowed, and for the local
120	node, anything can be passed.
121
122	=item $local_port = create_port
123
124	Create a new local port object. See the next section for allowed methods.
125
126	=cut
127
128	sub create_port {
129	my $id = "$AnyEvent::MP::Base::UNIQ." . ++$AnyEvent::MP::Base::ID;
130
131	my $self = bless {
132	id => "$NODE#$id",
133	names => [$id],
134	}, "AnyEvent::MP::Port";
135
136	$AnyEvent::MP::Base::PORT{$id} = sub {
137	unshift @_, $self;
138
139	for (@{ $self->{rc0}{$_[1]} }) {
140	$_ && &{$_->[0]}
141	&& undef $_;
142	}
143
144	for (@{ $self->{rcv}{$_[1]} }) {
145	$_ && [@_[1 .. @{$_->[1]}]] ~~ $_->[1]
146	&& &{$_->[0]}
147	&& undef $_;
148	}
149
150	for (@{ $self->{any} }) {
151	$_ && [@_[0 .. $#{$_->[1]}]] ~~ $_->[1]
152	&& &{$_->[0]}
153	&& undef $_;
154	}
155	};
156
157	$self
158	}
159
160	=item $portid = create_miniport { }
161
162	Creates a "mini port", that is, a port without much #TODO
163
164	=cut
165
166	sub create_miniport(&) {
167	my $cb = shift;
168	my $id = "$AnyEvent::MP::Base::UNIQ." . ++$AnyEvent::MP::Base::ID;
169
170	$AnyEvent::MP::Base::PORT{$id} = sub {
171	&$cb
172	and delete $AnyEvent::MP::Base::PORT{$id};
173	};
174
175	"$NODE#$id"
176	}
177
178	package AnyEvent::MP::Port;
179
180	=back
181
182	=head1 METHODS FOR PORT OBJECTS
183
184	=over 4
185
186	=item "$port"
187
188	A port object stringifies to its port ID, so can be used directly for
189	C<snd> operations.
190
191	=cut
192
193	use overload
194	'""' => sub { $_[0]{id} },
195	fallback => 1;
196
197	=item $port->rcv (type => $callback->($port, @msg))
198
199	=item $port->rcv ($smartmatch => $callback->($port, @msg))
200
201	=item $port->rcv ([$smartmatch...] => $callback->($port, @msg))
202
203	Register a callback on the given port.
204
205	The callback has to return a true value when its work is done, after
206	which is will be removed, or a false value in which case it will stay
207	registered.
208
209	If the match is an array reference, then it will be matched against the
210	first elements of the message, otherwise only the first element is being
211	matched.
212
213	Any element in the match that is specified as C<_any_> (a function
214	exported by this module) matches any single element of the message.
215
216	While not required, it is highly recommended that the first matching
217	element is a string identifying the message. The one-string-only match is
218	also the most efficient match (by far).
219
220	=cut
221
222	sub rcv($@) {
223	my ($self, $match, $cb) = @_;
224
225	if (!ref $match) {
226	push @{ $self->{rc0}{$match} }, [$cb];
227	} elsif (("ARRAY" eq ref $match && !ref $match->[0])) {
228	my ($type, @match) = @$match;
229	@match
230	? push @{ $self->{rcv}{$match->[0]} }, [$cb, \@match]
231	: push @{ $self->{rc0}{$match->[0]} }, [$cb];
232	} else {
233	push @{ $self->{any} }, [$cb, $match];
234	}
235	}
236
237	=item $port->register ($name)
238
239	Registers the given port under the well known name C<$name>. If the name
240	already exists it is replaced.
241
242	A port can only be registered under one well known name.
243
244	=cut
245
246	sub register {
247	my ($self, $name) = @_;
248
249	$self->{wkname} = $name;
250	$AnyEvent::MP::Base::WKP{$name} = "$self";
251	}
252
253	=item $port->destroy
254
255	Explicitly destroy/remove/nuke/vaporise the port.
256
257	Ports are normally kept alive by there mere existance alone, and need to
258	be destroyed explicitly.
259
260	=cut
261
262	sub destroy {
263	my ($self) = @_;
264
265	delete $AnyEvent::MP::Base::WKP{ $self->{wkname} };
266
267	delete $AnyEvent::MP::Base::PORT{$_}
268	for @{ $self->{names} };
269	}
270
271	=back
272
273	=head1 FUNCTIONS FOR NODES
274
275	=over 4
276
277	=item mon $noderef, $callback->($noderef, $status, $)
278
279	Monitors the given noderef.
280
281	=item become_public endpoint...
282
283	Tells the node to become a public node, i.e. reachable from other nodes.
284
285	If no arguments are given, or the first argument is C<undef>, then
286	AnyEvent::MP tries to bind on port C<4040> on all IP addresses that the
287	local nodename resolves to.
288
289	Otherwise the first argument must be an array-reference with transport
290	endpoints ("ip:port", "hostname:port") or port numbers (in which case the
291	local nodename is used as hostname). The endpoints are all resolved and
292	will become the node reference.
293
294	=cut
295
296	=back
297
298	=head1 NODE MESSAGES
299
300	Nodes understand the following messages sent to them. Many of them take
301	arguments called C<@reply>, which will simply be used to compose a reply
302	message - C<$reply[0]> is the port to reply to, C<$reply[1]> the type and
303	the remaining arguments are simply the message data.
304
305	=over 4
306
307	=cut
308
309	=item wkp => $name, @reply
310
311	Replies with the port ID of the specified well-known port, or C<undef>.
312
313	=item devnull => ...
314
315	Generic data sink/CPU heat conversion.
316
317	=item relay => $port, @msg
318
319	Simply forwards the message to the given port.
320
321	=item eval => $string[ @reply]
322
323	Evaluates the given string. If C<@reply> is given, then a message of the
324	form C<@reply, $@, @evalres> is sent.
325
326	Example: crash another node.
327
328	snd $othernode, eval => "exit";
329
330	=item time => @reply
331
332	Replies the the current node time to C<@reply>.
333
334	Example: tell the current node to send the current time to C<$myport> in a
335	C<timereply> message.
336
337	snd $NODE, time => $myport, timereply => 1, 2;
338	# => snd $myport, timereply => 1, 2, <time>
339
340	=back
341
342	=head1 SEE ALSO
343
344	L<AnyEvent>.
345
346	=head1 AUTHOR
347
348	Marc Lehmann <schmorp@schmorp.de>
349	http://home.schmorp.de/
350
351	=cut
352
353	1
354