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

Comparing Coro/Coro.pm (file contents):
Revision 1.198 by root, Sun Sep 21 01:23:26 2008 UTC vs.
Revision 1.229 by root, Thu Nov 20 06:32:55 2008 UTC

56 56
57=cut 57=cut
58 58
59package Coro; 59package Coro;
60 60
61use strict; 61use strict qw(vars subs);
62no warnings "uninitialized"; 62no warnings "uninitialized";
63 63
64use Coro::State; 64use Coro::State;
65 65
66use base qw(Coro::State Exporter); 66use base qw(Coro::State Exporter);
67 67
68our $idle; # idle handler 68our $idle; # idle handler
69our $main; # main coroutine 69our $main; # main coroutine
70our $current; # current coroutine 70our $current; # current coroutine
71 71
72our $VERSION = 4.746; 72our $VERSION = 5.0;
73 73
74our @EXPORT = qw(async async_pool cede schedule terminate current unblock_sub); 74our @EXPORT = qw(async async_pool cede schedule terminate current unblock_sub);
75our %EXPORT_TAGS = ( 75our %EXPORT_TAGS = (
76 prio => [qw(PRIO_MAX PRIO_HIGH PRIO_NORMAL PRIO_LOW PRIO_IDLE PRIO_MIN)], 76 prio => [qw(PRIO_MAX PRIO_HIGH PRIO_NORMAL PRIO_LOW PRIO_IDLE PRIO_MIN)],
77); 77);
86coroutines, it is mainly useful to compare again C<$Coro::current>, to see 86coroutines, it is mainly useful to compare again C<$Coro::current>, to see
87whether you are running in the main program or not. 87whether you are running in the main program or not.
88 88
89=cut 89=cut
90 90
91$main = new Coro; 91# $main is now being initialised by Coro::State
92 92
93=item $Coro::current 93=item $Coro::current
94 94
95The coroutine object representing the current coroutine (the last 95The coroutine object representing the current coroutine (the last
96coroutine that the Coro scheduler switched to). The initial value is 96coroutine that the Coro scheduler switched to). The initial value is
97C<$main> (of course). 97C<$Coro::main> (of course).
98 98
99This variable is B<strictly> I<read-only>. You can take copies of the 99This variable is B<strictly> I<read-only>. You can take copies of the
100value stored in it and use it as any other coroutine object, but you must 100value stored in it and use it as any other coroutine object, but you must
101not otherwise modify the variable itself. 101not otherwise modify the variable itself.
102 102
103=cut 103=cut
104
105$main->{desc} = "[main::]";
106
107# maybe some other module used Coro::Specific before...
108$main->{_specific} = $current->{_specific}
109 if $current;
110
111_set_current $main;
112 104
113sub current() { $current } # [DEPRECATED] 105sub current() { $current } # [DEPRECATED]
114 106
115=item $Coro::idle 107=item $Coro::idle
116 108
152 $self->_destroy 144 $self->_destroy
153 or return; 145 or return;
154 146
155 # call all destruction callbacks 147 # call all destruction callbacks
156 $_->(@{$self->{_status}}) 148 $_->(@{$self->{_status}})
157 for @{(delete $self->{_on_destroy}) || []}; 149 for @{ delete $self->{_on_destroy} || [] };
158} 150}
159 151
160# this coroutine is necessary because a coroutine 152# this coroutine is necessary because a coroutine
161# cannot destroy itself. 153# cannot destroy itself.
162my @destroy; 154our @destroy;
163my $manager; 155our $manager;
164 156
165$manager = new Coro sub { 157$manager = new Coro sub {
166 while () { 158 while () {
167 (shift @destroy)->_cancel 159 (shift @destroy)->_cancel
168 while @destroy; 160 while @destroy;
169 161
170 &schedule; 162 &schedule;
171 } 163 }
172}; 164};
173$manager->desc ("[coro manager]"); 165$manager->{desc} = "[coro manager]";
174$manager->prio (PRIO_MAX); 166$manager->prio (PRIO_MAX);
175 167
176=back 168=back
177 169
178=head2 SIMPLE COROUTINE CREATION 170=head2 SIMPLE COROUTINE CREATION
220Similar to C<async>, but uses a coroutine pool, so you should not call 212Similar to C<async>, but uses a coroutine pool, so you should not call
221terminate or join on it (although you are allowed to), and you get a 213terminate or join on it (although you are allowed to), and you get a
222coroutine that might have executed other code already (which can be good 214coroutine that might have executed other code already (which can be good
223or bad :). 215or bad :).
224 216
225On the plus side, this function is faster than creating (and destroying) 217On the plus side, this function is about twice as fast as creating (and
226a completely new coroutine, so if you need a lot of generic coroutines in 218destroying) a completely new coroutine, so if you need a lot of generic
227quick successsion, use C<async_pool>, not C<async>. 219coroutines in quick successsion, use C<async_pool>, not C<async>.
228 220
229The code block is executed in an C<eval> context and a warning will be 221The code block is executed in an C<eval> context and a warning will be
230issued in case of an exception instead of terminating the program, as 222issued in case of an exception instead of terminating the program, as
231C<async> does. As the coroutine is being reused, stuff like C<on_destroy> 223C<async> does. As the coroutine is being reused, stuff like C<on_destroy>
232will not work in the expected way, unless you call terminate or cancel, 224will not work in the expected way, unless you call terminate or cancel,
235 227
236The priority will be reset to C<0> after each run, tracing will be 228The priority will be reset to C<0> after each run, tracing will be
237disabled, the description will be reset and the default output filehandle 229disabled, the description will be reset and the default output filehandle
238gets restored, so you can change all these. Otherwise the coroutine will 230gets restored, so you can change all these. Otherwise the coroutine will
239be re-used "as-is": most notably if you change other per-coroutine global 231be re-used "as-is": most notably if you change other per-coroutine global
240stuff such as C<$/> you I<must needs> to revert that change, which is most 232stuff such as C<$/> you I<must needs> revert that change, which is most
241simply done by using local as in: C< local $/ >. 233simply done by using local as in: C<< local $/ >>.
242 234
243The pool size is limited to C<8> idle coroutines (this can be adjusted by 235The idle pool size is limited to C<8> idle coroutines (this can be
244changing $Coro::POOL_SIZE), and there can be as many non-idle coros as 236adjusted by changing $Coro::POOL_SIZE), but there can be as many non-idle
245required. 237coros as required.
246 238
247If you are concerned about pooled coroutines growing a lot because a 239If you are concerned about pooled coroutines growing a lot because a
248single C<async_pool> used a lot of stackspace you can e.g. C<async_pool 240single C<async_pool> used a lot of stackspace you can e.g. C<async_pool
249{ terminate }> once per second or so to slowly replenish the pool. In 241{ terminate }> once per second or so to slowly replenish the pool. In
250addition to that, when the stacks used by a handler grows larger than 16kb 242addition to that, when the stacks used by a handler grows larger than 16kb
255our $POOL_SIZE = 8; 247our $POOL_SIZE = 8;
256our $POOL_RSS = 16 * 1024; 248our $POOL_RSS = 16 * 1024;
257our @async_pool; 249our @async_pool;
258 250
259sub pool_handler { 251sub pool_handler {
260 my $cb;
261
262 while () { 252 while () {
263 eval { 253 eval {
264 while () { 254 &{&_pool_handler} while 1;
265 _pool_1 $cb;
266 &$cb;
267 _pool_2 $cb;
268 &schedule;
269 }
270 }; 255 };
271 256
272 if ($@) {
273 last if $@ eq "\3async_pool terminate\2\n";
274 warn $@; 257 warn $@ if $@;
275 }
276 } 258 }
277}
278
279sub async_pool(&@) {
280 # this is also inlined into the unlock_scheduler
281 my $coro = (pop @async_pool) || new Coro \&pool_handler;
282
283 $coro->{_invoke} = [@_];
284 $coro->ready;
285
286 $coro
287} 259}
288 260
289=back 261=back
290 262
291=head2 STATIC METHODS 263=head2 STATIC METHODS
313>> on that once some event happens, and last you call C<schedule> to put 285>> on that once some event happens, and last you call C<schedule> to put
314yourself to sleep. Note that a lot of things can wake your coroutine up, 286yourself to sleep. Note that a lot of things can wake your coroutine up,
315so you need to check whether the event indeed happened, e.g. by storing the 287so you need to check whether the event indeed happened, e.g. by storing the
316status in a variable. 288status in a variable.
317 289
318The canonical way to wait on external events is this: 290See B<HOW TO WAIT FOR A CALLBACK>, below, for some ways to wait for callbacks.
319
320 {
321 # remember current coroutine
322 my $current = $Coro::current;
323
324 # register a hypothetical event handler
325 on_event_invoke sub {
326 # wake up sleeping coroutine
327 $current->ready;
328 undef $current;
329 };
330
331 # call schedule until event occurred.
332 # in case we are woken up for other reasons
333 # (current still defined), loop.
334 Coro::schedule while $current;
335 }
336 291
337=item cede 292=item cede
338 293
339"Cede" to other coroutines. This function puts the current coroutine into 294"Cede" to other coroutines. This function puts the current coroutine into
340the ready queue and calls C<schedule>, which has the effect of giving 295the ready queue and calls C<schedule>, which has the effect of giving
365program calls this function, there will be some one-time resource leak. 320program calls this function, there will be some one-time resource leak.
366 321
367=cut 322=cut
368 323
369sub terminate { 324sub terminate {
370 $current->cancel (@_); 325 $current->{_status} = [@_];
326 push @destroy, $current;
327 $manager->ready;
328 do { &schedule } while 1;
371} 329}
372 330
373sub killall { 331sub killall {
374 for (Coro::State::list) { 332 for (Coro::State::list) {
375 $_->cancel 333 $_->cancel
396See C<async> and C<Coro::State::new> for additional info about the 354See C<async> and C<Coro::State::new> for additional info about the
397coroutine environment. 355coroutine environment.
398 356
399=cut 357=cut
400 358
401sub _run_coro { 359sub _terminate {
402 terminate &{+shift}; 360 terminate &{+shift};
403}
404
405sub new {
406 my $class = shift;
407
408 $class->SUPER::new (\&_run_coro, @_)
409} 361}
410 362
411=item $success = $coroutine->ready 363=item $success = $coroutine->ready
412 364
413Put the given coroutine into the end of its ready queue (there is one 365Put the given coroutine into the end of its ready queue (there is one
430 382
431=cut 383=cut
432 384
433sub cancel { 385sub cancel {
434 my $self = shift; 386 my $self = shift;
435 $self->{_status} = [@_];
436 387
437 if ($current == $self) { 388 if ($current == $self) {
438 push @destroy, $self; 389 terminate @_;
439 $manager->ready;
440 &schedule while 1;
441 } else { 390 } else {
391 $self->{_status} = [@_];
442 $self->_cancel; 392 $self->_cancel;
443 } 393 }
444} 394}
395
396=item $coroutine->schedule_to
397
398Puts the current coroutine to sleep (like C<Coro::schedule>), but instead
399of continuing with the next coro from the ready queue, always switch to
400the given coroutine object (regardless of priority etc.). The readyness
401state of that coroutine isn't changed.
402
403This is an advanced method for special cases - I'd love to hear about any
404uses for this one.
405
406=item $coroutine->cede_to
407
408Like C<schedule_to>, but puts the current coroutine into the ready
409queue. This has the effect of temporarily switching to the given
410coroutine, and continuing some time later.
411
412This is an advanced method for special cases - I'd love to hear about any
413uses for this one.
414
415=item $coroutine->throw ([$scalar])
416
417If C<$throw> is specified and defined, it will be thrown as an exception
418inside the coroutine at the next convenient point in time. Otherwise
419clears the exception object.
420
421Coro will check for the exception each time a schedule-like-function
422returns, i.e. after each C<schedule>, C<cede>, C<< Coro::Semaphore->down
423>>, C<< Coro::Handle->readable >> and so on. Most of these functions
424detect this case and return early in case an exception is pending.
425
426The exception object will be thrown "as is" with the specified scalar in
427C<$@>, i.e. if it is a string, no line number or newline will be appended
428(unlike with C<die>).
429
430This can be used as a softer means than C<cancel> to ask a coroutine to
431end itself, although there is no guarantee that the exception will lead to
432termination, and if the exception isn't caught it might well end the whole
433program.
434
435You might also think of C<throw> as being the moral equivalent of
436C<kill>ing a coroutine with a signal (in this case, a scalar).
445 437
446=item $coroutine->join 438=item $coroutine->join
447 439
448Wait until the coroutine terminates and return any values given to the 440Wait until the coroutine terminates and return any values given to the
449C<terminate> or C<cancel> functions. C<join> can be called concurrently 441C<terminate> or C<cancel> functions. C<join> can be called concurrently
511higher values mean lower priority, just as in unix). 503higher values mean lower priority, just as in unix).
512 504
513=item $olddesc = $coroutine->desc ($newdesc) 505=item $olddesc = $coroutine->desc ($newdesc)
514 506
515Sets (or gets in case the argument is missing) the description for this 507Sets (or gets in case the argument is missing) the description for this
516coroutine. This is just a free-form string you can associate with a coroutine. 508coroutine. This is just a free-form string you can associate with a
509coroutine.
517 510
518This method simply sets the C<< $coroutine->{desc} >> member to the given string. You 511This method simply sets the C<< $coroutine->{desc} >> member to the given
519can modify this member directly if you wish. 512string. You can modify this member directly if you wish.
520
521=item $coroutine->throw ([$scalar])
522
523If C<$throw> is specified and defined, it will be thrown as an exception
524inside the coroutine at the next convinient point in time (usually after
525it gains control at the next schedule/transfer/cede). Otherwise clears the
526exception object.
527
528The exception object will be thrown "as is" with the specified scalar in
529C<$@>, i.e. if it is a string, no line number or newline will be appended
530(unlike with C<die>).
531
532This can be used as a softer means than C<cancel> to ask a coroutine to
533end itself, although there is no guarentee that the exception will lead to
534termination, and if the exception isn't caught it might well end the whole
535program.
536 513
537=cut 514=cut
538 515
539sub desc { 516sub desc {
540 my $old = $_[0]{desc}; 517 my $old = $_[0]{desc};
632# return immediately and can be reused) and because we cannot cede 609# return immediately and can be reused) and because we cannot cede
633# inside an event callback. 610# inside an event callback.
634our $unblock_scheduler = new Coro sub { 611our $unblock_scheduler = new Coro sub {
635 while () { 612 while () {
636 while (my $cb = pop @unblock_queue) { 613 while (my $cb = pop @unblock_queue) {
637 # this is an inlined copy of async_pool 614 &async_pool (@$cb);
638 my $coro = (pop @async_pool) || new Coro \&pool_handler;
639 615
640 $coro->{_invoke} = $cb;
641 $coro->ready;
642 cede; # for short-lived callbacks, this reduces pressure on the coro pool 616 # for short-lived callbacks, this reduces pressure on the coro pool
617 # as the chance is very high that the async_poll coro will be back
618 # in the idle state when cede returns
619 cede;
643 } 620 }
644 schedule; # sleep well 621 schedule; # sleep well
645 } 622 }
646}; 623};
647$unblock_scheduler->desc ("[unblock_sub scheduler]"); 624$unblock_scheduler->{desc} = "[unblock_sub scheduler]";
648 625
649sub unblock_sub(&) { 626sub unblock_sub(&) {
650 my $cb = shift; 627 my $cb = shift;
651 628
652 sub { 629 sub {
653 unshift @unblock_queue, [$cb, @_]; 630 unshift @unblock_queue, [$cb, @_];
654 $unblock_scheduler->ready; 631 $unblock_scheduler->ready;
655 } 632 }
656} 633}
657 634
635=item $cb = Coro::rouse_cb
636
637Create and return a "rouse callback". That's a code reference that, when
638called, will save its arguments and notify the owner coroutine of the
639callback.
640
641See the next function.
642
643=item @args = Coro::rouse_wait [$cb]
644
645Wait for the specified rouse callback (or the last one tht was created in
646this coroutine).
647
648As soon as the callback is invoked (or when the calback was invoked before
649C<rouse_wait>), it will return a copy of the arguments originally passed
650to the rouse callback.
651
652See the section B<HOW TO WAIT FOR A CALLBACK> for an actual usage example.
653
658=back 654=back
659 655
660=cut 656=cut
661 657
6621; 6581;
663 659
660=head1 HOW TO WAIT FOR A CALLBACK
661
662It is very common for a coroutine to wait for some callback to be
663called. This occurs naturally when you use coroutines in an otherwise
664event-based program, or when you use event-based libraries.
665
666These typically register a callback for some event, and call that callback
667when the event occured. In a coroutine, however, you typically want to
668just wait for the event, simplyifying things.
669
670For example C<< AnyEvent->child >> registers a callback to be called when
671a specific child has exited:
672
673 my $child_watcher = AnyEvent->child (pid => $pid, cb => sub { ... });
674
675But from withina coroutine, you often just want to write this:
676
677 my $status = wait_for_child $pid;
678
679Coro offers two functions specifically designed to make this easy,
680C<Coro::rouse_cb> and C<Coro::rouse_wait>.
681
682The first function, C<rouse_cb>, generates and returns a callback that,
683when invoked, will save it's arguments and notify the coroutine that
684created the callback.
685
686The second function, C<rouse_wait>, waits for the callback to be called
687(by calling C<schedule> to go to sleep) and returns the arguments
688originally passed to the callback.
689
690Using these functions, it becomes easy to write the C<wait_for_child>
691function mentioned above:
692
693 sub wait_for_child($) {
694 my ($pid) = @_;
695
696 my $watcher = AnyEvent->child (pid => $pid, cb => Coro::rouse_cb);
697
698 my ($rpid, $rstatus) = Coro::rouse_wait;
699 $rstatus
700 }
701
702In the case where C<rouse_cb> and C<rouse_wait> are not flexible enough,
703you can roll your own, using C<schedule>:
704
705 sub wait_for_child($) {
706 my ($pid) = @_;
707
708 # store the current coroutine in $current,
709 # and provide result variables for the closure passed to ->child
710 my $current = $Coro::current;
711 my ($done, $rstatus);
712
713 # pass a closure to ->child
714 my $watcher = AnyEvent->child (pid => $pid, cb => sub {
715 $rstatus = $_[1]; # remember rstatus
716 $done = 1; # mark $rstatus as valud
717 });
718
719 # wait until the closure has been called
720 schedule while !$done;
721
722 $rstatus
723 }
724
725
664=head1 BUGS/LIMITATIONS 726=head1 BUGS/LIMITATIONS
727
728=over 4
729
730=item fork with pthread backend
731
732When Coro is compiled using the pthread backend (which isn't recommended
733but required on many BSDs as their libcs are completely broken), then
734coroutines will not survive a fork. There is no known workaround except to
735fix your libc and use a saner backend.
736
737=item perl process emulation ("threads")
665 738
666This module is not perl-pseudo-thread-safe. You should only ever use this 739This module is not perl-pseudo-thread-safe. You should only ever use this
667module from the same thread (this requirement might be removed in the 740module from the same thread (this requirement might be removed in the
668future to allow per-thread schedulers, but Coro::State does not yet allow 741future to allow per-thread schedulers, but Coro::State does not yet allow
669this). I recommend disabling thread support and using processes, as this 742this). I recommend disabling thread support and using processes, as having
670is much faster and uses less memory. 743the windows process emulation enabled under unix roughly halves perl
744performance, even when not used.
745
746=item coroutine switching not signal safe
747
748You must not switch to another coroutine from within a signal handler
749(only relevant with %SIG - most event libraries provide safe signals).
750
751That means you I<MUST NOT> call any function that might "block" the
752current coroutine - C<cede>, C<schedule> C<< Coro::Semaphore->down >> or
753anything that calls those. Everything else, including calling C<ready>,
754works.
755
756=back
757
671 758
672=head1 SEE ALSO 759=head1 SEE ALSO
673 760
674Event-Loop integration: L<Coro::AnyEvent>, L<Coro::EV>, L<Coro::Event>. 761Event-Loop integration: L<Coro::AnyEvent>, L<Coro::EV>, L<Coro::Event>.
675 762

Diff Legend

Removed lines
+ Added lines
< Changed lines
> Changed lines