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

Comparing AnyEvent-MP/MP.pm (file contents):
Revision 1.29 by root, Tue Aug 4 23:16:57 2009 UTC vs.
Revision 1.34 by root, Wed Aug 5 23:50:46 2009 UTC

…		…
53		53
54	=item port id - C<noderef#portname>	54	=item port id - C<noderef#portname>
55		55
56	A port id is normaly the concatenation of a noderef, a hash-mark (C<#>) as	56	A port id is normaly the concatenation of a noderef, a hash-mark (C<#>) as
57	separator, and a port name (a printable string of unspecified format). An	57	separator, and a port name (a printable string of unspecified format). An
58	exception is the the node port, whose ID is identical to it's node	58	exception is the the node port, whose ID is identical to its node
59	reference.	59	reference.
60		60
61	=item node	61	=item node
62		62
63	A node is a single process containing at least one port - the node	63	A node is a single process containing at least one port - the node
…		…
103	use base "Exporter";	103	use base "Exporter";
104		104
105	our $VERSION = '0.1';	105	our $VERSION = '0.1';
106	our @EXPORT = qw(	106	our @EXPORT = qw(
107	NODE $NODE *SELF node_of _any_	107	NODE $NODE *SELF node_of _any_
108	resolve_node	108	resolve_node initialise_node
109	become_slave become_public
110	snd rcv mon kil reg psub	109	snd rcv mon kil reg psub
111	port	110	port
112	);	111	);
113		112
114	our $SELF;	113	our $SELF;
…		…
124	The C<NODE> function returns, and the C<$NODE> variable contains	123	The C<NODE> function returns, and the C<$NODE> variable contains
125	the noderef of the local node. The value is initialised by a call	124	the noderef of the local node. The value is initialised by a call
126	to C<become_public> or C<become_slave>, after which all local port	125	to C<become_public> or C<become_slave>, after which all local port
127	identifiers become invalid.	126	identifiers become invalid.
128		127
129	=item $noderef = node_of $portid	128	=item $noderef = node_of $port
130		129
131	Extracts and returns the noderef from a portid or a noderef.	130	Extracts and returns the noderef from a portid or a noderef.
		131
		132	=item initialise_node $noderef, $seednode, $seednode...
		133
		134	=item initialise_node "slave/", $master, $master...
		135
		136	Before a node can talk to other nodes on the network it has to initialise
		137	itself - the minimum a node needs to know is it's own name, and optionally
		138	it should know the noderefs of some other nodes in the network.
		139
		140	This function initialises a node - it must be called exactly once (or
		141	never) before calling other AnyEvent::MP functions.
		142
		143	All arguments are noderefs, which can be either resolved or unresolved.
		144
		145	There are two types of networked nodes, public nodes and slave nodes:
		146
		147	=over 4
		148
		149	=item public nodes
		150
		151	For public nodes, C<$noderef> must either be a (possibly unresolved)
		152	noderef, in which case it will be resolved, or C<undef> (or missing), in
		153	which case the noderef will be guessed.
		154
		155	Afterwards, the node will bind itself on all endpoints and try to connect
		156	to all additional C<$seednodes> that are specified. Seednodes are optional
		157	and can be used to quickly bootstrap the node into an existing network.
		158
		159	=item slave nodes
		160
		161	When the C<$noderef> is the special string C<slave/>, then the node will
		162	become a slave node. Slave nodes cannot be contacted from outside and will
		163	route most of their traffic to the master node that they attach to.
		164
		165	At least one additional noderef is required: The node will try to connect
		166	to all of them and will become a slave attached to the first node it can
		167	successfully connect to.
		168
		169	=back
		170
		171	This function will block until all nodes have been resolved and, for slave
		172	nodes, until it has successfully established a connection to a master
		173	server.
		174
		175	Example: become a public node listening on the default node.
		176
		177	initialise_node;
		178
		179	Example: become a public node, and try to contact some well-known master
		180	servers to become part of the network.
		181
		182	initialise_node undef, "master1", "master2";
		183
		184	Example: become a public node listening on port C<4041>.
		185
		186	initialise_node 4041;
		187
		188	Example: become a public node, only visible on localhost port 4044.
		189
		190	initialise_node "locahost:4044";
		191
		192	Example: become a slave node to any of the specified master servers.
		193
		194	initialise_node "slave/", "master1", "192.168.13.17", "mp.example.net";
132		195
133	=item $cv = resolve_node $noderef	196	=item $cv = resolve_node $noderef
134		197
135	Takes an unresolved node reference that may contain hostnames and	198	Takes an unresolved node reference that may contain hostnames and
136	abbreviated IDs, resolves all of them and returns a resolved node	199	abbreviated IDs, resolves all of them and returns a resolved node
…		…
168		231
169	Due to some quirks in how perl exports variables, it is impossible to	232	Due to some quirks in how perl exports variables, it is impossible to
170	just export C<$SELF>, all the symbols called C<SELF> are exported by this	233	just export C<$SELF>, all the symbols called C<SELF> are exported by this
171	module, but only C<$SELF> is currently used.	234	module, but only C<$SELF> is currently used.
172		235
173	=item snd $portid, type => @data	236	=item snd $port, type => @data
174		237
175	=item snd $portid, @msg	238	=item snd $port, @msg
176		239
177	Send the given message to the given port ID, which can identify either	240	Send the given message to the given port ID, which can identify either
178	a local or a remote port, and can be either a string or soemthignt hat	241	a local or a remote port, and can be either a string or soemthignt hat
179	stringifies a sa port ID (such as a port object :).	242	stringifies a sa port ID (such as a port object :).
180		243
…		…
190	JSON is used, then only strings, numbers and arrays and hashes consisting	253	JSON is used, then only strings, numbers and arrays and hashes consisting
191	of those are allowed (no objects). When Storable is used, then anything	254	of those are allowed (no objects). When Storable is used, then anything
192	that Storable can serialise and deserialise is allowed, and for the local	255	that Storable can serialise and deserialise is allowed, and for the local
193	node, anything can be passed.	256	node, anything can be passed.
194		257
195	=item kil $portid[, @reason]
196
197	Kill the specified port with the given C<@reason>.
198
199	If no C<@reason> is specified, then the port is killed "normally" (linked
200	ports will not be kileld, or even notified).
201
202	Otherwise, linked ports get killed with the same reason (second form of
203	C<mon>, see below).
204
205	Runtime errors while evaluating C<rcv> callbacks or inside C<psub> blocks
206	will be reported as reason C<< die => $@ >>.
207
208	Transport/communication errors are reported as C<< transport_error =>
209	$message >>.
210
211	=item $guard = mon $portid, $cb->(@reason)
212
213	=item $guard = mon $portid, $otherport
214
215	=item $guard = mon $portid, $otherport, @msg
216
217	Monitor the given port and do something when the port is killed.
218
219	In the first form, the callback is simply called with any number
220	of C<@reason> elements (no @reason means that the port was deleted
221	"normally"). Note also that I<< the callback B<must> never die >>, so use
222	C<eval> if unsure.
223
224	In the second form, the other port will be C<kil>'ed with C<@reason>, iff
225	a @reason was specified, i.e. on "normal" kils nothing happens, while
226	under all other conditions, the other port is killed with the same reason.
227
228	In the last form, a message of the form C<@msg, @reason> will be C<snd>.
229
230	Example: call a given callback when C<$port> is killed.
231
232	mon $port, sub { warn "port died because of <@_>\n" };
233
234	Example: kill ourselves when C<$port> is killed abnormally.
235
236	mon $port, $self;
237
238	Example: send us a restart message another C<$port> is killed.
239
240	mon $port, $self => "restart";
241
242	=cut
243
244	sub mon {
245	my ($noderef, $port, $cb) = ((split /#/, shift, 2), shift);
246
247	my $node = $NODE{$noderef} \|\| add_node $noderef;
248
249	#TODO: ports must not be references
250	if (!ref $cb or "AnyEvent::MP::Port" eq ref $cb) {
251	if (@_) {
252	# send a kill info message
253	my (@msg) = ($cb, @_);
254	$cb = sub { snd @msg, @_ };
255	} else {
256	# simply kill other port
257	my $port = $cb;
258	$cb = sub { kil $port, @_ if @_ };
259	}
260	}
261
262	$node->monitor ($port, $cb);
263
264	defined wantarray
265	and AnyEvent::Util::guard { $node->unmonitor ($port, $cb) }
266	}
267
268	=item $guard = mon_guard $port, $ref, $ref...
269
270	Monitors the given C<$port> and keeps the passed references. When the port
271	is killed, the references will be freed.
272
273	Optionally returns a guard that will stop the monitoring.
274
275	This function is useful when you create e.g. timers or other watchers and
276	want to free them when the port gets killed:
277
278	$port->rcv (start => sub {
279	my $timer; $timer = mon_guard $port, AE::timer 1, 1, sub {
280	undef $timer if 0.9 < rand;
281	});
282	});
283
284	=cut
285
286	sub mon_guard {
287	my ($port, @refs) = @_;
288
289	mon $port, sub { 0 && @refs }
290	}
291
292	=item lnk $port1, $port2
293
294	Link two ports. This is simply a shorthand for:
295
296	mon $port1, $port2;
297	mon $port2, $port1;
298
299	It means that if either one is killed abnormally, the other one gets
300	killed as well.
301
302	=item $local_port = port	258	=item $local_port = port
303		259
304	Create a new local port object that supports message matching.	260	Create a new local port object that can be used either as a pattern
		261	matching port ("full port") or a single-callback port ("miniport"),
		262	depending on how C<rcv> callbacks are bound to the object.
305		263
306	=item $portid = port { my @msg = @_; $finished }	264	=item $port = port { my @msg = @_; $finished }
307		265
308	Creates a "mini port", that is, a very lightweight port without any	266	Creates a "miniport", that is, a very lightweight port without any pattern
309	pattern matching behind it, and returns its ID.	267	matching behind it, and returns its ID. Semantically the same as creating
		268	a port and calling C<rcv $port, $callback> on it.
310		269
311	The block will be called for every message received on the port. When the	270	The block will be called for every message received on the port. When the
312	callback returns a true value its job is considered "done" and the port	271	callback returns a true value its job is considered "done" and the port
313	will be destroyed. Otherwise it will stay alive.	272	will be destroyed. Otherwise it will stay alive.
314		273
315	The message will be passed as-is, no extra argument (i.e. no port id) will	274	The message will be passed as-is, no extra argument (i.e. no port id) will
316	be passed to the callback.	275	be passed to the callback.
317		276
318	If you need the local port id in the callback, this works nicely:	277	If you need the local port id in the callback, this works nicely:
319		278
320	my $port; $port = miniport {	279	my $port; $port = port {
321	snd $otherport, reply => $port;	280	snd $otherport, reply => $port;
322	};	281	};
323		282
324	=cut	283	=cut
		284
		285	sub rcv($@);
325		286
326	sub port(;&) {	287	sub port(;&) {
327	my $id = "$UNIQ." . $ID++;	288	my $id = "$UNIQ." . $ID++;
328	my $port = "$NODE#$id";	289	my $port = "$NODE#$id";
329		290
330	if (@_) {	291	if (@_) {
		292	rcv $port, shift;
		293	} else {
		294	$PORT{$id} = sub { }; # nop
		295	}
		296
		297	$port
		298	}
		299
		300	=item reg $port, $name
		301
		302	Registers the given port under the name C<$name>. If the name already
		303	exists it is replaced.
		304
		305	A port can only be registered under one well known name.
		306
		307	A port automatically becomes unregistered when it is killed.
		308
		309	=cut
		310
		311	sub reg(@) {
		312	my ($port, $name) = @_;
		313
		314	$REG{$name} = $port;
		315	}
		316
		317	=item rcv $port, $callback->(@msg)
		318
		319	Replaces the callback on the specified miniport (after converting it to
		320	one if required).
		321
		322	=item rcv $port, tagstring => $callback->(@msg), ...
		323
		324	=item rcv $port, $smartmatch => $callback->(@msg), ...
		325
		326	=item rcv $port, [$smartmatch...] => $callback->(@msg), ...
		327
		328	Register callbacks to be called on matching messages on the given full
		329	port (after converting it to one if required).
		330
		331	The callback has to return a true value when its work is done, after
		332	which is will be removed, or a false value in which case it will stay
		333	registered.
		334
		335	The global C<$SELF> (exported by this module) contains C<$port> while
		336	executing the callback.
		337
		338	Runtime errors wdurign callback execution will result in the port being
		339	C<kil>ed.
		340
		341	If the match is an array reference, then it will be matched against the
		342	first elements of the message, otherwise only the first element is being
		343	matched.
		344
		345	Any element in the match that is specified as C<_any_> (a function
		346	exported by this module) matches any single element of the message.
		347
		348	While not required, it is highly recommended that the first matching
		349	element is a string identifying the message. The one-string-only match is
		350	also the most efficient match (by far).
		351
		352	=cut
		353
		354	sub rcv($@) {
		355	my $port = shift;
		356	my ($noderef, $portid) = split /#/, $port, 2;
		357
		358	($NODE{$noderef} \|\| add_node $noderef) == $NODE{""}
		359	or Carp::croak "$port: rcv can only be called on local ports, caught";
		360
		361	if (@_ == 1) {
331	my $cb = shift;	362	my $cb = shift;
		363	delete $PORT_DATA{$portid};
332	$PORT{$id} = sub {	364	$PORT{$portid} = sub {
333	local $SELF = $port;	365	local $SELF = $port;
334	eval {	366	eval {
335	&$cb	367	&$cb
336	and kil $id;	368	and kil $port;
337	};	369	};
338	_self_die if $@;	370	_self_die if $@;
339	};	371	};
340	} else {	372	} else {
		373	my $self = $PORT_DATA{$portid} \|\|= do {
341	my $self = bless {	374	my $self = bless {
342	id => "$NODE#$id",	375	id => $port,
343	}, "AnyEvent::MP::Port";	376	}, "AnyEvent::MP::Port";
344		377
345	$PORT_DATA{$id} = $self;
346	$PORT{$id} = sub {	378	$PORT{$portid} = sub {
347	local $SELF = $port;	379	local $SELF = $port;
348		380
349	eval {	381	eval {
350	for (@{ $self->{rc0}{$_[0]} }) {	382	for (@{ $self->{rc0}{$_[0]} }) {
351	$_ && &{$_->[0]}	383	$_ && &{$_->[0]}
352	&& undef $_;	384	&& undef $_;
353	}	385	}
354		386
355	for (@{ $self->{rcv}{$_[0]} }) {	387	for (@{ $self->{rcv}{$_[0]} }) {
356	$_ && [@_[1 .. @{$_->[1]}]] ~~ $_->[1]	388	$_ && [@_[1 .. @{$_->[1]}]] ~~ $_->[1]
357	&& &{$_->[0]}	389	&& &{$_->[0]}
358	&& undef $_;	390	&& undef $_;
359	}	391	}
360		392
361	for (@{ $self->{any} }) {	393	for (@{ $self->{any} }) {
362	$_ && [@_[0 .. $#{$_->[1]}]] ~~ $_->[1]	394	$_ && [@_[0 .. $#{$_->[1]}]] ~~ $_->[1]
363	&& &{$_->[0]}	395	&& &{$_->[0]}
364	&& undef $_;	396	&& undef $_;
		397	}
365	}	398	};
		399	_self_die if $@;
366	};	400	};
367	_self_die if $@;	401
		402	$self
368	};	403	};
369	}
370		404
371	$port
372	}
373
374	=item reg $portid, $name
375
376	Registers the given port under the name C<$name>. If the name already
377	exists it is replaced.
378
379	A port can only be registered under one well known name.
380
381	A port automatically becomes unregistered when it is killed.
382
383	=cut
384
385	sub reg(@) {
386	my ($portid, $name) = @_;
387
388	$REG{$name} = $portid;
389	}
390
391	=item rcv $portid, tagstring => $callback->(@msg), ...
392
393	=item rcv $portid, $smartmatch => $callback->(@msg), ...
394
395	=item rcv $portid, [$smartmatch...] => $callback->(@msg), ...
396
397	Register callbacks to be called on matching messages on the given port.
398
399	The callback has to return a true value when its work is done, after
400	which is will be removed, or a false value in which case it will stay
401	registered.
402
403	The global C<$SELF> (exported by this module) contains C<$portid> while
404	executing the callback.
405
406	Runtime errors wdurign callback execution will result in the port being
407	C<kil>ed.
408
409	If the match is an array reference, then it will be matched against the
410	first elements of the message, otherwise only the first element is being
411	matched.
412
413	Any element in the match that is specified as C<_any_> (a function
414	exported by this module) matches any single element of the message.
415
416	While not required, it is highly recommended that the first matching
417	element is a string identifying the message. The one-string-only match is
418	also the most efficient match (by far).
419
420	=cut
421
422	sub rcv($@) {
423	my ($noderef, $port) = split /#/, shift, 2;
424
425	($NODE{$noderef} \|\| add_node $noderef) == $NODE{""}
426	or Carp::croak "$noderef#$port: rcv can only be called on local ports, caught";
427
428	my $self = $PORT_DATA{$port}
429	or Carp::croak "$noderef#$port: rcv can only be called on message matching ports, caught";
430
431	"AnyEvent::MP::Port" eq ref $self	405	"AnyEvent::MP::Port" eq ref $self
432	or Carp::croak "$noderef#$port: rcv can only be called on message matching ports, caught";	406	or Carp::croak "$port: rcv can only be called on message matching ports, caught";
433		407
434	while (@_) {	408	while (@_) {
435	my ($match, $cb) = splice @_, 0, 2;	409	my ($match, $cb) = splice @_, 0, 2;
436		410
437	if (!ref $match) {	411	if (!ref $match) {
438	push @{ $self->{rc0}{$match} }, [$cb];	412	push @{ $self->{rc0}{$match} }, [$cb];
439	} elsif (("ARRAY" eq ref $match && !ref $match->[0])) {	413	} elsif (("ARRAY" eq ref $match && !ref $match->[0])) {
440	my ($type, @match) = @$match;	414	my ($type, @match) = @$match;
441	@match	415	@match
442	? push @{ $self->{rcv}{$match->[0]} }, [$cb, \@match]	416	? push @{ $self->{rcv}{$match->[0]} }, [$cb, \@match]
443	: push @{ $self->{rc0}{$match->[0]} }, [$cb];	417	: push @{ $self->{rc0}{$match->[0]} }, [$cb];
444	} else {	418	} else {
445	push @{ $self->{any} }, [$cb, $match];	419	push @{ $self->{any} }, [$cb, $match];
		420	}
446	}	421	}
447	}	422	}
		423
		424	$port
448	}	425	}
449		426
450	=item $closure = psub { BLOCK }	427	=item $closure = psub { BLOCK }
451		428
452	Remembers C<$SELF> and creates a closure out of the BLOCK. When the	429	Remembers C<$SELF> and creates a closure out of the BLOCK. When the
…		…
483	$res	460	$res
484	}	461	}
485	}	462	}
486	}	463	}
487		464
488	=back	465	=item $guard = mon $port, $cb->(@reason)
489		466
490	=head1 FUNCTIONS FOR NODES	467	=item $guard = mon $port, $otherport
491		468
492	=over 4	469	=item $guard = mon $port, $otherport, @msg
493		470
494	=item become_public $noderef	471	Monitor the given port and do something when the port is killed.
495		472
496	Tells the node to become a public node, i.e. reachable from other nodes.	473	In the first form, the callback is simply called with any number
		474	of C<@reason> elements (no @reason means that the port was deleted
		475	"normally"). Note also that I<< the callback B<must> never die >>, so use
		476	C<eval> if unsure.
