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

Comparing AnyEvent-MP/MP.pm (file contents):
Revision 1.6 by root, Sat Aug 1 10:02:33 2009 UTC vs.
Revision 1.17 by root, Mon Aug 3 08:35:40 2009 UTC

…		…
72		72
73	=cut	73	=cut
74		74
75	package AnyEvent::MP;	75	package AnyEvent::MP;
76		76
77	use AnyEvent::MP::Util ();
78	use AnyEvent::MP::Node;	77	use AnyEvent::MP::Base;
79	use AnyEvent::MP::Transport;
80		78
81	use utf8;
82	use common::sense;	79	use common::sense;
83		80
84	use Carp ();	81	use Carp ();
85		82
86	use AE ();	83	use AE ();
87		84
88	use base "Exporter";	85	use base "Exporter";
89		86
90	our $VERSION = '0.01';	87	our $VERSION = '0.02';
91	our @EXPORT = qw(NODE $NODE $PORT snd rcv _any_);	88	our @EXPORT = qw(
92		89	NODE $NODE $PORT snd rcv _any_
93	our $DEFAULT_SECRET;	90	create_port create_port_on
94	our $DEFAULT_PORT = "4040";	91	create_miniport
95		92	become_slave become_public
96	our $CONNECT_INTERVAL = 5; # new connect every 5s, at least	93	);
97	our $CONNECT_TIMEOUT = 30; # includes handshake
98
99	sub default_secret {
100	unless (defined $DEFAULT_SECRET) {
101	if (open my $fh, "<$ENV{HOME}/.aemp-secret") {
102	sysread $fh, $DEFAULT_SECRET, -s $fh;
103	} else {
104	$DEFAULT_SECRET = AnyEvent::MP::Util::nonce 32;
105	}
106	}
107
108	$DEFAULT_SECRET
109	}
110		94
111	=item NODE / $NODE	95	=item NODE / $NODE
112		96
113	The C<NODE ()> function and the C<$NODE> variable contain the noderef of	97	The C<NODE ()> function and the C<$NODE> variable contain the noderef of
114	the local node. The value is initialised by a call to C<become_public> or	98	the local node. The value is initialised by a call to C<become_public> or
115	C<become_slave>, after which all local port identifiers become invalid.	99	C<become_slave>, after which all local port identifiers become invalid.
116		100
117	=cut
118
119	our $UNIQ = sprintf "%x.%x", $$, time; # per-process/node unique cookie
120	our $ID = "a0";
121	our $PUBLIC = 0;
122	our $NODE;
123	our $PORT;
124
125	our %NODE; # node id to transport mapping, or "undef", for local node
126	our %PORT; # local ports
127	our %LISTENER; # local transports
128
129	sub NODE() { $NODE }
130
131	{
132	use POSIX ();
133	my $nodename = (POSIX::uname)[1];
134	$NODE = "$$\@$nodename";
135	}
136
137	sub _ANY_() { 1 }
138	sub _any_() { \&_ANY_ }
139
140	sub add_node {
141	my ($noderef) = @_;
142
143	return $NODE{$noderef}
144	if exists $NODE{$noderef};
145
146	for (split /,/, $noderef) {
147	return $NODE{$noderef} = $NODE{$_}
148	if exists $NODE{$_};
149	}
150
151	# for indirect sends, use a different class
152	my $node = new AnyEvent::MP::Node::Direct $noderef;
153
154	$NODE{$_} = $node
155	for $noderef, split /,/, $noderef;
156
157	$node
158	}
159
160	=item snd $portid, type => @data	101	=item snd $portid, type => @data
161		102
162	=item snd $portid, @msg	103	=item snd $portid, @msg
163		104
164	Send the given message to the given port ID, which can identify either a	105	Send the given message to the given port ID, which can identify either
165	local or a remote port.	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 :).
166		108
167	While the message can be about anything, it is highly recommended to use	109	While the message can be about anything, it is highly recommended to use a
168	a constant string as first element.	110	string as first element (a portid, or some word that indicates a request
		111	type etc.).
