cvsroot/Coro-MP/MP.pm

=head1 NAME

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

=head1 SYNOPSIS

   use Coro::MP;

   # exports everything that AnyEvent::MP exports as well.
   # new stuff compared to AnyEvent::MP:

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

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

   # attach to an existing port
   spawn $NODE, "::initfunc";
   sub ::initfunc {
      rcv_async $SELF, sub {
         ...
      };
   }

   # simple "tag" receives:
   my ($pid) = get "pid", 30
      or die "no pid message received after 30s";

   # conditional receive
   my ($tag, @data) = get_cond { $_[0] =~ /^group1_/ };
   my @next_msg = get_cond { 1 } 30; # 30s timeout

   # run thread in port context
   peval_async $port, sub {
      die "kill the port\n";
   };

   # synchronous "cal"
   my @retval = syncal 30, $port, tag => $data;

=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, see L<AnyEvent::MP::Intro>, which also covers
C<Coro::MP>.

=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 = (@AnyEvent::MP::EXPORT, qw(
   port_async rcv_async get get_cond syncal peval_async
));
our @EXPORT_OK = (@AnyEvent::MP::EXPORT_OK);

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

   my $coro = async_pool {
      # we do it inside the thread, because this hopefully gives us the correct $SELF,
      # as $SELF should not be localised when threads are switched.
      $Coro::current->swap_sv (\$SELF, \$port);
      eval { $threadcb->() };
      kil $SELF, die => $@ if $@;
   };

   # killing the port cancels the coro
   # delaying kil messages inside aemp guarantees
   # (hopefully) that $coro != $Coro::current.
   mon $port, sub { $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, psub

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 peval

This function works exactly as C<AnyEvent::MP::peval> - 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 somewhat annoying
syntax, there is a C<peval_async> 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 feel fine 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 = "$AnyEvent::MP::Kernel::UNIQ." . $AnyEvent::MP::Kernel::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>.

Example: send a "log" message to C<$SELF> and then get and print it.

   snd $SELF, log => "text";
   my ($text) = get "log";
   print "log message: $text\n";

Example: receive C<p1> and C<p2> messages, regardless of the order they
arrive in on the port.

   my @p1 = get "p1";
   my @21 = get "p2";

Example: assume a message with tag C<now> is already in the queue and fetch
it. If no message was there, do not wait, but die.

   my @msg = get "now", 0
      or die "expected now emssage to be there, but it wasn't";

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

Example: fetch the next message, wait as long as necessary.

   my @msg = get_cond { 1 };

Example: fetch the next message whose tag starts with C<group1_>.

   my ($tag, @data) = get_cond { $_[0] =~ /^group1_/ };

Example: check whether a message with tag C<child_exit> and a second
elemet of C<$pid> is in the queue already.

   if (
      my (undef, $pid, $status) =
         get_cond {
            $_[0] eq "child_exit" && $_[1] == $pid
         } 0
   ) {
      warn "child $pid did exit with status $status\n";
   }

Example: implement a server that reacts to C<log>, C<exit> and C<reverse>
messages, and exits after 30 seconds of idling.

   my $reverser = port_async {
      while() {
         get_cond {
            $_[0] eq "exit" and return sub {
               last; # yes, this is valid
            };
            $_[0] eq "log" and return sub {
               print "log: $_[1]\n";
            };
            $_[0] eq "reverse" and return sub {
               my (undef, $text, @reply) = @_;
               snd @reply, scalar reverse $text;
            };

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

=cut

sub _true { 1 }

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);

   $cond ||= \&_true;

   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 = peval_async $port, sub { 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.

Example: execute some Coro::AIO code concurrently in another thread, but
make sure any errors C<kil> the originating port.

   port_async {
      ...
      peval_async $SELF, {
         # $SELF set, but cannot call get etc. here

         my $fh = aio_open ...
            or die "open: $!";

         aio_close $fh;
      };
   };

=cut

sub peval_async($$) {
   _new_coro $_[0], $_[1]
}

=item @reply = syncal $timeout, $port => @msg

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.

Example: call the string reverse example from C<get_cond>.

   my $reversed = syncal 1, $reverse, reverse => "Rotator";

=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.16
Committed:	Wed Sep 26 18:34:02 2012 UTC (13 years, 11 months ago) by root
Branch:	MAIN
CVS Tags:	HEAD
Changes since 1.15:	+1 -1 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	root	1.7	# exports everything that AnyEvent::MP exports as well.
10			# new stuff compared to AnyEvent::MP:
11	root	1.1
12	root	1.7	# creating/using ports from threads
13	root	1.2	my $port = port_async {
14	root	1.1	# thread context, $SELF is set to $port
15
16			# returning will "kil" the $port with an empty reason
17			};
18
19	root	1.7	# attach to an existing port
20			spawn $NODE, "::initfunc";
21			sub ::initfunc {
22			rcv_async $SELF, sub {
23			...
24			};
25			}
26
27			# simple "tag" receives:
28			my ($pid) = get "pid", 30
29			or die "no pid message received after 30s";
30
31			# conditional receive
32			my ($tag, @data) = get_cond { $_[0] =~ /^group1_/ };
33			my @next_msg = get_cond { 1 } 30; # 30s timeout
34
35			# run thread in port context
36	root	1.12	peval_async $port, sub {
37	root	1.7	die "kill the port\n";
38	root	1.1	};
39
40	root	1.7	# synchronous "cal"
41	root	1.11	my @retval = syncal 30, $port, tag => $data;
42	root	1.1
43			=head1 DESCRIPTION
44
45			This module (-family) implements a simple message passing framework.
46
47			Despite its simplicity, you can securely message other processes running
48			on the same or other hosts, and you can supervise entities remotely.
49
50			This module depends heavily on L<AnyEvent::MP>, in fact, many functions
51			exported by this module are identical to AnyEvent::MP functions. This
52			module family is simply the Coro API to AnyEvent::MP.
53
54			Care has been taken to stay compatible with AnyEvent::MP, even if
55			sometimes this required a less natural API (C<spawn> should indeed spawn a
56			thread, not just call an initfunc for example).
57
58	root	1.14	For an introduction, see L<AnyEvent::MP::Intro>, which also covers
59			C<Coro::MP>.
60	root	1.1
61			=head1 VARIABLES/FUNCTIONS
62
63			=over 4
64
65			=cut
66
67			package Coro::MP;
68
69			use common::sense;
70
71			use Carp ();
72
73			use AnyEvent::MP::Kernel;
74	root	1.4	use AnyEvent::MP;
75	root	1.1	use Coro;
76			use Coro::AnyEvent ();
77
78			use AE ();
79
80			use base "Exporter";
81
82			our $VERSION = "0.1";
83
84	root	1.7	our @EXPORT = (@AnyEvent::MP::EXPORT, qw(
85			port_async rcv_async get get_cond syncal peval_async
86			));
87			our @EXPORT_OK = (@AnyEvent::MP::EXPORT_OK);
88	root	1.1
89	root	1.4	sub _new_coro {
90			my ($port, $threadcb) = @_;
91
92			my $coro = async_pool {
93	root	1.15	# we do it inside the thread, because this hopefully gives us the correct $SELF,
94			# as $SELF should not be localised when threads are switched.
95	root	1.16	$Coro::current->swap_sv (\$SELF, \$port);
96	root	1.4	eval { $threadcb->() };
97			kil $SELF, die => $@ if $@;
98			};
99
100			# killing the port cancels the coro
101	root	1.7	# delaying kil messages inside aemp guarantees
102	root	1.6	# (hopefully) that $coro != $Coro::current.
103			mon $port, sub { $coro->cancel (@_) };
104	root	1.4
105			# cancelling the coro kills the port
106			$coro->on_destroy (sub { kil $port, @_ });
107	root	1.1
108	root	1.4	$coro
109	root	1.1	}
110
111			=item NODE, $NODE, node_of, configure
112
113			=item $SELF, *SELF, SELF, %SELF, @SELF...
114
115	root	1.7	=item snd, mon, kil, psub
116	root	1.1
117			These variables and functions work exactly as in AnyEvent::MP, in fact,
118	root	1.3	they are exactly the same functions, and are used in much the same way.
119
120			=item rcv
121
122			This function works exactly as C<AnyEvent::MP::rcv>, and is in fact
123			compatible with Coro::MP ports. However, the canonical way to receive
124			messages with Coro::MP is to use C<get> or C<get_cond>.
125
126			=item port
127
128			This function is exactly the same as C<AnyEvent::MP::port> and creates new
129			ports. You can attach a thread to them by calling C<rcv_async> or you can
130			do a create and attach in one operation using C<port_async>.
131
132	root	1.7	=item peval
133	root	1.3
134	root	1.8	This function works exactly as C<AnyEvent::MP::peval> - you could use it to
135	root	1.3	run callbacks within a port context (good for monitoring), but you cannot
136			C<get> messages unless the callback executes within the thread attached to
137			the port.
138
139	root	1.8	Since creating a thread with port context requires somewhat annoying
140			syntax, there is a C<peval_async> function that handles that for you -
141			note that within such a thread, you still cannot C<get> messages.
142	root	1.1
143			=item spawn
144
145	root	1.3	This function is identical to C<AnyEvent::MP::spawn>. This means that
146	root	1.1	it doesn't spawn a new thread as one would expect, but simply calls an
147			init function. The init function, however, can attach a new thread easily:
148
149			sub initfun {
150			my (@args) = @_;
151
152	root	1.3	rcv_async $SELF, sub {
153	root	1.1	# thread-code
154			};
155			}
156
157	root	1.3	=item cal
158
159			This function is identical to C<AnyEvent::MP::cal>. The easiest way to
160			make a synchronous call is to use Coro's rouse functionality:
161
162			# send 1, 2, 3 to $port and wait up to 30s for reply
163			cal $port, 1, 2, 3, rouse_cb, 30;
164			my @reply = rouse_wait;
165
166	root	1.9	You can also use C<syncal> if you want, and feel fine with learning yet
167	root	1.4	another function with a weird name:
168	root	1.3
169			my @reply = syncal 30, $port, 1, 2, 3;
170
171	root	1.1	=item $local_port = port_async { ... }
172
173			Creates a new local port, and returns its ID. A new thread is created and
174			attached to the port (see C<rcv_async>, below, for details).
175
176			=cut
177
178			sub rcv_async($$);
179
180			sub port_async(;&) {
181	root	1.13	my $id = "$AnyEvent::MP::Kernel::UNIQ." . $AnyEvent::MP::Kernel::ID++;
182	root	1.1	my $port = "$NODE#$id";
183
184			@_
185			? rcv_async $port, shift
186			: AnyEvent::MP::rcv $port, undef;
187
188			$port
189			}
190
191			=item rcv_async $port, $threadcb
192
193			This function creates and attaches a thread on a port. The thread is set
194			to execute C<$threadcb> and is put into the ready queue. The thread will
195			receive all messages not filtered away by tagged receive callbacks (as set
196			by C<AnyEvent::MP::rcv>) - it simply replaces the default callback of an
197			AnyEvent::MP port.
198
199			The special variable C<$SELF> will be set to C<$port> during thread
200			execution.
201
202			When C<$threadcb> returns or the thread is canceled, the return/cancel
203			values become the C<kil> reason.
204
205			It is not allowed to call C<rcv_async> more than once on a given port.
206
207			=cut
208
209			sub rcv_async($$) {
210			my ($port, $threadcb) = @_;
211
212			my (@queue, $coro);
213
214			AnyEvent::MP::rcv $port, sub {
215			push @queue, \@_; # TODO, take copy?
216			$coro->ready; # TODO, maybe too many unwanted wake-ups?
217			};
218
219	root	1.4	$coro = _new_coro $port, $threadcb;
220	root	1.1	$coro->{_coro_mp_queue} = \@queue;
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	root	1.7	Example: send a "log" message to C<$SELF> and then get and print it.
239
240			snd $SELF, log => "text";
241			my ($text) = get "log";
242			print "log message: $text\n";
243
244			Example: receive C<p1> and C<p2> messages, regardless of the order they
245			arrive in on the port.
246
247			my @p1 = get "p1";
248			my @21 = get "p2";
249
250			Example: assume a message with tag C<now> is already in the queue and fetch
251			it. If no message was there, do not wait, but die.
252
253			my @msg = get "now", 0
254			or die "expected now emssage to be there, but it wasn't";
255
256	root	1.1	=cut
257
258			sub get($;$) {
259			my ($tag, $timeout) = @_;
260
261			my $queue = $Coro::current->{_coro_mp_queue}
262			or Carp::croak "Coro::MP::get called from thread not attached to any port";
263
264			my $i;
265
266			while () {
267			$queue->[$_][0] eq $tag
268			and return @{ splice @$queue, $_, 1 }
269			for $i..$#$queue;
270
271			$i = @$queue;
272
273			# wait for more messages
274			if (ref $timeout) {
275			schedule;
276			defined $i or return; # timeout
277
278			} elsif (defined $timeout) {
279			$timeout or return;
280
281			my $current = $Coro::current;
282			$timeout = AE::timer $timeout, 0, sub {
283			undef $i;
284			$current->ready;
285			};
286			} else {
287			$timeout = \$i; # dummy
288			}
289			}
290			}
291
292	root	1.3	=item @msg = get_cond { condition... } [$timeout]
293
294			Similarly to C<get>, looks for a matching message. Unlike C<get>,
295			"matching" is not defined by a tag alone, but by a predicate, a piece of
296	root	1.4	code that is executed on each candidate message in turn, with C<@_> set to
297			the message contents.
298	root	1.3
299	root	1.4	The predicate code is supposed to return the empty list if the message
300			didn't match. If it returns anything else, then the message is removed
301			from the queue and returned to the caller.
302	root	1.3
303	root	1.4	In addition, if the predicate returns a code reference, then it is
304			immediately called invoked on the removed message.
305	root	1.3
306			If a C<$timeout> is specified and is not C<undef>, then, after this many
307			seconds have been passed without a matching message arriving, the empty
308			list will be returned.
309
310	root	1.7	Example: fetch the next message, wait as long as necessary.
311
312			my @msg = get_cond { 1 };
313
314			Example: fetch the next message whose tag starts with C<group1_>.
315
316			my ($tag, @data) = get_cond { $_[0] =~ /^group1_/ };
317
318			Example: check whether a message with tag C<child_exit> and a second
319			elemet of C<$pid> is in the queue already.
320
321			if (
322			my (undef, $pid, $status) =
323			get_cond {
324			$_[0] eq "child_exit" && $_[1] == $pid
325			} 0
326			) {
327			warn "child $pid did exit with status $status\n";
328			}
329
330			Example: implement a server that reacts to C<log>, C<exit> and C<reverse>
331			messages, and exits after 30 seconds of idling.
332
333			my $reverser = port_async {
334			while() {
335			get_cond {
336			$_[0] eq "exit" and return sub {
337			last; # yes, this is valid
338			};
339			$_[0] eq "log" and return sub {
340			print "log: $_[1]\n";
341			};
342			$_[0] eq "reverse" and return sub {
343			my (undef, $text, @reply) = @_;
344			snd @reply, scalar reverse $text;
345			};
346
347			die "unexpected message $_[0] received";
348			} 30
349			or last;
350			}
351			};
352	root	1.3
353			=cut
354
355	root	1.7	sub _true { 1 }
356
357			sub get_cond(;&$) {
358	root	1.3	my ($cond, $timeout) = @_;
359
360			my $queue = $Coro::current->{_coro_mp_queue}
361			or Carp::croak "Coro::MP::get_cond called from thread not attached to any port";
362
363			my ($i, $ok);
364
365	root	1.7	$cond \|\|= \&_true;
366
367	root	1.3	while () {
368			do
369			{
370			local *_ = $queue->[$_];
371			if ($ok = &$cond) {
372			splice @$queue, $_, 1;
373			&$ok if "CODE" eq ref $ok;
374			return @_;
375			}
376			}
377			for $i..$#$queue;
378
379			$i = @$queue;
380
381			# wait for more messages
382			if (ref $timeout) {
383			schedule;
384			defined $i or return; # timeout
385
386			} elsif (defined $timeout) {
387			$timeout or return;
388
389			my $current = $Coro::current;
390			$timeout = AE::timer $timeout, 0, sub {
391			undef $i;
392			$current->ready;
393			};
394			} else {
395			$timeout = \$i; # dummy
396			}
397			}
398			}
399
400	root	1.12	=item $async = peval_async $port, sub { BLOCK }
401	root	1.3
402			Sometimes you want to run a thread within a port context, for error
403			handling.
404
405			This function creates a new, ready, thread (using C<Coro::async>), sets
406			C<$SELF> to the the current value of C<$SELF> while it executing, and
407			calls the given BLOCK.
408
409			This is very similar to C<psub> - note that while the BLOCK exeuctes in
410			C<$SELF> port context, you cannot call C<get>, as C<$SELF> can only be
411			attached to one thread.
412
413	root	1.7	Example: execute some Coro::AIO code concurrently in another thread, but
414			make sure any errors C<kil> the originating port.
415
416			port_async {
417			...
418	root	1.12	peval_async $SELF, {
419	root	1.7	# $SELF set, but cannot call get etc. here
420
421			my $fh = aio_open ...
422			or die "open: $!";
423
424			aio_close $fh;
425			};
426			};
427
428	root	1.3	=cut
429
430	root	1.7	sub peval_async($$) {
431			_new_coro $_[0], $_[1]
432	root	1.1	}
433
434	root	1.12	=item @reply = syncal $timeout, $port => @msg
435	root	1.1
436	root	1.3	The synchronous form of C<cal>, a simple form of RPC - it sends a message
437			to the given C<$port> with the given contents (C<@msg>), but adds a reply
438			port to the message.
439	root	1.1
440			The reply port is created temporarily just for the purpose of receiving
441			the reply, and will be C<kil>ed when no longer needed.
442
443	root	1.3	Then it will wait until a reply message arrives, which will be returned to
444			the caller.
445	root	1.1
446	root	1.3	If the C<$timeout> is defined, then after this many seconds, when no
447			message has arrived, the port will be C<kil>ed and an empty list will be
448			returned.
449	root	1.1
450	root	1.3	If the C<$timeout> is undef, then the local port will monitor the remote
451			port instead, so it eventually gets cleaned-up.
452	root	1.1
453	root	1.7	Example: call the string reverse example from C<get_cond>.
454
455			my $reversed = syncal 1, $reverse, reverse => "Rotator";
456
457	root	1.3	=cut
458	root	1.1
459	root	1.3	sub syncal($@) {
460			my ($timeout, @msg) = @_;
461	root	1.1
462	root	1.3	cal @msg, Coro::rouse_cb, $timeout;
463			Coro::rouse_wait
464	root	1.1	}
465
466			=back
467
468			=head1 SEE ALSO
469
470			L<AnyEvent::MP::Intro> - a gentle introduction.
471
472			L<AnyEvent::MP> - like Coro::MP, but event-based.
473
474			L<AnyEvent>.
475
476			=head1 AUTHOR
477
478			Marc Lehmann <schmorp@schmorp.de>
479			http://home.schmorp.de/
480
481			=cut
482
483			1
484