view release on metacpan or  search on metacpan

lib/EV/Gearman.pm  view on Meta::CPAN

The handle binds the per-job event callbacks to the right submission
even when many submissions are in flight at once.

Worker lifecycle:

    CAN_DO(func)         -->                              # advertise
                         repeat ----------------------+
    GRAB_JOB             -->                          |
                         <--   JOB_ASSIGN(h,fn,wl)    |   # got work
                                 ... user callback   |
    WORK_COMPLETE(h, r)  -->                          |
                         -------------- or -----------+
                         <--   NO_JOB                 |
    PRE_SLEEP            -->                          |
                         <--   NOOP                   |   # wake-up
                         ----------------------------+

C<EV::Gearman> drives that state machine in C; the per-function
callback only sees a ready-to-process L<EV::Gearman::Job>.

=head1 ENCODING

All function names, payloads, results, and handles are byte strings
on the wire. Encode UTF-8 yourself before passing data in:

    use Encode;
    $cli->submit_job(reverse => encode_utf8($str), sub {
        my $result = decode_utf8($_[0] // '');
        ...
    });

Workload and result values can contain arbitrary bytes including
embedded NULs — Gearman's framing puts the payload last in the
packet and uses the header's length field, so it is not NUL-bounded.

=head1 CALLBACK CONVENTIONS

Every command callback receives C<($result, $err)>. On success
C<$err> is C<undef>; on failure C<$err> is a string like
C<"disconnected">, C<"job failed">, C<"command timeout">, or text
forwarded from the server (e.g. C<"INVALID_FUNCTION_NAME: ...">).

Callback exceptions are caught with C<G_EVAL> and surfaced via
C<warn> so a stray C<die> from your code never unwinds the libev
event loop. Use C<EV::break> to abort the loop deliberately.

=head1 CONSTRUCTOR

=head2 new(%options)

    my $g = EV::Gearman->new(
        host             => '127.0.0.1',
        port             => 4730,
        on_error         => sub { warn "@_" },
        on_connect       => sub { ... },
        on_disconnect    => sub { ... },
        connect_timeout  => 5_000,    # ms
        command_timeout  => 30_000,   # ms
        reconnect        => 1,
        reconnect_delay  => 1000,     # ms
        keepalive        => 60,       # seconds (TCP only)
        exceptions       => 1,        # request "exceptions" option
        client_id        => "worker-$$",
        grab_unique      => 1,        # use GRAB_JOB_UNIQ
    );

If C<host> (or C<path>) is given, a non-blocking connect starts
immediately. With neither, the object is unconfigured; call
C<< $g->connect >> / C<< $g->connect_unix >> later.

All keys default to C<undef> unless noted. Booleans accept any Perl
truthy value.

=head3 Connection

=over

=item C<host =E<gt> $str>

=item C<port =E<gt> $int>

TCP host and port. Default port: C<4730>. Mutually exclusive with
C<path>.

Name resolution is currently synchronous: a non-numeric C<host> is
passed straight to C<getaddrinfo>, which can block the event loop
for the system resolver timeout. Pass an IP literal (or pre-resolve
once) to keep reconnect cycles fully non-blocking.

=item C<path =E<gt> $str>

Unix-domain socket path. Mutually exclusive with C<host>.

Note that a C<connect(2)> to a unix socket completes inline, so the
connection is fully established (and C<on_connect> has already fired)
before C<new> returns — a C<< $g->on_connect(...) >> assigned after
construction will never fire for it. Pass C<on_connect> to the
constructor instead (the same can theoretically happen for a TCP
connect that completes immediately).

=item C<loop =E<gt> $ev_loop>

EV loop to attach to. Default: C<EV::default_loop>.

=item C<priority =E<gt> $num>

EV watcher priority in C<-2 .. +2>. Higher = serviced before other
EV watchers in the same iteration. Default C<0>.

=item C<keepalive =E<gt> $seconds>

TCP keepalive idle interval. C<0> disables. Ignored on Unix sockets.

=back

=head3 Timeouts

=over

=item C<connect_timeout =E<gt> $ms>

Abort an in-progress non-blocking connect after this many ms. C<0>
= no timeout (default).

=item C<command_timeout =E<gt> $ms>

Per-request timeout. The request at the head of the pending queue is
given this many ms from the moment it is written to the socket; if it
is still unanswered when its budget expires, the connection is torn
down with C<"command timeout">. The budget is independent of
unrelated traffic: other packets arriving meanwhile neither extend
the head request's budget nor shorten it, and once the head is
answered the next request's own budget applies. A slow reply that
keeps dribbling in is therefore safe as long as the request completes
within its budget — and a genuinely stuck request dies on schedule
even on an otherwise busy connection. C<0> = no timeout (default).

=back

=head3 Reconnect

=over

=item C<reconnect =E<gt> $bool>

Enable automatic reconnect on transport errors.

=item C<reconnect_delay =E<gt> $ms>

Wait this many ms before each reconnect attempt. Default C<1000>.
The delay is always honored via a timer, so even C<0> defers
through the event loop (no synchronous retry recursion).

=item C<max_reconnect_attempts =E<gt> $num>

Give up after this many consecutive failures and emit
C<"max reconnect attempts reached">. C<0> = unlimited (default).

=back

After a reconnect, all worker C<CAN_DO>/C<CAN_DO_TIMEOUT>
registrations and the C<exceptions> option are re-sent
automatically.

=head3 Worker / option flags

=over

=item C<exceptions =E<gt> $bool>

If true, the C<exceptions> option is sent on every connect, so
foreground clients receive C<WORK_EXCEPTION> packets. For workers,

lib/EV/Gearman.pm  view on Meta::CPAN

gearmand answers a single-line C<ERR> instead); everything else is
treated as single-line.

