AnyEvent-MP
view release on metacpan or search on metacpan
TODO: maybe disbale warnings bydefault?TODO: listener-scopes (10.0.0.1:4040@vpn) andconnect-scopes ("vpn,public")TODO: document env-variable usageTODO: make node objects responsibleforkeepalive?faq: can't se anythingfaq: all is asynchfaq: how to interface to non-perl nodes?TODO: check gproto, nproto, onconnectTODO: limiting reconnecting speedwhenunreachable? somehowusesame interval timers asforseeding and keepalive?TODO: multiple profiles? also somedefaultprofiles?TODO: export keepalive?TODO:$guard= con$cb->($up)TODO: aempreadlinesupportTODO: gleeco re: AE::MP::DataConn -TODO: version both in MP.pm and MP/Config.pm because of cpan indexer2.02 Sun Jul 29 04:22:53 CEST 2018- hardcode version in MP.pm to help the CPAN indexer.
# monitoringmon$port,$cb->(@msg)# callback is invoked on deathmon$port,$localport# kill localport on abnormal deathmon$port,$localport,@msg# send message on death# temporarily execute code in port contextpeval$port,sub{die"kill the port!"};# execute callbacks in $SELF port contextmy$timer= AE::timer 1, 0, psub {die"kill the port, delayed";};# distributed database - modificationdb_set$family=>$subkey[=>$value]# add a subkeydb_del$family=>$subkey...# delete one or more subkeysdb_reg$family=>$port[=>$value]# register a port# distributed database - queriesdb_family$family=>$cb->(\%familyhash)
closure is executed, sets up the environment in the same way as in C<rcv>callbacks, i.e. runtime errors will cause the port to get C<kil>ed.The effect is basically asifit returned C<<sub{ peval$SELF,sub{BLOCK },@_} >>.This is usefulwhenyou register callbacks from C<rcv> callbacks:rcvdelayed_reply=>sub{my($delay,@reply) =@_;my$timer= AE::timer$delay, 0, psub {snd@reply,$SELF;};};=cutsub psub(&) {my $cb = shift;my $port = $SELF
and ($cb+= 0, Guard::guard {$node->unmonitor ($port,$cb) })}=item $guard = mon_guard $port, $ref, $ref...Monitors the given C<$port> and keeps the passed references. When the portis killed, the references will be freed.Optionally returns a guard that will stop the monitoring.This function is useful when you create e.g. timers or other watchers andwant to free them when the port gets killed (note the use of C<psub>):$port->rcv (start => sub {my $timer; $timer = mon_guard $port, AE::timer 1, 1, psub {undef $timer if 0.9 < rand;});});=cutsubmon_guard {my($port,@refs) =@_;#TODO: mon-less form?
This is simply a utility function that comes in handy attimes- theAnyEvent::MP author is not convinced of the wisdom of having it, though,so it may go away in the future.=cutsub after($@) {my ($timeout, @action) = @_;my $t; $t = AE::timer $timeout, 0, sub {undef $t;ref $action[0]? $action[0](): snd @action;};}#=item $cb2 = timeout $seconds, $cb[, @args]=item cal $port, @msg, $callback[, $timeout]
my$timeout=ref$_[-1] ?undef:pop;my$cb=pop;my$port= port {undef$timeout;kil$SELF;&$cb;};if(defined$timeout) {$timeout= AE::timer$timeout, 0,sub{undef$timeout;kil$port;$cb->();};}else{mon$_[0],sub{kil$port;$cb->();};}
MP/DataConn.pm view on Meta::CPAN
our$ID="a";our%STATE;# another node tells us to await a connectionsub_expect {my($id,$port,$timeout,$initfunc,@initdata) =@_;$STATE{$id} = {id=>$id,to=> (AE::timer$timeout, 0,sub{$STATE{$id}{done}(undef);}),done=>sub{my($hdl,$error) =@_;%{delete$STATE{$id}} = ();if(defined$hdl) {(AnyEvent::MP::Kernel::load_func$initfunc)->(@initdata,$hdl);}else{
MP/DataConn.pm view on Meta::CPAN
my$state=$STATE{$id}orreturn;my$addr=$AnyEvent::MP::Global::addr{$node};@$addrorreturn$state->{done}(undef,"$node: no listeners found");# I love hardcoded constants !$state->{next} = AE::timer 0, 2,sub{my$endpoint=shift@$addrorreturndelete$state->{next};my($host,$port) = AnyEvent::Socket::parse_hostport$endpointorreturn;my$transport;$transport= AnyEvent::MP::Transport::mp_connect$host,$port,protocol=>"aemp-dataconn",local_greeting=> {dataconn_id=>$id},
MP/DataConn.pm view on Meta::CPAN
my$port=$SELFor Carp::croak"AnyEvent::MP::DataConn::connect_to must be called in port context";$node= node_of$node;my$id= (++$ID) ."\@$NODE";# damn, why do my simple state hashes resemble objects so quicklymy$state=$STATE{$id} = {id=> (++$ID) ."\@$NODE",to=> (AE::timer$timeout, 0,sub{$STATE{$id}{done}(undef,"$node: unable to establish connection within $timeout seconds");}),done=>sub{my($hdl,$error) =@_;delete$AnyEvent::MP::Global::ON_SETUP{$id};%{delete$STATE{$id}} = ();if(defined$hdl) {$cb->($hdl);
MP/Kernel.pm view on Meta::CPAN
our$DELAY_TIMER;our@DELAY_QUEUE;our$delay_run=sub{(shift@DELAY_QUEUEorreturnundef$DELAY_TIMER)->()while1;};subdelay($) {push@DELAY_QUEUE,shift;$DELAY_TIMER||= AE::timer 0, 0,$delay_run;}=item $AnyEvent::MP::Kernel::SRCNODEDuring execution of a message callback, this variable contains the node IDof the origin node.The main use of this variable is for debugging output - there are probablyvery few other cases where you need to know the source node ID.
MP/Kernel.pm view on Meta::CPAN
if(@seeds) {# start connection attempt for every seed we are not connected to yetseed_connect$_forgrep!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 timerundef$SEED_WATCHER;}}subseed_again {$SEED_RETRY= (1 +rand) * 0.6;$SEED_WATCHER||= AE::timer 0, 0, \&seed_all;}# sets new seed list, starts connectingsubset_seeds(@) {%SEED_NODE= ();%NODE_SEED= ();%SEED_CONNECT= ();@SEED_NODE{@_} = ();
MP/Kernel.pm view on Meta::CPAN
. (join" ",keys%KEEPALIVE_DOWN).".";(add_node$_)->connectforkeys%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;}subkeepalive_again {$KEEPALIVE_RETRY= (1 +rand) * 0.3;keepalive_all;}subkeepalive_add {returnif$KEEPALIVE{$_[0]}++;
returnif$self->{transport};returnif$self->{connect_w};# we unweaken the node reference, in case it was weakened before$AnyEvent::MP::Kernel::NODE{$self->{id}}=$AnyEvent::MP::Kernel::NODE{$self->{id}};Scalar::Util::weaken$self;$self->{connect_to} ||= AE::timer$AnyEvent::MP::Kernel::CONFIG->{connect_interval}, 0,sub{$self->transport_error (transport_error=>$self->{id},"connect timeout");};# maybe @$addresses?my$addresses=$AnyEvent::MP::Kernel::GLOBAL_DB{"'l"}{$self->{id}};if($addresses) {$self->connect_to ($addresses);}else{# on global nodes, all bets are off now - we either know the node, or we don't
my$monitor=$AnyEvent::MP::Kernel::CONFIG->{monitor_timeout};my$interval=$AnyEvent::MP::Kernel::CONFIG->{connect_interval};$interval= ($monitor-$interval) /@$addressesif($monitor-$interval) /@$addresses<$interval;$interval= 0.4if$interval< 0.4;my@endpoints=reverse@$addresses;$self->{connect_w} = AE::timer 0,$interval* (0.9 + 0.1 *rand),sub{my$endpoint=pop@endpointsorreturn;AE::log9=>"connecting to $self->{id} at $endpoint";$self->{trial}{$endpoint} ||=do{my($host,$port) = AnyEvent::Socket::parse_hostport$endpointorreturnAE::logcritical=>"$self->{id}: '$endpoint' is not a resolved node reference.";AnyEvent::MP::Transport::mp_connect
subtransport_reset {my($self) =@_;Scalar::Util::weaken$self;$self->{send} =sub{if(++$DELAY> 0) {my$msg=$_[0];push@DELAY,sub{ AnyEvent::MP::Kernel::_inject (@$msg) };$DELAY_W||= AE::timer 0, 0,$send_delayed;return;}local$AnyEvent::MP::Kernel::SRCNODE=$AnyEvent::MP::Kernel::NODE;AnyEvent::MP::Kernel::_inject (@{$_[0] });};}subtransport_connect {my($self,$tp) =@_;
AE::log9=>"I refuse to talk to myself ($tp->{peerhost}:$tp->{peerport})";}subkill{my(undef,@args) =@_;# we _always_ delay kil's, to avoid calling mon callbacks# from anything but the event loop context.$DELAY= 1;push@DELAY,sub{ AnyEvent::MP::Kernel::_kill (@args) };$DELAY_W||= AE::timer 0, 0,$send_delayed;}submonitor {# maybe always delay, too?if($DELAY_W) {my@args=@_;push@DELAY,sub{ AnyEvent::MP::Kernel::_monitor (@args) };return;}&AnyEvent::MP::Kernel::_monitor;
# monitoringmon$port,$cb->(@msg)# callback is invoked on deathmon$port,$localport# kill localport on abnormal deathmon$port,$localport,@msg# send message on death# temporarily execute code in port contextpeval$port,sub{die"kill the port!"};# execute callbacks in $SELF port contextmy$timer= AE::timer 1, 0, psub {die"kill the port, delayed";};# distributed database - modificationdb_set$family=>$subkey[=>$value]# add a subkeydb_del$family=>$subkey...# delete one or more subkeysdb_reg$family=>$port[=>$value]# register a port# distributed database - queriesdb_family$family=>$cb->(\%familyhash)
"rcv"callbacks, i.e. runtime errors will cause the port to get"kil"ed.The effect is basically asifit returned "sub{ peval$SELF,sub{BLOCK },@_}".This is usefulwhenyou register callbacks from"rcv"callbacks:rcvdelayed_reply=>sub{my($delay,@reply) =@_;my$timer= AE::timer$delay, 0, psub {snd@reply,$SELF;};};$guard= mon$port,$rcvport# kill $rcvport when $port dies$guard= mon$port# kill $SELF when $port dies$guard= mon$port,$cb->(@reason)# call $cb when $port dies$guard= mon$port,$rcvport,@msg# send a message when $port diesMonitor thegivenport anddosomethingwhenthe port is killed ormessages to it were lost, and optionallyreturna guard that can be
Example:sendus a restart messagewhenanother$portis killed.mon$port,$self=>"restart";$guard= mon_guard$port,$ref,$ref...Monitors thegiven$portand keeps the passed references. When theport is killed, the references will be freed.Optionally returns a guard that will stop the monitoring.This function is usefulwhenyou create e.g. timers or otherwatchers and want to free themwhenthe port gets killed (note the$port->rcv (start=>sub{my$timer;$timer= mon_guard$port, AE::timer 1, 1, psub {undef$timerif0.9 <rand;});});kil$port[,@reason]Kill the specified portwiththegiven@reason.Ifno@reasonis specified, then the port is killed"normally"-monitor callback will be invoked, but the kil will not cause linkedports ("mon $mport, $lport"form) to get killed.
delete$to{$node};@neigh=grep$_ne$NODE,@neigh;$node," -> ", (join" ",@neigh),"\n";formy$neigh(@neigh) {unless($seen{$neigh}++) {$cv->begin;$to{$neigh} = AE::timer 15, 0,sub{"$neigh (timeout)\n";$exit= 1;$cv->end;};AnyEvent::MP::Kernel::eval_on$neigh,"AnyEvent::MP::Kernel::up_nodes"=>$SELF=>$neigh;}}$cv->end;};
};$cv->recv;}subnode_eval {my($node,$expr) =@_;init;my$cv= AE::cv;my$to= AE::timer 5, 0,sub{exit1 };AnyEvent::MP::Kernel::eval_on$node,$expr, port {&$cv};mon$node,$cv;my($err,@res) =$cv->recv;die"$err @res"iflength$err;+(substrJSON::XS->new->encode (\@res), 1, -1),"\n";}subdocmd;our%CMD= (snd=>sub{my$port=shift@ARGV;init;snd$port,@ARGV;@ARGV= ();my$cv= AE::cv;my$to= AE::timer 5, 0,sub{exit1 };mon$port,$cv;my$reply= portsub{&$cv};snd node_of$port,snd=>$reply,"message sent successfully";join" ",$cv->recv,"\n";},cal=>sub{my$port=shift@ARGV;init;
( run in 0.549 second using v1.01-cache-2.11-cpan-87723dcf8b7 )