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.104 by root, Thu Jan 4 23:49:27 2007 UTC vs.
Revision 1.196 by root, Sat Aug 30 03:07:46 2008 UTC

2 2
3Coro - coroutine process abstraction 3Coro - coroutine process abstraction
4 4
5=head1 SYNOPSIS 5=head1 SYNOPSIS
6 6
7 use Coro; 7 use Coro;
8 8
9 async { 9 async {
10 # some asynchronous thread of execution 10 # some asynchronous thread of execution
11 print "2\n";
12 cede; # yield back to main
13 print "4\n";
11 }; 14 };
12 15 print "1\n";
13 # alternatively create an async coroutine like this: 16 cede; # yield to coroutine
14 17 print "3\n";
15 sub some_func : Coro { 18 cede; # and again
16 # some more async code 19
17 } 20 # use locking
18 21 my $lock = new Coro::Semaphore;
19 cede; 22 my $locked;
23
24 $lock->down;
25 $locked = 1;
26 $lock->up;
20 27
21=head1 DESCRIPTION 28=head1 DESCRIPTION
22 29
23This module collection manages coroutines. Coroutines are similar 30This module collection manages coroutines. Coroutines are similar to
24to threads but don't run in parallel at the same time even on SMP 31threads but don't (in general) run in parallel at the same time even
25machines. The specific flavor of coroutine use din this module also 32on SMP machines. The specific flavor of coroutine used in this module
26guarentees you that it will not switch between coroutines unless 33also guarantees you that it will not switch between coroutines unless
27necessary, at easily-identified points in your program, so locking and 34necessary, at easily-identified points in your program, so locking and
28parallel access are rarely an issue, making coroutine programming much 35parallel access are rarely an issue, making coroutine programming much
29safer than threads programming. 36safer and easier than threads programming.
30 37
31(Perl, however, does not natively support real threads but instead does a 38Unlike a normal perl program, however, coroutines allow you to have
32very slow and memory-intensive emulation of processes using threads. This 39multiple running interpreters that share data, which is especially useful
33is a performance win on Windows machines, and a loss everywhere else). 40to code pseudo-parallel processes and for event-based programming, such as
41multiple HTTP-GET requests running concurrently. See L<Coro::AnyEvent> to
42learn more.
43
44Coroutines are also useful because Perl has no support for threads (the so
45called "threads" that perl offers are nothing more than the (bad) process
46emulation coming from the Windows platform: On standard operating systems
47they serve no purpose whatsoever, except by making your programs slow and
48making them use a lot of memory. Best disable them when building perl, or
49aks your software vendor/distributor to do it for you).
34 50
35In this module, coroutines are defined as "callchain + lexical variables + 51In this module, coroutines are defined as "callchain + lexical variables +
36@_ + $_ + $@ + $/ + C stack), that is, a coroutine has its own callchain, 52@_ + $_ + $@ + $/ + C stack), that is, a coroutine has its own callchain,
37its own set of lexicals and its own set of perls most important global 53its own set of lexicals and its own set of perls most important global
38variables. 54variables (see L<Coro::State> for more configuration).
39 55
40=cut 56=cut
41 57
42package Coro; 58package Coro;
43 59
50 66
51our $idle; # idle handler 67our $idle; # idle handler
52our $main; # main coroutine 68our $main; # main coroutine
53our $current; # current coroutine 69our $current; # current coroutine
54 70
55our $VERSION = '3.3'; 71our $VERSION = 4.745;
56 72
57our @EXPORT = qw(async cede schedule terminate current unblock_sub); 73our @EXPORT = qw(async async_pool cede schedule terminate current unblock_sub);
58our %EXPORT_TAGS = ( 74our %EXPORT_TAGS = (
59 prio => [qw(PRIO_MAX PRIO_HIGH PRIO_NORMAL PRIO_LOW PRIO_IDLE PRIO_MIN)], 75 prio => [qw(PRIO_MAX PRIO_HIGH PRIO_NORMAL PRIO_LOW PRIO_IDLE PRIO_MIN)],
60); 76);
61our @EXPORT_OK = (@{$EXPORT_TAGS{prio}}, qw(nready)); 77our @EXPORT_OK = (@{$EXPORT_TAGS{prio}}, qw(nready));
62 78
63{
64 my @async;
65 my $init;
66
67 # this way of handling attributes simply is NOT scalable ;()
68 sub import {
69 no strict 'refs';
70
71 Coro->export_to_level (1, @_);
72
73 my $old = *{(caller)[0]."::MODIFY_CODE_ATTRIBUTES"}{CODE};
74 *{(caller)[0]."::MODIFY_CODE_ATTRIBUTES"} = sub {
75 my ($package, $ref) = (shift, shift);
76 my @attrs;
77 for (@_) {
78 if ($_ eq "Coro") {
79 push @async, $ref;
80 unless ($init++) {
81 eval q{
82 sub INIT {
83 &async(pop @async) while @async;
84 }
85 };
86 }
87 } else {
88 push @attrs, $_;
89 }
90 }
91 return $old ? $old->($package, $ref, @attrs) : @attrs;
92 };
93 }
94
95}
96
97=over 4 79=over 4
98 80
99=item $main 81=item $Coro::main
100 82
101This coroutine represents the main program. 83This variable stores the coroutine object that represents the main
84program. While you cna C<ready> it and do most other things you can do to
85coroutines, it is mainly useful to compare again C<$Coro::current>, to see
86whether you are running in the main program or not.
102 87
103=cut 88=cut
104 89
105$main = new Coro; 90$main = new Coro;
106 91
107=item $current (or as function: current) 92=item $Coro::current
108 93
109The current coroutine (the last coroutine switched to). The initial value 94The coroutine object representing the current coroutine (the last
95coroutine that the Coro scheduler switched to). The initial value is
110is C<$main> (of course). 96C<$main> (of course).
111 97
112This variable is B<strictly> I<read-only>. It is provided for performance 98This variable is B<strictly> I<read-only>. You can take copies of the
113reasons. If performance is not essentiel you are encouraged to use the 99value stored in it and use it as any other coroutine object, but you must
114C<Coro::current> function instead. 100not otherwise modify the variable itself.
115 101
116=cut 102=cut
103
104$main->{desc} = "[main::]";
117 105
118# maybe some other module used Coro::Specific before... 106# maybe some other module used Coro::Specific before...
119$main->{specific} = $current->{specific} 107$main->{_specific} = $current->{_specific}
120 if $current; 108 if $current;
121 109
122_set_current $main; 110_set_current $main;
123 111
124sub current() { $current } 112sub current() { $current } # [DEPRECATED]
125 113
126=item $idle 114=item $Coro::idle
127 115
128A callback that is called whenever the scheduler finds no ready coroutines 116This variable is mainly useful to integrate Coro into event loops. It is
129to run. The default implementation prints "FATAL: deadlock detected" and 117usually better to rely on L<Coro::AnyEvent> or LC<Coro::EV>, as this is
130exits, because the program has no other way to continue. 118pretty low-level functionality.
119
120This variable stores a callback that is called whenever the scheduler
121finds no ready coroutines to run. The default implementation prints
122"FATAL: deadlock detected" and exits, because the program has no other way
123to continue.
131 124
132This hook is overwritten by modules such as C<Coro::Timer> and 125This 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 126C<Coro::AnyEvent> to wait on an external event that hopefully wake up a
134coroutine so the scheduler can run it. 127coroutine so the scheduler can run it.
135 128
129Note that the callback I<must not>, under any circumstances, block
130the current coroutine. Normally, this is achieved by having an "idle
131coroutine" that calls the event loop and then blocks again, and then
132readying that coroutine in the idle handler.
133
134See L<Coro::Event> or L<Coro::AnyEvent> for examples of using this
135technique.
136
136Please note that if your callback recursively invokes perl (e.g. for event 137Please note that if your callback recursively invokes perl (e.g. for event
137handlers), then it must be prepared to be called recursively. 138handlers), then it must be prepared to be called recursively itself.
138 139
139=cut 140=cut
140 141
141$idle = sub { 142$idle = sub {
142 require Carp; 143 require Carp;
149 # free coroutine data and mark as destructed 150 # free coroutine data and mark as destructed
150 $self->_destroy 151 $self->_destroy
151 or return; 152 or return;
152 153
153 # call all destruction callbacks 154 # call all destruction callbacks
154 $_->(@{$self->{status}}) 155 $_->(@{$self->{_status}})
155 for @{(delete $self->{destroy_cb}) || []}; 156 for @{(delete $self->{_on_destroy}) || []};
156} 157}
157 158
158# this coroutine is necessary because a coroutine 159# this coroutine is necessary because a coroutine
159# cannot destroy itself. 160# cannot destroy itself.
160my @destroy; 161my @destroy;
166 while @destroy; 167 while @destroy;
167 168
168 &schedule; 169 &schedule;
169 } 170 }
170}; 171};
171 172$manager->desc ("[coro manager]");
172$manager->prio (PRIO_MAX); 173$manager->prio (PRIO_MAX);
173 174
174# static methods. not really.
175
176=back 175=back
177 176
178=head2 STATIC METHODS 177=head2 SIMPLE COROUTINE CREATION
179
180Static methods are actually functions that operate on the current coroutine only.
181 178
182=over 4 179=over 4
183 180
184=item async { ... } [@args...] 181=item async { ... } [@args...]
185 182
186Create a new asynchronous coroutine and return it's coroutine object 183Create a new coroutine and return it's coroutine object (usually
187(usually unused). When the sub returns the new coroutine is automatically 184unused). The coroutine will be put into the ready queue, so
185it will start running automatically on the next scheduler run.
186
187The first argument is a codeblock/closure that should be executed in the
188coroutine. When it returns argument returns the coroutine is automatically
188terminated. 189terminated.
189 190
190Calling C<exit> in a coroutine will not work correctly, so do not do that. 191The remaining arguments are passed as arguments to the closure.
191 192
192When the coroutine dies, the program will exit, just as in the main 193See the C<Coro::State::new> constructor for info about the coroutine
193program. 194environment in which coroutines are executed.
194 195
196Calling C<exit> in a coroutine will do the same as calling exit outside
197the coroutine. Likewise, when the coroutine dies, the program will exit,
198just as it would in the main program.
199
200If you do not want that, you can provide a default C<die> handler, or
201simply avoid dieing (by use of C<eval>).
202
195 # create a new coroutine that just prints its arguments 203Example: Create a new coroutine that just prints its arguments.
204
196 async { 205 async {
197 print "@_\n"; 206 print "@_\n";
198 } 1,2,3,4; 207 } 1,2,3,4;
199 208
200=cut 209=cut
203 my $coro = new Coro @_; 212 my $coro = new Coro @_;
204 $coro->ready; 213 $coro->ready;
205 $coro 214 $coro
206} 215}
207 216
217=item async_pool { ... } [@args...]
218
219Similar to C<async>, but uses a coroutine pool, so you should not call
220terminate or join on it (although you are allowed to), and you get a
221coroutine that might have executed other code already (which can be good
222or bad :).
223
224On the plus side, this function is faster than creating (and destroying)
225a completely new coroutine, so if you need a lot of generic coroutines in
226quick successsion, use C<async_pool>, not C<async>.
227
228The code block is executed in an C<eval> context and a warning will be
229issued in case of an exception instead of terminating the program, as
230C<async> does. As the coroutine is being reused, stuff like C<on_destroy>
231will not work in the expected way, unless you call terminate or cancel,
232which somehow defeats the purpose of pooling (but is fine in the
233exceptional case).
234
235The priority will be reset to C<0> after each run, tracing will be
236disabled, the description will be reset and the default output filehandle
237gets restored, so you can change all these. Otherwise the coroutine will
238be re-used "as-is": most notably if you change other per-coroutine global
239stuff such as C<$/> you I<must needs> to revert that change, which is most
240simply done by using local as in: C< local $/ >.
241
242The pool size is limited to C<8> idle coroutines (this can be adjusted by
243changing $Coro::POOL_SIZE), and there can be as many non-idle coros as
244required.
245
246If you are concerned about pooled coroutines growing a lot because a
247single C<async_pool> used a lot of stackspace you can e.g. C<async_pool
248{ terminate }> once per second or so to slowly replenish the pool. In
249addition to that, when the stacks used by a handler grows larger than 16kb
250(adjustable via $Coro::POOL_RSS) it will also be destroyed.
251
252=cut
253
254our $POOL_SIZE = 8;
255our $POOL_RSS = 16 * 1024;
256our @async_pool;
257
258sub pool_handler {
259 my $cb;
260
261 while () {
262 eval {
263 while () {
264 _pool_1 $cb;
265 &$cb;
266 _pool_2 $cb;
267 &schedule;
268 }
269 };
270
271 if ($@) {
272 last if $@ eq "\3async_pool terminate\2\n";
273 warn $@;
274 }
275 }
276}
277
278sub async_pool(&@) {
279 # this is also inlined into the unlock_scheduler
280 my $coro = (pop @async_pool) || new Coro \&pool_handler;
281
282 $coro->{_invoke} = [@_];
283 $coro->ready;
284
285 $coro
286}
287
288=back
289
290=head2 STATIC METHODS
291
292Static methods are actually functions that operate on the current coroutine.
293
294=over 4
295
208=item schedule 296=item schedule
209 297
210Calls the scheduler. Please note that the current coroutine will not be put 298Calls the scheduler. The scheduler will find the next coroutine that is
299to be run from the ready queue and switches to it. The next coroutine
300to be run is simply the one with the highest priority that is longest
301in its ready queue. If there is no coroutine ready, it will clal the
302C<$Coro::idle> hook.
303
304Please note that the current coroutine will I<not> be put into the ready
211into the ready queue, so calling this function usually means you will 305queue, so calling this function usually means you will never be called
212never be called again unless something else (e.g. an event handler) calls 306again unless something else (e.g. an event handler) calls C<< ->ready >>,
213ready. 307thus waking you up.
308
309This makes C<schedule> I<the> generic method to use to block the current
310coroutine and wait for events: first you remember the current coroutine in
311a variable, then arrange for some callback of yours to call C<< ->ready
312>> on that once some event happens, and last you call C<schedule> to put
313yourself to sleep. Note that a lot of things can wake your coroutine up,
314so you need to check whether the event indeed happened, e.g. by storing the
315status in a variable.
214 316
215The canonical way to wait on external events is this: 317The canonical way to wait on external events is this:
216 318
217 { 319 {
218 # remember current coroutine 320 # remember current coroutine
223 # wake up sleeping coroutine 325 # wake up sleeping coroutine
224 $current->ready; 326 $current->ready;
225 undef $current; 327 undef $current;
226 }; 328 };
227 329
228 # call schedule until event occured. 330 # call schedule until event occurred.
229 # in case we are woken up for other reasons 331 # in case we are woken up for other reasons
230 # (current still defined), loop. 332 # (current still defined), loop.
231 Coro::schedule while $current; 333 Coro::schedule while $current;
232 } 334 }
233 335
234=item cede 336=item cede
235 337
236"Cede" to other coroutines. This function puts the current coroutine into the 338"Cede" to other coroutines. This function puts the current coroutine into
237ready queue and calls C<schedule>, which has the effect of giving up the 339the ready queue and calls C<schedule>, which has the effect of giving
238current "timeslice" to other coroutines of the same or higher priority. 340up the current "timeslice" to other coroutines of the same or higher
341priority. Once your coroutine gets its turn again it will automatically be
342resumed.
343
344This function is often called C<yield> in other languages.
239 345
240=item Coro::cede_notself 346=item Coro::cede_notself
241 347
242Works like cede, but is not exported by default and will cede to any 348Works like cede, but is not exported by default and will cede to I<any>
243coroutine, regardless of priority, once. 349coroutine, regardless of priority. This is useful sometimes to ensure
350progress is made.
244 351
245=item terminate [arg...] 352=item terminate [arg...]
246 353
247Terminates the current coroutine with the given status values (see L<cancel>). 354Terminates the current coroutine with the given status values (see L<cancel>).
355
356=item killall
357
358Kills/terminates/cancels all coroutines except the currently running
359one. This is useful after a fork, either in the child or the parent, as
360usually only one of them should inherit the running coroutines.
361
362Note that while this will try to free some of the main programs resources,
363you cannot free all of them, so if a coroutine that is not the main
364program calls this function, there will be some one-time resource leak.
248 365
249=cut 366=cut
250 367
251sub terminate { 368sub terminate {
252 $current->cancel (@_); 369 $current->cancel (@_);
253} 370}
254 371
372sub killall {
373 for (Coro::State::list) {
374 $_->cancel
375 if $_ != $current && UNIVERSAL::isa $_, "Coro";
376 }
377}
378
255=back 379=back
256 380
257# dynamic methods
258
259=head2 COROUTINE METHODS 381=head2 COROUTINE METHODS
260 382
261These are the methods you can call on coroutine objects. 383These are the methods you can call on coroutine objects (or to create
384them).
262 385
263=over 4 386=over 4
264 387
265=item new Coro \&sub [, @args...] 388=item new Coro \&sub [, @args...]
266 389
267Create a new coroutine and return it. When the sub returns the coroutine 390Create a new coroutine and return it. When the sub returns, the coroutine
268automatically terminates as if C<terminate> with the returned values were 391automatically terminates as if C<terminate> with the returned values were
269called. To make the coroutine run you must first put it into the ready queue 392called. To make the coroutine run you must first put it into the ready
270by calling the ready method. 393queue by calling the ready method.
271 394
272Calling C<exit> in a coroutine will not work correctly, so do not do that. 395See C<async> and C<Coro::State::new> for additional info about the
396coroutine environment.
273 397
274=cut 398=cut
275 399
276sub _run_coro { 400sub _run_coro {
277 terminate &{+shift}; 401 terminate &{+shift};
283 $class->SUPER::new (\&_run_coro, @_) 407 $class->SUPER::new (\&_run_coro, @_)
284} 408}
285 409
286=item $success = $coroutine->ready 410=item $success = $coroutine->ready
287 411
288Put the given coroutine into the ready queue (according to it's priority) 412Put the given coroutine into the end of its ready queue (there is one
289and return true. If the coroutine is already in the ready queue, do nothing 413queue for each priority) and return true. If the coroutine is already in
290and return false. 414the ready queue, do nothing and return false.
415
416This ensures that the scheduler will resume this coroutine automatically
417once all the coroutines of higher priority and all coroutines of the same
418priority that were put into the ready queue earlier have been resumed.
291 419
292=item $is_ready = $coroutine->is_ready 420=item $is_ready = $coroutine->is_ready
293 421
294Return wether the coroutine is currently the ready queue or not, 422Return whether the coroutine is currently the ready queue or not,
295 423
296=item $coroutine->cancel (arg...) 424=item $coroutine->cancel (arg...)
297 425
298Terminates the given coroutine and makes it return the given arguments as 426Terminates the given coroutine and makes it return the given arguments as
299status (default: the empty list). Never returns if the coroutine is the 427status (default: the empty list). Never returns if the coroutine is the
301 429
302=cut 430=cut
303 431
304sub cancel { 432sub cancel {
305 my $self = shift; 433 my $self = shift;
306 $self->{status} = [@_]; 434 $self->{_status} = [@_];
307 435
308 if ($current == $self) { 436 if ($current == $self) {
309 push @destroy, $self; 437 push @destroy, $self;
310 $manager->ready; 438 $manager->ready;
311 &schedule while 1; 439 &schedule while 1;
315} 443}
316 444
317=item $coroutine->join 445=item $coroutine->join
318 446
319Wait until the coroutine terminates and return any values given to the 447Wait until the coroutine terminates and return any values given to the
320C<terminate> or C<cancel> functions. C<join> can be called multiple times 448C<terminate> or C<cancel> functions. C<join> can be called concurrently
321from multiple coroutine. 449from multiple coroutines, and all will be resumed and given the status
450return once the C<$coroutine> terminates.
322 451
323=cut 452=cut
324 453
325sub join { 454sub join {
326 my $self = shift; 455 my $self = shift;
327 456
328 unless ($self->{status}) { 457 unless ($self->{_status}) {
329 my $current = $current; 458 my $current = $current;
330 459
331 push @{$self->{destroy_cb}}, sub { 460 push @{$self->{_on_destroy}}, sub {
332 $current->ready; 461 $current->ready;
333 undef $current; 462 undef $current;
334 }; 463 };
335 464
336 &schedule while $current; 465 &schedule while $current;
337 } 466 }
338 467
339 wantarray ? @{$self->{status}} : $self->{status}[0]; 468 wantarray ? @{$self->{_status}} : $self->{_status}[0];
340} 469}
341 470
342=item $coroutine->on_destroy (\&cb) 471=item $coroutine->on_destroy (\&cb)
343 472
344Registers a callback that is called when this coroutine gets destroyed, 473Registers a callback that is called when this coroutine gets destroyed,
345but before it is joined. The callback gets passed the terminate arguments, 474but before it is joined. The callback gets passed the terminate arguments,
346if any. 475if any, and I<must not> die, under any circumstances.
347 476
348=cut 477=cut
349 478
350sub on_destroy { 479sub on_destroy {
351 my ($self, $cb) = @_; 480 my ($self, $cb) = @_;
352 481
353 push @{ $self->{destroy_cb} }, $cb; 482 push @{ $self->{_on_destroy} }, $cb;
354} 483}
355 484
356=item $oldprio = $coroutine->prio ($newprio) 485=item $oldprio = $coroutine->prio ($newprio)
357 486
358Sets (or gets, if the argument is missing) the priority of the 487Sets (or gets, if the argument is missing) the priority of the
383=item $olddesc = $coroutine->desc ($newdesc) 512=item $olddesc = $coroutine->desc ($newdesc)
384 513
385Sets (or gets in case the argument is missing) the description for this 514Sets (or gets in case the argument is missing) the description for this
386coroutine. This is just a free-form string you can associate with a coroutine. 515coroutine. This is just a free-form string you can associate with a coroutine.
387 516
517This method simply sets the C<< $coroutine->{desc} >> member to the given string. You
518can modify this member directly if you wish.
519
520=item $coroutine->throw ([$scalar])
521
522If C<$throw> is specified and defined, it will be thrown as an exception
523inside the coroutine at the next convinient point in time (usually after
524it gains control at the next schedule/transfer/cede). Otherwise clears the
525exception object.
526
527The exception object will be thrown "as is" with the specified scalar in
528C<$@>, i.e. if it is a string, no line number or newline will be appended
529(unlike with C<die>).
530
531This can be used as a softer means than C<cancel> to ask a coroutine to
532end itself, although there is no guarentee that the exception will lead to
533termination, and if the exception isn't caught it might well end the whole
534program.
535
388=cut 536=cut
389 537
390sub desc { 538sub desc {
391 my $old = $_[0]{desc}; 539 my $old = $_[0]{desc};
392 $_[0]{desc} = $_[1] if @_ > 1; 540 $_[0]{desc} = $_[1] if @_ > 1;
400=over 4 548=over 4
401 549
402=item Coro::nready 550=item Coro::nready
403 551
404Returns the number of coroutines that are currently in the ready state, 552Returns the number of coroutines that are currently in the ready state,
405i.e. that can be swicthed to. The value C<0> means that the only runnable 553i.e. that can be switched to by calling C<schedule> directory or
554indirectly. The value C<0> means that the only runnable coroutine is the
406coroutine is the currently running one, so C<cede> would have no effect, 555currently running one, so C<cede> would have no effect, and C<schedule>
407and C<schedule> would cause a deadlock unless there is an idle handler 556would cause a deadlock unless there is an idle handler that wakes up some
408that wakes up some coroutines. 557coroutines.
409 558
410=item my $guard = Coro::guard { ... } 559=item my $guard = Coro::guard { ... }
411 560
412This creates and returns a guard object. Nothing happens until the objetc 561This creates and returns a guard object. Nothing happens until the object
413gets destroyed, in which case the codeblock given as argument will be 562gets 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 563executed. 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 564runtime error or when the coroutine gets canceled, as in both cases the
416guard block will be executed. The guard object supports only one method, 565guard block will be executed. The guard object supports only one method,
417C<< ->cancel >>, which will keep the codeblock from being executed. 566C<< ->cancel >>, which will keep the codeblock from being executed.
442 591
443 592
444=item unblock_sub { ... } 593=item unblock_sub { ... }
445 594
446This utility function takes a BLOCK or code reference and "unblocks" it, 595This utility function takes a BLOCK or code reference and "unblocks" it,
447returning the new coderef. This means that the new coderef will return 596returning a new coderef. Unblocking means that calling the new coderef
448immediately without blocking, returning nothing, while the original code 597will return immediately without blocking, returning nothing, while the
449ref will be called (with parameters) from within its own coroutine. 598original code ref will be called (with parameters) from within another
599coroutine.
450 600
451The reason this fucntion exists is that many event libraries (such as the 601The reason this function exists is that many event libraries (such as the
452venerable L<Event|Event> module) are not coroutine-safe (a weaker form 602venerable L<Event|Event> module) are not coroutine-safe (a weaker form
453of thread-safety). This means you must not block within event callbacks, 603of thread-safety). This means you must not block within event callbacks,
454otherwise you might suffer from crashes or worse. 604otherwise you might suffer from crashes or worse. The only event library
605currently known that is safe to use without C<unblock_sub> is L<EV>.
455 606
456This function allows your callbacks to block by executing them in another 607This function allows your callbacks to block by executing them in another
457coroutine where it is safe to block. One example where blocking is handy 608coroutine where it is safe to block. One example where blocking is handy
458is when you use the L<Coro::AIO|Coro::AIO> functions to save results to 609is when you use the L<Coro::AIO|Coro::AIO> functions to save results to
459disk. 610disk, for example.
460 611
461In short: simply use C<unblock_sub { ... }> instead of C<sub { ... }> when 612In short: simply use C<unblock_sub { ... }> instead of C<sub { ... }> when
462creating event callbacks that want to block. 613creating event callbacks that want to block.
463 614
464=cut 615If your handler does not plan to block (e.g. simply sends a message to
616another coroutine, or puts some other coroutine into the ready queue),
617there is no reason to use C<unblock_sub>.
465 618
466our @unblock_pool; 619Note that you also need to use C<unblock_sub> for any other callbacks that
620are indirectly executed by any C-based event loop. For example, when you
621use a module that uses L<AnyEvent> (and you use L<Coro::AnyEvent>) and it
622provides callbacks that are the result of some event callback, then you
623must not block either, or use C<unblock_sub>.
624
625=cut
626
467our @unblock_queue; 627our @unblock_queue;
468our $UNBLOCK_POOL_SIZE = 2;
469 628
470sub unblock_handler_ { 629# we create a special coro because we want to cede,
471 while () { 630# to reduce pressure on the coro pool (because most callbacks
472 my ($cb, @arg) = @{ delete $Coro::current->{arg} }; 631# return immediately and can be reused) and because we cannot cede
473 $cb->(@arg); 632# inside an event callback.
474
475 last if @unblock_pool >= $UNBLOCK_POOL_SIZE;
476 push @unblock_pool, $Coro::current;
477 schedule;
478 }
479}
480
481our $unblock_scheduler = async { 633our $unblock_scheduler = new Coro sub {
482 while () { 634 while () {
483 while (my $cb = pop @unblock_queue) { 635 while (my $cb = pop @unblock_queue) {
484 my $handler = (pop @unblock_pool or new Coro \&unblock_handler_); 636 # this is an inlined copy of async_pool
485 $handler->{arg} = $cb; 637 my $coro = (pop @async_pool) || new Coro \&pool_handler;
638
639 $coro->{_invoke} = $cb;
486 $handler->ready; 640 $coro->ready;
487 cede; 641 cede; # for short-lived callbacks, this reduces pressure on the coro pool
488 } 642 }
489 643 schedule; # sleep well
490 schedule;
491 } 644 }
492}; 645};
646$unblock_scheduler->desc ("[unblock_sub scheduler]");
493 647
494sub unblock_sub(&) { 648sub unblock_sub(&) {
495 my $cb = shift; 649 my $cb = shift;
496 650
497 sub { 651 sub {
498 push @unblock_queue, [$cb, @_]; 652 unshift @unblock_queue, [$cb, @_];
499 $unblock_scheduler->ready; 653 $unblock_scheduler->ready;
500 } 654 }
501} 655}
502 656
503=back 657=back
506 660
5071; 6611;
508 662
509=head1 BUGS/LIMITATIONS 663=head1 BUGS/LIMITATIONS
510 664
511 - you must make very sure that no coro is still active on global
512 destruction. very bad things might happen otherwise (usually segfaults).
513
514 - this module is not thread-safe. You should only ever use this module 665This module is not perl-pseudo-thread-safe. You should only ever use this
515 from the same thread (this requirement might be losened in the future 666module from the same thread (this requirement might be removed in the
516 to allow per-thread schedulers, but Coro::State does not yet allow 667future to allow per-thread schedulers, but Coro::State does not yet allow
517 this). 668this). I recommend disabling thread support and using processes, as this
669is much faster and uses less memory.
518 670
519=head1 SEE ALSO 671=head1 SEE ALSO
520 672
673Event-Loop integration: L<Coro::AnyEvent>, L<Coro::EV>, L<Coro::Event>.
674
675Debugging: L<Coro::Debug>.
676
521Support/Utility: L<Coro::Cont>, L<Coro::Specific>, L<Coro::State>, L<Coro::Util>. 677Support/Utility: L<Coro::Specific>, L<Coro::Util>.
522 678
523Locking/IPC: L<Coro::Signal>, L<Coro::Channel>, L<Coro::Semaphore>, L<Coro::SemaphoreSet>, L<Coro::RWLock>. 679Locking/IPC: L<Coro::Signal>, L<Coro::Channel>, L<Coro::Semaphore>, L<Coro::SemaphoreSet>, L<Coro::RWLock>.
524 680
525Event/IO: L<Coro::Timer>, L<Coro::Event>, L<Coro::Handle>, L<Coro::Socket>, L<Coro::Select>. 681IO/Timers: L<Coro::Timer>, L<Coro::Handle>, L<Coro::Socket>, L<Coro::AIO>.
526 682
527Embedding: L<Coro:MakeMaker> 683Compatibility: L<Coro::LWP>, L<Coro::BDB>, L<Coro::Storable>, L<Coro::Select>.
684
685XS API: L<Coro::MakeMaker>.
686
687Low level Configuration, Coroutine Environment: L<Coro::State>.
528 688
529=head1 AUTHOR 689=head1 AUTHOR
530 690
531 Marc Lehmann <schmorp@schmorp.de> 691 Marc Lehmann <schmorp@schmorp.de>
532 http://home.schmorp.de/ 692 http://home.schmorp.de/

Diff Legend

Removed lines
+ Added lines
< Changed lines
> Changed lines