view release on metacpan or  search on metacpan

MP/Kernel.pm  view on Meta::CPAN

our %NODE; # node id to transport mapping, or "undef", for local node
our (%PORT, %PORT_DATA); # local ports

our %RMON; # local ports monitored by remote nodes ($RMON{nodeid}{portid} == cb)
our %LMON; # monitored _local_ ports

#our $GLOBAL; # true if node is a global ("directory") node
our %BINDS;
our $BINDS; # our listeners, as arrayref

our $SRCNODE; # holds the sending node _object_ during _inject
our $GLOBAL;  # true when this is a global node (only set by AnyEvent::MP::Global)

# initialise names for non-networked operation
{
   # ~54 bits, for local port names, lowercase $ID appended
   my $now = AE::now;
   $UNIQ =
      (join "",
         map $alnum[$_],
            $$ / 62 % 62,
            $$ % 62,
            (int $now        ) % 62,
            (int $now *   100) % 62,
            (int $now * 10000) % 62,
      ) . nonce62 4
   ;

   # ~59 bits, for remote port names, one longer than $UNIQ and uppercase at the end to avoid clashes
   $RUNIQ = nonce62 10;
   $RUNIQ =~ s/(.)$/\U$1/;

   $NODE = "";
}

sub NODE() {
   $NODE
}

sub node_of($) {
   my ($node, undef) = split /#/, $_[0], 2;

   $node
}

BEGIN {
   *TRACE = $ENV{PERL_ANYEVENT_MP_TRACE}
      ? sub () { 1 }
      : sub () { 0 };
}

our $DELAY_TIMER;
our @DELAY_QUEUE;

our $delay_run = sub {
   (shift @DELAY_QUEUE or return undef $DELAY_TIMER)->() while 1;
};

sub delay($) {
   push @DELAY_QUEUE, shift;
   $DELAY_TIMER ||= AE::timer 0, 0, $delay_run;
}

=item $AnyEvent::MP::Kernel::SRCNODE

During execution of a message callback, this variable contains the node ID
of the origin node.

The main use of this variable is for debugging output - there are probably
very few other cases where you need to know the source node ID.

=cut

sub _inject {
   warn "RCV $SRCNODE -> " . eval { JSON::XS->new->encode (\@_) } . "\n" if TRACE && @_;

   &{ $PORT{+shift} or return };
}

# this function adds a node-ref, so you can send stuff to it
# it is basically the central routing component.
sub add_node {
   $NODE{$_[0]} || do {
      my ($node) = @_;

      length $node
         or Carp::croak "'undef' or the empty string are not valid node/port IDs";

      # registers itself in %NODE
      new AnyEvent::MP::Node::Remote $node
   }
}

sub snd(@) {
   my ($nodeid, $portid) = split /#/, shift, 2;

   warn "SND $nodeid <- " . eval { JSON::XS->new->encode ([$portid, @_]) } . "\n" if TRACE && @_;

   ($NODE{$nodeid} || add_node $nodeid)
      ->{send} (["$portid", @_]);
}

sub port_is_local($) {
   my ($nodeid, undef) = split /#/, $_[0], 2;

   $nodeid eq $NODE
}

=item snd_to_func $node, $func, @args

Expects a node ID and a name of a function. Asynchronously tries to call
this function with the given arguments on that node.

This function can be used to implement C<spawn>-like interfaces.

=cut

