[ViewVC] Diff of: cvs/Coro/Coro.pm

Comparing Coro/Coro.pm (file contents):
Revision 1.88 by root, Sun Nov 26 02:54:55 2006 UTC vs.
Revision 1.102 by root, Fri Dec 29 11:37:49 2006 UTC

  async {
     # some asynchronous thread of execution
  };
- # alternatively create an async process like this:
+ # alternatively create an async coroutine like this:
  sub some_func : Coro {
     # some more async code
  }
  cede;
 =head1 DESCRIPTION
-This module collection manages coroutines. Coroutines are similar to
+This module collection manages coroutines. Coroutines are similar
-threads but don't run in parallel.
+to threads but don't run in parallel at the same time even on SMP
+machines. The specific flavor of coroutine use din this module also
+guarentees you that it will not switch between coroutines unless
+necessary, at easily-identified points in your program, so locking and
+parallel access are rarely an issue, making coroutine programming much
+safer than threads programming.
+(Perl, however, does not natively support real threads but instead does a
+very slow and memory-intensive emulation of processes using threads. This
+is a performance win on Windows machines, and a loss everywhere else).
-In this module, coroutines are defined as "callchain + lexical variables
+In this module, coroutines are defined as "callchain + lexical variables +
-+ @_ + $_ + $@ + $^W + C stack), that is, a coroutine has it's own
+@_ + $_ + $@ + $/ + C stack), that is, a coroutine has its own callchain,
-callchain, it's own set of lexicals and it's own set of perl's most
+its own set of lexicals and its own set of perls most important global
-important global variables.
+variables.
 =cut
 package Coro;
 our $idle;    # idle handler
 our $main;    # main coroutine
 our $current; # current coroutine
-our $VERSION = '3.0';
+our $VERSION = '3.3';
-our @EXPORT = qw(async cede schedule terminate current);
+our @EXPORT = qw(async cede schedule terminate current unblock_sub);
 our %EXPORT_TAGS = (
       prio => [qw(PRIO_MAX PRIO_HIGH PRIO_NORMAL PRIO_LOW PRIO_IDLE PRIO_MIN)],
 );
-our @EXPORT_OK = @{$EXPORT_TAGS{prio}};
+our @EXPORT_OK = (@{$EXPORT_TAGS{prio}}, qw(nready));
 {
    my @async;
    my $init;
    # this way of handling attributes simply is NOT scalable ;()
    sub import {
       no strict 'refs';
-      Coro->export_to_level(1, @_);
+      Coro->export_to_level (1, @_);
       my $old = *{(caller)[0]."::MODIFY_CODE_ATTRIBUTES"}{CODE};
       *{(caller)[0]."::MODIFY_CODE_ATTRIBUTES"} = sub {
          my ($package, $ref) = (shift, shift);
          my @attrs;
 C<Coro::current> function instead.
 =cut
 # maybe some other module used Coro::Specific before...
-if ($current) {
-   $main->{specific} = $current->{specific};
+$main->{specific} = $current->{specific}
-}
+   if $current;
-$current = $main;
+_set_current $main;
 sub current() { $current }
 =item $idle
 A callback that is called whenever the scheduler finds no ready coroutines
 to run. The default implementation prints "FATAL: deadlock detected" and
-exits.
+exits, because the program has no other way to continue.
 This hook is overwritten by modules such as C<Coro::Timer> and
-C<Coro::Event> to wait on an external event that hopefully wakes up some
+C<Coro::Event> to wait on an external event that hopefully wake up a
-coroutine.
+coroutine so the scheduler can run it.
+Please note that if your callback recursively invokes perl (e.g. for event
+handlers), then it must be prepared to be called recursively.
 =cut
 $idle = sub {
-   print STDERR "FATAL: deadlock detected\n";
+   require Carp;
-   exit (51);
+   Carp::croak ("FATAL: deadlock detected");
 };
 # this coroutine is necessary because a coroutine
 # cannot destroy itself.
 my @destroy;
       # been readied multiple times. this is harmless since the manager
       # can be called as many times as neccessary and will always
       # remove itself from the runqueue
       while (@destroy) {
          my $coro = pop @destroy;
          $coro->{status} ||= [];
-         $_->ready for @{delete $coro->{join} || []};
+         $_->ready                for @{(delete $coro->{join}      ) || []};
+         $_->(@{$coro->{status}}) for @{(delete $coro->{destroy_cb}) || []};
          # the next line destroys the coro state, but keeps the
-         # process itself intact (we basically make it a zombie
+         # coroutine itself intact (we basically make it a zombie
-         # process that always runs the manager thread, so it's possible
+         # coroutine that always runs the manager thread, so it's possible
-         # to transfer() to this process).
+         # to transfer() to this coroutine).
          $coro->_clone_state_from ($manager);
       }
       &schedule;
    }
 };
 =back
 =head2 STATIC METHODS
-Static methods are actually functions that operate on the current process only.
+Static methods are actually functions that operate on the current coroutine only.
 =over 4
 =item async { ... } [@args...]
-Create a new asynchronous process and return it's process object
+Create a new asynchronous coroutine and return it's coroutine object
-(usually unused). When the sub returns the new process is automatically
+(usually unused). When the sub returns the new coroutine is automatically
 terminated.
+Calling C<exit> in a coroutine will not work correctly, so do not do that.
 When the coroutine dies, the program will exit, just as in the main
 program.
    # create a new coroutine that just prints its arguments
    $pid
 }
 =item schedule
