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

Comparing Coro/Coro.pm (file contents):
Revision 1.56 by pcg, Sat Nov 15 03:53:10 2003 UTC vs.
Revision 1.98 by root, Mon Dec 4 21:56:00 2006 UTC

8 8
9 async { 9 async {
10 # some asynchronous thread of execution 10 # some asynchronous thread of execution
11 }; 11 };
12 12
13 # alternatively create an async process like this: 13 # alternatively create an async coroutine like this:
14 14
15 sub some_func : Coro { 15 sub some_func : Coro {
16 # some more async code 16 # some more async code
17 } 17 }
18 18
19 cede; 19 cede;
20 20
21=head1 DESCRIPTION 21=head1 DESCRIPTION
22 22
23This module collection manages coroutines. Coroutines are similar to 23This module collection manages coroutines. Coroutines are similar
24threads but don't run in parallel. 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
26guarentees you that it will not switch between coroutines unless
27necessary, at easily-identified points in your program, so locking and
28parallel access are rarely an issue, making coroutine programming much
29safer than threads programming.
25 30
31(Perl, however, does not natively support real threads but instead does a
32very slow and memory-intensive emulation of processes using threads. This
33is a performance win on Windows machines, and a loss everywhere else).
34
26In this module, coroutines are defined as "callchain + lexical variables 35In this module, coroutines are defined as "callchain + lexical variables +
27+ @_ + $_ + $@ + $^W + C stack), that is, a coroutine has it's own 36@_ + $_ + $@ + $/ + C stack), that is, a coroutine has its own callchain,
28callchain, it's own set of lexicals and it's own set of perl's most 37its own set of lexicals and its own set of perls most important global
29important global variables. 38variables.
30 39
31=cut 40=cut
32 41
33package Coro; 42package Coro;
34 43
35BEGIN { eval { require warnings } && warnings->unimport ("uninitialized") } 44use strict;
45no warnings "uninitialized";
36 46
37use Coro::State; 47use Coro::State;
38 48
39use vars qw($idle $main $current); 49use base qw(Coro::State Exporter);
40 50
41use base Exporter; 51our $idle; # idle handler
52our $main; # main coroutine
53our $current; # current coroutine
42 54
43$VERSION = "0.9"; 55our $VERSION = '3.1';
44 56
45@EXPORT = qw(async cede schedule terminate current); 57our @EXPORT = qw(async cede schedule terminate current unblock_sub);
46%EXPORT_TAGS = ( 58our %EXPORT_TAGS = (
47 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)],
48); 60);
49@EXPORT_OK = @{$EXPORT_TAGS{prio}}; 61our @EXPORT_OK = (@{$EXPORT_TAGS{prio}}, qw(nready));
50 62
51{ 63{
52 my @async; 64 my @async;
53 my $init; 65 my $init;
54 66
55 # this way of handling attributes simply is NOT scalable ;() 67 # this way of handling attributes simply is NOT scalable ;()
56 sub import { 68 sub import {
69 no strict 'refs';
70
57 Coro->export_to_level(1, @_); 71 Coro->export_to_level (1, @_);
72
58 my $old = *{(caller)[0]."::MODIFY_CODE_ATTRIBUTES"}{CODE}; 73 my $old = *{(caller)[0]."::MODIFY_CODE_ATTRIBUTES"}{CODE};
59 *{(caller)[0]."::MODIFY_CODE_ATTRIBUTES"} = sub { 74 *{(caller)[0]."::MODIFY_CODE_ATTRIBUTES"} = sub {
60 my ($package, $ref) = (shift, shift); 75 my ($package, $ref) = (shift, shift);
61 my @attrs; 76 my @attrs;
62 for (@_) { 77 for (@_) {
89 104
90$main = new Coro; 105$main = new Coro;
91 106
92=item $current (or as function: current) 107=item $current (or as function: current)
93 108
94The current coroutine (the last coroutine switched to). The initial value is C<$main> (of course). 109The current coroutine (the last coroutine switched to). The initial value
110is C<$main> (of course).
111
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
114C<Coro::current> function instead.
95 115
96=cut 116=cut
97 117
98# maybe some other module used Coro::Specific before... 118# maybe some other module used Coro::Specific before...
99if ($current) {
100 $main->{specific} = $current->{specific}; 119$main->{specific} = $current->{specific}
101} 120 if $current;
102 121
103$current = $main; 122_set_current $main;
104 123
105sub current() { $current } 124sub current() { $current }
106 125
107=item $idle 126=item $idle
108 127
109The coroutine to switch to when no other coroutine is running. The default 128A callback that is called whenever the scheduler finds no ready coroutines
110implementation prints "FATAL: deadlock detected" and exits. 129to run. The default implementation prints "FATAL: deadlock detected" and
130exits, because the program has no other way to continue.
111 131
112=cut 132This hook is overwritten by modules such as C<Coro::Timer> and
133C<Coro::Event> to wait on an external event that hopefully wake up a
134coroutine so the scheduler can run it.
113 135
114# should be done using priorities :( 136Please note that if your callback recursively invokes perl (e.g. for event
115$idle = new Coro sub { 137handlers), then it must be prepared to be called recursively.
116 print STDERR "FATAL: deadlock detected\n"; 138
117 exit(51); 139=cut
140
141$idle = sub {
142 require Carp;
143 Carp::croak ("FATAL: deadlock detected");
118}; 144};
119 145
120# this coroutine is necessary because a coroutine 146# this coroutine is necessary because a coroutine
121# cannot destroy itself. 147# cannot destroy itself.
122my @destroy; 148my @destroy;
123my $manager;
124$manager = new Coro sub { 149my $manager; $manager = new Coro sub {
125 while() { 150 while () {
126 # by overwriting the state object with the manager we destroy it 151 # by overwriting the state object with the manager we destroy it
127 # while still being able to schedule this coroutine (in case it has 152 # while still being able to schedule this coroutine (in case it has
128 # been readied multiple times. this is harmless since the manager 153 # been readied multiple times. this is harmless since the manager
129 # can be called as many times as neccessary and will always 154 # can be called as many times as neccessary and will always
130 # remove itself from the runqueue 155 # remove itself from the runqueue
131 while (@destroy) { 156 while (@destroy) {
132 my $coro = pop @destroy; 157 my $coro = pop @destroy;
133 $coro->{status} ||= []; 158 $coro->{status} ||= [];
134 $_->ready for @{delete $coro->{join} || []}; 159 $_->ready for @{delete $coro->{join} || []};
135 $coro->{_coro_state} = $manager->{_coro_state}; 160
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);
136 } 166 }
137 &schedule; 167 &schedule;
138 } 168 }
139}; 169};
140 170
142 172
143=back 173=back
144 174
145=head2 STATIC METHODS 175=head2 STATIC METHODS
146 176
147Static methods are actually functions that operate on the current process only. 177Static methods are actually functions that operate on the current coroutine only.
148 178
149=over 4 179=over 4
150 180
151=item async { ... } [@args...] 181=item async { ... } [@args...]
152 182
153Create a new asynchronous process and return it's process object 183Create a new asynchronous coroutine and return it's coroutine object
154(usually unused). When the sub returns the new process is automatically 184(usually unused). When the sub returns the new coroutine is automatically
155terminated. 185terminated.
186
187Calling C<exit> in a coroutine will not work correctly, so do not do that.
188
189When the coroutine dies, the program will exit, just as in the main
190program.
156 191
157 # create a new coroutine that just prints its arguments 192 # create a new coroutine that just prints its arguments
158 async { 193 async {
159 print "@_\n"; 194 print "@_\n";
160 } 1,2,3,4; 195 } 1,2,3,4;
161 196
162The coderef you submit MUST NOT be a closure that refers to variables
163in an outer scope. This does NOT work. Pass arguments into it instead.
164
165=cut 197=cut
166 198
167sub async(&@) { 199sub async(&@) {
168 my $pid = new Coro @_; 200 my $pid = new Coro @_;
169 $manager->ready; # this ensures that the stack is cloned from the manager
170 $pid->ready; 201 $pid->ready;
171 $pid; 202 $pid
172} 203}
173 204
174=item schedule 205=item schedule
175 206
176Calls the scheduler. Please note that the current process will not be put 207Calls the scheduler. Please note that the current coroutine will not be put
177into the ready queue, so calling this function usually means you will 208into the ready queue, so calling this function usually means you will
178never be called again. 209never be called again unless something else (e.g. an event handler) calls
210ready.
179 211
180=cut 212The canonical way to wait on external events is this:
213
214 {
215 # remember current coroutine
216 my $current = $Coro::current;
217
218 # register a hypothetical event handler
219 on_event_invoke sub {
220 # wake up sleeping coroutine
221 $current->ready;
222 undef $current;
223 };
224
225 # call schedule until event occured.
226 # in case we are woken up for other reasons
227 # (current still defined), loop.
228 Coro::schedule while $current;
229 }
181 230
182=item cede 231=item cede
183 232
184"Cede" to other processes. This function puts the current process into the 233"Cede" to other coroutines. This function puts the current coroutine into the
185ready queue and calls C<schedule>, which has the effect of giving up the 234ready queue and calls C<schedule>, which has the effect of giving up the
186current "timeslice" to other coroutines of the same or higher priority. 235current "timeslice" to other coroutines of the same or higher priority.
187 236
188=cut
189
190=item terminate [arg...] 237=item terminate [arg...]
191 238
192Terminates the current process. 239Terminates the current coroutine with the given status values (see L<cancel>).
193
194Future versions of this function will allow result arguments.
195 240
196=cut 241=cut
197 242
198sub terminate { 243sub terminate {
199 $current->{status} = [@_];
200 $current->cancel; 244 $current->cancel (@_);
201 &schedule;
202 die; # NORETURN
203} 245}
204 246
205=back 247=back
206 248
207# dynamic methods 249# dynamic methods
208 250
209=head2 PROCESS METHODS 251=head2 COROUTINE METHODS
210 252
211These are the methods you can call on process objects. 253These are the methods you can call on coroutine objects.
212 254
213=over 4 255=over 4
214 256
215=item new Coro \&sub [, @args...] 257=item new Coro \&sub [, @args...]
216 258
217Create a new process and return it. When the sub returns the process 259Create a new coroutine and return it. When the sub returns the coroutine
218automatically terminates as if C<terminate> with the returned values were 260automatically terminates as if C<terminate> with the returned values were
219called. To make the process run you must first put it into the ready queue 261called. To make the coroutine run you must first put it into the ready queue
220by calling the ready method. 262by calling the ready method.
221 263
222=cut 264Calling C<exit> in a coroutine will not work correctly, so do not do that.
223 265
266=cut
267
224sub _newcoro { 268sub _run_coro {
225 terminate &{+shift}; 269 terminate &{+shift};
226} 270}
227 271
228sub new { 272sub new {
229 my $class = shift; 273 my $class = shift;
230 bless {
231 _coro_state => (new Coro::State $_[0] && \&_newcoro, @_),
232 }, $class;
233}
234 274
235=item $process->ready 275 $class->SUPER::new (\&_run_coro, @_)
276}
236 277
237Put the given process into the ready queue. 278=item $success = $coroutine->ready
238 279
239=cut 280Put the given coroutine into the ready queue (according to it's priority)
281and return true. If the coroutine is already in the ready queue, do nothing
282and return false.
240 283
241=item $process->cancel 284=item $is_ready = $coroutine->is_ready
242 285
243Like C<terminate>, but terminates the specified process instead. 286Return wether the coroutine is currently the ready queue or not,
287
288=item $coroutine->cancel (arg...)
289
290Terminates the given coroutine and makes it return the given arguments as
291status (default: the empty list).
244 292
245=cut 293=cut
246 294
247sub cancel { 295sub cancel {
296 my $self = shift;
297 $self->{status} = [@_];
248 push @destroy, $_[0]; 298 push @destroy, $self;
249 $manager->ready; 299 $manager->ready;
250 &schedule if $current == $_[0]; 300 &schedule if $current == $self;
251} 301}
252 302
253=item $process->join 303=item $coroutine->join
254 304
255Wait until the coroutine terminates and return any values given to the 305Wait until the coroutine terminates and return any values given to the
256C<terminate> function. C<join> can be called multiple times from multiple 306C<terminate> or C<cancel> functions. C<join> can be called multiple times
257processes. 307from multiple coroutine.
258 308
259=cut 309=cut
260 310
261sub join { 311sub join {
262 my $self = shift; 312 my $self = shift;
265 &schedule; 315 &schedule;
266 } 316 }
267 wantarray ? @{$self->{status}} : $self->{status}[0]; 317 wantarray ? @{$self->{status}} : $self->{status}[0];
268} 318}
269 319
270=item $oldprio = $process->prio($newprio) 320=item $oldprio = $coroutine->prio ($newprio)
271 321
272Sets (or gets, if the argument is missing) the priority of the 322Sets (or gets, if the argument is missing) the priority of the
273process. Higher priority processes get run before lower priority 323coroutine. Higher priority coroutines get run before lower priority
274processes. Priorities are small signed integers (currently -4 .. +3), 324coroutines. Priorities are small signed integers (currently -4 .. +3),
275that you can refer to using PRIO_xxx constants (use the import tag :prio 325that you can refer to using PRIO_xxx constants (use the import tag :prio
276to get then): 326to get then):
277 327
278 PRIO_MAX > PRIO_HIGH > PRIO_NORMAL > PRIO_LOW > PRIO_IDLE > PRIO_MIN 328 PRIO_MAX > PRIO_HIGH > PRIO_NORMAL > PRIO_LOW > PRIO_IDLE > PRIO_MIN
279 3 > 1 > 0 > -1 > -3 > -4 329 3 > 1 > 0 > -1 > -3 > -4
282 current->prio(PRIO_HIGH); 332 current->prio(PRIO_HIGH);
283 333
284The idle coroutine ($Coro::idle) always has a lower priority than any 334The idle coroutine ($Coro::idle) always has a lower priority than any
285existing coroutine. 335existing coroutine.
286 336
287Changing the priority of the current process will take effect immediately, 337Changing the priority of the current coroutine will take effect immediately,
288but changing the priority of processes in the ready queue (but not 338but changing the priority of coroutines in the ready queue (but not
289running) will only take effect after the next schedule (of that 339running) will only take effect after the next schedule (of that
290process). This is a bug that will be fixed in some future version. 340coroutine). This is a bug that will be fixed in some future version.
291 341
292=cut
293
294sub prio {
295 my $old = $_[0]{prio};
296 $_[0]{prio} = $_[1] if @_ > 1;
297 $old;
298}
299
300=item $newprio = $process->nice($change) 342=item $newprio = $coroutine->nice ($change)
301 343
302Similar to C<prio>, but subtract the given value from the priority (i.e. 344Similar to C<prio>, but subtract the given value from the priority (i.e.
303higher values mean lower priority, just as in unix). 345higher values mean lower priority, just as in unix).
304 346
305=cut
306
307sub nice {
308 $_[0]{prio} -= $_[1];
309}
310
311=item $olddesc = $process->desc($newdesc) 347=item $olddesc = $coroutine->desc ($newdesc)
312 348
313Sets (or gets in case the argument is missing) the description for this 349Sets (or gets in case the argument is missing) the description for this
314process. This is just a free-form string you can associate with a process. 350coroutine. This is just a free-form string you can associate with a coroutine.
315 351
316=cut 352=cut
317 353
318sub desc { 354sub desc {
319 my $old = $_[0]{desc}; 355 my $old = $_[0]{desc};
321 $old; 357 $old;
322} 358}
323 359
324=back 360=back
325 361
362=head2 GLOBAL FUNCTIONS
363
364=over 4
365
366=item Coro::nready
367
368Returns 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
370coroutine 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
372that wakes up some coroutines.
373
374=item unblock_sub { ... }
375
376This utility function takes a BLOCK or code reference and "unblocks" it,
377returning the new coderef. This means that the new coderef will return
378immediately without blocking, returning nothing, while the original code
379ref will be called (with parameters) from within its own coroutine.
380
381The reason this fucntion exists is that many event libraries (such as the
382venerable L<Event|Event> module) are not coroutine-safe (a weaker form
383of thread-safety). This means you must not block within event callbacks,
384otherwise you might suffer from crashes or worse.
385
386This function allows your callbacks to block by executing them in another
387coroutine where it is safe to block. One example where blocking is handy
388is when you use the L<Coro::AIO|Coro::AIO> functions to save results to
389disk.
390
391In short: simply use C<unblock_sub { ... }> instead of C<sub { ... }> when
392creating event callbacks that want to block.
393
394=cut
395
396our @unblock_pool;
397our @unblock_queue;
398our $UNBLOCK_POOL_SIZE = 2;
399
400sub unblock_handler_ {
401 while () {
402 my ($cb, @arg) = @{ delete $Coro::current->{arg} };
403 $cb->(@arg);
404
405 last if @unblock_pool >= $UNBLOCK_POOL_SIZE;
406 push @unblock_pool, $Coro::current;
407 schedule;
408 }
409}
410
411our $unblock_scheduler = async {
412 while () {
413 while (my $cb = pop @unblock_queue) {
414 my $handler = (pop @unblock_pool or new Coro \&unblock_handler_);
415 $handler->{arg} = $cb;
416 $handler->ready;
417 cede;
418 }
419
420 schedule;
421 }
422};
423
424sub unblock_sub(&) {
425 my $cb = shift;
426
427 sub {
428 push @unblock_queue, [$cb, @_];
429 $unblock_scheduler->ready;
430 }
431}
432
433=back
434
326=cut 435=cut
327 436
3281; 4371;
329 438
330=head1 BUGS/LIMITATIONS 439=head1 BUGS/LIMITATIONS
337 to allow per-thread schedulers, but Coro::State does not yet allow 446 to allow per-thread schedulers, but Coro::State does not yet allow
338 this). 447 this).
339 448
340=head1 SEE ALSO 449=head1 SEE ALSO
341 450
342L<Coro::Channel>, L<Coro::Cont>, L<Coro::Specific>, L<Coro::Semaphore>, 451Support/Utility: L<Coro::Cont>, L<Coro::Specific>, L<Coro::State>, L<Coro::Util>.
343L<Coro::Signal>, L<Coro::State>, L<Coro::Timer>, L<Coro::Event>, 452
344L<Coro::L<Coro::RWLock>, Handle>, L<Coro::Socket>. 453Locking/IPC: L<Coro::Signal>, L<Coro::Channel>, L<Coro::Semaphore>, L<Coro::SemaphoreSet>, L<Coro::RWLock>.
454
455Event/IO: L<Coro::Timer>, L<Coro::Event>, L<Coro::Handle>, L<Coro::Socket>, L<Coro::Select>.
456
457Embedding: L<Coro:MakeMaker>
345 458
346=head1 AUTHOR 459=head1 AUTHOR
347 460
348 Marc Lehmann <pcg@goof.com> 461 Marc Lehmann <schmorp@schmorp.de>
349 http://www.goof.com/pcg/marc/ 462 http://home.schmorp.de/
350 463
351=cut 464=cut
352 465

Diff Legend

Removed lines
+ Added lines
< Changed lines
> Changed lines