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

Comparing Coro/Coro.pm (file contents):
Revision 1.104 by root, Thu Jan 4 23:49:27 2007 UTC vs.
Revision 1.107 by root, Fri Jan 5 18:25:51 2007 UTC

… …
52our $main; # main coroutine 52our $main; # main coroutine
53our $current; # current coroutine 53our $current; # current coroutine
54 54
55our $VERSION = '3.3'; 55our $VERSION = '3.3';
56 56
57our @EXPORT = qw(async cede schedule terminate current unblock_sub); 57our @EXPORT = qw(async async_pool cede schedule terminate current unblock_sub);
58our %EXPORT_TAGS = ( 58our %EXPORT_TAGS = (
59 prio => [qw(PRIO_MAX PRIO_HIGH PRIO_NORMAL PRIO_LOW PRIO_IDLE PRIO_MIN)], 59 prio => [qw(PRIO_MAX PRIO_HIGH PRIO_NORMAL PRIO_LOW PRIO_IDLE PRIO_MIN)],
60); 60);
61our @EXPORT_OK = (@{$EXPORT_TAGS{prio}}, qw(nready)); 61our @EXPORT_OK = (@{$EXPORT_TAGS{prio}}, qw(nready));
62 62
… …
203 my $coro = new Coro @_; 203 my $coro = new Coro @_;
204 $coro->ready; 204 $coro->ready;
205 $coro 205 $coro
206} 206}
207 207
208=item async_pool { ... } [@args...]
209
210Similar to C<async>, but uses a coroutine pool, so you should not call
211terminate or join (although you are allowed to), and you get a coroutine
212that might have executed other code already (which can be good or bad :).
213
214Also, the block is executed in an C<eval> context and a warning will be
215issued in case of an exception instead of terminating the program, as C<async> does.
216
217The priority will be reset to C<0> after each job, otherwise the coroutine
218will be re-used "as-is".
219
220The pool size is limited to 8 idle coroutines (this can be adjusted by
221changing $Coro::POOL_SIZE), and there can be as many non-idle coros as
222required.
223
224If you are concerned about pooled coroutines growing a lot because a
225single C<async_pool> used a lot of stackspace you can e.g. C<async_pool {
226terminate }> once per second or so to slowly replenish the pool.
227
228=cut
229
230our $POOL_SIZE = 8;
231our @pool;
232
233sub pool_handler {
234 while () {
235 my ($cb, @arg) = @{ delete $current->{_invoke} };
236
237 eval {
238 $cb->(@arg);
239 };
240 warn $@ if $@;
241
242 last if @pool >= $POOL_SIZE;
243 push @pool, $current;
244
245 $current->prio (0);
246 schedule;
247 }
248}
249
250sub async_pool(&@) {
251 # this is also inlined into the unlock_scheduler
252 my $coro = (pop @pool or new Coro \&pool_handler);
253
254 $coro->{_invoke} = [@_];
255 $coro->ready;
256
257 $coro
258}
259
208=item schedule 260=item schedule
209 261
210Calls the scheduler. Please note that the current coroutine will not be put 262Calls the scheduler. Please note that the current coroutine will not be put
211into the ready queue, so calling this function usually means you will 263into the ready queue, so calling this function usually means you will
212never be called again unless something else (e.g. an event handler) calls 264never be called again unless something else (e.g. an event handler) calls
… …
235 287
236"Cede" to other coroutines. This function puts the current coroutine into the 288"Cede" to other coroutines. This function puts the current coroutine into the
237ready queue and calls C<schedule>, which has the effect of giving up the 289ready queue and calls C<schedule>, which has the effect of giving up the
238current "timeslice" to other coroutines of the same or higher priority. 290current "timeslice" to other coroutines of the same or higher priority.
239 291
292Returns true if at least one coroutine switch has happened.
293
240=item Coro::cede_notself 294=item Coro::cede_notself
241 295
242Works like cede, but is not exported by default and will cede to any 296Works like cede, but is not exported by default and will cede to any
243coroutine, regardless of priority, once. 297coroutine, regardless of priority, once.
298
299Returns true if at least one coroutine switch has happened.
244 300
245=item terminate [arg...] 301=item terminate [arg...]
246 302
247Terminates the current coroutine with the given status values (see L<cancel>). 303Terminates the current coroutine with the given status values (see L<cancel>).
248 304
… …
461In short: simply use C<unblock_sub { ... }> instead of C<sub { ... }> when 517In short: simply use C<unblock_sub { ... }> instead of C<sub { ... }> when
462creating event callbacks that want to block. 518creating event callbacks that want to block.
463 519
464=cut 520=cut
465 521
466our @unblock_pool;
467our @unblock_queue; 522our @unblock_queue;
468our $UNBLOCK_POOL_SIZE = 2;
469 523
470sub unblock_handler_ { 524# we create a special coro because we want to cede,
471 while () { 525# to reduce pressure on the coro pool (because most callbacks
472 my ($cb, @arg) = @{ delete $Coro::current->{arg} }; 526# return immediately and can be reused) and because we cannot cede
473 $cb->(@arg); 527# inside an event callback.
474
475 last if @unblock_pool >= $UNBLOCK_POOL_SIZE;
476 push @unblock_pool, $Coro::current;
477 schedule;
478 }
479}
480
481our $unblock_scheduler = async { 528our $unblock_scheduler = async {
482 while () { 529 while () {
483 while (my $cb = pop @unblock_queue) { 530 while (my $cb = pop @unblock_queue) {
531 # this is an inlined copy of async_pool
484 my $handler = (pop @unblock_pool or new Coro \&unblock_handler_); 532 my $coro = (pop @pool or new Coro \&pool_handler);
485 $handler->{arg} = $cb; 533
534 $coro->{_invoke} = $cb;
486 $handler->ready; 535 $coro->ready;
487 cede; 536 cede; # for short-lived callbacks, this reduces pressure on the coro pool
488 } 537 }
489 538 schedule; # sleep well
490 schedule;
491 } 539 }
492}; 540};
493 541
494sub unblock_sub(&) { 542sub unblock_sub(&) {
495 my $cb = shift; 543 my $cb = shift;
496 544
497 sub { 545 sub {
498 push @unblock_queue, [$cb, @_]; 546 unshift @unblock_queue, [$cb, @_];
499 $unblock_scheduler->ready; 547 $unblock_scheduler->ready;
500 } 548 }
501} 549}
502 550
503=back 551=back

Diff Legend

– Removed lines
+ Added lines
< Changed lines
> Changed lines