view release on metacpan or  search on metacpan

MP/Intro.pod  view on Meta::CPAN

   });

   $| = 1;
   print "> ";
   AnyEvent->condvar->recv;

The first thing the client does is to store the nick name (which is
expected as the only command line argument) in C<$nick>, for further
usage.

The next relevant thing is... finally... the supervisor:

   sub server_connect {
      my $db_mon;
      $db_mon = db_mon eg_chat_server => sub {
         return unless %{ $_[0] };
         undef $db_mon; # stop monitoring

This monitors the C<eg_chat_server> database family. It waits until a
chat server becomes available. When that happens, it "connects" to it
by creating a client port that receives and prints chat messages, and
monitoring it:

      $client = port { print "\r  \r@_\n> " };
      mon $client, sub {
         print "\rdisconnected @_\n";
         &server_connect;
      };

If the client port dies (for whatever reason), the "supervisor" will start
looking for a server again - the semantics of C<db_mon> ensure that it
will immediately find it if there is a server port.

After this, everything is ready: the client will send a C<join> message
with its local port to the server, and start monitoring it:

      $server = (keys %{ $_[0] })[0];

      snd $server, join => $client, $nick;
      mon $server, $client;
   }

This second monitor will ensure that, when the server port crashes or goes
away (e.g. due to network problems), the client port will be killed as
well. This tells the user that the client was disconnected, and will then
start to connect the server again.

The rest of the program deals with the boring details of actually invoking
the supervisor function to start the whole client process and handle the
actual terminal input, sending it to the server.

Now... the "supervisor" in this example is a bit of a cheat - it doesn't
really clean up much (because the cleanup done by AnyEvent::MP suffices),
and there isn't much of a restarting action either - if the server isn't
there because it crashed, well, it isn't there.

In the real world, one would often add a timeout that would trigger when
the server couldn't be found within some time limit, and then complain,
or even try to start a new server. Or the supervisor would have to do
some real cleanups, such as rolling back database transactions when the
database thread crashes. For this simple chat server, however, this simple
supervisor works fine. Hopefully future versions of AnyEvent::MP will
offer some predefined supervisors, for now you will have to code it on
your own.

You should now try to start the server and one or more clients in different
terminal windows (and the seed node):

   perl eg/chat_client nick1
   perl eg/chat_client nick2
   perl eg/chat_server
   aemp run profile seed

And then you can experiment with chatting, killing one or more clients, or
stopping and restarting the server, to see the monitoring in action.

The crucial point you should understand from this example is that
monitoring is usually symmetric: when you monitor some other port,
potentially on another node, that other port usually should monitor you,
too, so when the connection dies, both ports get killed, or at least both
sides can take corrective action. Exceptions are "servers" that serve
multiple clients at once and might only wish to clean up, and supervisors,
who of course should not normally get killed (unless they, too, have a
supervisor).

If you often think in object-oriented terms, then you can think of a port
as an object: C<port> is the constructor, the receive callbacks set by
C<rcv> act as methods, the C<kil> function becomes the explicit destructor
and C<mon> installs a destructor hook. Unlike conventional object oriented
programming, it can make sense to exchange port IDs more freely (for
example, to monitor one port from another), because it is cheap to send
port IDs over the network, and AnyEvent::MP blurs the distinction between
local and remote ports.

Lastly, there is ample room for improvement in this example: the server
should probably remember the nickname in the C<join> handler instead of
expecting it in every chat message, it should probably monitor itself, and
the client should not try to send any messages unless a server is actually
connected.

=head1 PART 3: TIMTOWTDI: Virtual Connections

The chat system developed in the previous sections is very "traditional"
in a way: you start some server(s) and some clients statically and they
start talking to each other.

Sometimes applications work more like "services": They can run on almost
any node and even talk to copies of themselves on other nodes in case they
are distributed. The L<AnyEvent::MP::Global> service for example monitors
nodes joining the network and sometimes even starts itself on other nodes.

One good way to design such services is to put them into a module and
create "virtual connections" to other nodes. We call this the "bridge
head" method, because you start by I<creating a remote port> (the bridge
head) and from that you start to bootstrap your application.

