cvsroot/Coro-MP/MP.pm

=head1 NAME

Coro::MP - erlang-style multi-processing/message-passing framework

=head1 SYNOPSIS

   use Coro::MP;

   $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
   configure;

   # ports are message destinations

   # sending messages
   snd $port, type => data...;
   snd $port, @msg;
   snd @msg_with_first_element_being_a_port;

   # creating/using ports
   my $port = port_async {
      # thread context, $SELF is set to $port

      # returning will "kil" the $port with an empty reason
   };

   # simple receive
   my $port = port {
      my (undef, @data) = get "tag";
   };
   snd $port, tag => "data0", "data1";

   # create a port on another node
   my $port = spawn $node, $initfunc, @initdata;

   # monitoring
   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 DESCRIPTION

This module (-family) implements a simple message passing framework.

Despite its simplicity, you can securely message other processes running
on the same or other hosts, and you can supervise entities remotely.

This module depends heavily on L<AnyEvent::MP>, in fact, many functions
exported by this module are identical to AnyEvent::MP functions. This
module family is simply the Coro API to AnyEvent::MP.

Care has been taken to stay compatible with AnyEvent::MP, even if
sometimes this required a less natural API (C<spawn> should indeed spawn a
thread, not just call an initfunc for example).

For an introduction to AnyEvent::MP, see the L<AnyEvent::MP::Intro> manual
page.

=head1 VARIABLES/FUNCTIONS

=over 4

=cut

package Coro::MP;

use common::sense;

use Carp ();

use AnyEvent::MP::Kernel;
use AnyEvent::MP;
use Coro;
use Coro::AnyEvent ();

use AE ();

use base "Exporter";

our $VERSION = "0.1";

our @EXPORT = qw(
   NODE $NODE *SELF node_of
   configure
   port snd rcv mon kil psub cal spawn

   port_async rcv_async get get_cond syncal pasync
);

sub _new_coro {
   my ($port, $threadcb) = @_;

   my $coro = async_pool {
      eval { $threadcb->() };
      kil $SELF, die => $@ if $@;
   };
   $coro->swap_sv (\$SELF, \$port);

   # killing the port cancels the coro
   mon $port, sub {
      # canceling ourselves is a bit evil...
      # however, this should really not be called in coro context...
      if ($coro == $Coro::current) {
         my @reason = @_;
         async_pool { $coro->cancel (@reason) };
      } else {
         $coro->cancel (@_);
      }
   };

   # cancelling the coro kills the port
   $coro->on_destroy (sub { kil $port, @_ });

   $coro
}

=item NODE, $NODE, node_of, configure

=item $SELF, *SELF, SELF, %SELF, @SELF...

=item snd, mon, kil

These variables and functions work exactly as in AnyEvent::MP, in fact,
they are exactly the same functions, and are used in much the same way.

=item rcv

This function works exactly as C<AnyEvent::MP::rcv>, and is in fact
compatible with Coro::MP ports. However, the canonical way to receive
messages with Coro::MP is to use C<get> or C<get_cond>.

=item port

This function is exactly the same as C<AnyEvent::MP::port> and creates new
ports. You can attach a thread to them by calling C<rcv_async> or you can
do a create and attach in one operation using C<port_async>.

=item psub

This function works exactly as C<AnyEvent::MP::psub> - you could use it to
run callbacks within a port context (good for monitoring), but you cannot
C<get> messages unless the callback executes within the thread attached to
the port.

Since creating a thread with port context requires somewhta annoying
syntax, there is a C<pasync> function that handles that for you - note
that within such a thread, you still cannot C<get> messages.

=item spawn

This function is identical to C<AnyEvent::MP::spawn>. This means that
it doesn't spawn a new thread as one would expect, but simply calls an
init function. The init function, however, can attach a new thread easily:

   sub initfun {
      my (@args) = @_;

      rcv_async $SELF, sub {
         # thread-code
      };
   }

