Add retry code to client and worker

If a Gearman server dies and another server is available, Gearman
clients and workers will automatically try connecting to the next server
in the list.  But a dying server can still cause the next request from a
client or worker to fail.  To help prevent work from being lost, this
revision adds code that will retry connections up to 10 times, using an
exponential back-off algorithm to determine delays between retries.

Author: Mike Raynham

Changes

@@ -2,4 +2,5 @@ Revision history for Gearman-Mesh
 
 0.01    Date/time
         First version, released on an unsuspecting world.
-
+0.02    2015-09-21
+        Added automatic retries to worker and client.

lib/Gearman/Mesh.pm

@@ -14,11 +14,11 @@ Gearman::Mesh - thin wrappers around Gearman::XS modules
 
 =head1 VERSION
 
-Version 0.01
+Version 0.02
 
 =cut
 
-our $VERSION = '0.01';
+our $VERSION = '0.02';
 
 =head1 DESCRIPTION
 

lib/Gearman/Mesh/Client.pm

@@ -32,6 +32,7 @@ use Gearman::Mesh qw(
 
 use Exporter 5.57 qw(import);
 use Gearman::XS 0.16 qw(:constants);
+use Time::HiRes ();
 
 our @EXPORT_OK   = @Gearman::XS::EXPORT_OK;
 our %EXPORT_TAGS = %Gearman::XS::EXPORT_TAGS;
@@ -59,28 +60,44 @@ BEGIN {
         eval join("\n",
             qq`package $pkg;`,
             qq`sub $job {`,
-             q`my $self = shift;`,
-             q`my $task = shift;`,
-             q`my $args = $self->serialize([@_]);`,
-            qq`my \@ret = \$self->{_delegate}->$job(\$task, \$args);`,
-             q`return if $ret[0] ne GEARMAN_SUCCESS;`,
-             q`return $ret[1];`,
+             q`    my $self    = shift;`,
+             q`    my $task    = shift;`,
+             q`    my $args    = $self->serialize([@_]);`,
+             q`    my $backoff = _backoff(10, 1000);`,
+             q`    my @ret;`,
+             q`    while (1) {`,
+            qq`        \@ret = \$self->{_delegate}->$job(\$task, \$args);`,
+             q`        last if defined $ret[0] && $ret[0] eq GEARMAN_SUCCESS;`,
+             q`        last unless $backoff->();`,
+             q`    }`,
+             q`    return $ret[1];`,
              q`}`,
         );
     }
 }
 
+sub _backoff {
+    my $tries = shift;
+    my $scale = shift;
+    my $try   = 0;
+    
+    sub {
+        return if ++$try > $tries;
+        Time::HiRes::usleep( $scale << int rand($try) );
+    }
+}
+
 =head1 NAME
 
 Gearman::Mesh::Client - A wrapper around Gearman::XS::Client
 
 =head1 VERSION
 
-Version 0.01
+Version 0.02
 
 =cut
 
-our $VERSION = '0.01';
+our $VERSION = '0.02';
 
 
 =head1 SYNOPSIS
@@ -112,7 +129,9 @@ This module optionally re-exports the constants from L<Gearman::XS>:
 =cut
 
 sub new {
-    shift->SUPER::new('Gearman::XS::Client', @_);
+    my $self = shift->SUPER::new('Gearman::XS::Client', @_);
+    $self->set_timeout(1_000);
+    $self;
 }
 
 =head1 METHODS
@@ -126,16 +145,10 @@ sub new {
  my $task = $client->add_task($function_name => $workload);
 
 Each of these methods are wrapped in a method that serializes C<$workload>
-before calling the equivalent L<Gearman::XS::Client> method:
-
- sub do {
-     my $self = shift;
-     my $task = shift;
-     my $args = $self->serialize([@_]);
-     my @ret = $self->{gearman_client}->do($task, $args);
-     return if $ret[0] ne GEARMAN_SUCCESS;
-     return $ret[1];
- }
+before calling the equivalent L<Gearman::XS::Client> method.
+
+Each method will retry up to ten times, using an exponential backoff algorithm
+to determine the delay between retries.
 
 Job methods (those whose names begin with C<do>) will return the job handle on
 success.  Task methods (those whose name begin with C<add_task>) will return a

lib/Gearman/Mesh/Worker.pm

@@ -41,11 +41,11 @@ Gearman::Mesh::Worker - The great new Gearman::Mesh::Worker!
 
 =head1 VERSION
 
-Version 0.01
+Version 0.02
 
 =cut
 
-our $VERSION = '0.01';
+our $VERSION = '0.02';
 
 =head1 SYNOPSIS
 
@@ -172,19 +172,43 @@ sub work_loop {
     my $options = $worker->options;
     my $timeout = $worker->timeout;
 
-    $worker->add_options(GEARMAN_WORKER_NON_BLOCKING);
     $worker->set_timeout(1_000);
 
+    my $work_ok = sub {
+        my $code = shift;
+        return 1 if $code eq GEARMAN_IO_WAIT
+                 || $code eq GEARMAN_NO_JOBS
+                 || $code eq GEARMAN_TIMEOUT;
+    };
+
+    my $wait_ok = sub {
+        my $code = shift;
+        return 1 if $code eq GEARMAN_SUCCESS
+                 || $code eq GEARMAN_NO_ACTIVE_FDS
+                 || $code eq GEARMAN_TIMEOUT;
+    };
+
     {
         my $continue = 1;
         my $handler  = sub {$continue = 0};
 
         local $SIG{TERM} = $handler;
         local $SIG{INT}  = $handler;
 
-        while ($continue) {
-            $worker->work;
-            $worker->wait;
+        TRY: while ($continue) {
+
+            my $work = $worker->work;
+            next TRY if $work == GEARMAN_SUCCESS;
+
+            sleep 1;
+
+            if ($work_ok->($work)) {
+                my $wait = $worker->wait;
+                next TRY if $wait_ok->($wait);
+            }
+
+            warn $worker->error;
+            sleep 5; 
         }
     }