ViewVC Help
View File | Revision Log | Show Annotations | Download File
/cvs/Coro/Coro.pm
Revision: 1.221
Committed: Tue Nov 18 05:51:38 2008 UTC (15 years, 6 months ago) by root
Branch: MAIN
Changes since 1.220: +1 -1 lines
Log Message:
many fixes

File Contents

# Content
1 =head1 NAME
2
3 Coro - coroutine process abstraction
4
5 =head1 SYNOPSIS
6
7 use Coro;
8
9 async {
10 # some asynchronous thread of execution
11 print "2\n";
12 cede; # yield back to main
13 print "4\n";
14 };
15 print "1\n";
16 cede; # yield to coroutine
17 print "3\n";
18 cede; # and again
19
20 # use locking
21 use Coro::Semaphore;
22 my $lock = new Coro::Semaphore;
23 my $locked;
24
25 $lock->down;
26 $locked = 1;
27 $lock->up;
28
29 =head1 DESCRIPTION
30
31 This module collection manages coroutines. Coroutines are similar to
32 threads but don't (in general) run in parallel at the same time even
33 on SMP machines. The specific flavor of coroutine used in this module
34 also guarantees you that it will not switch between coroutines unless
35 necessary, at easily-identified points in your program, so locking and
36 parallel access are rarely an issue, making coroutine programming much
37 safer and easier than threads programming.
38
39 Unlike a normal perl program, however, coroutines allow you to have
40 multiple running interpreters that share data, which is especially useful
41 to code pseudo-parallel processes and for event-based programming, such as
42 multiple HTTP-GET requests running concurrently. See L<Coro::AnyEvent> to
43 learn more.
44
45 Coroutines are also useful because Perl has no support for threads (the so
46 called "threads" that perl offers are nothing more than the (bad) process
47 emulation coming from the Windows platform: On standard operating systems
48 they serve no purpose whatsoever, except by making your programs slow and
49 making them use a lot of memory. Best disable them when building perl, or
50 aks your software vendor/distributor to do it for you).
51
52 In this module, coroutines are defined as "callchain + lexical variables +
53 @_ + $_ + $@ + $/ + C stack), that is, a coroutine has its own callchain,
54 its own set of lexicals and its own set of perls most important global
55 variables (see L<Coro::State> for more configuration).
56
57 =cut
58
59 package Coro;
60
61 use strict qw(vars subs);
62 no warnings "uninitialized";
63
64 use Coro::State;
65
66 use base qw(Coro::State Exporter);
67
68 our $idle; # idle handler
69 our $main; # main coroutine
70 our $current; # current coroutine
71
72 our $VERSION = 5.0;
73
74 our @EXPORT = qw(async async_pool cede schedule terminate current unblock_sub);
75 our %EXPORT_TAGS = (
76 prio => [qw(PRIO_MAX PRIO_HIGH PRIO_NORMAL PRIO_LOW PRIO_IDLE PRIO_MIN)],
77 );
78 our @EXPORT_OK = (@{$EXPORT_TAGS{prio}}, qw(nready));
79
80 =over 4
81
82 =item $Coro::main
83
84 This variable stores the coroutine object that represents the main
85 program. While you cna C<ready> it and do most other things you can do to
86 coroutines, it is mainly useful to compare again C<$Coro::current>, to see
87 whether you are running in the main program or not.
88
89 =cut
90
91 # $main is now being initialised by Coro::State
92
93 =item $Coro::current
94
95 The coroutine object representing the current coroutine (the last
96 coroutine that the Coro scheduler switched to). The initial value is
97 C<$Coro::main> (of course).
98
99 This variable is B<strictly> I<read-only>. You can take copies of the
100 value stored in it and use it as any other coroutine object, but you must
101 not otherwise modify the variable itself.
102
103 =cut
104
105 sub current() { $current } # [DEPRECATED]
106
107 =item $Coro::idle
108
109 This variable is mainly useful to integrate Coro into event loops. It is
110 usually better to rely on L<Coro::AnyEvent> or LC<Coro::EV>, as this is
111 pretty low-level functionality.
112
113 This variable stores a callback that is called whenever the scheduler
114 finds no ready coroutines to run. The default implementation prints
115 "FATAL: deadlock detected" and exits, because the program has no other way
116 to continue.
117
118 This hook is overwritten by modules such as C<Coro::Timer> and
119 C<Coro::AnyEvent> to wait on an external event that hopefully wake up a
120 coroutine so the scheduler can run it.
121
122 Note that the callback I<must not>, under any circumstances, block
123 the current coroutine. Normally, this is achieved by having an "idle
124 coroutine" that calls the event loop and then blocks again, and then
125 readying that coroutine in the idle handler.
126
127 See L<Coro::Event> or L<Coro::AnyEvent> for examples of using this
128 technique.
129
130 Please note that if your callback recursively invokes perl (e.g. for event
131 handlers), then it must be prepared to be called recursively itself.
132
133 =cut
134
135 $idle = sub {
136 require Carp;
137 Carp::croak ("FATAL: deadlock detected");
138 };
139
140 sub _cancel {
141 my ($self) = @_;
142
143 # free coroutine data and mark as destructed
144 $self->_destroy
145 or return;
146
147 # call all destruction callbacks
148 $_->(@{$self->{_status}})
149 for @{ delete $self->{_on_destroy} || [] };
150 }
151
152 # this coroutine is necessary because a coroutine
153 # cannot destroy itself.
154 my @destroy;
155 my $manager;
156
157 $manager = new Coro sub {
158 while () {
159 (shift @destroy)->_cancel
160 while @destroy;
161
162 &schedule;
163 }
164 };
165 $manager->{desc} = "[coro manager]";
166 $manager->prio (PRIO_MAX);
167
168 =back
169
170 =head2 SIMPLE COROUTINE CREATION
171
172 =over 4
173
174 =item async { ... } [@args...]
175
176 Create a new coroutine and return it's coroutine object (usually
177 unused). The coroutine will be put into the ready queue, so
178 it will start running automatically on the next scheduler run.
179
180 The first argument is a codeblock/closure that should be executed in the
181 coroutine. When it returns argument returns the coroutine is automatically
182 terminated.
183
184 The remaining arguments are passed as arguments to the closure.
185
186 See the C<Coro::State::new> constructor for info about the coroutine
187 environment in which coroutines are executed.
188
189 Calling C<exit> in a coroutine will do the same as calling exit outside
190 the coroutine. Likewise, when the coroutine dies, the program will exit,
191 just as it would in the main program.
192
193 If you do not want that, you can provide a default C<die> handler, or
194 simply avoid dieing (by use of C<eval>).
195
196 Example: Create a new coroutine that just prints its arguments.
197
198 async {
199 print "@_\n";
200 } 1,2,3,4;
201
202 =cut
203
204 sub async(&@) {
205 my $coro = new Coro @_;
206 $coro->ready;
207 $coro
208 }
209
210 =item async_pool { ... } [@args...]
211
212 Similar to C<async>, but uses a coroutine pool, so you should not call
213 terminate or join on it (although you are allowed to), and you get a
214 coroutine that might have executed other code already (which can be good
215 or bad :).
216
217 On the plus side, this function is faster than creating (and destroying)
218 a completly new coroutine, so if you need a lot of generic coroutines in
219 quick successsion, use C<async_pool>, not C<async>.
220
221 The code block is executed in an C<eval> context and a warning will be
222 issued in case of an exception instead of terminating the program, as
223 C<async> does. As the coroutine is being reused, stuff like C<on_destroy>
224 will not work in the expected way, unless you call terminate or cancel,
225 which somehow defeats the purpose of pooling (but is fine in the
226 exceptional case).
227
228 The priority will be reset to C<0> after each run, tracing will be
229 disabled, the description will be reset and the default output filehandle
230 gets restored, so you can change all these. Otherwise the coroutine will
231 be re-used "as-is": most notably if you change other per-coroutine global
232 stuff such as C<$/> you I<must needs> revert that change, which is most
233 simply done by using local as in: C<< local $/ >>.
234
235 The idle pool size is limited to C<8> idle coroutines (this can be
236 adjusted by changing $Coro::POOL_SIZE), but there can be as many non-idle
237 coros as required.
238
239 If you are concerned about pooled coroutines growing a lot because a
240 single C<async_pool> used a lot of stackspace you can e.g. C<async_pool
241 { terminate }> once per second or so to slowly replenish the pool. In
242 addition to that, when the stacks used by a handler grows larger than 16kb
243 (adjustable via $Coro::POOL_RSS) it will also be destroyed.
244
245 =cut
246
247 our $POOL_SIZE = 8;
248 our $POOL_RSS = 16 * 1024;
249 our @async_pool;
250
251 sub pool_handler {
252 my $cb;
253
254 while () {
255 eval {
256 while () {
257 _pool_1 $cb;
258 &$cb;
259 _pool_2 $cb;
260 &schedule;
261 }
262 };
263
264 if ($@) {
265 last if $@ eq "\3async_pool terminate\2\n";
266 warn $@;
267 }
268 }
269 }
270
271 sub async_pool(&@) {
272 # this is also inlined into the unblock_scheduler
273 my $coro = (pop @async_pool) || new Coro \&pool_handler;
274
275 $coro->{_invoke} = [@_];
276 $coro->ready;
277
278 $coro
279 }
280
281 =back
282
283 =head2 STATIC METHODS
284
285 Static methods are actually functions that operate on the current coroutine.
286
287 =over 4
288
289 =item schedule
290
291 Calls the scheduler. The scheduler will find the next coroutine that is
292 to be run from the ready queue and switches to it. The next coroutine
293 to be run is simply the one with the highest priority that is longest
294 in its ready queue. If there is no coroutine ready, it will clal the
295 C<$Coro::idle> hook.
296
297 Please note that the current coroutine will I<not> be put into the ready
298 queue, so calling this function usually means you will never be called
299 again unless something else (e.g. an event handler) calls C<< ->ready >>,
300 thus waking you up.
301
302 This makes C<schedule> I<the> generic method to use to block the current
303 coroutine and wait for events: first you remember the current coroutine in
304 a variable, then arrange for some callback of yours to call C<< ->ready
305 >> on that once some event happens, and last you call C<schedule> to put
306 yourself to sleep. Note that a lot of things can wake your coroutine up,
307 so you need to check whether the event indeed happened, e.g. by storing the
308 status in a variable.
309
310 The canonical way to wait on external events is this:
311
312 {
313 # remember current coroutine
314 my $current = $Coro::current;
315
316 # register a hypothetical event handler
317 on_event_invoke sub {
318 # wake up sleeping coroutine
319 $current->ready;
320 undef $current;
321 };
322
323 # call schedule until event occurred.
324 # in case we are woken up for other reasons
325 # (current still defined), loop.
326 Coro::schedule while $current;
327 }
328
329 =item cede
330
331 "Cede" to other coroutines. This function puts the current coroutine into
332 the ready queue and calls C<schedule>, which has the effect of giving
333 up the current "timeslice" to other coroutines of the same or higher
334 priority. Once your coroutine gets its turn again it will automatically be
335 resumed.
336
337 This function is often called C<yield> in other languages.
338
339 =item Coro::cede_notself
340
341 Works like cede, but is not exported by default and will cede to I<any>
342 coroutine, regardless of priority. This is useful sometimes to ensure
343 progress is made.
344
345 =item terminate [arg...]
346
347 Terminates the current coroutine with the given status values (see L<cancel>).
348
349 =item killall
350
351 Kills/terminates/cancels all coroutines except the currently running
352 one. This is useful after a fork, either in the child or the parent, as
353 usually only one of them should inherit the running coroutines.
354
355 Note that while this will try to free some of the main programs resources,
356 you cannot free all of them, so if a coroutine that is not the main
357 program calls this function, there will be some one-time resource leak.
358
359 =cut
360
361 sub terminate {
362 $current->cancel (@_);
363 }
364
365 sub killall {
366 for (Coro::State::list) {
367 $_->cancel
368 if $_ != $current && UNIVERSAL::isa $_, "Coro";
369 }
370 }
371
372 =back
373
374 =head2 COROUTINE METHODS
375
376 These are the methods you can call on coroutine objects (or to create
377 them).
378
379 =over 4
380
381 =item new Coro \&sub [, @args...]
382
383 Create a new coroutine and return it. When the sub returns, the coroutine
384 automatically terminates as if C<terminate> with the returned values were
385 called. To make the coroutine run you must first put it into the ready
386 queue by calling the ready method.
387
388 See C<async> and C<Coro::State::new> for additional info about the
389 coroutine environment.
390
391 =cut
392
393 sub _run_coro {
394 terminate &{+shift};
395 }
396
397 sub new {
398 my $class = shift;
399
400 $class->SUPER::new (\&_run_coro, @_)
401 }
402
403 =item $success = $coroutine->ready
404
405 Put the given coroutine into the end of its ready queue (there is one
406 queue for each priority) and return true. If the coroutine is already in
407 the ready queue, do nothing and return false.
408
409 This ensures that the scheduler will resume this coroutine automatically
410 once all the coroutines of higher priority and all coroutines of the same
411 priority that were put into the ready queue earlier have been resumed.
412
413 =item $is_ready = $coroutine->is_ready
414
415 Return whether the coroutine is currently the ready queue or not,
416
417 =item $coroutine->cancel (arg...)
418
419 Terminates the given coroutine and makes it return the given arguments as
420 status (default: the empty list). Never returns if the coroutine is the
421 current coroutine.
422
423 =cut
424
425 sub cancel {
426 my $self = shift;
427 $self->{_status} = [@_];
428
429 if ($current == $self) {
430 push @destroy, $self;
431 $manager->ready;
432 &schedule while 1;
433 } else {
434 $self->_cancel;
435 }
436 }
437
438 =item $coroutine->throw ([$scalar])
439
440 If C<$throw> is specified and defined, it will be thrown as an exception
441 inside the coroutine at the next convenient point in time (usually after
442 it gains control at the next schedule/transfer/cede). Otherwise clears the
443 exception object.
444
445 The exception object will be thrown "as is" with the specified scalar in
446 C<$@>, i.e. if it is a string, no line number or newline will be appended
447 (unlike with C<die>).
448
449 This can be used as a softer means than C<cancel> to ask a coroutine to
450 end itself, although there is no guarantee that the exception will lead to
451 termination, and if the exception isn't caught it might well end the whole
452 program.
453
454 You might also think of C<throw> as being the moral equivalent of
455 C<kill>ing a coroutine with a signal (in this case, a scalar).
456
457 =item $coroutine->join
458
459 Wait until the coroutine terminates and return any values given to the
460 C<terminate> or C<cancel> functions. C<join> can be called concurrently
461 from multiple coroutines, and all will be resumed and given the status
462 return once the C<$coroutine> terminates.
463
464 =cut
465
466 sub join {
467 my $self = shift;
468
469 unless ($self->{_status}) {
470 my $current = $current;
471
472 push @{$self->{_on_destroy}}, sub {
473 $current->ready;
474 undef $current;
475 };
476
477 &schedule while $current;
478 }
479
480 wantarray ? @{$self->{_status}} : $self->{_status}[0];
481 }
482
483 =item $coroutine->on_destroy (\&cb)
484
485 Registers a callback that is called when this coroutine gets destroyed,
486 but before it is joined. The callback gets passed the terminate arguments,
487 if any, and I<must not> die, under any circumstances.
488
489 =cut
490
491 sub on_destroy {
492 my ($self, $cb) = @_;
493
494 push @{ $self->{_on_destroy} }, $cb;
495 }
496
497 =item $oldprio = $coroutine->prio ($newprio)
498
499 Sets (or gets, if the argument is missing) the priority of the
500 coroutine. Higher priority coroutines get run before lower priority
501 coroutines. Priorities are small signed integers (currently -4 .. +3),
502 that you can refer to using PRIO_xxx constants (use the import tag :prio
503 to get then):
504
505 PRIO_MAX > PRIO_HIGH > PRIO_NORMAL > PRIO_LOW > PRIO_IDLE > PRIO_MIN
506 3 > 1 > 0 > -1 > -3 > -4
507
508 # set priority to HIGH
509 current->prio(PRIO_HIGH);
510
511 The idle coroutine ($Coro::idle) always has a lower priority than any
512 existing coroutine.
513
514 Changing the priority of the current coroutine will take effect immediately,
515 but changing the priority of coroutines in the ready queue (but not
516 running) will only take effect after the next schedule (of that
517 coroutine). This is a bug that will be fixed in some future version.
518
519 =item $newprio = $coroutine->nice ($change)
520
521 Similar to C<prio>, but subtract the given value from the priority (i.e.
522 higher values mean lower priority, just as in unix).
523
524 =item $olddesc = $coroutine->desc ($newdesc)
525
526 Sets (or gets in case the argument is missing) the description for this
527 coroutine. This is just a free-form string you can associate with a
528 coroutine.
529
530 This method simply sets the C<< $coroutine->{desc} >> member to the given
531 string. You can modify this member directly if you wish.
532
533 =cut
534
535 sub desc {
536 my $old = $_[0]{desc};
537 $_[0]{desc} = $_[1] if @_ > 1;
538 $old;
539 }
540
541 =back
542
543 =head2 GLOBAL FUNCTIONS
544
545 =over 4
546
547 =item Coro::nready
548
549 Returns the number of coroutines that are currently in the ready state,
550 i.e. that can be switched to by calling C<schedule> directory or
551 indirectly. The value C<0> means that the only runnable coroutine is the
552 currently running one, so C<cede> would have no effect, and C<schedule>
553 would cause a deadlock unless there is an idle handler that wakes up some
554 coroutines.
555
556 =item my $guard = Coro::guard { ... }
557
558 This creates and returns a guard object. Nothing happens until the object
559 gets destroyed, in which case the codeblock given as argument will be
560 executed. This is useful to free locks or other resources in case of a
561 runtime error or when the coroutine gets canceled, as in both cases the
562 guard block will be executed. The guard object supports only one method,
563 C<< ->cancel >>, which will keep the codeblock from being executed.
564
565 Example: set some flag and clear it again when the coroutine gets canceled
566 or the function returns:
567
568 sub do_something {
569 my $guard = Coro::guard { $busy = 0 };
570 $busy = 1;
571
572 # do something that requires $busy to be true
573 }
574
575 =cut
576
577 sub guard(&) {
578 bless \(my $cb = $_[0]), "Coro::guard"
579 }
580
581 sub Coro::guard::cancel {
582 ${$_[0]} = sub { };
583 }
584
585 sub Coro::guard::DESTROY {
586 ${$_[0]}->();
587 }
588
589
590 =item unblock_sub { ... }
591
592 This utility function takes a BLOCK or code reference and "unblocks" it,
593 returning a new coderef. Unblocking means that calling the new coderef
594 will return immediately without blocking, returning nothing, while the
595 original code ref will be called (with parameters) from within another
596 coroutine.
597
598 The reason this function exists is that many event libraries (such as the
599 venerable L<Event|Event> module) are not coroutine-safe (a weaker form
600 of thread-safety). This means you must not block within event callbacks,
601 otherwise you might suffer from crashes or worse. The only event library
602 currently known that is safe to use without C<unblock_sub> is L<EV>.
603
604 This function allows your callbacks to block by executing them in another
605 coroutine where it is safe to block. One example where blocking is handy
606 is when you use the L<Coro::AIO|Coro::AIO> functions to save results to
607 disk, for example.
608
609 In short: simply use C<unblock_sub { ... }> instead of C<sub { ... }> when
610 creating event callbacks that want to block.
611
612 If your handler does not plan to block (e.g. simply sends a message to
613 another coroutine, or puts some other coroutine into the ready queue),
614 there is no reason to use C<unblock_sub>.
615
616 Note that you also need to use C<unblock_sub> for any other callbacks that
617 are indirectly executed by any C-based event loop. For example, when you
618 use a module that uses L<AnyEvent> (and you use L<Coro::AnyEvent>) and it
619 provides callbacks that are the result of some event callback, then you
620 must not block either, or use C<unblock_sub>.
621
622 =cut
623
624 our @unblock_queue;
625
626 # we create a special coro because we want to cede,
627 # to reduce pressure on the coro pool (because most callbacks
628 # return immediately and can be reused) and because we cannot cede
629 # inside an event callback.
630 our $unblock_scheduler = new Coro sub {
631 while () {
632 while (my $cb = pop @unblock_queue) {
633 # this is an inlined copy of async_pool
634 my $coro = (pop @async_pool) || new Coro \&pool_handler;
635
636 $coro->{_invoke} = $cb;
637 $coro->ready;
638 cede; # for short-lived callbacks, this reduces pressure on the coro pool
639 }
640 schedule; # sleep well
641 }
642 };
643 $unblock_scheduler->{desc} = "[unblock_sub scheduler]";
644
645 sub unblock_sub(&) {
646 my $cb = shift;
647
648 sub {
649 unshift @unblock_queue, [$cb, @_];
650 $unblock_scheduler->ready;
651 }
652 }
653
654 =back
655
656 =cut
657
658 1;
659
660 =head1 BUGS/LIMITATIONS
661
662 =over 4
663
664 =item fork with pthread backend
665
666 When Coro is compiled using the pthread backend (which isn't recommended
667 but required on many BSDs as their libcs are completely broken), then
668 coroutines will not survive a fork. There is no known workaround except to
669 fix your libc and use a saner backend.
670
671 =item perl process emulation ("threads")
672
673 This module is not perl-pseudo-thread-safe. You should only ever use this
674 module from the same thread (this requirement might be removed in the
675 future to allow per-thread schedulers, but Coro::State does not yet allow
676 this). I recommend disabling thread support and using processes, as having
677 the windows process emulation enabled under unix roughly halves perl
678 performance, even when not used.
679
680 =item coroutine switching not signal safe
681
682 You must not switch to another coroutine from within a signal handler
683 (only relevant with %SIG - most event libraries provide safe signals).
684
685 That means you I<MUST NOT> call any function that might "block" the
686 current coroutine - C<cede>, C<schedule> C<< Coro::Semaphore->down >> or
687 anything that calls those. Everything else, including calling C<ready>,
688 works.
689
690 =back
691
692
693 =head1 SEE ALSO
694
695 Event-Loop integration: L<Coro::AnyEvent>, L<Coro::EV>, L<Coro::Event>.
696
697 Debugging: L<Coro::Debug>.
698
699 Support/Utility: L<Coro::Specific>, L<Coro::Util>.
700
701 Locking/IPC: L<Coro::Signal>, L<Coro::Channel>, L<Coro::Semaphore>, L<Coro::SemaphoreSet>, L<Coro::RWLock>.
702
703 IO/Timers: L<Coro::Timer>, L<Coro::Handle>, L<Coro::Socket>, L<Coro::AIO>.
704
705 Compatibility: L<Coro::LWP>, L<Coro::BDB>, L<Coro::Storable>, L<Coro::Select>.
706
707 XS API: L<Coro::MakeMaker>.
708
709 Low level Configuration, Coroutine Environment: L<Coro::State>.
710
711 =head1 AUTHOR
712
713 Marc Lehmann <schmorp@schmorp.de>
714 http://home.schmorp.de/
715
716 =cut
717