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.01';
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
}

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.8
Committed:	Sun Aug 2 14:44:37 2009 UTC (17 years, 1 month ago) by root
Branch:	MAIN
Changes since 1.7:	+114 -242 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			so do not use. This was uploaded mainly to resreve the CPAN namespace -
34			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.6	our $VERSION = '0.01';
88	root	1.8	our @EXPORT = qw(
89			NODE $NODE $PORT snd rcv _any_
90			create_port create_port_on
91			become_slave become_public
92			);
93	root	1.2
94	root	1.3	=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	root	1.8	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	root	1.3
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	root	1.8	=item $local_port = create_port
123
124			Create a new local port object. See the next section for allowed methods.
125
126	root	1.3	=cut
127
128	root	1.8	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	root	1.2
150	root	1.8	for (@{ $self->{any} }) {
151			$_ && [@_[0 .. $#{$_->[1]}]] ~~ $_->[1]
152			&& &{$_->[0]}
153			&& undef $_;
154			}
155			};
156	root	1.2
157	root	1.8	$self
158	root	1.3	}
159
160	root	1.8	package AnyEvent::MP::Port;
161
162			=back
163
164			=head1 METHODS FOR PORT OBJECTS
165
166			=over 4
167
168			=item "$port"
169
170			A port object stringifies to its port ID, so can be used directly for
171			C<snd> operations.
172
173			=cut
174
175			use overload
176			'""' => sub { $_[0]{id} },
177			fallback => 1;
178
179			=item $port->rcv (type => $callback->($port, @msg))
180	root	1.3
181	root	1.8	=item $port->rcv ($smartmatch => $callback->($port, @msg))
182	root	1.3
183	root	1.8	=item $port->rcv ([$smartmatch...] => $callback->($port, @msg))
184	root	1.3
185	root	1.8	Register a callback on the given port.
186	root	1.3
187			The callback has to return a true value when its work is done, after
188			which is will be removed, or a false value in which case it will stay
189			registered.
190
191			If the match is an array reference, then it will be matched against the
192			first elements of the message, otherwise only the first element is being
193			matched.
194
195			Any element in the match that is specified as C<_any_> (a function
196			exported by this module) matches any single element of the message.
197
198			While not required, it is highly recommended that the first matching
199			element is a string identifying the message. The one-string-only match is
200			also the most efficient match (by far).
201
202			=cut
203
204			sub rcv($@) {
205	root	1.8	my ($self, $match, $cb) = @_;
206	root	1.3
207			if (!ref $match) {
208	root	1.8	push @{ $self->{rc0}{$match} }, [$cb];
209	root	1.3	} elsif (("ARRAY" eq ref $match && !ref $match->[0])) {
210			my ($type, @match) = @$match;
211			@match
212	root	1.8	? push @{ $self->{rcv}{$match->[0]} }, [$cb, \@match]
213			: push @{ $self->{rc0}{$match->[0]} }, [$cb];
214	root	1.3	} else {
215	root	1.8	push @{ $self->{any} }, [$cb, $match];
216	root	1.3	}
217	root	1.2	}
218
219	root	1.8	=item $port->register ($name)
220	root	1.2
221	root	1.8	Registers the given port under the well known name C<$name>. If the name
222			already exists it is replaced.
223	root	1.2
224	root	1.8	A port can only be registered under one well known name.
225	root	1.3
226	root	1.8	=cut
227	root	1.3
228	root	1.8	sub register {
229			my ($self, $name) = @_;
230	root	1.3
231	root	1.8	$self->{wkname} = $name;
232			$AnyEvent::MP::Base::WKP{$name} = "$self";
233	root	1.1	}
234
235	root	1.8	=item $port->destroy
236	root	1.2
237	root	1.8	Explicitly destroy/remove/nuke/vaporise the port.
238	root	1.2
239	root	1.8	Ports are normally kept alive by there mere existance alone, and need to
240			be destroyed explicitly.
241	root	1.2
242	root	1.8	=cut
243	root	1.1
244	root	1.8	sub destroy {
245			my ($self) = @_;
246	root	1.1
247	root	1.8	delete $AnyEvent::MP::Base::WKP{ $self->{wkname} };
248	root	1.2
249	root	1.8	delete $AnyEvent::MP::Base::PORT{$_}
250			for @{ $self->{names} };
251	root	1.2	}
252
253	root	1.8	=back
254
255			=head1 FUNCTIONS FOR NODES
256
257			=over 4
258	root	1.2
259	root	1.8	=item mon $noderef, $callback->($noderef, $status, $)
260	root	1.2
261	root	1.8	Monitors the given noderef.
262	root	1.2
263	root	1.8	=item become_public endpoint...
264
265			Tells the node to become a public node, i.e. reachable from other nodes.
266
267			If no arguments are given, or the first argument is C<undef>, then
268			AnyEvent::MP tries to bind on port C<4040> on all IP addresses that the
269			local nodename resolves to.
270
271			Otherwise the first argument must be an array-reference with transport
272			endpoints ("ip:port", "hostname:port") or port numbers (in which case the
273			local nodename is used as hostname). The endpoints are all resolved and
274			will become the node reference.
275	root	1.2
276	root	1.8	=cut
277	root	1.1
278	root	1.4	=back
279
280			=head1 NODE MESSAGES
281
282	root	1.5	Nodes understand the following messages sent to them. Many of them take
283			arguments called C<@reply>, which will simply be used to compose a reply
284			message - C<$reply[0]> is the port to reply to, C<$reply[1]> the type and
285			the remaining arguments are simply the message data.
286	root	1.4
287			=over 4
288
289			=cut
290
291	root	1.8	=item wkp => $name, @reply
292	root	1.3
293	root	1.8	Replies with the port ID of the specified well-known port, or C<undef>.
294	root	1.3
295	root	1.7	=item devnull => ...
296
297			Generic data sink/CPU heat conversion.
298
299	root	1.4	=item relay => $port, @msg
300
301			Simply forwards the message to the given port.
302
303			=item eval => $string[ @reply]
304
305			Evaluates the given string. If C<@reply> is given, then a message of the
306	root	1.5	form C<@reply, $@, @evalres> is sent.
307
308			Example: crash another node.
309
310			snd $othernode, eval => "exit";
311	root	1.4
312			=item time => @reply
313
314			Replies the the current node time to C<@reply>.
315
316	root	1.5	Example: tell the current node to send the current time to C<$myport> in a
317			C<timereply> message.
318
319			snd $NODE, time => $myport, timereply => 1, 2;
320			# => snd $myport, timereply => 1, 2, <time>
321
322	root	1.2	=back
323
324	root	1.1	=head1 SEE ALSO
325
326			L<AnyEvent>.
327
328			=head1 AUTHOR
329
330			Marc Lehmann <schmorp@schmorp.de>
331			http://home.schmorp.de/
332
333			=cut
334
335			1
336