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::MP;
use AnyEvent::MP::Kernel;

$AnyEvent::MP::Kernel::WARN->(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 $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_chg2 => $_[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}};

   snd $SRCNODE->{id}, g_chg1 => $_[0], $_[1],
      defined $_[1] ? $GLOBAL_DB{$_[0]}{$_[1]} : $GLOBAL_DB{$_[0]};
};

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

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

=back

=head1 SEE ALSO

L<AnyEvent::MP>.

=head1 AUTHOR

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

=cut

1

Revision:	1.50
Committed:	Sat Mar 3 13:07:19 2012 UTC (12 years, 2 months ago) by root
Branch:	MAIN
Changes since 1.49:	+3 -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.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
34	root	1.1	use AnyEvent::MP;
35			use AnyEvent::MP::Kernel;
36	root	1.20
37	root	1.5	$AnyEvent::MP::Kernel::WARN->(7, "starting global service.");
38
39	root	1.8	#############################################################################
40	root	1.44	# node protocol parts for global nodes
41	root	1.8
42	root	1.49	package AnyEvent::MP::Kernel;
43	root	1.8
44	root	1.49	# TODO: this is ugly (classical use vars vs. our),
45			# maybe this should go into MP::Kernel
46	root	1.26
47	root	1.49	our %NODE;
48			our $NODE;
49			our $LISTENER;
50	root	1.46
51	root	1.49	our $GLOBAL;
52			our $MASTER;
53			our $MASTER_MON;
54			our $MASTER_TIMER;
55	root	1.46
56	root	1.49	our %GLOBAL_SLAVE;
57	root	1.26
58	root	1.49	our %GLOBAL_DB; # global db
59			our %LOCAL_DBS; # local databases of other global nodes
60			our %LOCAL_DB; # this node database
61	root	1.46
62	root	1.49	our $SRCNODE;
63			our %NODE_REQ;
64	root	1.46
65	root	1.49	# only in global code
66			our %GLOBAL_MON; # monitors {family}{"" or key}
67	root	1.47
68	root	1.49	sub other_globals() {
69			grep $_ ne $NODE && node_is_up $_, keys %{ $GLOBAL_DB{"'g"} }
70			}
71	root	1.47
72	root	1.49	# broadcasts a message to all other global nodes
73			sub g_broadcast {
74			snd $_, @_
75			for other_globals;
76			}
77	root	1.46
78	root	1.49	sub g_mon_check {
79			warn "g_mon_check<@_>\n";#d#
80	root	1.47
81	root	1.49	my %node = (
82			%{ $GLOBAL_MON{$_[1]}{$_[2]} },
83			%{ $GLOBAL_MON{$_[1]}{"" } },
84			%{ $GLOBAL_MON{"" }{"" } },
85			);
86	root	1.47
87	root	1.50	snd $_ => g_chg2 => $_[1], $_[2], @_ > 2 ? $_[3] : ()
88	root	1.49	for keys %node;
89			}
90	root	1.46
91	root	1.49	# add/replace a key in the database
92			sub g_add($$$$) {
93			$LOCAL_DBS{$_[0]}{$_[1]}{$_[2]} =
94			$GLOBAL_DB {$_[1]}{$_[2]} = $_[3];
95	root	1.46
96	root	1.49	g_broadcast g_add => $_[1] => $_[2] => $_[3]
97			if exists $GLOBAL_SLAVE{$_[0]};
98	root	1.46
99	root	1.49	warn "g_add<@_>\n";#d#
100			&g_mon_check;
101			}
102	root	1.46
103	root	1.49	# delete a key from the database
104			sub g_del($$$) {
105			delete $LOCAL_DBS{$_[0]}{$_[1]}{$_[2]};
106
107			g_broadcast g_del => $_[1] => $_[2]
108			if exists $GLOBAL_SLAVE{$_[0]};
109
110			delete $GLOBAL_DB{$_[1]}{$_[2]};
111
112			# check if other node maybe still has the key, then we don't delete, but add
113			for (values %LOCAL_DBS) {
114			if (exists $_->{$_[1]}{$_[2]}) {
115			$GLOBAL_DB{$_[1]}{$_[2]} = $_->{$_[1]}{$_[2]};
116			last;
117	root	1.46	}
118			}
119
120	root	1.49	warn "g_del<@_>\n";#d#
121			&g_mon_check;
122			}
123	root	1.46
124	root	1.49	# delete all keys from a database
125			sub g_clr($) {
126			my ($node) = @_;
127
128			my $db = $LOCAL_DBS{$node};
129			while (my ($f, $k) = each %$db) {
130			g_del $node, $f => $_
131			for keys %$k;
132	root	1.46	}
133
134	root	1.49	delete $LOCAL_DBS{$node};
135			}
136
137			# set the whole (node-local) database - previous value must be empty
138			sub g_set($$) {
139			my ($node, $db) = @_;
140
141			while (my ($f, $k) = each %$db) {
142			g_add $node, $f => $_ => delete $k->{$_}
143			for keys %$k;
144	root	1.46	}
145	root	1.49	}
146	root	1.46
147	root	1.49	# gather node databases from slaves
148	root	1.46
149	root	1.49	# other node wants to make us the master
150			$NODE_REQ{g_slave} = sub {
151			my ($db) = @_;
152	root	1.46
153	root	1.49	my $node = $SRCNODE->{id};
154			undef $GLOBAL_SLAVE{$node};
155			g_set $node, $db;
156			};
157	root	1.46
158	root	1.49	$NODE_REQ{g_add} = sub {
159			&g_add ($SRCNODE->{id}, @_);
160			};
161	root	1.46
162	root	1.49	$NODE_REQ{g_del} = sub {
163			&g_del ($SRCNODE->{id}, @_);
164			};
165	root	1.46
166	root	1.49	$NODE_REQ{g_set} = sub {
167			g_set $SRCNODE->{id}, $_[0];
168			};
169	root	1.46
170	root	1.49	$NODE_REQ{g_find} = sub {
171			my ($node) = @_;
172	root	1.46
173	root	1.49	snd $SRCNODE->{id}, g_found => $node, $GLOBAL_DB{"'l"}{$node};
174			};
175	root	1.46
176	root	1.49	# monitoring
177	root	1.47
178	root	1.49	sub g_slave_disconnect($) {
179			my ($node) = @_;
180	root	1.47
181	root	1.49	g_clr $node;
182	root	1.47
183	root	1.49	if (my $mon = delete $GLOBAL_SLAVE{$node}) {
184			while (my ($f, $fv) = each %$mon) {
185			delete $GLOBAL_MON{$f}{$_}
186			for keys %$fv;
187	root	1.47	}
188			}
189	root	1.49	}
190
191			# g_mon0 family key - stop monitoring
192			$NODE_REQ{g_mon0} = sub {
193			delete $GLOBAL_MON{$_[0]}{$_[1]}{$SRCNODE->{id}};
194			delete $GLOBAL_SLAVE{$SRCNODE->{id}}{$_[0]}{$_[1]};
195			};
196
197			# g_mon1 family key - start monitoring
198			$NODE_REQ{g_mon1} = sub {
199			undef $GLOBAL_SLAVE{$SRCNODE->{id}}{$_[0]}{$_[1]};
200			undef $GLOBAL_MON{$_[0]}{$_[1]}{$SRCNODE->{id}};
201
202	root	1.50	snd $SRCNODE->{id}, g_chg1 => $_[0], $_[1],
203			defined $_[1] ? $GLOBAL_DB{$_[0]}{$_[1]} : $GLOBAL_DB{$_[0]};
204	root	1.49	};
205	root	1.47
206	root	1.49	#############################################################################
207			# switch to global mode
208
209			$GLOBAL = 1;
210			$MASTER = $NODE;
211			undef $GLOBAL_SLAVE{$NODE}; # we are our own master (and slave)
212			$LOCAL_DBS{$NODE} = { %LOCAL_DB };
213
214			# regularly try to connect to global nodes - maybe use seeding code?
215			$MASTER_TIMER = AE::timer 0, $AnyEvent::MP::Kernel::CONFIG->{monitor_timeout}, sub {
216			(add_node $_)->connect
217			for keys %{ $GLOBAL_DB{"'g"} };
218			};
219
220			# instantly connect to other global nodes when we learn of them
221			# so we don't have to wait for the timer.
222			#TODO
223	root	1.47	# $GLOBAL_MON{"'g"}{""}{""} = sub {
224			# (add_node $_[1])->connect;
225			# };
226
227	root	1.49	# delete slaves on node-down
228			# clear slave db on node-down
229			$MASTER_MON = mon_nodes sub {
230			g_slave_disconnect $_[0] unless $_[1];
231			};
232
233			# tell everybody who connects that we are a global node
234			push @AnyEvent::MP::Transport::HOOK_GREET, sub {
235			$_[0]{local_greeting}{global} = 1;
236			};
237
238			# connect from a global node
239			sub g_global_connect {
240			my ($node) = @_;
241
242			# we need to set this currently, as to avoid race conditions
243			# because it takes a while until the other global node tells us it is global.
244
245			undef $GLOBAL_DB{"'g"}{$node};
246			undef $LOCAL_DBS{$node}{"'g"}{$node};
247
248			# global nodes send all local databases, merged,
249			# as their local database to global nodes
250			my %db;
251
252			for (values %LOCAL_DBS) {
253			while (my ($f, $fv) = each %$_) {
254			while (my ($k, $kv) = each %$fv) {
255			$db{$f}{$k} = $kv;
256	root	1.46	}
257			}
258	root	1.47	}
259
260	root	1.49	snd $node => g_set => \%db;
261	root	1.4	}
262
263	root	1.49	# send our database to every global node that connects
264			push @AnyEvent::MP::Transport::HOOK_CONNECT, sub {
265			return unless $_[0]{remote_greeting}{global};
266
267			g_global_connect $_[0]{remote_node};
268			};
269
270			# tell our master that we are global now
271			for (values %NODE) {
272			if ($_->{transport} && $_->{transport}{remote_greeting}{global}) {
273			snd $_->{id} => "g_global";
274			g_global_connect $_->{id};
275			}
276	root	1.20	}
277
278	root	1.49	$NODE_REQ{g_global} = sub {
279			g_slave_disconnect $SRCNODE->{id};
280			$SRCNODE->{transport}{remote_greeting}{global} = 1;
281			g_global_connect $SRCNODE->{id};
282			};
283	root	1.20
284	root	1.49	# now add us to the set of global nodes
285			db_set "'g" => $NODE => undef;
286	root	1.3
287	root	1.1	=back
288
289			=head1 SEE ALSO
290
291			L<AnyEvent::MP>.
292
293			=head1 AUTHOR
294
295			Marc Lehmann <schmorp@schmorp.de>
296			http://home.schmorp.de/
297
298			=cut
299
300			1
301