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

Comparing Coro/Coro.pm (file contents):
Revision 1.99 by root, Tue Dec 5 12:50:04 2006 UTC vs.
Revision 1.121 by root, Fri Apr 13 12:56:55 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.11'; 55our $VERSION = '3.56';
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
141$idle = sub { 141$idle = sub {
142 require Carp; 142 require Carp;
143 Carp::croak ("FATAL: deadlock detected"); 143 Carp::croak ("FATAL: deadlock detected");
144}; 144};
145 145
146sub _cancel {
147 my ($self) = @_;
148
149 # free coroutine data and mark as destructed
150 $self->_destroy
151 or return;
152
153 # call all destruction callbacks
154 $_->(@{$self->{status}})
155 for @{(delete $self->{destroy_cb}) || []};
156}
157
146# this coroutine is necessary because a coroutine 158# this coroutine is necessary because a coroutine
147# cannot destroy itself. 159# cannot destroy itself.
148my @destroy; 160my @destroy;
161my $manager;
162
149my $manager; $manager = new Coro sub { 163$manager = new Coro sub {
150 while () { 164 while () {
151 # by overwriting the state object with the manager we destroy it 165 (shift @destroy)->_cancel
152 # while still being able to schedule this coroutine (in case it has
153 # been readied multiple times. this is harmless since the manager
154 # can be called as many times as neccessary and will always
155 # remove itself from the runqueue
156 while (@destroy) { 166 while @destroy;
157 my $coro = pop @destroy;
158 $coro->{status} ||= [];
159 $_->ready for @{delete $coro->{join} || []};
160 167
161 # the next line destroys the coro state, but keeps the
162 # coroutine itself intact (we basically make it a zombie
163 # coroutine that always runs the manager thread, so it's possible
164 # to transfer() to this coroutine).
165 $coro->_clone_state_from ($manager);
166 }
167 &schedule; 168 &schedule;
168 } 169 }
169}; 170};
171
172$manager->prio (PRIO_MAX);
170 173
171# static methods. not really. 174# static methods. not really.
172 175
173=back 176=back
174 177
182 185
183Create a new asynchronous coroutine and return it's coroutine object 186Create a new asynchronous coroutine and return it's coroutine object
184(usually unused). When the sub returns the new coroutine is automatically 187(usually unused). When the sub returns the new coroutine is automatically
185terminated. 188terminated.
186 189
187Calling C<exit> in a coroutine will not work correctly, so do not do that. 190Calling C<exit> in a coroutine will try to do the same as calling exit
191outside the coroutine, but this is experimental. It is best not to rely on
192exit doing any cleanups or even not crashing.
188 193
189When the coroutine dies, the program will exit, just as in the main 194When the coroutine dies, the program will exit, just as in the main
190program. 195program.
191 196
192 # create a new coroutine that just prints its arguments 197 # create a new coroutine that just prints its arguments
195 } 1,2,3,4; 200 } 1,2,3,4;
196 201
197=cut 202=cut
198 203
199sub async(&@) { 204sub async(&@) {
200 my $pid = new Coro @_; 205 my $coro = new Coro @_;
201 $pid->ready; 206 $coro->ready;
202 $pid 207 $coro
208}
209
210=item async_pool { ... } [@args...]
211
212Similar to C<async>, but uses a coroutine pool, so you should not call
213terminate or join (although you are allowed to), and you get a coroutine
214that might have executed other code already (which can be good or bad :).
215
216Also, the block is executed in an C<eval> context and a warning will be
217issued in case of an exception instead of terminating the program, as
218C<async> does. As the coroutine is being reused, stuff like C<on_destroy>
219will not work in the expected way, unless you call terminate or cancel,
220which somehow defeats the purpose of pooling.
221
222The priority will be reset to C<0> after each job, otherwise the coroutine
223will be re-used "as-is".
224
225The pool size is limited to 8 idle coroutines (this can be adjusted by
226changing $Coro::POOL_SIZE), and there can be as many non-idle coros as
227required.
228
229If you are concerned about pooled coroutines growing a lot because a
230single C<async_pool> used a lot of stackspace you can e.g. C<async_pool {
231terminate }> once per second or so to slowly replenish the pool.
232
233=cut
234
235our $POOL_SIZE = 8;
236our @pool;
237
238sub pool_handler {
239 while () {
240 eval {
241 my ($cb, @arg) = @{ delete $current->{_invoke} or return };
242 $cb->(@arg);
243 };
244 warn $@ if $@;
245
246 last if @pool >= $POOL_SIZE;
247 push @pool, $current;
248
249 $current->save (Coro::State::SAVE_DEF);
250 $current->prio (0);
251 schedule;
252 }
253}
254
255sub async_pool(&@) {
256 # this is also inlined into the unlock_scheduler
257 my $coro = (pop @pool or new Coro \&pool_handler);
258
259 $coro->{_invoke} = [@_];
260 $coro->ready;
261
262 $coro
203} 263}
204 264
205=item schedule 265=item schedule
206 266
207Calls the scheduler. Please note that the current coroutine will not be put 267Calls the scheduler. Please note that the current coroutine will not be put
232 292
233"Cede" to other coroutines. This function puts the current coroutine into the 293"Cede" to other coroutines. This function puts the current coroutine into the
234ready queue and calls C<schedule>, which has the effect of giving up the 294ready queue and calls C<schedule>, which has the effect of giving up the
235current "timeslice" to other coroutines of the same or higher priority. 295current "timeslice" to other coroutines of the same or higher priority.
236 296
297Returns true if at least one coroutine switch has happened.
298
299=item Coro::cede_notself
300
301Works like cede, but is not exported by default and will cede to any
302coroutine, regardless of priority, once.
303
304Returns true if at least one coroutine switch has happened.
305
237=item terminate [arg...] 306=item terminate [arg...]
238 307
239Terminates the current coroutine with the given status values (see L<cancel>). 308Terminates the current coroutine with the given status values (see L<cancel>).
240 309
241=cut 310=cut
259Create a new coroutine and return it. When the sub returns the coroutine 328Create a new coroutine and return it. When the sub returns the coroutine
260automatically terminates as if C<terminate> with the returned values were 329automatically terminates as if C<terminate> with the returned values were
261called. To make the coroutine run you must first put it into the ready queue 330called. To make the coroutine run you must first put it into the ready queue
262by calling the ready method. 331by calling the ready method.
263 332
264Calling C<exit> in a coroutine will not work correctly, so do not do that. 333See C<async> for additional discussion.
265 334
266=cut 335=cut
267 336
268sub _run_coro { 337sub _run_coro {
269 terminate &{+shift}; 338 terminate &{+shift};
286Return wether the coroutine is currently the ready queue or not, 355Return wether the coroutine is currently the ready queue or not,
287 356
288=item $coroutine->cancel (arg...) 357=item $coroutine->cancel (arg...)
289 358
290Terminates the given coroutine and makes it return the given arguments as 359Terminates the given coroutine and makes it return the given arguments as
291status (default: the empty list). 360status (default: the empty list). Never returns if the coroutine is the
361current coroutine.
292 362
293=cut 363=cut
294 364
295sub cancel { 365sub cancel {
296 my $self = shift; 366 my $self = shift;
297 $self->{status} = [@_]; 367 $self->{status} = [@_];
368
369 if ($current == $self) {
298 push @destroy, $self; 370 push @destroy, $self;
299 $manager->ready; 371 $manager->ready;
300 &schedule if $current == $self; 372 &schedule while 1;
373 } else {
374 $self->_cancel;
375 }
301} 376}
302 377
303=item $coroutine->join 378=item $coroutine->join
304 379
305Wait until the coroutine terminates and return any values given to the 380Wait until the coroutine terminates and return any values given to the
308 383
309=cut 384=cut
310 385
311sub join { 386sub join {
312 my $self = shift; 387 my $self = shift;
388
313 unless ($self->{status}) { 389 unless ($self->{status}) {
314 push @{$self->{join}}, $current; 390 my $current = $current;
315 &schedule; 391
392 push @{$self->{destroy_cb}}, sub {
393 $current->ready;
394 undef $current;
395 };
396
397 &schedule while $current;
316 } 398 }
399
317 wantarray ? @{$self->{status}} : $self->{status}[0]; 400 wantarray ? @{$self->{status}} : $self->{status}[0];
401}
402
403=item $coroutine->on_destroy (\&cb)
404
405Registers a callback that is called when this coroutine gets destroyed,
406but before it is joined. The callback gets passed the terminate arguments,
407if any.
408
409=cut
410
411sub on_destroy {
412 my ($self, $cb) = @_;
413
414 push @{ $self->{destroy_cb} }, $cb;
318} 415}
319 416
320=item $oldprio = $coroutine->prio ($newprio) 417=item $oldprio = $coroutine->prio ($newprio)
321 418
322Sets (or gets, if the argument is missing) the priority of the 419Sets (or gets, if the argument is missing) the priority of the
369i.e. that can be swicthed to. The value C<0> means that the only runnable 466i.e. that can be swicthed to. The value C<0> means that the only runnable
370coroutine is the currently running one, so C<cede> would have no effect, 467coroutine is the currently running one, so C<cede> would have no effect,
371and C<schedule> would cause a deadlock unless there is an idle handler 468and C<schedule> would cause a deadlock unless there is an idle handler
372that wakes up some coroutines. 469that wakes up some coroutines.
373 470
471=item my $guard = Coro::guard { ... }
472
473This creates and returns a guard object. Nothing happens until the object
474gets destroyed, in which case the codeblock given as argument will be
475executed. This is useful to free locks or other resources in case of a
476runtime error or when the coroutine gets canceled, as in both cases the
477guard block will be executed. The guard object supports only one method,
478C<< ->cancel >>, which will keep the codeblock from being executed.
479
480Example: set some flag and clear it again when the coroutine gets canceled
481or the function returns:
482
483 sub do_something {
484 my $guard = Coro::guard { $busy = 0 };
485 $busy = 1;
486
487 # do something that requires $busy to be true
488 }
489
490=cut
491
492sub guard(&) {
493 bless \(my $cb = $_[0]), "Coro::guard"
494}
495
496sub Coro::guard::cancel {
497 ${$_[0]} = sub { };
498}
499
500sub Coro::guard::DESTROY {
501 ${$_[0]}->();
502}
503
504
374=item unblock_sub { ... } 505=item unblock_sub { ... }
375 506
376This utility function takes a BLOCK or code reference and "unblocks" it, 507This utility function takes a BLOCK or code reference and "unblocks" it,
377returning the new coderef. This means that the new coderef will return 508returning the new coderef. This means that the new coderef will return
378immediately without blocking, returning nothing, while the original code 509immediately without blocking, returning nothing, while the original code
391In short: simply use C<unblock_sub { ... }> instead of C<sub { ... }> when 522In short: simply use C<unblock_sub { ... }> instead of C<sub { ... }> when
392creating event callbacks that want to block. 523creating event callbacks that want to block.
393 524
394=cut 525=cut
395 526
396our @unblock_pool;
397our @unblock_queue; 527our @unblock_queue;
398our $UNBLOCK_POOL_SIZE = 2;
399 528
400sub unblock_handler_ { 529# we create a special coro because we want to cede,
401 while () { 530# to reduce pressure on the coro pool (because most callbacks
402 my ($cb, @arg) = @{ delete $Coro::current->{arg} }; 531# return immediately and can be reused) and because we cannot cede
403 $cb->(@arg); 532# inside an event callback.
404
405 last if @unblock_pool >= $UNBLOCK_POOL_SIZE;
406 push @unblock_pool, $Coro::current;
407 schedule;
408 }
409}
410
411our $unblock_scheduler = async { 533our $unblock_scheduler = async {
412 while () { 534 while () {
413 while (my $cb = pop @unblock_queue) { 535 while (my $cb = pop @unblock_queue) {
536 # this is an inlined copy of async_pool
414 my $handler = (pop @unblock_pool or new Coro \&unblock_handler_); 537 my $coro = (pop @pool or new Coro \&pool_handler);
415 $handler->{arg} = $cb; 538
539 $coro->{_invoke} = $cb;
416 $handler->ready; 540 $coro->ready;
417 cede; 541 cede; # for short-lived callbacks, this reduces pressure on the coro pool
418 } 542 }
419 543 schedule; # sleep well
420 schedule;
421 } 544 }
422}; 545};
423 546
424sub unblock_sub(&) { 547sub unblock_sub(&) {
425 my $cb = shift; 548 my $cb = shift;
426 549
427 sub { 550 sub {
428 push @unblock_queue, [$cb, @_]; 551 unshift @unblock_queue, [$cb, @_];
429 $unblock_scheduler->ready; 552 $unblock_scheduler->ready;
430 } 553 }
431} 554}
432 555
433=back 556=back

Diff Legend

Removed lines
+ Added lines
< Changed lines
> Changed lines