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

Comparing AnyEvent-MP/MP/Node.pm (file contents):
Revision 1.22 by root, Thu Aug 13 22:50:47 2009 UTC vs.
Revision 1.60 by root, Wed Mar 21 15:22:16 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
		11	This is an internal utility module, horrible to look at, so don't.
10		12
11	=cut	13	=cut
12		14
13	package AnyEvent::MP::Node;	15	package AnyEvent::MP::Node;
14		16
19	use AnyEvent::Socket ();	21	use AnyEvent::Socket ();
20		22
21	use AnyEvent::MP::Transport ();	23	use AnyEvent::MP::Transport ();
22		24
23	sub new {	25	sub new {
24	my ($self, $noderef) = @_;	26	my ($self, $id) = @_;
25		27
26	$self = bless { noderef => $noderef }, $self;	28	$self = bless { id => $id }, $self;
27		29
		30	$self->init;
28	$self->transport_reset;	31	$self->transport_reset;
29		32
30	$self	33	$self
		34	}
		35
		36	sub init {
		37	#
31	}	38	}
32		39
33	sub send {	40	sub send {
34	&{ shift->{send} }	41	&{ shift->{send} }
35	}	42	}
36		43
		44	# nodes reachable via the network
37	package AnyEvent::MP::Node::External;	45	package AnyEvent::MP::Node::Remote;
38		46
39	use base "AnyEvent::MP::Node";	47	use base "AnyEvent::MP::Node";
40		48
41	# called at init time, mostly sets {send}	49	# called at init time, mostly sets {send}
42	sub transport_reset {	50	sub transport_reset {
…		…
50	push @{$self->{queue}}, shift;	58	push @{$self->{queue}}, shift;
51	$self->connect;	59	$self->connect;
52	};	60	};
53	}	61	}
54		62
55	# called only after successful handshake	63	# called each time we fail to establish a connection,
		64	# or the existing connection failed
