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 between global nodes and
tries to have connections with all nodes in the network.

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
);

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

#############################################################################
# node protocol parts for global nodes

{
   package AnyEvent::MP::Kernel;

   # TODO: this is ugly, maybe this should go into MP::Kernel or a separate module #d#

   our %NODE;
   our $NODE;
   our $LISTENER;

   our $GLOBAL;
   our $MASTER;
   our $MASTER_MON;
   our $MASTER_TIMER;

   our %GLOBAL_SLAVE;

   our %GLOBAL_DB; # global db
   our %LOCAL_DBS; # local databases of other global nodes
   our %LOCAL_DB;  # this node database

   our $SRCNODE;
   our %node_req;

   # only in global code
   our %GLOBAL_MON; # monitors {family}{"" or key}

   sub other_globals() {
      grep $_ ne $NODE && node_is_up $_, keys %{ $GLOBAL_DB{"'g"} }
   }

   # broadcasts a message to all other global nodes
   sub g_broadcast {
      snd $_, @_
         for other_globals;
   }

   sub g_mon_check {
      warn "g_mon_check<@_>\n";#d#

      my %node = (
         %{ $GLOBAL_MON{$_[1]}{$_[2]} },
         %{ $GLOBAL_MON{$_[1]}{""   } },
         %{ $GLOBAL_MON{""   }{""   } },
      );

      snd $_ => g_chg1 => $_[1], $_[2], @_ > 2 ? $_[3] : ()
         for keys %node;
   }

   # add/replace a key in the database
   sub g_add($$$$) {
      $LOCAL_DBS{$_[0]}{$_[1]}{$_[2]} =
      $GLOBAL_DB       {$_[1]}{$_[2]} = $_[3];

      g_broadcast g_add => $_[1] => $_[2] => $_[3]
         if exists $GLOBAL_SLAVE{$_[0]};

      warn "g_add<@_>\n";#d#
      &g_mon_check;
   }

   # delete a key from the database
   sub g_del($$$) {
      delete $LOCAL_DBS{$_[0]}{$_[1]}{$_[2]};

      g_broadcast g_del => $_[1] => $_[2]
         if exists $GLOBAL_SLAVE{$_[0]};

      delete $GLOBAL_DB{$_[1]}{$_[2]};

      # check if other node maybe still has the key, then we don't delete, but add
      for (values %LOCAL_DBS) {
         if (exists $_->{$_[1]}{$_[2]}) {
            $GLOBAL_DB{$_[1]}{$_[2]} = $_->{$_[1]}{$_[2]};
            last;
         }
      }

      warn "g_del<@_>\n";#d#
      &g_mon_check;
   }

   # delete all keys from a database
   sub g_clr($) {
      my ($node) = @_;

      my $db = $LOCAL_DBS{$node};
      while (my ($f, $k) = each %$db) {
         g_del $node, $f => $_
            for keys %$k;
      }

      delete $LOCAL_DBS{$node};
   }

   # set the whole (node-local) database - previous value must be empty
   sub g_set($$) {
      my ($node, $db) = @_;

      while (my ($f, $k) = each %$db) {
         g_add $node, $f => $_ => delete $k->{$_}
            for keys %$k;
      }
   }

   # gather node databases from slaves

   # other node wants to make us the master
   $node_req{g_slave} = sub {
      my ($db) = @_;

      my $node = $SRCNODE->{id};
      undef $GLOBAL_SLAVE{$node};
      g_set $node, $db;
   };

   $node_req{g_add} = sub {
      &g_add ($SRCNODE->{id}, @_);
   };

   $node_req{g_del} = sub {
      &g_del ($SRCNODE->{id}, @_);
   };

   $node_req{g_set} = sub {
      g_set $SRCNODE->{id}, $_[0];
   };

   $node_req{g_find} = sub {
      my ($node) = @_;

      snd $SRCNODE->{id}, g_found => $node, $GLOBAL_DB{"'l"}{$node};
   };

   # monitoring

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

      g_clr $node;

      if (my $mon = delete $GLOBAL_SLAVE{$node}) {
         while (my ($f, $fv) = each %$mon) {
            delete $GLOBAL_MON{$f}{$_}
               for keys %$fv;
         }
      }
   }

   # g_mon0 family key - stop monitoring
   $node_req{g_mon0} = sub {
      delete $GLOBAL_MON{$_[0]}{$_[1]}{$SRCNODE->{id}};
      delete $GLOBAL_SLAVE{$SRCNODE->{id}}{$_[0]}{$_[1]};
   };

   # g_mon1 family key - start monitoring
   $node_req{g_mon1} = sub {
      undef $GLOBAL_SLAVE{$SRCNODE->{id}}{$_[0]}{$_[1]};
      undef $GLOBAL_MON{$_[0]}{$_[1]}{$SRCNODE->{id}};
      #d# generate lots of initial change requests, or one big?

      snd $SRCNODE->{id}, g_chg0 => $_[0], $_[1]
   };

   #############################################################################
   # switch to global mode

   $GLOBAL = 1;
   $MASTER = $NODE;
   undef $GLOBAL_SLAVE{$NODE}; # we are our own master (and slave)
   $LOCAL_DBS{$NODE} = { %LOCAL_DB };

   # regularly try to connect to global nodes - maybe use seeding code?
   $MASTER_TIMER = AE::timer 0, $AnyEvent::MP::Kernel::CONFIG->{monitor_timeout}, sub {
      (add_node $_)->connect
         for keys %{ $GLOBAL_DB{"'g"} };
   };

   # instantly connect to other global nodes when we learn of them
   # so we don't have to wait for the timer.
   #TODO
