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

Comparing Coro/Coro.pm (file contents):
Revision 1.100 by root, Tue Dec 12 13:56:45 2006 UTC vs.
Revision 1.103 by root, Thu Jan 4 20:14:19 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.2'; 55our $VERSION = '3.3';
56 56
57our @EXPORT = qw(async cede schedule terminate current unblock_sub); 57our @EXPORT = qw(async 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);
… …
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
… …
232 235
233"Cede" to other coroutines. This function puts the current coroutine into the 236"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 237ready queue and calls C<schedule>, which has the effect of giving up the
235current "timeslice" to other coroutines of the same or higher priority. 238current "timeslice" to other coroutines of the same or higher priority.
236 239
240=item Coro::cede_notself
241
242Works like cede, but is not exported by default and will cede to any
243coroutine, regardless of priority, once.
244
237=item terminate [arg...] 245=item terminate [arg...]
238 246
239Terminates the current coroutine with the given status values (see L<cancel>). 247Terminates the current coroutine with the given status values (see L<cancel>).
240 248
241=cut 249=cut
… …
286Return wether the coroutine is currently the ready queue or not, 294Return wether the coroutine is currently the ready queue or not,
287 295
288=item $coroutine->cancel (arg...) 296=item $coroutine->cancel (arg...)
289 297
290Terminates the given coroutine and makes it return the given arguments as 298Terminates the given coroutine and makes it return the given arguments as
291status (default: the empty list). 299status (default: the empty list). Never returns if the coroutine is the
300current coroutine.
292 301
293=cut 302=cut
294 303
295sub cancel { 304sub cancel {
296 my $self = shift; 305 my $self = shift;
297 $self->{status} = [@_]; 306 $self->{status} = [@_];
307
308 if ($current == $self) {
298 push @destroy, $self; 309 push @destroy, $self;
299 $manager->ready; 310 $manager->ready;
300 &schedule if $current == $self; 311 &schedule while 1;
312 } else {
313 $self->_cancel;
314 }
301} 315}
302 316
303=item $coroutine->join 317=item $coroutine->join
304 318
305Wait until the coroutine terminates and return any values given to the 319Wait until the coroutine terminates and return any values given to the
… …
308 322
309=cut 323=cut
310 324
311sub join { 325sub join {
312 my $self = shift; 326 my $self = shift;
327
313 unless ($self->{status}) { 328 unless ($self->{status}) {
314 push @{$self->{join}}, $current; 329 my $current = $current;
315 &schedule; 330
331 push @{$self->{destroy_cb}}, sub {
332 $current->ready;
333 undef $current;
334 };
335
336 &schedule while $current;
316 } 337 }
338
317 wantarray ? @{$self->{status}} : $self->{status}[0]; 339 wantarray ? @{$self->{status}} : $self->{status}[0];
340}
341
342=item $coroutine->on_destroy (\&cb)
343
344Registers a callback that is called when this coroutine gets destroyed,
345but before it is joined. The callback gets passed the terminate arguments,
346if any.
347
348=cut
349
350sub on_destroy {
351 my ($self, $cb) = @_;
352
353 push @{ $self->{destroy_cb} }, $cb;
318} 354}
319 355
320=item $oldprio = $coroutine->prio ($newprio) 356=item $oldprio = $coroutine->prio ($newprio)
321 357
322Sets (or gets, if the argument is missing) the priority of the 358Sets (or gets, if the argument is missing) the priority of the
… …
368Returns the number of coroutines that are currently in the ready state, 404Returns the number of coroutines that are currently in the ready state,
369i.e. that can be swicthed to. The value C<0> means that the only runnable 405i.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, 406coroutine 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 407and C<schedule> would cause a deadlock unless there is an idle handler
372that wakes up some coroutines. 408that wakes up some coroutines.
409
410=item my $guard = Coro::guard { ... }
411
412This creates and returns a guard object. Nothing happens until the objetc
413gets destroyed, in which case the codeblock given as argument will be
414executed. This is useful to free locks or other resources in case of a
415runtime error or when the coroutine gets canceled, as in both cases the
416guard block will be executed. The guard object supports only one method,
417C<< ->cancel >>, which will keep the codeblock from being executed.
418
419Example: set some flag and clear it again when the coroutine gets canceled
420or the function returns:
421
422 sub do_something {
423 my $guard = Coro::guard { $busy = 0 };
424 $busy = 1;
425
426 # do something that requires $busy to be true
427 }
428
429=cut
430
431sub guard(&) {
432 bless \(my $cb = $_[0]), "Coro::guard"
433}
434
435sub Coro::guard::cancel {
436 ${$_[0]} = sub { };
437}
438
439sub Coro::guard::DESTROY {
440 ${$_[0]}->();
441}
442
373 443
374=item unblock_sub { ... } 444=item unblock_sub { ... }
375 445
376This utility function takes a BLOCK or code reference and "unblocks" it, 446This utility function takes a BLOCK or code reference and "unblocks" it,
377returning the new coderef. This means that the new coderef will return 447returning the new coderef. This means that the new coderef will return

Diff Legend

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