EV-ClickHouse
view release on metacpan or search on metacpan
size => 8, # other %args pass through to ::new
);
$pool->query($sql, $cb);
$pool->insert($table, $data, $cb);
$pool->drain(sub { ... }); # all connections drained
$pool->finish;
Built-in connection pool. Each member is an independent
`EV::ClickHouse` with its own `auto_reconnect`, send queue, and
in-flight callback queue, so a hung query on one connection doesn't
block the others. Dispatch picks the least-busy connection; ties are
broken round-robin.
The Pool exposes per-pick dispatch via `query`, `insert`, `ping`,
`for_table`, `iterate`, `insert_streamer`; aggregate stats via
`size`, `pending_count`, `conns` (the underlying connection list);
and broadcast lifecycle methods `drain`, `finish`, `cancel`,
`skip_pending`, `reset` (each affects every member because the state
they touch is owned per connection, not per query). The broadcast
`cancel`, `skip_pending`, and `reset` methods wrap each per-member
call in `eval` so a member that croaks doesn't abort the broadcast;
the body yourself if you need them. Useful for one-off per-member
work that doesn't justify a new broadcast method (e.g. resetting a
counter, asking each member for `last_error_code`, kicking off a
custom probe).
Queries that need server-side state (temporary tables, session
variables) must use a single connection, not a Pool, since successive
calls may land on different members.
`$pool->with_session(sub { my ($conn, $release) = @_; ... })`
checks out a least-busy member and "pins" it for the duration of the
callback: while pinned, `_pick` avoids that member when other
callers request a connection (it remains selectable as a fallback if
every other member is unavailable). The callback must call
`$release->()` when its multi-query sequence completes - typically
from the innermost query's callback so the pin lasts across the
async chain.
$pool->with_session(sub {
my ($ch, $release) = @_;
$ch->query("create temporary table t (n UInt32)", sub {
#!/usr/bin/env perl
# Connection pool: fan out N concurrent queries through a fixed pool.
# Pool->_pick chooses the least-busy connection; ties round-robin.
use strict;
use warnings;
use EV;
use EV::ClickHouse;
my $pool = EV::ClickHouse::Pool->new(
host => $ENV{CLICKHOUSE_HOST} // '127.0.0.1',
port => $ENV{CLICKHOUSE_NATIVE_PORT} // 9000,
protocol => 'native',
size => 8,
lib/EV/ClickHouse.pm view on Meta::CPAN
return $cb->($err) if $err;
$self->{columns} = [ map { $_->{name} } @{ $info->{columns} } ];
$cb->(undef);
});
return;
}
package EV::ClickHouse::Pool;
use Scalar::Util qw(refaddr);
# Built-in connection pool. Round-robin dispatch with least-busy fallback;
# each connection is independent (own auto_reconnect, own send_queue),
# so a hung query on one doesn't block the others. Pass any EV::ClickHouse
# constructor option in %args; it's applied to every pool member.
#
# my $pool = EV::ClickHouse::Pool->new(host => 'ch', size => 10, ...);
# $pool->query($sql, $cb);
# $pool->insert($table, $data, $cb);
# $pool->drain(sub { ... }); # all connections drained
# $pool->finish;
sub new {
lib/EV/ClickHouse.pm view on Meta::CPAN
else { $ch->query($sql, $obs) }
1;
} or do {
$out[$i]{err} = "$@" || "fan_out: dispatch failed";
$deliver->();
};
}
return;
}
# Checkout-style pin: hand the user a least-busy member, run their cb
# with ($conn, $release). Until $release->() is called, the pool will
# avoid handing the same member to other callers via _pick — useful
# for temp tables / SET / session-state work that must land on the
# same connection across multiple queries.
sub with_session {
my ($self, $cb) = @_;
die "Usage: \$pool->with_session(\$cb)" unless ref($cb) eq 'CODE';
my $ch = $self->_pick;
my $r = refaddr $ch;
$self->{_pinned}{$r} = ($self->{_pinned}{$r} // 0) + 1;
lib/EV/ClickHouse.pm view on Meta::CPAN
size => 8, # other %args pass through to ::new
);
$pool->query($sql, $cb);
$pool->insert($table, $data, $cb);
$pool->drain(sub { ... }); # all connections drained
$pool->finish;
Built-in connection pool. Each member is an independent
C<EV::ClickHouse> with its own C<auto_reconnect>, send queue, and
in-flight callback queue, so a hung query on one connection doesn't
block the others. Dispatch picks the least-busy connection; ties are
broken round-robin.
The Pool exposes per-pick dispatch via C<query>, C<insert>, C<ping>,
C<for_table>, C<iterate>, C<insert_streamer>; aggregate stats via
C<size>, C<pending_count>, C<conns> (the underlying connection list);
and broadcast lifecycle methods C<drain>, C<finish>, C<cancel>,
C<skip_pending>, C<reset> (each affects every member because the state
they touch is owned per connection, not per query). The broadcast
C<cancel>, C<skip_pending>, and C<reset> methods wrap each per-member
call in C<eval> so a member that croaks doesn't abort the broadcast;
lib/EV/ClickHouse.pm view on Meta::CPAN
the body yourself if you need them. Useful for one-off per-member
work that doesn't justify a new broadcast method (e.g. resetting a
counter, asking each member for C<last_error_code>, kicking off a
custom probe).
Queries that need server-side state (temporary tables, session
variables) must use a single connection, not a Pool, since successive
calls may land on different members.
C<<< $pool->with_session(sub { my ($conn, $release) = @_; ... }) >>>
checks out a least-busy member and "pins" it for the duration of the
callback: while pinned, C<_pick> avoids that member when other
callers request a connection (it remains selectable as a fallback if
every other member is unavailable). The callback must call
C<$release-E<gt>()> when its multi-query sequence completes - typically
from the innermost query's callback so the pin lasts across the
async chain.
$pool->with_session(sub {
my ($ch, $release) = @_;
$ch->query("create temporary table t (n UInt32)", sub {
t/18_cancel.t view on Meta::CPAN
use Time::HiRes qw(time);
my $ch;
my ($next_ok, $start, $first_elapsed);
$ch = EV::ClickHouse->new(
host => $host,
port => $nat_port,
protocol => 'native',
on_connect => sub {
$start = time();
# Heavy aggregation that keeps the server busy long enough for
# the cancel to be observable. ClickHouse's sleep() can't be
# interrupted mid-call, so use a real workload instead.
$ch->query(
"select count() from numbers(50000000000) where number % 7 = 0",
sub {
$first_elapsed = time() - $start;
# Follow-up query to prove the connection is still good.
$ch->query("select 1+1", sub {
my ($rows, $err2) = @_;
$next_ok = !$err2 && $rows && $rows->[0][0] == 2;
t/28_pass2_coverage.t view on Meta::CPAN
});
EV::timer(0.1, 0, sub { $ch->cancel });
},
on_error => sub { },
);
run_with_timeout(5);
ok $ok, '$ch->reset from cancel cb survives outer teardown';
$ch->finish if $ch->is_connected;
}
# Regression: Pool::_pick prefers any idle member over a busy one
# (regression: the previous logic round-robined whenever ANY member was
# idle, which could route the next pick to a busy connection).
{
my $pool = EV::ClickHouse::Pool->new(
host => $host, port => $nport, protocol => 'native',
size => 3,
);
my @c = $pool->conns;
# Wait for all to connect.
my $cn = 0;
$_->ping(sub { $cn++ }) for @c;
my $w = EV::timer(0, 0.05, sub { EV::break if $cn >= 3 });
run_with_timeout(5);
undef $w;
# Force c[0] busy with a long sleep, leave c[1], c[2] idle.
$c[0]->query("select sleep(2)", sub { });
# Now _pick should NEVER return c[0] while it's busy.
my %picks;
$picks{ "$_" }++ for map { $pool->_pick } 1 .. 6;
ok !exists $picks{"$c[0]"},
'Pool::_pick avoids the busy member when others are idle';
eval { $c[0]->cancel };
eval { $pool->finish };
}
# Coverage: reconnect_jitter accepted, insert_streamer named-rows.
# 13. reconnect_jitter is accepted as a constructor option without croaking,
# and a fully-disabled value (0) keeps backoff deterministic.
{
my $ok;
t/30_pool_distribution.t view on Meta::CPAN
#!/usr/bin/env perl
# Pool fan-out distribution test.
# - Launch a Pool of 4 native connections.
# - Issue 200 quick selects.
# - Each query callback records which connection handled it (by refaddr).
# - Assert no member handled less than 10% or more than 50% of the load
# (i.e. roughly balanced - allowing for least-busy bias near startup).
use strict;
use warnings;
use Test::More;
use EV;
use EV::ClickHouse;
use IO::Socket::INET;
use Scalar::Util qw(refaddr);
my $host = $ENV{TEST_CLICKHOUSE_HOST} || '127.0.0.1';
my $nport = $ENV{TEST_CLICKHOUSE_NATIVE_PORT} || 9000;
( run in 2.980 seconds using v1.01-cache-2.11-cpan-600a1bdf6e4 )