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

Comparing Coro/Coro.pm (file contents):
Revision 1.103 by root, Thu Jan 4 20:14:19 2007 UTC vs.
Revision 1.114 by root, Wed Jan 24 16:22:08 2007 UTC

… …
50 50
51our $idle; # idle handler 51our $idle; # idle handler
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.5';
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
… …
198 } 1,2,3,4; 198 } 1,2,3,4;
199 199
200=cut 200=cut
201 201
202sub async(&@) { 202sub async(&@) {
203 my $pid = new Coro @_; 203 my $coro = new Coro @_;
204 $pid->ready; 204 $coro->ready;
205 $pid 205 $coro
206}
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 eval {
239 my ($cb, @arg) = @{ delete $current->{_invoke} or return };
240 $cb->(@arg);
241 };
242 warn $@ if $@;
243
244 last if @pool >= $POOL_SIZE;
245 push @pool, $current;
246
247 $current->prio (0);
248 schedule;
249 }
250}
251
252sub async_pool(&@) {
253 # this is also inlined into the unlock_scheduler
254 my $coro = (pop @pool or new Coro \&pool_handler);
255
256 $coro->{_invoke} = [@_];
257 $coro->ready;
258
259 $coro
206} 260}
207 261
208=item schedule 262=item schedule
209 263
210Calls the scheduler. Please note that the current coroutine will not be put 264Calls the scheduler. Please note that the current coroutine will not be put
… …
235 289
236"Cede" to other coroutines. This function puts the current coroutine into the 290"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 291ready queue and calls C<schedule>, which has the effect of giving up the
238current "timeslice" to other coroutines of the same or higher priority. 292current "timeslice" to other coroutines of the same or higher priority.
239 293
294Returns true if at least one coroutine switch has happened.
295
240=item Coro::cede_notself 296=item Coro::cede_notself
241 297
242Works like cede, but is not exported by default and will cede to any 298Works like cede, but is not exported by default and will cede to any
243coroutine, regardless of priority, once. 299coroutine, regardless of priority, once.
300
301Returns true if at least one coroutine switch has happened.
244 302
245=item terminate [arg...] 303=item terminate [arg...]
246 304
247Terminates the current coroutine with the given status values (see L<cancel>). 305Terminates the current coroutine with the given status values (see L<cancel>).
248 306
… …
461In short: simply use C<unblock_sub { ... }> instead of C<sub { ... }> when 519In short: simply use C<unblock_sub { ... }> instead of C<sub { ... }> when
462creating event callbacks that want to block. 520creating event callbacks that want to block.
463 521
464=cut 522=cut
465 523
466our @unblock_pool;
467our @unblock_queue; 524our @unblock_queue;
468our $UNBLOCK_POOL_SIZE = 2;
469 525
470sub unblock_handler_ { 526# we create a special coro because we want to cede,
471 while () { 527# to reduce pressure on the coro pool (because most callbacks
472 my ($cb, @arg) = @{ delete $Coro::current->{arg} }; 528# return immediately and can be reused) and because we cannot cede
473 $cb->(@arg); 529# 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 { 530our $unblock_scheduler = async {
482 while () { 531 while () {
483 while (my $cb = pop @unblock_queue) { 532 while (my $cb = pop @unblock_queue) {
533 # this is an inlined copy of async_pool
484 my $handler = (pop @unblock_pool or new Coro \&unblock_handler_); 534 my $coro = (pop @pool or new Coro \&pool_handler);
485 $handler->{arg} = $cb; 535
536 $coro->{_invoke} = $cb;
486 $handler->ready; 537 $coro->ready;
487 cede; 538 cede; # for short-lived callbacks, this reduces pressure on the coro pool
488 } 539 }
489 540 schedule; # sleep well
490 schedule;
491 } 541 }
492}; 542};
493 543
494sub unblock_sub(&) { 544sub unblock_sub(&) {
495 my $cb = shift; 545 my $cb = shift;
496 546
497 sub { 547 sub {
498 push @unblock_queue, [$cb, @_]; 548 unshift @unblock_queue, [$cb, @_];
499 $unblock_scheduler->ready; 549 $unblock_scheduler->ready;
500 } 550 }
501} 551}
502 552
503=back 553=back

Diff Legend

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