[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.18 by root, Mon Aug 3 21:35:03 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 mon del _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 mon $portid, sub { }
181		124
182	sub snd(@) {	125	#TODO monitor the given port
		126
		127	=cut
		128
		129	sub mon {
183	my ($noderef, $port) = split /#/, shift, 2;	130	my ($noderef, $port) = split /#/, shift, 2;
184		131
185	add_node $noderef	132	my $node = AnyEvent::MP::Base::add_node $noderef;
186	unless exists $NODE{$noderef};
187		133
188	$NODE{$noderef}->send (["$port", [@_]]);	134	my $cb = shift;
189	}
190		135
		136	$node->monitor ($port, $cb);
		137
		138	defined wantarray
		139	and AnyEvent::Util::guard { $node->unmonitor ($port, $cb) }
		140	}
		141
		142	=item $local_port = create_port
		143
		144	Create a new local port object. See the next section for allowed methods.
		145
		146	=cut
		147
		148	sub create_port {
		149	my $id = "$AnyEvent::MP::Base::UNIQ." . $AnyEvent::MP::Base::ID++;
		150
		151	my $self = bless {
		152	id => "$NODE#$id",
		153	names => [$id],
		154	}, "AnyEvent::MP::Port";
		155
		156	$AnyEvent::MP::Base::PORT{$id} = sub {
		157	unshift @_, $self;
		158
		159	for (@{ $self->{rc0}{$_[1]} }) {
		160	$_ && &{$_->[0]}
		161	&& undef $_;
		162	}
		163
		164	for (@{ $self->{rcv}{$_[1]} }) {
		165	$_ && [@_[1 .. @{$_->[1]}]] ~~ $_->[1]
		166	&& &{$_->[0]}
		167	&& undef $_;
		168	}
		169
		170	for (@{ $self->{any} }) {
		171	$_ && [@_[0 .. $#{$_->[1]}]] ~~ $_->[1]
		172	&& &{$_->[0]}
		173	&& undef $_;
		174	}
		175	};
		176
		177	$self
		178	}
		179
		180	=item $portid = miniport { my @msg = @_; $finished }
		181
		182	Creates a "mini port", that is, a very lightweight port without any
		183	pattern matching behind it, and returns its ID.
		184
		185	The block will be called for every message received on the port. When the
		186	callback returns a true value its job is considered "done" and the port
		187	will be destroyed. Otherwise it will stay alive.
		188
		189	The message will be passed as-is, no extra argument (i.e. no port id) will
		190	be passed to the callback.
		191
		192	If you need the local port id in the callback, this works nicely:
		193
		194	my $port; $port = miniport {
		195	snd $otherport, reply => $port;
		196	};
		197
		198	=cut
		199
		200	sub miniport(&) {
		201	my $cb = shift;
		202	my $id = "$AnyEvent::MP::Base::UNIQ." . $AnyEvent::MP::Base::ID++;
		203
		204	$AnyEvent::MP::Base::PORT{$id} = sub {
		205	&$cb
		206	and delete $AnyEvent::MP::Base::PORT{$id};
		207	};
		208
		209	"$NODE#$id"
		210	}
		211
		212	package AnyEvent::MP::Port;
		213
		214	=back
		215
		216	=head1 METHODS FOR PORT OBJECTS
		217
		218	=over 4
		219
		220	=item "$port"
		221
		222	A port object stringifies to its port ID, so can be used directly for
		223	C<snd> operations.
		224
		225	=cut
		226
		227	use overload
		228	'""' => sub { $_[0]{id} },
		229	fallback => 1;
		230
		231	sub TO_JSON { $_[0]{id} }
		232
191	=item rcv $portid, type => $callback->(@msg)	233	=item $port->rcv (type => $callback->($port, @msg))
192		234
193	=item rcv $portid, $smartmatch => $callback->(@msg)	235	=item $port->rcv ($smartmatch => $callback->($port, @msg))
194		236
195	=item rcv $portid, [$smartmatch...] => $callback->(@msg)	237	=item $port->rcv ([$smartmatch...] => $callback->($port, @msg))
196		238
197	Register a callback on the port identified by C<$portid>, which I<must> be	239	Register a callback on the given port.
198	a local port.
199		240
200	The callback has to return a true value when its work is done, after	241	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	242	which is will be removed, or a false value in which case it will stay
202	registered.	243	registered.
203		244
…		…
213	also the most efficient match (by far).	254	also the most efficient match (by far).
214		255
215	=cut	256	=cut
216		257
217	sub rcv($@) {	258	sub rcv($@) {
218	my ($port, $match, $cb) = @_;	259	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		260
232	if (!ref $match) {	261	if (!ref $match) {
233	push @{ $port->{rc0}{$match} }, [$cb];	262	push @{ $self->{rc0}{$match} }, [$cb];
234	} elsif (("ARRAY" eq ref $match && !ref $match->[0])) {	263	} elsif (("ARRAY" eq ref $match && !ref $match->[0])) {
235	my ($type, @match) = @$match;	264	my ($type, @match) = @$match;
236	@match	265	@match
237	? push @{ $port->{rcv}{$match->[0]} }, [$cb, \@match]	266	? push @{ $self->{rcv}{$match->[0]} }, [$cb, \@match]
238	: push @{ $port->{rc0}{$match->[0]} }, [$cb];	267	: push @{ $self->{rc0}{$match->[0]} }, [$cb];
239	} else {	268	} else {
240	push @{ $port->{any} }, [$cb, $match];	269	push @{ $self->{any} }, [$cb, $match];
241	}	270	}
242	}	271	}
243		272
244	sub _inject {	273	=item $port->register ($name)
245	my ($port, $msg) = @{+shift};
246		274
247	$port = $PORT{$port}	275	Registers the given port under the well known name C<$name>. If the name
248	or return;	276	already exists it is replaced.
249		277
250	@_ = @$msg;	278	A port can only be registered under one well known name.
251		279
252	for (@{ $port->{rc0}{$msg->[0]} }) {	280	=cut
253	$_ && &{$_->[0]}
254	&& undef $_;
255	}
256		281
257	for (@{ $port->{rcv}{$msg->[0]} }) {	282	sub register {
258	$_ && [@_[1..$#{$_->[1]}]] ~~ $_->[1]	283	my ($self, $name) = @_;
259	&& &{$_->[0]}
260	&& undef $_;
261	}
262		284
263	for (@{ $port->{any} }) {	285	$self->{wkname} = $name;
264	$_ && [@_[0..$#{$_->[1]}]] ~~ $_->[1]	286	$AnyEvent::MP::Base::WKP{$name} = "$self";
265	&& &{$_->[0]}
266	&& undef $_;
267	}
268	}	287	}
269		288
270	sub normalise_noderef($) {	289	=item $port->destroy
		290
		291	Explicitly destroy/remove/nuke/vaporise the port.
		292
		293	Ports are normally kept alive by there mere existance alone, and need to
		294	be destroyed explicitly.
		295
		296	=cut
		297
		298	sub destroy {
271	my ($noderef) = @_;	299	my ($self) = @_;
272		300
273	my $cv = AE::cv;	301	AnyEvent::MP::Base::del $self->{id};
274	my @res;
275		302
276	$cv->begin (sub {	303	delete $AnyEvent::MP::Base::WKP{ $self->{wkname} };
277	my %seen;
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		304
285	$noderef = $DEFAULT_PORT unless length $noderef;	305	delete $AnyEvent::MP::Base::PORT{$_}
286		306	for @{ $self->{names} };
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	}	307	}
337		308
338	sub become_public {	309	=back
339	return if $PUBLIC;
340		310
341	my $noderef = join ",", ref $_[0] ? @{+shift} : shift;	311	=head1 FUNCTIONS FOR NODES
342	my @args = @_;
343		312
344	$NODE = (normalise_noderef $noderef)->recv;	313	=over 4
345		314
346	for my $t (split /,/, $NODE) {	315	=item mon $noderef, $callback->($noderef, $status, $)
347	$NODE{$t} = $NODE{""};
348		316
349	my ($host, $port) = AnyEvent::Socket::parse_hostport $t;	317	Monitors the given noderef.
350		318
351	$LISTENER{$t} = AnyEvent::MP::Transport::mp_server $host, $port,	319	=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		320
359	$NODE{$tp->{remote_id}} = $_[0];	321	Tells the node to become a public node, i.e. reachable from other nodes.
360	},
361	sub {
362	my ($tp) = @_;
363		322
364	$NODE{"$tp->{peerhost}:$tp->{peerport}"} = $tp;	323	If no arguments are given, or the first argument is C<undef>, then
365	},	324	AnyEvent::MP tries to bind on port C<4040> on all IP addresses that the
366	;	325	local nodename resolves to.
367	}
368		326
369	$PUBLIC = 1;	327	Otherwise the first argument must be an array-reference with transport
370	}	328	endpoints ("ip:port", "hostname:port") or port numbers (in which case the
		329	local nodename is used as hostname). The endpoints are all resolved and
		330	will become the node reference.
		331
		332	=cut
371		333
372	=back	334	=back
373		335
374	=head1 NODE MESSAGES	336	=head1 NODE MESSAGES
375		337
…		…
380		342
381	=over 4	343	=over 4
382		344
383	=cut	345	=cut
384		346
385	#############################################################################	347	=item wkp => $name, @reply
386	# self node code
387		348
388	sub _new_port($) {	349	Replies with the port ID of the specified well-known port, or C<undef>.
389	my ($name) = @_;
390		350
391	my ($noderef, $portname) = split /#/, $name;	351	=item devnull => ...
392		352
393	$PORT{$name} =	353	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		354
402	=item relay => $port, @msg	355	=item relay => $port, @msg
403		356
404	Simply forwards the message to the given port.	357	Simply forwards the message to the given port.
405
406	=cut
407
408	rcv "", relay => \&snd;
409		358
410	=item eval => $string[ @reply]	359	=item eval => $string[ @reply]
411		360
412	Evaluates the given string. If C<@reply> is given, then a message of the	361	Evaluates the given string. If C<@reply> is given, then a message of the
413	form C<@reply, $@, @evalres> is sent.	362	form C<@reply, $@, @evalres> is sent.
414		363
415	Example: crash another node.	364	Example: crash another node.
416		365
417	snd $othernode, eval => "exit";	366	snd $othernode, eval => "exit";
418		367
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	368	=item time => @reply
428		369
429	Replies the the current node time to C<@reply>.	370	Replies the the current node time to C<@reply>.
430		371
431	Example: tell the current node to send the current time to C<$myport> in a	372	Example: tell the current node to send the current time to C<$myport> in a
432	C<timereply> message.	373	C<timereply> message.
433		374
434	snd $NODE, time => $myport, timereply => 1, 2;	375	snd $NODE, time => $myport, timereply => 1, 2;
435	# => snd $myport, timereply => 1, 2, <time>	376	# => snd $myport, timereply => 1, 2, <time>
436		377
437	=cut
438
439	rcv "", time => sub { shift; snd @_, AE::time };
440
441	=back	378	=back
442		379
443	=head1 SEE ALSO	380	=head1 SEE ALSO
444		381
445	L<AnyEvent>.	382	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.18 by root, Mon Aug 3 21:35:03 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.18 by root, Mon Aug 3 21:35:03 2009 UTC