-Calls the scheduler. Please note that the current process will not be put
+Calls the scheduler. Please note that the current coroutine will not be put
 into the ready queue, so calling this function usually means you will
-never be called again.
+never be called again unless something else (e.g. an event handler) calls
+ready.
-=cut
+The canonical way to wait on external events is this:
+   {
+      # remember current coroutine
+      my $current = $Coro::current;
+      # register a hypothetical event handler
+      on_event_invoke sub {
+         # wake up sleeping coroutine
+         $current->ready;
+         undef $current;
+      };
+      # call schedule until event occured.
+      # in case we are woken up for other reasons
+      # (current still defined), loop.
+      Coro::schedule while $current;
+   }
 =item cede
-"Cede" to other processes. This function puts the current process into the
+"Cede" to other coroutines. This function puts the current coroutine into the
 ready queue and calls C<schedule>, which has the effect of giving up the
 current "timeslice" to other coroutines of the same or higher priority.
-=cut
+=item Coro::cede_notself
+Works like cede, but is not exported by default and will cede to any
+coroutine, regardless of priority, once.
 =item terminate [arg...]
-Terminates the current process with the given status values (see L<cancel>).
+Terminates the current coroutine with the given status values (see L<cancel>).
 =cut
 sub terminate {
    $current->cancel (@_);
 =back
 # dynamic methods
-=head2 PROCESS METHODS
+=head2 COROUTINE METHODS
-These are the methods you can call on process objects.
+These are the methods you can call on coroutine objects.
 =over 4
 =item new Coro \&sub [, @args...]
-Create a new process and return it. When the sub returns the process
+Create a new coroutine and return it. When the sub returns the coroutine
 automatically terminates as if C<terminate> with the returned values were
-called. To make the process run you must first put it into the ready queue
+called. To make the coroutine run you must first put it into the ready queue
 by calling the ready method.
-=cut
+Calling C<exit> in a coroutine will not work correctly, so do not do that.
+=cut
-sub _new_coro {
+sub _run_coro {
    terminate &{+shift};
 }
 sub new {
    my $class = shift;
-   $class->SUPER::new (\&_new_coro, @_)
+   $class->SUPER::new (\&_run_coro, @_)
 }
-=item $process->ready
+=item $success = $coroutine->ready
-Put the given process into the ready queue.
+Put the given coroutine into the ready queue (according to it's priority)
+and return true. If the coroutine is already in the ready queue, do nothing
+and return false.
-=cut
+=item $is_ready = $coroutine->is_ready
+Return wether the coroutine is currently the ready queue or not,
-=item $process->cancel (arg...)
+=item $coroutine->cancel (arg...)
-Terminates the given process and makes it return the given arguments as
+Terminates the given coroutine and makes it return the given arguments as
 status (default: the empty list).
 =cut
 sub cancel {
    push @destroy, $self;
    $manager->ready;
    &schedule if $current == $self;
 }
-=item $process->join
+=item $coroutine->join
 Wait until the coroutine terminates and return any values given to the
 C<terminate> or C<cancel> functions. C<join> can be called multiple times
-from multiple processes.
+from multiple coroutine.
 =cut
 sub join {
    my $self = shift;
       &schedule;
    }
    wantarray ? @{$self->{status}} : $self->{status}[0];
 }
+=item $coroutine->on_destroy (\&cb)
+Registers a callback that is called when this coroutine gets destroyed,
+but before it is joined. The callback gets passed the terminate arguments,
+if any.
+=cut
+sub on_destroy {
+   my ($self, $cb) = @_;
+   push @{ $self->{destroy_cb} }, $cb;
+}
-=item $oldprio = $process->prio ($newprio)
+=item $oldprio = $coroutine->prio ($newprio)
 Sets (or gets, if the argument is missing) the priority of the
-process. Higher priority processes get run before lower priority
+coroutine. Higher priority coroutines get run before lower priority
-processes. Priorities are small signed integers (currently -4 .. +3),
+coroutines. Priorities are small signed integers (currently -4 .. +3),
 that you can refer to using PRIO_xxx constants (use the import tag :prio
 to get then):
    PRIO_MAX > PRIO_HIGH > PRIO_NORMAL > PRIO_LOW > PRIO_IDLE > PRIO_MIN
        3    >     1     >      0      >    -1    >    -3     >    -4
    current->prio(PRIO_HIGH);
 The idle coroutine ($Coro::idle) always has a lower priority than any
 existing coroutine.
-Changing the priority of the current process will take effect immediately,
+Changing the priority of the current coroutine will take effect immediately,
-but changing the priority of processes in the ready queue (but not
+but changing the priority of coroutines in the ready queue (but not
 running) will only take effect after the next schedule (of that
-process). This is a bug that will be fixed in some future version.
+coroutine). This is a bug that will be fixed in some future version.
-=item $newprio = $process->nice ($change)
+=item $newprio = $coroutine->nice ($change)
 Similar to C<prio>, but subtract the given value from the priority (i.e.
 higher values mean lower priority, just as in unix).
-=item $olddesc = $process->desc ($newdesc)
+=item $olddesc = $coroutine->desc ($newdesc)
 Sets (or gets in case the argument is missing) the description for this
-process. This is just a free-form string you can associate with a process.
+coroutine. This is just a free-form string you can associate with a coroutine.
 =cut
 sub desc {
    my $old = $_[0]{desc};
    $old;
 }
 =back
+=head2 GLOBAL FUNCTIONS
+=over 4
+=item Coro::nready
+Returns the number of coroutines that are currently in the ready state,
+i.e. that can be swicthed to. The value C<0> means that the only runnable
+coroutine is the currently running one, so C<cede> would have no effect,
+and C<schedule> would cause a deadlock unless there is an idle handler
+that wakes up some coroutines.
+=item unblock_sub { ... }
+This utility function takes a BLOCK or code reference and "unblocks" it,
+returning the new coderef. This means that the new coderef will return
+immediately without blocking, returning nothing, while the original code
+ref will be called (with parameters) from within its own coroutine.
+The reason this fucntion exists is that many event libraries (such as the
+venerable L<Event|Event> module) are not coroutine-safe (a weaker form
+of thread-safety). This means you must not block within event callbacks,
+otherwise you might suffer from crashes or worse.
+This function allows your callbacks to block by executing them in another
+coroutine where it is safe to block. One example where blocking is handy
+is when you use the L<Coro::AIO|Coro::AIO> functions to save results to
+disk.
+In short: simply use C<unblock_sub { ... }> instead of C<sub { ... }> when
+creating event callbacks that want to block.
+=cut
+our @unblock_pool;
+our @unblock_queue;
+our $UNBLOCK_POOL_SIZE = 2;
+sub unblock_handler_ {
+   while () {
+      my ($cb, @arg) = @{ delete $Coro::current->{arg} };
+      $cb->(@arg);
+      last if @unblock_pool >= $UNBLOCK_POOL_SIZE;
+      push @unblock_pool, $Coro::current;
+      schedule;
+   }
+}
+our $unblock_scheduler = async {
+   while () {
+      while (my $cb = pop @unblock_queue) {
+         my $handler = (pop @unblock_pool or new Coro \&unblock_handler_);
+         $handler->{arg} = $cb;
+         $handler->ready;
+         cede;
+      }
+      schedule;
+   }
+};
+sub unblock_sub(&) {
+   my $cb = shift;
+   sub {
+      push @unblock_queue, [$cb, @_];
+      $unblock_scheduler->ready;
+   }
+}
+=back
 =cut
 1;
 =head1 BUGS/LIMITATIONS

Diff Legend

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

Comparing Coro/Coro.pm (file contents): Revision 1.88 by root, Sun Nov 26 02:54:55 2006 UTC vs. Revision 1.102 by root, Fri Dec 29 11:37:49 2006 UTC

Diff Legend

Comparing Coro/Coro.pm (file contents):
Revision 1.88 by root, Sun Nov 26 02:54:55 2006 UTC vs.
Revision 1.102 by root, Fri Dec 29 11:37:49 2006 UTC