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

Comparing Coro/Coro.pm (file contents):
Revision 1.119 by root, Wed Mar 28 14:24:17 2007 UTC vs.
Revision 1.151 by root, Sat Oct 6 19:25:00 2007 UTC

20 20
21=head1 DESCRIPTION 21=head1 DESCRIPTION
22 22
23This module collection manages coroutines. Coroutines are similar 23This module collection manages coroutines. Coroutines are similar
24to threads but don't run in parallel at the same time even on SMP 24to threads but don't run in parallel at the same time even on SMP
25machines. The specific flavor of coroutine use din this module also 25machines. The specific flavor of coroutine used in this module also
26guarentees you that it will not switch between coroutines unless 26guarantees you that it will not switch between coroutines unless
27necessary, at easily-identified points in your program, so locking and 27necessary, at easily-identified points in your program, so locking and
28parallel access are rarely an issue, making coroutine programming much 28parallel access are rarely an issue, making coroutine programming much
29safer than threads programming. 29safer than threads programming.
30 30
31(Perl, however, does not natively support real threads but instead does a 31(Perl, however, does not natively support real threads but instead does a
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.55'; 55our $VERSION = '4.03';
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);
108 108
109The current coroutine (the last coroutine switched to). The initial value 109The current coroutine (the last coroutine switched to). The initial value
110is C<$main> (of course). 110is C<$main> (of course).
111 111
112This variable is B<strictly> I<read-only>. It is provided for performance 112This variable is B<strictly> I<read-only>. It is provided for performance
113reasons. If performance is not essentiel 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
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.
189 191
190Calling C<exit> in a coroutine will not work correctly, so do not do that. 192See the C<Coro::State::new> constructor for info about the coroutine
193environment.
191 194
192When the coroutine dies, the program will exit, just as in the main 195Calling C<exit> in a coroutine will do the same as calling exit outside
193program. 196the coroutine. Likewise, when the coroutine dies, the program will exit,
197just as it would in the main program.
194 198
195 # create a new coroutine that just prints its arguments 199 # create a new coroutine that just prints its arguments
196 async { 200 async {
197 print "@_\n"; 201 print "@_\n";
198 } 1,2,3,4; 202 } 1,2,3,4;
215issued in case of an exception instead of terminating the program, as 219issued 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> 220C<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, 221will not work in the expected way, unless you call terminate or cancel,
218which somehow defeats the purpose of pooling. 222which somehow defeats the purpose of pooling.
219 223
220The priority will be reset to C<0> after each job, otherwise the coroutine 224The priority will be reset to C<0> after each job, tracing will be
221will be re-used "as-is". 225disabled, the description will be reset and the default output filehandle
226gets restored, so you can change alkl these. Otherwise the coroutine will
227be re-used "as-is": most notably if you change other per-coroutine global
228stuff such as C<$/> you need to revert that change, which is most simply
229done by using local as in C< local $/ >.
222 230
223The pool size is limited to 8 idle coroutines (this can be adjusted by 231The 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 232changing $Coro::POOL_SIZE), and there can be as many non-idle coros as
225required. 233required.
226 234
227If you are concerned about pooled coroutines growing a lot because a 235If 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 { 236single 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. 237{ terminate }> once per second or so to slowly replenish the pool. In
238addition to that, when the stacks used by a handler grows larger than 16kb
239(adjustable with $Coro::POOL_RSS) it will also exit.
230 240
231=cut 241=cut
232 242
233our $POOL_SIZE = 8; 243our $POOL_SIZE = 8;
244our $POOL_RSS = 16 * 1024;
234our @pool; 245our @async_pool;
235 246
236sub pool_handler { 247sub pool_handler {
248 my $cb;
249
237 while () { 250 while () {
238 eval { 251 eval {
239 my ($cb, @arg) = @{ delete $current->{_invoke} or return }; 252 while () {
240 $cb->(@arg); 253 _pool_1 $cb;
254 &$cb;
255 _pool_2 $cb;
256 &schedule;
257 }
241 }; 258 };
259
260 last if $@ eq "\3async_pool terminate\2\n";
242 warn $@ if $@; 261 warn $@ if $@;
243
244 last if @pool >= $POOL_SIZE;
245 push @pool, $current;
246
247 $current->save (Coro::State::SAVE_DEF);
248 $current->prio (0);
249 schedule;
250 } 262 }
251} 263}
252 264
253sub async_pool(&@) { 265sub async_pool(&@) {
254 # this is also inlined into the unlock_scheduler 266 # this is also inlined into the unlock_scheduler
255 my $coro = (pop @pool or new Coro \&pool_handler); 267 my $coro = (pop @async_pool) || new Coro \&pool_handler;
256 268
257 $coro->{_invoke} = [@_]; 269 $coro->{_invoke} = [@_];
258 $coro->ready; 270 $coro->ready;
259 271
260 $coro 272 $coro
278 # wake up sleeping coroutine 290 # wake up sleeping coroutine
279 $current->ready; 291 $current->ready;
280 undef $current; 292 undef $current;
281 }; 293 };
282 294
283 # call schedule until event occured. 295 # call schedule until event occurred.
284 # in case we are woken up for other reasons 296 # in case we are woken up for other reasons
285 # (current still defined), loop. 297 # (current still defined), loop.
286 Coro::schedule while $current; 298 Coro::schedule while $current;
287 } 299 }
288 300
303 315
304=item terminate [arg...] 316=item terminate [arg...]
305 317
306Terminates the current coroutine with the given status values (see L<cancel>). 318Terminates the current coroutine with the given status values (see L<cancel>).
307 319
320=item killall
321
322Kills/terminates/cancels all coroutines except the currently running
323one. This is useful after a fork, either in the child or the parent, as
324usually only one of them should inherit the running coroutines.
325
308=cut 326=cut
309 327
310sub terminate { 328sub terminate {
311 $current->cancel (@_); 329 $current->cancel (@_);
330}
331
332sub killall {
333 for (Coro::State::list) {
334 $_->cancel
335 if $_ != $current && UNIVERSAL::isa $_, "Coro";
336 }
312} 337}
313 338
314=back 339=back
315 340
316# dynamic methods 341# dynamic methods
326Create a new coroutine and return it. When the sub returns the coroutine 351Create a new coroutine and return it. When the sub returns the coroutine
327automatically terminates as if C<terminate> with the returned values were 352automatically terminates as if C<terminate> with the returned values were
328called. To make the coroutine run you must first put it into the ready queue 353called. To make the coroutine run you must first put it into the ready queue
329by calling the ready method. 354by calling the ready method.
330 355
331Calling C<exit> in a coroutine will not work correctly, so do not do that. 356See C<async> and C<Coro::State::new> for additional info about the
357coroutine environment.
332 358
333=cut 359=cut
334 360
335sub _run_coro { 361sub _run_coro {
336 terminate &{+shift}; 362 terminate &{+shift};
360 386
361=cut 387=cut
362 388
363sub cancel { 389sub cancel {
364 my $self = shift; 390 my $self = shift;
365 $self->{status} = [@_]; 391 $self->{_status} = [@_];
366 392
367 if ($current == $self) { 393 if ($current == $self) {
368 push @destroy, $self; 394 push @destroy, $self;
369 $manager->ready; 395 $manager->ready;
370 &schedule while 1; 396 &schedule while 1;
374} 400}
375 401
376=item $coroutine->join 402=item $coroutine->join
377 403
378Wait until the coroutine terminates and return any values given to the 404Wait until the coroutine terminates and return any values given to the
379C<terminate> or C<cancel> functions. C<join> can be called multiple times 405C<terminate> or C<cancel> functions. C<join> can be called concurrently
380from multiple coroutine. 406from multiple coroutines.
381 407
382=cut 408=cut
383 409
384sub join { 410sub join {
385 my $self = shift; 411 my $self = shift;
386 412
387 unless ($self->{status}) { 413 unless ($self->{_status}) {
388 my $current = $current; 414 my $current = $current;
389 415
390 push @{$self->{destroy_cb}}, sub { 416 push @{$self->{_on_destroy}}, sub {
391 $current->ready; 417 $current->ready;
392 undef $current; 418 undef $current;
393 }; 419 };
394 420
395 &schedule while $current; 421 &schedule while $current;
396 } 422 }
397 423
398 wantarray ? @{$self->{status}} : $self->{status}[0]; 424 wantarray ? @{$self->{_status}} : $self->{_status}[0];
399} 425}
400 426
401=item $coroutine->on_destroy (\&cb) 427=item $coroutine->on_destroy (\&cb)
402 428
403Registers a callback that is called when this coroutine gets destroyed, 429Registers a callback that is called when this coroutine gets destroyed,
407=cut 433=cut
408 434
409sub on_destroy { 435sub on_destroy {
410 my ($self, $cb) = @_; 436 my ($self, $cb) = @_;
411 437
412 push @{ $self->{destroy_cb} }, $cb; 438 push @{ $self->{_on_destroy} }, $cb;
413} 439}
414 440
415=item $oldprio = $coroutine->prio ($newprio) 441=item $oldprio = $coroutine->prio ($newprio)
416 442
417Sets (or gets, if the argument is missing) the priority of the 443Sets (or gets, if the argument is missing) the priority of the
442=item $olddesc = $coroutine->desc ($newdesc) 468=item $olddesc = $coroutine->desc ($newdesc)
443 469
444Sets (or gets in case the argument is missing) the description for this 470Sets (or gets in case the argument is missing) the description for this
445coroutine. This is just a free-form string you can associate with a coroutine. 471coroutine. This is just a free-form string you can associate with a coroutine.
446 472
473This method simply sets the C<< $coroutine->{desc} >> member to the given string. You
474can modify this member directly if you wish.
475
476=item $coroutine->throw ([$scalar])
477
478If C<$throw> is specified and defined, it will be thrown as an exception
479inside the coroutine at the next convinient point in time (usually after
480it gains control at the next schedule/transfer/cede). Otherwise clears the
481exception object.
482
483The exception object will be thrown "as is" with the specified scalar in
484C<$@>, i.e. if it is a string, no line number or newline will be appended
485(unlike with C<die>).
486
487This can be used as a softer means than C<cancel> to ask a coroutine to
488end itself, although there is no guarentee that the exception will lead to
489termination, and if the exception isn't caught it might well end the whole
490program.
491
447=cut 492=cut
448 493
449sub desc { 494sub desc {
450 my $old = $_[0]{desc}; 495 my $old = $_[0]{desc};
451 $_[0]{desc} = $_[1] if @_ > 1; 496 $_[0]{desc} = $_[1] if @_ > 1;
459=over 4 504=over 4
460 505
461=item Coro::nready 506=item Coro::nready
462 507
463Returns the number of coroutines that are currently in the ready state, 508Returns the number of coroutines that are currently in the ready state,
464i.e. that can be swicthed to. The value C<0> means that the only runnable 509i.e. that can be switched to. The value C<0> means that the only runnable
465coroutine is the currently running one, so C<cede> would have no effect, 510coroutine is the currently running one, so C<cede> would have no effect,
466and C<schedule> would cause a deadlock unless there is an idle handler 511and C<schedule> would cause a deadlock unless there is an idle handler
467that wakes up some coroutines. 512that wakes up some coroutines.
468 513
469=item my $guard = Coro::guard { ... } 514=item my $guard = Coro::guard { ... }
505This utility function takes a BLOCK or code reference and "unblocks" it, 550This utility function takes a BLOCK or code reference and "unblocks" it,
506returning the new coderef. This means that the new coderef will return 551returning the new coderef. This means that the new coderef will return
507immediately without blocking, returning nothing, while the original code 552immediately without blocking, returning nothing, while the original code
508ref will be called (with parameters) from within its own coroutine. 553ref will be called (with parameters) from within its own coroutine.
509 554
510The reason this fucntion exists is that many event libraries (such as the 555The reason this function exists is that many event libraries (such as the
511venerable L<Event|Event> module) are not coroutine-safe (a weaker form 556venerable L<Event|Event> module) are not coroutine-safe (a weaker form
512of thread-safety). This means you must not block within event callbacks, 557of thread-safety). This means you must not block within event callbacks,
513otherwise you might suffer from crashes or worse. 558otherwise you might suffer from crashes or worse.
514 559
515This function allows your callbacks to block by executing them in another 560This function allows your callbacks to block by executing them in another
526 571
527# we create a special coro because we want to cede, 572# we create a special coro because we want to cede,
528# to reduce pressure on the coro pool (because most callbacks 573# to reduce pressure on the coro pool (because most callbacks
529# return immediately and can be reused) and because we cannot cede 574# return immediately and can be reused) and because we cannot cede
530# inside an event callback. 575# inside an event callback.
531our $unblock_scheduler = async { 576our $unblock_scheduler = new Coro sub {
532 while () { 577 while () {
533 while (my $cb = pop @unblock_queue) { 578 while (my $cb = pop @unblock_queue) {
534 # this is an inlined copy of async_pool 579 # this is an inlined copy of async_pool
535 my $coro = (pop @pool or new Coro \&pool_handler); 580 my $coro = (pop @async_pool) || new Coro \&pool_handler;
536 581
537 $coro->{_invoke} = $cb; 582 $coro->{_invoke} = $cb;
538 $coro->ready; 583 $coro->ready;
539 cede; # for short-lived callbacks, this reduces pressure on the coro pool 584 cede; # for short-lived callbacks, this reduces pressure on the coro pool
540 } 585 }
541 schedule; # sleep well 586 schedule; # sleep well
542 } 587 }
543}; 588};
589$unblock_scheduler->desc ("[unblock_sub scheduler]");
544 590
545sub unblock_sub(&) { 591sub unblock_sub(&) {
546 my $cb = shift; 592 my $cb = shift;
547 593
548 sub { 594 sub {
561 607
562 - you must make very sure that no coro is still active on global 608 - you must make very sure that no coro is still active on global
563 destruction. very bad things might happen otherwise (usually segfaults). 609 destruction. very bad things might happen otherwise (usually segfaults).
564 610
565 - this module is not thread-safe. You should only ever use this module 611 - this module is not thread-safe. You should only ever use this module
566 from the same thread (this requirement might be losened in the future 612 from the same thread (this requirement might be loosened in the future
567 to allow per-thread schedulers, but Coro::State does not yet allow 613 to allow per-thread schedulers, but Coro::State does not yet allow
568 this). 614 this).
569 615
570=head1 SEE ALSO 616=head1 SEE ALSO
571 617
572Support/Utility: L<Coro::Cont>, L<Coro::Specific>, L<Coro::State>, L<Coro::Util>. 618Support/Utility: L<Coro::Specific>, L<Coro::State>, L<Coro::Util>.
573 619
574Locking/IPC: L<Coro::Signal>, L<Coro::Channel>, L<Coro::Semaphore>, L<Coro::SemaphoreSet>, L<Coro::RWLock>. 620Locking/IPC: L<Coro::Signal>, L<Coro::Channel>, L<Coro::Semaphore>, L<Coro::SemaphoreSet>, L<Coro::RWLock>.
575 621
576Event/IO: L<Coro::Timer>, L<Coro::Event>, L<Coro::Handle>, L<Coro::Socket>, L<Coro::Select>. 622Event/IO: L<Coro::Timer>, L<Coro::Event>, L<Coro::Handle>, L<Coro::Socket>, L<Coro::Select>.
577 623

Diff Legend

Removed lines
+ Added lines
< Changed lines
> Changed lines