EV-ClickHouse
view release on metacpan or search on metacpan
ClickHouse.xs view on Meta::CPAN
char *http_decoded; /* chunked body accumulated so far (NULL = none) */
size_t http_decoded_len;
size_t http_decoded_cap;
size_t http_chunk_off; /* offset into recv_buf of next unparsed chunk hdr */
int http_chunk_active; /* 1 = a partially decoded chunked body is held */
/* native protocol state */
char *server_name;
char *server_display_name;
char *server_timezone;
unsigned int server_version_major, server_version_minor, server_revision;
unsigned int server_version_patch;
int native_state; /* NATIVE_IDLE, NATIVE_WAIT_HELLO, NATIVE_WAIT_RESULT, ... */
AV *native_rows; /* accumulate rows across Data blocks */
char *insert_data; /* pending TabSeparated data for two-phase INSERT */
size_t insert_data_len;
SV *insert_av; /* pending AV* of AV*s for arrayref INSERT */
char *insert_err; /* deferred error from unsupported INSERT encoding */
/* queues */
ngx_queue_t cb_queue;
ngx_queue_t send_queue;
int pending_count;
int send_count;
/* options */
char *session_id;
char *query_log_comment; /* prepended as a SQL block comment per query */
int compress;
double connect_timeout;
HV *default_settings; /* connection-level ClickHouse settings */
SV *on_connect;
SV *on_error;
SV *on_progress;
SV *on_disconnect;
SV *on_query_complete; /* fires after each query (success or error) */
SV *on_query_start; /* fires when a query is dispatched */
SV *on_log; /* native SERVER_LOG packets */
SV *on_failover; /* multi-host: ($oh, $op, $nh, $np, $msg) */
char *last_tls_error; /* OpenSSL error from last failed handshake */
char **failover_hosts; /* parallel arrays: hosts + ports. */
unsigned int *failover_ports; /* NULL if multi-host failover disabled. */
int failover_n;
int failover_idx;
unsigned int failover_default_port;
double query_start_time; /* ev_now() captured in pipeline_advance */
int tls_skip_verify;
double query_timeout;
size_t max_query_size; /* 0 = unlimited; client-side croak guard */
size_t max_recv_buffer; /* 0 = unlimited; defensive recv ceiling */
int http_basic_auth; /* 0=X-ClickHouse-{User,Key} (default);
* 1=Authorization: Basic ... (for proxies) */
int auto_reconnect;
uint32_t decode_flags;
AV *native_col_names; /* column names from last native result */
AV *native_col_types; /* column type strings from last native result */
SV *on_drain; /* callback fired when pending_count drops to 0 */
char *last_query_id; /* query_id of the last dispatched query */
SV *on_trace; /* debug trace callback */
ev_timer ka_timer; /* keepalive timer */
double keepalive; /* keepalive interval (0 = disabled) */
int ka_timing;
int callback_depth;
/* error info from last SERVER_EXCEPTION or HTTP error */
int32_t last_error_code;
/* profile info from last SERVER_PROFILE_INFO */
uint64_t profile_rows;
uint64_t profile_bytes;
uint64_t profile_rows_before_limit;
/* totals / extremes from last native query */
AV *native_totals;
AV *native_extremes;
/* reconnect backoff */
double reconnect_delay;
double reconnect_max_delay;
double reconnect_jitter; /* fractional [0, 1+]: actual delay
picks uniformly in [d, d*(1+jitter)] */
int reconnect_attempts;
int reconnect_max_attempts; /* 0 = unlimited */
int pending_addendum_finish; /* set when addendum partially written;
* io_cb completes finish_connect after drain */
ev_timer reconnect_timer;
int reconnect_timing;
/* on_progress throttling (0 = fire every packet) */
double progress_period;
double progress_last; /* ev_now() of last on_progress dispatch */
uint64_t progress_acc[5]; /* coalesced totals since last dispatch */
/* LowCardinality cross-block dictionary state */
SV ***lc_dicts; /* array of dictionaries, one per column */
uint64_t *lc_dict_sizes; /* size of each dictionary */
int lc_num_cols; /* number of columns with LC state */
};
struct ev_ch_cb_s {
SV *cb;
int raw; /* return raw response body instead of parsed rows */
SV *on_data; /* per-query streaming callback (fires per block) */
SV *on_complete;/* per-query on_query_complete override (or NULL) */
double query_timeout; /* per-query timeout (0=use default) */
ngx_queue_t queue;
};
struct ev_ch_send_s {
char *data; /* full HTTP request or native packet */
size_t data_len;
SV *cb;
char *insert_data; /* deferred TSV data for native INSERT */
size_t insert_data_len;
SV *insert_av; /* deferred AV* data for native INSERT */
int raw; /* return raw response body */
SV *on_data; /* per-query streaming callback */
SV *on_complete; /* per-query on_query_complete override */
double query_timeout; /* per-query timeout */
char *query_id; /* query_id for tracking */
ngx_queue_t queue;
};
/* Forward declarations for helpers defined further down (or in xs/io.c)
* but called from earlier code in this file or from xs/*.c included
* before the definition site. */
static void timer_cb(EV_P_ ev_timer *w, int revents);
static void stop_keepalive(ev_clickhouse_t *self);
static void schedule_reconnect(ev_clickhouse_t *self);
static void lc_free_dicts(ev_clickhouse_t *self);
static void start_reading(ev_clickhouse_t *self);
static void stop_reading(ev_clickhouse_t *self);
static void start_writing(ev_clickhouse_t *self);
static void stop_writing(ev_clickhouse_t *self);
static void emit_error(ev_clickhouse_t *self, const char *msg);
static void emit_trace(ev_clickhouse_t *self, const char *fmt, ...);
static int cleanup_connection(ev_clickhouse_t *self);
static int cancel_pending(ev_clickhouse_t *self, const char *errmsg);
static int check_destroyed(ev_clickhouse_t *self);
static char *safe_strdup(const char *s);
static void failover_free(ev_clickhouse_t *self);
static int finish_connect(ev_clickhouse_t *self);
static int try_write(ev_clickhouse_t *self);
static int pipeline_advance(ev_clickhouse_t *self);
/* Two helpers shared between the PL_dirty and normal arms of DESTROY:
* adding a new on_* slot or persistent string requires touching only
* one place. on_failover is omitted from CONNECTION_HANDLERS because
* failover_free() handles it together with the host ring. */
#define CLEAR_CONNECTION_HANDLERS(self) do { \
CLEAR_SV((self)->on_connect); \
CLEAR_SV((self)->on_error); \
CLEAR_SV((self)->on_progress); \
CLEAR_SV((self)->on_disconnect); \
CLEAR_SV((self)->on_query_complete); \
CLEAR_SV((self)->on_query_start); \
CLEAR_SV((self)->on_log); \
CLEAR_SV((self)->on_drain); \
CLEAR_SV((self)->on_trace); \
} while (0)
#define CLEAR_PERSISTENT_STATE(self) do { \
CLEAR_STR((self)->last_tls_error); \
CLEAR_STR((self)->last_query_id); \
CLEAR_STR((self)->host); \
CLEAR_STR((self)->user); \
CLEAR_STR((self)->password); \
CLEAR_STR((self)->database); \
CLEAR_STR((self)->session_id); \
CLEAR_STR((self)->query_log_comment); \
CLEAR_STR((self)->tls_ca_file); \
CLEAR_STR((self)->tls_cert_file); \
CLEAR_STR((self)->tls_key_file); \
CLEAR_STR((self)->server_name); \
CLEAR_STR((self)->server_display_name); \
CLEAR_STR((self)->server_timezone); \
CLEAR_STR((self)->insert_err); \
CLEAR_STR((self)->recv_buf); \
CLEAR_STR((self)->http_decoded); \
CLEAR_STR((self)->send_buf); \
CLEAR_SV((self)->native_rows); \
CLEAR_SV((self)->native_col_names); \
CLEAR_SV((self)->native_col_types); \
CLEAR_SV((self)->native_totals); \
CLEAR_SV((self)->native_extremes); \
CLEAR_SV((self)->default_settings); \
} while (0)
#include "xs/macros.h"
#include "xs/queues.c"
/* --- watcher helpers --- */
static void start_reading(ev_clickhouse_t *self) {
if (!self->reading && self->fd >= 0) {
ev_io_start(self->loop, &self->rio);
self->reading = 1;
}
}
static void stop_reading(ev_clickhouse_t *self) {
if (self->reading) {
ev_io_stop(self->loop, &self->rio);
self->reading = 0;
}
}
static void start_writing(ev_clickhouse_t *self) {
if (!self->writing && self->fd >= 0) {
ev_io_start(self->loop, &self->wio);
self->writing = 1;
}
}
static void stop_writing(ev_clickhouse_t *self) {
if (self->writing) {
ev_io_stop(self->loop, &self->wio);
self->writing = 0;
}
}
static void stop_timing(ev_clickhouse_t *self) {
if (self->timing) {
ev_timer_stop(self->loop, &self->timer);
self->timing = 0;
}
}
static int check_destroyed(ev_clickhouse_t *self) {
if (self->magic == EV_CH_FREED && self->callback_depth == 0) {
Safefree(self);
return 1;
}
return 0;
}
/* Free the per-connection failover host list (allocated by setter). */
/* Free just the host-list arrays. Called from _set_failover before
* re-populating + from failover_free below. Keeps on_failover alive. */
static void failover_free_hosts(ev_clickhouse_t *self) {
if (self->failover_hosts) {
for (int i = 0; i < self->failover_n; i++)
if (self->failover_hosts[i]) Safefree(self->failover_hosts[i]);
Safefree(self->failover_hosts);
self->failover_hosts = NULL;
}
if (self->failover_ports) {
Safefree(self->failover_ports);
self->failover_ports = NULL;
}
self->failover_n = 0;
self->failover_idx = 0;
}
/* Full failover-state teardown including the on_failover SV. DESTROY only. */
static void failover_free(ev_clickhouse_t *self) {
failover_free_hosts(self);
if (self->on_failover) {
SvREFCNT_dec(self->on_failover);
self->on_failover = NULL;
}
}
/* Word-boundary case-insensitive match against the failover-trigger
* keyword set. Used by emit_error to decide whether to rotate the
* multi-host ring. Returns 1 on match, 0 otherwise. */
static int failover_msg_match(const char *msg) {
static const char *KEYWORDS[] = {
"connect", "refused", "timeout", "unreachable", "route",
"reset", "closed", "broken", "down", "dns", "resolution",
"getaddrinfo", NULL,
};
if (!msg) return 0;
for (const char **k = KEYWORDS; *k; k++) {
size_t kl = strlen(*k);
const char *p = msg;
for (;;) {
const char *m = NULL;
for (const char *q = p; *q; q++) {
if (strncasecmp(q, *k, kl) == 0) { m = q; break; }
}
if (!m) break;
int boundary_l = (m == msg) || !isalnum((unsigned char)m[-1]);
int boundary_r = !isalnum((unsigned char)m[kl]);
if (boundary_l && boundary_r) return 1;
p = m + 1;
}
}
return 0;
}
/* Advance failover ring + fire on_failover callback (if set). Caller is
* responsible for callback_depth bookkeeping. */
static void failover_advance(ev_clickhouse_t *self, const char *msg) {
if (!self->failover_hosts || self->failover_n <= 0) return;
if (!failover_msg_match(msg)) return;
char *old_host = self->host ? safe_strdup(self->host) : NULL;
unsigned int old_port = self->port;
self->failover_idx = (self->failover_idx + 1) % self->failover_n;
ClickHouse.xs view on Meta::CPAN
}
static void invoke_cb(SV *cb) {
call_sv(cb, G_DISCARD | G_EVAL);
WARN_AND_CLEAR_ERRSV("callback");
SvREFCNT_dec(cb);
}
/* Invoke `cb` with (undef, errmsg). Caller manages callback_depth. */
static void invoke_err_cb(SV *cb, const char *errmsg) {
dSP;
ENTER;
SAVETMPS;
PUSHMARK(SP);
PUSHs(&PL_sv_undef);
PUSHs(sv_2mortal(newSVpv(errmsg, 0)));
PUTBACK;
invoke_cb(cb);
FREETMPS;
LEAVE;
}
/* Fire on_query_complete with (query_id, rows, bytes, error_code, duration_s, errmsg).
* Caller must guard callback_depth around any invoke that follows. Safe to
* call when on_query_complete is unset. `override` (when non-NULL) is the
* per-query override hook from the settings hashref; it REPLACES the
* connection-level handler for this query so per-query instrumentation
* doesn't double-count against global metrics. */
static void fire_on_query_complete_ex(ev_clickhouse_t *self, const char *errmsg,
SV *override) {
SV *target = override ? override : self->on_query_complete;
if (!target) return;
double dur = self->query_start_time > 0
? ev_now(self->loop) - self->query_start_time : 0.0;
self->callback_depth++;
{
dSP;
ENTER; SAVETMPS;
PUSHMARK(SP);
EXTEND(SP, 6);
PUSHs(self->last_query_id
? sv_2mortal(newSVpv(self->last_query_id, 0)) : &PL_sv_undef);
PUSHs(sv_2mortal(newSVuv(self->profile_rows)));
PUSHs(sv_2mortal(newSVuv(self->profile_bytes)));
PUSHs(sv_2mortal(newSViv(self->last_error_code)));
PUSHs(sv_2mortal(newSVnv(dur)));
PUSHs(errmsg ? sv_2mortal(newSVpv(errmsg, 0)) : &PL_sv_undef);
PUTBACK;
PINNED_CALL_SV(target, G_DISCARD | G_EVAL);
WARN_AND_CLEAR_ERRSV("on_query_complete");
FREETMPS; LEAVE;
}
self->callback_depth--;
/* Reset so a subsequent fire for a never-dispatched cancelled
* query (e.g. from cancel_pending draining send_queue) sees
* query_start_time == 0 and reports dur = 0.0, not a stale
* duration carried over from the previous in-flight query. */
self->query_start_time = 0;
}
/* IS_KEEPALIVE_CB is defined in xs/macros.h; keepalive_noop_cb is the
* sentinel that backs it (declared below this comment). */
static int deliver_error(ev_clickhouse_t *self, const char *errmsg) {
SV *oqc = NULL;
SV *cb = pop_cb_ex(self, &oqc);
if (cb == NULL) {
fire_on_query_complete_ex(self, errmsg, oqc);
if (oqc) SvREFCNT_dec(oqc);
return check_destroyed(self);
}
self->callback_depth++;
if (!IS_KEEPALIVE_CB(cb)) fire_on_query_complete_ex(self, errmsg, oqc);
invoke_err_cb(cb, errmsg);
if (oqc) SvREFCNT_dec(oqc);
self->callback_depth--;
return check_destroyed(self);
}
/* Returns 1 if self was freed. */
static int deliver_rows(ev_clickhouse_t *self, AV *rows) {
SV *oqc = NULL;
SV *cb = pop_cb_ex(self, &oqc);
if (cb == NULL) {
if (rows) SvREFCNT_dec((SV*)rows);
fire_on_query_complete_ex(self, NULL, oqc);
if (oqc) SvREFCNT_dec(oqc);
return check_destroyed(self);
}
self->callback_depth++;
if (!IS_KEEPALIVE_CB(cb)) fire_on_query_complete_ex(self, NULL, oqc);
{
dSP;
ENTER;
SAVETMPS;
PUSHMARK(SP);
PUSHs(rows ? sv_2mortal(newRV_noinc((SV*)rows)) : &PL_sv_undef);
PUTBACK;
invoke_cb(cb);
FREETMPS;
LEAVE;
}
if (oqc) SvREFCNT_dec(oqc);
self->callback_depth--;
return check_destroyed(self);
}
/* Deliver raw response body as scalar string. Returns 1 if self was freed. */
static int deliver_raw_body(ev_clickhouse_t *self, const char *data, size_t len) {
SV *oqc = NULL;
SV *cb = pop_cb_ex(self, &oqc);
if (cb == NULL) {
fire_on_query_complete_ex(self, NULL, oqc);
if (oqc) SvREFCNT_dec(oqc);
return check_destroyed(self);
}
self->callback_depth++;
if (!IS_KEEPALIVE_CB(cb)) fire_on_query_complete_ex(self, NULL, oqc);
ClickHouse.xs view on Meta::CPAN
}
/* Write a SQL block comment "<slashstar> <cmt> <starslash> " to buf.
* Returns bytes written, or 0 if cmt is NULL. Both call sites reserve
* strlen(cmt) + 7 bytes for the comment plus the 7 framing bytes. */
static size_t qlc_emit_prefix(char *buf, const char *cmt) {
size_t cl, off = 0;
if (!cmt) return 0;
cl = strlen(cmt);
memcpy(buf + off, "/* ", 3); off += 3;
memcpy(buf + off, cmt, cl); off += cl;
memcpy(buf + off, " */ ", 4); off += 4;
return off;
}
static int has_http_unsafe_chars(const char *s) {
/* XS gives us NUL-terminated C strings, so only reject CR/LF. */
if (!s) return 0;
for (; *s; s++)
if (*s == '\r' || *s == '\n') return 1;
return 0;
}
/* Drop the first `n` bytes from recv_buf, shifting any remaining bytes left. */
static inline void recv_consume(struct ev_clickhouse_s *self, size_t n) {
/* Saturating: a user callback fired from inside the parser may
* call reset() which zeroes recv_len; an unguarded subtract would
* underflow size_t to a huge value and the next ch_read would write
* past the buffer. */
if (n >= self->recv_len) { self->recv_len = 0; return; }
memmove(self->recv_buf, self->recv_buf + n, self->recv_len - n);
self->recv_len -= n;
}
static void ensure_send_cap(struct ev_clickhouse_s *self, size_t need);
/* Replace send_buf content with `src` (heap-allocated, freed here). */
static inline void send_replace(struct ev_clickhouse_s *self, char *src, size_t len) {
ensure_send_cap(self, len);
Copy(src, self->send_buf, len, char);
self->send_len = len;
self->send_pos = 0;
Safefree(src);
}
static int is_ip_literal(const char *s) {
struct in_addr a4;
struct in6_addr a6;
return (inet_pton(AF_INET, s, &a4) == 1 ||
inet_pton(AF_INET6, s, &a6) == 1);
}
/* Tears down the socket + per-connection state, then fires on_disconnect.
* Returns 1 if on_disconnect freed self (caller must not touch self), else 0. */
static int cleanup_connection(ev_clickhouse_t *self) {
int was_connected = self->connected;
if (was_connected) emit_trace(self, "disconnect");
stop_reading(self);
stop_writing(self);
stop_keepalive(self);
stop_timing(self);
#ifdef HAVE_OPENSSL
if (self->ssl) {
SSL_shutdown(self->ssl);
SSL_free(self->ssl);
self->ssl = NULL;
}
if (self->ssl_ctx) {
SSL_CTX_free(self->ssl_ctx);
self->ssl_ctx = NULL;
}
#endif
if (self->fd >= 0) {
close(self->fd);
self->fd = -1;
}
self->connected = 0;
self->connecting = 0;
self->dns_pending = 0; /* finish/reset interrupts async DNS */
self->send_len = 0;
self->send_pos = 0;
self->recv_len = 0;
if (self->http_decoded) Safefree(self->http_decoded);
self->http_decoded = NULL;
self->http_decoded_len = 0;
self->http_decoded_cap = 0;
self->http_chunk_off = 0;
self->http_chunk_active = 0;
self->send_count = 0;
self->pending_addendum_finish = 0;
self->native_state = NATIVE_IDLE;
CLEAR_SV(self->native_rows);
CLEAR_SV(self->native_col_names);
CLEAR_SV(self->native_col_types);
CLEAR_SV(self->native_totals);
CLEAR_SV(self->native_extremes);
lc_free_dicts(self);
CLEAR_INSERT(self);
CLEAR_STR(self->insert_err);
/* Fire on_disconnect AFTER state is reset, so a handler that queues
* new queries or calls reconnect sees clean state. The handler can
* drop the last $ch ref (-> DESTROY); propagate that so callers
* don't touch a freed self. */
if (was_connected && self->on_disconnect)
return fire_zero_arg_cb(self, self->on_disconnect, "disconnect");
return 0;
}
/* Fire user error callback + on_query_complete (per-query override or
* connection-level — fire_on_query_complete_ex falls back). Honors the
* documented "fires after every query (success or error)" contract for
* cancelled queries. oqc is consumed (refcount-dec'd) here.
* HTTP keepalive PINGs are suppressed to match the success-path behavior:
* users instrumenting via on_query_complete shouldn't see spurious zero-
* row completions for pings they didn't initiate. */
static void fire_err_and_complete(ev_clickhouse_t *self, SV *cb,
const char *errmsg, SV *oqc) {
/* Fire on_query_complete BEFORE the user cb to match the order
* used by deliver_error / deliver_rows on the normal path —
* instrumentation observers expect the global hook to run first.
* Keepalive PINGs are suppressed to match the success-path
* behavior so observers don't see spurious zero-row completions. */
if (!IS_KEEPALIVE_CB(cb))
fire_on_query_complete_ex(self, errmsg, oqc);
if (oqc) SvREFCNT_dec(oqc);
/* A user error callback may drop the last ref to $ch (DESTROY runs
* deferred while callback_depth > 0). invoke_err_cb itself is safe
* either way — caller's outer magic check handles the next loop. */
invoke_err_cb(cb, errmsg);
}
/* Drain in-flight cb_queue, delivering errmsg to each callback and resetting
* send_count. Caller manages callback_depth. */
static void drain_cb_queue(ev_clickhouse_t *self, const char *errmsg) {
while (!ngx_queue_empty(&self->cb_queue)) {
SV *oqc = NULL;
SV *cb = pop_cb_ex(self, &oqc);
if (cb == NULL) break;
fire_err_and_complete(self, cb, errmsg, oqc);
if (self->magic != EV_CH_MAGIC) break;
}
self->send_count = 0;
}
/* Returns 1 if self was freed. */
static int cancel_pending(ev_clickhouse_t *self, const char *errmsg) {
self->callback_depth++;
while (!ngx_queue_empty(&self->send_queue)) {
ngx_queue_t *q = ngx_queue_head(&self->send_queue);
ev_ch_send_t *send = ngx_queue_data(q, ev_ch_send_t, queue);
SV *cb = send->cb;
SV *oqc = send->on_complete; /* transfer ownership for fire below */
send->on_complete = NULL;
ngx_queue_remove(q);
Safefree(send->data);
CLEAR_INSERT(send);
CLEAR_SV(send->on_data);
release_send(send);
self->pending_count--;
fire_err_and_complete(self, cb, errmsg, oqc);
if (self->magic != EV_CH_MAGIC) break;
}
drain_cb_queue(self, errmsg);
self->callback_depth--;
return check_destroyed(self);
}
/* --- I/O helpers (with optional TLS) --- */
static ssize_t ch_read(ev_clickhouse_t *self, void *buf, size_t len) {
#ifdef HAVE_OPENSSL
if (self->ssl) {
int ssl_len = (len > (size_t)INT_MAX) ? INT_MAX : (int)len;
int ret = SSL_read(self->ssl, buf, ssl_len);
if (ret <= 0) {
int err = SSL_get_error(self->ssl, ret);
if (err == SSL_ERROR_WANT_READ) {
errno = EAGAIN;
ClickHouse.xs view on Meta::CPAN
static void write_native_settings(native_buf_t *b, HV *defaults, HV *overrides) {
HE *entry;
if (overrides) {
hv_iterinit(overrides);
while ((entry = hv_iternext(overrides))) {
I32 klen;
STRLEN vlen;
char *key = hv_iterkey(entry, &klen);
char *val = SvPV(hv_iterval(overrides, entry), vlen);
if (is_client_only_key(key, klen)) continue;
/* param_* keys go in the parameters block, not settings */
if (klen > 6 && memcmp(key, "param_", 6) == 0) continue;
nbuf_varuint(b, (uint64_t)klen);
nbuf_append(b, key, (size_t)klen);
nbuf_u8(b, 0); /* is_important = 0 */
nbuf_varuint(b, (uint64_t)vlen);
nbuf_append(b, val, vlen);
}
}
if (defaults) {
hv_iterinit(defaults);
while ((entry = hv_iternext(defaults))) {
I32 klen;
STRLEN vlen;
char *key = hv_iterkey(entry, &klen);
char *val = SvPV(hv_iterval(defaults, entry), vlen);
if (overrides && hv_exists(overrides, key, klen))
continue;
if (is_client_only_key(key, klen)) continue;
if (klen > 6 && memcmp(key, "param_", 6) == 0) continue;
nbuf_varuint(b, (uint64_t)klen);
nbuf_append(b, key, (size_t)klen);
nbuf_u8(b, 0); /* is_important = 0 */
nbuf_varuint(b, (uint64_t)vlen);
nbuf_append(b, val, vlen);
}
}
}
/* Textually included so all helpers stay file-local statics and the
* compiler sees one translation unit. Order matters: each file may
* call into helpers defined above it, but not below. xs/io.c must
* come last because it's the only file that reaches into every other
* subsystem (TCP/TLS, HTTP, native, types). */
#include "xs/codecs.c"
#include "xs/proto_http.c"
#include "xs/proto_native_build.c"
#include "xs/types.c"
#include "xs/proto_native_parse.c"
#include "xs/io.c"
/* --- XS interface --- */
MODULE = EV::ClickHouse PACKAGE = EV::ClickHouse
BOOT:
{
I_EV_API("EV::ClickHouse");
ch_openssl_init();
/* Permanent no-op CV used for internal callbacks (HTTP keepalive ping). */
keepalive_noop_cb = newRV_inc((SV*)get_cv("EV::ClickHouse::__keepalive_noop", GV_ADD));
/* Per-process rand() seed so reconnect_jitter desynchronises forks
* (otherwise every worker generates the same sequence and the
* jitter is uniform across the herd, defeating its purpose). */
srand((unsigned)time(NULL) ^ (unsigned)getpid());
}
EV::ClickHouse
_new(char *class, EV::Loop loop)
CODE:
{
PERL_UNUSED_VAR(class);
Newxz(RETVAL, 1, ev_clickhouse_t);
RETVAL->magic = EV_CH_MAGIC;
RETVAL->loop = loop;
RETVAL->fd = -1;
RETVAL->protocol = PROTO_HTTP;
ngx_queue_init(&RETVAL->cb_queue);
ngx_queue_init(&RETVAL->send_queue);
Newx(RETVAL->recv_buf, RECV_BUF_INIT, char);
RETVAL->recv_cap = RECV_BUF_INIT;
Newx(RETVAL->send_buf, SEND_BUF_INIT, char);
RETVAL->send_cap = SEND_BUF_INIT;
ev_init(&RETVAL->timer, timer_cb);
RETVAL->timer.data = (void *)RETVAL;
}
OUTPUT:
RETVAL
void
DESTROY(EV::ClickHouse self)
CODE:
{
if (self->magic != EV_CH_MAGIC) return;
stop_reading(self);
stop_writing(self);
stop_timing(self);
stop_keepalive(self);
if (self->reconnect_timing) {
ev_timer_stop(self->loop, &self->reconnect_timer);
self->reconnect_timing = 0;
}
if (PL_dirty) {
self->magic = EV_CH_FREED;
while (!ngx_queue_empty(&self->send_queue)) {
ngx_queue_t *q = ngx_queue_head(&self->send_queue);
ev_ch_send_t *send = ngx_queue_data(q, ev_ch_send_t, queue);
ngx_queue_remove(q);
Safefree(send->data);
CLEAR_INSERT(send);
CLEAR_SV(send->on_data);
CLEAR_SV(send->on_complete);
SvREFCNT_dec(send->cb);
release_send(send);
}
while (!ngx_queue_empty(&self->cb_queue)) {
ngx_queue_t *q = ngx_queue_head(&self->cb_queue);
ev_ch_cb_t *cbt = ngx_queue_data(q, ev_ch_cb_t, queue);
ngx_queue_remove(q);
CLEAR_SV(cbt->on_data);
CLEAR_SV(cbt->on_complete);
SvREFCNT_dec(cbt->cb);
release_cbt(cbt);
}
#ifdef HAVE_OPENSSL
if (self->ssl) { SSL_free(self->ssl); self->ssl = NULL; }
if (self->ssl_ctx) { SSL_CTX_free(self->ssl_ctx); self->ssl_ctx = NULL; }
#endif
if (self->fd >= 0) close(self->fd);
CLEAR_CONNECTION_HANDLERS(self);
CLEAR_PERSISTENT_STATE(self);
failover_free(self);
CLEAR_INSERT(self);
lc_free_dicts(self);
Safefree(self);
return;
}
if (cancel_pending(self, "object destroyed"))
return; /* inner DESTROY already freed self */
/* A user callback fired from cancel_pending may have called
* $ch->reset() which re-arms watchers / opens a fresh fd.
* Stop everything again before tearing the struct down so the
* EV loop can't dispatch into freed memory. */
stop_reading(self);
stop_writing(self);
stop_timing(self);
stop_keepalive(self);
if (self->reconnect_timing) {
ev_timer_stop(self->loop, &self->reconnect_timer);
self->reconnect_timing = 0;
}
#ifdef HAVE_OPENSSL
if (self->ssl) {
SSL_shutdown(self->ssl);
SSL_free(self->ssl);
self->ssl = NULL;
}
if (self->ssl_ctx) {
SSL_CTX_free(self->ssl_ctx);
self->ssl_ctx = NULL;
}
#endif
if (self->fd >= 0) {
close(self->fd);
self->fd = -1;
}
self->loop = NULL;
self->connected = 0;
CLEAR_CONNECTION_HANDLERS(self);
CLEAR_PERSISTENT_STATE(self);
failover_free(self);
lc_free_dicts(self);
CLEAR_INSERT(self);
self->magic = EV_CH_FREED;
if (self->callback_depth == 0) {
Safefree(self);
}
/* else: check_destroyed() will Safefree when callback_depth reaches 0 */
}
void
_set_tls_ca_file(EV::ClickHouse self, const char *path)
CODE:
{
CLEAR_STR(self->tls_ca_file);
self->tls_ca_file = safe_strdup(path);
}
void
_set_tls_cert_file(EV::ClickHouse self, const char *path)
CODE:
{
CLEAR_STR(self->tls_cert_file);
self->tls_cert_file = safe_strdup(path);
}
void
_set_tls_key_file(EV::ClickHouse self, const char *path)
CODE:
{
CLEAR_STR(self->tls_key_file);
self->tls_key_file = safe_strdup(path);
ClickHouse.xs view on Meta::CPAN
OUTPUT:
RETVAL
SV *
last_totals(EV::ClickHouse self)
CODE:
{
RETVAL = self->native_totals ? newRV_inc((SV*)self->native_totals)
: &PL_sv_undef;
}
OUTPUT:
RETVAL
SV *
last_extremes(EV::ClickHouse self)
CODE:
{
RETVAL = self->native_extremes ? newRV_inc((SV*)self->native_extremes)
: &PL_sv_undef;
}
OUTPUT:
RETVAL
SV *
profile_rows_before_limit(EV::ClickHouse self)
CODE:
{
RETVAL = newSVuv(self->profile_rows_before_limit);
}
OUTPUT:
RETVAL
SV *
profile_rows(EV::ClickHouse self)
CODE:
{
RETVAL = newSVuv(self->profile_rows);
}
OUTPUT:
RETVAL
SV *
profile_bytes(EV::ClickHouse self)
CODE:
{
RETVAL = newSVuv(self->profile_bytes);
}
OUTPUT:
RETVAL
SV*
on_trace(EV::ClickHouse self, SV *handler = NULL)
CODE:
{
RETVAL = handler_accessor(&self->on_trace, handler, items > 1);
}
OUTPUT:
RETVAL
void
_set_keepalive(EV::ClickHouse self, double val)
CODE:
{
self->keepalive = val;
}
void
_set_reconnect_delay(EV::ClickHouse self, double val)
CODE:
{
self->reconnect_delay = val;
}
void
_set_reconnect_max_delay(EV::ClickHouse self, double val)
CODE:
{
self->reconnect_max_delay = val;
}
void
_set_reconnect_jitter(EV::ClickHouse self, double val)
CODE:
{
self->reconnect_jitter = val < 0 ? 0 : val;
}
void
_set_reconnect_max_attempts(EV::ClickHouse self, int val)
CODE:
{
self->reconnect_max_attempts = val;
}
void
_set_progress_period(EV::ClickHouse self, double val)
CODE:
{
self->progress_period = val;
}
void
drain(EV::ClickHouse self, SV *cb)
CODE:
{
if (!(SvROK(cb) && SvTYPE(SvRV(cb)) == SVt_PVCV))
croak("drain callback must be a CODE reference");
CLEAR_SV(self->on_drain);
if (self->pending_count == 0 && ngx_queue_empty(&self->send_queue)) {
/* Nothing pending — fire immediately */
(void)fire_zero_arg_cb(self, cb, "drain");
} else {
self->on_drain = SvREFCNT_inc(cb);
}
}
void
cancel(EV::ClickHouse self)
CODE:
{
if (self->protocol == PROTO_NATIVE && self->send_count > 0) {
/* Send CLIENT_CANCEL packet */
native_buf_t pkt;
nbuf_init(&pkt);
( run in 1.583 second using v1.01-cache-2.11-cpan-14f38c9f855 )