56	sub transport_error {	65	sub transport_error {
57	my ($self, @reason) = @_;	66	my ($self, @reason) = @_;
58		67
59	my $no_transport = !$self->{transport};	68	my $no_transport = !$self->{transport};
60		69
…		…
62	delete $self->{connect_to};	71	delete $self->{connect_to};
63		72
64	delete $self->{queue};	73	delete $self->{queue};
65	$self->transport_reset;	74	$self->transport_reset;
66		75
		76	if (my $mon = delete $self->{lmon}) {
		77	$_->(@reason) for map @$_, values %$mon;
		78	}
		79
67	AnyEvent::MP::Kernel::_inject_nodeevent ($self, 0, @reason)	80	AnyEvent::MP::Kernel::_inject_nodeevent ($self, 0, @reason)
68	unless $no_transport;	81	unless $no_transport;
69		82
70	if (my $mon = delete $self->{lmon}) {	83	# if we are here and are idle, we nuke ourselves
71	$_->(@reason) for map @$_, values %$mon;	84	delete $AnyEvent::MP::Kernel::NODE{$self->{id}}
72	}	85	unless $self->{transport} \|\| $self->{connect_to};
73	}	86	}
74		87
75	# called after handshake was successful	88	# called after handshake was successful
76	sub transport_connect {	89	sub transport_connect {
77	my ($self, $transport) = @_;	90	my ($self, $transport) = @_;
78		91
		92	delete $self->{trial};
		93
79	$self->transport_error (transport_error => "switched connections")	94	$self->transport_error (transport_error => $self->{id}, "switched connections")
80	if $self->{transport};	95	if $self->{transport};
81		96
82	delete $self->{connect_w};	97	delete $self->{connect_w};
83	delete $self->{connect_to};	98	delete $self->{connect_to};
84		99
85	$self->{transport} = $transport;	100	$self->{transport} = $transport;
86		101
87	my $transport_send = $transport->can ("send");	102	my $transport_send = $transport->{send};
88
89	$self->{send} = sub {
90	$transport_send->($transport, $_[0]);
91	};
92		103
93	AnyEvent::MP::Kernel::_inject_nodeevent ($self, 1);	104	AnyEvent::MP::Kernel::_inject_nodeevent ($self, 1);
94		105
		106	$self->{send} = $transport_send;
		107
95	$transport->send ($_)	108	$transport_send->($_)
96	for @{ delete $self->{queue} \|\| [] };	109	for @{ delete $self->{queue} \|\| [] };
97	}	110	}
98		111
99	sub connect {	112	sub connect {
100	my ($self) = @_;	113	my ($self) = @_;
101		114
		115	return if $self->{transport};
		116	return if $self->{connect_w};
		117
102	Scalar::Util::weaken $self;	118	Scalar::Util::weaken $self;
103		119
104	$self->{connect_to} \|\|= AE::timer	120	$self->{connect_to} \|\|= AE::timer $AnyEvent::MP::Kernel::CONFIG->{connect_interval}, 0, sub {
105	$AnyEvent::MP::Config::CFG{monitor_timeout} \|\| $AnyEvent::MP::Kernel::MONITOR_TIMEOUT,
106	0,
107	sub {
108	$self->transport_error (transport_error => $self->{noderef}, "unable to connect");	121	$self->transport_error (transport_error => $self->{id}, "unable to connect");
		122	};
		123
		124	# maybe @$addresses?
		125	my $addresses = $AnyEvent::MP::Kernel::GLOBAL_DB{"'l"}{$self->{id}};
		126
		127	if ($addresses) {
		128	$self->connect_to ($addresses);
		129	} else {
		130	# on global nodes, all bets are off now - we either know the node, or we don't
		131	unless ($AnyEvent::MP::Kernel::GLOBAL) {
		132	AnyEvent::MP::Kernel::g_find ($self->{id});
		133	}
		134	}
		135	}
		136
		137	sub connect_to {
		138	my ($self, $addresses) = @_;
		139
		140	return if $self->{transport};
		141	return if $self->{connect_w};
		142
		143	return unless @$addresses;
		144
		145	$AnyEvent::MP::Kernel::WARN->(9, "connecting to $self->{id} with [@$addresses]");
		146
		147	my $monitor = $AnyEvent::MP::Kernel::CONFIG->{monitor_timeout};
		148	my $interval = $AnyEvent::MP::Kernel::CONFIG->{connect_interval};
		149
		150	$interval = ($monitor - $interval) / @$addresses
		151	if ($monitor - $interval) / @$addresses < $interval;
		152
		153	$interval = 0.4 if $interval < 0.4;
		154
		155	my @endpoints = reverse @$addresses;
		156
		157	$self->{connect_w} = AE::timer 0, $interval * (0.9 + 0.1 * rand), sub {
		158	my $endpoint = pop @endpoints;
		159
		160	$AnyEvent::MP::Kernel::WARN->(9, "connecting to $self->{id} at $endpoint");
		161
		162	$self->{trial}{$endpoint} \|\|= do {
		163	my ($host, $port) = AnyEvent::Socket::parse_hostport $endpoint
		164	or return $AnyEvent::MP::Kernel::WARN->(1, "$self->{id}: not a resolved node reference.");
		165
		166	AnyEvent::MP::Transport::mp_connect
		167	$host, $port,
		168	sub { delete $self->{trial}{$endpoint} },
109	};	169	};
		170	};
		171	}
110		172
111	unless ($self->{connect_w}) {	173	sub connect {
		174	my ($self) = @_;
		175
		176	return if $self->{transport};
		177	return if $self->{connect_w};
		178
		179	Scalar::Util::weaken $self;
		180
		181	my $monitor = $AnyEvent::MP::Kernel::CONFIG->{monitor_timeout};
		182
		183	$self->{connect_to} \|\|= AE::timer $monitor, 0, sub {
		184	$self->transport_error (transport_error => $self->{id}, "unable to connect");
		185	};
		186
		187	# maybe @$addresses?
		188	my @addresses = @{ $AnyEvent::MP::Kernel::GLOBAL_DB{"'l"}{$self->{id}} };
		189
		190	unless (@addresses) {
		191	# on global nodes, all bets are off now - we either know the node, or we don't
		192	unless ($AnyEvent::MP::Kernel::GLOBAL) {
		193	AnyEvent::MP::Kernel::g_find ($self->{id});
		194	}
		195	return;
		196	}
		197
		198	$AnyEvent::MP::Kernel::WARN->(9, "connecting to $self->{id} with [@addresses]");
		199
		200	my $interval = $AnyEvent::MP::Kernel::CONFIG->{connect_interval};
		201
		202	$interval = ($monitor - $interval) / @addresses
		203	if ($monitor - $interval) / @addresses < $interval;
		204
		205	$interval = 0.4 if $interval < 0.4;
		206
112	my @endpoints;	207	my @endpoints;
113	my %trial;
114		208
115	$self->{connect_w} = AE::timer	209	$self->{connect_w} = AE::timer 0, $interval * (0.9 + 0.1 * rand), sub {
116	0,	210	@endpoints = @addresses
117	$AnyEvent::MP::Config::CFG{connect_interval} \|\| $AnyEvent::MP::Kernel::CONNECT_INTERVAL,
118	sub {
119	@endpoints = split /,/, $self->{noderef}
120	unless @endpoints;	211	unless @endpoints;
121		212
122	my $endpoint = shift @endpoints;	213	my $endpoint = shift @endpoints;
123		214
		215	$AnyEvent::MP::Kernel::WARN->(9, "connecting to $self->{id} at $endpoint");
		216
124	$trial{$endpoint} \|\|= do {	217	$self->{trial}{$endpoint} \|\|= do {
125	my ($host, $port) = AnyEvent::Socket::parse_hostport $endpoint	218	my ($host, $port) = AnyEvent::Socket::parse_hostport $endpoint
126	or return $AnyEvent::MP::Kernel::WARN->("$self->{noderef}: not a resolved node reference.");	219	or return $AnyEvent::MP::Kernel::WARN->(1, "$self->{id}: not a resolved node reference.");
127		220
128	AnyEvent::MP::Transport::mp_connect	221	AnyEvent::MP::Transport::mp_connect
129	$host, $port,	222	$host, $port,
130	sub { delete $trial{$endpoint} }	223	sub { delete $self->{trial}{$endpoint} },
131	;
132	};
133	}
134	;	224	};
135	}	225	};
136	}	226	}
137		227
138	sub kill {	228	sub kill {
139	my ($self, $port, @reason) = @_;	229	my ($self, $port, @reason) = @_;
140		230
141	$self->send (["", kil => $port, @reason]);	231	$self->{send} (["", kil => $port, @reason]);
142	}	232	}
143		233
144	sub monitor {	234	sub monitor {
145	my ($self, $portid, $cb) = @_;	235	my ($self, $portid, $cb) = @_;
146		236
147	my $list = $self->{lmon}{$portid} \|\|= [];	237	my $list = $self->{lmon}{$portid} \|\|= [];
148		238
149	$self->send (["", mon1 => $portid])	239	$self->send (["", mon1 => $portid])
150	unless @$list;	240	unless @$list \|\| !length $portid;
151		241
152	push @$list, $cb;	242	push @$list, $cb;
153	}	243	}
154		244
155	sub unmonitor {	245	sub unmonitor {
…		…
164	$self->send (["", mon0 => $portid]);	254	$self->send (["", mon0 => $portid]);
165	delete $self->{monitor}{$portid};	255	delete $self->{monitor}{$portid};
166	}	256	}
167	}	257	}
168		258
169	package AnyEvent::MP::Node::Direct;	259	package AnyEvent::MP::Node::Self;
170		260
171	use base "AnyEvent::MP::Node::External";
172
173	package AnyEvent::MP::Node::Indirect;
174
175	use base "AnyEvent::MP::Node::Direct";	261	use base "AnyEvent::MP::Node";
		262
		263	sub connect {
		264	# we are trivially connected
		265	}
		266
		267	# delay every so often to avoid recursion, also used to delay after spawn
		268	our $DELAY = -50;
		269	our @DELAY;
		270	our $DELAY_W;
		271
		272	sub _send_delayed {
		273	$AnyEvent::MP::Kernel::SRCNODE = $AnyEvent::MP::Kernel::NODE;
		274	(shift @DELAY)->()
		275	while @DELAY;
		276	undef $DELAY_W;
		277	$DELAY = -50;
		278	}
