AnyEvent-MP/MP/Global.pm

=head1 NAME

AnyEvent::MP::Global - network backbone services

=head1 SYNOPSIS

   use AnyEvent::MP::Global;

=head1 DESCRIPTION

This module is usually run (or started on) seed nodes and provides a
variety of services to connected nodes, such as the distributed database.

The global nodes form a fully-meshed network, that is, all global nodes
currently maintain connections to all other global nodes.

Loading this module (e.g. as a service) transforms the local node into a
global node. There are no user-servicable parts inside.

For a limited time, this module also exports some AEMP 1.x compatibility
functions (C<grp_reg>, C<grp_get> and C<grp_mon>).

=cut

package AnyEvent::MP::Global;

use common::sense;
use Carp ();

use AnyEvent ();

use AnyEvent::MP;
use AnyEvent::MP::Kernel;

AE::log 7 => "starting global service.";

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

package AnyEvent::MP::Kernel;

# TODO: this is ugly (classical use vars vs. our),
# maybe this should go into MP::Kernel

our %NODE;
our $NODE;

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; # the origin node id
our %NODE_REQ;

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

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

# add/replace/del inside a family in the database
# @$dle must not contain any key in %$set
sub g_upd {
   my ($node, $family, $set, $del) = @_;

   my $ldb = $LOCAL_DBS{$node}{$family} ||= {};
   my $gdb = $GLOBAL_DB       {$family} ||= {};

   # add/replace keys
   while (my ($k, $v) = each %$set) {
      $ldb->{$k} =
      $gdb->{$k} = $v;
   }

   my @del; # actual deletes

   # take care of deletes
   keydel:
   for my $k (@$del) {
      delete $ldb->{$k};
      delete $gdb->{$k};

      # check if some other node still has the key, then we don't delete, but change
      for (values %LOCAL_DBS) {
         if (exists $_->{$family}{$k}) {
            $set->{$k} = $gdb->{$k} = $_->{$family}{$k};

            next keydel;
         }
      }

      push @del, $k;
   }

   # family could be empty now
   delete $GLOBAL_DB{$family} unless %$gdb;
   delete $LOCAL_DBS{$node}{$family} unless %$ldb;

   g_broadcast g_upd => $family, $set, \@del
      if exists $GLOBAL_SLAVE{$node};

   # tell subscribers we have changed the family
   snd $_ => g_chg2 => $family, $set, \@del
      for keys %{ $GLOBAL_MON{$family} };
}

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

   while (my ($f, $k) = each %$db) {
      g_upd $node, $f, $k;
   }
}

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

   my $db = $LOCAL_DBS{$node};

   while (my ($f, $k) = each %$db) {
      g_upd $node, $f, undef, [keys %$k];
   }

   delete $LOCAL_DBS{$node};
}

# gather node databases from slaves

# other node wants to make us the master
$NODE_REQ{g_slave} = sub {
   my ($db) = @_
      or return; # empty g_slave is used to start global service

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

$NODE_REQ{g_set} = sub {
   &g_set ($SRCNODE, @_);
};

$NODE_REQ{g_upd} = sub {
   &g_upd ($SRCNODE, @_);
};

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

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

$NODE_REQ{g_db_family} = sub {
   my ($family, $id) = @_;
   snd $SRCNODE, g_reply => $id, $GLOBAL_DB{$family} || {};
};

$NODE_REQ{g_db_keys} = sub {
   my ($family, $id) = @_;
   snd $SRCNODE, g_reply => $id, [keys %{ $GLOBAL_DB{$family} } ];
};

$NODE_REQ{g_db_values} = sub {
   my ($family, $id) = @_;
   snd $SRCNODE, g_reply => $id, [values %{ $GLOBAL_DB{$family} } ];
};

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

         delete $GLOBAL_MON{$f}
            unless %{ $GLOBAL_MON{$f} };
      }
   }
}

# g_mon0 family - stop monitoring
$NODE_REQ{g_mon0} = sub {
   delete $GLOBAL_MON{$_[0]}{$SRCNODE};
   delete $GLOBAL_MON{$_[0]} unless %{ $GLOBAL_MON{$_[0]} };

   delete $GLOBAL_SLAVE{$SRCNODE}{$_[0]};
};

# g_mon1 family key - start monitoring
$NODE_REQ{g_mon1} = sub {
   undef $GLOBAL_SLAVE{$SRCNODE}{$_[0]};
   undef $GLOBAL_MON{$_[0]}{$SRCNODE};

   snd $SRCNODE, g_chg1 => $_[0], $GLOBAL_DB{$_[0]};
};

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

