EV-Gearman
view release on metacpan or search on metacpan
xt/03_reconnect_under_load.t
xt/04_destroy_in_callback.t
xt/05_async_concurrency.t
xt/06_binary_payload.t
xt/07_admin_protocol.t
xt/08_event_routing.t
xt/09_spawned_gearmand.t
xt/10_gearmand_restart.t
xt/11_event_storm.t
xt/12_huge_payload.t
xt/13_long_running_keepalive.t
xt/14_fork_safety.t
xt/15_handle_uniqueness.t
xt/16_buffer_shrink.t
xt/89_leak.t
xt/90_valgrind.t
xt/91_asan.t
xt/92_pod.t
xt/93_pod_coverage.t
xt/94_manifest.t
xt/95_kwalitee.t
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 `host` (or `path`) is given, a non-blocking connect starts
immediately. With neither, the object is unconfigured; call
`$g->connect` / `$g->connect_unix` later.
All keys default to `undef` unless noted. Booleans accept any Perl
- `loop => $ev_loop`
EV loop to attach to. Default: `EV::default_loop`.
- `priority => $num`
EV watcher priority in `-2 .. +2`. Higher = serviced before other
EV watchers in the same iteration. Default `0`.
- `keepalive => $seconds`
TCP keepalive idle interval. `0` disables. Ignored on Unix sockets.
### Timeouts
- `connect_timeout => $ms`
Abort an in-progress non-blocking connect after this many ms. `0`
= no timeout (default).
- `command_timeout => $ms`
# 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 `new` options (`host`, `port`, `path`,
`exceptions`, `client_id`, `grab_unique`, ...) are set once at
construction and have no accessor.
`reconnect` is the exception — it is a setter only; pass `0`/`1`
plus optional new delay and attempt cap:
sequence and re-registers worker abilities.
When the `EV::Gearman` object goes out of scope, every pending
and active callback fires once with `(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 `$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 `EV::Gearman`,
that object's DESTROY is correctly deferred and run once unwound.
eg/reconnect.pl view on Meta::CPAN
use warnings;
use EV;
use EV::Gearman;
my $g = EV::Gearman->new(
host => '127.0.0.1',
port => 4730,
reconnect => 1,
reconnect_delay => 500,
max_reconnect_attempts => 0, # unlimited
keepalive => 30,
on_connect => sub { warn "connected\n" },
on_disconnect => sub { warn "disconnected\n" },
on_error => sub { warn "error: @_\n" },
);
$g->register_function(echo => sub { $_[0]->workload });
$g->work;
EV::run;
lib/EV/Gearman.pm view on Meta::CPAN
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
lib/EV/Gearman.pm view on Meta::CPAN
=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>
lib/EV/Gearman.pm view on Meta::CPAN
=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:
lib/EV/Gearman.pm view on Meta::CPAN
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.
lib/EV/Gearman/Job.pm view on Meta::CPAN
C<WORK_COMPLETE>. C<die> becomes C<WORK_FAIL>. The job methods
below are still available for sending intermediate events.
In B<async> mode, the callback returns immediately; you must
explicitly call C<complete>, C<fail>, or C<exception> later. The
job object can be stashed in a closure or any other long-lived
container — it outlives the connection safely.
If the underlying L<EV::Gearman> connection has been destroyed by
the time you call a job method, the call C<croak>s with
C<"client destroyed">. If the client is alive but currently
disconnected (even with reconnect armed), the call C<croak>s with
C<"not connected">: gearmand forgets the job when the connection
drops, so a packet queued for the next session would only earn a
C<JOB_NOT_FOUND> error there. The job holds an internal tombstone
reference that keeps the connection's control block allocated
(but torn down) until every job referencing it is released, so
this check is sound — it never reads freed memory. The
back-pointer is stored as perl magic, not a hash key: user code,
hash walkers, and serializers cannot see or clobber it, and a
job hash that did not come from a C<JOB_ASSIGN> croaks with
src/EV__Gearman.xs view on Meta::CPAN
int in_active_cleanup;
/* Live EV::Gearman::Job objects holding a tombstone reference on
this struct (via sv_magicext on the job HV). While > 0 the
struct is never Safefreed — a DESTROYed connection becomes an
inert tombstone until the last job releases it, so job_resolve's
magic-word check never reads freed memory. */
U32 job_refs;
/* Options */
int priority;
int keepalive;
/* Options sent on connect */
int opt_exceptions;
};
/* ================================================================
* Forward declarations
* ================================================================ */
static void io_cb(EV_P_ ev_io *w, int revents);
src/EV__Gearman.xs view on Meta::CPAN
static void cmd_timeout_cb(EV_P_ ev_timer *w, int revents);
static void start_reading(ev_gm_t *self);
static void stop_reading(ev_gm_t *self);
static void start_writing(ev_gm_t *self);
static void stop_writing(ev_gm_t *self);
static void start_connect(pTHX_ ev_gm_t *self);
static void cleanup_connection(pTHX_ ev_gm_t *self);
static void emit_error(pTHX_ ev_gm_t *self, const char *msg);
static void handle_disconnect(pTHX_ ev_gm_t *self, const char *reason);
static void schedule_reconnect(pTHX_ ev_gm_t *self);
static void apply_keepalive(ev_gm_t *self);
static void report_connect_error(pTHX_ ev_gm_t *self, const char *errbuf);
static void finish_connect_success(pTHX_ ev_gm_t *self);
static void stop_connect_timer(ev_gm_t *self);
static void stop_reconnect_timer(ev_gm_t *self);
static int check_destroyed(ev_gm_t *self);
static void clear_addrinfo(ev_gm_t *self);
static void connect_try_from(pTHX_ ev_gm_t *self, const char *prev_err);
static void cancel_pending(pTHX_ ev_gm_t *self, const char *reason);
static void cancel_waiting(pTHX_ ev_gm_t *self, const char *reason);
static void cancel_active(pTHX_ ev_gm_t *self, const char *reason);
src/EV__Gearman.xs view on Meta::CPAN
}
static void emit_connect(pTHX_ ev_gm_t *self) {
invoke_handler(aTHX_ self, self->on_connect, NULL, "on_connect");
}
static void emit_disconnect(pTHX_ ev_gm_t *self) {
invoke_handler(aTHX_ self, self->on_disconnect, NULL, "on_disconnect");
}
static void apply_keepalive(ev_gm_t *self) {
if (self->keepalive <= 0 || self->path) return;
int one = 1;
setsockopt(self->fd, SOL_SOCKET, SO_KEEPALIVE, &one, sizeof(one));
#ifdef TCP_KEEPIDLE
setsockopt(self->fd, IPPROTO_TCP, TCP_KEEPIDLE,
&self->keepalive, sizeof(self->keepalive));
#endif
}
/* Set O_NONBLOCK and FD_CLOEXEC on an already-open socket fd. The
CLOEXEC step prevents the fd leaking into child processes spawned
via fork+exec while the connection is up. Returns 0 on success,
-1 on failure (caller should close and error out). */
static int gm_set_socket_flags(int fd) {
int fl = fcntl(fd, F_GETFL);
if (fl < 0 || fcntl(fd, F_SETFL, fl | O_NONBLOCK) < 0) return -1;
src/EV__Gearman.xs view on Meta::CPAN
}
if (reason) {
emit_error(aTHX_ self, reason);
if (check_destroyed(self)) return;
}
}
static void schedule_reconnect(pTHX_ ev_gm_t *self) {
/* DESTROY during a callback leaves a zombie (magic FREED, struct
kept alive by callback_depth); arming a timer on it would point
libev at freed memory once the depth unwinds. */
if (self->magic != GM_MAGIC_ALIVE) return;
if (self->reconnect_timer_active) return;
if (self->max_reconnect_attempts > 0 &&
self->reconnect_attempts >= self->max_reconnect_attempts) {
emit_error(aTHX_ self, "max reconnect attempts reached");
return;
}
self->reconnect_attempts++;
ev_tstamp delay = (ev_tstamp)self->reconnect_delay_ms / 1000.0;
src/EV__Gearman.xs view on Meta::CPAN
self->callback_depth--;
if (check_destroyed(self)) return;
if (!self->intentional_disconnect && self->reconnect)
schedule_reconnect(aTHX_ self);
}
static void finish_connect_success(pTHX_ ev_gm_t *self) {
self->reconnect_attempts = 0;
start_reading(self);
apply_keepalive(self);
send_options_and_id(aTHX_ self);
register_all_functions(aTHX_ self);
emit_connect(aTHX_ self);
if (check_destroyed(self)) return;
send_pending_waits(aTHX_ self);
if (check_destroyed(self)) return;
src/EV__Gearman.xs view on Meta::CPAN
break;
}
default:
break;
}
}
/* Dispatch a JOB_ASSIGN[_UNIQ]: invoke registered function callback,
* deliver result back to server (for sync mode). Async mode keeps a
* job object alive in Perl-land until user invokes complete/fail. */
static void worker_dispatch_job(pTHX_ ev_gm_t *self, ev_gm_req_t *r,
int with_unique, const char *body, size_t body_len)
{
int total = with_unique ? 4 : 3;
size_t hl, fl, ul, wl;
const char *h = gm_arg(body, body_len, 0, total, &hl);
const char *f = gm_arg(body, body_len, 1, total, &fl);
const char *u = with_unique ? gm_arg(body, body_len, 2, total, &ul) : NULL;
const char *w = gm_arg(body, body_len, with_unique ? 3 : 2, total, &wl);
src/EV__Gearman.xs view on Meta::CPAN
}
SvREFCNT_dec(jobref);
send_work_event(aTHX_ self, GM_CMD_WORK_FAIL, h, hl, NULL, 0);
worker_continue(aTHX_ self);
return;
}
/* Snapshot fn->async before invoking the user callback: the
callback is allowed to call cant_do() or reset_abilities(),
which would free `fn`, leaving us with a stale pointer. The
SV behind fn->cb stays alive during call_sv via Perl's own
sub-context refcount, so reading it is safe. */
int is_async = fn->async;
self->callback_depth++;
dSP;
ENTER; SAVETMPS;
PUSHMARK(SP);
XPUSHs(jobref);
PUTBACK;
int count = call_sv(fn->cb, is_async ? (G_DISCARD | G_EVAL) : (G_SCALAR | G_EVAL));
src/EV__Gearman.xs view on Meta::CPAN
sv_setsv(ERRSV, &PL_sv_undef);
if (!is_async && count > 0) {
retval = POPs;
SvREFCNT_inc(retval);
}
PUTBACK;
FREETMPS; LEAVE;
self->callback_depth--;
/* Release our owned jobref; if the callback stashed the job (async
mode) its own reference keeps the job alive. */
SvREFCNT_dec(jobref);
if (self->magic == GM_MAGIC_FREED) {
if (retval) SvREFCNT_dec(retval);
if (err_sv) SvREFCNT_dec(err_sv);
return;
}
if (!is_async) {
if (had_error) {
src/EV__Gearman.xs view on Meta::CPAN
ngx_queue_init(&RETVAL->wait_queue);
ngx_queue_init(&RETVAL->active_jobs);
ngx_queue_init(&RETVAL->functions);
Newx(RETVAL->rbuf, BUF_INIT_SIZE, char);
RETVAL->rbuf_cap = BUF_INIT_SIZE;
Newx(RETVAL->wbuf, BUF_INIT_SIZE, char);
RETVAL->wbuf_cap = BUF_INIT_SIZE;
/* Default error handler. eval_pv hands back the result without
transferring a reference we own (it stays mortal), so take one
to keep the sub alive past the current statement. */
RETVAL->on_error = eval_pv("sub { warn \"EV::Gearman error: @_\\n\" }", TRUE);
SvREFCNT_inc_simple_void_NN(RETVAL->on_error);
SV *host_sv = NULL, *path_sv = NULL;
int port = 4730;
int do_reconnect = 0, reconnect_delay = 1000, max_reconnect_attempts = 0;
RETVAL->loop = EV_DEFAULT;
int i;
for (i = 1; i < items; i += 2) {
src/EV__Gearman.xs view on Meta::CPAN
}
else if (strEQ(k, "on_connect")) {
if (SvOK(v) && SvROK(v)) RETVAL->on_connect = newSVsv(v);
}
else if (strEQ(k, "on_disconnect")) {
if (SvOK(v) && SvROK(v)) RETVAL->on_disconnect = newSVsv(v);
}
else if (strEQ(k, "connect_timeout")) RETVAL->connect_timeout_ms = SvIV(v);
else if (strEQ(k, "command_timeout")) RETVAL->command_timeout_ms = SvIV(v);
else if (strEQ(k, "priority")) RETVAL->priority = SvIV(v);
else if (strEQ(k, "keepalive")) RETVAL->keepalive = SvIV(v);
else if (strEQ(k, "reconnect")) do_reconnect = SvTRUE(v) ? 1 : 0;
else if (strEQ(k, "reconnect_delay")) reconnect_delay = SvIV(v);
else if (strEQ(k, "max_reconnect_attempts")) max_reconnect_attempts = SvIV(v);
else if (strEQ(k, "exceptions")) RETVAL->opt_exceptions = SvTRUE(v) ? 1 : 0;
else if (strEQ(k, "client_id")) {
if (SvOK(v)) RETVAL->client_id = savepv(SvPV_nolen(v));
}
else if (strEQ(k, "grab_unique")) RETVAL->worker_grab_uniq = SvTRUE(v) ? 1 : 0;
else if (strEQ(k, "loop")) {
if (!SvROK(v) || !sv_derived_from(v, "EV::Loop")) {
src/EV__Gearman.xs view on Meta::CPAN
} else {
ev_set_priority(&self->wio, self->priority);
}
}
RETVAL = self->priority;
}
OUTPUT:
RETVAL
int
keepalive(EV::Gearman self, ...)
CODE:
{
if (items > 1) {
self->keepalive = SvIV(ST(1));
if (self->keepalive < 0) self->keepalive = 0;
if (self->connected && self->fd >= 0)
apply_keepalive(self);
}
RETVAL = self->keepalive;
}
OUTPUT:
RETVAL
# ===== Job object methods (called from Perl via $job->complete etc.) =====
MODULE = EV::Gearman PACKAGE = EV::Gearman::Job
void
_send_event(SV *job_sv, int kind, SV *data_sv = &PL_sv_undef)
t/00_load.t view on Meta::CPAN
submit_job_bg submit_job_high_bg submit_job_low_bg
submit_job_epoch
get_status get_status_unique
option set_client_id
can_do cant_do reset_abilities
register_function unregister_function
work work_one work_stop grab_job all_yours
admin server_status server_workers server_version maxqueue
on_error on_connect on_disconnect
pending_count waiting_count active_count
connect_timeout command_timeout reconnect priority keepalive
);
ok(EV::Gearman::Job->can($_), "Job method $_") for qw(
handle function unique workload data
complete fail exception send_data warning status
);
done_testing;
t/05_misc.t view on Meta::CPAN
# ===== accessors =====
{
my $g = EV::Gearman->new;
$g->connect_timeout(1234);
is $g->connect_timeout, 1234, 'connect_timeout setter/getter';
$g->command_timeout(5678);
is $g->command_timeout, 5678, 'command_timeout setter/getter';
$g->priority(1);
is $g->priority, 1, 'priority setter/getter';
$g->keepalive(60);
is $g->keepalive, 60, 'keepalive setter/getter';
is $g->pending_count, 0, 'pending_count starts at 0';
is $g->waiting_count, 0, 'waiting_count starts at 0';
is $g->active_count, 0, 'active_count starts at 0';
ok !$g->is_connected, 'not connected';
}
# ===== sending commands before connect =====
{
my $g = EV::Gearman->new;
eval { $g->echo("x", sub {}) };
t/07_edge_cases.t view on Meta::CPAN
EV::run;
ok !$job, 'grab_job got no job';
is $err, 'no job', '"no job" error string';
}
# 7) accessor round-trip
{
my $g = EV::Gearman->new;
$g->priority(2);
is $g->priority, 2, 'priority round-trip';
$g->keepalive(120);
is $g->keepalive, 120, 'keepalive round-trip';
}
# 8) is_connected progresses through the connecting → connected phases
{
my $g = EV::Gearman->new(host => $host, port => $port);
ok $g->is_connected, 'is_connected true while connecting';
$g->on_connect(sub { EV::break });
my $w = EV::timer 3, 0, sub { EV::break };
EV::run;
ok $g->is_connected, 'is_connected true after on_connect';
t/17_arg_validation.t view on Meta::CPAN
# every documented key is still accepted — the croak must not be
# over-eager
my $g = EV::Gearman->new(
on_error => sub {},
on_connect => sub {},
on_disconnect => sub {},
connect_timeout => 100,
command_timeout => 100,
priority => 1,
keepalive => 0,
reconnect => 1,
reconnect_delay => 100,
max_reconnect_attempts => 2,
exceptions => 1,
client_id => 't17',
grab_unique => 1,
loop => EV::default_loop,
);
ok $g, 'all non-connect documented keys accepted';
my $g2 = EV::Gearman->new(host => '127.0.0.1', port => 4730, on_error => sub {});
t/17_arg_validation.t view on Meta::CPAN
{
my $err;
my $g = EV::Gearman->new(on_error => sub { $err = $_[0] });
$g->connect_unix('/tmp/' . ('x' x 200));
like $err, qr/path too long/, 'connect_unix: overlong path -> on_error';
}
# Payload over GM_MAX_PACKET (256 MiB) is rejected before any request
# is allocated. Needs one ~256 MiB transient scalar (a bounded single
# allocation, not list construction) and no gearmand: the object is
# "alive" (connecting) so the guard is reached, and the croak fires
# during the synchronous submit before any I/O.
{
my $g = EV::Gearman->new(host => '127.0.0.1', port => 4730);
my $huge = 'x' x (256 * 1024 * 1024 + 1);
eval { $g->submit_job('f', $huge, sub {}) };
like $@, qr/payload too large/, 'submit_job: oversized payload croaks';
undef $huge;
$g->reconnect(0); # stop any reconnect attempts before teardown
}
t/21_custom_loop.t view on Meta::CPAN
# loop => $ev_loop: the documented option must actually work. An
# EV::Loop object is a blessed scalar holding the loop pointer in its
# IV slot; reading the (NULL) PV slot used to store a NULL loop and
# segfault at the first ev_io_start. Also: bad values must croak, and
# the client must hold the loop alive.
use strict;
use warnings;
use Test::More;
use IO::Socket::INET;
use Scalar::Util qw(weaken);
use EV;
use EV::Gearman;
my $host = $ENV{TEST_GEARMAN_HOST} || '127.0.0.1';
my $port = $ENV{TEST_GEARMAN_PORT} || 4730;
t/21_custom_loop.t view on Meta::CPAN
# --- echo round trip on a custom loop, driven by $loop->run ---
my $loop = EV::Loop->new;
my ($res, $err);
my $g = EV::Gearman->new(loop => $loop, host => $host, port => $port);
$g->echo('custom-loop', sub { ($res, $err) = @_; $loop->break });
my $guard = $loop->timer(5, 0, sub { fail 'custom loop echo timeout'; $loop->break });
$loop->run;
is $err, undef, 'no error on custom loop';
is $res, 'custom-loop', 'echo round trip on custom EV::Loop';
# --- the client holds the loop alive ---
my $weak;
my $g2;
{
my $l2 = EV::Loop->new;
$weak = $l2;
weaken($weak);
$g2 = EV::Gearman->new(loop => $l2, host => $host, port => $port);
} # $l2's own strong ref dies here
ok defined($weak), 'loop survives after caller drops the only strong ref';
t/25_job_lifetime.t view on Meta::CPAN
my $probe = IO::Socket::INET->new(
PeerAddr => $host, PeerPort => $port, Proto => 'tcp', Timeout => 1,
);
plan skip_all => "no gearmand at $host:$port" unless $probe;
close $probe;
my $func = "lifetime_$$";
# T-D1-1: async worker stashes the job, the client object is destroyed
# from inside the callback, and a later job method must croak
# "client destroyed" — with the memory kept alive by the job's
# tombstone reference, not read from freed heap.
{
my $w = EV::Gearman->new(host => $host, port => $port);
my $c = EV::Gearman->new(host => $host, port => $port);
my $job;
$w->register_function($func => { async => 1 }, sub {
$job = $_[0];
undef $w; # DESTROY with a job outstanding
EV::break;
});
xt/03_reconnect_under_load.t view on Meta::CPAN
$w->work;
$g = EV::timer 1, 0, sub { EV::break };
EV::run;
ok $reconnects >= 2, "worker reconnected (got $reconnects)";
# Submit a job to verify the function is still registered after
# reconnect — the XS code re-sends CAN_DO on connect.
my $cli = EV::Gearman->new(host => $host, port => $port);
my ($r, $e);
$cli->on_connect(sub {
$cli->submit_job($func, "still-alive", sub { ($r, $e) = @_; EV::break });
});
$g = EV::timer 5, 0, sub { fail "post-reconnect timeout"; EV::break };
EV::run;
is $r, 'STILL-ALIVE', 'worker function works after reconnect';
}
done_testing;
xt/13_long_running_keepalive.t view on Meta::CPAN
# A long-idle async worker / client should not be torn down by the
# OS while waiting for a slow job. Exercises TCP keepalive path —
# without it, NATs / load balancers can drop idle connections.
use strict;
use warnings;
use Test::More;
use IO::Socket::INET;
use EV;
use EV::Gearman;
my $host = $ENV{TEST_GEARMAN_HOST} || '127.0.0.1';
my $port = $ENV{TEST_GEARMAN_PORT} || 4730;
my $probe = IO::Socket::INET->new(
PeerAddr => $host, PeerPort => $port, Proto => 'tcp', Timeout => 1,
);
plan skip_all => "no gearmand at $host:$port" unless $probe;
close $probe;
# Quick smoke: keepalive option flips socket options without croaking
# and survives a short idle round-trip. We can't actually verify the
# kernel sent KEEPIDLE probes without packet capture; this is a
# regression test for the option-plumbing path.
my $cli = EV::Gearman->new(
host => $host, port => $port, keepalive => 30,
);
my $wkr = EV::Gearman->new(
host => $host, port => $port, keepalive => 30,
);
my $func = "ka_$$";
my @timers; # retain the timer watchers across the callback return
$wkr->register_function($func => { async => 1 }, sub {
my $job = shift;
push @timers, EV::timer 1.0, 0, sub { $job->complete("kept-alive") };
});
$wkr->work;
is $cli->keepalive, 30, 'client keepalive set';
is $wkr->keepalive, 30, 'worker keepalive set';
my ($r, $e);
$cli->submit_job($func, "go", sub { ($r, $e) = @_; EV::break });
my $g = EV::timer 5, 0, sub { fail "ka timeout"; EV::break };
EV::run;
is $r, 'kept-alive', 'job completed under keepalive';
is $e, undef, 'no error';
# Setter at runtime
$cli->keepalive(60);
is $cli->keepalive, 60, 'keepalive runtime setter takes effect';
$cli->keepalive(0);
is $cli->keepalive, 0, 'keepalive can be cleared';
done_testing;
xt/89_leak.t view on Meta::CPAN
}
$settle = EV::timer 0.3, 0, sub { EV::break };
});
my $w = EV::timer 3, 0, sub { EV::break };
EV::run;
undef $g;
pass 'admin intermix cleanup';
}
# 6. Async worker DESTROYed with stashed jobs outstanding (T-D1-4):
# the client struct must become an inert tombstone kept alive by the
# jobs' magic references, job methods must croak "client destroyed",
# and releasing the jobs must free the tombstone — no definite leak,
# no use-after-free. (This is the scenario xt/90 and xt/91 drive.)
{
my $w = EV::Gearman->new(host => $host, port => $port);
my $c = EV::Gearman->new(host => $host, port => $port);
my @stash;
$w->register_function('xt_leak_tomb_'.$$ => { async => 1 }, sub {
push @stash, $_[0];
EV::break if @stash == 2;
( run in 1.603 second using v1.01-cache-2.11-cpan-14f38c9f855 )