This file is indexed.

/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