Bug #59423 for POE-Component-SpreadClient: Driver doesn't read all pending messages

Thu Jul 15 15:40:39 2010 rdb [...] cpan.org - Ticket created

Subject:

Driver doesn't read all pending messages

So when using PoCo::SpreadClient on a busy spread network, that gets many messages/sec, I get disconnected alot (especially when there is a burst of traffic). My program logic reconnects, but research indicates that when there are too many messages "backed up" for a client, the daemon will disconnect that client. I looked in the code and the Driver at its core simply reads a message & dispatches it. however, this can get "behind" and the pending messages stack up until the client is kicked off. The solution, of course, would be to read until there are no more messages (or at least several at a time), and hold those messages in an internal buffer, to be dispatched as appropriate.

Fri Jan 07 18:19:27 2011 rdb [...] cpan.org - Correspondence added 240 min

As a workaround to the bug I filed previously, I've re-implemented PoCo::SpreadClient around Wheel::Run. Attached is the tgz for POE::Comonent::SpreadClient::Fork, which uses PoCo::SpreadClient code and libs, but wraps them under a Wheel::Run. The lib/ folder can be extracted directly into your distribution, it's a drop-in replacement, by simply using SpreadClient::Fork instead of SpreadClient.

Subject:

POE-Component-SpreadClient-Fork-0.03.tar.gz

Download POE-Component-SpreadClient-Fork-0.03.tar.gz
application/gzip 3.8k

Message body not shown because it is not plain text.

Fri Jan 07 18:19:27 2011 rdb [...] cpan.org - Status changed from 'new' to 'open'

Sat Jan 15 00:07:06 2011 rdb [...] cpan.org - Correspondence added

OK, tried a different approach this time. Added Multi-Reads to the input Driver. If the there are pending messages after the initial read, the Driver will read up to MAX_READS messages at a time. And this patches cleanly against the source And this works with the standard module *and* my ::Fork patch.

Subject:

multi.patch

--- ../POE-Component-SpreadClient-0.08/lib/POE/Driver/SpreadClient.pm 2009-02-01 18:47:35.000000000 -0800 +++ lib/POE/Driver/SpreadClient.pm 2011-01-14 20:41:00.000000000 -0800 @@ -9,6 +9,8 @@ # Import some stuff use Spread; +use constant MAX_READS => 16; + sub new { my $type = shift; my $mbox = shift; @@ -18,9 +20,19 @@ sub get { my $self = shift; + my $reads_performed = 1; + my @buf = (); + + # read once: + push @buf, [ Spread::receive( $$self ) ]; + + # Spread::poll returns 0 if no messages pending; + while(Spread::poll($$self) and ++$reads_performed <= MAX_READS) { + push @buf, [ Spread::receive( $$self ) ]; + + } - # this returns all undef if we're disconnected - return [ [ Spread::receive( $$self ) ] ]; + return [ @buf ]; } 1;

Tue Feb 01 23:43:18 2011 rdb [...] cpan.org - Correspondence added

Last patch wasn't quite right, because RW_GotPacket was written to ONLY read one packet from the filter, regardless of how many had been delivered. This patch includes: multi_read patch to Driver, and bugfix to Component to process all messages delivered.

Subject:

poco_spreadclient_multi_read.patch

diff -urw POE-Component-SpreadClient-0.08/lib/POE/Component/SpreadClient.pm POE-Component-SpreadClient-0.09/lib/POE/Component/SpreadClient.pm --- POE-Component-SpreadClient-0.08/lib/POE/Component/SpreadClient.pm 2009-02-01 18:47:35.000000000 -0800 +++ POE-Component-SpreadClient-0.09/lib/POE/Component/SpreadClient.pm 2011-02-01 20:32:26.000000000 -0800 @@ -4,7 +4,7 @@ # Initialize our version $LastChangedRevision: 9 $ use vars qw( $VERSION ); -$VERSION = '0.08'; +$VERSION = '0.09'; # Load our stuff use 5.006; # to silence Perl::Critic's Compatibility::ProhibitThreeArgumentOpen @@ -438,7 +438,8 @@ } sub RW_GotPacket : State { - my( $type, $sender, $groups, $mess_type, $endian, $message ) = @{ @{ $_[ARG0] }[0] }; + for my $packet ( @{ $_[ARG0] } ) { + my( $type, $sender, $groups, $mess_type, $endian, $message ) = @{ $packet }; # Check for disconnect if ( ! defined $type ) { @@ -523,6 +524,7 @@ } } + } # All done! return; } diff -urw POE-Component-SpreadClient-0.08/lib/POE/Driver/SpreadClient.pm POE-Component-SpreadClient-0.09/lib/POE/Driver/SpreadClient.pm --- POE-Component-SpreadClient-0.08/lib/POE/Driver/SpreadClient.pm 2011-02-01 20:37:53.000000000 -0800 +++ POE-Component-SpreadClient-0.09/lib/POE/Driver/SpreadClient.pm 2011-02-01 20:38:41.000000000 -0800 @@ -4,11 +4,13 @@ # Our version stuff use vars qw( $VERSION ); -$VERSION = '0.08'; +$VERSION = '0.09'; # Import some stuff use Spread; +use constant MAX_READS => 256; + sub new { my $type = shift; my $mbox = shift; @@ -18,9 +20,18 @@ sub get { my $self = shift; + my $reads_performed = 1; + my @buf = (); + + # read once: + push @buf, [ Spread::receive( $$self ) ]; + + # Spread::poll returns 0 if no messages pending; + while(Spread::poll($$self) and ++$reads_performed <= MAX_READS) { + push @buf, [ Spread::receive( $$self ) ]; + } - # this returns all undef if we're disconnected - return [ [ Spread::receive( $$self ) ] ]; + return [ @buf ]; } 1;

Wed Feb 02 18:22:32 2011 APOCAL [...] cpan.org - Correspondence added

Hello, Big thanks for your work on this! I've integrated your patch(es) and released 0.09 on CPAN. Once you have the time to test it and confirm that I didn't screw anything up please let me know and I'll close this ticket :) -- ~Apocalypse

Thu Feb 03 21:47:16 2011 APOCAL [...] cpan.org - Correspondence added

Thanks again for double-checking! This is now fully resolved :) -- ~Apocalypse

Thu Feb 03 21:47:17 2011 APOCAL [...] cpan.org - Status changed from 'open' to 'resolved'