ViewVC Help
View File | Revision Log | Show Annotations | Download File
/cvs/AnyEvent-MP/MP/Node.pm
(Generate patch)

Comparing AnyEvent-MP/MP/Node.pm (file contents):
Revision 1.58 by root, Fri Mar 2 19:21:16 2012 UTC vs.
Revision 1.59 by root, Sat Mar 3 11:38:43 2012 UTC

… …
92 delete $self->{trial}; 92 delete $self->{trial};
93 93
94 $self->transport_error (transport_error => $self->{id}, "switched connections") 94 $self->transport_error (transport_error => $self->{id}, "switched connections")
95 if $self->{transport}; 95 if $self->{transport};
96 96
97 delete $self->{connect_addr};
98 delete $self->{connect_w}; 97 delete $self->{connect_w};
99 delete $self->{connect_to}; 98 delete $self->{connect_to};
100 99
101 $self->{transport} = $transport; 100 $self->{transport} = $transport;
102 101
… …
106 105
107 $self->{send} = $transport_send; 106 $self->{send} = $transport_send;
108 107
109 $transport_send->($_) 108 $transport_send->($_)
110 for @{ delete $self->{queue} || [] }; 109 for @{ delete $self->{queue} || [] };
110}
111
112sub connect {
113 my ($self) = @_;
114
115 return if $self->{transport};
116 return if $self->{connect_w};
117
118 Scalar::Util::weaken $self;
119
120 $self->{connect_to} ||= AE::timer $AnyEvent::MP::Kernel::CONFIG->{connect_interval}, 0, sub {
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
137sub 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} },
169 };
170 };
111} 171}
112 172
113sub connect { 173sub connect {
114 my ($self) = @_; 174 my ($self) = @_;
115 175
… …
132 unless ($AnyEvent::MP::Kernel::GLOBAL) { 192 unless ($AnyEvent::MP::Kernel::GLOBAL) {
133 AnyEvent::MP::Kernel::g_find ($self->{id}); 193 AnyEvent::MP::Kernel::g_find ($self->{id});
134 } 194 }
135 return; 195 return;
136 } 196 }
137
138 $self->{connect_addr} = \@addresses; # a bit weird, but efficient
139 197
140 $AnyEvent::MP::Kernel::WARN->(9, "connecting to $self->{id} with [@addresses]"); 198 $AnyEvent::MP::Kernel::WARN->(9, "connecting to $self->{id} with [@addresses]");
141 199
142 my $interval = $AnyEvent::MP::Kernel::CONFIG->{connect_interval}; 200 my $interval = $AnyEvent::MP::Kernel::CONFIG->{connect_interval};
143 201

Diff Legend

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