#   $GLOBAL_MON{"'g"}{""}{""} = sub {
#      (add_node $_[1])->connect;
#   };

   # delete slaves on node-down
   # clear slave db on node-down
   $MASTER_MON = mon_nodes sub {
      g_slave_disconnect $_[0] unless $_[1];
   };

   # tell everybody who connects that we are a global node
   push @AnyEvent::MP::Transport::HOOK_GREET, sub {
      $_[0]{local_greeting}{global} = 1;
   };

   # connect from a global node
   sub g_global_connect {
      my ($node) = @_;

      # we need to set this currently, as to avoid race conditions
      # because it takes a while until the other global node tells us it is global.

      undef $GLOBAL_DB{"'g"}{$node};
      undef $LOCAL_DBS{$node}{"'g"}{$node};

      # global nodes send all local databases, merged,
      # as their local database to global nodes
      my %db;

      for (values %LOCAL_DBS) {
         while (my ($f, $fv) = each %$_) {
            while (my ($k, $kv) = each %$fv) {
               $db{$f}{$k} = $kv;
            }
         }
      }

      snd $node => g_set => \%db;
   }

   # send our database to every global node that connects
   push @AnyEvent::MP::Transport::HOOK_CONNECT, sub {
      return unless $_[0]{remote_greeting}{global};

      g_global_connect $_[0]{remote_node};
   };

   # tell our master that we are global now
   for (values %NODE) {
      if ($_->{transport} && $_->{transport}{remote_greeting}{global}) {
         snd $_->{id} => "g_global";
         g_global_connect $_->{id};
      }
   }

   $node_req{g_global} = sub {
      g_slave_disconnect $SRCNODE->{id};
      $SRCNODE->{transport}{remote_greeting}{global} = 1;
      g_global_connect $SRCNODE->{id};
   };

   #d#d disconnect everybody to bootstrap development grr

#   $_->transport_error # remove Self::transport_error
#      for values %NODE;

   # now add us to the set of global nodes
   ldb_set "'g" => $NODE => undef;
}

=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

# 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";

   defined 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($) {
}

=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 actually non-empty.

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

=cut

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

=back

=head1 SEE ALSO

L<AnyEvent::MP>.

