ViewVC Help
View File | Revision Log | Show Annotations | Download File
/cvs/Coro/Coro.pm
Revision: 1.152
Committed: Sun Oct 7 13:53:37 2007 UTC (16 years, 7 months ago) by root
Branch: MAIN
Changes since 1.151: +28 -15 lines
Log Message:
*** empty log message ***

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