=item cal

This function is identical to C<AnyEvent::MP::cal>. The easiest way to
make a synchronous call is to use Coro's rouse functionality:

   # send 1, 2, 3 to $port and wait up to 30s for reply
   cal $port, 1, 2, 3, rouse_cb, 30;
   my @reply = rouse_wait;

You can also use C<syncal> if you want, and are ok with learning yet
another function with a weird name:

   my @reply = syncal 30, $port, 1, 2, 3;

=item $local_port = port_async { ... }

Creates a new local port, and returns its ID. A new thread is created and
attached to the port (see C<rcv_async>, below, for details).

=cut

sub rcv_async($$);

sub port_async(;&) {
   my $id = "$UNIQ." . $ID++;
   my $port = "$NODE#$id";

   @_
      ? rcv_async $port, shift
      : AnyEvent::MP::rcv $port, undef;

   $port
}

=item rcv_async $port, $threadcb

This function creates and attaches a thread on a port. The thread is set
to execute C<$threadcb> and is put into the ready queue. The thread will
receive all messages not filtered away by tagged receive callbacks (as set
by C<AnyEvent::MP::rcv>) - it simply replaces the default callback of an
AnyEvent::MP port.

The special variable C<$SELF> will be set to C<$port> during thread
execution.

When C<$threadcb> returns or the thread is canceled, the return/cancel
values become the C<kil> reason.

It is not allowed to call C<rcv_async> more than once on a given port.

=cut

sub rcv_async($$) {
   my ($port, $threadcb) = @_;

   my (@queue, $coro);

   AnyEvent::MP::rcv $port, sub {
      push @queue, \@_; # TODO, take copy?
      $coro->ready; # TODO, maybe too many unwanted wake-ups?
   };

   $coro = _new_coro $port, $threadcb;
   $coro->{_coro_mp_queue} = \@queue;
}

=item @msg = get $tag

=item @msg = get $tag, $timeout

Find, dequeue and return the next message with the specified C<$tag>. If
no matching message is currently queued, wait up to C<$timeout> seconds
(or forever if no C<$timeout> has been specified or it is C<undef>) for
one to arrive.

Returns the message with the initial tag removed. In case of a timeout,
the empty list. The function I<must> be called in list context.

Note that empty messages cannot be distinguished from a timeout when using
C<rcv>.

=cut

sub get($;$) {
   my ($tag, $timeout) = @_;

   my $queue = $Coro::current->{_coro_mp_queue}
      or Carp::croak "Coro::MP::get called from thread not attached to any port";

   my $i;

   while () {
      $queue->[$_][0] eq $tag
         and return @{ splice @$queue, $_, 1 }
         for $i..$#$queue;

      $i = @$queue;

      # wait for more messages
      if (ref $timeout) {
         schedule;
         defined $i or return; # timeout

      } elsif (defined $timeout) {
         $timeout or return;

         my $current = $Coro::current;
         $timeout = AE::timer $timeout, 0, sub {
            undef $i;
            $current->ready;
         };
      } else {
         $timeout = \$i; # dummy
      }
   }
}

=item @msg = get_cond { condition... } [$timeout]

Similarly to C<get>, looks for a matching message. Unlike C<get>,
"matching" is not defined by a tag alone, but by a predicate, a piece of
code that is executed on each candidate message in turn, with C<@_> set to
the message contents.

The predicate code is supposed to return the empty list if the message
didn't match. If it returns anything else, then the message is removed
from the queue and returned to the caller.

In addition, if the predicate returns a code reference, then it is
immediately called invoked on the removed message.

If a C<$timeout> is specified and is not C<undef>, then, after this many
seconds have been passed without a matching message arriving, the empty
list will be returned.

TODO

=cut

