[ViewVC] Diff of: cvs/AnyEvent-MP/MP.pm

Comparing AnyEvent-MP/MP.pm (file contents):
Revision 1.2 by root, Fri Jul 31 20:55:46 2009 UTC vs.
Revision 1.15 by root, Sun Aug 2 19:25:27 2009 UTC

…		…
27	This module (-family) implements a simple message passing framework.	27	This module (-family) implements a simple message passing framework.
28		28
29	Despite its simplicity, you can securely message other processes running	29	Despite its simplicity, you can securely message other processes running
30	on the same or other hosts.	30	on the same or other hosts.
31		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
32	=head1 CONCEPTS	36	=head1 CONCEPTS
33		37
34	=over 4	38	=over 4
35		39
36	=item port	40	=item port
37		41
38	A port is something you can send messages to with the C<snd> function, and	42	A port is something you can send messages to with the C<snd> function, and
39	you can register C<rcv> handlers with. All C<rcv> handlers will receive	43	you can register C<rcv> handlers with. All C<rcv> handlers will receive
40	messages they match, messages will not be queued.	44	messages they match, messages will not be queued.
41		45
42	=item port id - C<pid@host#portname>	46	=item port id - C<noderef#portname>
43		47
44	A port id is always the node id, a hash-mark (C<#>) as separator, followed	48	A port id is always the noderef, a hash-mark (C<#>) as separator, followed
45	by a port name.	49	by a port name (a printable string of unspecified format).
46
47	A port name can be a well known port (basically an identifier/bareword),
48	or a generated name, consisting of node id, a dot (C<.>), and an
49	identifier.
50		50
51	=item node	51	=item node
52		52
53	A node is a single process containing at least one port - the node	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,	54	port. You can send messages to node ports to let them create new ports,
55	among other things.	55	among other things.
56		56
57	Initially, nodes are either private (single-process only) or hidden	57	Initially, nodes are either private (single-process only) or hidden
58	(connected to a father node only). Only when they epxlicitly "go public"	58	(connected to a master node only). Only when they epxlicitly "become
59	can you send them messages form unrelated other nodes.	59	public" can you send them messages from unrelated other nodes.
60		60
61	Public nodes automatically connect to all other public nodes in a network	61	=item noderef - C<host:port,host:port...>, C<id@noderef>, C<id>
62	when they connect, creating a full mesh.
63		62
64	=item node id - C<host:port>, C<id@host>, C<id>
65
66	A node ID is a string that either uniquely identifies a given node (For	63	A noderef is a string that either uniquely identifies a given node (for
67	private and hidden nodes), or contains a recipe on how to reach a given	64	private and hidden nodes), or contains a recipe on how to reach a given
68	node (for public nodes).	65	node (for public nodes).
69		66
70	=back	67	=back
71		68
72	=head1 FUNCTIONS	69	=head1 VARIABLES/FUNCTIONS
73		70
74	=over 4	71	=over 4
75		72
76	=cut	73	=cut
77		74
78	package AnyEvent::MP;	75	package AnyEvent::MP;
79		76
80	use AnyEvent::MP::Util ();
81	use AnyEvent::MP::Node;	77	use AnyEvent::MP::Base;
82	use AnyEvent::MP::Transport;
83		78
84	use utf8;
85	use common::sense;	79	use common::sense;
86		80
87	use Carp ();	81	use Carp ();
88		82
89	use AE ();	83	use AE ();
90		84
91	use base "Exporter";	85	use base "Exporter";
92		86
93	our $VERSION = '0.0';	87	our $VERSION = '0.02';
94	our @EXPORT = qw(NODE $NODE $PORT snd rcv _any_);	88	our @EXPORT = qw(
		89	NODE $NODE $PORT snd rcv _any_
		90	create_port create_port_on
		91	create_miniport
		92	become_slave become_public
		93	);
95		94
96	our $DEFAULT_SECRET;	95	=item NODE / $NODE
97	our $DEFAULT_PORT = "4040";
98		96
99	our $CONNECT_INTERVAL = 5; # new connect every 5s, at least	97	The C<NODE ()> function and the C<$NODE> variable contain the noderef of
100	our $CONNECT_TIMEOUT = 30; # includes handshake	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.
101		100
102	sub default_secret {	101	=item snd $portid, type => @data
103	unless (defined $DEFAULT_SECRET) {	102
104	if (open my $fh, "<$ENV{HOME}/.aemp-secret") {	103	=item snd $portid, @msg
105	sysread $fh, $DEFAULT_SECRET, -s $fh;	104
106	} else {	105	Send the given message to the given port ID, which can identify either
107	$DEFAULT_SECRET = AnyEvent::MP::Util::nonce 32;	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 $local_port = create_port
		124
		125	Create a new local port object. See the next section for allowed methods.
		126
		127	=cut
		128
		129	sub create_port {
		130	my $id = "$AnyEvent::MP::Base::UNIQ." . ++$AnyEvent::MP::Base::ID;
		131
		132	my $self = bless {
		133	id => "$NODE#$id",
		134	names => [$id],
		135	}, "AnyEvent::MP::Port";
		136
		137	$AnyEvent::MP::Base::PORT{$id} = sub {
		138	unshift @_, $self;
		139
		140	for (@{ $self->{rc0}{$_[1]} }) {
		141	$_ && &{$_->[0]}
		142	&& undef $_;
108	}	143	}
		144
		145	for (@{ $self->{rcv}{$_[1]} }) {
		146	$_ && [@_[1 .. @{$_->[1]}]] ~~ $_->[1]
		147	&& &{$_->[0]}
		148	&& undef $_;
		149	}
		150
		151	for (@{ $self->{any} }) {
		152	$_ && [@_[0 .. $#{$_->[1]}]] ~~ $_->[1]
		153	&& &{$_->[0]}
		154	&& undef $_;
		155	}
		156	};
		157
		158	$self
		159	}
		160
		161	=item $portid = miniport { my @msg = @_; $finished }
		162
		163	Creates a "mini port", that is, a very lightweight port without any
		164	pattern matching behind it, and returns its ID.
		165
		166	The block will be called for every message received on the port. When the
		167	callback returns a true value its job is considered "done" and the port
		168	will be destroyed. Otherwise it will stay alive.
		169
		170	The message will be passed as-is, no extra argument (ie.. no port id) will
		171	be passed to the callback.
		172
		173	If you need the local port id in the callback, this works nicely:
		174
		175	my $port; $port = miniport {
		176	snd $otherport, reply => $port;
		177	};
		178
		179	=cut
		180
		181	sub miniport(&) {
		182	my $cb = shift;
		183	my $id = "$AnyEvent::MP::Base::UNIQ." . ++$AnyEvent::MP::Base::ID;
		184
		185	$AnyEvent::MP::Base::PORT{$id} = sub {
		186	&$cb
		187	and delete $AnyEvent::MP::Base::PORT{$id};
		188	};
		189
		190	"$NODE#$id"
		191	}
		192
		193	package AnyEvent::MP::Port;
		194
		195	=back
		196
		197	=head1 METHODS FOR PORT OBJECTS
		198
		199	=over 4
		200
		201	=item "$port"
		202
		203	A port object stringifies to its port ID, so can be used directly for
		204	C<snd> operations.
		205
		206	=cut
		207
		208	use overload
		209	'""' => sub { $_[0]{id} },
		210	fallback => 1;
		211
		212	=item $port->rcv (type => $callback->($port, @msg))
		213
		214	=item $port->rcv ($smartmatch => $callback->($port, @msg))
		215
		216	=item $port->rcv ([$smartmatch...] => $callback->($port, @msg))
		217
		218	Register a callback on the given port.
		219
		220	The callback has to return a true value when its work is done, after
		221	which is will be removed, or a false value in which case it will stay
		222	registered.
		223
		224	If the match is an array reference, then it will be matched against the
		225	first elements of the message, otherwise only the first element is being
		226	matched.
		227
		228	Any element in the match that is specified as C<_any_> (a function
		229	exported by this module) matches any single element of the message.
		230
		231	While not required, it is highly recommended that the first matching
		232	element is a string identifying the message. The one-string-only match is
		233	also the most efficient match (by far).
		234
		235	=cut
		236
		237	sub rcv($@) {
		238	my ($self, $match, $cb) = @_;
		239
		240	if (!ref $match) {
		241	push @{ $self->{rc0}{$match} }, [$cb];
		242	} elsif (("ARRAY" eq ref $match && !ref $match->[0])) {
		243	my ($type, @match) = @$match;
		244	@match
		245	? push @{ $self->{rcv}{$match->[0]} }, [$cb, \@match]
		246	: push @{ $self->{rc0}{$match->[0]} }, [$cb];
		247	} else {
		248	push @{ $self->{any} }, [$cb, $match];
109	}	249	}
110
111	$DEFAULT_SECRET
112	}	250	}
113		251
114	our $UNIQ = sprintf "%x.%x", $$, time; # per-process/node unique cookie	252	=item $port->register ($name)
115	our $PUBLIC = 0;
116	our $NODE;
117	our $PORT;
118		253
119	our %NODE; # node id to transport mapping, or "undef", for local node	254	Registers the given port under the well known name C<$name>. If the name
120	our %PORT; # local ports	255	already exists it is replaced.
121	our %LISTENER; # local transports
122		256
123	sub NODE() { $NODE }	257	A port can only be registered under one well known name.
124		258
125	{	259	=cut
126	use POSIX ();
127	my $nodename = (POSIX::uname)[1];
128	$NODE = "$$\@$nodename";
129	}
130		260
131	sub _ANY_() { 1 }	261	sub register {
132	sub _any_() { \&_ANY_ }	262	my ($self, $name) = @_;
133		263
134	sub add_node {	264	$self->{wkname} = $name;
		265	$AnyEvent::MP::Base::WKP{$name} = "$self";
		266	}
		267
		268	=item $port->destroy
		269
		270	Explicitly destroy/remove/nuke/vaporise the port.
		271
		272	Ports are normally kept alive by there mere existance alone, and need to
		273	be destroyed explicitly.
		274
		275	=cut
		276
		277	sub destroy {
135	my ($noderef) = @_;	278	my ($self) = @_;
136		279
137	return $NODE{$noderef}	280	delete $AnyEvent::MP::Base::WKP{ $self->{wkname} };
138	if exists $NODE{$noderef};
139		281
140	for (split /,/, $noderef) {	282	delete $AnyEvent::MP::Base::PORT{$_}
141	return $NODE{$noderef} = $NODE{$_}	283	for @{ $self->{names} };
142	if exists $NODE{$_};
143	}
144
145	# for indirect sends, use a different class
146	my $node = new AnyEvent::MP::Node::Direct $noderef;
147
148	$NODE{$_} = $node
149	for $noderef, split /,/, $noderef;
150
151	$node
152	}	284	}
153		285
154	sub snd($@) {	286	=back
155	my ($noderef, $port) = split /#/, shift, 2;
156		287
157	add_node $noderef	288	=head1 FUNCTIONS FOR NODES
158	unless exists $NODE{$noderef};
159		289
160	$NODE{$noderef}->send ([$port, [@_]]);	290	=over 4
161	}
162		291
163	sub _inject {	292	=item mon $noderef, $callback->($noderef, $status, $)
164	my ($port, $msg) = @{+shift};
165		293
166	$port = $PORT{$port}	294	Monitors the given noderef.
167	or return;
168		295
169	use Data::Dumper;	296	=item become_public endpoint...
170	warn Dumper $msg;
171	}
172		297
173	sub normalise_noderef($) {	298	Tells the node to become a public node, i.e. reachable from other nodes.
174	my ($noderef) = @_;
175		299
176	my $cv = AE::cv;	300	If no arguments are given, or the first argument is C<undef>, then
177	my @res;	301	AnyEvent::MP tries to bind on port C<4040> on all IP addresses that the
		302	local nodename resolves to.
178		303
179	$cv->begin (sub {	304	Otherwise the first argument must be an array-reference with transport
180	my %seen;	305	endpoints ("ip:port", "hostname:port") or port numbers (in which case the
181	my @refs;	306	local nodename is used as hostname). The endpoints are all resolved and
182	for (sort { $a->[0] <=> $b->[0] } @res) {	307	will become the node reference.
183	push @refs, $_->[1] unless $seen{$_->[1]}++
184	}
185	shift->send (join ",", @refs);
186	});
187		308
188	$noderef = $DEFAULT_PORT unless length $noderef;	309	=cut
189		310
190	my $idx;	311	=back
191	for my $t (split /,/, $noderef) {
192	my $pri = ++$idx;
193
194	#TODO: this should be outside normalise_noderef and in become_public
195	if ($t =~ /^\d*$/) {
196	my $nodename = (POSIX::uname)[1];
197		312
198	$cv->begin;	313	=head1 NODE MESSAGES
199	AnyEvent::Socket::resolve_sockaddr $nodename, $t \|\| "aemp=$DEFAULT_PORT", "tcp", 0, undef, sub {
200	for (@_) {
201	my ($service, $host) = AnyEvent::Socket::unpack_sockaddr $_->[3];
202	push @res, [
203	$pri += 1e-5,
204	AnyEvent::Socket::format_hostport AnyEvent::Socket::format_address $host, $service
205	];
206	}
207	$cv->end;
208	};
209		314
210	# my (undef, undef, undef, undef, @ipv4) = gethostbyname $nodename;	315	Nodes understand the following messages sent to them. Many of them take
211	#	316	arguments called C<@reply>, which will simply be used to compose a reply
212	# for (@ipv4) {	317	message - C<$reply[0]> is the port to reply to, C<$reply[1]> the type and
213	# push @res, [	318	the remaining arguments are simply the message data.
214	# $pri,
215	# AnyEvent::Socket::format_hostport AnyEvent::Socket::format_address $_, $t \|\| $DEFAULT_PORT,
216	# ];
217	# }
218	} else {
219	my ($host, $port) = AnyEvent::Socket::parse_hostport $t, "aemp=$DEFAULT_PORT"
220	or Carp::croak "$t: unparsable transport descriptor";
221		319
222	$cv->begin;	320	=over 4
223	AnyEvent::Socket::resolve_sockaddr $host, $port, "tcp", 0, undef, sub {
224	for (@_) {
225	my ($service, $host) = AnyEvent::Socket::unpack_sockaddr $_->[3];
226	push @res, [
227	$pri += 1e-5,
228	AnyEvent::Socket::format_hostport AnyEvent::Socket::format_address $host, $service
229	];
230	}
231	$cv->end;
232	}
233	}
234	}
235		321
236	$cv->end;	322	=cut
237		323
238	$cv	324	=item wkp => $name, @reply
239	}
240		325
241	sub become_public {	326	Replies with the port ID of the specified well-known port, or C<undef>.
242	return if $PUBLIC;
243		327
244	my $noderef = join ",", ref $_[0] ? @{+shift} : shift;	328	=item devnull => ...
245	my @args = @_;
246		329
247	$NODE = (normalise_noderef $noderef)->recv;	330	Generic data sink/CPU heat conversion.
248		331
249	my $self = new AnyEvent::MP::Node::Self noderef => $NODE;	332	=item relay => $port, @msg
250		333
251	$NODE{""} = $self; # empty string == local node	334	Simply forwards the message to the given port.
252		335
253	for my $t (split /,/, $NODE) {	336	=item eval => $string[ @reply]
254	$NODE{$t} = $self;
255		337
256	my ($host, $port) = AnyEvent::Socket::parse_hostport $t;	338	Evaluates the given string. If C<@reply> is given, then a message of the
		339	form C<@reply, $@, @evalres> is sent.
257		340
258	$LISTENER{$t} = AnyEvent::MP::Transport::mp_server $host, $port,	341	Example: crash another node.
259	@args,
260	on_error => sub {
261	die "on_error<@_>\n";#d#
262	},
263	on_connect => sub {
264	my ($tp) = @_;
265		342
266	$NODE{$tp->{remote_id}} = $_[0];	343	snd $othernode, eval => "exit";
267	},
268	sub {
269	my ($tp) = @_;
270		344
271	$NODE{"$tp->{peerhost}:$tp->{peerport}"} = $tp;	345	=item time => @reply
272	},
273	;
274	}
275		346
276	$PUBLIC = 1;	347	Replies the the current node time to C<@reply>.
277	}	348
		349	Example: tell the current node to send the current time to C<$myport> in a
		350	C<timereply> message.
		351
		352	snd $NODE, time => $myport, timereply => 1, 2;
		353	# => snd $myport, timereply => 1, 2, <time>
278		354
279	=back	355	=back
280		356
281	=head1 SEE ALSO	357	=head1 SEE ALSO
282		358

Diff Legend

-–
+Removed lines
-+
+Added lines
-<
+Changed lines
->
+Changed lines

Comparing AnyEvent-MP/MP.pm (file contents): Revision 1.2 by root, Fri Jul 31 20:55:46 2009 UTC vs. Revision 1.15 by root, Sun Aug 2 19:25:27 2009 UTC

Diff Legend

Comparing AnyEvent-MP/MP.pm (file contents):
Revision 1.2 by root, Fri Jul 31 20:55:46 2009 UTC vs.
Revision 1.15 by root, Sun Aug 2 19:25:27 2009 UTC