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

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

   # tell subscribers we have added or replaced a key
   snd $_ => g_chg2 => $_[1], $_[2], $_[3]
      for keys %{ $GLOBAL_MON{$_[1]} };
}

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

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

   delete $GLOBAL_DB{$_[1]}{$_[2]};
   delete $GLOBAL_DB{$_[1]} unless %{ $GLOBAL_DB{$_[1]} }; # could be moved below for

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

         # tell subscribers we have added or replaced a key
         snd $_ => g_chg2 => $_[1], $_[2], $val
            for keys %{ $GLOBAL_MON{$_[1]} };

         last;
      }
   }

   # tell subscribers key is actually gone
   snd $_ => g_chg2 => $_[1], $_[2]
      for keys %{ $GLOBAL_MON{$_[1]} };
}

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

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

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

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

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

   delete $GLOBAL_SLAVE{$SRCNODE->{id}}{$_[0]};
};

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

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