diff --git a/bdb/bdb.c b/bdb/bdb.c index 3f5d1595db..93f778a34f 100644 --- a/bdb/bdb.c +++ b/bdb/bdb.c @@ -705,15 +705,27 @@ void bdb_fingerprint_rtstats_set(const unsigned char *fingerprint, size_t fplen, bb_berkdb_fingerprint_rtstats_set(fingerprint, fplen, has_main_entry); } +void bdb_fingerprint_rtstats_set_write(const unsigned char *fingerprint, size_t fplen, int has_main_entry) +{ + bb_berkdb_fingerprint_rtstats_set_write(fingerprint, fplen, has_main_entry); +} + void bdb_fingerprint_rtstats_clear(void) { bb_berkdb_fingerprint_rtstats_clear(); } int bdb_fingerprint_rtstats_get(const unsigned char *fingerprint, size_t fplen, uint64_t *n_pagein_read, - uint64_t *n_pagein_read_io) + uint64_t *n_pagein_read_io, uint64_t *n_write_pagein_read, + uint64_t *n_write_pagein_read_io) +{ + return bb_berkdb_fingerprint_rtstats_get(fingerprint, fplen, n_pagein_read, n_pagein_read_io, n_write_pagein_read, + n_write_pagein_read_io); +} + +void bdb_fingerprint_rtstats_foreach(bdb_fingerprint_rtstats_enum_fn fn, void *arg) { - return bb_berkdb_fingerprint_rtstats_get(fingerprint, fplen, n_pagein_read, n_pagein_read_io); + bb_berkdb_fingerprint_rtstats_foreach((bb_berkdb_fingerprint_rtstats_enum_fn)fn, arg); } /* Call this any time to get process wide stats (which get updated locklessly) diff --git a/bdb/bdb_api.h b/bdb/bdb_api.h index cb4c191ed2..74920fd2fe 100644 --- a/bdb/bdb_api.h +++ b/bdb/bdb_api.h @@ -1373,9 +1373,15 @@ const struct berkdb_thread_stats *bdb_get_thread_stats(void); const struct berkdb_thread_stats *bdb_get_process_stats(void); void bdb_fingerprint_rtstats_set(const unsigned char *fingerprint, size_t fplen, int has_main_entry); +void bdb_fingerprint_rtstats_set_write(const unsigned char *fingerprint, size_t fplen, int has_main_entry); void bdb_fingerprint_rtstats_clear(void); int bdb_fingerprint_rtstats_get(const unsigned char *fingerprint, size_t fplen, uint64_t *n_pagein_read, - uint64_t *n_pagein_read_io); + uint64_t *n_pagein_read_io, uint64_t *n_write_pagein_read, + uint64_t *n_write_pagein_read_io); +/* counts[4] = {n_pagein_read, n_pagein_read_io, n_write_pagein_read, n_write_pagein_read_io} */ +typedef void (*bdb_fingerprint_rtstats_enum_fn)(const unsigned char *fingerprint, const uint64_t *counts, + int has_main_entry, void *arg); +void bdb_fingerprint_rtstats_foreach(bdb_fingerprint_rtstats_enum_fn fn, void *arg); /* Format and print the thread stats. printfn() is a function which accepts * a line to print (\n\0 terminated) and a context pointer. Its return value diff --git a/berkdb/build/db.h b/berkdb/build/db.h index b1321cd088..a81ec7c28a 100644 --- a/berkdb/build/db.h +++ b/berkdb/build/db.h @@ -3130,10 +3130,16 @@ void bb_berkdb_thread_stats_reset(void); void bb_berkdb_fingerprint_rtstats_init(void); void bb_berkdb_fingerprint_rtstats_set(const unsigned char *fingerprint, size_t fplen, int has_main_entry); +void bb_berkdb_fingerprint_rtstats_set_write(const unsigned char *fingerprint, size_t fplen, int has_main_entry); void bb_berkdb_fingerprint_rtstats_clear(void); void bb_berkdb_fingerprint_rtstats_bump_pagein(int did_io); int bb_berkdb_fingerprint_rtstats_get(const unsigned char *fingerprint, size_t fplen, - uint64_t *n_pagein_read, uint64_t *n_pagein_read_io); + uint64_t *n_pagein_read, uint64_t *n_pagein_read_io, + uint64_t *n_write_pagein_read, uint64_t *n_write_pagein_read_io); +/* counts[4] = {n_pagein_read, n_pagein_read_io, n_write_pagein_read, n_write_pagein_read_io} */ +typedef void (*bb_berkdb_fingerprint_rtstats_enum_fn)(const unsigned char *fingerprint, + const uint64_t *counts, int has_main_entry, void *arg); +void bb_berkdb_fingerprint_rtstats_foreach(bb_berkdb_fingerprint_rtstats_enum_fn fn, void *arg); extern int gbl_bb_berkdb_enable_thread_stats; extern int gbl_bb_berkdb_enable_lock_timing; diff --git a/berkdb/mp/mp_fingerprint_rtstats.c b/berkdb/mp/mp_fingerprint_rtstats.c index 07d60cc427..2aee3fa970 100644 --- a/berkdb/mp/mp_fingerprint_rtstats.c +++ b/berkdb/mp/mp_fingerprint_rtstats.c @@ -36,17 +36,23 @@ struct fingerprint_rtstats { unsigned char fingerprint[FP_RTSTATS_KEYSZ]; /* hash key, must be first */ uint64_t n_pagein_read; /* every successful __memp_fget_internal() * return attributed to this fingerprint - * (cache hit or miss) */ + * during SQL read execution (cache hit or + * miss) */ uint64_t n_pagein_read_io; /* subset of n_pagein_read where the fetch * required an actual disk read */ + uint64_t n_write_pagein_read; /* same as n_pagein_read, but for page-ins + * generated while the master applies a write + * schedule for this fingerprint */ + uint64_t n_write_pagein_read_io; /* subset of n_write_pagein_read that + * required an actual disk read */ int has_main_entry; /* 1 if gbl_fingerprint_hash already had an - * entry for this fingerprint when this - * stat entry was created (set once, at - * creation time, not refreshed). Lets a - * future write-side commit distinguish - * "this machine has full query info" from - * "we only know the fingerprint + counts". - * Unused by read-side logic. */ + * entry for this fingerprint when this stat + * entry was created (set once at creation, + * never refreshed). Diagnostic only: the + * comdb2_fingerprints systable derives its + * has_query_info column from live + * gbl_fingerprint_hash membership, not from + * this (possibly stale) flag. */ }; static hash_t *gbl_fingerprint_rtstats_hash; @@ -63,12 +69,21 @@ int gbl_fingerprint_rtstats_max_entries = 5000; static struct fingerprint_rtstats gbl_fingerprint_rtstats_nofingerprint; static pthread_key_t fingerprint_rtstats_key; +/* + * Per-thread mode flag paired with fingerprint_rtstats_key. NULL (the default) + * means the armed thread is doing SQL read execution; non-NULL means it is a + * master block-processor thread applying a write schedule. bump_pagein() reads + * it to decide which pair of counters to increment. Must be per-thread (not on + * the shared entry) because many threads share one fingerprint entry. + */ +static pthread_key_t fingerprint_rtstats_write_key; static int fingerprint_rtstats_inited = 0; void bb_berkdb_fingerprint_rtstats_init(void) { Pthread_key_create(&fingerprint_rtstats_key, NULL); + Pthread_key_create(&fingerprint_rtstats_write_key, NULL); Pthread_mutex_lock(&gbl_fingerprint_rtstats_hash_mu); if (gbl_fingerprint_rtstats_hash == NULL) @@ -79,18 +94,22 @@ bb_berkdb_fingerprint_rtstats_init(void) } /* - * Called once per statement, as soon as its fingerprint is known and - * before cursor traversal begins. Finds-or-creates the stat entry for - * this fingerprint and points the calling thread's TLS slot at it. + * Find-or-create the stat entry for this fingerprint and point the calling + * thread's TLS slot at it, recording whether the thread is in write-apply mode. + * Shared by the read-side set (is_write=0) and the master write-side set + * (is_write=1). */ -void -bb_berkdb_fingerprint_rtstats_set(const unsigned char *fingerprint, size_t fplen, int has_main_entry) +static void +fingerprint_rtstats_set_internal(const unsigned char *fingerprint, size_t fplen, + int has_main_entry, int is_write) { struct fingerprint_rtstats *t; if (!fingerprint_rtstats_inited || fingerprint == NULL || fplen != FP_RTSTATS_KEYSZ) return; + Pthread_setspecific(fingerprint_rtstats_write_key, is_write ? (void *)1 : NULL); + Pthread_mutex_lock(&gbl_fingerprint_rtstats_hash_mu); if (gbl_fingerprint_rtstats_hash == NULL) gbl_fingerprint_rtstats_hash = hash_init(FP_RTSTATS_KEYSZ); @@ -125,6 +144,29 @@ bb_berkdb_fingerprint_rtstats_set(const unsigned char *fingerprint, size_t fplen Pthread_setspecific(fingerprint_rtstats_key, t); } +/* + * Called once per statement, as soon as its fingerprint is known and + * before cursor traversal begins. Finds-or-creates the stat entry for + * this fingerprint and points the calling thread's TLS slot at it. + */ +void +bb_berkdb_fingerprint_rtstats_set(const unsigned char *fingerprint, size_t fplen, int has_main_entry) +{ + fingerprint_rtstats_set_internal(fingerprint, fplen, has_main_entry, 0); +} + +/* + * Master write-side arm. Called on the block-processor thread when it receives + * an OSQL_FINGERPRINT op, before it applies the write ops that follow. Page-ins + * generated until the next clear are attributed to this fingerprint's write + * counters. + */ +void +bb_berkdb_fingerprint_rtstats_set_write(const unsigned char *fingerprint, size_t fplen, int has_main_entry) +{ + fingerprint_rtstats_set_internal(fingerprint, fplen, has_main_entry, 1); +} + /* Called when the statement is done (or on error paths). */ void bb_berkdb_fingerprint_rtstats_clear(void) @@ -132,6 +174,7 @@ bb_berkdb_fingerprint_rtstats_clear(void) if (!fingerprint_rtstats_inited) return; Pthread_setspecific(fingerprint_rtstats_key, NULL); + Pthread_setspecific(fingerprint_rtstats_write_key, NULL); } /* @@ -145,15 +188,24 @@ void bb_berkdb_fingerprint_rtstats_bump_pagein(int did_io) { struct fingerprint_rtstats *t = NULL; + int is_write = 0; - if (fingerprint_rtstats_inited) + if (fingerprint_rtstats_inited) { t = pthread_getspecific(fingerprint_rtstats_key); + is_write = pthread_getspecific(fingerprint_rtstats_write_key) != NULL; + } if (t == NULL) t = &gbl_fingerprint_rtstats_nofingerprint; - ATOMIC_ADD64(t->n_pagein_read, 1); - if (did_io) - ATOMIC_ADD64(t->n_pagein_read_io, 1); + if (is_write) { + ATOMIC_ADD64(t->n_write_pagein_read, 1); + if (did_io) + ATOMIC_ADD64(t->n_write_pagein_read_io, 1); + } else { + ATOMIC_ADD64(t->n_pagein_read, 1); + if (did_io) + ATOMIC_ADD64(t->n_pagein_read_io, 1); + } } /* @@ -165,13 +217,16 @@ bb_berkdb_fingerprint_rtstats_bump_pagein(int did_io) */ int bb_berkdb_fingerprint_rtstats_get(const unsigned char *fingerprint, size_t fplen, - uint64_t *n_pagein_read, uint64_t *n_pagein_read_io) + uint64_t *n_pagein_read, uint64_t *n_pagein_read_io, + uint64_t *n_write_pagein_read, uint64_t *n_write_pagein_read_io) { struct fingerprint_rtstats *t; int found = 0; *n_pagein_read = 0; *n_pagein_read_io = 0; + *n_write_pagein_read = 0; + *n_write_pagein_read_io = 0; if (!fingerprint_rtstats_inited || fingerprint == NULL || fplen != FP_RTSTATS_KEYSZ) return 0; @@ -184,6 +239,8 @@ bb_berkdb_fingerprint_rtstats_get(const unsigned char *fingerprint, size_t fplen * ATOMIC_ADD64, so load them atomically too. */ *n_pagein_read = ATOMIC_LOAD64(t->n_pagein_read); *n_pagein_read_io = ATOMIC_LOAD64(t->n_pagein_read_io); + *n_write_pagein_read = ATOMIC_LOAD64(t->n_write_pagein_read); + *n_write_pagein_read_io = ATOMIC_LOAD64(t->n_write_pagein_read_io); found = 1; } } @@ -191,3 +248,37 @@ bb_berkdb_fingerprint_rtstats_get(const unsigned char *fingerprint, size_t fplen return found; } + +/* + * Enumerate every rtstats entry for reporting (the comdb2_fingerprints + * systable). Invokes fn once per entry under the hash mutex; the callback + * must not call back into this subsystem (it would deadlock) and must not + * block. Counters are passed as counts[4] = {n_pagein_read, n_pagein_read_io, + * n_write_pagein_read, n_write_pagein_read_io} so no struct type needs to be + * shared across the bdb/berkdb boundary. + */ +void +bb_berkdb_fingerprint_rtstats_foreach(bb_berkdb_fingerprint_rtstats_enum_fn fn, void *arg) +{ + struct fingerprint_rtstats *t; + void *hash_cur; + unsigned int hash_cur_buk; + + if (!fingerprint_rtstats_inited || fn == NULL) + return; + + Pthread_mutex_lock(&gbl_fingerprint_rtstats_hash_mu); + if (gbl_fingerprint_rtstats_hash != NULL) { + t = hash_first(gbl_fingerprint_rtstats_hash, &hash_cur, &hash_cur_buk); + while (t != NULL) { + uint64_t counts[4]; + counts[0] = ATOMIC_LOAD64(t->n_pagein_read); + counts[1] = ATOMIC_LOAD64(t->n_pagein_read_io); + counts[2] = ATOMIC_LOAD64(t->n_write_pagein_read); + counts[3] = ATOMIC_LOAD64(t->n_write_pagein_read_io); + fn(t->fingerprint, counts, t->has_main_entry, arg); + t = hash_next(gbl_fingerprint_rtstats_hash, &hash_cur, &hash_cur_buk); + } + } + Pthread_mutex_unlock(&gbl_fingerprint_rtstats_hash_mu); +} diff --git a/db/db_fingerprint.c b/db/db_fingerprint.c index d643470fe4..9690a4dd5a 100644 --- a/db/db_fingerprint.c +++ b/db/db_fingerprint.c @@ -78,6 +78,33 @@ int clear_fingerprints(int *plans_count) return count; } +/* + * Returns 1 if gbl_fingerprint_hash already has an entry for this fingerprint + * (i.e. this node has the full normalized-SQL / query info), 0 otherwise. Used + * by the master write-side arm to record has_main_entry. + */ +int fingerprint_has_main_entry(const unsigned char fingerprint[FINGERPRINTSZ]) +{ + int found = 0; + Pthread_mutex_lock(&gbl_fingerprint_hash_mu); + if (gbl_fingerprint_hash != NULL && hash_find(gbl_fingerprint_hash, fingerprint) != NULL) + found = 1; + Pthread_mutex_unlock(&gbl_fingerprint_hash_mu); + return found; +} + +/* + * Returns 1 if the fingerprint is all-zeros -- calc_fingerprint() produces this + * for a statement with no normalized SQL. There is nothing useful to attribute + * to such a fingerprint, so the write-side send paths skip it rather than pool + * unrelated traffic under a single all-zero key. + */ +int fingerprint_is_zero(const unsigned char fingerprint[FINGERPRINTSZ]) +{ + static const unsigned char zero[FINGERPRINTSZ] = {0}; + return memcmp(fingerprint, zero, FINGERPRINTSZ) == 0; +} + void calc_fingerprint(const char *zNormSql, size_t *pnNormSql, unsigned char fingerprint[FINGERPRINTSZ]) { memset(fingerprint, 0, FINGERPRINTSZ); diff --git a/db/db_tunables.c b/db/db_tunables.c index 920e84d15b..09819f0b42 100644 --- a/db/db_tunables.c +++ b/db/db_tunables.c @@ -308,6 +308,7 @@ extern uint32_t gbl_rand_elect_min_ms; extern int gbl_rand_elect_max_ms; extern int gbl_handle_buf_add_latency_ms; extern int gbl_osql_send_startgen; +extern int gbl_osql_send_fingerprint; extern int gbl_create_default_user; extern int gbl_allow_neg_column_size; extern int gbl_client_heartbeat_ms; diff --git a/db/db_tunables.h b/db/db_tunables.h index 571519df1a..55999a9d8f 100644 --- a/db/db_tunables.h +++ b/db/db_tunables.h @@ -1744,6 +1744,12 @@ REGISTER_TUNABLE("osql_send_startgen", TUNABLE_BOOLEAN, &gbl_osql_send_startgen, EXPERIMENTAL | INTERNAL, NULL, NULL, NULL, NULL); +REGISTER_TUNABLE("osql_send_fingerprint", + "Send SQL fingerprint in the osql stream so the master can " + "attribute write-apply page-in I/O to it. Keep off until the " + "whole cluster is upgraded. (Default: off)", + TUNABLE_BOOLEAN, &gbl_osql_send_fingerprint, EXPERIMENTAL | INTERNAL, NULL, NULL, NULL, NULL); + REGISTER_TUNABLE("client_heartbeat_ms", "Number of milliseconds between client api heartbeats. " "(Default: 100)", diff --git a/db/osqlblockproc.c b/db/osqlblockproc.c index 42b7d30383..5964662d4a 100644 --- a/db/osqlblockproc.c +++ b/db/osqlblockproc.c @@ -1209,6 +1209,11 @@ static int apply_changes(struct ireq *iq, blocksql_tran_t *tran, void *iq_tran, out_rc = process_this_session(iq, iq_tran, iq->sorese, &bdberr, nops, err, dbc, dbc_ins, func); + /* An OSQL_FINGERPRINT op may have armed this (pooled) block-processor + * thread's write-side fingerprint TLS while applying the session above; + * clear it so page-ins from unrelated later work aren't misattributed. */ + bdb_fingerprint_rtstats_clear(); + Pthread_mutex_unlock(&tran->store_mtx); /* close the cursor */ diff --git a/db/osqlcomm.c b/db/osqlcomm.c index ae40ac5955..ccc29af5fc 100644 --- a/db/osqlcomm.c +++ b/db/osqlcomm.c @@ -58,6 +58,7 @@ #include "sc_logic.h" #include "eventlog.h" #include +#include "fingerprint.h" #define MAX_CLUSTER REPMAX @@ -1715,6 +1716,83 @@ osqlcomm_usedb_rpl_uuid_type_get(osql_usedb_rpl_uuid_t *p_osql_usedb_uuid_rpl, return p_buf; } +typedef struct osql_fingerprint { + unsigned char fingerprint[FINGERPRINTSZ]; /* MD5 of normalized SQL */ +} osql_fingerprint_t; + +enum { OSQLCOMM_FINGERPRINT_TYPE_LEN = FINGERPRINTSZ }; + +BB_COMPILE_TIME_ASSERT(osqlcomm_fingerprint_type_len, sizeof(osql_fingerprint_t) == OSQLCOMM_FINGERPRINT_TYPE_LEN); + +static uint8_t *osqlcomm_fingerprint_type_put(const osql_fingerprint_t *p_osql_fingerprint, uint8_t *p_buf, + const uint8_t *p_buf_end) +{ + if (p_buf_end < p_buf || OSQLCOMM_FINGERPRINT_TYPE_LEN > p_buf_end - p_buf) + return NULL; + + /* raw opaque bytes, no byte-swap (like a uuid) */ + p_buf = + buf_no_net_put(&(p_osql_fingerprint->fingerprint), sizeof(p_osql_fingerprint->fingerprint), p_buf, p_buf_end); + + return p_buf; +} + +static const uint8_t *osqlcomm_fingerprint_type_get(osql_fingerprint_t *p_osql_fingerprint, const uint8_t *p_buf, + const uint8_t *p_buf_end) +{ + if (p_buf_end < p_buf || OSQLCOMM_FINGERPRINT_TYPE_LEN > p_buf_end - p_buf) + return NULL; + + p_buf = + buf_no_net_get(&(p_osql_fingerprint->fingerprint), sizeof(p_osql_fingerprint->fingerprint), p_buf, p_buf_end); + + return p_buf; +} + +typedef struct osql_fingerprint_rpl { + osql_rpl_t hd; + osql_fingerprint_t dt; +} osql_fingerprint_rpl_t; + +enum { OSQLCOMM_FINGERPRINT_RPL_TYPE_LEN = OSQLCOMM_RPL_TYPE_LEN + OSQLCOMM_FINGERPRINT_TYPE_LEN }; + +BB_COMPILE_TIME_ASSERT(osqlcomm_fingerprint_rpl_type_len, + sizeof(osql_fingerprint_rpl_t) == OSQLCOMM_FINGERPRINT_RPL_TYPE_LEN); + +static uint8_t *osqlcomm_fingerprint_rpl_type_put(const osql_fingerprint_rpl_t *p_fingerprint_rpl, uint8_t *p_buf, + uint8_t *p_buf_end) +{ + if (p_buf_end < p_buf || OSQLCOMM_FINGERPRINT_RPL_TYPE_LEN > (p_buf_end - p_buf)) + return NULL; + + p_buf = osqlcomm_rpl_type_put(&(p_fingerprint_rpl->hd), p_buf, p_buf_end); + p_buf = osqlcomm_fingerprint_type_put(&(p_fingerprint_rpl->dt), p_buf, p_buf_end); + + return p_buf; +} + +typedef struct osql_fingerprint_rpl_uuid { + osql_uuid_rpl_t hd; + osql_fingerprint_t dt; +} osql_fingerprint_rpl_uuid_t; + +enum { OSQLCOMM_FINGERPRINT_RPL_UUID_TYPE_LEN = OSQLCOMM_UUID_RPL_TYPE_LEN + OSQLCOMM_FINGERPRINT_TYPE_LEN }; + +BB_COMPILE_TIME_ASSERT(osqlcomm_fingerprint_rpl_uuid_type_len, + sizeof(osql_fingerprint_rpl_uuid_t) == OSQLCOMM_FINGERPRINT_RPL_UUID_TYPE_LEN); + +static uint8_t *osqlcomm_fingerprint_uuid_rpl_type_put(const osql_fingerprint_rpl_uuid_t *p_fingerprint_uuid_rpl, + uint8_t *p_buf, uint8_t *p_buf_end) +{ + if (p_buf_end < p_buf || OSQLCOMM_FINGERPRINT_RPL_UUID_TYPE_LEN > (p_buf_end - p_buf)) + return NULL; + + p_buf = osqlcomm_uuid_rpl_type_put(&(p_fingerprint_uuid_rpl->hd), p_buf, p_buf_end); + p_buf = osqlcomm_fingerprint_type_put(&(p_fingerprint_uuid_rpl->dt), p_buf, p_buf_end); + + return p_buf; +} + typedef struct osql_index { unsigned long long seq; int ixnum; @@ -4121,6 +4199,69 @@ int osql_send_usedb(osql_target_t *target, unsigned long long rqid, uuid_t uuid, return rc; } +/** + * Send OSQL_FINGERPRINT op. + * + * Carries the originating statement's SQL fingerprint (MD5 of the normalized + * SQL, already computed on the replicant) to the master, so the master can + * attribute the write-apply page-in I/O it generates to that fingerprint. The + * payload is a fixed-size opaque 16-byte blob, so unlike USEDB there is no + * trailing send. + */ +int osql_send_fingerprint(osql_target_t *target, unsigned long long rqid, uuid_t uuid, const unsigned char *fingerprint, + int type) +{ + int msglen; + int rc = 0; + + uint8_t buf[(int)OSQLCOMM_FINGERPRINT_RPL_UUID_TYPE_LEN > (int)OSQLCOMM_FINGERPRINT_RPL_TYPE_LEN + ? OSQLCOMM_FINGERPRINT_RPL_UUID_TYPE_LEN + : OSQLCOMM_FINGERPRINT_RPL_TYPE_LEN]; + + if (check_master(target)) + return OSQL_SEND_ERROR_WRONGMASTER; + + if (rqid == OSQL_RQID_USE_UUID) { + osql_fingerprint_rpl_uuid_t fp_uuid_rpl = {{0}}; + uint8_t *p_buf = buf; + uint8_t *p_buf_end = (p_buf + OSQLCOMM_FINGERPRINT_RPL_UUID_TYPE_LEN); + + msglen = OSQLCOMM_FINGERPRINT_RPL_UUID_TYPE_LEN; + + fp_uuid_rpl.hd.type = OSQL_FINGERPRINT; + comdb2uuidcpy(fp_uuid_rpl.hd.uuid, uuid); + memcpy(fp_uuid_rpl.dt.fingerprint, fingerprint, FINGERPRINTSZ); + + if (!(p_buf = osqlcomm_fingerprint_uuid_rpl_type_put(&fp_uuid_rpl, p_buf, p_buf_end))) { + logmsg(LOGMSG_ERROR, "%s:%s returns NULL\n", __func__, "osqlcomm_fingerprint_uuid_rpl_type_put"); + return -1; + } + type = osql_net_type_to_net_uuid_type(NET_OSQL_SOCK_RPL); + } else { + osql_fingerprint_rpl_t fp_rpl = {{0}}; + uint8_t *p_buf = buf; + uint8_t *p_buf_end = (p_buf + OSQLCOMM_FINGERPRINT_RPL_TYPE_LEN); + + msglen = OSQLCOMM_FINGERPRINT_RPL_TYPE_LEN; + + fp_rpl.hd.type = OSQL_FINGERPRINT; + fp_rpl.hd.sid = rqid; + memcpy(fp_rpl.dt.fingerprint, fingerprint, FINGERPRINTSZ); + + if (!(p_buf = osqlcomm_fingerprint_rpl_type_put(&fp_rpl, p_buf, p_buf_end))) { + logmsg(LOGMSG_ERROR, "%s:%s returns NULL\n", __func__, "osqlcomm_fingerprint_rpl_type_put"); + return -1; + } + } + + rc = target->send(target, type, &buf, msglen, 0, NULL, 0); + + if (rc) + logmsg(LOGMSG_ERROR, "%s target->send returns rc=%d\n", __func__, rc); + + return rc; +} + /** * Send UPDCOLS op * It handles remote/local connectivity @@ -7422,6 +7563,22 @@ int osql_process_packet(struct ireq *iq, uuid_t uuid, void *trans, char **pmsg, } } break; + case OSQL_FINGERPRINT: { + osql_fingerprint_t dt = {{0}}; + /* Fixed-size payload; p_buf_end is derived from the struct size (as + * OSQL_USEDB / OSQL_STARTGEN do), relying on the net layer to deliver + * the whole message. The _get below still bounds-checks against it. */ + const uint8_t *p_buf_end = p_buf + sizeof(osql_fingerprint_t); + if (!osqlcomm_fingerprint_type_get(&dt, p_buf, p_buf_end)) { + logmsg(LOGMSG_ERROR, "%s: failed to read OSQL_FINGERPRINT\n", __func__); + break; + } + /* Arm this block-processor thread so the write-apply page-ins that + * follow are attributed to the originating statement's fingerprint. + * Cleared when the session finishes applying (process_this_session). */ + bdb_fingerprint_rtstats_set_write(dt.fingerprint, FINGERPRINTSZ, fingerprint_has_main_entry(dt.fingerprint)); + } break; + case OSQL_UPDREC: case OSQL_UPDATE: { osql_upd_t dt; diff --git a/db/osqlcomm.h b/db/osqlcomm.h index d911c572b9..abba5f5b63 100644 --- a/db/osqlcomm.h +++ b/db/osqlcomm.h @@ -60,6 +60,14 @@ int osql_comm_send_socksqlreq(osql_target_t *target, const char *sql, int sqlen, int osql_send_usedb(osql_target_t *target, unsigned long long rqid, uuid_t uuid, char *tablename, int type, unsigned long long version); +/** + * Send FINGERPRINT op + * Carries the originating statement's SQL fingerprint to the master for + * write-I/O accounting. It handles remote/local connectivity. + */ +int osql_send_fingerprint(osql_target_t *target, unsigned long long rqid, uuid_t uuid, const unsigned char *fingerprint, + int type); + /** * Send INDEX op * It handles remote/local connectivity diff --git a/db/osqlrpltypes.h b/db/osqlrpltypes.h index 04148179ab..0594e4794d 100644 --- a/db/osqlrpltypes.h +++ b/db/osqlrpltypes.h @@ -54,7 +54,8 @@ XMACRO_OSQL_RPL_TYPES( OSQL_DONE_WITH_EFFECTS, 28, "OSQL_DONE_WITH_EFFECTS" ) XMACRO_OSQL_RPL_TYPES( OSQL_PREPARE, 29, "OSQL_PREPARE" ) /* participant should prepare */ \ XMACRO_OSQL_RPL_TYPES( OSQL_DIST_TXNID, 30, "OSQL_DIST_TXNID" ) /* send dist-txnid to coordinator */ \ XMACRO_OSQL_RPL_TYPES( OSQL_PARTICIPANT, 31, "OSQL_PARTICIPANT" ) /* a participant (to coordinator) */ \ -XMACRO_OSQL_RPL_TYPES( MAX_OSQL_TYPES, 32, "OSQL_MAX") +XMACRO_OSQL_RPL_TYPES( OSQL_FINGERPRINT, 32, "OSQL_FINGERPRINT" ) /* SQL fingerprint for master write-I/O accounting */ \ +XMACRO_OSQL_RPL_TYPES( MAX_OSQL_TYPES, 33, "OSQL_MAX") // clang-format on diff --git a/db/osqlshadtbl.c b/db/osqlshadtbl.c index ad1ca30b91..7650b85be1 100644 --- a/db/osqlshadtbl.c +++ b/db/osqlshadtbl.c @@ -67,6 +67,17 @@ typedef struct rec_flags { int flags; } rec_flags_t; +extern int gbl_osql_send_fingerprint; + +/* Per-row SQL fingerprint kept in a shad_tbl side hash. Keyed by seq for + * inserts/updates and by genid for deletes -- the same keys the drain loop uses + * -- so the originating statement's fingerprint can be emitted to the master + * (OSQL_FINGERPRINT) ahead of each op at commit time. */ +typedef struct rec_fingerprint { + unsigned long long seq; /* seq (ins/upd) or genid (del) */ + unsigned char fingerprint[FINGERPRINTSZ]; +} rec_fingerprint_t; + static shad_tbl_t *get_shadtbl(struct BtCursor *pCur); static shad_tbl_t *open_shadtbl(struct BtCursor *pCur); static shad_tbl_t *create_shadtbl(struct BtCursor *pCur, @@ -173,6 +184,10 @@ static int destroy_shadtbl(shad_tbl_t *tbl) destroy_idx_hash(tbl->ins_rec_hash); if (tbl->upd_rec_hash) destroy_idx_hash(tbl->upd_rec_hash); + if (tbl->ins_fp_hash) + destroy_idx_hash(tbl->ins_fp_hash); + if (tbl->del_fp_hash) + destroy_idx_hash(tbl->del_fp_hash); if (tbl->delidx_tbl) destroy_tablecursor(tbl->env->bdb_env, tbl->delidx_cur, tbl->delidx_tbl, @@ -515,6 +530,8 @@ static shad_tbl_t *create_shadtbl(struct BtCursor *pCur, hash_init_o(offsetof(rec_flags_t, seq), sizeof(unsigned long long)); tbl->upd_rec_hash = hash_init_o(offsetof(rec_flags_t, seq), sizeof(unsigned long long)); + tbl->ins_fp_hash = hash_init_o(offsetof(rec_fingerprint_t, seq), sizeof(unsigned long long)); + tbl->del_fp_hash = hash_init_o(offsetof(rec_fingerprint_t, seq), sizeof(unsigned long long)); listc_abl(&clnt->osql.shadtbls, tbl); pCur->shadtbl = tbl; @@ -906,6 +923,53 @@ static int get_rec_flags(struct sqlclntstate *clnt, shad_tbl_t *tbl, return 0; } +/* Remember the current statement's fingerprint for this row, keyed the same way + * the drain loop later looks it up (seq for ins/upd, genid for del). No-op when + * the feature is off or the fingerprint is unset, so nothing is allocated in the + * common case. */ +static int save_rec_fingerprint(struct sqlclntstate *clnt, shad_tbl_t *tbl, unsigned long long key, int is_del) +{ + hash_t *h; + rec_fingerprint_t tmp; + rec_fingerprint_t *rf; + + if (!gbl_osql_send_fingerprint || fingerprint_is_zero(clnt->work.aFingerprint)) + return 0; + + h = is_del ? tbl->del_fp_hash : tbl->ins_fp_hash; + if (h == NULL) + return 0; + + tmp.seq = key; + if (hash_find(h, &tmp) != NULL) /* already recorded for this key */ + return 0; + + rf = calloc(1, sizeof(rec_fingerprint_t)); + if (!rf) { + logmsg(LOGMSG_ERROR, "%s: unable to allocate %zu bytes\n", __func__, sizeof(rec_fingerprint_t)); + return -1; + } + rf->seq = key; + memcpy(rf->fingerprint, clnt->work.aFingerprint, FINGERPRINTSZ); + hash_add(h, rf); + + return 0; +} + +/* Returns a pointer to the 16-byte fingerprint stored for this row, or NULL. */ +static const unsigned char *get_rec_fingerprint(shad_tbl_t *tbl, unsigned long long key, int is_del) +{ + hash_t *h = is_del ? tbl->del_fp_hash : tbl->ins_fp_hash; + rec_fingerprint_t tmp; + rec_fingerprint_t *rf; + + if (h == NULL) + return NULL; + tmp.seq = key; + rf = hash_find(h, &tmp); + return rf ? rf->fingerprint : NULL; +} + /* * NOTE: * Handle upd table for multiple updates of synthetic rows @@ -1082,6 +1146,10 @@ int osql_save_updrec(struct BtCursor *pCur, struct sql_thread *thd, char *pData, return -1; } + if (save_rec_fingerprint(thd->clnt, tbl, tmp, 0 /* upd keyed by seq */)) { + return -1; + } + #ifdef TEST_OSQL uuidstr_t us; fprintf(stdout, @@ -1164,6 +1232,10 @@ int osql_save_insrec(struct BtCursor *pCur, struct sql_thread *thd, char *pData, return -1; } + if (save_rec_fingerprint(thd->clnt, tbl, tmp, 0 /* ins */)) { + return -1; + } + tbl->seq = increment_seq(tbl->seq); /*++tbl->seq;*/ @@ -1210,6 +1282,10 @@ int osql_save_delrec(struct BtCursor *pCur, struct sql_thread *thd) return -1; } + if (save_rec_fingerprint(thd->clnt, tbl, pCur->genid, 1 /* del */)) { + return -1; + } + thd->clnt->osql.dirty = 1; return 0; @@ -1531,6 +1607,10 @@ int osql_shadtbl_process(struct sqlclntstate *clnt, int *nops, int *bdberr, if (rc) return -1; + /* Force the first row of this drain to (re)send its fingerprint: the master + * clears its write-side fingerprint TLS after each applied session. */ + osql->fingerprint_sent = 0; + LISTC_FOR_EACH(&osql->shadtbls, tbl, linkv) { /* we need to reset any cached nops in tbl */ @@ -1653,6 +1733,33 @@ int osql_shadtbl_cleartbls(struct sqlclntstate *clnt) /****************************************** INTERNALS * **************************************/ +/* Emit an OSQL_FINGERPRINT op for this row's fingerprint, ahead of the row op + * it describes, so the master attributes the write-apply I/O to it. Deduped + * against the last fingerprint sent this drain (osql->last_fingerprint / + * fingerprint_sent), so contiguous same-statement rows cost one op. fingerprint + * is NULL when nothing was recorded for the row (feature off, or zero fp). */ +static int process_local_shadtbl_fingerprint(struct sqlclntstate *clnt, const unsigned char *fingerprint, + int osql_nettype) +{ + osqlstate_t *osql = &clnt->osql; + int rc; + + if (fingerprint == NULL) + return 0; + if (osql->fingerprint_sent && memcmp(osql->last_fingerprint, fingerprint, FINGERPRINTSZ) == 0) + return 0; + + rc = osql_send_fingerprint(&osql->target, osql->rqid, osql->uuid, fingerprint, osql_nettype); + if (rc) + return rc; + + memcpy(osql->last_fingerprint, fingerprint, FINGERPRINTSZ); + osql->fingerprint_sent = 1; + osql->replicant_numops++; + DEBUG_PRINT_NUMOPS(); + return 0; +} + static int process_local_shadtbl_usedb(struct sqlclntstate *clnt, char *tablename, int tableversion) { @@ -1705,6 +1812,10 @@ static int process_local_shadtbl_skp(struct sqlclntstate *clnt, shad_tbl_t *tbl, return SQLITE_TOOBIG; } + rc = process_local_shadtbl_fingerprint(clnt, get_rec_fingerprint(tbl, genid, 1 /* del */), osql_nettype); + if (rc) + return SQLITE_INTERNAL; + if (osql->is_reorder_on) { rc = osql_send_delrec(&osql->target, osql->rqid, osql->uuid, genid, @@ -1997,6 +2108,10 @@ static int process_local_shadtbl_add(struct sqlclntstate *clnt, shad_tbl_t *tbl, else if (rc == IX_FND) goto next; + rc = process_local_shadtbl_fingerprint(clnt, get_rec_fingerprint(tbl, key, 0 /* ins */), osql_nettype); + if (rc) + return SQLITE_INTERNAL; + if (osql->is_reorder_on) { rc = osql_send_insrec(&osql->target, osql->rqid, osql->uuid, key, (gbl_partial_indexes && tbl->ix_partial) @@ -2112,6 +2227,12 @@ static int process_local_shadtbl_upd(struct sqlclntstate *clnt, shad_tbl_t *tbl, ((tbl->nops + crt_nops) > clnt->osql_max_trans)) { return SQLITE_TOOBIG; } + + rc = process_local_shadtbl_fingerprint(clnt, get_rec_fingerprint(tbl, seq, 0 /* upd keyed by seq */), + osql_nettype); + if (rc) + return SQLITE_INTERNAL; + if (osql->is_reorder_on) { rc = osql_send_updrec(&osql->target, osql->rqid, osql->uuid, genid, (gbl_partial_indexes && tbl->ix_partial) diff --git a/db/osqlshadtbl.h b/db/osqlshadtbl.h index 0fe5f96d41..fbae4d4547 100644 --- a/db/osqlshadtbl.h +++ b/db/osqlshadtbl.h @@ -40,6 +40,10 @@ struct shad_tbl { hash_t *ins_rec_hash; hash_t *upd_rec_hash; + hash_t *ins_fp_hash; /* per-row SQL fingerprint, keyed by seq (inserts and + updates) */ + hash_t *del_fp_hash; /* per-row SQL fingerprint, keyed by genid (deletes) */ + struct tmp_table *upd_tbl; /* all updates go here also, key=tbl->seq, data=original_genid */ struct tmp_table *blb_tbl; /* all blobs go here, diff --git a/db/osqlsqlthr.c b/db/osqlsqlthr.c index 0eda3a8475..54fb1e024e 100644 --- a/db/osqlsqlthr.c +++ b/db/osqlsqlthr.c @@ -938,6 +938,7 @@ static int osql_sock_restart(struct sqlclntstate *clnt, int maxretries, int keep sentops = 0; osql->replicant_numops = 0; + osql->fingerprint_sent = 0; /* re-send fingerprint for the new session */ if (osql->tablename) { free(osql->tablename); osql->tablename = NULL; @@ -1017,6 +1018,12 @@ static int osql_sock_restart(struct sqlclntstate *clnt, int maxretries, int keep int gbl_random_blkseq_replays; int gbl_osql_send_startgen = 1; +/* When set, replicants send an OSQL_FINGERPRINT op ahead of write ops so the + * master can attribute write-apply page-in I/O to the originating statement's + * fingerprint. Default OFF: an older master fails the whole transaction on an + * unknown osql op, so this must stay off until the entire cluster is upgraded. */ +int gbl_osql_send_fingerprint = 0; + static inline int sock_restart_retryable_rcode(int restart_rc) { switch (restart_rc) { @@ -1445,6 +1452,25 @@ static int osql_send_usedb_logic_int(char *tablename, struct sqlclntstate *clnt, if (clnt->dml_tables && !hash_find_readonly(clnt->dml_tables, tablename)) hash_add(clnt->dml_tables, strdup(tablename)); + /* Tell the master which fingerprint the following write ops belong to, so + * it can attribute its write-apply page-in I/O. Sent (and re-sent) only + * when the statement's fingerprint changes within this session -- done + * ahead of the usedb dedup below so a new statement on the same table is + * still attributed correctly. */ + if (gbl_osql_send_fingerprint && !fingerprint_is_zero(clnt->work.aFingerprint) && + (!osql->fingerprint_sent || memcmp(osql->last_fingerprint, clnt->work.aFingerprint, FINGERPRINTSZ) != 0)) { + do { + rc = osql_send_fingerprint(&osql->target, osql->rqid, osql->uuid, clnt->work.aFingerprint, nettype); + RESTART_SOCKSQL; + } while (restarted); + if (rc) + return rc; + memcpy(osql->last_fingerprint, clnt->work.aFingerprint, FINGERPRINTSZ); + osql->fingerprint_sent = 1; + osql->replicant_numops++; + DEBUG_PRINT_NUMOPS(); + } + int tablenamelen = strlen(tablename) + 1; /*including trailing 0*/ if (osql->tablename) { if (osql->tablenamelen == tablenamelen && @@ -1608,6 +1634,7 @@ static int osql_send_commit_logic(struct sqlclntstate *clnt, int retries, int ne osql->tablename = NULL; osql->tablenamelen = 0; } + osql->fingerprint_sent = 0; /* re-send fingerprint for the next transaction */ osql->tran_ops = 0; /* reset transaction size counter*/ if (!clnt->dbtran.trans_has_sp) { diff --git a/db/sql.h b/db/sql.h index 1b84d2b471..330d944d4b 100644 --- a/db/sql.h +++ b/db/sql.h @@ -144,6 +144,9 @@ typedef struct osqlstate { uuid_t uuid; /* session id, take 2 */ char *tablename; /* malloc-ed cache of send tablename for usedb */ int tablenamelen; /* tablename length */ + unsigned char last_fingerprint[FINGERPRINTSZ]; /* last fingerprint sent to + the master this session */ + int fingerprint_sent; /* set once we've sent an OSQL_FINGERPRINT */ int sentops; /* number of operations per statement */ int tran_ops; /* actual number of operations for a transaction */ int replicant_numops; /* total num of ops sent by replicant to master which @@ -1608,6 +1611,8 @@ void restore_thd_cost_and_reset(struct sqlthdstate *thd, Vdbe *pVdbe); void clnt_query_cost(struct sqlthdstate *thd, double *pCost, int64_t *pPrepMs); int clear_fingerprints(int *plans_count); +int fingerprint_has_main_entry(const unsigned char fingerprint[FINGERPRINTSZ]); +int fingerprint_is_zero(const unsigned char fingerprint[FINGERPRINTSZ]); void calc_fingerprint(const char *zNormSql, size_t *pnNormSql, unsigned char fingerprint[FINGERPRINTSZ]); void add_fingerprint(struct sqlclntstate *, sqlite3_stmt *, struct string_ref *, const char *, int64_t, int64_t, diff --git a/sqlite/ext/comdb2/fingerprints.c b/sqlite/ext/comdb2/fingerprints.c index 7eaaa196fb..5071abde86 100644 --- a/sqlite/ext/comdb2/fingerprints.c +++ b/sqlite/ext/comdb2/fingerprints.c @@ -35,8 +35,13 @@ struct fingerprint_track_systbl { char *zNormSql; /* The normalized SQL query */ size_t nNormSql; /* Length of normalized SQL query */ char *excluded; /* 'Y' if excluded from longreqs */ - int64_t total_pagein_read; /* Cumulative bufferpool page-ins (hit+miss) */ + int64_t total_pagein_read; /* Cumulative read (SQL) bufferpool page-ins (hit+miss) */ int64_t total_pagein_read_io; /* Subset of the above that required disk I/O */ + int64_t total_write_pagein_read; /* Cumulative write-apply page-ins on the master */ + int64_t total_write_pagein_read_io; /* Subset of the above that required disk I/O */ + char *has_query_info; /* 'Y' if this node has the full query text (a + gbl_fingerprint_hash entry); 'N' for rtstats-only + fingerprints (e.g. master write-apply accounting) */ char fp[FINGERPRINTSZ*2+1]; }; @@ -54,65 +59,146 @@ static void release_callback(void *data, int npoints) } } +/* Counts every rtstats entry, an upper bound on how many rtstats-only rows we + * may need to append (before knowing the overlap with gbl_fingerprint_hash). */ +static void rtstats_count_cb(const unsigned char *fingerprint, const uint64_t *counts, + int has_main_entry, void *arg) +{ + (void)fingerprint; (void)counts; (void)has_main_entry; + (*(int *)arg)++; +} + +struct rtstats_merge_ctx { + hash_t *seen; /* fingerprints already emitted from gbl_fingerprint_hash */ + struct fingerprint_track_systbl *pFp; /* output row array */ + int capacity; /* allocated rows */ + int copied; /* next free row index */ +}; + +/* Appends one row per rtstats fingerprint that has no gbl_fingerprint_hash + * entry -- i.e. fingerprints known only via write-apply accounting (typically + * on the master). These have no query text, so has_query_info='N'. + * + * Only entries with write-apply activity are surfaced. A fingerprint can also + * be rtstats-only transiently on any node -- the read-side arms an entry at + * prepare time, but gbl_fingerprint_hash is populated only when the statement + * finishes -- so an in-flight query (including this very systable scan) would + * otherwise appear here with no query text. Those have no write page-ins, so + * gating on the write counters keeps the read-side view unchanged and limits + * the new rows to genuine master write-apply fingerprints. */ +static void rtstats_merge_cb(const unsigned char *fingerprint, const uint64_t *counts, + int has_main_entry, void *arg) +{ + struct rtstats_merge_ctx *ctx = (struct rtstats_merge_ctx *)arg; + struct fingerprint_track_systbl *row; + (void)has_main_entry; + + if (counts[2] == 0 && counts[3] == 0) + return; /* no write-apply activity -- read-side-only/in-flight entry */ + if (ctx->seen != NULL && hash_find(ctx->seen, fingerprint) != NULL) + return; /* already emitted above with full query info */ + if (ctx->copied >= ctx->capacity) + return; /* raced with new inserts; snapshot is best-effort */ + + row = &ctx->pFp[ctx->copied]; + util_tohex(row->fp, (char *)fingerprint, FINGERPRINTSZ); + row->fingerprint = row->fp; + row->zNormSql = NULL; + row->nNormSql = 0; + row->excluded = + reqlog_fingerprint_is_excluded((char *)fingerprint) ? "Y" : "N"; + row->has_query_info = "N"; + row->total_pagein_read = counts[0]; + row->total_pagein_read_io = counts[1]; + row->total_write_pagein_read = counts[2]; + row->total_write_pagein_read_io = counts[3]; + ctx->copied++; +} + static int fingerprints_callback(void **data, int *npoints) { int rc = SQLITE_OK; + int rtstats_total = 0; + int count = 0; + int capacity; + int copied = 0; + struct fingerprint_track_systbl *pFp = NULL; + hash_t *seen = NULL; + unsigned char *seen_keys = NULL; /* raw 16-byte keys backing `seen` */ + *npoints = 0; *data = NULL; + + /* Upper bound on rtstats-only rows we might append. Taken before the + * fingerprint-hash lock (this only takes the rtstats hash lock). */ + bdb_fingerprint_rtstats_foreach(rtstats_count_cb, &rtstats_total); + Pthread_mutex_lock(&gbl_fingerprint_hash_mu); - if (gbl_fingerprint_hash != NULL) { - int count; + if (gbl_fingerprint_hash != NULL) hash_info(gbl_fingerprint_hash, NULL, NULL, NULL, NULL, &count, NULL, NULL); + + capacity = count + rtstats_total; + if (capacity > 0) { + pFp = calloc(capacity, sizeof(struct fingerprint_track_systbl)); if (count > 0) { - struct fingerprint_track_systbl *pFp = calloc(count, - sizeof(struct fingerprint_track_systbl)); - if (pFp != NULL) { - struct fingerprint_track *pEntry; - int copied = 0; - void *hash_cur; - unsigned int hash_cur_buk; - pEntry = hash_first(gbl_fingerprint_hash, &hash_cur, &hash_cur_buk); - while (pEntry != NULL) { - assert( copiedfingerprint, FINGERPRINTSZ); - pFp[copied].fingerprint = pFp[copied].fp; - pFp[copied].count = pEntry->count; - pFp[copied].cost = pEntry->cost; - pFp[copied].time = pEntry->time; - pFp[copied].prepTime = pEntry->prepTime; - pFp[copied].rows = pEntry->rows; - if (reqlog_fingerprint_is_excluded((char *)pEntry->fingerprint)) { - pFp[copied].excluded = "Y"; - } else { - pFp[copied].excluded = "N"; - } - if (pEntry->zNormSql != NULL) { - pFp[copied].zNormSql = strdup(pEntry->zNormSql); - pFp[copied].nNormSql = strlen(pEntry->zNormSql); - assert( pFp[copied].nNormSql==pEntry->nNormSql ); - } - { - uint64_t n_pagein_read, n_pagein_read_io; - bdb_fingerprint_rtstats_get(pEntry->fingerprint, FINGERPRINTSZ, - &n_pagein_read, &n_pagein_read_io); - pFp[copied].total_pagein_read = n_pagein_read; - pFp[copied].total_pagein_read_io = n_pagein_read_io; - } - copied++; - pEntry = hash_next(gbl_fingerprint_hash, &hash_cur, &hash_cur_buk); - } - if (rc == SQLITE_OK) { - *data = pFp; - *npoints = count; - } else { - release_callback(pFp, count); - } - } else { - rc = SQLITE_NOMEM; + seen = hash_init(FINGERPRINTSZ); + seen_keys = malloc((size_t)count * FINGERPRINTSZ); + } + } + + if (pFp == NULL || (count > 0 && (seen == NULL || seen_keys == NULL))) { + rc = (capacity > 0) ? SQLITE_NOMEM : SQLITE_OK; + } else if (count > 0) { + struct fingerprint_track *pEntry; + void *hash_cur; + unsigned int hash_cur_buk; + pEntry = hash_first(gbl_fingerprint_hash, &hash_cur, &hash_cur_buk); + while (pEntry != NULL && copied < count) { + uint64_t r = 0, rio = 0, wr = 0, wrio = 0; + util_tohex(pFp[copied].fp, (char *)pEntry->fingerprint, FINGERPRINTSZ); + pFp[copied].fingerprint = pFp[copied].fp; + pFp[copied].count = pEntry->count; + pFp[copied].cost = pEntry->cost; + pFp[copied].time = pEntry->time; + pFp[copied].prepTime = pEntry->prepTime; + pFp[copied].rows = pEntry->rows; + pFp[copied].excluded = + reqlog_fingerprint_is_excluded((char *)pEntry->fingerprint) ? "Y" : "N"; + pFp[copied].has_query_info = "Y"; + if (pEntry->zNormSql != NULL) { + pFp[copied].zNormSql = strdup(pEntry->zNormSql); + pFp[copied].nNormSql = strlen(pEntry->zNormSql); } + bdb_fingerprint_rtstats_get(pEntry->fingerprint, FINGERPRINTSZ, + &r, &rio, &wr, &wrio); + pFp[copied].total_pagein_read = r; + pFp[copied].total_pagein_read_io = rio; + pFp[copied].total_write_pagein_read = wr; + pFp[copied].total_write_pagein_read_io = wrio; + /* remember this fingerprint so the rtstats merge below skips it */ + memcpy(&seen_keys[copied * FINGERPRINTSZ], pEntry->fingerprint, FINGERPRINTSZ); + hash_add(seen, &seen_keys[copied * FINGERPRINTSZ]); + copied++; + pEntry = hash_next(gbl_fingerprint_hash, &hash_cur, &hash_cur_buk); } } Pthread_mutex_unlock(&gbl_fingerprint_hash_mu); + + /* Append fingerprints known only via rtstats (no query text). Runs without + * the fingerprint-hash lock; `seen` is a private snapshot. */ + if (rc == SQLITE_OK && pFp != NULL) { + struct rtstats_merge_ctx ctx = {seen, pFp, capacity, copied}; + bdb_fingerprint_rtstats_foreach(rtstats_merge_cb, &ctx); + copied = ctx.copied; + *data = pFp; + *npoints = copied; + } else if (pFp != NULL) { + release_callback(pFp, copied); + } + + if (seen != NULL) + hash_free(seen); + free(seen_keys); return rc; } @@ -143,9 +229,15 @@ int systblFingerprintsInit(sqlite3 *db) offsetof(struct fingerprint_track_systbl, total_pagein_read), CDB2_INTEGER, "total_pagein_read_io", -1, offsetof(struct fingerprint_track_systbl, total_pagein_read_io), + CDB2_INTEGER, "total_write_pagein_read", -1, + offsetof(struct fingerprint_track_systbl, total_write_pagein_read), + CDB2_INTEGER, "total_write_pagein_read_io", -1, + offsetof(struct fingerprint_track_systbl, total_write_pagein_read_io), CDB2_CSTRING, "normalized_sql", -1, offsetof(struct fingerprint_track_systbl, zNormSql), CDB2_CSTRING, "excluded_from_longreqs", -1, offsetof(struct fingerprint_track_systbl, excluded), + CDB2_CSTRING, "has_query_info", -1, + offsetof(struct fingerprint_track_systbl, has_query_info), SYSTABLE_END_OF_FIELDS); } diff --git a/tests/fingerprints.test/README b/tests/fingerprints.test/README index 06794d47f4..8d4b1c3c38 100644 --- a/tests/fingerprints.test/README +++ b/tests/fingerprints.test/README @@ -14,6 +14,30 @@ executed inside a single stored procedure (Lua inner statements), verifying each gets its own independently-tracked page-in counters rather than one statement's counters leaking into the other's. +"t12.req" (driven by "fpwrite.sh") verifies the write-side counterpart: writes +are applied on the master, which attributes the page-in I/O it generates to the +originating statement's fingerprint (comdb2_fingerprints.total_write_pagein_read +/ total_write_pagein_read_io). This relies on the "osql_send_fingerprint" +tunable, enabled in lrl.options, which has the replicant transmit the +fingerprint to the master. Because writes are applied on the master -- which +need not be the node the client connected to -- the test looks the fingerprint +up on the connected node (where the query text exists) and then asserts its +write counters on the master by fingerprint hash. As with t10/t11, only +structural invariants are checked (write page-ins were attributed, and the +disk-I/O subset never exceeds the total), since exact counts are not +deterministic. + +"t13.req" (driven by "fpwrite2.sh") verifies the same master write-apply +attribution for the non-SOSQL isolation levels -- read committed, snapshot +isolation, and serial -- which buffer writes in shadow tables and drain them to +the master at commit. serial is enabled via "enable_serial_isolation" in +lrl.options (which also brings up the snapshot machinery); the levels are then +selected per transaction with "set transaction ...". For each level it runs an +UPDATE and DELETE against pre-existing (committed) rows and asserts those +fingerprints carry write page-ins on the master. Rows updated/deleted must be +pre-existing: deleting a row inserted in the same transaction cancels in the +shadow table and never reaches the master. + -------------------------------- SPECIAL NOTES -------------------------------- The "t03.req" test file purposely excludes the following fingerprints from its diff --git a/tests/fingerprints.test/fpwrite.sh b/tests/fingerprints.test/fpwrite.sh new file mode 100755 index 0000000000..b08890f18b --- /dev/null +++ b/tests/fingerprints.test/fpwrite.sh @@ -0,0 +1,36 @@ +#!/usr/bin/env bash + +# Verifies the master attributes write-apply page-in I/O to the originating +# statement's fingerprint, exposed via comdb2_fingerprints.total_write_pagein_read. +# +# Writes issued on $SP_HOST are offloaded to and applied on the master, so the +# write-pagein counters accumulate on the *master* -- which may not be $SP_HOST. +# We therefore look the fingerprint up on $SP_HOST (where the query text, and +# hence normalized_sql, exists) and then assert its write counters on the master +# by fingerprint hash. This works whether or not $SP_HOST is the master: in a +# cluster the master has an rtstats-only row (has_query_info='N', no query text) +# keyed by the same fingerprint we transmitted. +# +# gbl_osql_send_fingerprint is enabled for this db via lrl.options. + +# Generate write-apply I/O on the master. All five inserts normalize to one +# fingerprint. Suppress setup output -- only the final assertion is diffed. +cdb2sql --host $SP_HOST $SP_OPTIONS - > /dev/null 2>&1 <<'EOF' +CREATE TABLE fp_wpagein(x INTEGER)$$ +INSERT INTO fp_wpagein(x) VALUES(1) +INSERT INTO fp_wpagein(x) VALUES(2) +INSERT INTO fp_wpagein(x) VALUES(3) +INSERT INTO fp_wpagein(x) VALUES(4) +INSERT INTO fp_wpagein(x) VALUES(5) +EOF + +# The INSERT fingerprint (from the node that ran the SQL engine). +fp=$(cdb2sql --tabs --host $SP_HOST $SP_OPTIONS "SELECT fingerprint FROM comdb2_fingerprints WHERE normalized_sql LIKE 'INSERT%fp_wpagein%'") + +# The master (where the writes were actually applied). +master=$(cdb2sql --tabs --host $SP_HOST $SP_OPTIONS "SELECT host FROM comdb2_cluster WHERE is_master='Y'") + +# Page counts are not deterministic across environments, so assert structural +# invariants only: write-apply page-ins were attributed (>0), and the disk-I/O +# subset never exceeds the total. +cdb2sql --host $master $SP_OPTIONS "SELECT (total_write_pagein_read > 0) AS writes_present, (total_write_pagein_read_io <= total_write_pagein_read) AS io_le_total FROM comdb2_fingerprints WHERE fingerprint='$fp'" diff --git a/tests/fingerprints.test/fpwrite2.sh b/tests/fingerprints.test/fpwrite2.sh new file mode 100755 index 0000000000..a4f4a74158 --- /dev/null +++ b/tests/fingerprints.test/fpwrite2.sh @@ -0,0 +1,48 @@ +#!/usr/bin/env bash + +# Write-side fingerprint accounting under the non-SOSQL isolation levels +# (read committed, snapshot isolation, serial). These buffer writes in shadow +# tables and drain them to the master at commit; this checks the master +# attributes the write-apply page-ins to the originating statement's fingerprint +# for each level. +# +# serial is enabled via enable_serial_isolation in lrl.options (which also brings +# up the snapshot machinery); read committed needs no server enabling. +# +# For each level we first create committed rows under the default level, then run +# an UPDATE + DELETE + INSERT transaction under the level being tested. We assert +# on the UPDATE and DELETE fingerprints because those statements ran *only* under +# the level -- a clean signal. (The INSERT is exercised too, but its fingerprint +# is shared with the SOSQL pre-inserts, so it is not asserted. The rows updated +# and deleted are pre-existing/committed, since deleting a row inserted in the +# same transaction cancels in the shadow table and never reaches the master.) + +master=$(cdb2sql --tabs --host $SP_HOST $SP_OPTIONS "SELECT host FROM comdb2_cluster WHERE is_master='Y'") + +check_level() { + local level="$1" tbl="$2" + + cdb2sql --host $SP_HOST $SP_OPTIONS - > /dev/null 2>&1 < 0) AS writes_present, (total_write_pagein_read_io <= total_write_pagein_read) AS io_le_total FROM comdb2_fingerprints WHERE fingerprint='$ufp'" + cdb2sql --host $master $SP_OPTIONS "SELECT '$level' AS level, 'delete' AS op, (total_write_pagein_read > 0) AS writes_present, (total_write_pagein_read_io <= total_write_pagein_read) AS io_le_total FROM comdb2_fingerprints WHERE fingerprint='$dfp'" +} + +check_level "read committed" fp_iso_recom +check_level "snapshot isolation" fp_iso_snap +check_level "serial" fp_iso_serial diff --git a/tests/fingerprints.test/lrl.options b/tests/fingerprints.test/lrl.options index 4cba46063d..a7e5c57057 100644 --- a/tests/fingerprints.test/lrl.options +++ b/tests/fingerprints.test/lrl.options @@ -1 +1,3 @@ verbose_normalized_queries +osql_send_fingerprint 1 +enable_serial_isolation diff --git a/tests/fingerprints.test/t12.req b/tests/fingerprints.test/t12.req new file mode 100644 index 0000000000..e69de29bb2 diff --git a/tests/fingerprints.test/t12.req.args b/tests/fingerprints.test/t12.req.args new file mode 100644 index 0000000000..e69de29bb2 diff --git a/tests/fingerprints.test/t12.req.out b/tests/fingerprints.test/t12.req.out new file mode 100644 index 0000000000..c66494738a --- /dev/null +++ b/tests/fingerprints.test/t12.req.out @@ -0,0 +1 @@ +(writes_present=1, io_le_total=1) diff --git a/tests/fingerprints.test/t12.req.tool b/tests/fingerprints.test/t12.req.tool new file mode 100644 index 0000000000..0b6e3b8082 --- /dev/null +++ b/tests/fingerprints.test/t12.req.tool @@ -0,0 +1 @@ +./fpwrite.sh diff --git a/tests/fingerprints.test/t13.req b/tests/fingerprints.test/t13.req new file mode 100644 index 0000000000..e69de29bb2 diff --git a/tests/fingerprints.test/t13.req.args b/tests/fingerprints.test/t13.req.args new file mode 100644 index 0000000000..e69de29bb2 diff --git a/tests/fingerprints.test/t13.req.out b/tests/fingerprints.test/t13.req.out new file mode 100644 index 0000000000..e502c5e29a --- /dev/null +++ b/tests/fingerprints.test/t13.req.out @@ -0,0 +1,6 @@ +(level='read committed', op='update', writes_present=1, io_le_total=1) +(level='read committed', op='delete', writes_present=1, io_le_total=1) +(level='snapshot isolation', op='update', writes_present=1, io_le_total=1) +(level='snapshot isolation', op='delete', writes_present=1, io_le_total=1) +(level='serial', op='update', writes_present=1, io_le_total=1) +(level='serial', op='delete', writes_present=1, io_le_total=1) diff --git a/tests/fingerprints.test/t13.req.tool b/tests/fingerprints.test/t13.req.tool new file mode 100644 index 0000000000..c779dac0cb --- /dev/null +++ b/tests/fingerprints.test/t13.req.tool @@ -0,0 +1 @@ +./fpwrite2.sh