sub get_cond(&;$) {
   my ($cond, $timeout) = @_;

   my $queue = $Coro::current->{_coro_mp_queue}
      or Carp::croak "Coro::MP::get_cond called from thread not attached to any port";

   my ($i, $ok);

   while () {
      do
         {
            local *_ = $queue->[$_];
            if ($ok = &$cond) {
               splice @$queue, $_, 1;
               &$ok if "CODE" eq ref $ok;
               return @_;
            }
         }
      for $i..$#$queue;

      $i = @$queue;

      # wait for more messages
      if (ref $timeout) {
         schedule;
         defined $i or return; # timeout

      } elsif (defined $timeout) {
         $timeout or return;

         my $current = $Coro::current;
         $timeout = AE::timer $timeout, 0, sub {
            undef $i;
            $current->ready;
         };
      } else {
         $timeout = \$i; # dummy
      }
   }
}

=item $async = pasync { BLOCK }

Sometimes you want to run a thread within a port context, for error
handling.

This function creates a new, ready, thread (using C<Coro::async>), sets
C<$SELF> to the the current value of C<$SELF> while it executing, and
calls the given BLOCK.

This is very similar to C<psub> - note that while the BLOCK exeuctes in
C<$SELF> port context, you cannot call C<get>, as C<$SELF> can only be
attached to one thread.

=cut

sub pasync(&) {
   warn "new coro with $SELF\n";#d#
   _new_coro $SELF, $_[0]
}

=item @reply = syncal $port, @msg, $callback[, $timeout]

The synchronous form of C<cal>, a simple form of RPC - it 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 C<kil>ed when no longer needed.

Then it will wait until a reply message arrives, which will be returned to
the caller.

If the C<$timeout> is defined, then after this many seconds, when no
message has arrived, the port will be C<kil>ed and an empty list will be
returned.

If the C<$timeout> is undef, then the local port will monitor the remote
port instead, so it eventually gets cleaned-up.

=cut

sub syncal($@) {
   my ($timeout, @msg) = @_;

   cal @msg, Coro::rouse_cb, $timeout;
   Coro::rouse_wait
}

=back

=head1 SEE ALSO

L<AnyEvent::MP::Intro> - a gentle introduction.

L<AnyEvent::MP> - like Coro::MP, but event-based.

L<AnyEvent>.

=head1 AUTHOR

 Marc Lehmann <schmorp@schmorp.de>
 http://home.schmorp.de/

=cut

1

