| … | |
… | |
| 140 | return if $self->{transport}; |
140 | return if $self->{transport}; |
| 141 | return if $self->{connect_w}; |
141 | return if $self->{connect_w}; |
| 142 | |
142 | |
| 143 | return unless @$addresses; |
143 | return unless @$addresses; |
| 144 | |
144 | |
| 145 | $AnyEvent::MP::Kernel::WARN->(9, "connecting to $self->{id} with [@$addresses]"); |
145 | AE::log 9 => "connecting to $self->{id} with [@$addresses]"; |
| 146 | |
146 | |
| 147 | my $monitor = $AnyEvent::MP::Kernel::CONFIG->{monitor_timeout}; |
147 | my $monitor = $AnyEvent::MP::Kernel::CONFIG->{monitor_timeout}; |
| 148 | my $interval = $AnyEvent::MP::Kernel::CONFIG->{connect_interval}; |
148 | my $interval = $AnyEvent::MP::Kernel::CONFIG->{connect_interval}; |
| 149 | |
149 | |
| 150 | $interval = ($monitor - $interval) / @$addresses |
150 | $interval = ($monitor - $interval) / @$addresses |
| … | |
… | |
| 155 | my @endpoints = reverse @$addresses; |
155 | my @endpoints = reverse @$addresses; |
| 156 | |
156 | |
| 157 | $self->{connect_w} = AE::timer 0, $interval * (0.9 + 0.1 * rand), sub { |
157 | $self->{connect_w} = AE::timer 0, $interval * (0.9 + 0.1 * rand), sub { |
| 158 | my $endpoint = pop @endpoints; |
158 | my $endpoint = pop @endpoints; |
| 159 | |
159 | |
| 160 | $AnyEvent::MP::Kernel::WARN->(9, "connecting to $self->{id} at $endpoint"); |
160 | AE::log 9 => "connecting to $self->{id} at $endpoint"; |
| 161 | |
161 | |
| 162 | $self->{trial}{$endpoint} ||= do { |
162 | $self->{trial}{$endpoint} ||= do { |
| 163 | my ($host, $port) = AnyEvent::Socket::parse_hostport $endpoint |
163 | my ($host, $port) = AnyEvent::Socket::parse_hostport $endpoint |
| 164 | or return $AnyEvent::MP::Kernel::WARN->(1, "$self->{id}: not a resolved node reference."); |
164 | or return AE::log critical => "$self->{id}: '$endpoint' is not a resolved node reference."; |
| 165 | |
|
|
| 166 | AnyEvent::MP::Transport::mp_connect |
|
|
| 167 | $host, $port, |
|
|
| 168 | sub { delete $self->{trial}{$endpoint} }, |
|
|
| 169 | }; |
|
|
| 170 | }; |
|
|
| 171 | } |
|
|
| 172 | |
|
|
| 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 | |
|
|
| 207 | my @endpoints; |
|
|
| 208 | |
|
|
| 209 | $self->{connect_w} = AE::timer 0, $interval * (0.9 + 0.1 * rand), sub { |
|
|
| 210 | @endpoints = @addresses |
|
|
| 211 | unless @endpoints; |
|
|
| 212 | |
|
|
| 213 | my $endpoint = shift @endpoints; |
|
|
| 214 | |
|
|
| 215 | $AnyEvent::MP::Kernel::WARN->(9, "connecting to $self->{id} at $endpoint"); |
|
|
| 216 | |
|
|
| 217 | $self->{trial}{$endpoint} ||= do { |
|
|
| 218 | my ($host, $port) = AnyEvent::Socket::parse_hostport $endpoint |
|
|
| 219 | or return $AnyEvent::MP::Kernel::WARN->(1, "$self->{id}: not a resolved node reference."); |
|
|
| 220 | |
165 | |
| 221 | AnyEvent::MP::Transport::mp_connect |
166 | AnyEvent::MP::Transport::mp_connect |
| 222 | $host, $port, |
167 | $host, $port, |
| 223 | sub { delete $self->{trial}{$endpoint} }, |
168 | sub { delete $self->{trial}{$endpoint} }, |
| 224 | }; |
169 | }; |
| … | |
… | |
| 296 | } |
241 | } |
| 297 | |
242 | |
| 298 | sub transport_connect { |
243 | sub transport_connect { |
| 299 | my ($self, $tp) = @_; |
244 | my ($self, $tp) = @_; |
| 300 | |
245 | |
| 301 | $AnyEvent::MP::Kernel::WARN->(9, "I refuse to talk to myself ($tp->{peerhost}:$tp->{peerport})"); |
246 | AE::log 9 => "I refuse to talk to myself ($tp->{peerhost}:$tp->{peerport})"; |
| 302 | } |
247 | } |
| 303 | |
248 | |
| 304 | sub kill { |
249 | sub kill { |
| 305 | my (undef, @args) = @_; |
250 | my (undef, @args) = @_; |
| 306 | |
251 | |
