| 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 | |
| … | |
… | |
| 54 | |
56 | |
| 55 | $self->{send} = sub { |
57 | $self->{send} = sub { |
| 56 | push @{$self->{queue}}, shift; |
58 | push @{$self->{queue}}, shift; |
| 57 | $self->connect; |
59 | $self->connect; |
| 58 | }; |
60 | }; |
| 59 | |
|
|
| 60 | $self->connect |
|
|
| 61 | if $self->{autoconnect}; |
|
|
| 62 | } |
61 | } |
| 63 | |
62 | |
| 64 | # called only after successful handshake |
63 | # called only after successful handshake |
| 65 | sub transport_error { |
64 | sub transport_error { |
| 66 | my ($self, @reason) = @_; |
65 | my ($self, @reason) = @_; |
| … | |
… | |
| 71 | delete $self->{connect_to}; |
70 | delete $self->{connect_to}; |
| 72 | |
71 | |
| 73 | delete $self->{queue}; |
72 | delete $self->{queue}; |
| 74 | $self->transport_reset; |
73 | $self->transport_reset; |
| 75 | |
74 | |
| 76 | AnyEvent::MP::Kernel::_inject_nodeevent ($self, 0, @reason) |
|
|
| 77 | unless $no_transport; |
|
|
| 78 | |
|
|
| 79 | if (my $mon = delete $self->{lmon}) { |
75 | if (my $mon = delete $self->{lmon}) { |
| 80 | $_->(@reason) for map @$_, values %$mon; |
76 | $_->(@reason) for map @$_, values %$mon; |
| 81 | } |
77 | } |
|
|
78 | |
|
|
79 | AnyEvent::MP::Kernel::_inject_nodeevent ($self, 0, @reason) |
|
|
80 | unless $no_transport; |
| 82 | } |
81 | } |
| 83 | |
82 | |
| 84 | # called after handshake was successful |
83 | # called after handshake was successful |
| 85 | sub transport_connect { |
84 | sub transport_connect { |
| 86 | my ($self, $transport) = @_; |
85 | my ($self, $transport) = @_; |
| … | |
… | |
| 119 | 0, |
118 | 0, |
| 120 | sub { |
119 | sub { |
| 121 | $self->transport_error (transport_error => $self->{id}, "unable to connect"); |
120 | $self->transport_error (transport_error => $self->{id}, "unable to connect"); |
| 122 | }; |
121 | }; |
| 123 | |
122 | |
|
|
123 | return unless @addresses; |
|
|
124 | |
| 124 | $AnyEvent::MP::Kernel::WARN->(9, "connecting to $self->{id} with [@addresses]"); |
125 | $AnyEvent::MP::Kernel::WARN->(9, "connecting to $self->{id} with [@addresses]"); |
| 125 | return unless @addresses; |
|
|
| 126 | |
126 | |
| 127 | unless ($self->{connect_w}) { |
127 | unless ($self->{connect_w}) { |
| 128 | my @endpoints; |
128 | my @endpoints; |
| 129 | |
129 | |
| 130 | $self->{connect_w} = AE::timer |
130 | $self->{connect_w} = AE::timer |
