AnyEvent-MP/MP/Global.pm

=head1 NAME

AnyEvent::MP::Global - some network-global services

=head1 SYNOPSIS

   use AnyEvent::MP::Global;

=head1 DESCRIPTION

This module maintains a fully-meshed network, if possible, and tries to
ensure that we are connected to at least one other node.

It also manages named port groups - ports can register themselves in any
number of groups that will be available network-wide, which is great for
discovering services.

Running it on one node will automatically run it on all nodes, although,
at the moment, the global service is started by default anyways.

=head1 GLOBALS AND FUNCTIONS

=over 4

=cut

package AnyEvent::MP::Global;

use common::sense;
use Carp ();

use AnyEvent ();
use AnyEvent::Util ();

use AnyEvent::MP;
use AnyEvent::MP::Kernel;
use AnyEvent::MP::Transport ();

use base "Exporter";

our @EXPORT = qw(
   grp_reg
   grp_get
   grp_mon
);

our $VERSION = $AnyEvent::MP::VERSION;

our %addr; # port ID => [address...] mapping

our %port; # our rendezvous port on the other side
our %lreg; # local registry, name => [pid...]
our %lmon; # local registry monitoring name,pid => mon
our %greg; # global regstry, name => pid => undef
our %gmon; # group monitoring, group => [$cb...]

our $nodecnt;

$AnyEvent::MP::Kernel::WARN->(7, "starting global service.");

#############################################################################
# seednodes

our @SEEDS;
our %SEEDS; # just to check whether a seed is a seed
our %SEED_CONNECT;
our $SEED_WATCHER;

push @AnyEvent::MP::Transport::HOOK_CONNECT, sub {
   my $peer = $_[0]{local_greeting}{peeraddr};
   return unless exists $SEEDS{$peer};
   $SEED_CONNECT{$peer} = 2;
};

push @AnyEvent::MP::Transport::HOOK_CONNECTED, sub {
   my $peer = $_[0]{local_greeting}{peeraddr};
   return unless exists $SEEDS{$peer};
   $SEEDS{$peer} = $_[0]{remote_node};
};

push @AnyEvent::MP::Transport::HOOK_DESTROY, sub {
   delete $SEED_CONNECT{$_[0]{local_greeting}{peeraddr}};
};

sub seed_connect {
   my ($seed) = @_;

   my ($host, $port) = AnyEvent::Socket::parse_hostport $seed
      or Carp::croak "$seed: unparsable seed address";

   return if $SEED_CONNECT{$seed};

   $AnyEvent::MP::Kernel::WARN->(9, "trying connect to seed node $seed.");

   # ughhh
   $SEED_CONNECT{$seed} ||= AnyEvent::MP::Transport::mp_connect $host, $port,
      seed => $seed,
      sub {
         $SEED_CONNECT{$seed} = 1;
      },
   ;
}

sub more_seeding {
   my $int = List::Util::max 1,
                $AnyEvent::MP::Kernel::CONFIG->{connect_interval}
                * ($nodecnt ? keys %AnyEvent::MP::Kernel::NODE : 1)
                - rand;

   $SEED_WATCHER = AE::timer $int, 0, \&more_seeding;

   @SEEDS = keys %SEEDS unless @SEEDS;
   return unless @SEEDS;

   seed_connect splice @SEEDS, rand @SEEDS, 1;
}

sub avoid_seed($) {
   delete $SEEDS{$_[0]};
}

sub set_seeds(@) {
   @SEEDS{@_} = ();

   $SEED_WATCHER ||= AE::timer 5, $AnyEvent::MP::Kernel::MONITOR_TIMEOUT, \&more_seeding;

   for (1 .. keys %SEEDS) {
      after 0.100 * rand, \&more_seeding;
   }
}

# returns all (up) seed nodes, or all nodes if no seednodes are up/known
sub _route_nodes {
   my @seeds = grep node_is_up $_, values %SEEDS;
   @seeds = up_nodes unless @seeds;
   @seeds
}

#############################################################################

