AnyEvent-MP/MP/Intro.pod

=head1 Message Passing for the non-blocked Mind

=head2 Introduction and Terminology

This is a Tutorial about how to get the swing of the new L<AnyEvent::MP>
module. Which allows us to transparently pass messages to our own process and
to other process on other or the same host.

What kind of messages? Well, basically a message here means a list of Perl
strings, numbers, hashes and arrays, mostly everything that can be expressed as
L<JSON> string.

And next you might ask: between which entities are those messages "passed"?
Basically between C<nodes>, which are basically your applications (as in
processes) that use L<AnyEvent::MP> that run either on the same or different
hosts.

In this Tutorial I'll show you how to write a simple chat server based on
L<AnyEvent::MP>.

=head2 The Chat Client

Ok, lets start by implementing the "frontend" of the client. We will delay the
explanation and the code of the server until we finished the client, as the
most complex things actually happen in the client.

We will use L<AnyEvent::Handle> to do non-blocking IO read on standard input:

   #!perl
   use AnyEvent;
   use AnyEvent::Handle;

   sub send_message {
      die "This is where we will send the messages to the server"
         . "in the next step of this tutorial.\n"
   }

   # make an AnyEvent condition variable for the 'quit' condition
   # (when we want to exit the client).
   my $quit_cv = AnyEvent->condvar;

   my $stdin_hdl = AnyEvent::Handle->new (
      fh      => \*STDIN,
      on_read => sub {
         my ($hdl) = @_;

         $hdl->push_read (line => sub {
            my ($hdl, $line) = @_;

            if ($line =~ /^\/quit/) { # /quit will end the client
               $quit_cv->send;

            } else {
               send_message ($line);
            }
         });
      }
   );

   $quit_cv->recv;

This is now a very basic client. Explaining explicitly what L<AnyEvent::Handle>
does or what a I<condvar> is all about is out of scope of this document, please
consult L<AnyEvent::Intro> or the manpages for L<AnyEvent> and
L<AnyEvent::Handle>.

=head2 First Step Into Messaging

Now we take a look at L<AnyEvent::MP>. We need to know what to do in
C<send_message>. This is an example how it might look like:

   ... # the use lines from the above snippet

   use AnyEvent::MP;

   sub send_message {
      my ($msg) = @_;

      snd $server_port, message => $msg;
   }

   ... # the rest of the above script

The C<snd> function is exported by L<AnyEvent::MP>, it stands for 'send a
message'. The first argument is the I<port> (a I<port> is something that can
receive messages) of the server which will receive the message . How we get it
will be explained in the next step.  The next arguments of C<snd> are
C<'message'> and C<$msg> are the first two elements of the I<message> (a
I<message> in L<AnyEvent::MP> is a be a simple list of values, which can be
sent to I<ports>).

Thats all fine so far, but how do we get the C<$server_port>? Well, we will
need to get the unique I<port id> of the server's port where he wants to
receive all the incoming chat messages. A I<port id> is unfortunately a very
unique string, which we are unable to know in advance. But L<AnyEvent::MP>
supports the concept of 'well known ports', which is basically a port on the
server side registered under a well known name. For example, the server has a
port for receiving chat messages with a unique I<port id> and registered it
under the name C<"chatter">.

As I<messages> can only be sent to a I<port id> and not just to a name we have
to ask the server I<node> what I<port id> has the well known port with the
name C<"chatter">

Another new term, what is a I<node>: The messaging network that can be created with
L<AnyEvent::MP> consists of I<nodes>. A I<node> handles all the connection and
low level message sending logic for it's application. The application in this
case is the server. Also every client has/is a I<node>.

=head2 Getting The Chatter Port