# 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 other_globals;
};

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

   my $node = $NODE{$SRCNODE};
   $node->{transport}{remote_greeting}{global} = 1;

   g_global_connect $SRCNODE;
};

# enable global mode
$GLOBAL = 1;

# global nodes are their own masters - this
# resends global requests and sets the local database.
master_set $NODE;

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

#############################################################################
# compatibility functions for aemp 1.0

package AnyEvent::MP::Global;

use base "Exporter";
our @EXPORT = qw(grp_reg grp_get grp_mon);

sub grp_reg($$) {
   &db_reg
}

sub grp_get($) {
   my @ports = keys %{ $AnyEvent::MP::Kernel::GLOBAL_DB{$_[0]} };

   @ports ? \@ports : undef
}

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

   db_mon $grp => sub {
      my ($ports, $add, $chg, $del) = @_;

      $cb->([keys %$ports], $add, $del);
   };
}

=head1 SEE ALSO

L<AnyEvent::MP>.

=head1 AUTHOR

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

=cut

1

Revision:	1.62
Committed:	Fri Mar 23 21:16:25 2012 UTC (14 years, 6 months ago) by root
Branch:	MAIN
Changes since 1.61:	+1 -1 lines
Log Message:	* empty log message *
#	User	Rev	Content
1	root	1.1	=head1 NAME
2
3	root	1.58	AnyEvent::MP::Global - network backbone services
4	root	1.1
5			=head1 SYNOPSIS
6
7			use AnyEvent::MP::Global;
8
9			=head1 DESCRIPTION
10
11	root	1.58	This module is usually run (or started on) seed nodes and provides a
12			variety of services to connected nodes, such as the distributed database.
13	root	1.1
14	root	1.58	The global nodes form a fully-meshed network, that is, all global nodes
15			currently maintain connections to all other global nodes.
16	root	1.1
17	root	1.58	Loading this module (e.g. as a service) transforms the local node into a
18			global node. There are no user-servicable parts inside.
19	root	1.1
20	root	1.61	For a limited time, this module also exports some AEMP 1.x compatibility
21			functions (C<grp_reg>, C<grp_get> and C<grp_mon>).
22
23	root	1.1	=cut
24
25			package AnyEvent::MP::Global;
26
27			use common::sense;
28			use Carp ();
29
30	root	1.8	use AnyEvent ();
31	root	1.4
32	root	1.1	use AnyEvent::MP;
33			use AnyEvent::MP::Kernel;
34	root	1.20
35	root	1.60	AE::log 7 => "starting global service.";
36	root	1.5
37	root	1.8	#############################################################################
38	root	1.44	# node protocol parts for global nodes
39	root	1.8
40	root	1.49	package AnyEvent::MP::Kernel;
41	root	1.8
42	root	1.49	# TODO: this is ugly (classical use vars vs. our),
43			# maybe this should go into MP::Kernel
44	root	1.26
45	root	1.49	our %NODE;
46			our $NODE;
47	root	1.46
48	root	1.49	our $GLOBAL;
49			our $MASTER;
50			our $MASTER_MON;
51			our $MASTER_TIMER;
52	root	1.46
53	root	1.49	our %GLOBAL_SLAVE;
54	root	1.26
55	root	1.49	our %GLOBAL_DB; # global db
56			our %LOCAL_DBS; # local databases of other global nodes
57			our %LOCAL_DB; # this node database
58	root	1.46
59	root	1.59	our $SRCNODE; # the origin node id
60	root	1.49	our %NODE_REQ;
61	root	1.46
62	root	1.49	# only in global code
63	root	1.51	our %GLOBAL_MON; # monitors {family}
64	root	1.47
65	root	1.49	sub other_globals() {
66			grep $_ ne $NODE && node_is_up $_, keys %{ $GLOBAL_DB{"'g"} }
67			}
68	root	1.47
69	root	1.49	# broadcasts a message to all other global nodes
70			sub g_broadcast {
71			snd $_, @_
72			for other_globals;
73			}
74	root	1.46
75	root	1.57	# add/replace/del inside a family in the database
76			# @$dle must not contain any key in %$set
77			sub g_upd {
78			my ($node, $family, $set, $del) = @_;
79
80			my $ldb = $LOCAL_DBS{$node}{$family} \|\|= {};
81			my $gdb = $GLOBAL_DB {$family} \|\|= {};
82
83			# add/replace keys
84			while (my ($k, $v) = each %$set) {
85			$ldb->{$k} =
86			$gdb->{$k} = $v;
87			}
88
89			my @del; # actual deletes
90
91			# take care of deletes
92			keydel:
93			for my $k (@$del) {
94			delete $ldb->{$k};
95			delete $gdb->{$k};
96
97			# check if some other node still has the key, then we don't delete, but change
98			for (values %LOCAL_DBS) {
99			if (exists $_->{$family}{$k}) {
100			$set->{$k} = $gdb->{$k} = $_->{$family}{$k};
101	root	1.46
102	root	1.57	next keydel;
103			}
104			}
105	root	1.46
106	root	1.57	push @del, $k;
107			}
108	root	1.51
109	root	1.57	# family could be empty now
110			delete $GLOBAL_DB{$family} unless %$gdb;
111			delete $LOCAL_DBS{$node}{$family} unless %$ldb;
112	root	1.51
113	root	1.57	g_broadcast g_upd => $family, $set, \@del
114			if exists $GLOBAL_SLAVE{$node};
115	root	1.46
116	root	1.57	# tell subscribers we have changed the family
117			snd $_ => g_chg2 => $family, $set, \@del
118			for keys %{ $GLOBAL_MON{$family} };
119	root	1.51	}
120
121			# set the whole (node-local) database - previous value must be empty
122			sub g_set($$) {
123			my ($node, $db) = @_;
124
125			while (my ($f, $k) = each %$db) {
126	root	1.57	g_upd $node, $f, $k;
127	root	1.51	}
128	root	1.49	}
129	root	1.46
130	root	1.49	# delete all keys from a database
131			sub g_clr($) {
132			my ($node) = @_;
133
134			my $db = $LOCAL_DBS{$node};
135	root	1.57
136	root	1.49	while (my ($f, $k) = each %$db) {
137	root	1.57	g_upd $node, $f, undef, [keys %$k];
138	root	1.46	}
139
140	root	1.49	delete $LOCAL_DBS{$node};
141			}
142
143			# gather node databases from slaves
144	root	1.46
145	root	1.49	# other node wants to make us the master
146			$NODE_REQ{g_slave} = sub {
147	root	1.58	my ($db) = @_
148			or return; # empty g_slave is used to start global service
149	root	1.46
150	root	1.59	my $node = $SRCNODE;
151	root	1.49	undef $GLOBAL_SLAVE{$node};
152			g_set $node, $db;
153			};
154	root	1.46
155	root	1.57	$NODE_REQ{g_set} = sub {
156	root	1.61	&g_set ($SRCNODE, @_);
157	root	1.49	};
158	root	1.46
159	root	1.57	$NODE_REQ{g_upd} = sub {
160	root	1.61	&g_upd ($SRCNODE, @_);
161	root	1.49	};
162	root	1.46
163	root	1.49	$NODE_REQ{g_find} = sub {
164			my ($node) = @_;
165	root	1.46
166	root	1.59	snd $SRCNODE, g_found => $node, $GLOBAL_DB{"'l"}{$node};
167	root	1.49	};
168	root	1.46
169	root	1.55	$NODE_REQ{g_db_family} = sub {
170			my ($family, $id) = @_;
171	root	1.59	snd $SRCNODE, g_reply => $id, $GLOBAL_DB{$family} \|\| {};
172	root	1.55	};
173
174			$NODE_REQ{g_db_keys} = sub {
175			my ($family, $id) = @_;
176	root	1.59	snd $SRCNODE, g_reply => $id, [keys %{ $GLOBAL_DB{$family} } ];
177	root	1.55	};
178
179			$NODE_REQ{g_db_values} = sub {
180			my ($family, $id) = @_;
181	root	1.59	snd $SRCNODE, g_reply => $id, [values %{ $GLOBAL_DB{$family} } ];
182	root	1.55	};
183
184	root	1.49	# monitoring
185	root	1.47
186	root	1.49	sub g_slave_disconnect($) {
187			my ($node) = @_;
188	root	1.47
189	root	1.49	g_clr $node;
190	root	1.47
191	root	1.49	if (my $mon = delete $GLOBAL_SLAVE{$node}) {
192			while (my ($f, $fv) = each %$mon) {
193			delete $GLOBAL_MON{$f}{$_}
194			for keys %$fv;
195	root	1.51
196			delete $GLOBAL_MON{$f}
197			unless %{ $GLOBAL_MON{$f} };
198	root	1.47	}
199			}
200	root	1.49	}
201
202	root	1.51	# g_mon0 family - stop monitoring
203	root	1.49	$NODE_REQ{g_mon0} = sub {
204	root	1.59	delete $GLOBAL_MON{$_[0]}{$SRCNODE};
205	root	1.51	delete $GLOBAL_MON{$_[0]} unless %{ $GLOBAL_MON{$_[0]} };
206
207	root	1.59	delete $GLOBAL_SLAVE{$SRCNODE}{$_[0]};
208	root	1.49	};
209
210			# g_mon1 family key - start monitoring
211			$NODE_REQ{g_mon1} = sub {
212	root	1.59	undef $GLOBAL_SLAVE{$SRCNODE}{$_[0]};
213			undef $GLOBAL_MON{$_[0]}{$SRCNODE};
214	root	1.49
215	root	1.59	snd $SRCNODE, g_chg1 => $_[0], $GLOBAL_DB{$_[0]};
216	root	1.49	};
217	root	1.47
218	root	1.49	#############################################################################
219			# switch to global mode
220
221			# regularly try to connect to global nodes - maybe use seeding code?
222			$MASTER_TIMER = AE::timer 0, $AnyEvent::MP::Kernel::CONFIG->{monitor_timeout}, sub {
223			(add_node $_)->connect
224	root	1.62	for other_globals;
225	root	1.49	};
226
227			# instantly connect to other global nodes when we learn of them
228			# so we don't have to wait for the timer.
229			#TODO
230	root	1.47	# $GLOBAL_MON{"'g"}{""}{""} = sub {
231			# (add_node $_[1])->connect;
232			# };
233
234	root	1.49	# delete slaves on node-down
235			# clear slave db on node-down
236			$MASTER_MON = mon_nodes sub {
237			g_slave_disconnect $_[0] unless $_[1];
238			};
239
240			# tell everybody who connects that we are a global node
241			push @AnyEvent::MP::Transport::HOOK_GREET, sub {
242			$_[0]{local_greeting}{global} = 1;
243			};
244
245			# connect from a global node
246			sub g_global_connect {
247			my ($node) = @_;
248
249			# we need to set this currently, as to avoid race conditions
250			# because it takes a while until the other global node tells us it is global.
251
252			undef $GLOBAL_DB{"'g"}{$node};
253			undef $LOCAL_DBS{$node}{"'g"}{$node};
254
255			# global nodes send all local databases, merged,
256			# as their local database to global nodes
257			my %db;
258
259			for (values %LOCAL_DBS) {
260			while (my ($f, $fv) = each %$_) {
261			while (my ($k, $kv) = each %$fv) {
262			$db{$f}{$k} = $kv;
263	root	1.46	}
264			}
265	root	1.47	}
266
267	root	1.49	snd $node => g_set => \%db;
268	root	1.4	}
269
270	root	1.49	# send our database to every global node that connects
271			push @AnyEvent::MP::Transport::HOOK_CONNECT, sub {
272			return unless $_[0]{remote_greeting}{global};
273
274			g_global_connect $_[0]{remote_node};
275			};
276
277			# tell our master that we are global now
278			for (values %NODE) {
279			if ($_->{transport} && $_->{transport}{remote_greeting}{global}) {
280			snd $_->{id} => "g_global";
281			g_global_connect $_->{id};
282			}
283	root	1.20	}
284
285	root	1.49	$NODE_REQ{g_global} = sub {
286	root	1.59	g_slave_disconnect $SRCNODE;
287
288			my $node = $NODE{$SRCNODE};
289			$node->{transport}{remote_greeting}{global} = 1;
290
291			g_global_connect $SRCNODE;
292	root	1.49	};
293	root	1.20
294	root	1.56	# enable global mode
295			$GLOBAL = 1;
296
297			# global nodes are their own masters - this
298			# resends global requests and sets the local database.
299			master_set $NODE;
300
301	root	1.49	# now add us to the set of global nodes
302			db_set "'g" => $NODE => undef;
303	root	1.3
304	root	1.58	#############################################################################
305			# compatibility functions for aemp 1.0
306
307	root	1.61	package AnyEvent::MP::Global;
308
309			use base "Exporter";
310			our @EXPORT = qw(grp_reg grp_get grp_mon);
311
312			sub grp_reg($$) {
313			&db_reg
314			}
315
316			sub grp_get($) {
317			my @ports = keys %{ $AnyEvent::MP::Kernel::GLOBAL_DB{$_[0]} };
318
319			@ports ? \@ports : undef
320			}
321
322			sub grp_mon($$) {
323			my ($grp, $cb) = @_;
324
325			db_mon $grp => sub {
326			my ($ports, $add, $chg, $del) = @_;
327
328			$cb->([keys %$ports], $add, $del);
329			};
330			}
331	root	1.1
332			=head1 SEE ALSO
333
334			L<AnyEvent::MP>.
335
336			=head1 AUTHOR
337
338			Marc Lehmann <schmorp@schmorp.de>
339			http://home.schmorp.de/
340
341			=cut
342
343			1
344