497		477
498	The first argument is the (unresolved) node reference of the local node	478	In the second form, the other port will be C<kil>'ed with C<@reason>, iff
499	(if missing then the empty string is used).	479	a @reason was specified, i.e. on "normal" kils nothing happens, while
		480	under all other conditions, the other port is killed with the same reason.
500		481
501	It is quite common to not specify anything, in which case the local node	482	In the last form, a message of the form C<@msg, @reason> will be C<snd>.
502	tries to listen on the default port, or to only specify a port number, in
503	which case AnyEvent::MP tries to guess the local addresses.
504		483
		484	Example: call a given callback when C<$port> is killed.
		485
		486	mon $port, sub { warn "port died because of <@_>\n" };
		487
		488	Example: kill ourselves when C<$port> is killed abnormally.
		489
		490	mon $port, $self;
		491
		492	Example: send us a restart message another C<$port> is killed.
		493
		494	mon $port, $self => "restart";
		495
505	=cut	496	=cut
		497
		498	sub mon {
		499	my ($noderef, $port) = split /#/, shift, 2;
		500
		501	my $node = $NODE{$noderef} \|\| add_node $noderef;
		502
		503	my $cb = shift;
		504
		505	unless (ref $cb) {
		506	if (@_) {
		507	# send a kill info message
		508	my (@msg) = ($cb, @_);
		509	$cb = sub { snd @msg, @_ };
		510	} else {
		511	# simply kill other port
		512	my $port = $cb;
		513	$cb = sub { kil $port, @_ if @_ };
		514	}
		515	}
		516
		517	$node->monitor ($port, $cb);
		518
		519	defined wantarray
		520	and AnyEvent::Util::guard { $node->unmonitor ($port, $cb) }
		521	}
		522
		523	=item $guard = mon_guard $port, $ref, $ref...
		524
		525	Monitors the given C<$port> and keeps the passed references. When the port
		526	is killed, the references will be freed.
		527
		528	Optionally returns a guard that will stop the monitoring.
		529
		530	This function is useful when you create e.g. timers or other watchers and
		531	want to free them when the port gets killed:
		532
		533	$port->rcv (start => sub {
		534	my $timer; $timer = mon_guard $port, AE::timer 1, 1, sub {
		535	undef $timer if 0.9 < rand;
		536	});
		537	});
		538
		539	=cut
		540
		541	sub mon_guard {
		542	my ($port, @refs) = @_;
		543
		544	mon $port, sub { 0 && @refs }
		545	}
		546
		547	=item lnk $port1, $port2
		548
		549	Link two ports. This is simply a shorthand for:
		550
		551	mon $port1, $port2;
		552	mon $port2, $port1;
		553
		554	It means that if either one is killed abnormally, the other one gets
		555	killed as well.
		556
		557	=item kil $port[, @reason]
		558
		559	Kill the specified port with the given C<@reason>.
		560
		561	If no C<@reason> is specified, then the port is killed "normally" (linked
		562	ports will not be kileld, or even notified).
		563
		564	Otherwise, linked ports get killed with the same reason (second form of
		565	C<mon>, see below).
		566
		567	Runtime errors while evaluating C<rcv> callbacks or inside C<psub> blocks
		568	will be reported as reason C<< die => $@ >>.
		569
		570	Transport/communication errors are reported as C<< transport_error =>
		571	$message >>.
506		572
507	=back	573	=back
508		574
509	=head1 NODE MESSAGES	575	=head1 NODE MESSAGES
510		576

Diff Legend

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

Comparing AnyEvent-MP/MP.pm (file contents): Revision 1.29 by root, Tue Aug 4 23:16:57 2009 UTC vs. Revision 1.34 by root, Wed Aug 5 23:50:46 2009 UTC

Diff Legend

Comparing AnyEvent-MP/MP.pm (file contents):
Revision 1.29 by root, Tue Aug 4 23:16:57 2009 UTC vs.
Revision 1.34 by root, Wed Aug 5 23:50:46 2009 UTC