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

Comparing Coro/Coro.pm (file contents):
Revision 1.125 by root, Fri Apr 27 19:35:58 2007 UTC vs.
Revision 1.145 by root, Wed Oct 3 16:03:17 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.62'; 55our $VERSION = '4.0';
56 56
57our @EXPORT = qw(async async_pool 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);
113reasons. If performance is not essential you are encouraged to use the 113reasons. If performance is not essential you are encouraged to use the
114C<Coro::current> function instead. 114C<Coro::current> function instead.
115 115
116=cut 116=cut
117 117
118$main->{desc} = "[main::]";
119
118# maybe some other module used Coro::Specific before... 120# maybe some other module used Coro::Specific before...
119$main->{specific} = $current->{specific} 121$main->{_specific} = $current->{_specific}
120 if $current; 122 if $current;
121 123
122_set_current $main; 124_set_current $main;
123 125
124sub current() { $current } 126sub current() { $current }
149 # free coroutine data and mark as destructed 151 # free coroutine data and mark as destructed
150 $self->_destroy 152 $self->_destroy
151 or return; 153 or return;
152 154
153 # call all destruction callbacks 155 # call all destruction callbacks
154 $_->(@{$self->{status}}) 156 $_->(@{$self->{_status}})
155 for @{(delete $self->{destroy_cb}) || []}; 157 for @{(delete $self->{_on_destroy}) || []};
156} 158}
157 159
158# this coroutine is necessary because a coroutine 160# this coroutine is necessary because a coroutine
159# cannot destroy itself. 161# cannot destroy itself.
160my @destroy; 162my @destroy;
166 while @destroy; 168 while @destroy;
167 169
168 &schedule; 170 &schedule;
169 } 171 }
170}; 172};
171 173$manager->desc ("[coro manager]");
172$manager->prio (PRIO_MAX); 174$manager->prio (PRIO_MAX);
173 175
174# static methods. not really. 176# static methods. not really.
175 177
176=back 178=back
184=item async { ... } [@args...] 186=item async { ... } [@args...]
185 187
186Create a new asynchronous coroutine and return it's coroutine object 188Create a new asynchronous coroutine and return it's coroutine object
187(usually unused). When the sub returns the new coroutine is automatically 189(usually unused). When the sub returns the new coroutine is automatically
188terminated. 190terminated.
191
192See the C<Coro::State::new> constructor for info about the coroutine
193environment.
189 194
190Calling C<exit> in a coroutine will do the same as calling exit outside 195Calling C<exit> in a coroutine will do the same as calling exit outside
191the coroutine. Likewise, when the coroutine dies, the program will exit, 196the coroutine. Likewise, when the coroutine dies, the program will exit,
192just as it would in the main program. 197just as it would in the main program.
193 198
222The pool size is limited to 8 idle coroutines (this can be adjusted by 227The pool size is limited to 8 idle coroutines (this can be adjusted by
223changing $Coro::POOL_SIZE), and there can be as many non-idle coros as 228changing $Coro::POOL_SIZE), and there can be as many non-idle coros as
224required. 229required.
225 230
226If you are concerned about pooled coroutines growing a lot because a 231If you are concerned about pooled coroutines growing a lot because a
227single C<async_pool> used a lot of stackspace you can e.g. C<async_pool { 232single C<async_pool> used a lot of stackspace you can e.g. C<async_pool
228terminate }> once per second or so to slowly replenish the pool. 233{ terminate }> once per second or so to slowly replenish the pool. In
234addition to that, when the stacks used by a handler grows larger than 16kb
235(adjustable with $Coro::POOL_RSS) it will also exit.
229 236
230=cut 237=cut
231 238
232our $POOL_SIZE = 8; 239our $POOL_SIZE = 8;
240our $POOL_RSS = 16 * 1024;
233our @pool; 241our @async_pool;
234 242
235sub pool_handler { 243sub pool_handler {
244 my $cb;
245
236 while () { 246 while () {
237 eval { 247 eval {
238 my ($cb, @arg) = @{ delete $current->{_invoke} or return }; 248 while () {
239 $cb->(@arg); 249 _pool_1 $cb;
250 &$cb;
251 _pool_2 $cb;
252 &schedule;
253 }
240 }; 254 };
255
256 last if $@ eq "\3terminate\2\n";
241 warn $@ if $@; 257 warn $@ if $@;
242
243 last if @pool >= $POOL_SIZE;
244 push @pool, $current;
245
246 $current->save (Coro::State::SAVE_DEF);
247 $current->prio (0);
248 schedule;
249 } 258 }
250} 259}
251 260
252sub async_pool(&@) { 261sub async_pool(&@) {
253 # this is also inlined into the unlock_scheduler 262 # this is also inlined into the unlock_scheduler
254 my $coro = (pop @pool or new Coro \&pool_handler); 263 my $coro = (pop @async_pool) || new Coro \&pool_handler;
255 264
256 $coro->{_invoke} = [@_]; 265 $coro->{_invoke} = [@_];
257 $coro->ready; 266 $coro->ready;
258 267
259 $coro 268 $coro
302 311
303=item terminate [arg...] 312=item terminate [arg...]
304 313
305Terminates the current coroutine with the given status values (see L<cancel>). 314Terminates the current coroutine with the given status values (see L<cancel>).
306 315
316=item killall
317
318Kills/terminates/cancels all coroutines except the currently running
319one. This is useful after a fork, either in the child or the parent, as
320usually only one of them should inherit the running coroutines.
321
307=cut 322=cut
308 323
309sub terminate { 324sub terminate {
310 $current->cancel (@_); 325 $current->cancel (@_);
326}
327
328sub killall {
329 for (Coro::State::list) {
330 $_->cancel
331 if $_ != $current && UNIVERSAL::isa $_, "Coro";
332 }
311} 333}
312 334
313=back 335=back
314 336
315# dynamic methods 337# dynamic methods
325Create a new coroutine and return it. When the sub returns the coroutine 347Create a new coroutine and return it. When the sub returns the coroutine
326automatically terminates as if C<terminate> with the returned values were 348automatically terminates as if C<terminate> with the returned values were
327called. To make the coroutine run you must first put it into the ready queue 349called. To make the coroutine run you must first put it into the ready queue
328by calling the ready method. 350by calling the ready method.
329 351
330See C<async> for additional discussion. 352See C<async> and C<Coro::State::new> for additional info about the
353coroutine environment.
331 354
332=cut 355=cut
333 356
334sub _run_coro { 357sub _run_coro {
335 terminate &{+shift}; 358 terminate &{+shift};
359 382
360=cut 383=cut
361 384
362sub cancel { 385sub cancel {
363 my $self = shift; 386 my $self = shift;
364 $self->{status} = [@_]; 387 $self->{_status} = [@_];
365 388
366 if ($current == $self) { 389 if ($current == $self) {
367 push @destroy, $self; 390 push @destroy, $self;
368 $manager->ready; 391 $manager->ready;
369 &schedule while 1; 392 &schedule while 1;
373} 396}
374 397
375=item $coroutine->join 398=item $coroutine->join
376 399
377Wait until the coroutine terminates and return any values given to the 400Wait until the coroutine terminates and return any values given to the
378C<terminate> or C<cancel> functions. C<join> can be called multiple times 401C<terminate> or C<cancel> functions. C<join> can be called concurrently
379from multiple coroutine. 402from multiple coroutines.
380 403
381=cut 404=cut
382 405
383sub join { 406sub join {
384 my $self = shift; 407 my $self = shift;
385 408
386 unless ($self->{status}) { 409 unless ($self->{_status}) {
387 my $current = $current; 410 my $current = $current;
388 411
389 push @{$self->{destroy_cb}}, sub { 412 push @{$self->{_on_destroy}}, sub {
390 $current->ready; 413 $current->ready;
391 undef $current; 414 undef $current;
392 }; 415 };
393 416
394 &schedule while $current; 417 &schedule while $current;
395 } 418 }
396 419
397 wantarray ? @{$self->{status}} : $self->{status}[0]; 420 wantarray ? @{$self->{_status}} : $self->{_status}[0];
398} 421}
399 422
400=item $coroutine->on_destroy (\&cb) 423=item $coroutine->on_destroy (\&cb)
401 424
402Registers a callback that is called when this coroutine gets destroyed, 425Registers a callback that is called when this coroutine gets destroyed,
406=cut 429=cut
407 430
408sub on_destroy { 431sub on_destroy {
409 my ($self, $cb) = @_; 432 my ($self, $cb) = @_;
410 433
411 push @{ $self->{destroy_cb} }, $cb; 434 push @{ $self->{_on_destroy} }, $cb;
412} 435}
413 436
414=item $oldprio = $coroutine->prio ($newprio) 437=item $oldprio = $coroutine->prio ($newprio)
415 438
416Sets (or gets, if the argument is missing) the priority of the 439Sets (or gets, if the argument is missing) the priority of the
440 463
441=item $olddesc = $coroutine->desc ($newdesc) 464=item $olddesc = $coroutine->desc ($newdesc)
442 465
443Sets (or gets in case the argument is missing) the description for this 466Sets (or gets in case the argument is missing) the description for this
444coroutine. This is just a free-form string you can associate with a coroutine. 467coroutine. This is just a free-form string you can associate with a coroutine.
468
469This method simply sets the C<< $coroutine->{desc} >> member to the given string. You
470can modify this member directly if you wish.
445 471
446=cut 472=cut
447 473
448sub desc { 474sub desc {
449 my $old = $_[0]{desc}; 475 my $old = $_[0]{desc};
525 551
526# we create a special coro because we want to cede, 552# we create a special coro because we want to cede,
527# to reduce pressure on the coro pool (because most callbacks 553# to reduce pressure on the coro pool (because most callbacks
528# return immediately and can be reused) and because we cannot cede 554# return immediately and can be reused) and because we cannot cede
529# inside an event callback. 555# inside an event callback.
530our $unblock_scheduler = async { 556our $unblock_scheduler = new Coro sub {
531 while () { 557 while () {
532 while (my $cb = pop @unblock_queue) { 558 while (my $cb = pop @unblock_queue) {
533 # this is an inlined copy of async_pool 559 # this is an inlined copy of async_pool
534 my $coro = (pop @pool or new Coro \&pool_handler); 560 my $coro = (pop @async_pool) || new Coro \&pool_handler;
535 561
536 $coro->{_invoke} = $cb; 562 $coro->{_invoke} = $cb;
537 $coro->ready; 563 $coro->ready;
538 cede; # for short-lived callbacks, this reduces pressure on the coro pool 564 cede; # for short-lived callbacks, this reduces pressure on the coro pool
539 } 565 }
540 schedule; # sleep well 566 schedule; # sleep well
541 } 567 }
542}; 568};
569$unblock_scheduler->desc ("[unblock_sub scheduler]");
543 570
544sub unblock_sub(&) { 571sub unblock_sub(&) {
545 my $cb = shift; 572 my $cb = shift;
546 573
547 sub { 574 sub {

Diff Legend

Removed lines
+ Added lines
< Changed lines
> Changed lines