/usr/share/perl5/Gearman/Task.pm is in libgearman-client-perl 1.11-3.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338 339 340 341 342 343 344 345 346 347 348 349 350 351 352 353 | package Gearman::Task;
use strict;
use Carp ();
use String::CRC32 ();
use Gearman::Taskset;
use Gearman::Util;
BEGIN {
my $storable = eval { require Storable; 1 }
if !defined &RECEIVE_EXCEPTIONS || RECEIVE_EXCEPTIONS();
$storable ||= 0;
if (defined &RECEIVE_EXCEPTIONS) {
die "Exceptions support requires Storable: $@";
} else {
eval "sub RECEIVE_EXCEPTIONS () { $storable }";
die "Couldn't define RECEIVE_EXCEPTIONS: $@\n" if $@;
}
}
# constructor, given: ($func, $argref, $opts);
sub new {
my $class = shift;
my $self = $class;
$self = fields::new($class) unless ref $self;
$self->{func} = shift
or Carp::croak("No function given");
$self->{argref} = shift || do { my $empty = ""; \$empty; };
Carp::croak("Argref not a scalar reference") unless ref $self->{argref} eq "SCALAR";
my $opts = shift || {};
for my $k (qw( uniq
on_complete on_exception on_fail on_retry on_status
retry_count timeout high_priority try_timeout
)) {
$self->{$k} = delete $opts->{$k};
}
$self->{retry_count} ||= 0;
$self->{is_finished} = 0; # bool: if success or fail has been called yet on this.
if (%{$opts}) {
Carp::croak("Unknown option(s): " . join(", ", sort keys %$opts));
}
$self->{retries_done} = 0;
return $self;
}
sub run_hook {
my Gearman::Task $self = shift;
my $hookname = shift || return;
my $hook = $self->{hooks}->{$hookname};
return unless $hook;
eval { $hook->(@_) };
warn "Gearman::Task hook '$hookname' threw error: $@\n" if $@;
}
sub add_hook {
my Gearman::Task $self = shift;
my $hookname = shift || return;
if (@_) {
$self->{hooks}->{$hookname} = shift;
} else {
delete $self->{hooks}->{$hookname};
}
}
sub is_finished {
my Gearman::Task $task = $_[0];
return $task->{is_finished};
}
sub taskset {
my Gearman::Task $task = shift;
# getter
return $task->{taskset} unless @_;
# setter
my Gearman::Taskset $ts = shift;
$task->{taskset} = $ts;
my $merge_on = $task->{uniq} && $task->{uniq} eq "-" ?
$task->{argref} : \ $task->{uniq};
if ($$merge_on) {
my $hash_num = _hashfunc($merge_on);
$task->{jssock} = $ts->_get_hashed_sock($hash_num);
} else {
$task->{jssock} = $ts->_get_default_sock;
}
return $task->{taskset};
}
# returns undef on non-uniq packet, or the hash value (0-32767) if uniq
sub hash {
my Gearman::Task $task = shift;
my $merge_on = $task->{uniq} && $task->{uniq} eq "-" ?
$task->{argref} : \ $task->{uniq};
if ($$merge_on) {
return _hashfunc( $merge_on );
} else {
return undef;
}
}
# returns number in range [0,32767] given a scalarref
sub _hashfunc {
return (String::CRC32::crc32(${ shift() }) >> 16) & 0x7fff;
}
sub pack_submit_packet {
my Gearman::Task $task = shift;
my Gearman::Client $client = shift;
my $is_background = shift;
my $mode = $is_background ?
"submit_job_bg" :
($task->{high_priority} ?
"submit_job_high" :
"submit_job");
my $func = $task->{func};
if (my $prefix = $client && $client->prefix) {
$func = join "\t", $prefix, $task->{func};
}
return Gearman::Util::pack_req_command($mode,
join("\0",
$func || '',
$task->{uniq} || '',
${ $task->{argref} } || ''));
}
sub fail {
my Gearman::Task $task = shift;
my $reason = shift;
return if $task->{is_finished};
# try to retry, if we can
if ($task->{retries_done} < $task->{retry_count}) {
$task->{retries_done}++;
$task->{on_retry}->($task->{retries_done}) if $task->{on_retry};
$task->handle(undef);
return $task->{taskset}->add_task($task);
}
$task->final_fail($reason);
}
sub final_fail {
my Gearman::Task $task = $_[0];
my $reason = $_[1];
return if $task->{is_finished};
$task->{is_finished} = $_[1] || 1;
$task->run_hook('final_fail', $task);
$task->{on_fail}->($reason) if $task->{on_fail};
$task->{on_post_hooks}->() if $task->{on_post_hooks};
$task->wipe;
return undef;
}
sub exception {
my Gearman::Task $task = shift;
return unless RECEIVE_EXCEPTIONS;
my $exception_ref = shift;
my $exception = Storable::thaw($$exception_ref);
$task->{on_exception}->($$exception) if $task->{on_exception};
return;
}
sub complete {
my Gearman::Task $task = shift;
return if $task->{is_finished};
my $result_ref = shift;
$task->{is_finished} = 'complete';
$task->run_hook('complete', $task);
$task->{on_complete}->($result_ref) if $task->{on_complete};
$task->{on_post_hooks}->() if $task->{on_post_hooks};
$task->wipe;
}
sub status {
my Gearman::Task $task = shift;
return if $task->{is_finished};
return unless $task->{on_status};
my ($nu, $de) = @_;
$task->{on_status}->($nu, $de);
}
# getter/setter for the fully-qualified handle of form "IP:port//shandle" where
# shandle is an opaque handle specific to the job server running on IP:port
sub handle {
my Gearman::Task $task = shift;
return $task->{handle} unless @_;
return $task->{handle} = shift;
}
sub set_on_post_hooks {
my Gearman::Task $task = shift;
my $code = shift;
$task->{on_post_hooks} = $code;
}
sub wipe {
my Gearman::Task $task = shift;
foreach my $f (qw(on_post_hooks on_complete on_fail on_retry on_status hooks)) {
$task->{$f} = undef;
}
}
sub func {
my Gearman::Task $task = shift;
return $task->{func};
}
sub timeout {
my Gearman::Task $task = shift;
return $task->{timeout} unless @_;
return $task->{timeout} = shift;
}
1;
__END__
=head1 NAME
Gearman::Task - a task in Gearman, from the point of view of a client
=head1 SYNOPSIS
my $task = Gearman::Task->new("add", "1+2", {
.....
};
$taskset->add_task($task);
$client->do_task($task);
$client->dispatch_background($task);
=head1 DESCRIPTION
I<Gearman::Task> is a Gearman::Client's representation of a task to be
done.
=head1 USAGE
=head2 Gearman::Task->new($func, $arg, \%options)
Creates a new I<Gearman::Task> object, and returns the object.
I<$func> is the function name to be run. (that you have a worker registered to process)
I<$arg> is an opaque scalar or scalarref representing the argument(s)
to pass to the distributed function. If you want to pass multiple
arguments, you must encode them somehow into this one. That's up to
you and your worker.
I<%options> can contain:
=over 4
=item * uniq
A key which indicates to the server that other tasks with the same
function name and key will be merged into one. That is, the task
will be run just once, but all the listeners waiting on that job
will get the response multiplexed back to them.
Uniq may also contain the magic value "-" (a single hyphen) which
means the uniq key is the contents of the args.
=item * on_complete
A subroutine reference to be invoked when the task is completed. The
subroutine will be passed a reference to the return value from the worker
process.
=item * on_fail
A subroutine reference to be invoked when the task fails (or fails for
the last time, if retries were specified). No arguments are
passed to this callback. This callback won't be called after a failure
if more retries are still possible.
=item * on_retry
A subroutine reference to be invoked when the task fails, but is about
to be retried.
Is passed one argument, what retry attempt number this is. (starts with 1)
=item * on_status
A subroutine reference to be invoked if the task emits status updates.
Arguments passed to the subref are ($numerator, $denominator), where those
are left up to the client and job to determine.
=item * retry_count
Number of times job will be retried if there are failures. Defaults to 0.
=item * high_priority
Boolean, whether this job should take priority over other jobs already
enqueued.
=item * timeout
Automatically fail, calling your on_fail callback, after this many
seconds have elapsed without an on_fail or on_complete being
called. Defaults to 0, which means never. Bypasses any retry_count
remaining.
=item * try_timeout
Automatically fail, calling your on_retry callback (or on_fail if out of
retries), after this many seconds have elapsed. Defaults to 0, which means
never.
=back
=head2 $task->is_finished
Returns bool: whether or not task is totally done (on_failure or
on_complete callback has been called)
=cut
|