ViewVC Help
View File | Revision Log | Show Annotations | Download File
/cvs/AnyEvent-MP/MP.pm
(Generate patch)

Comparing AnyEvent-MP/MP.pm (file contents):
Revision 1.10 by root, Sun Aug 2 18:05:43 2009 UTC vs.
Revision 1.28 by root, Tue Aug 4 22:16:54 2009 UTC

4 4
5=head1 SYNOPSIS 5=head1 SYNOPSIS
6 6
7 use AnyEvent::MP; 7 use AnyEvent::MP;
8 8
9 NODE # returns this node identifier
10 $NODE # contains this node identifier 9 $NODE # contains this node's noderef
10 NODE # returns this node's noderef
11 NODE $port # returns the noderef of the port
11 12
12 snd $port, type => data...; 13 snd $port, type => data...;
14
15 $SELF # receiving/own port id in rcv callbacks
13 16
14 rcv $port, smartmatch => $cb->($port, @msg); 17 rcv $port, smartmatch => $cb->($port, @msg);
15 18
16 # examples: 19 # examples:
17 rcv $port2, ping => sub { snd $_[0], "pong"; 0 }; 20 rcv $port2, ping => sub { snd $_[0], "pong"; 0 };
27This module (-family) implements a simple message passing framework. 30This module (-family) implements a simple message passing framework.
28 31
29Despite its simplicity, you can securely message other processes running 32Despite its simplicity, you can securely message other processes running
30on the same or other hosts. 33on the same or other hosts.
31 34
35For an introduction to this module family, see the L<AnyEvent::MP::Intro>
36manual page.
37
32At the moment, this module family is severly brokena nd underdocumented, 38At the moment, this module family is severly broken and underdocumented,
33so do not use. This was uploaded mainly to resreve the CPAN namespace - 39so do not use. This was uploaded mainly to reserve the CPAN namespace -
34stay tuned! 40stay tuned! The basic API should be finished, however.
35 41
36=head1 CONCEPTS 42=head1 CONCEPTS
37 43
38=over 4 44=over 4
39 45
82 88
83use AE (); 89use AE ();
84 90
85use base "Exporter"; 91use base "Exporter";
86 92
87our $VERSION = '0.02'; 93our $VERSION = '0.1';
88our @EXPORT = qw( 94our @EXPORT = qw(
89 NODE $NODE $PORT snd rcv _any_ 95 NODE $NODE *SELF node_of _any_
90 create_port create_port_on
91 become_slave become_public 96 become_slave become_public
97 snd rcv mon kil reg psub
98 port
92); 99);
93 100
101our $SELF;
102
103sub _self_die() {
104 my $msg = $@;
105 $msg =~ s/\n+$// unless ref $msg;
106 kil $SELF, die => $msg;
107}
108
94=item NODE / $NODE 109=item $thisnode = NODE / $NODE
95 110
96The C<NODE ()> function and the C<$NODE> variable contain the noderef of 111The C<NODE> function returns, and the C<$NODE> variable contains
97the local node. The value is initialised by a call to C<become_public> or 112the noderef of the local node. The value is initialised by a call
98C<become_slave>, after which all local port identifiers become invalid. 113to C<become_public> or C<become_slave>, after which all local port
114identifiers become invalid.
115
116=item $noderef = node_of $portid
117
118Extracts and returns the noderef from a portid or a noderef.
119
120=item $SELF
121
122Contains the current port id while executing C<rcv> callbacks or C<psub>
123blocks.
124
125=item SELF, %SELF, @SELF...
126
127Due to some quirks in how perl exports variables, it is impossible to
128just export C<$SELF>, all the symbols called C<SELF> are exported by this
129module, but only C<$SELF> is currently used.
99 130
100=item snd $portid, type => @data 131=item snd $portid, type => @data
101 132
102=item snd $portid, @msg 133=item snd $portid, @msg
103 134
117JSON is used, then only strings, numbers and arrays and hashes consisting 148JSON is used, then only strings, numbers and arrays and hashes consisting
118of those are allowed (no objects). When Storable is used, then anything 149of those are allowed (no objects). When Storable is used, then anything
119that Storable can serialise and deserialise is allowed, and for the local 150that Storable can serialise and deserialise is allowed, and for the local
120node, anything can be passed. 151node, anything can be passed.
121 152
122=item $local_port = create_port 153=item kil $portid[, @reason]
123 154
124Create a new local port object. See the next section for allowed methods. 155Kill the specified port with the given C<@reason>.
125 156
126=cut 157If no C<@reason> is specified, then the port is killed "normally" (linked
158ports will not be kileld, or even notified).
127 159
128sub create_port { 160Otherwise, linked ports get killed with the same reason (second form of
129 my $id = "$AnyEvent::MP::Base::UNIQ." . ++$AnyEvent::MP::Base::ID; 161C<mon>, see below).
130 162
131 my $self = bless { 163Runtime errors while evaluating C<rcv> callbacks or inside C<psub> blocks
132 id => "$NODE#$id", 164will be reported as reason C<< die => $@ >>.
133 names => [$id],
134 }, "AnyEvent::MP::Port";
135 165
136 $AnyEvent::MP::Base::PORT{$id} = sub { 166Transport/communication errors are reported as C<< transport_error =>
137 unshift @_, $self; 167$message >>.
138 168
139 for (@{ $self->{rc0}{$_[1]} }) { 169=item $guard = mon $portid, $cb->(@reason)
140 $_ && &{$_->[0]} 170
141 && undef $_; 171=item $guard = mon $portid, $otherport
172
173=item $guard = mon $portid, $otherport, @msg
174
175Monitor the given port and do something when the port is killed.
176
177In the first form, the callback is simply called with any number
178of C<@reason> elements (no @reason means that the port was deleted
179"normally"). Note also that I<< the callback B<must> never die >>, so use
180C<eval> if unsure.
181
182In the second form, the other port will be C<kil>'ed with C<@reason>, iff
183a @reason was specified, i.e. on "normal" kils nothing happens, while
184under all other conditions, the other port is killed with the same reason.
185
186In the last form, a message of the form C<@msg, @reason> will be C<snd>.
187
188Example: call a given callback when C<$port> is killed.
189
190 mon $port, sub { warn "port died because of <@_>\n" };
191
192Example: kill ourselves when C<$port> is killed abnormally.
193
194 mon $port, $self;
195
196Example: send us a restart message another C<$port> is killed.
197
198 mon $port, $self => "restart";
199
200=cut
201
202sub mon {
203 my ($noderef, $port, $cb) = ((split /#/, shift, 2), shift);
204
205 my $node = $NODE{$noderef} || add_node $noderef;
206
207 #TODO: ports must not be references
208 if (!ref $cb or "AnyEvent::MP::Port" eq ref $cb) {
209 if (@_) {
210 # send a kill info message
211 my (@msg) = ($cb, @_);
212 $cb = sub { snd @msg, @_ };
213 } else {
214 # simply kill other port
215 my $port = $cb;
216 $cb = sub { kil $port, @_ if @_ };
142 } 217 }
218 }
143 219
144 for (@{ $self->{rcv}{$_[1]} }) { 220 $node->monitor ($port, $cb);
145 $_ && [@_[1 .. @{$_->[1]}]] ~~ $_->[1] 221
146 && &{$_->[0]} 222 defined wantarray
147 && undef $_; 223 and AnyEvent::Util::guard { $node->unmonitor ($port, $cb) }
224}
225
226=item $guard = mon_guard $port, $ref, $ref...
227
228Monitors the given C<$port> and keeps the passed references. When the port
229is killed, the references will be freed.
230
231Optionally returns a guard that will stop the monitoring.
232
233This function is useful when you create e.g. timers or other watchers and
234want to free them when the port gets killed:
235
236 $port->rcv (start => sub {
237 my $timer; $timer = mon_guard $port, AE::timer 1, 1, sub {
238 undef $timer if 0.9 < rand;
148 } 239 });
240 });
149 241
150 for (@{ $self->{any} }) { 242=cut
151 $_ && [@_[0 .. $#{$_->[1]}]] ~~ $_->[1] 243
152 && &{$_->[0]} 244sub mon_guard {
153 && undef $_; 245 my ($port, @refs) = @_;
154 } 246
247 mon $port, sub { 0 && @refs }
248}
249
250=item lnk $port1, $port2
251
252Link two ports. This is simply a shorthand for:
253
254 mon $port1, $port2;
255 mon $port2, $port1;
256
257It means that if either one is killed abnormally, the other one gets
258killed as well.
259
260=item $local_port = port
261
262Create a new local port object that supports message matching.
263
264=item $portid = port { my @msg = @_; $finished }
265
266Creates a "mini port", that is, a very lightweight port without any
267pattern matching behind it, and returns its ID.
268
269The block will be called for every message received on the port. When the
270callback returns a true value its job is considered "done" and the port
271will be destroyed. Otherwise it will stay alive.
272
273The message will be passed as-is, no extra argument (i.e. no port id) will
274be passed to the callback.
275
276If you need the local port id in the callback, this works nicely:
277
278 my $port; $port = miniport {
279 snd $otherport, reply => $port;
155 }; 280 };
156 281
157 $self 282=cut
283
284sub port(;&) {
285 my $id = "$UNIQ." . $ID++;
286 my $port = "$NODE#$id";
287
288 if (@_) {
289 my $cb = shift;
290 $PORT{$id} = sub {
291 local $SELF = $port;
292 eval {
293 &$cb
294 and kil $id;
295 };
296 _self_die if $@;
297 };
298 } else {
299 my $self = bless {
300 id => "$NODE#$id",
301 }, "AnyEvent::MP::Port";
302
303 $PORT_DATA{$id} = $self;
304 $PORT{$id} = sub {
305 local $SELF = $port;
306
307 eval {
308 for (@{ $self->{rc0}{$_[0]} }) {
309 $_ && &{$_->[0]}
310 && undef $_;
311 }
312
313 for (@{ $self->{rcv}{$_[0]} }) {
314 $_ && [@_[1 .. @{$_->[1]}]] ~~ $_->[1]
315 && &{$_->[0]}
316 && undef $_;
317 }
318
319 for (@{ $self->{any} }) {
320 $_ && [@_[0 .. $#{$_->[1]}]] ~~ $_->[1]
321 && &{$_->[0]}
322 && undef $_;
323 }
324 };
325 _self_die if $@;
326 };
327 }
328
329 $port
158} 330}
159 331
160=item $portid = create_miniport { } 332=item reg $portid, $name
161 333
162Creates a "mini port", that is, a port without much #TODO 334Registers the given port under the name C<$name>. If the name already
335exists it is replaced.
163 336
164=cut 337A port can only be registered under one well known name.
165 338
166sub create_miniport(&) { 339A port automatically becomes unregistered when it is killed.
167 my $cb = shift;
168 my $id = "$AnyEvent::MP::Base::UNIQ." . ++$AnyEvent::MP::Base::ID;
169 340
170 $AnyEvent::MP::Base::PORT{$id} = sub { 341=cut
171 &$cb
172 and delete $AnyEvent::MP::Base::PORT{$id};
173 };
174 342
175 "$NODE#$id" 343sub reg(@) {
344 my ($portid, $name) = @_;
345
346 $REG{$name} = $portid;
176} 347}
177 348
178package AnyEvent::MP::Port; 349=item rcv $portid, tagstring => $callback->(@msg), ...
179 350
180=back 351=item rcv $portid, $smartmatch => $callback->(@msg), ...
181 352
182=head1 METHODS FOR PORT OBJECTS
183
184=over 4
185
186=item "$port"
187
188A port object stringifies to its port ID, so can be used directly for
189C<snd> operations.
190
191=cut
192
193use overload
194 '""' => sub { $_[0]{id} },
195 fallback => 1;
196
197=item $port->rcv (type => $callback->($port, @msg))
198
199=item $port->rcv ($smartmatch => $callback->($port, @msg))
200
201=item $port->rcv ([$smartmatch...] => $callback->($port, @msg)) 353=item rcv $portid, [$smartmatch...] => $callback->(@msg), ...
202 354
203Register a callback on the given port. 355Register callbacks to be called on matching messages on the given port.
204 356
205The callback has to return a true value when its work is done, after 357The callback has to return a true value when its work is done, after
206which is will be removed, or a false value in which case it will stay 358which is will be removed, or a false value in which case it will stay
207registered. 359registered.
208 360
361The global C<$SELF> (exported by this module) contains C<$portid> while
362executing the callback.
363
364Runtime errors wdurign callback execution will result in the port being
365C<kil>ed.
366
209If the match is an array reference, then it will be matched against the 367If the match is an array reference, then it will be matched against the
210first elements of the message, otherwise only the first element is being 368first elements of the message, otherwise only the first element is being
211matched. 369matched.
212 370
213Any element in the match that is specified as C<_any_> (a function 371Any element in the match that is specified as C<_any_> (a function
218also the most efficient match (by far). 376also the most efficient match (by far).
219 377
220=cut 378=cut
221 379
222sub rcv($@) { 380sub rcv($@) {
223 my ($self, $match, $cb) = @_; 381 my ($noderef, $port) = split /#/, shift, 2;
224 382
383 ($NODE{$noderef} || add_node $noderef) == $NODE{""}
384 or Carp::croak "$noderef#$port: rcv can only be called on local ports, caught";
385
386 my $self = $PORT_DATA{$port}
387 or Carp::croak "$noderef#$port: rcv can only be called on message matching ports, caught";
388
389 "AnyEvent::MP::Port" eq ref $self
390 or Carp::croak "$noderef#$port: rcv can only be called on message matching ports, caught";
391
392 while (@_) {
393 my ($match, $cb) = splice @_, 0, 2;
394
225 if (!ref $match) { 395 if (!ref $match) {
226 push @{ $self->{rc0}{$match} }, [$cb]; 396 push @{ $self->{rc0}{$match} }, [$cb];
227 } elsif (("ARRAY" eq ref $match && !ref $match->[0])) { 397 } elsif (("ARRAY" eq ref $match && !ref $match->[0])) {
228 my ($type, @match) = @$match; 398 my ($type, @match) = @$match;
229 @match 399 @match
230 ? push @{ $self->{rcv}{$match->[0]} }, [$cb, \@match] 400 ? push @{ $self->{rcv}{$match->[0]} }, [$cb, \@match]
231 : push @{ $self->{rc0}{$match->[0]} }, [$cb]; 401 : push @{ $self->{rc0}{$match->[0]} }, [$cb];
232 } else { 402 } else {
233 push @{ $self->{any} }, [$cb, $match]; 403 push @{ $self->{any} }, [$cb, $match];
404 }
234 } 405 }
235} 406}
236 407
237=item $port->register ($name) 408=item $closure = psub { BLOCK }
238 409
239Registers the given port under the well known name C<$name>. If the name 410Remembers C<$SELF> and creates a closure out of the BLOCK. When the
240already exists it is replaced. 411closure is executed, sets up the environment in the same way as in C<rcv>
412callbacks, i.e. runtime errors will cause the port to get C<kil>ed.
241 413
242A port can only be registered under one well known name. 414This is useful when you register callbacks from C<rcv> callbacks:
243 415
244=cut 416 rcv delayed_reply => sub {
417 my ($delay, @reply) = @_;
418 my $timer = AE::timer $delay, 0, psub {
419 snd @reply, $SELF;
420 };
421 };
245 422
246sub register { 423=cut
247 my ($self, $name) = @_;
248 424
249 $self->{wkname} = $name; 425sub psub(&) {
250 $AnyEvent::MP::Base::WKP{$name} = "$self"; 426 my $cb = shift;
427
428 my $port = $SELF
429 or Carp::croak "psub can only be called from within rcv or psub callbacks, not";
430
431 sub {
432 local $SELF = $port;
433
434 if (wantarray) {
435 my @res = eval { &$cb };
436 _self_die if $@;
437 @res
438 } else {
439 my $res = eval { &$cb };
440 _self_die if $@;
441 $res
442 }
443 }
251} 444}
252 445
253=item $port->destroy
254
255Explicitly destroy/remove/nuke/vaporise the port.
256
257Ports are normally kept alive by there mere existance alone, and need to
258be destroyed explicitly.
259
260=cut
261
262sub destroy {
263 my ($self) = @_;
264
265 delete $AnyEvent::MP::Base::WKP{ $self->{wkname} };
266
267 delete $AnyEvent::MP::Base::PORT{$_}
268 for @{ $self->{names} };
269}
270
271=back 446=back
272 447
273=head1 FUNCTIONS FOR NODES 448=head1 FUNCTIONS FOR NODES
274 449
275=over 4 450=over 4
276
277=item mon $noderef, $callback->($noderef, $status, $)
278
279Monitors the given noderef.
280 451
281=item become_public endpoint... 452=item become_public endpoint...
282 453
283Tells the node to become a public node, i.e. reachable from other nodes. 454Tells the node to become a public node, i.e. reachable from other nodes.
284 455
304 475
305=over 4 476=over 4
306 477
307=cut 478=cut
308 479
309=item wkp => $name, @reply 480=item lookup => $name, @reply
310 481
311Replies with the port ID of the specified well-known port, or C<undef>. 482Replies with the port ID of the specified well-known port, or C<undef>.
312 483
313=item devnull => ... 484=item devnull => ...
314 485
337 snd $NODE, time => $myport, timereply => 1, 2; 508 snd $NODE, time => $myport, timereply => 1, 2;
338 # => snd $myport, timereply => 1, 2, <time> 509 # => snd $myport, timereply => 1, 2, <time>
339 510
340=back 511=back
341 512
513=head1 AnyEvent::MP vs. Distributed Erlang
514
515AnyEvent::MP got lots of its ideas from distributed erlang (erlang node
516== aemp node, erlang process == aemp port), so many of the documents and
517programming techniques employed by erlang apply to AnyEvent::MP. Here is a
518sample:
519
520 http://www.erlang.se/doc/programming_rules.shtml
521 http://erlang.org/doc/getting_started/part_frame.html # chapters 3 and 4
522 http://erlang.org/download/erlang-book-part1.pdf # chapters 5 and 6
523 http://erlang.org/download/armstrong_thesis_2003.pdf # chapters 4 and 5
524
525Despite the similarities, there are also some important differences:
526
527=over 4
528
529=item * Node references contain the recipe on how to contact them.
530
531Erlang relies on special naming and DNS to work everywhere in the
532same way. AEMP relies on each node knowing it's own address(es), with
533convenience functionality.
534
535This means that AEMP requires a less tightly controlled environment at the
536cost of longer node references and a slightly higher management overhead.
537
538=item * Erlang uses processes and a mailbox, AEMP does not queue.
539
540Erlang uses processes that selctively receive messages, and therefore
541needs a queue. AEMP is event based, queuing messages would serve no useful
542purpose.
543
544(But see L<Coro::MP> for a more erlang-like process model on top of AEMP).
545
546=item * Erlang sends are synchronous, AEMP sends are asynchronous.
547
548Sending messages in erlang is synchronous and blocks the process. AEMP
549sends are immediate, connection establishment is handled in the
550background.
551
552=item * Erlang can silently lose messages, AEMP cannot.
553
554Erlang makes few guarantees on messages delivery - messages can get lost
555without any of the processes realising it (i.e. you send messages a, b,
556and c, and the other side only receives messages a and c).
557
558AEMP guarantees correct ordering, and the guarantee that there are no
559holes in the message sequence.
560
561=item * In erlang, processes can be declared dead and later be found to be
562alive.
563
564In erlang it can happen that a monitored process is declared dead and
565linked processes get killed, but later it turns out that the process is
566still alive - and can receive messages.
567
568In AEMP, when port monitoring detects a port as dead, then that port will
569eventually be killed - it cannot happen that a node detects a port as dead
570and then later sends messages to it, finding it is still alive.
571
572=item * Erlang can send messages to the wrong port, AEMP does not.
573
574In erlang it is quite possible that a node that restarts reuses a process
575ID known to other nodes for a completely different process, causing
576messages destined for that process to end up in an unrelated process.
577
578AEMP never reuses port IDs, so old messages or old port IDs floating
579around in the network will not be sent to an unrelated port.
580
581=item * Erlang uses unprotected connections, AEMP uses secure
582authentication and can use TLS.
583
584AEMP can use a proven protocol - SSL/TLS - to protect connections and
585securely authenticate nodes.
586
587=item * The AEMP protocol is optimised for both text-based and binary
588communications.
589
590The AEMP protocol, unlike the erlang protocol, supports both
591language-independent text-only protocols (good for debugging) and binary,
592language-specific serialisers (e.g. Storable).
593
594It has also been carefully designed to be implementable in other languages
595with a minimum of work while gracefully degrading fucntionality to make the
596protocol simple.
597
598=back
599
342=head1 SEE ALSO 600=head1 SEE ALSO
343 601
344L<AnyEvent>. 602L<AnyEvent>.
345 603
346=head1 AUTHOR 604=head1 AUTHOR

Diff Legend

Removed lines
+ Added lines
< Changed lines
> Changed lines