Since that sounds rather theoretical, let us redesign the chat server and
client using this design method.

As usual, we start with the full program - here is the server:

MP/Intro.pod  view on Meta::CPAN

because of network problems (extremely unlikely, but monitoring, after
all, is implemented by passing a message, and messages can get lost), then
this connection loss will eventually trigger the monitoring action. On the
remote node (which in return monitors the client) the port will also be
cleaned up on connection loss. When the remote node comes up again and our
monitoring message can be delivered, it will instantly fail because the
port has been cleaned up in the meantime.

If your head is spinning by now, that's fine - just keep in mind, after
creating a port using C<spawn>, monitor it on the local node, and monitor
"the other side" from the remote node, and all will be cleaned up just
fine.

=head2 Services

Above it was mentioned that C<spawn> automatically loads modules. This can
be exploited in various useful ways.

Assume for a moment you put the server into a file called
F<mymod/chatserver.pm> reachable from the current directory. Then you
could run a node there with:

   aemp run

The other nodes could C<spawn> the server by using
C<mymod::chatserver::client_connect> as init function - without any other
configuration.

Likewise, when you have some service that starts automatically when loaded
(similar to AnyEvent::MP::Global), then you can configure this service
statically:

   aemp profile mysrvnode services mymod::service::
   aemp run profile mysrvnode

And the module will automatically be loaded in the node, as specifying a
module name (with C<::>-suffix) will simply load the module, which is then
free to do whatever it wants.

Of course, you can also do it in the much more standard way by writing
a module (e.g. C<BK::Backend::IRC>), installing it as part of a module
distribution and then configure nodes. For example, if I wanted to run the
Bummskraut IRC backend on a machine named "ruth", I could do this:

   aemp profile ruth addservice BK::Backend::IRC::

And any F<aemp run> on that host will automatically have the Bummskraut
IRC backend running.

There are plenty of possibilities you can use - it's all up to you how you
structure your application.

=head1 PART 4: Coro::MP - selective receive

Not all problems lend themselves naturally to an event-based solution:
sometimes things are easier if you can decide in what order you want to
receive messages, regardless of the order in which they were sent.

In these cases, L<Coro::MP> can provide a nice solution: instead of
registering callbacks for each message type, C<Coro::MP> attaches a
(coro-) thread to a port. The thread can then opt to selectively receive
messages it is interested in. Other messages are not lost, but queued, and
can be received at a later time.

The C<Coro::MP> module is not part of L<AnyEvent::MP>, but a separate
module. It is, however, tightly integrated into C<AnyEvent::MP> - the
ports it creates are fully compatible to C<AnyEvent::MP> ports.

In fact, C<Coro::MP> is more of an extension than a separate module: all
functions exported by C<AnyEvent::MP> are exported by it as well.

To illustrate how programing with C<Coro::MP> looks like, consider the
following (slightly contrived) example: Let's implement a server that
accepts a C<< (write_file =>, $port, $path) >> message with a (source)
port and a filename, followed by as many C<< (data => $port, $data) >>
messages as required to fill the file, followed by an empty C<< (data =>
$port) >> message.

The server only writes a single file at a time, other requests will stay
in the queue until the current file has been finished.

Here is an example implementation that uses L<Coro::AIO> and largely
ignores error handling:

   my $ioserver = port_async {
      while () {
         my ($tag, $port, $path) = get_cond;

         $tag eq "write_file"
            or die "only write_file messages expected";

         my $fh = aio_open $path, O_WRONLY|O_CREAT, 0666
            or die "$path: $!";

         while () {
            my (undef, undef, $data) = get_cond {
               $_[0] eq "data" && $_[1] eq $port
            } 5
               or die "timeout waiting for data message from $port\n";

            length $data or last;

            aio_write $fh, undef, undef, $data, 0;
         };
      }
   };

   mon $ioserver, sub {
      warn "ioserver was killed: @_\n";
   }; 