sub snd_to_func($$;@) {
   my $nodeid = shift;

   # on $NODE, we artificially delay... (for spawn)

MP/Kernel.pm  view on Meta::CPAN

#############################################################################
# seed management, try to keep connections to all seeds at all times

our %SEED_NODE;    # seed ID => node ID|undef
our %NODE_SEED;    # map node ID to seed ID
our %SEED_CONNECT; # $seed => transport_connector | 1=connected | 2=connecting
our $SEED_WATCHER;
our $SEED_RETRY;
our %GLOBAL_NODE;  # global => undef

sub seed_connect {
   my ($seed) = @_;

   my ($host, $port) = AnyEvent::Socket::parse_hostport $seed
      or Carp::croak "$seed: unparsable seed address";

   AE::log 9 => "trying connect to seed node $seed.";

   $SEED_CONNECT{$seed} ||= AnyEvent::MP::Transport::mp_connect
      $host, $port,
      on_greeted => sub {
         # called after receiving remote greeting, learn remote node name

         # we rely on untrusted data here (the remote node name) this is
         # hopefully ok, as this can at most be used for DOSing, which is easy
         # when you can do MITM anyway.

         # if we connect to ourselves, nuke this seed, but make sure we act like a seed
         if ($_[0]{remote_node} eq $AnyEvent::MP::Kernel::NODE) {
            require AnyEvent::MP::Global; # every seed becomes a global node currently
            delete $SEED_NODE{$seed};
         } else {
            $SEED_NODE{$seed} = $_[0]{remote_node};
            $NODE_SEED{$_[0]{remote_node}} = $seed;

            # also start global service, in case it isn't running
            # since we probably switch conenctions, maybe we don't need to do this here?
            snd $_[0]{remote_node}, "g_slave";
         }
      },
      sub {
         delete $SEED_CONNECT{$seed};
      }
   ;
}

sub seed_all {
   my @seeds = grep
      !(defined $SEED_NODE{$_} && node_is_up $SEED_NODE{$_}),
      keys %SEED_NODE;

   if (@seeds) {
      # start connection attempt for every seed we are not connected to yet
      seed_connect $_
         for grep !exists $SEED_CONNECT{$_}, @seeds;

      $SEED_RETRY = $SEED_RETRY * 2;
      $SEED_RETRY = $AnyEvent::MP::Kernel::CONFIG->{monitor_timeout}
         if $SEED_RETRY > $AnyEvent::MP::Kernel::CONFIG->{monitor_timeout};

      $SEED_WATCHER = AE::timer $SEED_RETRY, 0, \&seed_all;

   } else {
      # all seeds connected or connecting, no need to restart timer
      undef $SEED_WATCHER;
   }
}

sub seed_again {
   $SEED_RETRY = (1 + rand) * 0.6;
   $SEED_WATCHER ||= AE::timer 0, 0, \&seed_all;
}

# sets new seed list, starts connecting
sub set_seeds(@) {
   %SEED_NODE     = ();
   %NODE_SEED     = ();
   %SEED_CONNECT  = ();

   @SEED_NODE{@_} = ();

   seed_again;
}

# normal nodes only record global node connections
$NODE_REQ{g_global} = sub {
   undef $GLOBAL_NODE{$SRCNODE};
};

mon_nodes sub {
   delete $GLOBAL_NODE{$_[0]}
     unless $_[1];

   return unless exists $NODE_SEED{$_[0]};

   if ($_[1]) {
      # each time a connection to a seed node goes up, make
      # sure it runs the global service.
      snd $_[0], "g_slave";
   } else {
      # if we lost the connection to a seed node, make sure we are seeding
      seed_again;
   }
};

#############################################################################
# keepalive code - used to kepe conenctions to certain nodes alive
# only used by global code atm., but ought to be exposed somehow.
#TODO: should probbaly be done directly by node objects

our $KEEPALIVE_RETRY;
our $KEEPALIVE_WATCHER;
our %KEEPALIVE; # we want to keep these nodes alive
our %KEEPALIVE_DOWN; # nodes that are down currently

sub keepalive_all {
   AE::log 9 => "keepalive: trying to establish connections with: "
                . (join " ", keys %KEEPALIVE_DOWN)
                . ".";

   (add_node $_)->connect
      for keys %KEEPALIVE_DOWN;

   $KEEPALIVE_RETRY = $KEEPALIVE_RETRY * 2;
   $KEEPALIVE_RETRY = $AnyEvent::MP::Kernel::CONFIG->{monitor_timeout}
      if $KEEPALIVE_RETRY > $AnyEvent::MP::Kernel::CONFIG->{monitor_timeout};

   $KEEPALIVE_WATCHER = AE::timer $KEEPALIVE_RETRY, 0, \&keepalive_all;
}

sub keepalive_again {
   $KEEPALIVE_RETRY = (1 + rand) * 0.3;
   keepalive_all;
}

sub keepalive_add {
   return if $KEEPALIVE{$_[0]}++;

   return if node_is_up $_[0];
   undef $KEEPALIVE_DOWN{$_[0]};
   keepalive_again;
}

sub keepalive_del {
   return if --$KEEPALIVE{$_[0]};

   delete $KEEPALIVE     {$_[0]};
   delete $KEEPALIVE_DOWN{$_[0]};

   undef $KEEPALIVE_WATCHER
      unless %KEEPALIVE_DOWN;
}

mon_nodes sub {
   return unless exists $KEEPALIVE{$_[0]};

   if ($_[1]) {
      delete $KEEPALIVE_DOWN{$_[0]};

      undef $KEEPALIVE_WATCHER
         unless %KEEPALIVE_DOWN;
   } else {
      # lost the conenction, try to connect again
      undef $KEEPALIVE_DOWN{$_[0]};
      keepalive_again;
   }
};

#############################################################################
# talk with/to global nodes

# protocol messages:
#
# sent by global nodes
# g_global                  - global nodes send this to all others
#
# database protocol
# g_slave database          - make other global node master of the sender
# g_set database            - global node's database to other global nodes
# g_upd family set del      - update single family (any to global)
#
# slave <-> global protocol
# g_find node               - query addresses for node (slave to global)
# g_found node binds        - node addresses (global to slave)
# g_db_family family id     - send g_reply with data (global to slave)
# g_db_keys   family id     - send g_reply with data (global to slave)
# g_db_values family id     - send g_reply with data (global to slave)
# g_reply id result         - result of any query (global to slave)