busy results from the CPAN

busy

EV-ClickHouse

view release on metacpan or search on metacpan


# Register a callback that fires once the buffered row count drops
# to low_water (defaults to high_water/2; 0 if high_water not set).
# Pairs with on_high_water to close the backpressure loop:
#   on_high_water => sub { $producer->pause },
#   $streamer->await_drain(sub { $producer->resume });
# Fires synchronously if the buffer is already at/below the threshold.
sub await_drain {
    my ($self, $cb) = @_;
    die "Usage: \$streamer->await_drain(\$cb)" unless ref($cb) eq 'CODE';
    my $low = $self->{low_water} || 0;
    # Fire synchronously when nothing is in flight AND the buffer is
    # at/below low_water: there's no flush pending so no waiter would
    # ever fire otherwise.
    return $cb->(undef) if !$self->{in_flight} && @{ $self->{buffer} } <= $low;
    push @{ $self->{drain_waiters} }, $cb;
    return;
}

# Discard buffered rows + sticky error without finishing. Useful for
# "retry after permanent error" patterns where the producer wants to
# wipe the slate clean (typically after a schema-level fix) and keep
# pushing into the same streamer object. The underlying $ch is NOT
# touched - any in-flight batch already on the wire still completes.
sub reset {
    my ($self) = @_;
    $self->{buffer}            = [];
    $self->{sticky_err}        = undef;
    $self->{high_water_active} = 0;
    # Deliver any pending finish/drain callbacks with a reset error
    # rather than silently dropping them - quiet loss of a finish cb
    # leaves the producer waiting forever.
    my $pf = delete $self->{pending_finish};
    my @dw = @{ delete $self->{drain_waiters} || [] };
    $self->{drain_waiters} = [];
    $pf->(undef, 'streamer reset') if $pf;
    $_->('streamer reset') for @dw;
    return $self;
}

