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 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
   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 calxxxTODO $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 calxxxTODO(@) {
   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.2
Committed:	Fri Oct 2 00:00:22 2009 UTC (16 years, 11 months ago) by root
Branch:	MAIN
Changes since 1.1:	+5 -17 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			use AnyEvent::MP qw(snd rcv mon kil psub spawn); # TODO, should use AnyEvent::MP::Kernel only
77			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.2	snd rcv mon kil psub spawn
90	root	1.1	port_async rcv_async get get_cond
91			);
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			=item snd, rcv, mon, kil, psub
106
107			These variables and functions work exactly as in AnyEvent::MP, in fact,
108			they are usually exactly the same functions.
109
110			=item spawn
111
112			This function is also identical to C<AnyEvent::MP::spawn>. This means that
113			it doesn't spawn a new thread as one would expect, but simply calls an
114			init function. The init function, however, can attach a new thread easily:
115
116			sub initfun {
117			my (@args) = @_;
118
119			attach $SELF, sub {
120			# thread-code
121			};
122			}
123
124			=item $local_port = port_async { ... }
125
126			Creates a new local port, and returns its ID. A new thread is created and
127			attached to the port (see C<rcv_async>, below, for details).
128
129			=cut
130
131			sub rcv_async($$);
132
133			sub port_async(;&) {
134			my $id = "$UNIQ." . $ID++;
135			my $port = "$NODE#$id";
136
137			@_
138			? rcv_async $port, shift
139			: AnyEvent::MP::rcv $port, undef;
140
141			$port
142			}
143
144			=item rcv_async $port, $threadcb
145
146			This function creates and attaches a thread on a port. The thread is set
147			to execute C<$threadcb> and is put into the ready queue. The thread will
148			receive all messages not filtered away by tagged receive callbacks (as set
149			by C<AnyEvent::MP::rcv>) - it simply replaces the default callback of an
150			AnyEvent::MP port.
151
152			The special variable C<$SELF> will be set to C<$port> during thread
153			execution.
154
155			When C<$threadcb> returns or the thread is canceled, the return/cancel
156			values become the C<kil> reason.
157
158			It is not allowed to call C<rcv_async> more than once on a given port.
159
160			=cut
161
162			sub rcv_async($$) {
163			my ($port, $threadcb) = @_;
164
165			my (@queue, $coro);
166
167			AnyEvent::MP::rcv $port, sub {
168			push @queue, \@_; # TODO, take copy?
169			$coro->ready; # TODO, maybe too many unwanted wake-ups?
170			};
171
172			$coro = async {
173			# this SELF swapping is horrible
174			my $old_self;
175			Coro::on_enter { $old_self = $SELF; $SELF = $port };
176			Coro::on_leave { $SELF = $old_self };
177
178			$threadcb->();
179			};
180			$coro->{_coro_mp_queue} = \@queue;
181
182			mon $port, sub { $coro->cancel (@_) };
183			$coro->on_destroy (sub { kil $port, @_ });
184			}
185
186			=item @msg = get $tag
187
188			=item @msg = get $tag, $timeout
189
190			Find, dequeue and return the next message with the specified C<$tag>. If
191			no matching message is currently queued, wait up to C<$timeout> seconds
192			(or forever if no C<$timeout> has been specified or it is C<undef>) for
193			one to arrive.
194
195			Returns the message with the initial tag removed. In case of a timeout,
196			the empty list. The function I<must> be called in list context.
197
198			Note that empty messages cannot be distinguished from a timeout when using
199			C<rcv>.
200
201			=cut
202
203			sub get($;$) {
204			my ($tag, $timeout) = @_;
205
206			my $queue = $Coro::current->{_coro_mp_queue}
207			or Carp::croak "Coro::MP::get called from thread not attached to any port";
208
209			my $i;
210
211			while () {
212			$queue->[$_][0] eq $tag
213			and return @{ splice @$queue, $_, 1 }
214			for $i..$#$queue;
215
216			$i = @$queue;
217
218			# wait for more messages
219			if (ref $timeout) {
220			schedule;
221			defined $i or return; # timeout
222
223			} elsif (defined $timeout) {
224			$timeout or return;
225
226			my $current = $Coro::current;
227			$timeout = AE::timer $timeout, 0, sub {
228			undef $i;
229			$current->ready;
230			};
231			} else {
232			$timeout = \$i; # dummy
233			}
234			}
235			}
236
237			sub get_cond {
238			die "nyi";
239			}
240
241	root	1.2	=item calxxxTODO $port, @msg, $callback[, $timeout]
242	root	1.1
243			A simple form of RPC - sends a message to the given C<$port> with the
244			given contents (C<@msg>), but adds a reply port to the message.
245
246			The reply port is created temporarily just for the purpose of receiving
247			the reply, and will be C<kil>ed when no longer needed.
248
249			A reply message sent to the port is passed to the C<$callback> as-is.
250
251			If an optional time-out (in seconds) is given and it is not C<undef>,
252			then the callback will be called without any arguments after the time-out
253			elapsed and the port is C<kil>ed.
254
255			If no time-out is given, then the local port will monitor the remote port
256			instead, so it eventually gets cleaned-up.
257
258			Currently this function returns the temporary port, but this "feature"
259			might go in future versions unless you can make a convincing case that
260			this is indeed useful for something.
261
262			=cut
263
264	root	1.2	sub calxxxTODO(@) {
265	root	1.1	die "nyi";
266			}
267
268			=back
269
270			=head1 SEE ALSO
271
272			L<AnyEvent::MP::Intro> - a gentle introduction.
273
274			L<AnyEvent::MP> - like Coro::MP, but event-based.
275
276			L<AnyEvent>.
277
278			=head1 AUTHOR
279
280			Marc Lehmann <schmorp@schmorp.de>
281			http://home.schmorp.de/
282
283			=cut
284
285			1
286