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

Comparing Coro/Coro.pm (file contents):
Revision 1.91 by root, Fri Dec 1 02:17:37 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
 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
    $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 unless something else (e.g. an event handler) calls
 ready.
 The canonical way to wait on external events is this:
    {
-      # remember current process
+      # remember current coroutine
       my $current = $Coro::current;
       # register a hypothetical event handler
       on_event_invoke sub {
          # wake up sleeping coroutine
       # in case we are woken up for other reasons
       # (current still defined), loop.
       Coro::schedule while $current;
    }
-=cut
 =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.
 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 $success = $process->ready
+=item $success = $coroutine->ready
-Put the given process into the ready queue (according to it's priority)
+Put the given coroutine into the ready queue (according to it's priority)
-and return true. If the process is already in the ready queue, do nothing
+and return true. If the coroutine is already in the ready queue, do nothing
 and return false.
-=item $is_ready = $process->is_ready
+=item $is_ready = $coroutine->is_ready
-Return wether the process is currently the ready queue or not,
+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.91 by root, Fri Dec 1 02:17:37 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.91 by root, Fri Dec 1 02:17:37 2006 UTC vs.
Revision 1.102 by root, Fri Dec 29 11:37:49 2006 UTC