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.108 by root, Fri Jan 5 20:00:49 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
216C<async> does. As the coroutine is being reused, stuff like C<on_destroy>
217will not work in the expected way, unless you call terminate or cancel,
218which somehow defeats the purpose of pooling.
219
220The priority will be reset to C<0> after each job, otherwise the coroutine
221will be re-used "as-is".
222
223The pool size is limited to 8 idle coroutines (this can be adjusted by
224changing $Coro::POOL_SIZE), and there can be as many non-idle coros as
225required.
226
227If you are concerned about pooled coroutines growing a lot because a
228single C<async_pool> used a lot of stackspace you can e.g. C<async_pool {
229terminate }> once per second or so to slowly replenish the pool.
230
231=cut
232
233our $POOL_SIZE = 8;
234our @pool;
235
236sub pool_handler {
237 while () {
238 my ($cb, @arg) = @{ delete $current->{_invoke} };
239
240 eval {
241 $cb->(@arg);
242 };
243 warn $@ if $@;
244
245 last if @pool >= $POOL_SIZE;
246 push @pool, $current;
247
248 $current->prio (0);
249 schedule;
250 }
251}
252
253sub async_pool(&@) {
254 # this is also inlined into the unlock_scheduler
255 my $coro = (pop @pool or new Coro \&pool_handler);
256
257 $coro->{_invoke} = [@_];
258 $coro->ready;
259
260 $coro
261}
262
208=item schedule 263=item schedule
209 264
210Calls the scheduler. Please note that the current coroutine will not be put 265Calls the scheduler. Please note that the current coroutine will not be put
211into the ready queue, so calling this function usually means you will 266into the ready queue, so calling this function usually means you will
212never be called again unless something else (e.g. an event handler) calls 267never be called again unless something else (e.g. an event handler) calls
… …
235 290
236"Cede" to other coroutines. This function puts the current coroutine into the 291"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 292ready queue and calls C<schedule>, which has the effect of giving up the
238current "timeslice" to other coroutines of the same or higher priority. 293current "timeslice" to other coroutines of the same or higher priority.
239 294
295Returns true if at least one coroutine switch has happened.
296
240=item Coro::cede_notself 297=item Coro::cede_notself
241 298
242Works like cede, but is not exported by default and will cede to any 299Works like cede, but is not exported by default and will cede to any
243coroutine, regardless of priority, once. 300coroutine, regardless of priority, once.
301
302Returns true if at least one coroutine switch has happened.
244 303
245=item terminate [arg...] 304=item terminate [arg...]
246 305
247Terminates the current coroutine with the given status values (see L<cancel>). 306Terminates the current coroutine with the given status values (see L<cancel>).
248 307
… …
461In short: simply use C<unblock_sub { ... }> instead of C<sub { ... }> when 520In short: simply use C<unblock_sub { ... }> instead of C<sub { ... }> when
462creating event callbacks that want to block. 521creating event callbacks that want to block.
463 522
464=cut 523=cut
465 524
466our @unblock_pool;
467our @unblock_queue; 525our @unblock_queue;
468our $UNBLOCK_POOL_SIZE = 2;
469 526
470sub unblock_handler_ { 527# we create a special coro because we want to cede,
471 while () { 528# to reduce pressure on the coro pool (because most callbacks
472 my ($cb, @arg) = @{ delete $Coro::current->{arg} }; 529# return immediately and can be reused) and because we cannot cede
473 $cb->(@arg); 530# 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 { 531our $unblock_scheduler = async {
482 while () { 532 while () {
483 while (my $cb = pop @unblock_queue) { 533 while (my $cb = pop @unblock_queue) {
534 # this is an inlined copy of async_pool
484 my $handler = (pop @unblock_pool or new Coro \&unblock_handler_); 535 my $coro = (pop @pool or new Coro \&pool_handler);
485 $handler->{arg} = $cb; 536
537 $coro->{_invoke} = $cb;
486 $handler->ready; 538 $coro->ready;
487 cede; 539 cede; # for short-lived callbacks, this reduces pressure on the coro pool
488 } 540 }
489 541 schedule; # sleep well
490 schedule;
491 } 542 }
492}; 543};
493 544
494sub unblock_sub(&) { 545sub unblock_sub(&) {
495 my $cb = shift; 546 my $cb = shift;
496 547
497 sub { 548 sub {
498 push @unblock_queue, [$cb, @_]; 549 unshift @unblock_queue, [$cb, @_];
499 $unblock_scheduler->ready; 550 $unblock_scheduler->ready;
500 } 551 }
501} 552}
502 553
503=back 554=back

Diff Legend

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