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

Comparing cvsroot/Coro/Coro.pm (file contents):
Revision 1.274 by root, Sat Dec 12 01:30:26 2009 UTC vs.
Revision 1.294 by root, Fri May 6 21:15:17 2011 UTC

40points in your program, so locking and parallel access are rarely an 40points in your program, so locking and parallel access are rarely an
41issue, making thread programming much safer and easier than using other 41issue, making thread programming much safer and easier than using other
42thread models. 42thread models.
43 43
44Unlike the so-called "Perl threads" (which are not actually real threads 44Unlike the so-called "Perl threads" (which are not actually real threads
45but only the windows process emulation (see section of same name for more 45but only the windows process emulation (see section of same name for
46details) ported to unix, and as such act as processes), Coro provides 46more details) ported to UNIX, and as such act as processes), Coro
47a full shared address space, which makes communication between threads 47provides a full shared address space, which makes communication between
48very easy. And Coro's threads are fast, too: disabling the Windows 48threads very easy. And coro threads are fast, too: disabling the Windows
49process emulation code in your perl and using Coro can easily result in 49process emulation code in your perl and using Coro can easily result in
50a two to four times speed increase for your programs. A parallel matrix 50a two to four times speed increase for your programs. A parallel matrix
51multiplication benchmark runs over 300 times faster on a single core than 51multiplication benchmark (very communication-intensive) runs over 300
52perl's pseudo-threads on a quad core using all four cores. 52times faster on a single core than perls pseudo-threads on a quad core
53using all four cores.
53 54
54Coro achieves that by supporting multiple running interpreters that share 55Coro achieves that by supporting multiple running interpreters that share
55data, which is especially useful to code pseudo-parallel processes and 56data, which is especially useful to code pseudo-parallel processes and
56for event-based programming, such as multiple HTTP-GET requests running 57for event-based programming, such as multiple HTTP-GET requests running
57concurrently. See L<Coro::AnyEvent> to learn more on how to integrate Coro 58concurrently. See L<Coro::AnyEvent> to learn more on how to integrate Coro
63variables (see L<Coro::State> for more configuration and background info). 64variables (see L<Coro::State> for more configuration and background info).
64 65
65See also the C<SEE ALSO> section at the end of this document - the Coro 66See also the C<SEE ALSO> section at the end of this document - the Coro
66module family is quite large. 67module family is quite large.
67 68
69=head1 CORO THREAD LIFE CYCLE
70
71During the long and exciting (or not) life of a coro thread, it goes
72through a number of states:
73
74=over 4
75
76=item 1. Creation
77
78The first thing in the life of a coro thread is it's creation -
79obviously. The typical way to create a thread is to call the C<async
80BLOCK> function:
81
82 async {
83 # thread code goes here
84 };
85
86You can also pass arguments, which are put in C<@_>:
87
88 async {
89 print $_[1]; # prints 2
90 } 1, 2, 3;
91
92This creates a new coro thread and puts it into the ready queue, meaning
93it will run as soon as the CPU is free for it.
94
95C<async> will return a coro object - you can store this for future
96reference or ignore it, the thread itself will keep a reference to it's
97thread object - threads are alive on their own.
98
99Another way to create a thread is to call the C<new> constructor with a
100code-reference:
101
102 new Coro sub {
103 # thread code goes here
104 }, @optional_arguments;
105
106This is quite similar to calling C<async>, but the important difference is
107that the new thread is not put into the ready queue, so the thread will
108not run until somebody puts it there. C<async> is, therefore, identical to
109this sequence:
110
111 my $coro = new Coro sub {
112 # thread code goes here
113 };
114 $coro->ready;
115 return $coro;
116
117=item 2. Startup
118
119When a new coro thread is created, only a copy of the code reference
120and the arguments are stored, no extra memory for stacks and so on is
121allocated, keeping the coro thread in a low-memory state.
122
123Only when it actually starts executing will all the resources be finally
124allocated.
125
126The optional arguments specified at coro creation are available in C<@_>,
127similar to function calls.
128
129=item 3. Running / Blocking
130
131A lot can happen after the coro thread has started running. Quite usually,
132it will not run to the end in one go (because you could use a function
133instead), but it will give up the CPU regularly because it waits for
134external events.
135
136As long as a coro thread runs, it's coro object is available in the global
137variable C<$Coro::current>.
138
139The low-level way to give up the CPU is to call the scheduler, which
140selects a new coro thread to run:
141
142 Coro::schedule;
143
144Since running threads are not in the ready queue, calling the scheduler
145without doing anything else will block the coro thread forever - you need
146to arrange either for the coro to put woken up (readied) by some other
147event or some other thread, or you can put it into the ready queue before
148scheduling:
149
150 # this is exactly what Coro::cede does
151 $Coro::current->ready;
152 Coro::schedule;
153
154All the higher-level synchronisation methods (Coro::Semaphore,
155Coro::rouse_*...) are actually implemented via C<< ->ready >> and C<<
156Coro::schedule >>.
157
158While the coro thread is running it also might get assigned a C-level
159thread, or the C-level thread might be unassigned from it, as the Coro
160runtime wishes. A C-level thread needs to be assigned when your perl
161thread calls into some C-level function and that function in turn calls
162perl and perl then wants to switch coroutines. This happens most often
163when you run an event loop and block in the callback, or when perl
164itself calls some function such as C<AUTOLOAD> or methods via the C<tie>
165mechanism.
166
167=item 4. Termination
168
169Many threads actually terminate after some time. There are a number of
170ways to terminate a coro thread, the simplest is returning from the
171top-level code reference:
172
173 async {
174 # after returning from here, the coro thread is terminated
175 };
176
177 async {
178 return if 0.5 < rand; # terminate a little earlier, maybe
179 print "got a chance to print this\n";
180 # or here
181 };
182
183Any values returned from the coroutine can be recovered using C<< ->join
184>>:
185
186 my $coro = async {
187 "hello, world\n" # return a string
188 };
189
190 my $hello_world = $coro->join;
191
192 print $hello_world;
193
194Another way to terminate is to call C<< Coro::terminate >>, which at any
195subroutine call nesting level:
196
197 async {
198 Coro::terminate "return value 1", "return value 2";
199 };
200
201And yet another way is to C<< ->cancel >> (or C<< ->safe_cancel >>) the
202coro thread from another thread:
203
204 my $coro = async {
205 exit 1;
206 };
207
208 $coro->cancel; # an also accept values for ->join to retrieve
209
210Cancellation I<can> be dangerous - it's a bit like calling C<exit>
211without actually exiting, and might leave C libraries and XS modules in
212a weird state. Unlike other thread implementations, however, Coro is
213exceptionally safe with regards to cancellation, as perl will always be
214in a consistent state, and for those cases where you want to do truly
215marvellous things with your coro while it is being cancelled, there is
216even a C<< ->safe_cancel >> method.
217
218So, cancelling a thread that runs in an XS event loop might not be the
219best idea, but any other combination that deals with perl only (cancelling
220when a thread is in a C<tie> method or an C<AUTOLOAD> for example) is
221safe.
222
223=item 5. Cleanup
224
225Threads will allocate various resources. Most but not all will be returned
226when a thread terminates, during clean-up.
227
228Cleanup is quite similar to throwing an uncaught exception: perl will
229work it's way up through all subroutine calls and blocks. On it's way, it
230will release all C<my> variables, undo all C<local>'s and free any other
231resources truly local to the thread.
232
233So, a common way to free resources is to keep them referenced only by my
234variables:
235
236 async {
237 my $big_cache = new Cache ...;
238 };
239
240If there are no other references, then the C<$big_cache> object will be
241freed when the thread terminates, regardless of how it does so.
242
243What it does C<NOT> do is unlock any Coro::Semaphores or similar
244resources, but that's where the C<guard> methods come in handy:
245
246 my $sem = new Coro::Semaphore;
247
248 async {
249 my $lock_guard = $sem->guard;
250 # if we reutrn, or die or get cancelled, here,
251 # then the semaphore will be "up"ed.
252 };
253
254The C<Guard::guard> function comes in handy for any custom cleanup you
255might want to do:
256
257 async {
258 my $window = new Gtk2::Window "toplevel";
259 # The window will not be cleaned up automatically, even when $window
260 # gets freed, so use a guard to ensure it's destruction
261 # in case of an error:
262 my $window_guard = Guard::guard { $window->destroy };
263
264 # we are safe here
265 };
266
267Last not least, C<local> can often be handy, too, e.g. when temporarily
268replacing the coro thread description:
269
270 sub myfunction {
271 local $Coro::current->{desc} = "inside myfunction(@_)";
272
273 # if we return or die here, the description will be restored
274 }
275
276=item 6. Viva La Zombie Muerte
277
278Even after a thread has terminated and cleaned up it's resources, the coro
279object still is there and stores the return values of the thread. Only in
280this state will the coro object be "reference counted" in the normal perl
281sense: the thread code keeps a reference to it when it is active, but not
282after it has terminated.
283
284The means the coro object gets freed automatically when the thread has
285terminated and cleaned up and there arenot other references.
286
287If there are, the coro object will stay around, and you can call C<<
288->join >> as many times as you wish to retrieve the result values:
289
290 async {
291 print "hi\n";
292 1
293 };
294
295 # run the async above, and free everything before returning
296 # from Coro::cede:
297 Coro::cede;
298
299 {
300 my $coro = async {
301 print "hi\n";
302 1
303 };
304
305 # run the async above, and clean up, but do not free the coro
306 # object:
307 Coro::cede;
308
309 # optionally retrieve the result values
310 my @results = $coro->join;
311
312 # now $coro goes out of scope, and presumably gets freed
313 };
314
315=back
316
68=cut 317=cut
69 318
70package Coro; 319package Coro;
71 320
72use common::sense; 321use common::sense;
81 330
82our $idle; # idle handler 331our $idle; # idle handler
83our $main; # main coro 332our $main; # main coro
84our $current; # current coro 333our $current; # current coro
85 334
86our $VERSION = 5.21; 335our $VERSION = 5.372;
87 336
88our @EXPORT = qw(async async_pool cede schedule terminate current unblock_sub rouse_cb rouse_wait); 337our @EXPORT = qw(async async_pool cede schedule terminate current unblock_sub rouse_cb rouse_wait);
89our %EXPORT_TAGS = ( 338our %EXPORT_TAGS = (
90 prio => [qw(PRIO_MAX PRIO_HIGH PRIO_NORMAL PRIO_LOW PRIO_IDLE PRIO_MIN)], 339 prio => [qw(PRIO_MAX PRIO_HIGH PRIO_NORMAL PRIO_LOW PRIO_IDLE PRIO_MIN)],
91); 340);
131 380
132The default implementation dies with "FATAL: deadlock detected.", followed 381The default implementation dies with "FATAL: deadlock detected.", followed
133by a thread listing, because the program has no other way to continue. 382by a thread listing, because the program has no other way to continue.
134 383
135This hook is overwritten by modules such as C<Coro::EV> and 384This hook is overwritten by modules such as C<Coro::EV> and
136C<Coro::AnyEvent> to wait on an external event that hopefully wake up a 385C<Coro::AnyEvent> to wait on an external event that hopefully wakes up a
137coro so the scheduler can run it. 386coro so the scheduler can run it.
138 387
139See L<Coro::EV> or L<Coro::AnyEvent> for examples of using this technique. 388See L<Coro::EV> or L<Coro::AnyEvent> for examples of using this technique.
140 389
141=cut 390=cut
142 391
392# ||= because other modules could have provided their own by now
143$idle = new Coro sub { 393$idle ||= new Coro sub {
144 require Coro::Debug; 394 require Coro::Debug;
145 die "FATAL: deadlock detected.\n" 395 die "FATAL: deadlock detected.\n"
146 . Coro::Debug::ps_listing (); 396 . Coro::Debug::ps_listing ();
147}; 397};
148 398
151our @destroy; 401our @destroy;
152our $manager; 402our $manager;
153 403
154$manager = new Coro sub { 404$manager = new Coro sub {
155 while () { 405 while () {
156 Coro::State::cancel shift @destroy 406 _destroy shift @destroy
157 while @destroy; 407 while @destroy;
158 408
159 &schedule; 409 &schedule;
160 } 410 }
161}; 411};
295coro, regardless of priority. This is useful sometimes to ensure 545coro, regardless of priority. This is useful sometimes to ensure
296progress is made. 546progress is made.
297 547
298=item terminate [arg...] 548=item terminate [arg...]
299 549
300Terminates the current coro with the given status values (see L<cancel>). 550Terminates the current coro with the given status values (see
551L<cancel>). The values will not be copied, but referenced directly.
301 552
302=item Coro::on_enter BLOCK, Coro::on_leave BLOCK 553=item Coro::on_enter BLOCK, Coro::on_leave BLOCK
303 554
304These function install enter and leave winders in the current scope. The 555These function install enter and leave winders in the current scope. The
305enter block will be executed when on_enter is called and whenever the 556enter block will be executed when on_enter is called and whenever the
479Returns true iff this Coro object has been suspended. Suspended Coros will 730Returns true iff this Coro object has been suspended. Suspended Coros will
480not ever be scheduled. 731not ever be scheduled.
481 732
482=item $coro->cancel (arg...) 733=item $coro->cancel (arg...)
483 734
484Terminates the given Coro and makes it return the given arguments as 735Terminates the given Coro thread and makes it return the given arguments as
485status (default: the empty list). Never returns if the Coro is the 736status (default: an empty list). Never returns if the Coro is the
486current Coro. 737current Coro.
487 738
488=cut 739This is a rather brutal way to free a coro, with some limitations - if
740the thread is inside a C callback that doesn't expect to be canceled,
741bad things can happen, or if the cancelled thread insists on running
742complicated cleanup handlers that rely on it'S thread context, things will
743not work.
489 744
490sub cancel { 745Sometimes it is safer to C<< ->throw >> an exception, or use C<<
491 my $self = shift; 746->safe_cancel >>.
492 747
493 if ($current == $self) { 748The arguments are not copied, but instead will be referenced directly
494 terminate @_; 749(e.g. if you pass C<$var> and after the call change that variable, then
495 } else { 750you might change the return values passed to e.g. C<join>, so don't do
496 $self->{_status} = [@_]; 751that).
497 Coro::State::cancel $self; 752
753The resources of the Coro are usually freed (or destructed) before this
754call returns, but this can be delayed for an indefinite amount of time, as
755in some cases the manager thread has to run first to actually destruct the
756Coro object.
757
758=item $coro->safe_cancel ($arg...)
759
760Works mostly like C<< ->cancel >>, but is inherently "safer", and
761consequently, can fail with an exception in cases the thread is not in a
762cancellable state.
763
764This method works a bit like throwing an exception that cannot be caught
765- specifically, it will clean up the thread from within itself, so all
766cleanup handlers (e.g. C<guard> blocks) are run with full thread context
767and can block if they wish.
768
769A thread is safe-cancellable if it either hasn't been run yet, or
770it has no C context attached and is inside an SLF function.
771
772The latter two basically mean that the thread isn't currently inside a
773perl callback called from some C function (usually XS modules) and isn't
774currently inside some C function itself.
775
776This call always returns true when it could cancel the thread, or croaks
777with an error otherwise, so you can write things like this:
778
779 if (! eval { $coro->safe_cancel }) {
780 warn "unable to cancel thread: $@";
498 } 781 }
499}
500 782
501=item $coro->schedule_to 783=item $coro->schedule_to
502 784
503Puts the current coro to sleep (like C<Coro::schedule>), but instead 785Puts the current coro to sleep (like C<Coro::schedule>), but instead
504of continuing with the next coro from the ready queue, always switch to 786of continuing with the next coro from the ready queue, always switch to
542 824
543=item $coro->join 825=item $coro->join
544 826
545Wait until the coro terminates and return any values given to the 827Wait until the coro terminates and return any values given to the
546C<terminate> or C<cancel> functions. C<join> can be called concurrently 828C<terminate> or C<cancel> functions. C<join> can be called concurrently
547from multiple coro, and all will be resumed and given the status 829from multiple threads, and all will be resumed and given the status
548return once the C<$coro> terminates. 830return once the C<$coro> terminates.
549 831
550=cut 832=cut
551 833
552sub join { 834sub join {
566 wantarray ? @{$self->{_status}} : $self->{_status}[0]; 848 wantarray ? @{$self->{_status}} : $self->{_status}[0];
567} 849}
568 850
569=item $coro->on_destroy (\&cb) 851=item $coro->on_destroy (\&cb)
570 852
571Registers a callback that is called when this coro gets destroyed, 853Registers a callback that is called when this coro thread gets destroyed,
572but before it is joined. The callback gets passed the terminate arguments, 854that is, after it's resources have been freed but before it is joined. The
855callback gets passed the terminate/cancel arguments, if any, and I<must
573if any, and I<must not> die, under any circumstances. 856not> die, under any circumstances.
857
858There can be any number of C<on_destroy> callbacks per coro, and there is
859no way currently to remove a callback once added.
574 860
575=cut 861=cut
576 862
577sub on_destroy { 863sub on_destroy {
578 my ($self, $cb) = @_; 864 my ($self, $cb) = @_;
581} 867}
582 868
583=item $oldprio = $coro->prio ($newprio) 869=item $oldprio = $coro->prio ($newprio)
584 870
585Sets (or gets, if the argument is missing) the priority of the 871Sets (or gets, if the argument is missing) the priority of the
586coro. Higher priority coro get run before lower priority 872coro thread. Higher priority coro get run before lower priority
587coro. Priorities are small signed integers (currently -4 .. +3), 873coros. Priorities are small signed integers (currently -4 .. +3),
588that you can refer to using PRIO_xxx constants (use the import tag :prio 874that you can refer to using PRIO_xxx constants (use the import tag :prio
589to get then): 875to get then):
590 876
591 PRIO_MAX > PRIO_HIGH > PRIO_NORMAL > PRIO_LOW > PRIO_IDLE > PRIO_MIN 877 PRIO_MAX > PRIO_HIGH > PRIO_NORMAL > PRIO_LOW > PRIO_IDLE > PRIO_MIN
592 3 > 1 > 0 > -1 > -3 > -4 878 3 > 1 > 0 > -1 > -3 > -4
593 879
594 # set priority to HIGH 880 # set priority to HIGH
595 current->prio (PRIO_HIGH); 881 current->prio (PRIO_HIGH);
596 882
597The idle coro ($Coro::idle) always has a lower priority than any 883The idle coro thread ($Coro::idle) always has a lower priority than any
598existing coro. 884existing coro.
599 885
600Changing the priority of the current coro will take effect immediately, 886Changing the priority of the current coro will take effect immediately,
601but changing the priority of coro in the ready queue (but not 887but changing the priority of a coro in the ready queue (but not running)
602running) will only take effect after the next schedule (of that 888will only take effect after the next schedule (of that coro). This is a
603coro). This is a bug that will be fixed in some future version. 889bug that will be fixed in some future version.
604 890
605=item $newprio = $coro->nice ($change) 891=item $newprio = $coro->nice ($change)
606 892
607Similar to C<prio>, but subtract the given value from the priority (i.e. 893Similar to C<prio>, but subtract the given value from the priority (i.e.
608higher values mean lower priority, just as in unix). 894higher values mean lower priority, just as in UNIX's nice command).
609 895
610=item $olddesc = $coro->desc ($newdesc) 896=item $olddesc = $coro->desc ($newdesc)
611 897
612Sets (or gets in case the argument is missing) the description for this 898Sets (or gets in case the argument is missing) the description for this
613coro. This is just a free-form string you can associate with a 899coro thread. This is just a free-form string you can associate with a
614coro. 900coro.
615 901
616This method simply sets the C<< $coro->{desc} >> member to the given 902This method simply sets the C<< $coro->{desc} >> member to the given
617string. You can modify this member directly if you wish. 903string. You can modify this member directly if you wish, and in fact, this
904is often preferred to indicate major processing states that cna then be
905seen for example in a L<Coro::Debug> session:
906
907 sub my_long_function {
908 local $Coro::current->{desc} = "now in my_long_function";
909 ...
910 $Coro::current->{desc} = "my_long_function: phase 1";
911 ...
912 $Coro::current->{desc} = "my_long_function: phase 2";
913 ...
914 }
618 915
619=cut 916=cut
620 917
621sub desc { 918sub desc {
622 my $old = $_[0]{desc}; 919 my $old = $_[0]{desc};
659returning a new coderef. Unblocking means that calling the new coderef 956returning a new coderef. Unblocking means that calling the new coderef
660will return immediately without blocking, returning nothing, while the 957will return immediately without blocking, returning nothing, while the
661original code ref will be called (with parameters) from within another 958original code ref will be called (with parameters) from within another
662coro. 959coro.
663 960
664The reason this function exists is that many event libraries (such as the 961The reason this function exists is that many event libraries (such as
665venerable L<Event|Event> module) are not thread-safe (a weaker form 962the venerable L<Event|Event> module) are not thread-safe (a weaker form
666of reentrancy). This means you must not block within event callbacks, 963of reentrancy). This means you must not block within event callbacks,
667otherwise you might suffer from crashes or worse. The only event library 964otherwise you might suffer from crashes or worse. The only event library
668currently known that is safe to use without C<unblock_sub> is L<EV>. 965currently known that is safe to use without C<unblock_sub> is L<EV> (but
966you might still run into deadlocks if all event loops are blocked).
669 967
670Coro will try to catch you when you block in the event loop 968Coro will try to catch you when you block in the event loop
671("FATAL:$Coro::IDLE blocked itself"), but this is just best effort and 969("FATAL:$Coro::IDLE blocked itself"), but this is just best effort and
672only works when you do not run your own event loop. 970only works when you do not run your own event loop.
673 971
744 1042
745=back 1043=back
746 1044
747=cut 1045=cut
748 1046
1047for my $module (qw(Channel RWLock Semaphore SemaphoreSet Signal Specific)) {
1048 my $old = defined &{"Coro::$module\::new"} && \&{"Coro::$module\::new"};
1049
1050 *{"Coro::$module\::new"} = sub {
1051 require "Coro/$module.pm";
1052
1053 # some modules have their new predefined in State.xs, some don't
1054 *{"Coro::$module\::new"} = $old
1055 if $old;
1056
1057 goto &{"Coro::$module\::new"};
1058 };
1059}
1060
7491; 10611;
750 1062
751=head1 HOW TO WAIT FOR A CALLBACK 1063=head1 HOW TO WAIT FOR A CALLBACK
752 1064
753It is very common for a coro to wait for some callback to be 1065It is very common for a coro to wait for some callback to be
856ithreads (for example, that memory or files would be shared), showing his 1168ithreads (for example, that memory or files would be shared), showing his
857lack of understanding of this area - if it is hard to understand for Chip, 1169lack of understanding of this area - if it is hard to understand for Chip,
858it is probably not obvious to everybody). 1170it is probably not obvious to everybody).
859 1171
860What follows is an ultra-condensed version of my talk about threads in 1172What follows is an ultra-condensed version of my talk about threads in
861scripting languages given onthe perl workshop 2009: 1173scripting languages given on the perl workshop 2009:
862 1174
863The so-called "ithreads" were originally implemented for two reasons: 1175The so-called "ithreads" were originally implemented for two reasons:
864first, to (badly) emulate unix processes on native win32 perls, and 1176first, to (badly) emulate unix processes on native win32 perls, and
865secondly, to replace the older, real thread model ("5.005-threads"). 1177secondly, to replace the older, real thread model ("5.005-threads").
866 1178

Diff Legend

Removed lines
+ Added lines
< Changed lines
> Changed lines