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 qw(snd rcv mon kil psub cal spawn); # TODO, should use AnyEvent::MP::Kernel only
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
);

our $SELF;

sub _self_die() {
   my $msg = $@;
   $msg =~ s/\n+$// unless ref $msg;
   kil $SELF, die => $msg;
}

=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:

   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 = async {
      # this SELF swapping is horrible
      my $old_self;
      Coro::on_enter { $old_self = $SELF; $SELF = $port };
      Coro::on_leave { $SELF = $old_self };

      $threadcb->();
   };
   $coro->{_coro_mp_queue} = \@queue;

   mon $port, sub { $coro->cancel (@_) };
   $coro->on_destroy (sub { kil $port, @_ });
}

=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.

Also unlike C<get>, 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.

AS an exception, if the predicate returns a code reference, then it is
immediately 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(&) {
   die "nyi";
}

=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.3
Committed:	Fri Oct 2 14:32:10 2009 UTC (16 years, 11 months ago) by root
Branch:	MAIN
Changes since 1.2:	+137 -22 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.3	use AnyEvent::MP qw(snd rcv mon kil psub cal spawn); # TODO, should use AnyEvent::MP::Kernel only
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			port_async rcv_async get get_cond syncal pasync
91	root	1.1	);
92
93			our $SELF;
94
95			sub _self_die() {
96			my $msg = $@;
97			$msg =~ s/\n+$// unless ref $msg;
98			kil $SELF, die => $msg;
99			}
100
101			=item NODE, $NODE, node_of, configure
102
103			=item $SELF, *SELF, SELF, %SELF, @SELF...
104
105	root	1.3	=item snd, mon, kil
106	root	1.1
107			These variables and functions work exactly as in AnyEvent::MP, in fact,
108	root	1.3	they are exactly the same functions, and are used in much the same way.
109
110			=item rcv
111
112			This function works exactly as C<AnyEvent::MP::rcv>, and is in fact
113			compatible with Coro::MP ports. However, the canonical way to receive
114			messages with Coro::MP is to use C<get> or C<get_cond>.
115
116			=item port
117
118			This function is exactly the same as C<AnyEvent::MP::port> and creates new
119			ports. You can attach a thread to them by calling C<rcv_async> or you can
120			do a create and attach in one operation using C<port_async>.
121
122			=item psub
123
124			This function works exactly as C<AnyEvent::MP::psub> - you could use it to
125			run callbacks within a port context (good for monitoring), but you cannot
126			C<get> messages unless the callback executes within the thread attached to
127			the port.
128
129			Since creating a thread with port context requires somewhta annoying
130			syntax, there is a C<pasync> function that handles that for you - note
131			that within such a thread, you still cannot C<get> messages.
132	root	1.1
133			=item spawn
134
135	root	1.3	This function is identical to C<AnyEvent::MP::spawn>. This means that
136	root	1.1	it doesn't spawn a new thread as one would expect, but simply calls an
137			init function. The init function, however, can attach a new thread easily:
138
139			sub initfun {
140			my (@args) = @_;
141
142	root	1.3	rcv_async $SELF, sub {
143	root	1.1	# thread-code
144			};
145			}
146
147	root	1.3	=item cal
148
149			This function is identical to C<AnyEvent::MP::cal>. The easiest way to
150			make a synchronous call is to use Coro's rouse functionality:
151
152			# send 1, 2, 3 to $port and wait up to 30s for reply
153			cal $port, 1, 2, 3, rouse_cb, 30;
154			my @reply = rouse_wait;
155
156			You can also use C<syncal> if you want, and are ok with learning yet
157			another function:
158
159			my @reply = syncal 30, $port, 1, 2, 3;
160
161	root	1.1	=item $local_port = port_async { ... }
162
163			Creates a new local port, and returns its ID. A new thread is created and
164			attached to the port (see C<rcv_async>, below, for details).
165
166			=cut
167
168			sub rcv_async($$);
169
170			sub port_async(;&) {
171			my $id = "$UNIQ." . $ID++;
172			my $port = "$NODE#$id";
173
174			@_
175			? rcv_async $port, shift
176			: AnyEvent::MP::rcv $port, undef;
177
178			$port
179			}
180
181			=item rcv_async $port, $threadcb
182
183			This function creates and attaches a thread on a port. The thread is set
184			to execute C<$threadcb> and is put into the ready queue. The thread will
185			receive all messages not filtered away by tagged receive callbacks (as set
186			by C<AnyEvent::MP::rcv>) - it simply replaces the default callback of an
187			AnyEvent::MP port.
188
189			The special variable C<$SELF> will be set to C<$port> during thread
190			execution.
191
192			When C<$threadcb> returns or the thread is canceled, the return/cancel
193			values become the C<kil> reason.
194
195			It is not allowed to call C<rcv_async> more than once on a given port.
196
197			=cut
198
199			sub rcv_async($$) {
200			my ($port, $threadcb) = @_;
201
202			my (@queue, $coro);
203
204			AnyEvent::MP::rcv $port, sub {
205			push @queue, \@_; # TODO, take copy?
206			$coro->ready; # TODO, maybe too many unwanted wake-ups?
207			};
208
209			$coro = async {
210			# this SELF swapping is horrible
211			my $old_self;
212			Coro::on_enter { $old_self = $SELF; $SELF = $port };
213			Coro::on_leave { $SELF = $old_self };
214
215			$threadcb->();
216			};
217			$coro->{_coro_mp_queue} = \@queue;
218
219			mon $port, sub { $coro->cancel (@_) };
220			$coro->on_destroy (sub { kil $port, @_ });
221			}
222
223			=item @msg = get $tag
224
225			=item @msg = get $tag, $timeout
226
227			Find, dequeue and return the next message with the specified C<$tag>. If
228			no matching message is currently queued, wait up to C<$timeout> seconds
229			(or forever if no C<$timeout> has been specified or it is C<undef>) for
230			one to arrive.
231
232			Returns the message with the initial tag removed. In case of a timeout,
233			the empty list. The function I<must> be called in list context.
234
235			Note that empty messages cannot be distinguished from a timeout when using
236			C<rcv>.
237
238			=cut
239
240			sub get($;$) {
241			my ($tag, $timeout) = @_;
242
243			my $queue = $Coro::current->{_coro_mp_queue}
244			or Carp::croak "Coro::MP::get called from thread not attached to any port";
245
246			my $i;
247
248			while () {
249			$queue->[$_][0] eq $tag
250			and return @{ splice @$queue, $_, 1 }
251			for $i..$#$queue;
252
253			$i = @$queue;
254
255			# wait for more messages
256			if (ref $timeout) {
257			schedule;
258			defined $i or return; # timeout
259
260			} elsif (defined $timeout) {
261			$timeout or return;
262
263			my $current = $Coro::current;
264			$timeout = AE::timer $timeout, 0, sub {
265			undef $i;
266			$current->ready;
267			};
268			} else {
269			$timeout = \$i; # dummy
270			}
271			}
272			}
273
274	root	1.3	=item @msg = get_cond { condition... } [$timeout]
275
276			Similarly to C<get>, looks for a matching message. Unlike C<get>,
277			"matching" is not defined by a tag alone, but by a predicate, a piece of
278			code that is executed on each candidate message in turn.
279
280			Also unlike C<get>, the predicate code is supposed to return the empty
281			list if the message didn't match. If it returns anything else, then the
282			message is removed from the queue and returned to the caller.
283
284			AS an exception, if the predicate returns a code reference, then it is
285			immediately invoked on the removed message.
286
287			If a C<$timeout> is specified and is not C<undef>, then, after this many
288			seconds have been passed without a matching message arriving, the empty
289			list will be returned.
290
291			TODO
292
293			=cut
294
295	root	1.1	sub get_cond {
296	root	1.3	my ($cond, $timeout) = @_;
297
298			my $queue = $Coro::current->{_coro_mp_queue}
299			or Carp::croak "Coro::MP::get_cond called from thread not attached to any port";
300
301			my ($i, $ok);
302
303			while () {
304			do
305			{
306			local *_ = $queue->[$_];
307			if ($ok = &$cond) {
308			splice @$queue, $_, 1;
309			&$ok if "CODE" eq ref $ok;
310			return @_;
311			}
312			}
313			for $i..$#$queue;
314
315			$i = @$queue;
316
317			# wait for more messages
318			if (ref $timeout) {
319			schedule;
320			defined $i or return; # timeout
321
322			} elsif (defined $timeout) {
323			$timeout or return;
324
325			my $current = $Coro::current;
326			$timeout = AE::timer $timeout, 0, sub {
327			undef $i;
328			$current->ready;
329			};
330			} else {
331			$timeout = \$i; # dummy
332			}
333			}
334			}
335
336			=item $async = pasync { BLOCK }
337
338			Sometimes you want to run a thread within a port context, for error
339			handling.
340
341			This function creates a new, ready, thread (using C<Coro::async>), sets
342			C<$SELF> to the the current value of C<$SELF> while it executing, and
343			calls the given BLOCK.
344
345			This is very similar to C<psub> - note that while the BLOCK exeuctes in
346			C<$SELF> port context, you cannot call C<get>, as C<$SELF> can only be
347			attached to one thread.
348
349			=cut
350
351			sub pasync(&) {
352	root	1.1	die "nyi";
353			}
354
355	root	1.3	=item @reply = syncal $port, @msg, $callback[, $timeout]
356	root	1.1
357	root	1.3	The synchronous form of C<cal>, a simple form of RPC - it sends a message
358			to the given C<$port> with the given contents (C<@msg>), but adds a reply
359			port to the message.
360	root	1.1
361			The reply port is created temporarily just for the purpose of receiving
362			the reply, and will be C<kil>ed when no longer needed.
363
364	root	1.3	Then it will wait until a reply message arrives, which will be returned to
365			the caller.
366	root	1.1
367	root	1.3	If the C<$timeout> is defined, then after this many seconds, when no
368			message has arrived, the port will be C<kil>ed and an empty list will be
369			returned.
370	root	1.1
371	root	1.3	If the C<$timeout> is undef, then the local port will monitor the remote
372			port instead, so it eventually gets cleaned-up.
373	root	1.1
374	root	1.3	=cut
375	root	1.1
376	root	1.3	sub syncal($@) {
377			my ($timeout, @msg) = @_;
378	root	1.1
379	root	1.3	cal @msg, Coro::rouse_cb, $timeout;
380			Coro::rouse_wait
381	root	1.1	}
382
383			=back
384
385			=head1 SEE ALSO
386
387			L<AnyEvent::MP::Intro> - a gentle introduction.
388
389			L<AnyEvent::MP> - like Coro::MP, but event-based.
390
391			L<AnyEvent>.
392
393			=head1 AUTHOR
394
395			Marc Lehmann <schmorp@schmorp.de>
396			http://home.schmorp.de/
397
398			=cut
399
400			1
401