=head1 AUTHOR

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

=cut

1

Revision:	1.47
Committed:	Fri Mar 2 19:19:21 2012 UTC (14 years, 7 months ago) by root
Branch:	MAIN
Changes since 1.46:	+101 -30 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.42	This module maintains a fully-meshed network between global nodes and
12			tries to have connections with all nodes in the network.
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.5	$AnyEvent::MP::Kernel::WARN->(7, "starting global service.");
48
49	root	1.8	#############################################################################
50	root	1.44	# node protocol parts for global nodes
51	root	1.8
52	root	1.44	{
53			package AnyEvent::MP::Kernel;
54	root	1.8
55	root	1.44	# TODO: this is ugly, maybe this should go into MP::Kernel or a separate module #d#
56	root	1.26
57	root	1.46	our %NODE;
58			our $NODE;
59			our $LISTENER;
60
61	root	1.44	our $GLOBAL;
62	root	1.45	our $MASTER;
63	root	1.47	our $MASTER_MON;
64			our $MASTER_TIMER;
65	root	1.46
66	root	1.44	our %GLOBAL_SLAVE;
67	root	1.26
68	root	1.46	our %GLOBAL_DB; # global db
69			our %LOCAL_DBS; # local databases of other global nodes
70			our %LOCAL_DB; # this node database
71
72			our $SRCNODE;
73			our %node_req;
74
75	root	1.47	# only in global code
76			our %GLOBAL_MON; # monitors {family}{"" or key}
77
78			sub other_globals() {
79			grep $_ ne $NODE && node_is_up $_, keys %{ $GLOBAL_DB{"'g"} }
80			}
81
82	root	1.46	# broadcasts a message to all other global nodes
83			sub g_broadcast {
84			snd $_, @_
85			for other_globals;
86			}
87
88			sub g_mon_check {
89			warn "g_mon_check<@_>\n";#d#
90	root	1.47
91			my %node = (
92			%{ $GLOBAL_MON{$_[1]}{$_[2]} },
93			%{ $GLOBAL_MON{$_[1]}{"" } },
94			%{ $GLOBAL_MON{"" }{"" } },
95			);
96
97			snd $_ => g_chg1 => $_[1], $_[2], @_ > 2 ? $_[3] : ()
98			for keys %node;
99	root	1.46	}
100
101			# add/replace a key in the database
102			sub g_add($$$$) {
103			$LOCAL_DBS{$_[0]}{$_[1]}{$_[2]} =
104			$GLOBAL_DB {$_[1]}{$_[2]} = $_[3];
105
106			g_broadcast g_add => $_[1] => $_[2] => $_[3]
107			if exists $GLOBAL_SLAVE{$_[0]};
108
109			warn "g_add<@_>\n";#d#
110			&g_mon_check;
111			}
112
113			# delete a key from the database
114			sub g_del($$$) {
115			delete $LOCAL_DBS{$_[0]}{$_[1]}{$_[2]};
116
117			g_broadcast g_del => $_[1] => $_[2]
118			if exists $GLOBAL_SLAVE{$_[0]};
119
120			delete $GLOBAL_DB{$_[1]}{$_[2]};
121
122			# check if other node maybe still has the key, then we don't delete, but add
123			for (values %LOCAL_DBS) {
124			if (exists $_->{$_[1]}{$_[2]}) {
125			$GLOBAL_DB{$_[1]}{$_[2]} = $_->{$_[1]}{$_[2]};
126			last;
127			}
128			}
129
130			warn "g_del<@_>\n";#d#
131			&g_mon_check;
132			}
133
134			# delete all keys from a database
135			sub g_clr($) {
136			my ($node) = @_;
137
138			my $db = $LOCAL_DBS{$node};
139			while (my ($f, $k) = each %$db) {
140			g_del $node, $f => $_
141			for keys %$k;
142			}
143
144			delete $LOCAL_DBS{$node};
145			}
146
147			# set the whole (node-local) database - previous value must be empty
148			sub g_set($$) {
149			my ($node, $db) = @_;
150
151			while (my ($f, $k) = each %$db) {
152			g_add $node, $f => $_ => delete $k->{$_}
153			for keys %$k;
154			}
155			}
156
157			# gather node databases from slaves
158
159			# other node wants to make us the master
160			$node_req{g_slave} = sub {
161			my ($db) = @_;
162
163			my $node = $SRCNODE->{id};
164			undef $GLOBAL_SLAVE{$node};
165			g_set $node, $db;
166			};
167
168			$node_req{g_add} = sub {
169			&g_add ($SRCNODE->{id}, @_);
170			};
171
172			$node_req{g_del} = sub {
173			&g_del ($SRCNODE->{id}, @_);
174			};
175
176			$node_req{g_set} = sub {
177			g_set $SRCNODE->{id}, $_[0];
178			};
179
180			$node_req{g_find} = sub {
181			my ($node) = @_;
182
183			snd $SRCNODE->{id}, g_found => $node, $GLOBAL_DB{"'l"}{$node};
184			};
185
186	root	1.47	# monitoring
187
188			sub g_slave_disconnect($) {
189			my ($node) = @_;
190
191			g_clr $node;
192
193			if (my $mon = delete $GLOBAL_SLAVE{$node}) {
194			while (my ($f, $fv) = each %$mon) {
195			delete $GLOBAL_MON{$f}{$_}
196			for keys %$fv;
197			}
198			}
199			}
200
201			# g_mon0 family key - stop monitoring
202			$node_req{g_mon0} = sub {
203			delete $GLOBAL_MON{$_[0]}{$_[1]}{$SRCNODE->{id}};
204			delete $GLOBAL_SLAVE{$SRCNODE->{id}}{$_[0]}{$_[1]};
205			};
206
207			# g_mon1 family key - start monitoring
208			$node_req{g_mon1} = sub {
209			undef $GLOBAL_SLAVE{$SRCNODE->{id}}{$_[0]}{$_[1]};
210			undef $GLOBAL_MON{$_[0]}{$_[1]}{$SRCNODE->{id}};
211			#d# generate lots of initial change requests, or one big?
212
213			snd $SRCNODE->{id}, g_chg0 => $_[0], $_[1]
214			};
215
216	root	1.46	#############################################################################
217	root	1.44	# switch to global mode
218	root	1.46
219	root	1.44	$GLOBAL = 1;
220			$MASTER = $NODE;
221	root	1.46	undef $GLOBAL_SLAVE{$NODE}; # we are our own master (and slave)
222	root	1.47	$LOCAL_DBS{$NODE} = { %LOCAL_DB };
223
224			# regularly try to connect to global nodes - maybe use seeding code?
225			$MASTER_TIMER = AE::timer 0, $AnyEvent::MP::Kernel::CONFIG->{monitor_timeout}, sub {
226			(add_node $_)->connect
227			for keys %{ $GLOBAL_DB{"'g"} };
228			};
229	root	1.22
230	root	1.47	# instantly connect to other global nodes when we learn of them
231			# so we don't have to wait for the timer.
232			#TODO
233			# $GLOBAL_MON{"'g"}{""}{""} = sub {
234			# (add_node $_[1])->connect;
235			# };
236
237			# delete slaves on node-down
238	root	1.46	# clear slave db on node-down
239	root	1.47	$MASTER_MON = mon_nodes sub {
240			g_slave_disconnect $_[0] unless $_[1];
241	root	1.44	};
242	root	1.20
243	root	1.44	# tell everybody who connects that we are a global node
244			push @AnyEvent::MP::Transport::HOOK_GREET, sub {
245			$_[0]{local_greeting}{global} = 1;
246			};
247	root	1.4
248	root	1.47	# connect from a global node
249			sub g_global_connect {
250			my ($node) = @_;
251
252			# we need to set this currently, as to avoid race conditions
253			# because it takes a while until the other global node tells us it is global.
254
255			undef $GLOBAL_DB{"'g"}{$node};
256			undef $LOCAL_DBS{$node}{"'g"}{$node};
257	root	1.46
258			# global nodes send all local databases, merged,
259			# as their local database to global nodes
260			my %db;
261
262			for (values %LOCAL_DBS) {
263	root	1.47	while (my ($f, $fv) = each %$_) {
264			while (my ($k, $kv) = each %$fv) {
265			$db{$f}{$k} = $kv;
266	root	1.46	}
267			}
268			}
269
270	root	1.47	snd $node => g_set => \%db;
271			}
272
273			# send our database to every global node that connects
274			push @AnyEvent::MP::Transport::HOOK_CONNECT, sub {
275			return unless $_[0]{remote_greeting}{global};
276
277			g_global_connect $_[0]{remote_node};
278			};
279
280			# tell our master that we are global now
281			for (values %NODE) {
282			if ($_->{transport} && $_->{transport}{remote_greeting}{global}) {
283			snd $_->{id} => "g_global";
284			g_global_connect $_->{id};
285			}
286			}
287
288			$node_req{g_global} = sub {
289			g_slave_disconnect $SRCNODE->{id};
290			$SRCNODE->{transport}{remote_greeting}{global} = 1;
291			g_global_connect $SRCNODE->{id};
292	root	1.44	};
293	root	1.4
294	root	1.46	#d#d disconnect everybody to bootstrap development grr
295
296	root	1.47	# $_->transport_error # remove Self::transport_error
297			# for values %NODE;
298	root	1.46
299			# now add us to the set of global nodes
300			ldb_set "'g" => $NODE => undef;
301	root	1.4	}
302
303	root	1.20	=item $guard = grp_reg $group, $port
304	root	1.4
305			Register the given (local!) port in the named global group C<$group>.
306
307			The port will be unregistered automatically when the port is destroyed.
308
309			When not called in void context, then a guard object will be returned that
310			will also cause the name to be unregistered when destroyed.
311
312			=cut
313
314			# register local port
315	root	1.20	sub grp_reg($$) {
316			my ($group, $port) = @_;
317	root	1.4
318			port_is_local $port
319	root	1.20	or Carp::croak "AnyEvent::MP::Global::grp_reg can only be called for local ports, caught";
320
321	root	1.44	defined wantarray && AnyEvent::Util::guard { unregister ($port, $group) }
322	root	1.4	}
323
324	root	1.20	=item $ports = grp_get $group
325	root	1.15
326			Returns all the ports currently registered to the given group (as
327	root	1.20	read-only(!) array reference). When the group has no registered members,
328	root	1.15	return C<undef>.
329
330			=cut
331
332	root	1.20	sub grp_get($) {
333			}
334
335			=item $guard = grp_mon $group, $callback->($ports, $add, $del)
336
337			Installs a monitor on the given group. Each time there is a change it
338			will be called with the current group members as an arrayref as the
339			first argument. The second argument only contains ports added, the third
340			argument only ports removed.
341
342			Unlike C<grp_get>, all three arguments will always be array-refs, even if
343			the array is empty. None of the arrays must be modified in any way.
344
345			The first invocation will be with the first two arguments set to the
346			current members, as if all of them were just added, but only when the
347	root	1.39	group is actually non-empty.
348	root	1.20
349			Optionally returns a guard object that uninstalls the watcher when it is
350			destroyed.
351
352			=cut
353
354			sub grp_mon($$) {
355			my ($grp, $cb) = @_;
356	root	1.4	}
357	root	1.3
358	root	1.1	=back
359
360			=head1 SEE ALSO
361
362			L<AnyEvent::MP>.
363
364			=head1 AUTHOR
365
366			Marc Lehmann <schmorp@schmorp.de>
367			http://home.schmorp.de/
368
369			=cut
370
371			1
372