169		112
170	The message data effectively becomes read-only after a call to this	113	The message data effectively becomes read-only after a call to this
171	function: modifying any argument is not allowed and can cause many	114	function: modifying any argument is not allowed and can cause many
172	problems.	115	problems.
173		116
…		…
175	JSON is used, then only strings, numbers and arrays and hashes consisting	118	JSON is used, then only strings, numbers and arrays and hashes consisting
176	of those are allowed (no objects). When Storable is used, then anything	119	of those are allowed (no objects). When Storable is used, then anything
177	that Storable can serialise and deserialise is allowed, and for the local	120	that Storable can serialise and deserialise is allowed, and for the local
178	node, anything can be passed.	121	node, anything can be passed.
179		122
180	=cut	123	=item $local_port = create_port
181		124
182	sub snd(@) {	125	Create a new local port object. See the next section for allowed methods.
183	my ($noderef, $port) = split /#/, shift, 2;
184		126
185	add_node $noderef	127	=cut
186	unless exists $NODE{$noderef};
187		128
188	$NODE{$noderef}->send (["$port", [@_]]);	129	sub create_port {
189	}	130	my $id = "$AnyEvent::MP::Base::UNIQ." . $AnyEvent::MP::Base::ID++;
190		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 $_;
		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 (i.e. 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
191	=item rcv $portid, type => $callback->(@msg)	212	=item $port->rcv (type => $callback->($port, @msg))
192		213
193	=item rcv $portid, $smartmatch => $callback->(@msg)	214	=item $port->rcv ($smartmatch => $callback->($port, @msg))
194		215
195	=item rcv $portid, [$smartmatch...] => $callback->(@msg)	216	=item $port->rcv ([$smartmatch...] => $callback->($port, @msg))
196		217
197	Register a callback on the port identified by C<$portid>, which I<must> be	218	Register a callback on the given port.
198	a local port.
199		219
200	The callback has to return a true value when its work is done, after	220	The callback has to return a true value when its work is done, after
201	which is will be removed, or a false value in which case it will stay	221	which is will be removed, or a false value in which case it will stay
202	registered.	222	registered.
203		223
…		…
213	also the most efficient match (by far).	233	also the most efficient match (by far).
214		234
215	=cut	235	=cut
216		236
217	sub rcv($@) {	237	sub rcv($@) {
218	my ($port, $match, $cb) = @_;	238	my ($self, $match, $cb) = @_;
219
220	my $port = $PORT{$port}
221	or do {
222	my ($noderef, $lport) = split /#/, $port;
223	"AnyEvent::MP::Node::Self" eq ref $NODE{$noderef}
224	or Carp::croak "$port: can only rcv on local ports";
225
226	$PORT{$lport}
227	or Carp::croak "$port: port does not exist";
228
229	$PORT{$port} = $PORT{$lport} # also return
230	};
231		239
232	if (!ref $match) {	240	if (!ref $match) {
233	push @{ $port->{rc0}{$match} }, [$cb];	241	push @{ $self->{rc0}{$match} }, [$cb];
234	} elsif (("ARRAY" eq ref $match && !ref $match->[0])) {	242	} elsif (("ARRAY" eq ref $match && !ref $match->[0])) {
235	my ($type, @match) = @$match;	243	my ($type, @match) = @$match;
236	@match	244	@match
237	? push @{ $port->{rcv}{$match->[0]} }, [$cb, \@match]	245	? push @{ $self->{rcv}{$match->[0]} }, [$cb, \@match]
238	: push @{ $port->{rc0}{$match->[0]} }, [$cb];	246	: push @{ $self->{rc0}{$match->[0]} }, [$cb];
239	} else {	247	} else {
240	push @{ $port->{any} }, [$cb, $match];	248	push @{ $self->{any} }, [$cb, $match];
241	}	249	}
242	}	250	}
243		251
244	sub _inject {	252	=item $port->register ($name)
245	my ($port, $msg) = @{+shift};
246		253
247	$port = $PORT{$port}	254	Registers the given port under the well known name C<$name>. If the name
248	or return;	255	already exists it is replaced.
249		256
250	@_ = @$msg;	257	A port can only be registered under one well known name.
251		258
252	for (@{ $port->{rc0}{$msg->[0]} }) {	259	=cut
253	$_ && &{$_->[0]}
254	&& undef $_;
255	}
256		260
257	for (@{ $port->{rcv}{$msg->[0]} }) {	261	sub register {
258	$_ && [@_[1..$#{$_->[1]}]] ~~ $_->[1]	262	my ($self, $name) = @_;
259	&& &{$_->[0]}
260	&& undef $_;
261	}
262		263
263	for (@{ $port->{any} }) {	264	$self->{wkname} = $name;
264	$_ && [@_[0..$#{$_->[1]}]] ~~ $_->[1]	265	$AnyEvent::MP::Base::WKP{$name} = "$self";
265	&& &{$_->[0]}
266	&& undef $_;
267	}
268	}	266	}
269		267
270	sub normalise_noderef($) {	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 {
271	my ($noderef) = @_;	278	my ($self) = @_;
272		279
273	my $cv = AE::cv;	280	delete $AnyEvent::MP::Base::WKP{ $self->{wkname} };
274	my @res;
275		281
276	$cv->begin (sub {	282	delete $AnyEvent::MP::Base::PORT{$_}
277	my %seen;	283	for @{ $self->{names} };
278	my @refs;
279	for (sort { $a->[0] <=> $b->[0] } @res) {
280	push @refs, $_->[1] unless $seen{$_->[1]}++
281	}
282	shift->send (join ",", @refs);
283	});
284
285	$noderef = $DEFAULT_PORT unless length $noderef;
286
287	my $idx;
288	for my $t (split /,/, $noderef) {
289	my $pri = ++$idx;
290
291	#TODO: this should be outside normalise_noderef and in become_public
292	if ($t =~ /^\d*$/) {
293	my $nodename = (POSIX::uname)[1];
294
295	$cv->begin;
296	AnyEvent::Socket::resolve_sockaddr $nodename, $t \|\| "aemp=$DEFAULT_PORT", "tcp", 0, undef, sub {
297	for (@_) {
298	my ($service, $host) = AnyEvent::Socket::unpack_sockaddr $_->[3];
299	push @res, [
300	$pri += 1e-5,
301	AnyEvent::Socket::format_hostport AnyEvent::Socket::format_address $host, $service
302	];
303	}
304	$cv->end;
305	};
306
307	# my (undef, undef, undef, undef, @ipv4) = gethostbyname $nodename;
308	#
309	# for (@ipv4) {
310	# push @res, [
311	# $pri,
312	# AnyEvent::Socket::format_hostport AnyEvent::Socket::format_address $_, $t \|\| $DEFAULT_PORT,
313	# ];
314	# }
315	} else {
316	my ($host, $port) = AnyEvent::Socket::parse_hostport $t, "aemp=$DEFAULT_PORT"
317	or Carp::croak "$t: unparsable transport descriptor";
318
319	$cv->begin;
320	AnyEvent::Socket::resolve_sockaddr $host, $port, "tcp", 0, undef, sub {
321	for (@_) {
322	my ($service, $host) = AnyEvent::Socket::unpack_sockaddr $_->[3];
323	push @res, [
324	$pri += 1e-5,
325	AnyEvent::Socket::format_hostport AnyEvent::Socket::format_address $host, $service
326	];
327	}
328	$cv->end;
329	}
330	}
331	}
332
333	$cv->end;
334
335	$cv
336	}	284	}
337		285
338	sub become_public {	286	=back
339	return if $PUBLIC;
340		287
341	my $noderef = join ",", ref $_[0] ? @{+shift} : shift;	288	=head1 FUNCTIONS FOR NODES
342	my @args = @_;
343		289
344	$NODE = (normalise_noderef $noderef)->recv;	290	=over 4
345		291
346	for my $t (split /,/, $NODE) {	292	=item mon $noderef, $callback->($noderef, $status, $)
347	$NODE{$t} = $NODE{""};
348		293
349	my ($host, $port) = AnyEvent::Socket::parse_hostport $t;	294	Monitors the given noderef.
350		295
351	$LISTENER{$t} = AnyEvent::MP::Transport::mp_server $host, $port,	296	=item become_public endpoint...
352	@args,
353	on_error => sub {
354	die "on_error<@_>\n";#d#
355	},
356	on_connect => sub {
357	my ($tp) = @_;
358		297
359	$NODE{$tp->{remote_id}} = $_[0];	298	Tells the node to become a public node, i.e. reachable from other nodes.
360	},
361	sub {
362	my ($tp) = @_;
363		299
364	$NODE{"$tp->{peerhost}:$tp->{peerport}"} = $tp;	300	If no arguments are given, or the first argument is C<undef>, then
365	},	301	AnyEvent::MP tries to bind on port C<4040> on all IP addresses that the
366	;	302	local nodename resolves to.
367	}
368		303
369	$PUBLIC = 1;	304	Otherwise the first argument must be an array-reference with transport
370	}	305	endpoints ("ip:port", "hostname:port") or port numbers (in which case the
		306	local nodename is used as hostname). The endpoints are all resolved and
		307	will become the node reference.
		308
		309	=cut
371		310
372	=back	311	=back
373		312
374	=head1 NODE MESSAGES	313	=head1 NODE MESSAGES
375		314
…		…
380		319
381	=over 4	320	=over 4
382		321
383	=cut	322	=cut
384		323
385	#############################################################################	324	=item wkp => $name, @reply
386	# self node code
387		325
388	sub _new_port($) {	326	Replies with the port ID of the specified well-known port, or C<undef>.
389	my ($name) = @_;
390		327
391	my ($noderef, $portname) = split /#/, $name;	328	=item devnull => ...
392		329
393	$PORT{$name} =	330	Generic data sink/CPU heat conversion.
394	$PORT{$portname} = {
395	names => [$name, $portname],
396	};
397	}
398
399	$NODE{""} = new AnyEvent::MP::Node::Self noderef => $NODE;
400	_new_port "";
401		331
402	=item relay => $port, @msg	332	=item relay => $port, @msg
403		333
404	Simply forwards the message to the given port.	334	Simply forwards the message to the given port.
405
406	=cut
407
408	rcv "", relay => \&snd;
409		335
410	=item eval => $string[ @reply]	336	=item eval => $string[ @reply]
411		337
412	Evaluates the given string. If C<@reply> is given, then a message of the	338	Evaluates the given string. If C<@reply> is given, then a message of the
413	form C<@reply, $@, @evalres> is sent.	339	form C<@reply, $@, @evalres> is sent.
414		340
415	Example: crash another node.	341	Example: crash another node.
416		342
417	snd $othernode, eval => "exit";	343	snd $othernode, eval => "exit";
418		344
419	=cut
420
421	rcv "", eval => sub {
422	my (undef, $string, @reply) = @_;
423	my @res = eval $string;
424	snd @reply, "$@", @res if @reply;
425	};
426
427	=item time => @reply	345	=item time => @reply
428		346
429	Replies the the current node time to C<@reply>.	347	Replies the the current node time to C<@reply>.
430		348
431	Example: tell the current node to send the current time to C<$myport> in a	349	Example: tell the current node to send the current time to C<$myport> in a
432	C<timereply> message.	350	C<timereply> message.
433		351
434	snd $NODE, time => $myport, timereply => 1, 2;	352	snd $NODE, time => $myport, timereply => 1, 2;
435	# => snd $myport, timereply => 1, 2, <time>	353	# => snd $myport, timereply => 1, 2, <time>
436		354
437	=cut
438
439	rcv "", time => sub { shift; snd @_, AE::time };
440
441	=back	355	=back
442		356
443	=head1 SEE ALSO	357	=head1 SEE ALSO
444		358
445	L<AnyEvent>.	359	L<AnyEvent>.

Diff Legend

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

Comparing AnyEvent-MP/MP.pm (file contents): Revision 1.6 by root, Sat Aug 1 10:02:33 2009 UTC vs. Revision 1.17 by root, Mon Aug 3 08:35:40 2009 UTC

Diff Legend

Comparing AnyEvent-MP/MP.pm (file contents):
Revision 1.6 by root, Sat Aug 1 10:02:33 2009 UTC vs.
Revision 1.17 by root, Mon Aug 3 08:35:40 2009 UTC