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

Comparing Coro/Coro.pm (file contents):
Revision 1.202 by root, Tue Sep 30 17:12:34 2008 UTC vs.
Revision 1.228 by root, Thu Nov 20 03:14:49 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.8; 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->throw ([$scalar])
397
398If C<$throw> is specified and defined, it will be thrown as an exception
399inside the coroutine at the next convenient point in time. Otherwise
400clears the exception object.
401
402Coro will check for the exception each time a schedule-like-function
403returns, i.e. after each C<schedule>, C<cede>, C<< Coro::Semaphore->down
404>>, C<< Coro::Handle->readable >> and so on. Most of these functions
405detect this case and return early in case an exception is pending.
406
407The exception object will be thrown "as is" with the specified scalar in
408C<$@>, i.e. if it is a string, no line number or newline will be appended
409(unlike with C<die>).
410
411This can be used as a softer means than C<cancel> to ask a coroutine to
412end itself, although there is no guarantee that the exception will lead to
413termination, and if the exception isn't caught it might well end the whole
414program.
415
416You might also think of C<throw> as being the moral equivalent of
417C<kill>ing a coroutine with a signal (in this case, a scalar).
445 418
446=item $coroutine->join 419=item $coroutine->join
447 420
448Wait until the coroutine terminates and return any values given to the 421Wait until the coroutine terminates and return any values given to the
449C<terminate> or C<cancel> functions. C<join> can be called concurrently 422C<terminate> or C<cancel> functions. C<join> can be called concurrently
511higher values mean lower priority, just as in unix). 484higher values mean lower priority, just as in unix).
512 485
513=item $olddesc = $coroutine->desc ($newdesc) 486=item $olddesc = $coroutine->desc ($newdesc)
514 487
515Sets (or gets in case the argument is missing) the description for this 488Sets (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. 489coroutine. This is just a free-form string you can associate with a
490coroutine.
517 491
518This method simply sets the C<< $coroutine->{desc} >> member to the given string. You 492This method simply sets the C<< $coroutine->{desc} >> member to the given
519can modify this member directly if you wish. 493string. 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 494
537=cut 495=cut
538 496
539sub desc { 497sub desc {
540 my $old = $_[0]{desc}; 498 my $old = $_[0]{desc};
632# return immediately and can be reused) and because we cannot cede 590# return immediately and can be reused) and because we cannot cede
633# inside an event callback. 591# inside an event callback.
634our $unblock_scheduler = new Coro sub { 592our $unblock_scheduler = new Coro sub {
635 while () { 593 while () {
636 while (my $cb = pop @unblock_queue) { 594 while (my $cb = pop @unblock_queue) {
637 # this is an inlined copy of async_pool 595 &async_pool (@$cb);
638 my $coro = (pop @async_pool) || new Coro \&pool_handler;
639 596
640 $coro->{_invoke} = $cb;
641 $coro->ready;
642 cede; # for short-lived callbacks, this reduces pressure on the coro pool 597 # for short-lived callbacks, this reduces pressure on the coro pool
598 # as the chance is very high that the async_poll coro will be back
599 # in the idle state when cede returns
600 cede;
643 } 601 }
644 schedule; # sleep well 602 schedule; # sleep well
645 } 603 }
646}; 604};
647$unblock_scheduler->desc ("[unblock_sub scheduler]"); 605$unblock_scheduler->{desc} = "[unblock_sub scheduler]";
648 606
649sub unblock_sub(&) { 607sub unblock_sub(&) {
650 my $cb = shift; 608 my $cb = shift;
651 609
652 sub { 610 sub {
653 unshift @unblock_queue, [$cb, @_]; 611 unshift @unblock_queue, [$cb, @_];
654 $unblock_scheduler->ready; 612 $unblock_scheduler->ready;
655 } 613 }
656} 614}
657 615
616=item $cb = Coro::rouse_cb
617
618Create and return a "rouse callback". That's a code reference that, when
619called, will save its arguments and notify the owner coroutine of the
620callback.
621
622See the next function.
623
624=item @args = Coro::rouse_wait [$cb]
625
626Wait for the specified rouse callback (or the last one tht was created in
627this coroutine).
628
629As soon as the callback is invoked (or when the calback was invoked before
630C<rouse_wait>), it will return a copy of the arguments originally passed
631to the rouse callback.
632
633See the section B<HOW TO WAIT FOR A CALLBACK> for an actual usage example.
634
658=back 635=back
659 636
660=cut 637=cut
661 638
6621; 6391;
663 640
641=head1 HOW TO WAIT FOR A CALLBACK
642
643It is very common for a coroutine to wait for some callback to be
644called. This occurs naturally when you use coroutines in an otherwise
645event-based program, or when you use event-based libraries.
646
647These typically register a callback for some event, and call that callback
648when the event occured. In a coroutine, however, you typically want to
649just wait for the event, simplyifying things.
650
651For example C<< AnyEvent->child >> registers a callback to be called when
652a specific child has exited:
653
654 my $child_watcher = AnyEvent->child (pid => $pid, cb => sub { ... });
655
656But from withina coroutine, you often just want to write this:
657
658 my $status = wait_for_child $pid;
659
660Coro offers two functions specifically designed to make this easy,
661C<Coro::rouse_cb> and C<Coro::rouse_wait>.
662
663The first function, C<rouse_cb>, generates and returns a callback that,
664when invoked, will save it's arguments and notify the coroutine that
665created the callback.
666
667The second function, C<rouse_wait>, waits for the callback to be called
668(by calling C<schedule> to go to sleep) and returns the arguments
669originally passed to the callback.
670
671Using these functions, it becomes easy to write the C<wait_for_child>
672function mentioned above:
673
674 sub wait_for_child($) {
675 my ($pid) = @_;
676
677 my $watcher = AnyEvent->child (pid => $pid, cb => Coro::rouse_cb);
678
679 my ($rpid, $rstatus) = Coro::rouse_wait;
680 $rstatus
681 }
682
683In the case where C<rouse_cb> and C<rouse_wait> are not flexible enough,
684you can roll your own, using C<schedule>:
685
686 sub wait_for_child($) {
687 my ($pid) = @_;
688
689 # store the current coroutine in $current,
690 # and provide result variables for the closure passed to ->child
691 my $current = $Coro::current;
692 my ($done, $rstatus);
693
694 # pass a closure to ->child
695 my $watcher = AnyEvent->child (pid => $pid, cb => sub {
696 $rstatus = $_[1]; # remember rstatus
697 $done = 1; # mark $rstatus as valud
698 });
699
700 # wait until the closure has been called
701 schedule while !$done;
702
703 $rstatus
704 }
705
706
664=head1 BUGS/LIMITATIONS 707=head1 BUGS/LIMITATIONS
708
709=over 4
710
711=item fork with pthread backend
712
713When Coro is compiled using the pthread backend (which isn't recommended
714but required on many BSDs as their libcs are completely broken), then
715coroutines will not survive a fork. There is no known workaround except to
716fix your libc and use a saner backend.
717
718=item perl process emulation ("threads")
665 719
666This module is not perl-pseudo-thread-safe. You should only ever use this 720This 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 721module from the same thread (this requirement might be removed in the
668future to allow per-thread schedulers, but Coro::State does not yet allow 722future to allow per-thread schedulers, but Coro::State does not yet allow
669this). I recommend disabling thread support and using processes, as this 723this). I recommend disabling thread support and using processes, as having
670is much faster and uses less memory. 724the windows process emulation enabled under unix roughly halves perl
725performance, even when not used.
726
727=item coroutine switching not signal safe
728
729You must not switch to another coroutine from within a signal handler
730(only relevant with %SIG - most event libraries provide safe signals).
731
732That means you I<MUST NOT> call any function that might "block" the
733current coroutine - C<cede>, C<schedule> C<< Coro::Semaphore->down >> or
734anything that calls those. Everything else, including calling C<ready>,
735works.
736
737=back
738
671 739
672=head1 SEE ALSO 740=head1 SEE ALSO
673 741
674Event-Loop integration: L<Coro::AnyEvent>, L<Coro::EV>, L<Coro::Event>. 742Event-Loop integration: L<Coro::AnyEvent>, L<Coro::EV>, L<Coro::Event>.
675 743

Diff Legend

Removed lines
+ Added lines
< Changed lines
> Changed lines