view release on metacpan or  search on metacpan

MP/Kernel.pm  view on Meta::CPAN

   my $db = $MON_DB{$family};

   my (@a, @c);

   while (my ($k, $v) = each %$set) {
      exists $db->{$k}
         ? push @c, $k
         : push @a, $k;
      $db->{$k} = $v;
   }

   delete @$db{@$del};

   $_->($db, \@a, \@c, $del)
      for values %{ $LOCAL_MON{$family} };
};

#############################################################################
# configure

sub nodename {
   require POSIX;
   (POSIX::uname ())[1]
}

sub _resolve($) {
   my ($nodeid) = @_;

   my $cv = AE::cv;
   my @res;

   $cv->begin (sub {
      my %seen;
      my @refs;
      for (sort { $a->[0] <=> $b->[0] } @res) {
         push @refs, $_->[1] unless $seen{$_->[1]}++
      }
      shift->send (@refs);
   });

   my $idx;
   for my $t (split /,/, $nodeid) {
      my $pri = ++$idx;

      $t = length $t ? nodename . ":$t" : nodename
         if $t =~ /^\d*$/;
      
      my ($host, $port) = AnyEvent::Socket::parse_hostport $t, 0
         or Carp::croak "$t: unparsable transport descriptor";

      $port = "0" if $port eq "*";

      if ($host eq "*") {
         $cv->begin;

         my $get_addr = sub {
            my @addr;

            require Net::Interface;

            # Net::Interface hangs on some systems, so hope for the best
            local $SIG{ALRM} = 'DEFAULT';
            alarm 2;

            for my $if (Net::Interface->interfaces) {
               # we statically lower-prioritise ipv6 here, TODO :()
               for $_ ($if->address (Net::Interface::AF_INET ())) {
                  next if /^\x7f/; # skip localhost etc.
                  push @addr, $_;
               }
               for ($if->address (Net::Interface::AF_INET6 ())) {
                  #next if $if->scope ($_) <= 2;
                  next unless /^[\x20-\x3f\xfc\xfd]/; # global unicast, site-local unicast
                  push @addr, $_;
               }
            }

            alarm 0;

            @addr
         };

         my @addr;

         if (AnyEvent::WIN32) {
            @addr = $get_addr->();
         } else {
            # use a child process, as Net::Interface is big, and we need it only once.

            pipe my $r, my $w
               or die "pipe: $!";

            if (fork eq 0) {
               close $r;
               syswrite $w, pack "(C/a*)*", $get_addr->();
               require POSIX;
               POSIX::_exit (0);
            } else {
               close $w;

               my $addr;

               1 while sysread $r, $addr, 1024, length $addr;
               
               @addr = unpack "(C/a*)*", $addr;
            }
         }

         for my $ip (@addr) {
            push @res, [
               $pri += 1e-5,
               AnyEvent::Socket::format_hostport AnyEvent::Socket::format_address $ip, $port
            ];
         }
         $cv->end;
      } else {
         $cv->begin;
         AnyEvent::Socket::resolve_sockaddr $host, $port, "tcp", 0, undef, sub {
            for (@_) {
               my ($service, $host) = AnyEvent::Socket::unpack_sockaddr $_->[3];
               push @res, [