176		279
177	sub transport_reset {	280	sub transport_reset {
178	my ($self) = @_;	281	my ($self) = @_;
179		282
180	# as an optimisation, immediately nuke slave nodes
181	delete $AnyEvent::MP::Kernel::NODE{$self->{noderef}}
182	if $self->{transport};
183
184	$self->SUPER::transport_reset;
185	}
186
187	sub connect {
188	my ($self) = @_;
189
190	$self->transport_error (transport_error => $self->{noderef}, "unable to connect to indirect node");
191	}
192
193	package AnyEvent::MP::Node::Self;
194
195	use base "AnyEvent::MP::Node";
196
197	sub connect {
198	# we are trivially connected
199	}
200
201	sub transport_reset {
202	my ($self) = @_;
203
204	Scalar::Util::weaken $self;	283	Scalar::Util::weaken $self;
205		284
206	$self->{send} = sub {	285	$self->{send} = sub {
		286	if ($DELAY++ >= 0) {
		287	my $msg = $_[0];
		288	push @DELAY, sub { AnyEvent::MP::Kernel::_inject (@$msg) };
		289	$DELAY_W \|\|= AE::timer 0, 0, \&_send_delayed;
		290	return;
		291	}
		292
207	local $AnyEvent::MP::Kernel::SRCNODE = $self;	293	local $AnyEvent::MP::Kernel::SRCNODE = $AnyEvent::MP::Kernel::NODE;
208	AnyEvent::MP::Kernel::_inject (@{ $_[0] });	294	AnyEvent::MP::Kernel::_inject (@{ $_[0] });
209	};	295	};
210	}	296	}
211		297
		298	sub transport_connect {
		299	my ($self, $tp) = @_;
		300
		301	$AnyEvent::MP::Kernel::WARN->(9, "I refuse to talk to myself ($tp->{peerhost}:$tp->{peerport})");
		302	}
		303
