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

Comparing AnyEvent-MP/MP/Node.pm (file contents):
Revision 1.7 by root, Tue Aug 4 14:10:51 2009 UTC vs.
Revision 1.66 by root, Sat Mar 24 00:48:54 2012 UTC

…		…
1	=head1 NAME	1	=head1 NAME
2		2
3	AnyEvent::MP::Node - a single processing node (CPU/process...)	3	AnyEvent::MP::Node - represent a node
4		4
5	=head1 SYNOPSIS	5	=head1 SYNOPSIS
6		6
7	use AnyEvent::MP::Node;	7	use AnyEvent::MP::Node;
8		8
9	=head1 DESCRIPTION	9	=head1 DESCRIPTION
10		10
		11	This is an internal utility module, horrible to look at, so don't.
		12
11	=cut	13	=cut
12		14
13	package AnyEvent::MP::Node;	15	package AnyEvent::MP::Node; # base class for nodes
14		16
15	use common::sense;	17	use common::sense;
16		18
17	use AE ();
18	use AnyEvent::Util ();	19	use AnyEvent ();
19	use AnyEvent::Socket ();	20	use AnyEvent::Socket ();
20		21
21	use AnyEvent::MP::Transport ();	22	use AnyEvent::MP::Transport ();
22		23
23	use base Exporter::;	24	sub new {
		25	my ($self, $id) = @_;
24		26
25	our $VERSION = '0.0';	27	$self = bless { id => $id }, $self;
26		28
27	our $DEFAULT_PORT = "4040";	29	# register
		30	$AnyEvent::MP::Kernel::NODE{$id} = $self;
28		31
29	sub normalise_noderef($) {	32	$self->init;
		33	$self->transport_reset;
		34
		35	$self
		36	}
		37
		38	sub DESTROY {
		39	# unregister
		40	delete $AnyEvent::MP::Kernel::NODE{$_[0]{id}};
		41	}
		42
		43	sub init {
		44	#
		45	}
		46
		47	sub send {
		48	&{ shift->{send} }
		49	}
		50
		51	# nodes reachable via the network
		52	package AnyEvent::MP::Node::Remote; # a remote node
		53
		54	use base "AnyEvent::MP::Node";
		55
		56	# called at init time, mostly sets {send}
		57	sub transport_reset {
30	my ($noderef) = @_;	58	my ($self) = @_;
31		59
32	my $cv = AE::cv;	60	delete $self->{transport};
33	my @res;
34		61
35	$cv->begin (sub {	62	Scalar::Util::weaken $self;
36	my %seen;	63
37	my @refs;	64	$self->{send} = sub {
38	for (sort { $a->[0] <=> $b->[0] } @res) {	65	push @{$self->{queue}}, shift;
39	push @refs, $_->[1] unless $seen{$_->[1]}++	66	$self->connect;
		67	};
		68	}
		69
		70	# called each time we fail to establish a connection,
		71	# or the existing connection failed
		72	sub transport_error {
		73	my ($self, @reason) = @_;
		74
		75	my $no_transport = !$self->{transport};
		76
		77	delete $self->{connect_w};
		78	delete $self->{connect_to};
		79
		80	delete $self->{queue};
		81	$self->transport_reset;
		82
		83	if (my $mon = delete $self->{lmon}) {
		84	$_->(@reason) for map @$_, values %$mon;
		85	}
		86
		87	AnyEvent::MP::Kernel::_inject_nodeevent ($self, 0, @reason)
		88	unless $no_transport;
		89
		90	# we weaken the node reference, so it can go away if unused
		91	Scalar::Util::weaken $AnyEvent::MP::Kernel::NODE{$self->{id}}
		92	unless $self->{connect_to};
		93
		94	AE::log 9 => "@reason";
		95	}
		96
		97	# called after handshake was successful
		98	sub transport_connect {
		99	my ($self, $transport) = @_;
		100
		101	delete $self->{trial};
		102
		103	$self->transport_error (transport_error => $self->{id}, "switched connections")
		104	if $self->{transport};
		105
		106	delete $self->{connect_w};
		107	delete $self->{connect_to};
		108
		109	$self->{transport} = $transport;
		110
		111	my $transport_send = $transport->{send};
		112
		113	AnyEvent::MP::Kernel::_inject_nodeevent ($self, 1);
		114
		115	$self->{send} = $transport_send;
		116
		117	$transport_send->($_)
		118	for @{ delete $self->{queue} \|\| [] };
		119	}
		120
		121	sub connect {
		122	my ($self) = @_;
		123
		124	return if $self->{transport};
		125	return if $self->{connect_w};
		126
		127	# we unweaken the node reference, in case it was weakened before
		128	$AnyEvent::MP::Kernel::NODE{$self->{id}}
		129	= $AnyEvent::MP::Kernel::NODE{$self->{id}};
		130
		131	Scalar::Util::weaken $self;
		132
		133	$self->{connect_to} \|\|= AE::timer $AnyEvent::MP::Kernel::CONFIG->{connect_interval}, 0, sub {
		134	$self->transport_error (transport_error => $self->{id}, "connect timeout");
		135	};
		136
		137	# maybe @$addresses?
		138	my $addresses = $AnyEvent::MP::Kernel::GLOBAL_DB{"'l"}{$self->{id}};
		139
		140	if ($addresses) {
		141	$self->connect_to ($addresses);
		142	} else {
		143	# on global nodes, all bets are off now - we either know the node, or we don't
		144	if ($AnyEvent::MP::Kernel::GLOBAL) {
		145	$self->transport_error (transport_error => $self->{id}, "no known address");
		146	} else {
		147	AnyEvent::MP::Kernel::g_find ($self->{id});
40	}	148	}
41	shift->send (join ",", @refs);
42	});	149	}
		150	}
43		151
44	$noderef = $DEFAULT_PORT unless length $noderef;	152	sub connect_to {
		153	my ($self, $addresses) = @_;
45		154
46	my $idx;	155	return if $self->{transport};
47	for my $t (split /,/, $noderef) {	156	return if $self->{connect_w};
48	my $pri = ++$idx;
49
50	#TODO: this should be outside normalise_noderef and in become_public
51	if ($t =~ /^\d*$/) {
52	require POSIX;
53	my $nodename = (POSIX::uname ())[1];
54		157
55	$cv->begin;	158	unless (@$addresses) {
56	AnyEvent::Socket::resolve_sockaddr $nodename, $t \|\| "aemp=$DEFAULT_PORT", "tcp", 0, undef, sub {	159	$self->transport_error (transport_error => $self->{id}, "no known address");
57	for (@_) {	160	return;
58	my ($service, $host) = AnyEvent::Socket::unpack_sockaddr $_->[3];	161	}
59	push @res, [	162
60	$pri += 1e-5,	163	AE::log 9 => "connecting to $self->{id} with [@$addresses]";
61	AnyEvent::Socket::format_hostport AnyEvent::Socket::format_address $host, $service
62	];
63	}
64	$cv->end;
65	};
66		164
67	# my (undef, undef, undef, undef, @ipv4) = gethostbyname $nodename;	165	my $monitor = $AnyEvent::MP::Kernel::CONFIG->{monitor_timeout};
68	#	166	my $interval = $AnyEvent::MP::Kernel::CONFIG->{connect_interval};
69	# for (@ipv4) {	167
70	# push @res, [	168	$interval = ($monitor - $interval) / @$addresses
71	# $pri,	169	if ($monitor - $interval) / @$addresses < $interval;
72	# AnyEvent::Socket::format_hostport AnyEvent::Socket::format_address $_, $t \|\| $DEFAULT_PORT,	170
73	# ];	171	$interval = 0.4 if $interval < 0.4;
74	# }	172
75	} else {	173	my @endpoints = reverse @$addresses;
		174
		175	$self->{connect_w} = AE::timer 0, $interval * (0.9 + 0.1 * rand), sub {
		176	my $endpoint = pop @endpoints;
		177
		178	AE::log 9 => "connecting to $self->{id} at $endpoint";
		179
		180	$self->{trial}{$endpoint} \|\|= do {
		181	use Carp;AnyEvent::Socket::parse_hostport $endpoint or Carp::confess "not a reslved blabla\n";#d#
76	my ($host, $port) = AnyEvent::Socket::parse_hostport $t, "aemp=$DEFAULT_PORT"	182	my ($host, $port) = AnyEvent::Socket::parse_hostport $endpoint
77	or Carp::croak "$t: unparsable transport descriptor";	183	or return AE::log critical => "$self->{id}: '$endpoint' is not a resolved node reference.";
78		184
79	$cv->begin;	185	AnyEvent::MP::Transport::mp_connect
80	AnyEvent::Socket::resolve_sockaddr $host, $port, "tcp", 0, undef, sub {	186	$host, $port,
81	for (@_) {	187	sub { delete $self->{trial}{$endpoint} },
82	my ($service, $host) = AnyEvent::Socket::unpack_sockaddr $_->[3];
83	push @res, [
84	$pri += 1e-5,
85	AnyEvent::Socket::format_hostport AnyEvent::Socket::format_address $host, $service
86	];
87	}
88	$cv->end;
89	}
90	}	188	};
91	}	189	};
92
93	$cv->end;
94
95	$cv
96	}
97
98	sub new {
99	my ($class, $noderef) = @_;
100
101	bless { noderef => $noderef }, $class
102	}
103
104	package AnyEvent::MP::Node::Direct;
105
106	use base "AnyEvent::MP::Node";
107
108	sub send {
109	my ($self, $msg) = @_;
110
111	if ($self->{transport}) {
112	$self->{transport}->send ($msg);
113	} elsif ($self->{queue}) {
114	push @{ $self->{queue} }, $msg;
115	} else {
116	$self->{queue} = [$msg];
117	$self->connect;
118	}
119	}	190	}
120		191
121	sub kill {	192	sub kill {
122	my ($self, $port, @reason) = @_;	193	my ($self, $port, @reason) = @_;
123		194
124	$self->send (["", kil => $port, @reason]);	195	$self->{send} (["", kil1 => $port, @reason]);
125
126	# delete $AnyEvent::MP::Base::PORT{$port};
127
128	# my $mon = delete $AnyEvent::MP::Base::LMON{$port}
129	# or return;
130
131	# $_->(@reason) for values %$mon;
132	}	196	}
133		197
134	sub monitor {	198	sub monitor {
135	my ($self, $portid, $cb) = @_;	199	my ($self, $portid, $cb) = @_;
136		200
137	return $cb->()
138	if $self->{failed};
139
140	my $list = $self->{lmon}{$portid} \|\|= [];	201	my $list = $self->{lmon}{$portid} \|\|= [];
141		202
142	$self->send (["", mon1 => $portid])	203	$self->send (["", mon1 => $portid])
143	unless @$list;	204	unless @$list \|\| !length $portid;
144		205
145	push @$list, $cb;	206	push @$list, $cb;
146	}	207	}
147		208
148	sub unmonitor {	209	sub unmonitor {
157	$self->send (["", mon0 => $portid]);	218	$self->send (["", mon0 => $portid]);
158	delete $self->{monitor}{$portid};	219	delete $self->{monitor}{$portid};
159	}	220	}
160	}	221	}
161		222
		223	package AnyEvent::MP::Node::Self; # the local node
		224
		225	use base "AnyEvent::MP::Node";
		226
		227	sub connect {
		228	# we are trivially connected
		229	}
		230
		231	# delay every so often to avoid recursion, also used to delay after spawn
		232	our $DELAY = -50;
		233	our @DELAY;
		234	our $DELAY_W;
		235
		236	our $send_delayed = sub {
		237	$AnyEvent::MP::Kernel::SRCNODE = $AnyEvent::MP::Kernel::NODE;
		238	(shift @DELAY)->()
		239	while @DELAY;
		240	undef $DELAY_W;
		241	$DELAY = -50;
		242	};
		243