A single-line reply beginning with C<ERR > (gearmand's text-protocol
error prefix, e.g. for an unrecognized command) is delivered as the
error argument: C<$cb-E<gt>(undef, "ERR UNKNOWN_COMMAND ...")> — not
as a successful result. The connection stays up.

If a command not in the multi-line set above nevertheless gets a
C<".\n">-terminated multi-line reply from the server, only the first
line is delivered; the leftover lines cannot be attributed to any
request and the connection is torn down with a protocol error.

Because the text protocol has no length prefix, a buffered reply is
capped at 16 MiB: a peer that keeps streaming without ever sending
the terminator is dropped with an C<"admin response too large">
error (and the normal reconnect logic applies).

=head2 server_status([$cb])

Tab-separated lines: C<FUNC \t TOTAL_JOBS \t RUNNING_JOBS \t WORKERS>.

=head2 server_workers([$cb])

One line per connected worker.

=head2 server_version([$cb])

Single-line reply (e.g. C<"OK 1.1.21+ds">).

=head2 maxqueue($func, $size, [$cb])

Set the per-function queue size cap. Reply is C<"OK\n">.

=head1 INTROSPECTION

=head2 pending_count

Number of requests sent and awaiting a response — binary client
requests, admin (text-protocol) commands, and the worker's in-flight
C<GRAB_JOB> all count.

=head2 waiting_count

Number of requests held in the local pre-connect queue.

=head2 active_count

Number of foreground jobs whose handle has been received but
which haven't yet completed.

=head1 ACCESSORS

These tunables have a getter / setter of the same name. Calling
without arguments reads the current value; with one argument, writes
and (where meaningful) takes effect immediately:

    $g->connect_timeout($ms);
    $g->command_timeout($ms);
    $g->priority($num);
    $g->keepalive($seconds);
    $g->on_error($cb);         # set; pass undef to clear
    $g->on_connect($cb);
    $g->on_disconnect($cb);

The remaining C<new> options (C<host>, C<port>, C<path>,
C<exceptions>, C<client_id>, C<grab_unique>, ...) are set once at
construction and have no accessor.

C<reconnect> is the exception — it is a setter only; pass C<0>/C<1>
plus optional new delay and attempt cap:

    $g->reconnect($enable, [$delay_ms], [$max_attempts]);

Omitting C<$delay_ms> / C<$max_attempts> leaves the previously
configured values unchanged.

=head2 reconnect_enabled

The getter that C<reconnect> lacks: returns true while automatic
reconnect is enabled, false otherwise. Takes no arguments.

=head1 LIFECYCLE AND DESTRUCTION

When a connection drops, the FIFO of pending requests is drained
with C<(undef, "disconnected")>; foreground active jobs are drained
with the same error. Reconnect (if enabled) re-runs the connect
sequence and re-registers worker abilities.

When the C<EV::Gearman> object goes out of scope, every pending
and active callback fires once with C<(undef, "disconnected")>,
then the FD is closed. The clean-shutdown idiom is:

    $g->disconnect;            # drains queues, fires on_disconnect
    undef $g;

If callbacks close over C<$g> (a common mistake — every reference
inside a closure keeps the object alive), break the cycle first:

    $g->on_error(undef);
    $g->on_connect(undef);
    $g->on_disconnect(undef);
    undef $g;

DESTROY is reentrancy-safe: if a callback fired during teardown
drops the last external reference to a separate C<EV::Gearman>,
that object's DESTROY is correctly deferred and run once unwound.

=head1 PERFORMANCE

Loopback benchmark on Linux, Perl 5.40, gearmand 1.1.21, single
worker (always L<EV::Gearman> so the worker isn't the bottleneck).

C<bench/benchmark.pl> measures one client by itself:

                                     ops/sec
    Pipelined foreground jobs        ~53,000
    Sequential round-trip            ~19,000
    Background submissions          ~280,000

C<bench/vs.pl> compares against the existing CPAN clients
(L<Gearman::Client> 2.004.015 sync, L<AnyEvent::Gearman> 0.10
async). Numbers in operations / second:

                          EV::Gearman   AnyEvent::Gearman   Gearman::Client
    pipelined foreground   ~51,000          ~5,200              n/a (1)
    sequential round-trip  ~19,000          ~5,900             ~5,400
    background submits    ~248,000          ~5,400              n/a (1)

    (1) Gearman::Client is synchronous — it has neither pipelining
        nor concurrent background submits.

EV::Gearman is roughly B<10x> the foreground throughput of
AnyEvent::Gearman, B<45x> the background submission rate, and
B<3x> the sequential round-trip rate. The gap comes from three
places:

=over

=item *

Pipelining is the default. Submitting N jobs in a tight loop
ships them in batched writes; responses are demultiplexed by
handle as they stream back. AnyEvent::Gearman is async but
serializes one request per round-trip, so it pays a full RTT per
job.

=item *

The protocol implementation is C/XS — packet encode/decode,
buffer growth, and FIFO bookkeeping run without per-call Perl
allocations.

=item *

The IO layer is direct C<ev_io> on the gearmand socket, so each
read/write involves no AnyEvent guard-object construction or
backend-dispatch overhead.