POE::Component::Schedule: Request for feature

Deven Parekh <[email protected]> Fri, 20 Nov 2009 15:31:27 -0600

Newsgroups	gmane.comp.lang.perl.poe
Message-ID	<[email protected]>

Hello Oliver,

I saw your other mail that you were re-factoring the
POE::Component::Schedule.
I was tempted to reply to that but did not want to pollute that thread.

First of all this is a great module. Thanks.

I have a feature request for this module for your consideration.
I was wondering if a capability could be defined to plug in a
custom iterator to use (instead of DateTime::Set). I was interested in
Algorithm::TokenBucket as the iterator.

Thanks in advance for the consideration.

Regards
Deven.
p.s: I gave it an attempt and came up with a version that works. Basically
      your module + some modification to _schedule and minors ones to
_client_event
     and of course the iterator type check in add. But was wondering if some
how the
     two can be combined? I attached the experimental module an example
script that
     uses it and the output of that script. I guess i can try subclass from
POCO::Schedule
     and just overload these three routines.

p.p.s: This is based on the original version of this module before
re-factoring.

output.txt (text/plain, 542 B)

 ./example_1.pl
start 1258751264.06958
tick 1258751265.74366 ( 00:00:01.674042 )
tick 1258751267.41223 ( 00:00:03.342612 )
tick 1258751269.0807 ( 00:00:05.011077 )
tick 1258751270.7492 ( 00:00:06.679584 )
tick 1258751272.41771 ( 00:00:08.348090 )
tick 1258751274.08622 ( 00:00:10.016599 )
tick 1258751275.75471 ( 00:00:11.685086 )
tick 1258751277.42326 ( 00:00:13.353643 )
tick 1258751279.09176 ( 00:00:15.022136 )
tick 1258751280.76028 ( 00:00:16.690656 )
tick 1258751282.42877 ( 00:00:18.359147 )
tick 1258751284.09731 ( 00:00:20.027685 )

ATkB.pm (application/octet-stream, 7.2 KB)

package POE::Component::Schedule::ATkB;
use 5.008;

our $VERSION = '0.03';

use strict;
use warnings;

use POE;

my $Singleton;
my $ID_Sequence = 'a';    # sequence is 'a', 'b', ..., 'z', 'aa', 'ab', ...
my %Schedule_Ticket;      # Hash helps remember alarm id for cancel.

#
# crank up the schedule session
#
sub spawn {
    my $class = shift;
    my %arg   = @_;

    if ( !defined $Singleton ) {

        $Singleton = POE::Session->create(
            inline_states => {
                _start => sub {
                    my ($k) = $_[KERNEL];

                    $k->alias_set( $arg{'Alias'} || $class );
                    $k->sig( 'SHUTDOWN', 'shutdown' );
                },

                schedule     => \&_schedule,
                client_event => \&_client_event,
                cancel       => \&_cancel,

                shutdown => sub {
                    #print "# $class shutdown\n";
                    my $k = $_[KERNEL];

                    # FIXME We are removing too much here!
                    $k->alarm_remove_all();

                    $k->sig_handled();
                },
                _stop => sub {
                    #print "# $class _stop\n";
                    $Singleton = undef;
                },
            },
        )->ID;
    }
}