212	sub kill {	304	sub kill {
213	my ($self, $port, @reason) = @_;	305	my (undef, @args) = @_;
214		306
215	delete $AnyEvent::MP::Kernel::PORT{$port};	307	# we _always_ delay kil's, to avoid calling mon callbacks
216	delete $AnyEvent::MP::Kernel::PORT_DATA{$port};	308	# from anything but the event loop context.
217		309	$DELAY = 1;
218	my $mon = delete $AnyEvent::MP::Kernel::LMON{$port}	310	push @DELAY, sub { AnyEvent::MP::Kernel::_kill (@args) };
219	or !@reason	311	$DELAY_W \|\|= AE::timer 0, 0, \&_send_delayed;
220	or $AnyEvent::MP::Kernel::WARN->("unmonitored local port $port died with reason: @reason");
221
222	$_->(@reason) for values %$mon;
223	}	312	}
224		313
225	sub monitor {	314	sub monitor {
226	my ($self, $portid, $cb) = @_;	315	# maybe always delay, too?
227		316	if ($DELAY_W) {
228	return $cb->(no_such_port => "cannot monitor nonexistent port")	317	my @args = @_;
229	unless exists $AnyEvent::MP::Kernel::PORT{$portid};	318	push @DELAY, sub { AnyEvent::MP::Kernel::_monitor (@args) };
230		319	return;
231	$AnyEvent::MP::Kernel::LMON{$portid}{$cb+0} = $cb;	320	}
		321	&AnyEvent::MP::Kernel::_monitor;
232	}	322	}
233		323
234	sub unmonitor {	324	sub unmonitor {
235	my ($self, $portid, $cb) = @_;	325	# no need to always delay
		326	if ($DELAY_W) {
		327	my @args = @_;
		328	push @DELAY, sub { AnyEvent::MP::Kernel::_unmonitor (@args) };
		329	return;
		330	}
236		331
237	delete $AnyEvent::MP::Kernel::LMON{$portid}{$cb+0};	332	&AnyEvent::MP::Kernel::_unmonitor;
238	}	333	}
239		334
240	=head1 SEE ALSO	335	=head1 SEE ALSO
241		336
242	L<AnyEvent::MP>.	337	L<AnyEvent::MP>.

Diff Legend

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

Comparing AnyEvent-MP/MP/Node.pm (file contents): Revision 1.22 by root, Thu Aug 13 22:50:47 2009 UTC vs. Revision 1.60 by root, Wed Mar 21 15:22:16 2012 UTC

Diff Legend

Comparing AnyEvent-MP/MP/Node.pm (file contents):
Revision 1.22 by root, Thu Aug 13 22:50:47 2009 UTC vs.
Revision 1.60 by root, Wed Mar 21 15:22:16 2012 UTC