162	sub set_transport {	244	sub transport_reset {
163	my ($self, $transport) = @_;
164
165	$self->clr_transport
166	if $self->{transport};
167
168	delete $self->{trial};
169	delete $self->{next_connect};
170	delete $self->{failed};
171
172	if (
173	exists $self->{remote_uniq}
174	&& $self->{remote_uniq} ne $transport->{remote_uniq}
175	) {
176	# uniq changed, drop queue
177	delete $self->{queue};
178	#TODO: "DOWN"
179	}
180
181	$self->{remote_uniq} = $transport->{remote_uniq};
182	$self->{transport} = $transport;
183
184	$transport->send ($_)
185	for @{ delete $self->{queue} \|\| [] };
186	}
187
188	sub clr_transport {
189	my ($self) = @_;	245	my ($self) = @_;
190		246
191	delete $self->{transport};
192	$self->{failed} = 1;
193
194	if (my $mon = delete $self->{monitor}) {
195	$_->() for map @$_, values %$mon;
196	}
197
198	$self->connect;
199	}
200
201	sub connect {
202	my ($self) = @_;
203
204	Scalar::Util::weaken $self;	247	Scalar::Util::weaken $self;
205		248
206	unless (exists $self->{n_noderef}) {	249	$self->{send} = sub {
207	return if $self->{n_noderef_}++;	250	if (++$DELAY > 0) {
208	(AnyEvent::MP::Node::normalise_noderef ($self->{noderef}))->cb (sub {	251	my $msg = $_[0];
		252	push @DELAY, sub { AnyEvent::MP::Kernel::_inject (@$msg) };
		253	$DELAY_W \|\|= AE::timer 0, 0, $send_delayed;
209	$self or return;	254	return;
210	delete $self->{n_noderef_};
211	my $noderef = shift->recv;
212
213	$self->{n_noderef} = $noderef;
214
215	$AnyEvent::MP::Base::NODE{$_} = $self
216	for split /,/, $noderef;
217
218	$self->connect;
219	});	255	}
		256
		257	local $AnyEvent::MP::Kernel::SRCNODE = $AnyEvent::MP::Kernel::NODE;
		258	AnyEvent::MP::Kernel::_inject (@{ $_[0] });
		259	};
		260	}
		261
		262	sub transport_connect {
		263	my ($self, $tp) = @_;
		264
		265	AE::log 9 => "I refuse to talk to myself ($tp->{peerhost}:$tp->{peerport})";
		266	}
		267
		268	sub kill {
		269	my (undef, @args) = @_;
		270
		271	# we _always_ delay kil's, to avoid calling mon callbacks
		272	# from anything but the event loop context.
		273	$DELAY = 1;
		274	push @DELAY, sub { AnyEvent::MP::Kernel::_kill (@args) };
		275	$DELAY_W \|\|= AE::timer 0, 0, $send_delayed;
		276	}
		277
		278	sub monitor {
		279	# maybe always delay, too?
		280	if ($DELAY_W) {
		281	my @args = @_;
		282	push @DELAY, sub { AnyEvent::MP::Kernel::_monitor (@args) };
220	return;	283	return;
221	}	284	}
222		285	&AnyEvent::MP::Kernel::_monitor;
223	$self->{retry} \|\|= [split /,/, $self->{n_noderef}];
224
225	my $endpoint = shift @{ $self->{retry} };
226
227	if (defined $endpoint) {
228	$self->{trial}{$endpoint} \|\|= do {
229	my ($host, $port) = AnyEvent::Socket::parse_hostport $endpoint
230	or return;
231
232	my ($w, $g);
233
234	$w = AE::timer $AnyEvent::MP::Base::CONNECT_TIMEOUT, 0, sub {
235	delete $self->{trial}{$endpoint};
236	};
237	$g = AnyEvent::MP::Transport::mp_connect
238	$host, $port,
239	sub {
240	delete $self->{trial}{$endpoint}
241	unless @_;
242	$g = shift;
243	};
244	;
245
246	[$w, \$g]
247	};
248	} else {
249	delete $self->{retry};
250	}
251
252	$self->{next_connect} = AE::timer $AnyEvent::MP::Base::CONNECT_INTERVAL, 0, sub {
253	$self->connect;
254	};
255	}
256
257	package AnyEvent::MP::Node::Self;
258
259	use base "AnyEvent::MP::Node";
260
261	sub set_transport {
262	die "FATAL error, set_transport was called";
263	}
264
265	sub send {
266	local $AnyEvent::MP::Base::SRCNODE = $_[0];
267	AnyEvent::MP::Base::_inject (@{ $_[1] });
268	}
269
270	sub kill {
271	my ($self, $port, @reason) = @_;
272
273	delete $AnyEvent::MP::Base::PORT{$port};
274
275	my $mon = delete $AnyEvent::MP::Base::LMON{$port}
276	or return;
277
278	$_->(@reason) for values %$mon;
279	}
280
281	sub monitor {
282	my ($self, $portid, $cb) = @_;
283
284	return $cb->()
285	unless exists $AnyEvent::MP::Base::PORT{$portid};
286
287	$AnyEvent::MP::Base::LMON{$portid}{$cb+0} = $cb;
288	}	286	}
289		287
290	sub unmonitor {	288	sub unmonitor {
291	my ($self, $portid, $cb) = @_;	289	# no need to always delay
		290	if ($DELAY_W) {
		291	my @args = @_;
		292	push @DELAY, sub { AnyEvent::MP::Kernel::_unmonitor (@args) };
		293	return;
		294	}
292		295
293	delete $AnyEvent::MP::Base::LMON{$portid}{$cb+0};	296	&AnyEvent::MP::Kernel::_unmonitor;
294	}	297	}
295		298
296	=head1 SEE ALSO	299	=head1 SEE ALSO
297		300
298	L<AnyEvent>.	301	L<AnyEvent::MP>.
299		302
300	=head1 AUTHOR	303	=head1 AUTHOR
301		304
302	Marc Lehmann <schmorp@schmorp.de>	305	Marc Lehmann <schmorp@schmorp.de>
303	http://home.schmorp.de/	306	http://home.schmorp.de/

Diff Legend

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

Comparing AnyEvent-MP/MP/Node.pm (file contents): Revision 1.7 by root, Tue Aug 4 14:10:51 2009 UTC vs. Revision 1.66 by root, Sat Mar 24 00:48:54 2012 UTC

Diff Legend

Comparing AnyEvent-MP/MP/Node.pm (file contents):
Revision 1.7 by root, Tue Aug 4 14:10:51 2009 UTC vs.
Revision 1.66 by root, Sat Mar 24 00:48:54 2012 UTC