#
# schedule the next event
#  ARG0 is a client session,
#  ARG1 is the client event name,
#  ARG2 is a DateTime::Set iterator
#  ARG3 is an schedule ticket
#  ARG4 .. $#_ are arguments to the client event
#
#--- Overload add and schedule
#
## schedule the next event
##  ARG0 is the schedule ticket
##
sub _schedule {
    my ( $k, $s, $e, $ds, $tix, @arg ) = @_[ KERNEL, ARG0 .. $#_ ];
    my @ds_state = $ds->state();
    my $burst_size = $ds_state[1];
    if ($ds->conform($burst_size) ) {
       #--- It is time to call the event
       for (my $i = 0; $i < $burst_size;$i++) {
          $k->yield( 'client_event', $s, $e, $ds, $tix, @arg );
       }
       $ds->count($burst_size);
    }

    #--- How long before we do another check
    my $n;
    $n = $ds->until($burst_size); 
    $Schedule_Ticket{$tix} = $poe_kernel->delay_set( 'schedule', $n, $s, $e, $ds, $tix, @arg );
}
    
#
# handle a client event and schedule the next one
#  ARG0 is a client session,
#  ARG1 is the client event name,
#  ARG2 is a DateTime::Set iterator
#  ARG3 is an schedule ticket
#  ARG4 .. $#_ are arguments to the client event
#
sub _client_event {
    my ( $k, $s, $e, $ds, $tix, @arg ) = @_[ KERNEL, ARG0 .. $#_ ];
       $k->post( $s, $e, @arg );
}

#
# cancel an alarm
#
sub _cancel {
    my ( $k, $id ) = @_[ KERNEL, ARG0 ];

    $k->alarm_remove($id);
}


sub add {

    my $class  = shift;
    my $ticket = $ID_Sequence++;    # get the next ticket;

    my ( $session, $event, $iterator, @args ) = @_;
    $iterator->isa('Algorithm::TokenBucket')
      or die __PACKAGE__ . "->add: third arg must be a Algorithm::TokenBucket";

    $class->spawn unless $Singleton;

    $poe_kernel->post( $poe_kernel->ID_id_to_session($Singleton),
        'schedule', $session, $event, $iterator, $ticket, @args, );
    $Schedule_Ticket{$ticket} = ();

    return bless \$ticket, ref $class || $class;
}


sub delete {
    my $self   = shift;
    my $ticket = $$self;

    $poe_kernel->post(
        $poe_kernel->ID_id_to_session($Singleton),
        'cancel', $Schedule_Ticket{$ticket},
    );
    delete $Schedule_Ticket{$ticket};
}

sub DESTROY {
    $_[0]->delete if exists $Schedule_Ticket{${$_[0]}};
}

{
    no warnings;
    *new = \&add;
}

1;
__END__

=head1 NAME

POE::Component::Schedule - Schedule POE events using DateTime::Set iterators

=head1 SYNOPSIS

    use POE qw(Component::Schedule);
    use DateTime::Set;

    $s1 = POE::Session->create(
        inline_states => {
            _start => sub {
                $_[HEAP]{sched} = POE::Component::Schedule->add(
                    $_[SESSION], Tick => DateTime::Set->from_recurrence(
                        after      => DateTime->now,
                        before     => DateTime->now->add(seconds => 3)
                        recurrence => sub {
                            return $_[0]->truncate( to => 'second' )->add( seconds => 1 )
                        },
                    ),
                );
            },
            Tick => sub {
                print 'tick ', scalar localtime, "\n";
            },
            remove_sched => sub {
                # Three ways to remove a schedule
                # The first one is only for API compatibility with POE::Component::Cron
                $_[HEAP]{sched}->delete;
                $_[HEAP]{sched} = undef;
                delete $_[HEAP]{sched};
            }
            _stop => sub {
                print "_stop\n";
            },
        },
    );

=head1 DESCRIPTION

This component encapsulates a session that sends events to client sessions
on a schedule as defined by a DateTime::Set iterator.

=head1 POE::Component::Schedule METHODS

=head2 spawn(Alias => I<name>)

No need to call this in normal use, add() and new() all crank
one of these up if it is needed. Start up a PoCo::Schedule. Returns a
handle that can then be added to.

=head2 add()

    my $sched = POE::Component::Schedule->add(
        $session_object,
        $event_name,
        $DateTime_Set_iterator,
        @event_args
    );

Add a set of events to the schedule. The C<$session_object> and C<$event_name> are passed
to POE without even checking to see if they are valid and so have the same
warnings as ->post() itself.
C<$session_object> must be a real L<POE::Session>, not a session ID. Else session
reference count will not be increased and the session may end before receiving all
events.

Returns a schedule handle. The event is removed from when the handle is not referenced
anymore.


=head2 new

new is an alias for add

=head1 SCHEDULE HANDLE METHODS

=head2 delete

Removes a schedule using the handle returned from ->add or ->new.

B<DEPRECATED>: Schedules are now automatically deleted when they are not
referenced anymore. So just setting the container variable to C<undef> will
delete the schedule.

=head1 SEE ALSO

L<POE>, L<DateTime::Set>, L<POE::Component::Cron>.

=head1 BUGS

You can look for information at:

=over 4

=item * RT: CPAN's request tracker

L<http://rt.cpan.org/NoAuth/Bugs.html?Dist=POE-Component-Schedule>

=item * AnnoCPAN: Annotated CPAN documentation

L<http://annocpan.org/dist/POE-Component-Schedule>

=item * CPAN Ratings

L<http://cpanratings.perl.org/p/POE-Component-Schedule>

=item * Search CPAN

L<http://search.cpan.org/dist/POE-Component-Schedule/>

=back


=head1 ACKNOWLEDGMENT

This module is a friendly fork of POE::Component::Cron to extract the generic
parts and isolate the Cron specific code in order to reduce dependencies on
other CPAN modules.

The orignal author of POE::Component::Cron is Chris Fedde.

See L<https://rt.cpan.org/Ticket/Display.html?id=44442>

=head1 AUTHORS

=over 4

=item Olivier MenguE<eacute>, C<<< [email protected] >>>

=item Chris Fedde, C<<< [email protected] >>>

=back

=head1 COPYRIGHT AND LICENSE

=over 4

=item Copyright E<copy> 2007-2008 Chris Fedde

=item Copyright E<copy> 2009 Olivier MenguE<eacute>

=back

This library is free software; you can redistribute it and/or modify
it under the same terms as Perl itself, either Perl version 5.8.3 or,
at your option, any later version of Perl 5 you may have available.

=cut

example_1.pl (application/octet-stream, 1.2 KB)

#!/usr/bin/perl
use strict;
use warnings;
use lib "/path/to/module"; #--- not needed if Schedule::ATkB is installed
use POE qw(Component::Schedule::ATkB);
use Algorithm::TokenBucket;
use Time::HiRes qw( time );
use Time::Elapse;
my $so_many_task = 3;
my $so_many_seconds = 5;
my $burst_size = 1;
my $rate = $so_many_task / $so_many_seconds;
my $elapsed_time;

print 'start ', time , "\n";
Time::Elapse->lapse($elapsed_time);
POE::Session->create(
        inline_states => {
            _start => sub {
               $_[KERNEL]->yield("add_sched");
            },
            Tick => sub {
                #--- Just to show that we do get Ticks at the said rate
                print 'tick ', time, " ( $elapsed_time )\n";
            },
            add_sched => sub {
                $_[HEAP]{sched} = POE::Component::Schedule::ATkB->add(
                    $_[SESSION], Tick => Algorithm::TokenBucket->new($rate, $burst_size),
                );
            },
            remove_sched => sub {
                $_[HEAP]{sched}->delete;
                #$_[HEAP]{sched} = undef;
                #delete $_[HEAP]{sched};
            },
            _stop => sub {
                print "_stop\n";
            },
        },
    );


POE::Kernel->run();
exit(0);