Let's go through it, section by section.

   my $ioserver = port_async {

Ports can be created by attaching a thread to an existing port via
C<rcv_async>, or as in this example, by calling C<port_async> with the
code to execute as a thread. The C<async> component comes from the fact
that threads are created using the C<Coro::async> function.

The thread runs in a normal port context (so C<$SELF> is set). In
addition, when the thread returns, it will be C<kil> I<normally>, i.e.
without a reason argument.

      while () {
         my ($tag, $port, $path) = get_cond;
            or die "only write_file messages expected";

The thread is supposed to serve many file writes, which is why it
executes in a loop. The first thing it does is fetch the next message,
using C<get_cond>, the "conditional message get". Without arguments, it
merely fetches the I<next> message from the queue, which I<must> be a
C<write_file> message.

The message contains the C<$path> to the file, which is then created:

         my $fh = aio_open $path, O_WRONLY|O_CREAT, 0666
            or die "$path: $!";

Then we enter a loop again, to serve as many C<data> messages as
necessary:

         while () {
            my (undef, undef, $data) = get_cond {
               $_[0] eq "data" && $_[1] eq $port
            } 5
               or die "timeout waiting for data message from $port\n";

This time, the condition is not empty, but instead a code block: similarly
to grep, the code block will be called with C<@_> set to each message in
the queue, and it has to return whether it wants to receive the message or
not.

In this case we are interested in C<data> messages (C<< $_[0] eq "data"
>>), whose first element is the source port (C<< $_[1] eq $port >>).

The condition must be this strict, as it is possible to receive both
C<write_file> messages and C<data> messages from other ports while we
handle the file writing.

The lone C<5> argument at the end is a timeout - when no matching message
is received within C<5> seconds, we assume an error and C<die>.

When an empty C<data> message is received we are done and can close the
file (which is done automatically as C<$fh> goes out of scope):

            length $data or last;

Otherwise we need to write the data:

            aio_write $fh, undef, undef, $data, 0;

And that's basically it. Note that every port thread should have some
kind of supervisor. In our case, the supervisor simply prints any error
message:

   mon $ioserver, sub {
      warn "ioserver was killed: @_\n";
   }; 

Here is a usage example:

   port_async {
      snd $ioserver, write_file => $SELF, "/tmp/unsafe";
      snd $ioserver, data => $SELF, "abc\n";
      snd $ioserver, data => $SELF, "def\n";
      snd $ioserver, data => $SELF;
   }; 

The messages are sent without any flow control or acknowledgement (feel
free to improve). Also, the source port does not actually need to be a
port - any unique ID will do - but port identifiers happen to be a simple
source of network-wide unique IDs.

Apart from C<get_cond> as seen above, there are other ways to receive
messages. The C<write_file> message above could also selectively be
received using a C<get> call:

   my ($port, $path) = get "write_file";

This is simpler, but when some other code part sends an unexpected message
to the C<$ioserver> it will stay in the queue forever. As a rule of thumb,
every threaded port should have a "fetch next message unconditionally"
somewhere, to avoid filling up the queue.

Finally, it is also possible to use more switch-like C<get_conds>:

  get_cond {
     $_[0] eq "msg1" and return sub {
        my (undef, @msg1_data) = @_;
        ...;
     };

     $_[0] eq "msg2" and return sub {
        my (undef, @msg2_data) = @_;
        ...;
     };

     die "unexpected message $_[0] received";
  };

=head1 THE END

This is the end of this introduction, but hopefully not the end of
your career as AEMP user. I hope the tutorial was enough to make the
basic concepts clear. Keep in mind that distributed programming is not
completely trivial, in fact, it's pretty complicated. We hope AEMP makes
it simpler and will be useful to create exciting new applications.

=head1 SEE ALSO

L<AnyEvent::MP>

L<AnyEvent::MP::Global>

L<Coro::MP>

L<AnyEvent>

=head1 AUTHOR

  Robin Redeker <elmex@ta-sa.org>
  Marc Lehmann <schmorp@schmorp.de>