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

Comparing Coro/Coro.pm (file contents):
Revision 1.7 by root, Fri Jul 13 13:05:38 2001 UTC vs.
Revision 1.58 by pcg, Fri Feb 13 23:17:41 2004 UTC

1=head1 NAME 1=head1 NAME
2 2
3Coro - create and manage simple coroutines 3Coro - coroutine process abstraction
4 4
5=head1 SYNOPSIS 5=head1 SYNOPSIS
6 6
7 use Coro; 7 use Coro;
8 8
9 $new = new Coro sub { 9 async {
10 print "in coroutine, switching back\n"; 10 # some asynchronous thread of execution
11 $new->transfer($main);
12 print "in coroutine again, switching back\n";
13 $new->transfer($main);
14 }; 11 };
15 12
16 $main = new Coro; 13 # alternatively create an async process like this:
17 14
18 print "in main, switching to coroutine\n"; 15 sub some_func : Coro {
19 $main->transfer($new); 16 # some more async code
20 print "back in main, switch to coroutine again\n"; 17 }
21 $main->transfer($new); 18
22 print "back in main\n"; 19 cede;
23 20
24=head1 DESCRIPTION 21=head1 DESCRIPTION
25 22
26This module implements coroutines. Coroutines, similar to continuations, 23This module collection manages coroutines. Coroutines are similar to
27allow you to run more than one "thread of execution" in parallel. Unlike 24threads but don't run in parallel.
28threads this, only voluntary switching is used so locking problems are
29greatly reduced.
30 25
31Although this is the "main" module of the Coro family it provides only 26In this module, coroutines are defined as "callchain + lexical variables
32low-level functionality. See L<Coro::Process> and related modules for a 27+ @_ + $_ + $@ + $^W + C stack), that is, a coroutine has it's own
33more useful process abstraction including scheduling. 28callchain, it's own set of lexicals and it's own set of perl's most
29important global variables.
30
31=cut
32
33package Coro;
34
35BEGIN { eval { require warnings } && warnings->unimport ("uninitialized") }
36
37use Coro::State;
38
39use vars qw($idle $main $current);
40
41use base Exporter;
42
43$VERSION = 0.95;
44
45@EXPORT = qw(async cede schedule terminate current);
46%EXPORT_TAGS = (
47 prio => [qw(PRIO_MAX PRIO_HIGH PRIO_NORMAL PRIO_LOW PRIO_IDLE PRIO_MIN)],
48);
49@EXPORT_OK = @{$EXPORT_TAGS{prio}};
50
51{
52 my @async;
53 my $init;
54
55 # this way of handling attributes simply is NOT scalable ;()
56 sub import {
57 Coro->export_to_level(1, @_);
58 my $old = *{(caller)[0]."::MODIFY_CODE_ATTRIBUTES"}{CODE};
59 *{(caller)[0]."::MODIFY_CODE_ATTRIBUTES"} = sub {
60 my ($package, $ref) = (shift, shift);
61 my @attrs;
62 for (@_) {
63 if ($_ eq "Coro") {
64 push @async, $ref;
65 unless ($init++) {
66 eval q{
67 sub INIT {
68 &async(pop @async) while @async;
69 }
70 };
71 }
72 } else {
73 push @attrs, $_;
74 }
75 }
76 return $old ? $old->($package, $ref, @attrs) : @attrs;
77 };
78 }
79
80}
34 81
35=over 4 82=over 4
36 83
37=cut 84=item $main
38 85
39package Coro; 86This coroutine represents the main program.
40 87
41BEGIN {
42 $VERSION = 0.03;
43
44 require XSLoader;
45 XSLoader::load Coro, $VERSION;
46}
47
48=item $coro = new [$coderef [, @args]]
49
50Create a new coroutine and return it. The first C<transfer> call to this
51coroutine will start execution at the given coderef. If, the subroutine
52returns it will be executed again.
53
54If the coderef is omitted this function will create a new "empty"
55coroutine, i.e. a coroutine that cannot be transfered to but can be used
56to save the current coroutine in.
57
58=cut 88=cut
89
90$main = new Coro;
91
92=item $current (or as function: current)
93
94The current coroutine (the last coroutine switched to). The initial value is C<$main> (of course).
95
96=cut
97
98# maybe some other module used Coro::Specific before...
99if ($current) {
100 $main->{specific} = $current->{specific};
101}
102
103$current = $main;
104
105sub current() { $current }
106
107=item $idle
108
109The coroutine to switch to when no other coroutine is running. The default
110implementation prints "FATAL: deadlock detected" and exits.
111
112=cut
113
114# should be done using priorities :(
115$idle = new Coro sub {
116 print STDERR "FATAL: deadlock detected\n";
117 exit(51);
118};
119
120# this coroutine is necessary because a coroutine
121# cannot destroy itself.
122my @destroy;
123my $manager;
124$manager = new Coro sub {
125 while () {
126 # by overwriting the state object with the manager we destroy it
127 # while still being able to schedule this coroutine (in case it has
128 # been readied multiple times. this is harmless since the manager
129 # can be called as many times as neccessary and will always
130 # remove itself from the runqueue
131 while (@destroy) {
132 my $coro = pop @destroy;
133 $coro->{status} ||= [];
134 $_->ready for @{delete $coro->{join} || []};
135 $coro->{_coro_state} = $manager->{_coro_state};
136 }
137 &schedule;
138 }
139};
140
141# static methods. not really.
142
143=back
144
145=head2 STATIC METHODS
146
147Static methods are actually functions that operate on the current process only.
148
149=over 4
150
151=item async { ... } [@args...]
152
153Create a new asynchronous process and return it's process object
154(usually unused). When the sub returns the new process is automatically
155terminated.
156
157 # create a new coroutine that just prints its arguments
158 async {
159 print "@_\n";
160 } 1,2,3,4;
161
162=cut
163
164sub async(&@) {
165 my $pid = new Coro @_;
166 $manager->ready; # this ensures that the stack is cloned from the manager
167 $pid->ready;
168 $pid;
169}
170
171=item schedule
172
173Calls the scheduler. Please note that the current process will not be put
174into the ready queue, so calling this function usually means you will
175never be called again.
176
177=cut
178
179=item cede
180
181"Cede" to other processes. This function puts the current process into the
182ready queue and calls C<schedule>, which has the effect of giving up the
183current "timeslice" to other coroutines of the same or higher priority.
184
185=cut
186
187=item terminate [arg...]
188
189Terminates the current process.
190
191Future versions of this function will allow result arguments.
192
193=cut
194
195sub terminate {
196 $current->{status} = [@_];
197 $current->cancel;
198 &schedule;
199 die; # NORETURN
200}
201
202=back
203
204# dynamic methods
205
206=head2 PROCESS METHODS
207
208These are the methods you can call on process objects.
209
210=over 4
211
212=item new Coro \&sub [, @args...]
213
214Create a new process and return it. When the sub returns the process
215automatically terminates as if C<terminate> with the returned values were
216called. To make the process run you must first put it into the ready queue
217by calling the ready method.
218
219=cut
220
221sub _newcoro {
222 terminate &{+shift};
223}
59 224
60sub new { 225sub new {
61 my $class = $_[0]; 226 my $class = shift;
62 my $proc = $_[1] || sub { die "tried to transfer to an empty coroutine" }; 227 bless {
63 bless _newprocess { 228 _coro_state => (new Coro::State $_[0] && \&_newcoro, @_),
64 do {
65 eval { &$proc };
66 if ($@) {
67 $error_msg = $@;
68 $error_coro = _newprocess { };
69 &transfer($error_coro, $error);
70 }
71 } while (1);
72 }, $class; 229 }, $class;
73} 230}
74 231
75=item $prev->transfer($next) 232=item $process->ready
76 233
77Save the state of the current subroutine in C<$prev> and switch to the 234Put the given process into the ready queue.
78coroutine saved in C<$next>.
79 235
80The "state" of a subroutine only ever includes scope, i.e. lexical 236=cut
81variables and the current execution state. It does not save/restore any
82global variables such as C<$_> or C<$@> or any other special or non
83special variables. So remember that every function call that might call
84C<transfer> (such as C<Coro::Channel::put>) might clobber any global
85and/or special variables. Yes, this is by design ;) You cna always create
86your own process abstraction model that saves these variables.
87 237
88The easiest way to do this is to create your own scheduling primitive like this: 238=item $process->cancel
89 239
90 sub schedule { 240Like C<terminate>, but terminates the specified process instead.
91 local ($_, $@, ...); 241
92 $old->transfer($new); 242=cut
243
244sub cancel {
245 push @destroy, $_[0];
246 $manager->ready;
247 &schedule if $current == $_[0];
248}
249
250=item $process->join
251
252Wait until the coroutine terminates and return any values given to the
253C<terminate> function. C<join> can be called multiple times from multiple
254processes.
255
256=cut
257
258sub join {
259 my $self = shift;
260 unless ($self->{status}) {
261 push @{$self->{join}}, $current;
262 &schedule;
93 } 263 }
94 264 wantarray ? @{$self->{status}} : $self->{status}[0];
95=cut
96
97# I call the _transfer function from a perl function
98# because that way perl saves all important things on
99# the stack. Actually, I'd do it from within XS, but
100# I couldn't get it to work.
101sub transfer {
102 _transfer($_[0], $_[1]);
103} 265}
104 266
105=item $error, $error_msg, $error_coro 267=item $oldprio = $process->prio($newprio)
106 268
107This coroutine will be called on fatal errors. C<$error_msg> and 269Sets (or gets, if the argument is missing) the priority of the
108C<$error_coro> return the error message and the error-causing coroutine 270process. Higher priority processes get run before lower priority
109(NOT an object) respectively. This API might change. 271processes. Priorities are small signed integers (currently -4 .. +3),
272that you can refer to using PRIO_xxx constants (use the import tag :prio
273to get then):
110 274
111=cut 275 PRIO_MAX > PRIO_HIGH > PRIO_NORMAL > PRIO_LOW > PRIO_IDLE > PRIO_MIN
276 3 > 1 > 0 > -1 > -3 > -4
112 277
113$error_msg = 278 # set priority to HIGH
114$error_coro = undef; 279 current->prio(PRIO_HIGH);
115 280
116$error = _newprocess { 281The idle coroutine ($Coro::idle) always has a lower priority than any
117 print STDERR "FATAL: $error_msg\nprogram aborted\n"; 282existing coroutine.
118 exit 50; 283
119}; 284Changing the priority of the current process will take effect immediately,
285but changing the priority of processes in the ready queue (but not
286running) will only take effect after the next schedule (of that
287process). This is a bug that will be fixed in some future version.
288
289=cut
290
291sub prio {
292 my $old = $_[0]{prio};
293 $_[0]{prio} = $_[1] if @_ > 1;
294 $old;
295}
296
297=item $newprio = $process->nice($change)
298
299Similar to C<prio>, but subtract the given value from the priority (i.e.
300higher values mean lower priority, just as in unix).
301
302=cut
303
304sub nice {
305 $_[0]{prio} -= $_[1];
306}
307
308=item $olddesc = $process->desc($newdesc)
309
310Sets (or gets in case the argument is missing) the description for this
311process. This is just a free-form string you can associate with a process.
312
313=cut
314
315sub desc {
316 my $old = $_[0]{desc};
317 $_[0]{desc} = $_[1] if @_ > 1;
318 $old;
319}
320
321=back
322
323=cut
120 324
1211; 3251;
122 326
123=back 327=head1 BUGS/LIMITATIONS
124 328
125=head1 BUGS 329 - you must make very sure that no coro is still active on global
330 destruction. very bad things might happen otherwise (usually segfaults).
126 331
127This module has not yet been extensively tested. 332 - this module is not thread-safe. You should only ever use this module
333 from the same thread (this requirement might be losened in the future
334 to allow per-thread schedulers, but Coro::State does not yet allow
335 this).
128 336
129=head1 SEE ALSO 337=head1 SEE ALSO
130 338
131L<Coro::Process>, L<Coro::Signal>. 339L<Coro::Channel>, L<Coro::Cont>, L<Coro::Specific>, L<Coro::Semaphore>,
340L<Coro::Signal>, L<Coro::State>, L<Coro::Timer>, L<Coro::Event>,
341L<Coro::L<Coro::RWLock>, Handle>, L<Coro::Socket>.
132 342
133=head1 AUTHOR 343=head1 AUTHOR
134 344
135 Marc Lehmann <pcg@goof.com> 345 Marc Lehmann <pcg@goof.com>
136 http://www.goof.com/pcg/marc/ 346 http://www.goof.com/pcg/marc/

Diff Legend

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