Revision:	1.4
Committed:	Fri Oct 2 20:36:35 2009 UTC (16 years, 11 months ago) by root
Branch:	MAIN
Changes since 1.3:	+38 -26 lines
Log Message:	* empty log message *
#	User	Rev	Content
1	root	1.1	=head1 NAME
2
3			Coro::MP - erlang-style multi-processing/message-passing framework
4
5			=head1 SYNOPSIS
6
7			use Coro::MP;
8
9			$NODE # contains this node's node ID
10			NODE # returns this node's node ID
11
12			$SELF # receiving/own port id in rcv callbacks
13
14			# initialise the node so it can send/receive messages
15			configure;
16
17			# ports are message destinations
18
19			# sending messages
20			snd $port, type => data...;
21			snd $port, @msg;
22			snd @msg_with_first_element_being_a_port;
23
24			# creating/using ports
25	root	1.2	my $port = port_async {
26	root	1.1	# thread context, $SELF is set to $port
27
28			# returning will "kil" the $port with an empty reason
29			};
30
31			# simple receive
32			my $port = port {
33	root	1.2	my (undef, @data) = get "tag";
34	root	1.1	};
35			snd $port, tag => "data0", "data1";
36
37			# create a port on another node
38			my $port = spawn $node, $initfunc, @initdata;
39
40			# monitoring
41			mon $localport, $cb->(@msg) # callback is invoked on death
42			mon $localport, $otherport # kill otherport on abnormal death
43			mon $localport, $otherport, @msg # send message on death
44
45			=head1 DESCRIPTION
46
47			This module (-family) implements a simple message passing framework.
48
49			Despite its simplicity, you can securely message other processes running
50			on the same or other hosts, and you can supervise entities remotely.
51
52			This module depends heavily on L<AnyEvent::MP>, in fact, many functions
53			exported by this module are identical to AnyEvent::MP functions. This
54			module family is simply the Coro API to AnyEvent::MP.
55
56			Care has been taken to stay compatible with AnyEvent::MP, even if
57			sometimes this required a less natural API (C<spawn> should indeed spawn a
58			thread, not just call an initfunc for example).
59
60			For an introduction to AnyEvent::MP, see the L<AnyEvent::MP::Intro> manual
61			page.
62
63			=head1 VARIABLES/FUNCTIONS
64
65			=over 4
66
67			=cut
68
69			package Coro::MP;
70
71			use common::sense;
72
73			use Carp ();
74
75			use AnyEvent::MP::Kernel;
76	root	1.4	use AnyEvent::MP;
77	root	1.1	use Coro;
78			use Coro::AnyEvent ();
79
80			use AE ();
81
82			use base "Exporter";
83
84			our $VERSION = "0.1";
85
86			our @EXPORT = qw(
87			NODE $NODE *SELF node_of
88			configure
89	root	1.3	port snd rcv mon kil psub cal spawn
90	root	1.4
91	root	1.3	port_async rcv_async get get_cond syncal pasync
92	root	1.1	);
93
94	root	1.4	sub _new_coro {
95			my ($port, $threadcb) = @_;
96
97			my $coro = async_pool {
98			eval { $threadcb->() };
99			kil $SELF, die => $@ if $@;
100			};
101			$coro->swap_sv (\$SELF, \$port);
102
103			# killing the port cancels the coro
104			mon $port, sub {
105			# canceling ourselves is a bit evil...
106			# however, this should really not be called in coro context...
107			if ($coro == $Coro::current) {
108			my @reason = @_;
109			async_pool { $coro->cancel (@reason) };
110			} else {
111			$coro->cancel (@_);
112			}
113			};
114
115			# cancelling the coro kills the port
116			$coro->on_destroy (sub { kil $port, @_ });
117	root	1.1
118	root	1.4	$coro
119	root	1.1	}
120
121			=item NODE, $NODE, node_of, configure
122
123			=item $SELF, *SELF, SELF, %SELF, @SELF...
124
125	root	1.3	=item snd, mon, kil
126	root	1.1
127			These variables and functions work exactly as in AnyEvent::MP, in fact,
128	root	1.3	they are exactly the same functions, and are used in much the same way.
129
130			=item rcv
131
132			This function works exactly as C<AnyEvent::MP::rcv>, and is in fact
133			compatible with Coro::MP ports. However, the canonical way to receive
134			messages with Coro::MP is to use C<get> or C<get_cond>.
135
136			=item port
137
138			This function is exactly the same as C<AnyEvent::MP::port> and creates new
139			ports. You can attach a thread to them by calling C<rcv_async> or you can
140			do a create and attach in one operation using C<port_async>.
141
142			=item psub
143
144			This function works exactly as C<AnyEvent::MP::psub> - you could use it to
145			run callbacks within a port context (good for monitoring), but you cannot
146			C<get> messages unless the callback executes within the thread attached to
147			the port.
148
149			Since creating a thread with port context requires somewhta annoying
150			syntax, there is a C<pasync> function that handles that for you - note
151			that within such a thread, you still cannot C<get> messages.
152	root	1.1
153			=item spawn
154
155	root	1.3	This function is identical to C<AnyEvent::MP::spawn>. This means that
156	root	1.1	it doesn't spawn a new thread as one would expect, but simply calls an
157			init function. The init function, however, can attach a new thread easily:
158
159			sub initfun {
160			my (@args) = @_;
161
162	root	1.3	rcv_async $SELF, sub {
163	root	1.1	# thread-code
164			};
165			}
166
167	root	1.3	=item cal
168
169			This function is identical to C<AnyEvent::MP::cal>. The easiest way to
170			make a synchronous call is to use Coro's rouse functionality:
171
172			# send 1, 2, 3 to $port and wait up to 30s for reply
173			cal $port, 1, 2, 3, rouse_cb, 30;
174			my @reply = rouse_wait;
175
176			You can also use C<syncal> if you want, and are ok with learning yet
177	root	1.4	another function with a weird name:
178	root	1.3
179			my @reply = syncal 30, $port, 1, 2, 3;
180
181	root	1.1	=item $local_port = port_async { ... }
182
183			Creates a new local port, and returns its ID. A new thread is created and
184			attached to the port (see C<rcv_async>, below, for details).
185
186			=cut
187
188			sub rcv_async($$);
189
190			sub port_async(;&) {
191			my $id = "$UNIQ." . $ID++;
192			my $port = "$NODE#$id";
193
194			@_
195			? rcv_async $port, shift
196			: AnyEvent::MP::rcv $port, undef;
197
198			$port
199			}
200
201			=item rcv_async $port, $threadcb
202
203			This function creates and attaches a thread on a port. The thread is set
204			to execute C<$threadcb> and is put into the ready queue. The thread will
205			receive all messages not filtered away by tagged receive callbacks (as set
206			by C<AnyEvent::MP::rcv>) - it simply replaces the default callback of an
207			AnyEvent::MP port.
208
209			The special variable C<$SELF> will be set to C<$port> during thread
210			execution.
211
212			When C<$threadcb> returns or the thread is canceled, the return/cancel
213			values become the C<kil> reason.
214
215			It is not allowed to call C<rcv_async> more than once on a given port.
216
217			=cut
218
219			sub rcv_async($$) {
220			my ($port, $threadcb) = @_;
221
222			my (@queue, $coro);
223
224			AnyEvent::MP::rcv $port, sub {
225			push @queue, \@_; # TODO, take copy?
226			$coro->ready; # TODO, maybe too many unwanted wake-ups?
227			};
228
229	root	1.4	$coro = _new_coro $port, $threadcb;
230	root	1.1	$coro->{_coro_mp_queue} = \@queue;
231			}
232
233			=item @msg = get $tag
234
235			=item @msg = get $tag, $timeout
236
237			Find, dequeue and return the next message with the specified C<$tag>. If
238			no matching message is currently queued, wait up to C<$timeout> seconds
239			(or forever if no C<$timeout> has been specified or it is C<undef>) for
240			one to arrive.
241
242			Returns the message with the initial tag removed. In case of a timeout,
243			the empty list. The function I<must> be called in list context.
244
245			Note that empty messages cannot be distinguished from a timeout when using
246			C<rcv>.
247
248			=cut
249
250			sub get($;$) {
251			my ($tag, $timeout) = @_;
252
253			my $queue = $Coro::current->{_coro_mp_queue}
254			or Carp::croak "Coro::MP::get called from thread not attached to any port";
255
256			my $i;
257
258			while () {
259			$queue->[$_][0] eq $tag
260			and return @{ splice @$queue, $_, 1 }
261			for $i..$#$queue;
262
263			$i = @$queue;
264
265			# wait for more messages
266			if (ref $timeout) {
267			schedule;
268			defined $i or return; # timeout
269
270			} elsif (defined $timeout) {
271			$timeout or return;
272
273			my $current = $Coro::current;
274			$timeout = AE::timer $timeout, 0, sub {
275			undef $i;
276			$current->ready;
277			};
278			} else {
279			$timeout = \$i; # dummy
280			}
281			}
282			}
283
284	root	1.3	=item @msg = get_cond { condition... } [$timeout]
285
286			Similarly to C<get>, looks for a matching message. Unlike C<get>,
287			"matching" is not defined by a tag alone, but by a predicate, a piece of
288	root	1.4	code that is executed on each candidate message in turn, with C<@_> set to
289			the message contents.
290	root	1.3
291	root	1.4	The predicate code is supposed to return the empty list if the message
292			didn't match. If it returns anything else, then the message is removed
293			from the queue and returned to the caller.
294	root	1.3
295	root	1.4	In addition, if the predicate returns a code reference, then it is
296			immediately called invoked on the removed message.
297	root	1.3
298			If a C<$timeout> is specified and is not C<undef>, then, after this many
299			seconds have been passed without a matching message arriving, the empty
300			list will be returned.
301
302			TODO
303
304			=cut
305
306	root	1.4	sub get_cond(&;$) {
307	root	1.3	my ($cond, $timeout) = @_;
308
309			my $queue = $Coro::current->{_coro_mp_queue}
310			or Carp::croak "Coro::MP::get_cond called from thread not attached to any port";
311
312			my ($i, $ok);
313
314			while () {
315			do
316			{
317			local *_ = $queue->[$_];
318			if ($ok = &$cond) {
319			splice @$queue, $_, 1;
320			&$ok if "CODE" eq ref $ok;
321			return @_;
322			}
323			}
324			for $i..$#$queue;
325
326			$i = @$queue;
327
328			# wait for more messages
329			if (ref $timeout) {
330			schedule;
331			defined $i or return; # timeout
332
333			} elsif (defined $timeout) {
334			$timeout or return;
335
336			my $current = $Coro::current;
337			$timeout = AE::timer $timeout, 0, sub {
338			undef $i;
339			$current->ready;
340			};
341			} else {
342			$timeout = \$i; # dummy
343			}
344			}
345			}
346
347			=item $async = pasync { BLOCK }
348
349			Sometimes you want to run a thread within a port context, for error
350			handling.
351
352			This function creates a new, ready, thread (using C<Coro::async>), sets
353			C<$SELF> to the the current value of C<$SELF> while it executing, and
354			calls the given BLOCK.
355
356			This is very similar to C<psub> - note that while the BLOCK exeuctes in
357			C<$SELF> port context, you cannot call C<get>, as C<$SELF> can only be
358			attached to one thread.
359
360			=cut
361
362			sub pasync(&) {
363	root	1.4	warn "new coro with $SELF\n";#d#
364			_new_coro $SELF, $_[0]
365	root	1.1	}
366
367	root	1.3	=item @reply = syncal $port, @msg, $callback[, $timeout]
368	root	1.1
369	root	1.3	The synchronous form of C<cal>, a simple form of RPC - it sends a message
370			to the given C<$port> with the given contents (C<@msg>), but adds a reply
371			port to the message.
372	root	1.1
373			The reply port is created temporarily just for the purpose of receiving
374			the reply, and will be C<kil>ed when no longer needed.
375
376	root	1.3	Then it will wait until a reply message arrives, which will be returned to
377			the caller.
378	root	1.1
379	root	1.3	If the C<$timeout> is defined, then after this many seconds, when no
380			message has arrived, the port will be C<kil>ed and an empty list will be
381			returned.
382	root	1.1
383	root	1.3	If the C<$timeout> is undef, then the local port will monitor the remote
384			port instead, so it eventually gets cleaned-up.
385	root	1.1
386	root	1.3	=cut
387	root	1.1
388	root	1.3	sub syncal($@) {
389			my ($timeout, @msg) = @_;
390	root	1.1
391	root	1.3	cal @msg, Coro::rouse_cb, $timeout;
392			Coro::rouse_wait
393	root	1.1	}
394
395			=back
396
397			=head1 SEE ALSO
398
399			L<AnyEvent::MP::Intro> - a gentle introduction.
400
401			L<AnyEvent::MP> - like Coro::MP, but event-based.
402
403			L<AnyEvent>.
404
405			=head1 AUTHOR
406
407			Marc Lehmann <schmorp@schmorp.de>
408			http://home.schmorp.de/
409
410			=cut
411
412			1
413