EV-ClickHouse
view release on metacpan or search on metacpan
ClickHouse.xs view on Meta::CPAN
#define SERVER_EXTREMES 8
#define SERVER_LOG 10
#define SERVER_TABLE_COLUMNS 11
#define SERVER_PROFILE_EVENTS 14
#define SERVER_TIMEZONE_UPDATE 17
/* Query kind */
#define QUERY_INITIAL 1
/* Query stage */
#define STAGE_COMPLETE 2
/* Native protocol states */
#define NATIVE_IDLE 0
#define NATIVE_WAIT_HELLO 1
#define NATIVE_WAIT_RESULT 2
#define NATIVE_WAIT_INSERT_META 3
/* Decode flags for column value formatting (opt-in) */
#define DECODE_DT_STR (1 << 0) /* Date/DateTime/DateTime64 → string */
#define DECODE_DEC_SCALE (1 << 1) /* Decimal → scaled NV */
#define DECODE_ENUM_STR (1 << 2) /* Enum → string label */
#define DECODE_NAMED_ROWS (1 << 3) /* results as arrayref of hashrefs */
struct ev_clickhouse_s {
unsigned int magic;
struct ev_loop *loop;
int fd;
ev_io rio, wio;
ev_timer timer;
int reading, writing, timing;
int connected, connecting;
int dns_pending; /* set while EV::cares is resolving the host;
query()/insert()/ping() queue against this
state so calls between new() and connect
don't croak with "not connected" */
unsigned int connect_gen; /* bumped by every start_connect; lets
fail_connection notice when the user
started a new connect from on_error */
int protocol; /* PROTO_HTTP or PROTO_NATIVE */
#ifdef HAVE_OPENSSL
SSL_CTX *ssl_ctx;
SSL *ssl;
#endif
int tls_enabled;
char *tls_ca_file;
char *tls_cert_file; /* client certificate (mutual TLS) */
char *tls_key_file; /* client private key (mutual TLS) */
/* connection params */
char *host, *user, *password, *database;
unsigned int port;
/* send/recv buffers */
char *send_buf;
size_t send_len, send_pos, send_cap;
char *recv_buf;
size_t recv_len, recv_cap;
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 */
ClickHouse.xs view on Meta::CPAN
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. */
ClickHouse.xs view on Meta::CPAN
* 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. */
( run in 0.699 second using v1.01-cache-2.11-cpan-b16cb0d3907 )