sub _change {
   my ($group, $add, $del) = @_;

   my $kv = $greg{$group} ||= {};

   delete @$kv{@$del};
   @$kv{@$add} = ();

   my $ports = [keys %$kv];
   $_->($ports, $add, $del) for @{ $gmon{$group} };
}

sub unreg_groups($) {
   my ($node) = @_;

   my $qr = qr/^\Q$node\E(?:#|$)/;
   my @del;

   while (my ($group, $ports) = each %greg) {
      @del = grep /$qr/, keys %$ports;
      _change $group, [], \@del
         if @del;
   }
}

sub set_groups($$) {
   my ($node, $lreg) = @_;

   while (my ($k, $v) = each %$lreg) {
      _change $k, $v, [];
   }
}

=item $guard = grp_reg $group, $port

Register the given (local!) port in the named global group C<$group>.

The port will be unregistered automatically when the port is destroyed.

When not called in void context, then a guard object will be returned that
will also cause the name to be unregistered when destroyed.

=cut

# unregister local port
sub unregister {
   my ($port, $group) = @_;

   delete $lmon{"$group\x00$port"};
   @{ $lreg{$group} } = grep $_ ne $port, @{ $lreg{$group} };

   _change $group, [], [$port];

   snd $_, reg0 => $group, $port
      for values %port;
}

# register local port
sub grp_reg($$) {
   my ($group, $port) = @_;

   port_is_local $port
      or Carp::croak "AnyEvent::MP::Global::grp_reg can only be called for local ports, caught";

   grep $_ eq $port, @{ $lreg{$group} }
      and Carp::croak "'$group': group already registered, cannot register a second time";

   $lmon{"$group\x00$port"} = mon $port, sub { unregister $port, $group };
   push @{ $lreg{$group} }, $port;

   snd $_, reg1 => $group, $port
      for values %port;

   _change $group, [$port], [];

   wantarray && AnyEvent::Util::guard { unregister $port, $group }
}

=item $ports = grp_get $group

Returns all the ports currently registered to the given group (as
read-only(!) array reference). When the group has no registered members,
return C<undef>.

=cut

sub grp_get($) {
   my @ports = keys %{ $greg{$_[0]} };
   @ports ? \@ports : undef
}

=item $guard = grp_mon $group, $callback->($ports, $add, $del)

Installs a monitor on the given group. Each time there is a change it
will be called with the current group members as an arrayref as the
first argument. The second argument only contains ports added, the third
argument only ports removed.

Unlike C<grp_get>, all three arguments will always be array-refs, even if
the array is empty. None of the arrays must be modified in any way.

The first invocation will be with the first two arguments set to the
current members, as if all of them were just added, but only when the
group is atcually non-empty.

Optionally returns a guard object that uninstalls the watcher when it is
destroyed.

=cut

sub grp_mon($$) {
   my ($grp, $cb) = @_;

   AnyEvent::MP::Kernel::delay sub {
      return unless $cb;

      push @{ $gmon{$grp} }, $cb;
      $cb->(((grp_get $grp) || return) x 2, []);
   };

   defined wantarray && AnyEvent::Util::::guard {
      my @mon = grep $_ != $cb, @{ delete $gmon{$grp} };
      $gmon{$grp} = \@mon if @mon;
      undef $cb;
   }
}

sub start_node {
   my ($node) = @_;

   return if exists $port{$node};
   return if $node eq $NODE; # do not connect to ourselves

   # establish connection
   my $port = $port{$node} = spawn $node, "AnyEvent::MP::Global::connect", 0, $NODE;

   mon $port, sub {
      unreg_groups $node;
      delete $port{$node};
   };

   snd $port, addr => $AnyEvent::MP::Kernel::LISTENER;
   snd $port, nodes => \%addr if %addr;
   snd $port, set => \%lreg if %lreg;
}

# other nodes connect via this
sub connect {
   my ($version, $node) = @_;

   # monitor them, silently die
   mon $node, psub { kil $SELF };

   rcv $SELF,
      addr => sub {
         my $addresses = shift;
         $AnyEvent::MP::Kernel::WARN->(9, "$node told us its addresses (@$addresses).");
         $addr{$node} = $addresses;

         # to help listener-less nodes, we broadcast new addresses to them unconditionally
         #TODO: should be done by a node finding out about a listener-less one
         if (@$addresses) {
            for my $other (values %AnyEvent::MP::Kernel::NODE) {
               if ($other->{transport}) {
                  if ($addr{$other->{id}} && !@{ $addr{$other->{id}} }) {
                     $AnyEvent::MP::Kernel::WARN->(9, "helping $other->{id} to find $node.");
                     snd $port{$other->{id}}, nodes => { $node => $addresses };
                  }
               }
            }
         }
      },
      nodes => sub {
         my ($kv) = @_;

         use JSON::XS;#d#
         my $kv_txt = JSON::XS->new->encode ($kv);#d#
         $AnyEvent::MP::Kernel::WARN->(9, "$node told us it knows about $kv_txt.");#d#

         while (my ($id, $addresses) = each %$kv) {
            my $node = AnyEvent::MP::Kernel::add_node $id;
            $node->connect (@$addresses);
            start_node $id;
         }
      },
      find => sub {
         my ($othernode) = @_;

         $AnyEvent::MP::Kernel::WARN->(9, "$node asked us to find $othernode.");
         snd $port{$node}, nodes => { $othernode => $addr{$othernode} }
            if $addr{$othernode};
      },
      set => sub {
         set_groups $node, shift;
      },
      reg0 => sub {
         _change $_[0], [], [$_[1]];
      },
      reg1 => sub {
         _change $_[0], [$_[1]], [];
      },
   ;
}

sub mon_node {
   my ($node, $is_up) = @_;

   if ($is_up) {
      ++$nodecnt;
      start_node $node;
   } else {
      --$nodecnt;
      more_seeding unless $nodecnt;
      unreg_groups $node;

      # forget about the node
      delete $addr{$node};
      # ask other nodes if they know the node
      snd $_, find => $node
         for values %port;
   }
   #warn "node<$node,$is_up>\n";#d#
}

mon_node $_, 1
   for up_nodes;

mon_nodes \&mon_node;

=back

=head1 SEE ALSO

L<AnyEvent::MP>.

=head1 AUTHOR

 Marc Lehmann <schmorp@schmorp.de>
 http://home.schmorp.de/

=cut

1

Revision:	1.23
Committed:	Tue Sep 8 01:42:14 2009 UTC (14 years, 8 months ago) by root
Branch:	MAIN
Changes since 1.22:	+1 -3 lines
Log Message:	* empty log message *
#	User	Rev	Content
1	root	1.1	=head1 NAME
2
3			AnyEvent::MP::Global - some network-global services
4
5			=head1 SYNOPSIS
6
7			use AnyEvent::MP::Global;
8
9			=head1 DESCRIPTION
10
11	root	1.15	This module maintains a fully-meshed network, if possible, and tries to
12			ensure that we are connected to at least one other node.
13	root	1.1
14	root	1.15	It also manages named port groups - ports can register themselves in any
15			number of groups that will be available network-wide, which is great for
16			discovering services.
17	root	1.1
18	root	1.15	Running it on one node will automatically run it on all nodes, although,
19			at the moment, the global service is started by default anyways.
20	root	1.3
21	root	1.1	=head1 GLOBALS AND FUNCTIONS
22
23			=over 4
24
25			=cut
26
27			package AnyEvent::MP::Global;
28
29			use common::sense;
30			use Carp ();
31
32	root	1.8	use AnyEvent ();
33	root	1.4	use AnyEvent::Util ();
34
35	root	1.1	use AnyEvent::MP;
36			use AnyEvent::MP::Kernel;
37	root	1.22	use AnyEvent::MP::Transport ();
38	root	1.1
39	root	1.20	use base "Exporter";
40
41			our @EXPORT = qw(
42			grp_reg
43			grp_get
44			grp_mon
45			);
46
47	root	1.1	our $VERSION = $AnyEvent::MP::VERSION;
48
49	root	1.8	our %addr; # port ID => [address...] mapping
50
51	root	1.4	our %port; # our rendezvous port on the other side
52			our %lreg; # local registry, name => [pid...]
53	root	1.20	our %lmon; # local registry monitoring name,pid => mon
54			our %greg; # global regstry, name => pid => undef
55	root	1.22	our %gmon; # group monitoring, group => [$cb...]
56	root	1.4
57	root	1.8	our $nodecnt;
58
59	root	1.5	$AnyEvent::MP::Kernel::WARN->(7, "starting global service.");
60
61	root	1.8	#############################################################################
62			# seednodes
63
64			our @SEEDS;
65	root	1.22	our %SEEDS; # just to check whether a seed is a seed
66	root	1.14	our %SEED_CONNECT;
67	root	1.8	our $SEED_WATCHER;
68
69	root	1.22	push @AnyEvent::MP::Transport::HOOK_CONNECT, sub {
70			my $peer = $_[0]{local_greeting}{peeraddr};
71			return unless exists $SEEDS{$peer};
72			$SEED_CONNECT{$peer} = 2;
73			};
74
75			push @AnyEvent::MP::Transport::HOOK_CONNECTED, sub {
76			my $peer = $_[0]{local_greeting}{peeraddr};
77			return unless exists $SEEDS{$peer};
78			$SEEDS{$peer} = $_[0]{remote_node};
79			};
80
81			push @AnyEvent::MP::Transport::HOOK_DESTROY, sub {
82			delete $SEED_CONNECT{$_[0]{local_greeting}{peeraddr}};
83			};
84
85	root	1.8	sub seed_connect {
86			my ($seed) = @_;
87
88			my ($host, $port) = AnyEvent::Socket::parse_hostport $seed
89			or Carp::croak "$seed: unparsable seed address";
90
91	root	1.22	return if $SEED_CONNECT{$seed};
92
93			$AnyEvent::MP::Kernel::WARN->(9, "trying connect to seed node $seed.");
94
95	root	1.8	# ughhh
96	root	1.14	$SEED_CONNECT{$seed} \|\|= AnyEvent::MP::Transport::mp_connect $host, $port,
97			seed => $seed,
98	root	1.8	sub {
99	root	1.22	$SEED_CONNECT{$seed} = 1;
100	root	1.8	},
101			;
102			}
103
104			sub more_seeding {
105	root	1.22	my $int = List::Util::max 1,
106			$AnyEvent::MP::Kernel::CONFIG->{connect_interval}
107			* ($nodecnt ? keys %AnyEvent::MP::Kernel::NODE : 1)
108			- rand;
109
110			$SEED_WATCHER = AE::timer $int, 0, \&more_seeding;
111
112			@SEEDS = keys %SEEDS unless @SEEDS;
113	root	1.8	return unless @SEEDS;
114
115	root	1.22	seed_connect splice @SEEDS, rand @SEEDS, 1;
116	root	1.8	}
117
118	root	1.14	sub avoid_seed($) {
119	root	1.23	delete $SEEDS{$_[0]};
120	root	1.14	}
121
122	root	1.8	sub set_seeds(@) {
123	root	1.22	@SEEDS{@_} = ();
124	root	1.8
125			$SEED_WATCHER \|\|= AE::timer 5, $AnyEvent::MP::Kernel::MONITOR_TIMEOUT, \&more_seeding;
126
127	root	1.22	for (1 .. keys %SEEDS) {
128			after 0.100 * rand, \&more_seeding;
129	root	1.19	}
130	root	1.8	}
131
132	root	1.22	# returns all (up) seed nodes, or all nodes if no seednodes are up/known
133			sub _route_nodes {
134			my @seeds = grep node_is_up $_, values %SEEDS;
135			@seeds = up_nodes unless @seeds;
136			@seeds
137			}
138
139	root	1.8	#############################################################################
140
141	root	1.20	sub _change {
142			my ($group, $add, $del) = @_;
143
144			my $kv = $greg{$group} \|\|= {};
145
146			delete @$kv{@$del};
147			@$kv{@$add} = ();
148
149			my $ports = [keys %$kv];
150			$_->($ports, $add, $del) for @{ $gmon{$group} };
151			}
152
153	root	1.4	sub unreg_groups($) {
154	root	1.8	my ($node) = @_;
155	root	1.4
156	root	1.8	my $qr = qr/^\Q$node\E(?:#\|$)/;
157	root	1.20	my @del;
158	root	1.4
159	root	1.20	while (my ($group, $ports) = each %greg) {
160			@del = grep /$qr/, keys %$ports;
161			_change $group, [], \@del
162			if @del;
163	root	1.4	}
164			}
165
166			sub set_groups($$) {
167	root	1.8	my ($node, $lreg) = @_;
168	root	1.5
169	root	1.6	while (my ($k, $v) = each %$lreg) {
170	root	1.20	_change $k, $v, [];
171	root	1.6	}
172	root	1.4	}
173
174	root	1.20	=item $guard = grp_reg $group, $port
175	root	1.4
176			Register the given (local!) port in the named global group C<$group>.
177
178			The port will be unregistered automatically when the port is destroyed.
179
180			When not called in void context, then a guard object will be returned that
181			will also cause the name to be unregistered when destroyed.
182
183			=cut
184
185			# unregister local port
186			sub unregister {
187			my ($port, $group) = @_;
188
189			delete $lmon{"$group\x00$port"};
190			@{ $lreg{$group} } = grep $_ ne $port, @{ $lreg{$group} };
191
192	root	1.20	_change $group, [], [$port];
193	root	1.4
194	root	1.20	snd $_, reg0 => $group, $port
195	root	1.4	for values %port;
196			}
197
198			# register local port
199	root	1.20	sub grp_reg($$) {
200			my ($group, $port) = @_;
201	root	1.4
202			port_is_local $port
203	root	1.20	or Carp::croak "AnyEvent::MP::Global::grp_reg can only be called for local ports, caught";
204
205			grep $_ eq $port, @{ $lreg{$group} }
206			and Carp::croak "'$group': group already registered, cannot register a second time";
207	root	1.4
208			$lmon{"$group\x00$port"} = mon $port, sub { unregister $port, $group };
209			push @{ $lreg{$group} }, $port;
210
211	root	1.20	snd $_, reg1 => $group, $port
212	root	1.4	for values %port;
213
214	root	1.20	_change $group, [$port], [];
215	root	1.4
216			wantarray && AnyEvent::Util::guard { unregister $port, $group }
217			}
218
219	root	1.20	=item $ports = grp_get $group
220	root	1.15
221			Returns all the ports currently registered to the given group (as
222	root	1.20	read-only(!) array reference). When the group has no registered members,
223	root	1.15	return C<undef>.
224
225			=cut
226
227	root	1.20	sub grp_get($) {
228			my @ports = keys %{ $greg{$_[0]} };
229			@ports ? \@ports : undef
230			}
231
232			=item $guard = grp_mon $group, $callback->($ports, $add, $del)
233
234			Installs a monitor on the given group. Each time there is a change it
235			will be called with the current group members as an arrayref as the
236			first argument. The second argument only contains ports added, the third
237			argument only ports removed.
238
239			Unlike C<grp_get>, all three arguments will always be array-refs, even if
240			the array is empty. None of the arrays must be modified in any way.
241
242			The first invocation will be with the first two arguments set to the
243			current members, as if all of them were just added, but only when the
244			group is atcually non-empty.
245
246			Optionally returns a guard object that uninstalls the watcher when it is
247			destroyed.
248
249			=cut
250
251			sub grp_mon($$) {
252			my ($grp, $cb) = @_;
253
254			AnyEvent::MP::Kernel::delay sub {
255			return unless $cb;
256
257			push @{ $gmon{$grp} }, $cb;
258			$cb->(((grp_get $grp) \|\| return) x 2, []);
259			};
260
261			defined wantarray && AnyEvent::Util::::guard {
262			my @mon = grep $_ != $cb, @{ delete $gmon{$grp} };
263			$gmon{$grp} = \@mon if @mon;
264			undef $cb;
265			}
266	root	1.15	}
267
268	root	1.4	sub start_node {
269	root	1.8	my ($node) = @_;
270	root	1.4
271	root	1.8	return if exists $port{$node};
272			return if $node eq $NODE; # do not connect to ourselves
273	root	1.4
274			# establish connection
275	root	1.8	my $port = $port{$node} = spawn $node, "AnyEvent::MP::Global::connect", 0, $NODE;
276
277	root	1.4	mon $port, sub {
278	root	1.8	unreg_groups $node;
279			delete $port{$node};
280	root	1.4	};
281	root	1.8
282	root	1.10	snd $port, addr => $AnyEvent::MP::Kernel::LISTENER;
283	root	1.17	snd $port, nodes => \%addr if %addr;
284	root	1.8	snd $port, set => \%lreg if %lreg;
285	root	1.4	}
286	root	1.3
287	root	1.5	# other nodes connect via this
288	root	1.3	sub connect {
289	root	1.8	my ($version, $node) = @_;
290	root	1.1
291	root	1.3	# monitor them, silently die
292	root	1.8	mon $node, psub { kil $SELF };
293	root	1.3
294	root	1.4	rcv $SELF,
295	root	1.8	addr => sub {
296			my $addresses = shift;
297	root	1.14	$AnyEvent::MP::Kernel::WARN->(9, "$node told us its addresses (@$addresses).");
298	root	1.8	$addr{$node} = $addresses;
299	root	1.9
300			# to help listener-less nodes, we broadcast new addresses to them unconditionally
301			#TODO: should be done by a node finding out about a listener-less one
302			if (@$addresses) {
303	root	1.18	for my $other (values %AnyEvent::MP::Kernel::NODE) {
304	root	1.9	if ($other->{transport}) {
305	root	1.12	if ($addr{$other->{id}} && !@{ $addr{$other->{id}} }) {
306			$AnyEvent::MP::Kernel::WARN->(9, "helping $other->{id} to find $node.");
307	root	1.17	snd $port{$other->{id}}, nodes => { $node => $addresses };
308	root	1.9	}
309			}
310			}
311			}
312	root	1.8	},
313	root	1.17	nodes => sub {
314	root	1.8	my ($kv) = @_;
315
316			use JSON::XS;#d#
317			my $kv_txt = JSON::XS->new->encode ($kv);#d#
318			$AnyEvent::MP::Kernel::WARN->(9, "$node told us it knows about $kv_txt.");#d#
319
320			while (my ($id, $addresses) = each %$kv) {
321			my $node = AnyEvent::MP::Kernel::add_node $id;
322			$node->connect (@$addresses);
323			start_node $id;
324	root	1.4	}
325			},
326	root	1.16	find => sub {
327	root	1.8	my ($othernode) = @_;
328
329			$AnyEvent::MP::Kernel::WARN->(9, "$node asked us to find $othernode.");
330	root	1.17	snd $port{$node}, nodes => { $othernode => $addr{$othernode} }
331	root	1.8	if $addr{$othernode};
332			},
333	root	1.4	set => sub {
334	root	1.8	set_groups $node, shift;
335	root	1.4	},
336	root	1.20	reg0 => sub {
337			_change $_[0], [], [$_[1]];
338			},
339			reg1 => sub {
340			_change $_[0], [$_[1]], [];
341			},
342	root	1.4	;
343	root	1.3	}
344	root	1.1
345			sub mon_node {
346	root	1.8	my ($node, $is_up) = @_;
347	root	1.1
348			if ($is_up) {
349	root	1.8	++$nodecnt;
350			start_node $node;
351	root	1.2	} else {
352	root	1.8	--$nodecnt;
353			more_seeding unless $nodecnt;
354			unreg_groups $node;
355
356			# forget about the node
357			delete $addr{$node};
358			# ask other nodes if they know the node
359	root	1.16	snd $_, find => $node
360	root	1.8	for values %port;
361	root	1.1	}
362	root	1.8	#warn "node<$node,$is_up>\n";#d#
363	root	1.1	}
364
365			mon_node $_, 1
366			for up_nodes;
367
368			mon_nodes \&mon_node;
369
370			=back
371
372			=head1 SEE ALSO
373
374			L<AnyEvent::MP>.
375
376			=head1 AUTHOR
377
378			Marc Lehmann <schmorp@schmorp.de>
379			http://home.schmorp.de/
380
381			=cut
382
383			1
384