# Discover column names from the target table via for_table, then
# enable named-rows mode by populating $self->{columns}. Callback
# receives undef on success, error string on failure. Useful when
# the producer side doesn't know (or shouldn't care about) the
# schema in advance.
sub columns_from_table {
    my ($self, $cb) = @_;
    die "Usage: \$streamer->columns_from_table(\$cb)" unless ref($cb) eq 'CODE';
    $self->{ch}->for_table($self->{table}, sub {
        my ($info, $err) = @_;
        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 {
    my ($class, %args) = @_;
    my $size = delete $args{size} || 4;
    die "Pool size must be >= 1" if $size < 1;
    # Circuit breaker per member. After `circuit_threshold` consecutive
    # query/insert failures, mark the member dead for `circuit_cooldown`
    # seconds; _pick skips dead members. 0 disables.
    my $threshold = delete $args{circuit_threshold} || 0;
    my $cooldown  = delete $args{circuit_cooldown}  || 30;
    my @conns;
    for (1 .. $size) {
        push @conns, EV::ClickHouse->new(%args);
    }
    bless {
        conns      => \@conns,
        idx        => 0,
        cb_thresh  => $threshold,
        cb_cool    => $cooldown,
        cb_state   => [ map { { fails => 0, dead_until => 0 } } @conns ],
    }, $class;
}

# Pick the connection with the fewest in-flight queries; ties broken by
# round-robin. With circuit_threshold > 0, dead members are skipped
# (unless all are dead - then the breaker is bypassed). Hot path - the
# implementation lives in XS for ~5x lower per-pick cost.
sub _pick { EV::ClickHouse::_pool_pick($_[0]) }

# Find the {fails,dead_until} slot for a given $ch (object identity match).
sub _slot_for {
    my ($self, $ch) = @_;
    my $r = refaddr $ch;
    for my $i (0 .. $#{ $self->{conns} }) {
        return $self->{cb_state}[$i] if refaddr($self->{conns}[$i]) == $r;
    }
    return undef;
}

# Wrap the user callback so the circuit breaker can observe success/failure.
# The user's $cb is the LAST argument of query/insert/ping (per the public
# API). The slot update itself is in XS (_breaker_observe) so the wrapper
# closure body is one XSUB call rather than a handful of Perl ops.
sub _cb_observer {
    my ($self, $ch, $user_cb, $observe_failures) = @_;
    $observe_failures //= 1;
    return $user_cb unless $self->{cb_thresh} && ref($user_cb) eq 'CODE';
    my $slot = ref($ch) ? $self->_slot_for($ch) : $self->{cb_state}[$ch];
    return $user_cb unless $slot;
    my $thresh = $self->{cb_thresh};
    my $cool   = $self->{cb_cool};
    sub {

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

# ($conn, $idx) per call. Useful for warm-up (preload dictionaries,
# set session-level variables, dispatch a probe per member). The
# callback is invoked synchronously in pool order; if it throws,
# subsequent members are still visited (errors silently swallowed,
# matching the broadcast cancel/skip_pending/reset convention).
sub with_each {
    my ($self, $cb) = @_;
    die "Usage: \$pool->with_each(\$cb)" unless ref($cb) eq 'CODE';
    my @c = @{ $self->{conns} };
    for my $i (0 .. $#c) { eval { $cb->($c[$i], $i) } }
    return;
}

# Broadcast the same SELECT to every member and collect per-member
# results. Useful for `system.replicas`-style diagnostics where each
# shard needs to be queried directly. Callback fires once with an
# arrayref of { member => $i, rows => [...], err => $msg }, ordered
# by member index. Per-query settings are honoured. Dead members are
# included in the result with a "circuit open" error string rather
# than dispatched â€” the breaker would refuse them anyway.
sub fan_out {
    my ($self, $sql, @rest) = @_;
    my $cb       = pop @rest;
    my %opts     = @rest;
    my $settings = delete $opts{settings};
    die "fan_out: callback required" unless ref($cb) eq 'CODE';
    my @c   = @{ $self->{conns} };
    die "fan_out: no members" unless @c;
    my @out = map { { member => $_, rows => undef, err => undef } } 0 .. $#c;
    my $left = scalar @c;
    my $deliver = sub { $cb->(\@out) unless --$left };
    my $now     = EV::time();
    for my $i (0 .. $#c) {
        my $ch = $c[$i];
        # Short-circuit dead members so a long cooldown doesn't stall fan_out.
        if ($self->{cb_thresh}
            && $self->{cb_state}[$i]{dead_until} > $now) {
            $out[$i]{err} = "fan_out: member $i circuit open";
            $deliver->();
            next;
        }
        my $obs = $self->_cb_observer($i, sub {
            ($out[$i]{rows}, $out[$i]{err}) = @_;
            $deliver->();
        });
        # Wrap each member's dispatch so a synchronous croak (e.g.
        # "not connected" before auto_reconnect catches up) doesn't
        # strand the rest of the callbacks waiting on $left.
        eval {
            if ($settings) { $ch->query($sql, $settings, $obs) }
            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;
    my $released = 0;
    my $release = sub {
        return if $released++;
        if (--$self->{_pinned}{$r} <= 0) { delete $self->{_pinned}{$r} }
    };
    eval { $cb->($ch, $release); 1 } or do {
        my $err = $@;
        $release->();
        die $err;
    };
    return;
}

# Drain when ALL connections have completed pending work.
sub drain {
    my ($self, $cb) = @_;
    my $left = scalar @{ $self->{conns} };
    my $err;
    for my $c (@{ $self->{conns} }) {
        $c->drain(sub {
            $err //= $_[0] if $_[0];
            $cb->($err) unless --$left;
        });
    }
}

sub finish { $_->finish for @{ $_[0]{conns} } }

# Coordinated graceful shutdown: drain every member, then finish. If
# the optional $grace_seconds elapses before all members drain, force
# finish and report a timeout in the callback. Callback receives undef
# on clean shutdown, an error string on per-member drain error or
# timeout. $cb is optional.
#
#   $pool->shutdown(10, sub {
#       my ($err) = @_;
#       warn "shutdown: $err" if $err;
#       EV::break;
#   });
sub shutdown {
    my ($self, $grace, $cb) = @_;
    # Two-arg form: $pool->shutdown($cb). Treat the coderef as the cb
    # with no grace timer rather than silently dropping it.
    ($grace, $cb) = (undef, $grace) if ref($grace) eq 'CODE';
    $cb //= sub { };
    die "Usage: \$pool->shutdown([\$grace_seconds], \$cb)" unless ref($cb) eq 'CODE';
    my $left  = scalar @{ $self->{conns} };
    my $err;
    my $timer;
    my $fired = 0;

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

C<query_duration_p($p)> returns the C<$p>-quantile in seconds (C<$p>
in C<[0,1]>); C<query_duration_count> returns the number of samples
currently buffered.

=head2 pending_queries

    for my $q (@{ $ch->pending_queries }) {
        printf "%s %s age=%.3fs\n",
               $q->{state}, $q->{query_id} // '-', $q->{age};
    }

Snapshot of pending queries: returns arrayref of hashrefs. The head
of the in-flight queue (if any) appears first with
C<state =E<gt> 'in_flight'>, C<query_id =E<gt> last_query_id>, and
C<age =E<gt> seconds since dispatch>. Queued entries follow with
C<state =E<gt> 'queued'> and C<age =E<gt> 0> (they have no dispatch
time yet). SQL/settings are not retained after enqueue and so are
not included.

=head2 dump_state

    my $h = $ch->dump_state;
    # { connected, connecting, dns_pending, pending_count,
    #   callback_depth, send_len/pos/cap, recv_len/cap, fd,
    #   protocol, server_revision, reconnect_attempts,
    #   host, port, send_count, compress, tls }

Read-only diagnostic snapshot of internal struct state. Intended for
debugging stuck connections; field set may shift between releases
(don't script against it in production).

=head2 for_json_paths

    $ch->for_json_paths('events', 'payload', sub {
        my ($paths, $err) = @_;
        for my $p (@$paths) { say "$p->{path} : $p->{type}" }
    });

Discovers the dynamic JSON path layout of a C<JSON>/C<Object('json')>
column. Internally walks the C<Map(String, String)> returned by
C<JSONAllPathsWithTypes(col)> with a single C<arrayJoin(mapKeys(m))>
(the map alias is preserved so each path's type is correlated via a
second lookup), dedupes, sorts by path, and returns
C<[ { path =E<gt> 'a.b.c', type =E<gt> 'Int64' }, ... ]>. Useful for
monitoring schema drift on weakly-typed columns.

=head1 EV::ClickHouse::Pool

    my $pool = EV::ClickHouse::Pool->new(
        host => 'ch', port => 9000, protocol => 'native',
        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;
per-member errors are silently discarded (the surviving members still
receive the call). Iterate C<conns> yourself if you need per-member
error handling.

C<<< $pool->with_each(sub { my ($conn, $idx) = @_; ... }) >>> calls
C<$cb> once per member, passing the connection object and its index.
Each per-member call is wrapped in C<eval> so a single croak does not
abort the iteration; per-member errors are silently discarded - wrap
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 {
            $ch->query("insert into t values (1),(2),(3)", sub {
                $ch->query("select sum(n) from t", sub {
                    my ($rows) = @_;
                    say $rows->[0][0];
                    $release->();
                });
            });
        });
    });

C<<< $pool->query_to($idx, $sql, $cb) >>> /
C<<< $pool->insert_to($idx, $table, $data, $cb) >>> force-routes a
call to a specific member without going through C<_pick>. Circuit
breaker observation still applies (success/failure is recorded
against that member). Useful for replica-targeted DDL, S3 ingest
that has to land on a chosen node, or sticky-affinity reads.

C<<< $pool->nominate($idx) >>> returns the underlying connection so
subsequent calls bypass the pool entirely. Use sparingly - calls
made directly on the nominated connection don't update the
circuit-breaker state.

C<<< $pool->hedged_query($sql, hedge =E<gt> 2, $cb) >>> dispatches
the same select to C<hedge> distinct random members and resolves
with whichever returns first. The callback receives
C<($rows, undef, $member_idx)> on success (so callers can attribute
wins per member) or C<(undef, $err)> if I<every> member fails.
Extra completions after the winner are silently discarded.
Recommended for tail-latency-sensitive selects on replicated tables.
B<Do not> use for insert - would silently double-write when the
server's dedupe window misses.

C<<< $pool->fan_out($sql, $cb) >>> sends the same select to I<every>
member and collects per-member results into one arrayref:

    $pool->fan_out("select hostName(), uptime()", sub {
        for my $r (@{ $_[0] }) {
            printf "[%d] err=%s rows=%s\n",
                   $r->{member}, $r->{err} // '-',
                   $r->{rows} ? scalar @{$r->{rows}} : '-';
        }
    });

Useful for shard-aware diagnostics (per-replica lag, distinct
C<system.*> values across the pool). Errors are per-member, not
aggregated - the callback always fires with a complete list. Pass
C<settings =E<gt> \%h> for per-query options.

B<Circuit breaker:> pass C<circuit_threshold =E<gt> N> at construction
to enable per-member fail-fast. After N consecutive query/insert/ping

( run in 0.753 second using v1.01-cache-2.11-cpan-600a1bdf6e4 )