Ok, lots of talk, now some code. Now we will actually get the C<$server_port>
from the backend:

   ...

   use AnyEvent::MP;

   my $resolved_cv = AnyEvent->condvar;

   my $client_port = create_port;

   my $server_node = "localhost:1299#";

   snd $server_node, wkp => "chatter", "$client_port", "resolved";

   my $server_port;

   # setup a receiver callback for the 'resolved' message:
   $client_port->rcv (resolved => sub {
      my ($client_port, $type, $chatter_port_id) = @_;

      print "Resolved the server port 'chatter' to $chatter_port_id\n";
      $server_port = $chatter_port_id;

      $resolved_cv->send;
      1
   });

   # lets block the client until we resolved the server port.
   $resolved_cv->recv;

   # now setup another receiver callback for the chat messages:
   $client_port->rcv (message => sub {
      my ($client_port, $type, $msg) = @_;

      print "chat> $msg\n";
      0
   });

   # send the server a 'join' message:
   snd $server_port, join => "$client_port";

   sub send_message { ...

Now that was a lot of new stuff. In order to ask the server and receive an
answer we need to have a I<port> where we can receive the answer.
This is what C<create_port> will do for us, it just creates a new local
port and returns us an object (that will btw. stringify to the I<port id>),
that we can use to receive messages.

Next thing is the C<$server_node>. In order to refer to another node we need
some kind of string to reference it. The I<noderef> is basically a comma
seperated list of C<host:port> pairs. We assume in this tutorial that the
server runs on your localhost at port 1299, this gives us the noderef
C<"localhost:1299">.

Now you might ask what the C<"#"> at the end in C<$server_node> the above
example is about. Well, what I didn't tell you yet is that each I<node> has a
default I<port> to receive messages. The default port is the empty string
C<"">. The I<default port> of a I<node> also provides some special services for
us, for example resolving a well known port to a I<port id>.

Now to this line:

   snd $server_node, wkp => "chatter", "$client_port", "resolved";

We send a message with first element being C<'wkp'> (standing for 'well known
port'). Then the well known port name that we want to resolve to a I<port id>:
C<"chatter">.  And in order for the server node to be able to send us back the
resolved I<port id> we have to tell it where to send the result message: The
result message will have as it's first argument the string C<"resolved"> and
will be sent to C<"$client_port"> (the I<port id> of our own just created
port).

Next this is this:

   $client_port->rcv (resolved => sub {
      my ($client_port, $type, $chatter_port_id) = @_;
      ...
      1
   });

This sets up a receiver on our own port for the result message with the first
argument being the string C<"resolved">. Receivers can match the contents of
the messages before actually 'sending' it to the given callback.

B<Please note> that the given callback has to return either a true or a false
value for indicating whether it is B<done> (true value) or still wants to
B<continue> (false value) receiving messages.

In this case we tell the C<$client_port> to look into the received messages and
look for the string C<"resolved"> in the first element of the message. If it is
found, the given callback will be called with the C<$client_port> as first
argument, and the message as the remaining arguments.

We name the first element of the message C<$type> in this case. It's a common
idiom to code the 'type' of a message into it's first element, this allows for
simple matching.

The result message will contain the I<port id> of the well known port C<"chatter">
as next element, and will be put in C<$chatter_port_id>.

Next we just assign C<$server_port> and return a 1 (a true value)
from the callback. It indicates that we are done and don't want to receive
further C<'resolved'> messages with this callback.

Now we continue to the rest of the client by calling C<send> on
C<$resolved_cv>.

First new step after this is setting up the chat message receiver callback.

   $client_port->rcv (message => sub {
      my ($client_port, $type, $msg) = @_;

      print "chat> $msg\n";
      0
   });

We assume that all messages that are broadcasted to all clients by the server
will contain as first element the string C<"message"> and the actual message as
second element. The callback returns a false value this time, to indicate that
it wants to continue receiving messages.

Last but not least we actually tell the server to send us
the new chat messages from other clients. We do so by sending the
message type C<'join'> followed by our own I<port id>.

   # send the server a 'join' message:
   snd $server_port, join => "$client_port";

Then the server knows where to send all the new messages to.

=head2 The Finished Client

This is the complete client script:

   #!perl
   use AnyEvent;
   use AnyEvent::Handle;
   use AnyEvent::MP;

   my $resolved_cv = AnyEvent->condvar;

   my $client_port = create_port;

   my $server_node = "localhost:1299#";

   snd $server_node, wkp => "chatter", "$client_port", "resolved";

   my $server_port;

   # setup a receiver callback for the 'resolved' message:
   $client_port->rcv (resolved => sub {
      my ($client_port, $type, $chatter_port_id) = @_;

      print "Resolved the server port 'chatter' to $chatter_port_id\n";
      $server_port = $chatter_port_id;

      $resolved_cv->send;
      1
   });

   # lets block the client until we resolved the server port.
   $resolved_cv->recv;

   # now setup another receiver callback for the chat messages:
   $client_port->rcv (message => sub {
      my ($client_port, $type, $msg) = @_;

      print "chat> $msg\n";
      0
   });

   # send the server a 'join' message:
   snd $server_port, join => "$client_port";

   sub send_message {
      my ($msg) = @_;

      snd $server_port, message => $msg;
   }

   # make an AnyEvent condition variable for the 'quit' condition
   # (when we want to exit the client).
   my $quit_cv = AnyEvent->condvar;

   my $stdin_hdl = AnyEvent::Handle->new (
      fh      => \*STDIN,
      on_read => sub {
         my ($hdl) = @_;

         $hdl->push_read (line => sub {
            my ($hdl, $line) = @_;

            if ($line =~ /^\/quit/) { # /quit will end the client
               $quit_cv->send;

            } else {
               send_message ($line);
            }
         });
      }
   );

   $quit_cv->recv;

=head2 The Server

Ok, now finally to the server. What do we need? Well, we need to setup
the well known port C<"chatter"> where all clients send their messages to.

Up and into code right now:

   #!perl
   use AnyEvent;
   use AnyEvent::MP;

   become_public "localhost:1299";

   my $chatter_port = create_port;
   $chatter_port->register ("chatter");

   my %client_ports;

   $chatter_port->rcv (join => sub {
      my ($chatter_port, $type, $client_port) = @_;

      $client_ports{$client_port} = 1;
      0
   });

   $chatter_port->rcv (message => sub {
      my ($chatter_port, $type, $msg) = @_;

      snd $_, message => $msg for keys %client_ports;
      0
   });

   AnyEvent->condvar->recv;

This is all. Looks much easier, doesn't it? I'll explain it only shortly, as
we had the discussion of the C<rcv> method in the client part of this tutorial
above.

First this:

   become_public "localhost:1299";

This will tell our I<node> to become a I<public> node, which means that it can
be contacted via TCP. The first argument should be the I<noderef> the server
wants to be reachable at. In this case it's the TCP port 1299 on localhost.

Next we bascially setup two receivers, one for the C<join> messages and
another one for the actual messages of type C<messsage>.

In the C<join> message we get the client's port, which we just remember in the
C<%client_ports> hash. In the receiver for the message type C<message> we will
just iterate through all known C<%client_ports> and relay the message to them.

And thats it.

=head2 The Remaining Problems

The shown implementation still has some bugs. For instance: How does the
server know that the client isn't there anymore, and can cleanup the
C<%client_ports> hash? And also the chat messages have no originator,
so we don't know who actually sent the message (which would be quite useful
for human-to-human interaction: to know who the other one is :).

But aside from these issues I hope this tutorial got you the swing of
L<AnyEvent::MP> and explained some common idioms.

=head1 SEE ALSO

L<AnyEvent>

L<AnyEvent::Handle>

L<AnyEvent::MP>

=head1 AUTHOR

  Robin Redeker <elmex@ta-sa.org>
Revision:	1.3
Committed:	Mon Aug 3 08:40:29 2009 UTC (14 years, 9 months ago) by elmex
Branch:	MAIN
Changes since 1.2:	+2 -2 lines
Log Message:	refering to AnyEvent::Intro.
#	Content
1	=head1 Message Passing for the non-blocked Mind
2
3	=head2 Introduction and Terminology
4
5	This is a Tutorial about how to get the swing of the new L<AnyEvent::MP>
6	module. Which allows us to transparently pass messages to our own process and
7	to other process on other or the same host.
8
9	What kind of messages? Well, basically a message here means a list of Perl
10	strings, numbers, hashes and arrays, mostly everything that can be expressed as
11	L<JSON> string.
12
13	And next you might ask: between which entities are those messages "passed"?
14	Basically between C<nodes>, which are basically your applications (as in
15	processes) that use L<AnyEvent::MP> that run either on the same or different
16	hosts.
17
18	In this Tutorial I'll show you how to write a simple chat server based on
19	L<AnyEvent::MP>.
20
21	=head2 The Chat Client
22
23	Ok, lets start by implementing the "frontend" of the client. We will delay the
24	explanation and the code of the server until we finished the client, as the
25	most complex things actually happen in the client.
26
27	We will use L<AnyEvent::Handle> to do non-blocking IO read on standard input:
28
29	#!perl
30	use AnyEvent;
31	use AnyEvent::Handle;
32
33	sub send_message {
34	die "This is where we will send the messages to the server"
35	. "in the next step of this tutorial.\n"
36	}
37
38	# make an AnyEvent condition variable for the 'quit' condition
39	# (when we want to exit the client).
40	my $quit_cv = AnyEvent->condvar;
41
42	my $stdin_hdl = AnyEvent::Handle->new (
43	fh => \*STDIN,
44	on_read => sub {
45	my ($hdl) = @_;
46
47	$hdl->push_read (line => sub {
48	my ($hdl, $line) = @_;
49
50	if ($line =~ /^\/quit/) { # /quit will end the client
51	$quit_cv->send;
52
53	} else {
54	send_message ($line);
55	}
56	});
57	}
58	);
59
60	$quit_cv->recv;
61
62	This is now a very basic client. Explaining explicitly what L<AnyEvent::Handle>
63	does or what a I<condvar> is all about is out of scope of this document, please
64	consult L<AnyEvent::Intro> or the manpages for L<AnyEvent> and
65	L<AnyEvent::Handle>.
66
67	=head2 First Step Into Messaging
68
69	Now we take a look at L<AnyEvent::MP>. We need to know what to do in
70	C<send_message>. This is an example how it might look like:
71
72	... # the use lines from the above snippet
73
74	use AnyEvent::MP;
75
76	sub send_message {
77	my ($msg) = @_;
78
79	snd $server_port, message => $msg;
80	}
81
82	... # the rest of the above script
83
84	The C<snd> function is exported by L<AnyEvent::MP>, it stands for 'send a
85	message'. The first argument is the I<port> (a I<port> is something that can
86	receive messages) of the server which will receive the message . How we get it
87	will be explained in the next step. The next arguments of C<snd> are
88	C<'message'> and C<$msg> are the first two elements of the I<message> (a
89	I<message> in L<AnyEvent::MP> is a be a simple list of values, which can be
90	sent to I<ports>).
91
92	Thats all fine so far, but how do we get the C<$server_port>? Well, we will
93	need to get the unique I<port id> of the server's port where he wants to
94	receive all the incoming chat messages. A I<port id> is unfortunately a very
95	unique string, which we are unable to know in advance. But L<AnyEvent::MP>
96	supports the concept of 'well known ports', which is basically a port on the
97	server side registered under a well known name. For example, the server has a
98	port for receiving chat messages with a unique I<port id> and registered it
99	under the name C<"chatter">.
100
101	As I<messages> can only be sent to a I<port id> and not just to a name we have
102	to ask the server I<node> what I<port id> has the well known port with the
103	name C<"chatter">
104
105	Another new term, what is a I<node>: The messaging network that can be created with
106	L<AnyEvent::MP> consists of I<nodes>. A I<node> handles all the connection and
107	low level message sending logic for it's application. The application in this
108	case is the server. Also every client has/is a I<node>.
109
110	=head2 Getting The Chatter Port
111
112	Ok, lots of talk, now some code. Now we will actually get the C<$server_port>
113	from the backend:
114
115	...
116
117	use AnyEvent::MP;
118
119	my $resolved_cv = AnyEvent->condvar;
120
121	my $client_port = create_port;
122
123	my $server_node = "localhost:1299#";
124
125	snd $server_node, wkp => "chatter", "$client_port", "resolved";
126
127	my $server_port;
128
129	# setup a receiver callback for the 'resolved' message:
130	$client_port->rcv (resolved => sub {
131	my ($client_port, $type, $chatter_port_id) = @_;
132
133	print "Resolved the server port 'chatter' to $chatter_port_id\n";
134	$server_port = $chatter_port_id;
135
136	$resolved_cv->send;
137	1
138	});
139
140	# lets block the client until we resolved the server port.
141	$resolved_cv->recv;
142
143	# now setup another receiver callback for the chat messages:
144	$client_port->rcv (message => sub {
145	my ($client_port, $type, $msg) = @_;
146
147	print "chat> $msg\n";
148	0
149	});
150
151	# send the server a 'join' message:
152	snd $server_port, join => "$client_port";
153
154	sub send_message { ...
155
156	Now that was a lot of new stuff. In order to ask the server and receive an
157	answer we need to have a I<port> where we can receive the answer.
158	This is what C<create_port> will do for us, it just creates a new local
159	port and returns us an object (that will btw. stringify to the I<port id>),
160	that we can use to receive messages.
161
162	Next thing is the C<$server_node>. In order to refer to another node we need
163	some kind of string to reference it. The I<noderef> is basically a comma
164	seperated list of C<host:port> pairs. We assume in this tutorial that the
165	server runs on your localhost at port 1299, this gives us the noderef
166	C<"localhost:1299">.
167
168	Now you might ask what the C<"#"> at the end in C<$server_node> the above
169	example is about. Well, what I didn't tell you yet is that each I<node> has a
170	default I<port> to receive messages. The default port is the empty string
171	C<"">. The I<default port> of a I<node> also provides some special services for
172	us, for example resolving a well known port to a I<port id>.
173
174	Now to this line:
175
176	snd $server_node, wkp => "chatter", "$client_port", "resolved";
177
178	We send a message with first element being C<'wkp'> (standing for 'well known
179	port'). Then the well known port name that we want to resolve to a I<port id>:
180	C<"chatter">. And in order for the server node to be able to send us back the
181	resolved I<port id> we have to tell it where to send the result message: The
182	result message will have as it's first argument the string C<"resolved"> and
183	will be sent to C<"$client_port"> (the I<port id> of our own just created
184	port).
185
186	Next this is this:
187
188	$client_port->rcv (resolved => sub {
189	my ($client_port, $type, $chatter_port_id) = @_;
190	...
191	1
192	});
193
194	This sets up a receiver on our own port for the result message with the first
195	argument being the string C<"resolved">. Receivers can match the contents of
196	the messages before actually 'sending' it to the given callback.
197
198	B<Please note> that the given callback has to return either a true or a false
199	value for indicating whether it is B<done> (true value) or still wants to
200	B<continue> (false value) receiving messages.
201
202	In this case we tell the C<$client_port> to look into the received messages and
203	look for the string C<"resolved"> in the first element of the message. If it is
204	found, the given callback will be called with the C<$client_port> as first
205	argument, and the message as the remaining arguments.
206
207	We name the first element of the message C<$type> in this case. It's a common
208	idiom to code the 'type' of a message into it's first element, this allows for
209	simple matching.
210
211	The result message will contain the I<port id> of the well known port C<"chatter">
212	as next element, and will be put in C<$chatter_port_id>.
213
214	Next we just assign C<$server_port> and return a 1 (a true value)
215	from the callback. It indicates that we are done and don't want to receive
216	further C<'resolved'> messages with this callback.
217
218	Now we continue to the rest of the client by calling C<send> on
219	C<$resolved_cv>.
220
221	First new step after this is setting up the chat message receiver callback.
222
223	$client_port->rcv (message => sub {
224	my ($client_port, $type, $msg) = @_;
225
226	print "chat> $msg\n";
227	0
228	});
229
230	We assume that all messages that are broadcasted to all clients by the server
231	will contain as first element the string C<"message"> and the actual message as
232	second element. The callback returns a false value this time, to indicate that
233	it wants to continue receiving messages.
234
235	Last but not least we actually tell the server to send us
236	the new chat messages from other clients. We do so by sending the
237	message type C<'join'> followed by our own I<port id>.
238
239	# send the server a 'join' message:
240	snd $server_port, join => "$client_port";
241
242	Then the server knows where to send all the new messages to.
243
244	=head2 The Finished Client
245
246	This is the complete client script:
247
248	#!perl
249	use AnyEvent;
250	use AnyEvent::Handle;
251	use AnyEvent::MP;
252
253	my $resolved_cv = AnyEvent->condvar;
254
255	my $client_port = create_port;
256
257	my $server_node = "localhost:1299#";
258
259	snd $server_node, wkp => "chatter", "$client_port", "resolved";
260
261	my $server_port;
262
263	# setup a receiver callback for the 'resolved' message:
264	$client_port->rcv (resolved => sub {
265	my ($client_port, $type, $chatter_port_id) = @_;
266
267	print "Resolved the server port 'chatter' to $chatter_port_id\n";
268	$server_port = $chatter_port_id;
269
270	$resolved_cv->send;
271	1
272	});
273
274	# lets block the client until we resolved the server port.
275	$resolved_cv->recv;
276
277	# now setup another receiver callback for the chat messages:
278	$client_port->rcv (message => sub {
279	my ($client_port, $type, $msg) = @_;
280
281	print "chat> $msg\n";
282	0
283	});
284
285	# send the server a 'join' message:
286	snd $server_port, join => "$client_port";
287
288	sub send_message {
289	my ($msg) = @_;
290
291	snd $server_port, message => $msg;
292	}
293
294	# make an AnyEvent condition variable for the 'quit' condition
295	# (when we want to exit the client).
296	my $quit_cv = AnyEvent->condvar;
297
298	my $stdin_hdl = AnyEvent::Handle->new (
299	fh => \*STDIN,
300	on_read => sub {
301	my ($hdl) = @_;
302
303	$hdl->push_read (line => sub {
304	my ($hdl, $line) = @_;
305
306	if ($line =~ /^\/quit/) { # /quit will end the client
307	$quit_cv->send;
308
309	} else {
310	send_message ($line);
311	}
312	});
313	}
314	);
315
316	$quit_cv->recv;
317
318	=head2 The Server
319
320	Ok, now finally to the server. What do we need? Well, we need to setup
321	the well known port C<"chatter"> where all clients send their messages to.
322
323	Up and into code right now:
324
325	#!perl
326	use AnyEvent;
327	use AnyEvent::MP;
328
329	become_public "localhost:1299";
330
331	my $chatter_port = create_port;
332	$chatter_port->register ("chatter");
333
334	my %client_ports;
335
336	$chatter_port->rcv (join => sub {
337	my ($chatter_port, $type, $client_port) = @_;
338
339	$client_ports{$client_port} = 1;
340	0
341	});
342
343	$chatter_port->rcv (message => sub {
344	my ($chatter_port, $type, $msg) = @_;
345
346	snd $_, message => $msg for keys %client_ports;
347	0
348	});
349
350	AnyEvent->condvar->recv;
351
352	This is all. Looks much easier, doesn't it? I'll explain it only shortly, as
353	we had the discussion of the C<rcv> method in the client part of this tutorial
354	above.
355
356	First this:
357
358	become_public "localhost:1299";
359
360	This will tell our I<node> to become a I<public> node, which means that it can
361	be contacted via TCP. The first argument should be the I<noderef> the server
362	wants to be reachable at. In this case it's the TCP port 1299 on localhost.
363
364	Next we bascially setup two receivers, one for the C<join> messages and
365	another one for the actual messages of type C<messsage>.
366
367	In the C<join> message we get the client's port, which we just remember in the
368	C<%client_ports> hash. In the receiver for the message type C<message> we will
369	just iterate through all known C<%client_ports> and relay the message to them.
370
371	And thats it.
372
373	=head2 The Remaining Problems
374
375	The shown implementation still has some bugs. For instance: How does the
376	server know that the client isn't there anymore, and can cleanup the
377	C<%client_ports> hash? And also the chat messages have no originator,
378	so we don't know who actually sent the message (which would be quite useful
379	for human-to-human interaction: to know who the other one is :).
380
381	But aside from these issues I hope this tutorial got you the swing of
382	L<AnyEvent::MP> and explained some common idioms.
383
384	=head1 SEE ALSO
385
386	L<AnyEvent>
387
388	L<AnyEvent::Handle>
389
390	L<AnyEvent::MP>
391
392	=head1 AUTHOR
393
394	Robin Redeker <elmex@ta-sa.org>