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 {
      # thread context, $SELF is set to $port

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

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

   # selective receive
   rcvx {
      if ($_[0] eq "incdata") {
         my (undef, $data, @reply) = @_;
         snd @reply, $data + 1;

      } else {
         next; # skip (but store) this message
      }
   } 30
      or die "timeout after 30s";

   # 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 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
   snd rcv mon kil psub spawn cal
   port_async rcv_async get get_cond
);

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, rcv, mon, kil, psub

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

=item spawn

This function is also 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) = @_;

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

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

sub get_cond {
   die "nyi";
}

=item cal $port, @msg, $callback[, $timeout]

A simple form of RPC - 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.

A reply message sent to the port is passed to the C<$callback> as-is.

If an optional time-out (in seconds) is given and it is not C<undef>,
then the callback will be called without any arguments after the time-out
elapsed and the port is C<kil>ed.

If no time-out is given, then the local port will monitor the remote port
instead, so it eventually gets cleaned-up.

Currently this function returns the temporary port, but this "feature"
might go in future versions unless you can make a convincing case that
this is indeed useful for something.

=cut

sub cal(@) {
   die "nyi";
}

=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.1
Committed:	Thu Oct 1 23:56:31 2009 UTC (16 years, 11 months ago) by root
Branch:	MAIN
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			my $port = port {
26			# 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			my (undef, @data) = rcv "tag";
34			};
35			snd $port, tag => "data0", "data1";
36
37			# selective receive
38			rcvx {
39			if ($_[0] eq "incdata") {
40			my (undef, $data, @reply) = @_;
41			snd @reply, $data + 1;
42
43			} else {
44			next; # skip (but store) this message
45			}
46			} 30
47			or die "timeout after 30s";
48
49			# create a port on another node
50			my $port = spawn $node, $initfunc, @initdata;
51
52			# monitoring
53			mon $localport, $cb->(@msg) # callback is invoked on death
54			mon $localport, $otherport # kill otherport on abnormal death
55			mon $localport, $otherport, @msg # send message on death
56
57			=head1 DESCRIPTION
58
59			This module (-family) implements a simple message passing framework.
60
61			Despite its simplicity, you can securely message other processes running
62			on the same or other hosts, and you can supervise entities remotely.
63
64			This module depends heavily on L<AnyEvent::MP>, in fact, many functions
65			exported by this module are identical to AnyEvent::MP functions. This
66			module family is simply the Coro API to AnyEvent::MP.
67
68			Care has been taken to stay compatible with AnyEvent::MP, even if
69			sometimes this required a less natural API (C<spawn> should indeed spawn a
70			thread, not just call an initfunc for example).
71
72			For an introduction to AnyEvent::MP, see the L<AnyEvent::MP::Intro> manual
73			page.
74
75			=head1 VARIABLES/FUNCTIONS
76
77			=over 4
78
79			=cut
80
81			package Coro::MP;
82
83			use common::sense;
84
85			use Carp ();
86
87			use AnyEvent::MP::Kernel;
88			use AnyEvent::MP qw(snd rcv mon kil psub spawn); # TODO, should use AnyEvent::MP::Kernel only
89			use Coro;
90			use Coro::AnyEvent ();
91
92			use AE ();
93
94			use base "Exporter";
95
96			our $VERSION = "0.1";
97
98			our @EXPORT = qw(
99			NODE $NODE *SELF node_of
100			configure
101			snd rcv mon kil psub spawn cal
102			port_async rcv_async get get_cond
103			);
104
105			our $SELF;
106
107			sub _self_die() {
108			my $msg = $@;
109			$msg =~ s/\n+$// unless ref $msg;
110			kil $SELF, die => $msg;
111			}
112
113			=item NODE, $NODE, node_of, configure
114
115			=item $SELF, *SELF, SELF, %SELF, @SELF...
116
117			=item snd, rcv, mon, kil, psub
118
119			These variables and functions work exactly as in AnyEvent::MP, in fact,
120			they are usually exactly the same functions.
121
122			=item spawn
123
124			This function is also identical to C<AnyEvent::MP::spawn>. This means that
125			it doesn't spawn a new thread as one would expect, but simply calls an
126			init function. The init function, however, can attach a new thread easily:
127
128			sub initfun {
129			my (@args) = @_;
130
131			attach $SELF, sub {
132			# thread-code
133			};
134			}
135
136			=item $local_port = port_async { ... }
137
138			Creates a new local port, and returns its ID. A new thread is created and
139			attached to the port (see C<rcv_async>, below, for details).
140
141			=cut
142
143			sub rcv_async($$);
144
145			sub port_async(;&) {
146			my $id = "$UNIQ." . $ID++;
147			my $port = "$NODE#$id";
148
149			@_
150			? rcv_async $port, shift
151			: AnyEvent::MP::rcv $port, undef;
152
153			$port
154			}
155
156			=item rcv_async $port, $threadcb
157
158			This function creates and attaches a thread on a port. The thread is set
159			to execute C<$threadcb> and is put into the ready queue. The thread will
160			receive all messages not filtered away by tagged receive callbacks (as set
161			by C<AnyEvent::MP::rcv>) - it simply replaces the default callback of an
162			AnyEvent::MP port.
163
164			The special variable C<$SELF> will be set to C<$port> during thread
165			execution.
166
167			When C<$threadcb> returns or the thread is canceled, the return/cancel
168			values become the C<kil> reason.
169
170			It is not allowed to call C<rcv_async> more than once on a given port.
171
172			=cut
173
174			sub rcv_async($$) {
175			my ($port, $threadcb) = @_;
176
177			my (@queue, $coro);
178
179			AnyEvent::MP::rcv $port, sub {
180			push @queue, \@_; # TODO, take copy?
181			$coro->ready; # TODO, maybe too many unwanted wake-ups?
182			};
183
184			$coro = async {
185			# this SELF swapping is horrible
186			my $old_self;
187			Coro::on_enter { $old_self = $SELF; $SELF = $port };
188			Coro::on_leave { $SELF = $old_self };
189
190			$threadcb->();
191			};
192			$coro->{_coro_mp_queue} = \@queue;
193
194			mon $port, sub { $coro->cancel (@_) };
195			$coro->on_destroy (sub { kil $port, @_ });
196			}
197
198			=item @msg = get $tag
199
200			=item @msg = get $tag, $timeout
201
202			Find, dequeue and return the next message with the specified C<$tag>. If
203			no matching message is currently queued, wait up to C<$timeout> seconds
204			(or forever if no C<$timeout> has been specified or it is C<undef>) for
205			one to arrive.
206
207			Returns the message with the initial tag removed. In case of a timeout,
208			the empty list. The function I<must> be called in list context.
209
210			Note that empty messages cannot be distinguished from a timeout when using
211			C<rcv>.
212
213			=cut
214
215			sub get($;$) {
216			my ($tag, $timeout) = @_;
217
218			my $queue = $Coro::current->{_coro_mp_queue}
219			or Carp::croak "Coro::MP::get called from thread not attached to any port";
220
221			my $i;
222
223			while () {
224			$queue->[$_][0] eq $tag
225			and return @{ splice @$queue, $_, 1 }
226			for $i..$#$queue;
227
228			$i = @$queue;
229
230			# wait for more messages
231			if (ref $timeout) {
232			schedule;
233			defined $i or return; # timeout
234
235			} elsif (defined $timeout) {
236			$timeout or return;
237
238			my $current = $Coro::current;
239			$timeout = AE::timer $timeout, 0, sub {
240			undef $i;
241			$current->ready;
242			};
243			} else {
244			$timeout = \$i; # dummy
245			}
246			}
247			}
248
249			sub get_cond {
250			die "nyi";
251			}
252
253			=item cal $port, @msg, $callback[, $timeout]
254
255			A simple form of RPC - sends a message to the given C<$port> with the
256			given contents (C<@msg>), but adds a reply port to the message.
257
258			The reply port is created temporarily just for the purpose of receiving
259			the reply, and will be C<kil>ed when no longer needed.
260
261			A reply message sent to the port is passed to the C<$callback> as-is.
262
263			If an optional time-out (in seconds) is given and it is not C<undef>,
264			then the callback will be called without any arguments after the time-out
265			elapsed and the port is C<kil>ed.
266
267			If no time-out is given, then the local port will monitor the remote port
268			instead, so it eventually gets cleaned-up.
269
270			Currently this function returns the temporary port, but this "feature"
271			might go in future versions unless you can make a convincing case that
272			this is indeed useful for something.
273
274			=cut
275
276			sub cal(@) {
277			die "nyi";
278			}
279
280			=back
281
282			=head1 SEE ALSO
283
284			L<AnyEvent::MP::Intro> - a gentle introduction.
285
286			L<AnyEvent::MP> - like Coro::MP, but event-based.
287
288			L<AnyEvent>.
289
290			=head1 AUTHOR
291
292			Marc Lehmann <schmorp@schmorp.de>
293			http://home.schmorp.de/
294
295			=cut
296
297			1
298