Changeset 2272 for branches/release-38/lib/MT/TheSchwartz.pm

Timestamp:

05/08/08 00:00:05 (23 months ago)

Author:

bchoate

Message:

Added 'randomly' flag to control whether jobs are shuffled when selecting work to do. Default does not randomize jobs within batch selected.

Files:

• branches/release-38/lib/MT/TheSchwartz.pm (modified) (3 diffs)

branches/release-38/lib/MT/TheSchwartz.pm

@@ -10 +10 @@
 use base qw( TheSchwartz );
 use MT::ObjectDriver::Driver::DBI;
+use List::Util qw( shuffle );
 
 my $instance;
+
+our $RANDOMIZE_JOBS = 0;
 
 sub instance {
@@ -33 +36 @@
     my (%param) = @_;
     my $workers = delete $param{workers} if exists $param{workers};
+    $RANDOMIZE_JOBS = delete $param{randomize} if exists $param{randomize};
 
     my $client = $class->SUPER::new(%param);
@@ -159 +163 @@
 }
 
+sub _grab_a_job {
+    my TheSchwartz $client = shift;
+    my $hashdsn = shift;
+    my $driver = $client->driver_for($hashdsn);
+
+    ## Got some jobs! Randomize them to avoid contention between workers.
+    my @jobs = $RANDOMIZE_JOBS ? shuffle(@_) : @_;
+
+  JOB:
+    while (my $job = shift @jobs) {
+        ## Convert the funcid to a funcname, based on this database's map.
+        $job->funcname( $client->funcid_to_name($driver, $hashdsn, $job->funcid) );
+
+        ## Update the job's grabbed_until column so that
+        ## no one else takes it.
+        my $worker_class = $job->funcname;
+        my $old_grabbed_until = $job->grabbed_until;
+
+        my $server_time = $client->get_server_time($driver)
+            or die "expected a server time";
+
+        $job->grabbed_until($server_time + ($worker_class->grab_for || 1));
+
+        ## Update the job in the database, and end the transaction.
+        if ($driver->update($job, { grabbed_until => $old_grabbed_until }) < 1) {
+            ## We lost the race to get this particular job--another worker must
+            ## have got it and already updated it. Move on to the next job.
+            $TheSchwartz::T_LOST_RACE->() if $TheSchwartz::T_LOST_RACE;
+            next JOB;
+        }
+
+        ## Now prepare the job, and return it.
+        my $handle = TheSchwartz::JobHandle->new({
+            dsn_hashed => $hashdsn,
+            jobid      => $job->jobid,
+        });
+        $handle->client($client);
+        $job->handle($handle);
+        return $job;
+    }
+
+    return undef;
+}
+
 1;