view release on metacpan or  search on metacpan

MP/Global.pm  view on Meta::CPAN

=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 List::Util ();

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;

use JSON::XS ();

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

# "import" from Kernel
our %NODE;
our $NODE;
#our $GLOBAL;
our $SRCNODE; # the origin node id
our %NODE_REQ;
our %GLOBAL_NODE;
our $GLOBAL;

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

our %GLOBAL_DB;    # all local databases, merged - empty on non-global nodes
our %LOCAL_DBS; # local databases of other nodes (global and slave)
our %LOCAL_DB;  # this node database

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

# add/replace/del inside a family in the database
# @$del 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} ||= {};

   my %local_set; # extra local set's created by deletes

   # add/replace keys
   while (my ($k, $v) = each %$set) {
      #TODO# optimize duplicate gdb-set's, to some extent, maybe
      # but is probably difficult and slow, so don't for the time being.

      $ldb->{$k} =
      $gdb->{$k} = $v;
   }

   my (@del_local, @del_global); # actual deletes for other global nodes / our slaves

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

      if (my @other = grep exists $LOCAL_DBS{$_}{$family}{$k}, keys %LOCAL_DBS) {
         # key exists in some other db shard(s)

         # if there is a local one, we have to update
         # otherwise, we update and delete on other globals

         if (my $local = List::Util::first { exists $GLOBAL_SLAVE{$_} } @other) {
            $set->{$k} =