diff --git a/lib/POE.pm b/lib/POE.pm index 04c58d0..5800bc2 100644 --- a/lib/POE.pm +++ b/lib/POE.pm @@ -8,6 +8,8 @@ use Carp qw( croak ); use vars qw($VERSION); $VERSION = '1.354'; # NOTE - Should be #.### (three decimal places) +use POE::Resource::Clock qw( monotime time walltime sleep mono2wall wall2mono ); + sub import { my $self = shift; diff --git a/lib/POE/Kernel.pm b/lib/POE/Kernel.pm index 42c3789..e28ae08 100644 --- a/lib/POE/Kernel.pm +++ b/lib/POE/Kernel.pm @@ -5,13 +5,15 @@ use strict; use vars qw($VERSION); $VERSION = '1.354'; # NOTE - Should be #.### (three decimal places) +use POE::Resource::Clock qw( monotime sleep mono2wall wall2mono walltime time ); + use POSIX qw(uname); use Errno qw(ESRCH EINTR ECHILD EPERM EINVAL EEXIST EAGAIN EWOULDBLOCK); use Carp qw(carp croak confess cluck); use Sys::Hostname qw(hostname); use IO::Handle (); use File::Spec (); -use Time::HiRes qw(time sleep); +#use Time::HiRes qw(time sleep); # People expect these to be lexical. @@ -235,8 +237,10 @@ sub EV_ARGS () { 4 } # \@event_parameters_arg0_etc, # sub EV_OWNER_FILE () { 5 } # $caller_filename_where_enqueued, sub EV_OWNER_LINE () { 6 } # $caller_line_where_enqueued, -sub EV_TIME () { 7 } # Maintained by POE::Queue (create time) +sub EV_FROMSTATE () { 7 } # $fromstate sub EV_SEQ () { 8 } # Maintained by POE::Queue (unique event ID) +sub EV_WALLTIME () { 9 } # Walltime when event was created (for alarms) +sub EV_DELTA () { 10 } # Seconds past walltime for event (for alarms) # ] # These are the names of POE's internal events. They're in constants @@ -632,7 +636,7 @@ sub _test_if_kernel_is_idle { $self->_data_ev_enqueue( $self, $self, EN_SIGNAL, ET_SIGNAL, [ 'IDLE' ], - __FILE__, __LINE__, undef, time(), + __FILE__, __LINE__, undef ); } @@ -723,7 +727,7 @@ sub signal { $self->_data_ev_enqueue( $session, $kr_active_session, EN_SIGNAL, ET_SIGNAL, [ $signal, @etc ], - (caller)[1,2], $kr_active_event, time(), + (caller)[1,2], $kr_active_event ); return 1; } @@ -861,7 +865,7 @@ sub _dispatch_event { my ( $self, $session, $source_session, $event, $type, $etc, - $file, $line, $fromstate, $time, $seq + $file, $line, $fromstate, $priority, $seq ) = @_; if (ASSERT_EVENTS) { @@ -960,7 +964,7 @@ sub _dispatch_event { $self->_dispatch_event( $target_session, $self, $target_event, ET_SIGNAL_RECURSIVE, [ @$etc, @$target_etc ], - $file, $line, $fromstate, time(), -__LINE__ + $file, $line, $fromstate, monotime(), -__LINE__ ); } } @@ -1095,7 +1099,7 @@ sub _dispatch_event { from_state => $fromstate, error_str => $exception, }, - ], __FILE__, __LINE__, undef, time() + ], __FILE__, __LINE__, undef ); } } @@ -1382,7 +1386,7 @@ sub _invoke_state { ) { $self->_data_ev_enqueue( $self, $self, EN_SIGNAL, ET_SIGNAL, [ 'ZOMBIE' ], - __FILE__, __LINE__, undef, time(), + __FILE__, __LINE__, undef ); } } @@ -1438,7 +1442,7 @@ sub session_alloc { my $return = $self->_dispatch_event( $session, $kr_active_session, EN_START, ET_START, \@args, - __FILE__, __LINE__, undef, time(), -__LINE__ + __FILE__, __LINE__, undef, monotime(), -__LINE__ ); unless($self->_data_ses_exists($new_sid)) { @@ -1455,7 +1459,7 @@ sub session_alloc { $self->_dispatch_event( $self->_data_ses_get_parent($session->ID), $self, EN_CHILD, ET_CHILD, [ CHILD_CREATE, $session, $return ], - __FILE__, __LINE__, undef, time(), -__LINE__ + __FILE__, __LINE__, undef, monotime(), -__LINE__ ); unless ($self->_data_ses_exists($new_sid)) { @@ -1469,7 +1473,7 @@ sub session_alloc { # to do its thing before it goes. $self->_data_ev_enqueue( $session, $session, EN_GC, ET_GC, [], - __FILE__, __LINE__, undef, time(), + __FILE__, __LINE__, undef ); } @@ -1500,7 +1504,7 @@ sub detach_myself { $self->_dispatch_event( $old_parent, $self, EN_CHILD, ET_CHILD, [ CHILD_LOSE, $kr_active_session, undef ], - (caller)[1,2], undef, time(), -__LINE__ + (caller)[1,2], undef, monotime(), -__LINE__ ) unless $kr_active_event_type & ET_START; @@ -1512,7 +1516,7 @@ sub detach_myself { $self->_dispatch_event( $kr_active_session, $self, EN_PARENT, ET_PARENT, [ $old_parent, $self ], - (caller)[1,2], undef, time(), -__LINE__ + (caller)[1,2], undef, monotime(), -__LINE__ ); $self->_data_ses_move_child($kr_active_session->ID, $self->ID); @@ -1557,7 +1561,7 @@ sub detach_child { $self->_dispatch_event( $kr_active_session, $self, EN_CHILD, ET_CHILD, [ CHILD_LOSE, $child_session, undef ], - (caller)[1,2], undef, time(), -__LINE__ + (caller)[1,2], undef, monotime(), -__LINE__ ); # Tell the new parent (kernel) that it's gaining a child. @@ -1568,7 +1572,7 @@ sub detach_child { $self->_dispatch_event( $child_session, $self, EN_PARENT, ET_PARENT, [ $kr_active_session, $self ], - (caller)[1,2], undef, time(), -__LINE__ + (caller)[1,2], undef, monotime(), -__LINE__ ); $self->_data_ses_move_child($child_session->ID, $self->ID); @@ -1631,7 +1635,7 @@ sub post { $self->_data_ev_enqueue( $session, $kr_active_session, $event_name, ET_POST, \@etc, - (caller)[1,2], $kr_active_event, time(), + (caller)[1,2], $kr_active_event ); return 1; } @@ -1655,7 +1659,7 @@ sub yield { $self->_data_ev_enqueue( $kr_active_session, $kr_active_session, $event_name, ET_POST, \@etc, - (caller)[1,2], $kr_active_event, time(), + (caller)[1,2], $kr_active_event ); undef; @@ -1706,7 +1710,7 @@ sub call { : $self->_dispatch_event( $session, $kr_active_session, $event_name, ET_CALL, \@etc, - (caller)[1,2], $kr_active_event, time(), -__LINE__ + (caller)[1,2], $kr_active_event, monotime(), -__LINE__ ) ); @@ -1724,7 +1728,7 @@ sub call { : $self->_dispatch_event( $session, $kr_active_session, $event_name, ET_CALL, \@etc, - (caller)[1,2], $kr_active_event, time(), -__LINE__ + (caller)[1,2], $kr_active_event, monotime(), -__LINE__ ) ); @@ -1742,7 +1746,7 @@ sub call { $self->_dispatch_event( $session, $kr_active_session, $event_name, ET_CALL, \@etc, - (caller)[1,2], $kr_active_event, time(), -__LINE__ + (caller)[1,2], $kr_active_event, monotime(), -__LINE__ ); } @@ -1822,9 +1826,10 @@ sub alarm_add { return 0; } -# Add a delay, which is just an alarm relative to the current time. +# Add a delay, which is like an alarm relative to the current time. sub delay { my ($self, $event_name, $delay, @etc) = ($poe_kernel, @_[1..$#_]); + my $pri = monotime(); if (ASSERT_USAGE) { _confess " must call delay() from a running session" @@ -1842,7 +1847,15 @@ sub delay { } if (defined $delay) { - $self->alarm($event_name, time() + $delay, @etc); + $self->_data_ev_clear_alarm_by_name($kr_active_session->ID(), $event_name); + + # Add the new alarm if it includes a time. Calling _data_ev_enqueue + # directly is faster than calling alarm_set to enqueue it. + $self->_data_ev_enqueue + ( $kr_active_session, $kr_active_session, + $event_name, ET_ALARM, [ @etc ], + (caller)[1,2], $kr_active_event, undef, $delay, $pri+$delay + ); } else { $self->alarm($event_name); @@ -1854,6 +1867,7 @@ sub delay { # Add a delay without clobbering previous delays of the same name. sub delay_add { my ($self, $event_name, $delay, @etc) = ($poe_kernel, @_[1..$#_]); + my $pri = monotime(); if (ASSERT_USAGE) { _confess " must call delay_add() from a running session" @@ -1872,7 +1886,11 @@ sub delay_add { return EINVAL; } - $self->alarm_add($event_name, time() + $delay, @etc); + $self->_data_ev_enqueue + ( $kr_active_session, $kr_active_session, + $event_name, ET_ALARM, [ @etc ], + (caller)[1,2], $kr_active_event, undef, $delay, $pri+$delay + ); return 0; } @@ -1976,7 +1994,8 @@ sub alarm_adjust { my $my_alarm = sub { $_[0]->[EV_SESSION] == $kr_active_session; }; - return $kr_queue->adjust_priority($alarm_id, $my_alarm, $delta); + + return $self->_data_ev_adjust( $alarm_id, $my_alarm, undef(), $delta ); } # A convenient function for setting alarms relative to now. It also @@ -1987,7 +2006,8 @@ sub delay_set { # Always always always grab time() ASAP, so that the eventual # time we set the alarm for is as close as possible to the time # at which they ASKED for the delay, not when we actually set it. - my $t = time(); + my $t = walltime(); + my $pri = monotime(); # And now continue as normal my ($self, $event_name, $seconds, @etc) = ($poe_kernel, @_[1..$#_]); @@ -2018,7 +2038,7 @@ sub delay_set { return $self->_data_ev_enqueue ( $kr_active_session, $kr_active_session, $event_name, ET_ALARM, [ @etc ], - (caller)[1,2], $kr_active_event, $t + $seconds, + (caller)[1,2], $kr_active_event, $t, $seconds, $pri+$seconds ); } @@ -2053,7 +2073,7 @@ sub delay_adjust { _warn(" adjusted event $alarm_id by $seconds seconds"); } - return $kr_queue->set_priority($alarm_id, $my_delay, time() + $seconds); + return $kr_queue->_data_ev_adjust($alarm_id, $my_delay, undef, $seconds ); } # Remove all alarms for the current session. @@ -2394,7 +2414,7 @@ sub _recalc_id { "-", $hostname, map { unpack "H*", $_ } map { pack "N", $_ } - (time(), $$, ++$kr_id_seq) + (monotime(), $$, ++$kr_id_seq) ); if (defined $old_id) { diff --git a/lib/POE/Loop.pm b/lib/POE/Loop.pm index ee033fc..d5edce7 100644 --- a/lib/POE/Loop.pm +++ b/lib/POE/Loop.pm @@ -315,7 +315,7 @@ Here's Gtk's: $self->_dispatch_event( $self, $self, EN_SIGNAL, ET_SIGNAL, [ 'UIDESTROY' ], - __FILE__, __LINE__, undef, time(), -__LINE__ + __FILE__, __LINE__, undef, monotime(), -__LINE__ ); } return 0; diff --git a/lib/POE/Loop/IO_Poll.pm b/lib/POE/Loop/IO_Poll.pm index 37b5fd7..d02b0bc 100644 --- a/lib/POE/Loop/IO_Poll.pm +++ b/lib/POE/Loop/IO_Poll.pm @@ -67,7 +67,7 @@ BEGIN { my %poll_fd_masks; # Allow $^T to change without affecting our internals. -my $start_time = $^T; +my $start_time = monotime(); #------------------------------------------------------------------------------ # Loop construction and destruction. @@ -95,7 +95,7 @@ sub loop_attach_uidestroy { # value. A "paused" time watcher is just a timeout for some future # time. -my $_next_event_time = time(); +my $_next_event_time = monotime(); sub loop_resume_time_watcher { $_next_event_time = $_[1]; @@ -106,7 +106,7 @@ sub loop_reset_time_watcher { } sub loop_pause_time_watcher { - $_next_event_time = time() + 3600; + $_next_event_time = monotime() + 3600; } # A static function; not some object method. @@ -232,7 +232,7 @@ sub loop_do_timeslice { my $timeout = $_next_event_time; - my $now = time(); + my $now = monotime(); if (defined $timeout) { $timeout -= $now; $timeout = 0 if $timeout < 0; diff --git a/lib/POE/Loop/PerlSignals.pm b/lib/POE/Loop/PerlSignals.pm index b267625..651db9f 100644 --- a/lib/POE/Loop/PerlSignals.pm +++ b/lib/POE/Loop/PerlSignals.pm @@ -39,7 +39,7 @@ sub _loop_signal_handler_generic_bottom { $poe_kernel->_data_ev_enqueue( $poe_kernel, $poe_kernel, EN_SIGNAL, ET_SIGNAL, [ $_[0] ], - __FILE__, __LINE__, undef, time() + __FILE__, __LINE__, undef ); $SIG{$_[0]} = \&_loop_signal_handler_generic; } @@ -62,7 +62,7 @@ sub _loop_signal_handler_pipe_bottom { $poe_kernel->_data_ev_enqueue( $poe_kernel, $poe_kernel, EN_SIGNAL, ET_SIGNAL, [ $_[0] ], - __FILE__, __LINE__, undef, time() + __FILE__, __LINE__, undef ); $SIG{$_[0]} = \&_loop_signal_handler_pipe; } diff --git a/lib/POE/Loop/Select.pm b/lib/POE/Loop/Select.pm index 10e2840..ddd1fd4 100644 --- a/lib/POE/Loop/Select.pm +++ b/lib/POE/Loop/Select.pm @@ -37,7 +37,7 @@ my @loop_vectors = ("", "", ""); my %loop_filenos; # Allow $^T to change without affecting our internals. -my $start_time = $^T; +my $start_time = monotime(); #------------------------------------------------------------------------------ # Loop construction and destruction. @@ -86,7 +86,7 @@ sub loop_attach_uidestroy { # value. A "paused" time watcher is just a timeout for some future # time. -my $_next_event_time = time(); +my $_next_event_time = monotime(); sub loop_resume_time_watcher { $_next_event_time = $_[1]; @@ -97,7 +97,7 @@ sub loop_reset_time_watcher { } sub loop_pause_time_watcher { - $_next_event_time = time() + 3600; + $_next_event_time = monotime() + 3600; } #------------------------------------------------------------------------------ @@ -156,7 +156,7 @@ sub loop_do_timeslice { my $timeout = $_next_event_time; - my $now = time(); + my $now = monotime(); if (defined $timeout) { $timeout -= $now; $timeout = 0 if $timeout < 0; diff --git a/lib/POE/Queue/Array.pm b/lib/POE/Queue/Array.pm index 3140dd6..c6746be 100644 --- a/lib/POE/Queue/Array.pm +++ b/lib/POE/Queue/Array.pm @@ -8,6 +8,7 @@ use vars qw($VERSION @ISA); $VERSION = '1.354'; # NOTE - Should be #.### (three decimal places) @ISA = qw(POE::Queue); +use POE::Resource::Clock qw( wall2mono mono2wall monotime walltime time sleep ); use Errno qw(ESRCH EPERM); use Carp qw(confess); @@ -419,6 +420,23 @@ sub peek_items { return @items; } +### Get one item from the queue + +sub get_item { + my ($self, $id, $filter) = @_; + + my $priority = $item_priority{$id}; + unless (defined $priority) { + $! = ESRCH; + return; + } + + # Find that darn item. + my $item_index = $self->_find_item($id, $priority); + return unless defined $item_index; + return $self->[$item_index]->[ITEM_PAYLOAD]; +} + 1; __END__ diff --git a/lib/POE/Resource/Aliases.pm b/lib/POE/Resource/Aliases.pm index 03d82ba..b99c168 100644 --- a/lib/POE/Resource/Aliases.pm +++ b/lib/POE/Resource/Aliases.pm @@ -67,6 +67,7 @@ sub _data_alias_finalize { sub _data_alias_add { my ($self, $session, $alias) = @_; +# _warn( "Session ", $session->ID, " is alias $alias\n" ); $self->_data_ses_refcount_inc($session->ID); $kr_aliases{$alias} = $session; $kr_ses_to_alias{$session->ID}->{$alias} = $session; @@ -79,6 +80,7 @@ sub _data_alias_add { sub _data_alias_remove { my ($self, $session, $alias) = @_; +# _warn( "Session ", $session->ID, " was alias $alias\n" ); delete $kr_aliases{$alias}; delete $kr_ses_to_alias{$session->ID}->{$alias}; $self->_data_ses_refcount_dec($session->ID); diff --git a/lib/POE/Resource/Clock.pm b/lib/POE/Resource/Clock.pm new file mode 100644 index 0000000..ac1b263 --- /dev/null +++ b/lib/POE/Resource/Clock.pm @@ -0,0 +1,238 @@ +# Manage a platonic, monotonic clock to keep the event queue ordered + +package POE::Resource::Clock; + +use vars qw($VERSION); +$VERSION = '1.354'; # NOTE - Should be #.### (three decimal places) + +use strict; + +use File::Spec; + +sub DEBUG () { 0 } + +sub do_X +{ + my( $X, $default ) = @_; + my $m = "USE_$X"; + return POE::Kernel->can( $m )->() if POE::Kernel->can( $m ); + my $k = "POE_USE_$X"; + return $ENV{$k} if exists $ENV{$k}; + return $default if defined $default; + return 1; +} + +sub exact_epoch +{ + my( $monoclock ) = @_; + + # Try to get the exact difference between the monotonic clock's epoch + # and the system clock's epoch. We do this by comparing the 2 for 0.25 second + # or 10 samples. To compensate for delays between calling time and get_time, + # we run in both order. Even so, I still see up to 10 mS divergence in my dev VM + # between invocations + my $N=0; + my $total = 0; + my $end = $monoclock->get_time() + 0.25; + while( $end > $monoclock->get_time() or $N < 20) { + my $hr = Time::HiRes::time(); + my $mono = $monoclock->get_time; + $total += $hr - $mono; + $N++; + $mono = $monoclock->get_time; + $hr = Time::HiRes::time(); + $total += $hr - $mono; + $N++; + } + DEBUG and warn "RT clock samples=$N"; + return $total/$N; +} + +sub get_epoch +{ + my( $monoclock, $wallclock ) = @_; + return $wallclock->get_time - $monoclock->get_time; +} + + +BEGIN { + my $done; + if( do_X( 'POSIXRT' ) ) { + eval { + require File::Spec->catfile( qw( POSIX RT Clock.pm ) ); + my $monoclock = POSIX::RT::Clock->new( 'monotonic' ); + my $wallclock = POSIX::RT::Clock->new( 'realtime' ); + *monotime = sub { return $monoclock->get_time; }; + *walltime = sub { return $wallclock->get_time; }; + *sleep = sub { $monoclock->sleep_deeply(@_) }; + $DB::single = 1; + if( do_X( 'STATIC_EPOCH' ) ) { + # This is where we cheat: without a static epoch the tests fail + # because they expect alarm(), alarm_set() to arrive in order + # Calling get_epoch() each time would preclude this + my $epoch = 0; + if( do_X( 'EXACT_EPOCH', 0 ) ) { + $epoch = exact_epoch( $monoclock, $wallclock ); + } + else { + $epoch = get_epoch( $monoclock, $wallclock ); + } + DEBUG and warn "POE clock epoch=$epoch"; + *wall2mono = sub { $_[0] - $epoch }; + *mono2wall = sub { $_[0] + $epoch }; + } + else { + *wall2mono = sub { $_[0] - get_epoch($monoclock, $wallclock) }; + *mono2wall = sub { $_[0] + get_epoch($monoclock, $wallclock) }; + } + $done = 1; + }; + if( DEBUG ) { + warn "POSIX::RT::Clock not installed: $@" if $@; + warn "POE clock using POSIX::RT::Clock" if $done; + } + } + if( !$done and do_X( 'HIRES' ) ) { + eval { + require File::Spec->catfile( qw( Time HiRes.pm ) ); + *monotime = \&Time::HiRes::time; + *walltime = \&Time::HiRes::time; + *sleep = \&Time::HiRes::sleep; + *wall2mono = sub { return $_[0] }; + *mono2wall = sub { return $_[0] }; + $done = 1; + }; + if( DEBUG ) { + warn "Time::HiRes not installed: $@" if $@; + warn "POE clock using Time::HiRes" if $done; + } + } + unless( $done ) { + # \&CORE::time fails :-( + *monotime = sub { CORE::time }; + *walltime = sub { CORE::time }; + *sleep = sub { CORE::sleep(@_) }; + *wall2mono = sub { return $_[0] }; + *mono2wall = sub { return $_[0] }; + warn "POE clock using CORE::time" if DEBUG; + } + *time = sub { Carp::confess( "This should be monotime" ) }; +} + +require Exporter; +our @EXPORT_OK = qw( monotime sleep walltime wall2mono mono2wall time ); +our @ISA = qw( Exporter ); + +1; + +__END__ + +=head1 NAME + +POE::Resource::Clock - internal clock used for ordering the queue + +=head1 SYNOPSIS + + sub POE::Kernel::USE_POSIXRT { 0 } + use POE; + +=head1 DESCRIPTION + +POE::Resource::Clock is a helper module for POE::Kernel. It provides the +features to keep an internal monotonic clock and a wall clock. It also +converts between this monotonic clock and the wall clock. + +The monotonic clock is used to keep an ordered queue of events. The wall +clock is used to comunicate the time with user code +(L, L). + +There are 3 possible clock sources in order of preference: +L, L and L. Only +C has a seperate monotonic and wall clock; the other two use the +same source for both clocks. + +Clock selection and behaviour is controled with the following: + +=head2 USE_POSIXRT + + export POE_USE_POSIXRT=0 + or + sub POE::Kernel::USE_POSIXRT { 0 } + +Uses the C clock source for queue priority and the C +clock source for wall clock. Not used if POSIX::RT::Clock is not installed +or your system does not have a C clock. + +Defaults to true. If you want the old POE behaviour, set this to 0. + +=head2 USE_STATIC_EPOCH + + export POE_USE_STATIC_EPOCH=0 + or + sub POE::Kernel::USE_STATIC_EPOCH { 0 } + +The epoch of the POSIX::RT::Clock monotonic is different from that of the +realtime clock. For instance on Linux 2.6.18, the monotonic clock is the +number of seconds since system boot. This epoch is used to convert from +walltime into monotonic time for L, +L and L. If +C is true (the default), then the epoch is calculated at +load time. If false, the epoch is calculated each time it is needed. + +Defaults to true. Only relevant for if using POSIX::RT::Clock. Long-running +POE servers should have this set to false so that system clock skew does +mess up the queue. + +It is important to point out that without a static epoch, the ordering of +the following two alarms is undefined. + + $poe_kernel->alarm_set( a1 => $time ); + $poe_kernel->alarm_set( a2 => $time ); + +=head2 USE_EXACT_EPOCH + + export POE_USE_EXACT_EPOCH=1 + or + sub POE::Kernel::USE_EACT_EPOCH { 1 } + +There currently no way to exactly get the monotonic clock's epoch. Instead +the difference between the current monotonic clock value to the realtime +clock's value is used. This is obviously inexact because there is a slight +delay between the 2 system calls. Setting USE_EXACT_EPOCH to true will +calculate an average of this difference over 250 ms or at least 20 samples. +What's more, the system calls are done in both orders (monotonic then +realtime, realtime then monotonic) to try and get a more eact value. + +Defaults to false. Only relevant if L is true. + + +=head2 USE_HIRES + + export POE_USE_HIRES=0 + or + sub POE::Kernel::USE_HIRES { 0 } + +Use L as both monotonic and wall clock source. This was POE's +previous default clock. + +Defaults to true. Only relevant if L is false. Set this to false to use +L. + + +=head1 SEE ALSO + +See L for general discussion about resources and the +classes that manage them. + +=head1 BUGS + +None known. + +=head1 AUTHORS & COPYRIGHTS + +Please see L for more information about authors and contributors. + +=cut + +# rocco // vim: ts=2 sw=2 expandtab +# TODO - Edit. diff --git a/lib/POE/Resource/Events.pm b/lib/POE/Resource/Events.pm index 213c9f4..6c419e7 100644 --- a/lib/POE/Resource/Events.pm +++ b/lib/POE/Resource/Events.pm @@ -58,13 +58,13 @@ sub _data_ev_finalize { ### Enqueue an event. sub FIFO_TIME_EPSILON () { 0.000001 } -my $last_fifo_time = time(); +my $last_fifo_time = monotime(); sub _data_ev_enqueue { my ( $self, $session, $source_session, $event, $type, $etc, - $file, $line, $fromstate, $time + $file, $line, $fromstate, $time, $delta, $priority ) = @_; my $sid = $session->ID; @@ -79,32 +79,42 @@ sub _data_ev_enqueue { } # This is awkward, but faster than using the fields individually. - my $event_to_enqueue = [ @_[1..8] ]; + my $event_to_enqueue = [ @_[(1+EV_SESSION) .. (1+EV_FROMSTATE)] ]; + if( defined $time ) { + $event_to_enqueue->[EV_WALLTIME] = $time; + $event_to_enqueue->[EV_DELTA] = $delta; + $priority ||= wall2mono( $time + ($delta||0) ); + } + else { + $priority ||= monotime(); + } my $new_id; my $old_head_priority = $kr_queue->get_next_priority(); unless ($type & ET_MASK_DELAYED) { - $time = $last_fifo_time + FIFO_TIME_EPSILON if $time <= $last_fifo_time; - $last_fifo_time = $time; + $priority = $last_fifo_time + FIFO_TIME_EPSILON if $priority <= $last_fifo_time; + $last_fifo_time = $priority; } - $new_id = $kr_queue->enqueue($time, $event_to_enqueue); + $new_id = $kr_queue->enqueue($priority, $event_to_enqueue); + $event_to_enqueue->[EV_SEQ] = $new_id; - if (TRACE_EVENTS) { + #_carp( Carp::longmess( " priority is much to far in the future" ) ) if $priority > 1354569908; + if (TRACE_EVENTS ) { _warn( " enqueued event $new_id ``$event'' from ", $self->_data_alias_loggable($source_session->ID), " to ", $self->_data_alias_loggable($sid), - " at $time" + " at $time, priority=$priority" ); } unless (defined $old_head_priority) { - $self->loop_resume_time_watcher($time); + $self->loop_resume_time_watcher($priority); } - elsif ($time < $old_head_priority) { - $self->loop_reset_time_watcher($time); + elsif ($priority < $old_head_priority) { + $self->loop_reset_time_watcher($priority); } # This is the counterpart to _data_ev_refcount_dec(). It's only @@ -121,6 +131,20 @@ sub _data_ev_enqueue { return $new_id; } +sub _data_ev_adjust +{ + my( $self, $alarm_id, $my_alarm, $time, $delta ) = @_; + # Doing it this way is faster + # XXX- However, if there has been a clock skew, the priority will have changed + # and we should recalculate priority from time+delta + $kr_queue->adjust_priority( $alarm_id, $my_alarm, $delta ); + my $event = $kr_queue->get_item( $alarm_id, $my_alarm ); + return unless $event; + $event->[EV_WALLTIME] = $time if defined $time; + $event->[EV_DELTA] += $delta if defined $delta; + return $event->[EV_WALLTIME]+$event->[EV_DELTA]; +} + ### Remove events sent to or from a specific session. sub _data_ev_clear_session { @@ -211,17 +235,18 @@ sub _data_ev_clear_alarm_by_id { $_[0]->[EV_SESSION]->ID() eq $sid; }; - my ($time, $id, $event) = $kr_queue->remove_item($alarm_id, $my_alarm); - return unless defined $time; + my ($pri, $id, $event) = $kr_queue->remove_item($alarm_id, $my_alarm); + return unless defined $pri; if (TRACE_EVENTS) { _warn( " removed event $id ``", $event->[EV_NAME], "'' to ", - $self->_data_alias_loggable($sid), " at $time" + $self->_data_alias_loggable($sid), " at $pri" ); } $self->_data_ev_refcount_dec( @$event[EV_SOURCE, EV_SESSION] ); + my $time = $event->[EV_WALLTIME] + ($event->[EV_DELTA]||0); return ($time, $event); } @@ -238,8 +263,9 @@ sub _data_ev_clear_alarm_by_session { my @removed; foreach ($kr_queue->remove_items($my_alarm)) { - my ($time, $event) = @$_[ITEM_PRIORITY, ITEM_PAYLOAD]; + my ($pri, $event) = @$_[ITEM_PRIORITY, ITEM_PAYLOAD]; $self->_data_ev_refcount_dec( @$event[EV_SOURCE, EV_SESSION] ); + my $time = $event->[EV_WALLTIME] + ($event->[EV_DELTA]||0); push @removed, [ $event->[EV_NAME], $time, @{$event->[EV_ARGS]} ]; } @@ -297,13 +323,13 @@ sub _data_ev_dispatch_due { } } - my $now = time(); + my $now = monotime(); my $next_time; while ( defined($next_time = $kr_queue->get_next_priority()) and $next_time <= $now ) { - my ($due_time, $id, $event) = $kr_queue->dequeue_next(); + my ($priority, $id, $event) = $kr_queue->dequeue_next(); if (TRACE_EVENTS) { _warn(" dispatching event $id ($event->[EV_NAME])"); @@ -312,7 +338,7 @@ sub _data_ev_dispatch_due { # TODO - Why can't we reverse these two lines? # TODO - Reversing them could avoid entering and removing GC marks. $self->_data_ev_refcount_dec($event->[EV_SOURCE], $event->[EV_SESSION]); - $self->_dispatch_event(@$event, $due_time, $id); + $self->_dispatch_event(@{$event}[EV_SESSION..EV_FROMSTATE], $priority, $id); # Stop the system if an unhandled exception occurred. # This wipes out all sessions and associated resources. diff --git a/lib/POE/Resource/FileHandles.pm b/lib/POE/Resource/FileHandles.pm index 0cffccd..bd57c2a 100644 --- a/lib/POE/Resource/FileHandles.pm +++ b/lib/POE/Resource/FileHandles.pm @@ -202,7 +202,7 @@ sub _data_handle_finalize { sub _data_handle_enqueue_ready { my ($self, $mode) = splice(@_, 0, 2); - my $now = time(); + my $now = monotime(); foreach my $fileno (@_) { if (ASSERT_DATA) { _trap "internal inconsistency: undefined fileno" unless defined $fileno; diff --git a/lib/POE/Resource/Sessions.pm b/lib/POE/Resource/Sessions.pm index dfb9879..a7f5d75 100644 --- a/lib/POE/Resource/Sessions.pm +++ b/lib/POE/Resource/Sessions.pm @@ -551,13 +551,13 @@ sub _data_ses_stop { $self->_dispatch_event( $parent, $self, EN_CHILD, ET_CHILD, [ CHILD_GAIN, $child ], - __FILE__, __LINE__, undef, time(), -__LINE__ + __FILE__, __LINE__, undef, monotime(), -__LINE__ ); $self->_dispatch_event( $child, $self, EN_PARENT, ET_PARENT, [ $self->_data_ses_get_parent($child->ID), $parent, ], - __FILE__, __LINE__, undef, time(), -__LINE__ + __FILE__, __LINE__, undef, monotime(), -__LINE__ ); } @@ -568,7 +568,7 @@ sub _data_ses_stop { my $stop_return = $self->_dispatch_event( $session, $self->get_active_session(), EN_STOP, ET_STOP, [], - __FILE__, __LINE__, undef, time(), -__LINE__ + __FILE__, __LINE__, undef, monotime(), -__LINE__ ); # If the departing session has a parent, notify it that the session @@ -578,7 +578,7 @@ sub _data_ses_stop { $self->_dispatch_event( $parent, $self, EN_CHILD, ET_CHILD, [ CHILD_LOSE, $session, $stop_return ], - __FILE__, __LINE__, undef, time(), -__LINE__ + __FILE__, __LINE__, undef, monotime(), -__LINE__ ); } diff --git a/lib/POE/Resource/Signals.pm b/lib/POE/Resource/Signals.pm index fea5168..8961393 100644 --- a/lib/POE/Resource/Signals.pm +++ b/lib/POE/Resource/Signals.pm @@ -588,7 +588,7 @@ sub _data_sig_enqueue_poll_event { $self->_data_ev_enqueue( $self, $self, EN_SCPOLL, ET_SCPOLL, [ $signal ], - __FILE__, __LINE__, undef, time(), + __FILE__, __LINE__, undef ); } else { return if $self->_data_ses_count() < 1; @@ -596,7 +596,7 @@ sub _data_sig_enqueue_poll_event { $self->_data_ev_enqueue( $self, $self, EN_SCPOLL, ET_SCPOLL, [ $signal ], - __FILE__, __LINE__, undef, time() + POE::Kernel::CHILD_POLLING_INTERVAL(), + __FILE__, __LINE__, undef, walltime(), POE::Kernel::CHILD_POLLING_INTERVAL(), ); } } @@ -610,7 +610,7 @@ sub _data_sig_handle_poll_event { if (TRACE_SIGNALS) { _warn( - " POE::Kernel is polling for signals at " . time() . + " POE::Kernel is polling for signals at " . monotime() . (USE_SIGCHLD ? " due to SIGCHLD" : "") ); } @@ -675,7 +675,7 @@ sub _data_sig_reap_pids { $self->_data_ev_enqueue( $ses_rec->[PID_SESSION], $self, $ses_rec->[PID_EVENT], ET_SIGCLD, [ 'CHLD', $pid, $?, @{$ses_rec->[PID_ARGS]} ], - __FILE__, __LINE__, undef, time(), + __FILE__, __LINE__, undef ); push @sessions_to_clear, $sid; } @@ -685,7 +685,7 @@ sub _data_sig_reap_pids { # Kick off a SIGCHLD cascade. $self->_data_ev_enqueue( $self, $self, EN_SIGNAL, ET_SIGNAL, [ 'CHLD', $pid, $? ], - __FILE__, __LINE__, undef, time(), + __FILE__, __LINE__, undef ); } elsif (TRACE_SIGNALS) {