--- AnyEvent-MP/MP.pm 2009/08/30 18:51:49 1.69 +++ AnyEvent-MP/MP.pm 2009/09/22 14:14:29 1.93 @@ -1,21 +1,20 @@ =head1 NAME -AnyEvent::MP - multi-processing/message-passing framework +AnyEvent::MP - erlang-style multi-processing/message-passing framework =head1 SYNOPSIS use AnyEvent::MP; - $NODE # contains this node's noderef - NODE # returns this node's noderef - NODE $port # returns the noderef of the port + $NODE # contains this node's node ID + NODE # returns this node's node ID $SELF # receiving/own port id in rcv callbacks # initialise the node so it can send/receive messages - initialise_node; + configure; - # ports are message endpoints + # ports are message destinations # sending messages snd $port, type => data...; @@ -23,31 +22,28 @@ snd @msg_with_first_element_being_a_port; # creating/using ports, the simple way - my $simple_port = port { my @msg = @_; 0 }; + my $simple_port = port { my @msg = @_ }; # creating/using ports, tagged message matching my $port = port; - rcv $port, ping => sub { snd $_[0], "pong"; 0 }; - rcv $port, pong => sub { warn "pong received\n"; 0 }; + rcv $port, ping => sub { snd $_[0], "pong" }; + rcv $port, pong => sub { warn "pong received\n" }; # create a port on another node my $port = spawn $node, $initfunc, @initdata; # monitoring - mon $port, $cb->(@msg) # callback is invoked on death - mon $port, $otherport # kill otherport on abnormal death - mon $port, $otherport, @msg # send message on death + mon $localport, $cb->(@msg) # callback is invoked on death + mon $localport, $otherport # kill otherport on abnormal death + mon $localport, $otherport, @msg # send message on death =head1 CURRENT STATUS - AnyEvent::MP - stable API, should work - AnyEvent::MP::Intro - outdated - AnyEvent::MP::Kernel - mostly stable - AnyEvent::MP::Global - mostly stable - AnyEvent::MP::Node - mostly stable, but internal anyways - AnyEvent::MP::Transport - mostly stable, but internal anyways - - stay tuned. + bin/aemp - stable. + AnyEvent::MP - stable API, should work. + AnyEvent::MP::Intro - explains most concepts. + AnyEvent::MP::Kernel - mostly stable API. + AnyEvent::MP::Global - stable API. =head1 DESCRIPTION @@ -59,15 +55,14 @@ For an introduction to this module family, see the L manual page and the examples under F. -At the moment, this module family is a bit underdocumented. - =head1 CONCEPTS =over 4 =item port -A port is something you can send messages to (with the C function). +Not to be confused with a TCP port, a "port" is something you can send +messages to (with the C function). Ports allow you to register C handlers that can match all or just some messages. Messages send to ports will not be queued, regardless of @@ -88,7 +83,7 @@ (no listening ports). Private nodes cannot talk to other private nodes currently. -=item node ID - C<[a-za-Z0-9_\-.:]+> +=item node ID - C<[A-Z_][a-zA-Z0-9_\-.:]*> A node ID is a string that uniquely identifies the node within a network. Depending on the configuration used, node IDs can look like a @@ -102,20 +97,33 @@ endpoints - binds. Currently, only standard C specifications can be used, which specify TCP ports to listen on. -=item seeds - C +=item seed nodes When a node starts, it knows nothing about the network. To teach the node about the network it first has to contact some other node within the network. This node is called a seed. -Seeds are transport endpoint(s) of as many nodes as one wants. Those nodes -are expected to be long-running, and at least one of those should always -be available. When nodes run out of connections (e.g. due to a network -error), they try to re-establish connections to some seednodes again to -join the network. +Apart from the fact that other nodes know them as seed nodes and they have +to have fixed listening addresses, seed nodes are perfectly normal nodes - +any node can function as a seed node for others. + +In addition to discovering the network, seed nodes are also used to +maintain the network and to connect nodes that otherwise would have +trouble connecting. They form the backbone of an AnyEvent::MP network. + +Seed nodes are expected to be long-running, and at least one seed node +should always be available. They should also be relatively responsive - a +seed node that blocks for long periods will slow down everybody else. + +=item seeds - C + +Seeds are transport endpoint(s) (usually a hostname/IP address and a +TCP port) of nodes thta should be used as seed nodes. -Apart from being sued for seeding, seednodes are not special in any way - -every public node can be a seednode. +The nodes listening on those endpoints are expected to be long-running, +and at least one of those should always be available. When nodes run out +of connections (e.g. due to a network error), they try to re-establish +connections to some seednodes again to join the network. =back @@ -141,8 +149,8 @@ our @EXPORT = qw( NODE $NODE *SELF node_of after - initialise_node - snd rcv mon mon_guard kil reg psub spawn + configure + snd rcv mon mon_guard kil psub spawn cal port ); @@ -158,69 +166,94 @@ The C function returns, and the C<$NODE> variable contains, the node ID of the node running in the current process. This value is initialised by -a call to C. +a call to C. =item $nodeid = node_of $port Extracts and returns the node ID from a port ID or a node ID. -=item initialise_node $profile_name, key => value... +=item configure $profile, key => value... + +=item configure key => value... Before a node can talk to other nodes on the network (i.e. enter -"distributed mode") it has to initialise itself - the minimum a node needs +"distributed mode") it has to configure itself - the minimum a node needs to know is its own name, and optionally it should know the addresses of some other nodes in the network to discover other nodes. -This function initialises a node - it must be called exactly once (or +This function configures a node - it must be called exactly once (or never) before calling other AnyEvent::MP functions. -The first argument is a profile name. If it is C or missing, then -the current nodename will be used instead (i.e. F). +=over 4 -The function first looks up the profile in the aemp configuration (see the -L commandline utility). the profile is calculated as follows: +=item step 1, gathering configuration from profiles -First, all remaining key => value pairs will be used. Then they will be -overwritten by any values specified in the global default configuration -(see the F utility), then the chain of profiles selected, if -any. That means that the values specified in the profile have highest -priority and the values specified via C have lowest -priority. +The function first looks up a profile in the aemp configuration (see the +L commandline utility). The profile name can be specified via the +named C parameter or can simply be the first parameter). If it is +missing, then the nodename (F) will be used as profile name. + +The profile data is then gathered as follows: + +First, all remaining key => value pairs (all of which are conveniently +undocumented at the moment) will be interpreted as configuration +data. Then they will be overwritten by any values specified in the global +default configuration (see the F utility), then the chain of +profiles chosen by the profile name (and any C attributes). + +That means that the values specified in the profile have highest priority +and the values specified directly via C have lowest priority, +and can only be used to specify defaults. If the profile specifies a node ID, then this will become the node ID of this process. If not, then the profile name will be used as node ID. The special node ID of C will be replaced by a random node ID. +=item step 2, bind listener sockets + The next step is to look up the binds in the profile, followed by binding aemp protocol listeners on all binds specified (it is possible and valid to have no binds, meaning that the node cannot be contacted form the outside. This means the node cannot talk to other nodes that also have no binds, but it can still talk to all "normal" nodes). -If the profile does not specify a binds list, then the node ID will be -treated as if it were of the form C, which will be resolved and -used as binds list. +If the profile does not specify a binds list, then a default of C<*> is +used, meaning the node will bind on a dynamically-assigned port on every +local IP address it finds. + +=item step 3, connect to seed nodes -Lastly, the seeds list from the profile is passed to the +As the last step, the seeds list from the profile is passed to the L module, which will then use it to keep -connectivity with at least on of those seed nodes at any point in time. +connectivity with at least one node at any point in time. + +=back -Example: become a distributed node listening on the guessed noderef, or -the one specified via C for the current node. This should be the -most common form of invocation for "daemon"-type nodes. +Example: become a distributed node using the local node name as profile. +This should be the most common form of invocation for "daemon"-type nodes. - initialise_node; + configure Example: become an anonymous node. This form is often used for commandline clients. - initialise_node "anon/"; + configure nodeid => "anon/"; + +Example: configure a node using a profile called seed, which si suitable +for a seed node as it binds on all local addresses on a fixed port (4040, +customary for aemp). + + # use the aemp commandline utility + # aemp profile seed nodeid anon/ binds '*:4040' + + # then use it + configure profile => "seed"; -Example: become a distributed node. If there is no profile of the given -name, or no binds list was specified, resolve C and bind -on the resulting addresses. + # or simply use aemp from the shell again: + # aemp run profile seed - initialise_node "localhost:4044"; + # or provide a nicer-to-remember nodeid + # aemp run profile seed nodeid "$(hostname)" =item $SELF @@ -352,9 +385,9 @@ sub rcv($@) { my $port = shift; - my ($noderef, $portid) = split /#/, $port, 2; + my ($nodeid, $portid) = split /#/, $port, 2; - $NODE{$noderef} == $NODE{""} + $NODE{$nodeid} == $NODE{""} or Carp::croak "$port: rcv can only be called on local ports, caught"; while (@_) { @@ -455,24 +488,13 @@ messages to it were lost, and optionally return a guard that can be used to stop monitoring again. -C effectively guarantees that, in the absence of hardware failures, -after starting the monitor, either all messages sent to the port will -arrive, or the monitoring action will be invoked after possible message -loss has been detected. No messages will be lost "in between" (after -the first lost message no further messages will be received by the -port). After the monitoring action was invoked, further messages might get -delivered again. - -Note that monitoring-actions are one-shot: once messages are lost (and a -monitoring alert was raised), they are removed and will not trigger again. - In the first form (callback), the callback is simply called with any number of C<@reason> elements (no @reason means that the port was deleted "normally"). Note also that I<< the callback B never die >>, so use C if unsure. In the second form (another port given), the other port (C<$rcvport>) -will be C'ed with C<@reason>, iff a @reason was specified, i.e. on +will be C'ed with C<@reason>, if a @reason was specified, i.e. on "normal" kils nothing happens, while under all other conditions, the other port is killed with the same reason. @@ -482,13 +504,33 @@ In the last form (message), a message of the form C<@msg, @reason> will be C. +Monitoring-actions are one-shot: once messages are lost (and a monitoring +alert was raised), they are removed and will not trigger again. + As a rule of thumb, monitoring requests should always monitor a port from a local port (or callback). The reason is that kill messages might get lost, just like any other message. Another less obvious reason is that -even monitoring requests can get lost (for exmaple, when the connection +even monitoring requests can get lost (for example, when the connection to the other node goes down permanently). When monitoring a port locally these problems do not exist. +C effectively guarantees that, in the absence of hardware failures, +after starting the monitor, either all messages sent to the port will +arrive, or the monitoring action will be invoked after possible message +loss has been detected. No messages will be lost "in between" (after +the first lost message no further messages will be received by the +port). After the monitoring action was invoked, further messages might get +delivered again. + +Inter-host-connection timeouts and monitoring depend on the transport +used. The only transport currently implemented is TCP, and AnyEvent::MP +relies on TCP to detect node-downs (this can take 10-15 minutes on a +non-idle connection, and usually around two hours for idle conenctions). + +This means that monitoring is good for program errors and cleaning up +stuff eventually, but they are no replacement for a timeout when you need +to ensure some maximum latency. + Example: call a given callback when C<$port> is killed. mon $port, sub { warn "port died because of <@_>\n" }; @@ -504,9 +546,9 @@ =cut sub mon { - my ($noderef, $port) = split /#/, shift, 2; + my ($nodeid, $port) = split /#/, shift, 2; - my $node = $NODE{$noderef} || add_node $noderef; + my $node = $NODE{$nodeid} || add_node $nodeid; my $cb = @_ ? shift : $SELF || Carp::croak 'mon: called with one argument only, but $SELF not set,'; @@ -525,6 +567,7 @@ $node->monitor ($port, $cb); defined wantarray + and $cb += 0 and AnyEvent::Util::guard { $node->unmonitor ($port, $cb) } } @@ -592,13 +635,19 @@ exists or it runs out of package names. The init function is then called with the newly-created port as context -object (C<$SELF>) and the C<@initdata> values as arguments. +object (C<$SELF>) and the C<@initdata> values as arguments. It I +call one of the C functions to set callbacks on C<$SELF>, otherwise +the port might not get created. A common idiom is to pass a local port, immediately monitor the spawned port, and in the remote init function, immediately monitor the passed local port. This two-way monitoring ensures that both ports get cleaned up when there is a problem. +C guarantees that the C<$initfunc> has no visible effects on the +caller before C returns (by delaying invocation when spawn is +called for the local node). + Example: spawn a chat server port on C<$othernode>. # this node, executed from within a port context: @@ -622,6 +671,7 @@ my $port = shift; my $init = shift; + # rcv will create the actual port local $SELF = "$NODE#$port"; eval { &{ load_func $init } @@ -630,16 +680,16 @@ } sub spawn(@) { - my ($noderef, undef) = split /#/, shift, 2; + my ($nodeid, undef) = split /#/, shift, 2; my $id = "$RUNIQ." . $ID++; $_[0] =~ /::/ or Carp::croak "spawn init function must be a fully-qualified name, caught"; - snd_to_func $noderef, "AnyEvent::MP::_spawn" => $id, @_; + snd_to_func $nodeid, "AnyEvent::MP::_spawn" => $id, @_; - "$noderef#$id" + "$nodeid#$id" } =item after $timeout, @msg @@ -666,6 +716,58 @@ }; } +=item cal $port, @msg, $callback[, $timeout] + +A simple form of RPC - sends a message to the given C<$port> with the +given contents (C<@msg>), but adds a reply port to the message. + +The reply port is created temporarily just for the purpose of receiving +the reply, and will be Ced when no longer needed. + +A reply message sent to the port is passed to the C<$callback> as-is. + +If an optional time-out (in seconds) is given and it is not C, +then the callback will be called without any arguments after the time-out +elapsed and the port is Ced. + +If no time-out is given, then the local port will monitor the remote port +instead, so it eventually gets cleaned-up. + +Currently this function returns the temporary port, but this "feature" +might go in future versions unless you can make a convincing case that +this is indeed useful for something. + +=cut + +sub cal(@) { + 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->(); + }; + } + + push @_, $port; + &snd; + + $port +} + =back =head1 AnyEvent::MP vs. Distributed Erlang @@ -688,7 +790,7 @@ Erlang relies on special naming and DNS to work everywhere in the same way. AEMP relies on each node somehow knowing its own address(es) (e.g. by -configuraiton or DNS), but will otherwise discover other odes itself. +configuration or DNS), but will otherwise discover other odes itself. =item * Erlang has a "remote ports are like local ports" philosophy, AEMP uses "local ports are like remote ports". @@ -711,7 +813,7 @@ needs a queue. AEMP is event based, queuing messages would serve no useful purpose. For the same reason the pattern-matching abilities of AnyEvent::MP are more limited, as there is little need to be able to -filter messages without dequeing them. +filter messages without dequeuing them. (But see L for a more Erlang-like process model on top of AEMP). @@ -827,6 +929,9 @@ L - network maintainance and port groups, to find your applications. +L - simple service to display log messages from +all nodes. + L. =head1 AUTHOR