diff --git a/src/Makefile b/src/Makefile index 54c46f56761..646be3c7abf 100644 --- a/src/Makefile +++ b/src/Makefile @@ -605,6 +605,9 @@ ENGINE_SERVER_OBJ = \ crdt/merge.o \ crdt/peer.o \ rdb_forkless.o \ + tier/vlog.o \ + tier/evict.o \ + tier/compact.o \ rdma.o \ release.o \ replication.o \ diff --git a/src/aof.c b/src/aof.c index c46414f4ec8..127ca170278 100644 --- a/src/aof.c +++ b/src/aof.c @@ -32,6 +32,7 @@ #include "listpack.h" #include "sds.h" #include "server.h" +#include "tier/tier.h" #include "ordered_index.h" #include "bio.h" #include "rio.h" @@ -1925,6 +1926,13 @@ int rioWriteBulkObject(rio *r, robj *obj) { } if (obj->encoding == OBJ_ENCODING_INT) { return rioWriteBulkLongLong(r, (long)objectGetVal(obj)); + } else if (obj->encoding == OBJ_ENCODING_TIERED) { + sds raw = tierReadRef(objectGetVal(obj)); + int n; + if (!raw) return 0; + n = rioWriteBulkString(r, raw, sdslen(raw)); + sdsfree(raw); + return n; } else if (sdsEncodedObject(obj)) { return rioWriteBulkString(r, objectGetVal(obj), sdslen(objectGetVal(obj))); } else { diff --git a/src/config.c b/src/config.c index 4492828aee8..69cf77ebb44 100644 --- a/src/config.c +++ b/src/config.c @@ -3483,6 +3483,8 @@ standardConfig static_configs[] = { createBoolConfig("syslog-enabled", NULL, IMMUTABLE_CONFIG, server.syslog_enabled, 0, NULL, NULL), createBoolConfig("cluster-enabled", NULL, IMMUTABLE_CONFIG, server.cluster_enabled, 0, NULL, NULL), createBoolConfig("crdt-enabled", NULL, MODIFIABLE_CONFIG, server.crdt_enabled, 0, NULL, NULL), + createBoolConfig("tiering-enabled", NULL, MODIFIABLE_CONFIG, server.tiering_enabled, 0, NULL, NULL), + createBoolConfig("tiering-debug-exit-before-rename", NULL, MODIFIABLE_CONFIG, server.tiering_debug_exit_before_rename, 0, NULL, NULL), createBoolConfig("appendonly", NULL, MODIFIABLE_CONFIG | DENY_LOADING_CONFIG, server.aof_enabled, 0, NULL, updateAppendOnly), createBoolConfig("cluster-allow-reads-when-down", NULL, MODIFIABLE_CONFIG, server.cluster_allow_reads_when_down, 0, NULL, NULL), createBoolConfig("cluster-allow-pubsubshard-when-down", NULL, MODIFIABLE_CONFIG, server.cluster_allow_pubsubshard_when_down, 1, NULL, NULL), @@ -3506,6 +3508,7 @@ standardConfig static_configs[] = { /* String Configs */ createStringConfig("aclfile", NULL, IMMUTABLE_CONFIG, ALLOW_EMPTY_STRING, server.acl_filename, "", NULL, NULL), createStringConfig("crdt-peers", NULL, MODIFIABLE_CONFIG, ALLOW_EMPTY_STRING, server.crdt_peers, "", NULL, NULL), + createStringConfig("tiering-path", NULL, MODIFIABLE_CONFIG, EMPTY_STRING_IS_NULL, server.tiering_path, "tier.vlog", NULL, NULL), createStringConfig("unixsocket", NULL, IMMUTABLE_CONFIG, EMPTY_STRING_IS_NULL, server.unixsocket, NULL, NULL, NULL), createStringConfig("unixsocketgroup", NULL, IMMUTABLE_CONFIG, EMPTY_STRING_IS_NULL, server.unix_ctx_config.group, NULL, NULL, NULL), createStringConfig("pidfile", NULL, IMMUTABLE_CONFIG, EMPTY_STRING_IS_NULL, server.pidfile, NULL, NULL, NULL), @@ -3571,6 +3574,10 @@ standardConfig static_configs[] = { createIntConfig("databases", NULL, IMMUTABLE_CONFIG, 1, INT_MAX, server.config_databases, 16, INTEGER_CONFIG, NULL, NULL), createIntConfig("crdt-region-id", NULL, MODIFIABLE_CONFIG, 1, 65535, server.crdt_region_id, 1, INTEGER_CONFIG, NULL, NULL), createLongLongConfig("crdt-tombstone-gc-ms", NULL, MODIFIABLE_CONFIG, 0, LLONG_MAX, server.crdt_tombstone_gc_ms, 60000, INTEGER_CONFIG, NULL, NULL), + createIntConfig("tiering-min-value", NULL, MODIFIABLE_CONFIG, 1, INT_MAX, server.tiering_min_value, 128, INTEGER_CONFIG, NULL, NULL), + createIntConfig("tiering-idle-seconds", NULL, MODIFIABLE_CONFIG, 0, INT_MAX, server.tiering_idle_seconds, 60, INTEGER_CONFIG, NULL, NULL), + createIntConfig("tiering-debug-crash-after", NULL, MODIFIABLE_CONFIG, 0, INT_MAX, server.tiering_debug_crash_after, 0, INTEGER_CONFIG, NULL, NULL), + createLongLongConfig("tiering-max-ram", NULL, MODIFIABLE_CONFIG, 0, LLONG_MAX, server.tiering_max_ram, 0, MEMORY_CONFIG, NULL, NULL), createIntConfig("cluster-databases", NULL, IMMUTABLE_CONFIG, 1, INT_MAX, server.config_databases_cluster, 1, INTEGER_CONFIG, NULL, NULL), createIntConfig("port", NULL, MODIFIABLE_CONFIG, 0, 65535, server.port, 6379, INTEGER_CONFIG, NULL, updatePort), /* TCP port. */ createIntConfig("io-threads", NULL, DEBUG_CONFIG | MODIFIABLE_CONFIG, 1, IO_THREADS_MAX_NUM, server.io_threads_num, 1, INTEGER_CONFIG, NULL, updateIOThreads), /* Single threaded by default */ diff --git a/src/db.c b/src/db.c index e403b172825..4aceb403be0 100644 --- a/src/db.c +++ b/src/db.c @@ -30,6 +30,7 @@ #include "server.h" #include "crdt/peer.h" #include "rdb_forkless.h" +#include "tier/tier.h" #include "listpack.h" #include "hotkeys.h" #include "ordered_index.h" @@ -127,6 +128,8 @@ robj *lookupKey(serverDb *db, robj *key, int flags) { val->lru = lrulfu_touch(val->lru); } + if (server.tiering_enabled && objectGetType(val) == OBJ_STRING && objectGetEncoding(val) == OBJ_ENCODING_TIERED) + val = tierPromote(db, key, val); if (!(flags & (LOOKUP_NOSTATS | LOOKUP_WRITE))) server.stat_keyspace_hits++; /* TODO: Use separate hits stats for WRITE */ } else { diff --git a/src/object.c b/src/object.c index 58c69fe555d..9d4c3fa91ec 100644 --- a/src/object.c +++ b/src/object.c @@ -30,6 +30,7 @@ #include "hashtable.h" #include "server.h" +#include "tier/tier.h" #include "ordered_index.h" #include "serverassert.h" #include "functions.h" @@ -619,6 +620,14 @@ robj *dupStringObject(const robj *o) { d->encoding = OBJ_ENCODING_INT; d->val_ptr = o->val_ptr; return d; + case OBJ_ENCODING_TIERED: { + sds raw = tierReadRef(objectGetVal(o)); + robj *copy; + if (!raw) serverPanic("Tiered string unreadable"); + copy = createRawStringObject(raw, sdslen(raw)); + sdsfree(raw); + return copy; + } default: serverPanic("Wrong encoding."); break; } } @@ -700,6 +709,9 @@ robj *createModuleObject(moduleType *mt, void *value) { void freeStringObject(robj *o) { if (objectGetEncoding(o) == OBJ_ENCODING_RAW) { sdsfree(objectGetVal(o)); + } else if (objectGetEncoding(o) == OBJ_ENCODING_TIERED) { + tierMarkDead(objectGetVal(o)); + zfree(objectGetVal(o)); } } @@ -1073,6 +1085,13 @@ robj *getDecodedObject(robj *o) { incrRefCount(o); return o; } + if (objectGetType(o) == OBJ_STRING && objectGetEncoding(o) == OBJ_ENCODING_TIERED) { + sds raw = tierReadRef(objectGetVal(o)); + if (!raw) serverPanic("Tiered string unreadable"); + dec = createStringObject(raw, sdslen(raw)); + sdsfree(raw); + return dec; + } if (objectGetType(o) == OBJ_STRING && objectGetEncoding(o) == OBJ_ENCODING_INT) { char buf[32]; diff --git a/src/rdb.c b/src/rdb.c index 7e7941f3b00..8385e324cf2 100644 --- a/src/rdb.c +++ b/src/rdb.c @@ -51,6 +51,7 @@ #include "compression_stream.h" #include "forkless.h" #include "rdb_forkless.h" +#include "tier/tier.h" #include #include @@ -596,6 +597,14 @@ ssize_t rdbSaveStringObject(rio *rdb, robj *obj) { return n; } } + if (obj->encoding == OBJ_ENCODING_TIERED) { + sds raw = tierReadRef(objectGetVal(obj)); + ssize_t n; + if (!raw) return -1; + n = rdbSaveRawString(rdb, (unsigned char *)raw, sdslen(raw)); + sdsfree(raw); + return n; + } if (obj->encoding == OBJ_ENCODING_INT) { return rdbSaveLongLongAsStringObject(rdb, (long)objectGetVal(obj)); } else { diff --git a/src/server.c b/src/server.c index 2583ddc3a17..b7a4e941971 100644 --- a/src/server.c +++ b/src/server.c @@ -61,6 +61,7 @@ #include "forkless.h" #include "crdt/peer.h" #include "rdb_forkless.h" +#include "tier/tier.h" #include "eval.h" #include "bgiteration.h" @@ -1696,6 +1697,7 @@ long long serverCron(struct aeEventLoop *eventLoop, long long id, void *clientDa databasesCron(); crdtCron(); snapshotForklessCron(); + tierCron(); /* Start a scheduled AOF rewrite if this was requested by the user while * a BGSAVE was in progress. We don't start the rewrite if there is an @@ -6741,6 +6743,11 @@ sds genValkeyInfoString(dict *section_dict, int all_sections, int everything) { "mem_overhead_db_hashtable_rehashing:%zu\r\n", mh->overhead_db_hashtable_rehashing, "active_defrag_running:%d\r\n", server.active_defrag_cpu_percent, "lazyfree_pending_objects:%zu\r\n", lazyfreeGetPendingObjectsCount(), + "tiering_enabled:%d\r\n", server.tiering_enabled, + "tiering_spills:%lld\r\n", tierSpills(), + "tiering_promotions:%lld\r\n", tierPromotions(), + "tiering_bytes:%lld\r\n", tierBytesOnDisk(), + "tiering_dead_bytes:%lld\r\n", tierDeadBytes(), "lazyfreed_objects:%zu\r\n", lazyfreeGetFreedObjectsCount())); freeMemoryOverheadData(mh); } diff --git a/src/server.h b/src/server.h index 84501eedcdb..577ffbc7638 100644 --- a/src/server.h +++ b/src/server.h @@ -843,6 +843,7 @@ typedef struct ValkeyModuleType moduleType; #define OBJ_ENCODING_LISTPACK 11 /* Encoded as a listpack */ #define OBJ_ENCODING_LISTPACK2 12 /* Encoded as a listpack with metadata tag */ #define OBJ_ENCODING_PATH_HASH 13 /* Path hash backed by a radix tree */ +#define OBJ_ENCODING_TIERED 14 /* String payload lives in the tiering value log */ #define OBJ_REFCOUNT_BITS 29 #define OBJ_SHARED_REFCOUNT ((1 << OBJ_REFCOUNT_BITS) - 1) /* Global object never destroyed. */ @@ -2223,6 +2224,13 @@ struct valkeyServer { long long crdt_tombstone_gc_ms; int snapshot_mode; /* SNAPSHOT_MODE_FORK or SNAPSHOT_MODE_FORKLESS */ int snapshot_forkless_slice_keys; /* Max keys written per cron tick in forkless mode */ + int tiering_enabled; + char *tiering_path; + int tiering_min_value; + long long tiering_max_ram; + int tiering_idle_seconds; + int tiering_debug_crash_after; + int tiering_debug_exit_before_rename; int key_load_delay; /* Delay in microseconds between keys while * loading aof or rdb. (for testings). negative * value means fractions of microseconds (on average). */ diff --git a/src/tier/compact.c b/src/tier/compact.c new file mode 100644 index 00000000000..05697fa6001 --- /dev/null +++ b/src/tier/compact.c @@ -0,0 +1,32 @@ +/* Gather live tier pointers and rewrite the value log. + * + * Copyright (c) 2026, Cache Me Outside contributors. + * Licensed under the BSD 3-Clause License, the same terms as the server. + */ +#include "tier/tier.h" + +int tierCompactLive(void) { + tierRef **live = NULL; + int n = 0, cap = 0, dbid; + int rc; + for (dbid = 0; dbid < server.dbnum; dbid++) { + serverDb *db = server.db[dbid]; + kvstoreIterator *it; + void *next; + if (!db) continue; + it = kvstoreIteratorInit(db->keys, 0); + while (kvstoreIteratorNext(it, &next)) { + robj *o = next; + if (objectGetType(o) != OBJ_STRING || objectGetEncoding(o) != OBJ_ENCODING_TIERED) continue; + if (n == cap) { + cap = cap ? cap * 2 : 16; + live = zrealloc(live, sizeof(*live) * (size_t)cap); + } + live[n++] = objectGetVal(o); + } + kvstoreIteratorRelease(it); + } + rc = tierCompact(live, n); + zfree(live); + return rc; +} diff --git a/src/tier/evict.c b/src/tier/evict.c new file mode 100644 index 00000000000..928b58dd302 --- /dev/null +++ b/src/tier/evict.c @@ -0,0 +1,85 @@ +/* Move cold large strings to the value log and bring them back on read. + * + * Copyright (c) 2026, Cache Me Outside contributors. + * Licensed under the BSD 3-Clause License, the same terms as the server. + */ +#include "tier/tier.h" + +/* Replace the string body with a log pointer and free the body now. + * dbReplaceValue() can leave the old allocation charged: a refcount above 1 + * only decrements, and the swap path hands the sds to a throwaway object that + * lazy-free may not reclaim before the next INFO. The key stays the same + * hashtable entry, so this only rewrites that entry's value pointer. */ +static int tierSpillObject(serverDb *db, robj *key, robj *o) { + sds raw; + tierRef *ref; + UNUSED(db); + UNUSED(key); + if (!o || objectGetType(o) != OBJ_STRING) return 0; + if (objectGetEncoding(o) != OBJ_ENCODING_RAW || o->hasembval) return 0; + /* Another owner still needs the bytes. Try again on a later tick. */ + if (objectGetRefcount(o) != 1) return 0; + raw = objectGetVal(o); + if (sdslen(raw) < (size_t)server.tiering_min_value) return 0; + if (server.tiering_idle_seconds > 0 && objectGetLRUIdleSecs(o) < (uint32_t)server.tiering_idle_seconds) return 0; + ref = tierAppend(raw, sdslen(raw)); + if (!ref) return 0; + objectSetEncoding(o, OBJ_ENCODING_TIERED); + objectSetVal(o, ref); + sdsfree(raw); + return 1; +} + +robj *tierPromote(serverDb *db, robj *key, robj *o) { + tierRef *ref; + sds raw; + robj *neu; + if (!o || objectGetEncoding(o) != OBJ_ENCODING_TIERED) return o; + ref = objectGetVal(o); + raw = tierReadRef(ref); + if (!raw) return o; + neu = createObject(OBJ_STRING, raw); + dbReplaceValue(db, key, &neu); + tierNotePromotion(); + return neu; +} + +static int tierVisit(serverDb *db, int budget) { + kvstoreIterator *it; + void *next; + int spilled = 0; + if (!db || kvstoreSize(db->keys) == 0 || budget <= 0) return 0; + it = kvstoreIteratorInit(db->keys, HASHTABLE_ITER_SAFE); + while (budget > 0 && kvstoreIteratorNext(it, &next)) { + robj *o = next; + sds keystr; + robj key; + budget--; + if (objectGetType(o) != OBJ_STRING) continue; + if (objectGetEncoding(o) == OBJ_ENCODING_TIERED) continue; + keystr = objectGetKey(o); + initStaticStringObject(key, keystr); + if (server.tiering_max_ram > 0 && (long long)zmalloc_used_memory() <= server.tiering_max_ram && + server.tiering_idle_seconds > 0 && objectGetLRUIdleSecs(o) < (uint32_t)server.tiering_idle_seconds) + continue; + if (tierSpillObject(db, &key, o)) spilled++; + if (server.tiering_max_ram > 0 && (long long)zmalloc_used_memory() <= server.tiering_max_ram && spilled > 0 && + server.tiering_idle_seconds == 0) + break; + } + kvstoreIteratorRelease(it); + return spilled; +} + +void tierCron(void) { + int dbid; + if (!server.tiering_enabled) return; + for (dbid = 0; dbid < server.dbnum; dbid++) { + if (server.db[dbid]) tierVisit(server.db[dbid], 64); + } + if (tierDeadBytes() > 4096 && tierDeadBytes() * 2 > tierBytesOnDisk()) { + /* Collected by compact.c */ + extern int tierCompactLive(void); + tierCompactLive(); + } +} diff --git a/src/tier/tier.h b/src/tier/tier.h new file mode 100644 index 00000000000..fbda1e5a08c --- /dev/null +++ b/src/tier/tier.h @@ -0,0 +1,29 @@ +#ifndef CMO_TIER_H +#define CMO_TIER_H + +#include "server.h" + +/* String values live in an append-only log. The object keeps only this + * pointer. OBJ_ENCODING_TIERED is defined in server.h. A command never sees + * that encoding: lookup promotes the value back to a normal string first. */ + +typedef struct tierRef { + uint64_t offset; + uint32_t len; + uint64_t crc; +} tierRef; + +void tierCron(void); +robj *tierPromote(serverDb *db, robj *key, robj *o); +tierRef *tierAppend(const char *buf, size_t len); +sds tierReadRef(const tierRef *ref); +void tierMarkDead(const tierRef *ref); +void tierFreeRef(tierRef *ref); +void tierNotePromotion(void); +int tierCompact(tierRef **live, int nlive); +long long tierBytesOnDisk(void); +long long tierSpills(void); +long long tierPromotions(void); +long long tierDeadBytes(void); + +#endif diff --git a/src/tier/vlog.c b/src/tier/vlog.c new file mode 100644 index 00000000000..78d64b2bc60 --- /dev/null +++ b/src/tier/vlog.c @@ -0,0 +1,226 @@ +/* Append-only value log. Records are checksummed. A torn tail (crash during + * append) is dropped the next time the file is opened. Compaction writes a + * new file and renames it over the old one, so a crash before rename leaves + * the previous log intact. + * + * Layout of one record: magic u32, length u32, crc64 u64, payload. + * Ideas from memcached extstore and KeyDB FLASH (BSD) stopped at "append-only + * value log plus a pointer". This file is not derived from either. + * + * Copyright (c) 2026, Cache Me Outside contributors. + * Licensed under the BSD 3-Clause License, the same terms as the server. + */ +#include "tier/tier.h" + +#include +#include +#include +#include + +#define TIER_MAGIC 0xC0E71067u + +static int tier_fd = -1; +static uint64_t tier_end = 0; +static long long tier_dead = 0; +static long long tier_spills = 0; +static long long tier_promotions = 0; +static int tier_torn = 0; + +static uint64_t tier_crc(const unsigned char *p, size_t n) { + uint64_t c = 0xcbf29ce484222325ULL; + size_t i; + for (i = 0; i < n; i++) { + c ^= p[i]; + c *= 0x100000001b3ULL; + } + return c; +} + +static int tier_open(void) { + if (tier_fd >= 0) return C_OK; + if (!server.tiering_path || !server.tiering_path[0]) return C_ERR; + tier_fd = open(server.tiering_path, O_RDWR | O_CREAT, 0644); + if (tier_fd < 0) { + serverLog(LL_WARNING, "Tiering open %s failed: %s", server.tiering_path, strerror(errno)); + return C_ERR; + } + tier_end = 0; + for (;;) { + unsigned char hdr[16]; + uint32_t magic, len; + uint64_t crc, got; + ssize_t n = pread(tier_fd, hdr, sizeof(hdr), (off_t)tier_end); + if (n == 0) break; + if (n != (ssize_t)sizeof(hdr)) { + tier_torn = 1; + break; + } + memcpy(&magic, hdr, 4); + memcpy(&len, hdr + 4, 4); + memcpy(&crc, hdr + 8, 8); + if (magic != TIER_MAGIC || len > 64 * 1024 * 1024) { + tier_torn = 1; + break; + } + { + unsigned char *buf = zmalloc(len ? len : 1); + ssize_t rn = pread(tier_fd, buf, len, (off_t)(tier_end + 16)); + if (rn != (ssize_t)len) { + zfree(buf); + tier_torn = 1; + break; + } + got = tier_crc(buf, len); + zfree(buf); + if (got != crc) { + tier_torn = 1; + break; + } + } + tier_end += 16 + len; + } + if (tier_torn) { + if (ftruncate(tier_fd, (off_t)tier_end) == 0) { + serverLog(LL_WARNING, "Tiering dropped a torn tail at offset %llu", (unsigned long long)tier_end); + } + } + return C_OK; +} + +tierRef *tierAppend(const char *buf, size_t len) { + unsigned char hdr[16]; + uint32_t magic = TIER_MAGIC; + uint32_t ulen = (uint32_t)len; + uint64_t crc; + tierRef *ref; + if (tier_open() != C_OK) return NULL; + if (len > 0xffffffffu) return NULL; + crc = tier_crc((const unsigned char *)buf, len); + memcpy(hdr, &magic, 4); + memcpy(hdr + 4, &ulen, 4); + memcpy(hdr + 8, &crc, 8); + if (pwrite(tier_fd, hdr, sizeof(hdr), (off_t)tier_end) != (ssize_t)sizeof(hdr)) return NULL; + if (len && pwrite(tier_fd, buf, len, (off_t)(tier_end + 16)) != (ssize_t)len) return NULL; + ref = zmalloc(sizeof(*ref)); + ref->offset = tier_end; + ref->len = ulen; + ref->crc = crc; + tier_end += 16 + len; + tier_spills++; + if (server.tiering_debug_crash_after > 0 && tier_spills >= server.tiering_debug_crash_after) { + unsigned char junk[8] = {1, 2, 3, 4, 5, 6, 7, 8}; + pwrite(tier_fd, junk, sizeof(junk), (off_t)tier_end); + _exit(0); + } + return ref; +} + +sds tierReadRef(const tierRef *ref) { + unsigned char hdr[16]; + uint32_t magic, len; + uint64_t crc; + sds out; + ssize_t n; + if (!ref || tier_open() != C_OK) return NULL; + n = pread(tier_fd, hdr, sizeof(hdr), (off_t)ref->offset); + if (n != (ssize_t)sizeof(hdr)) return NULL; + memcpy(&magic, hdr, 4); + memcpy(&len, hdr + 4, 4); + memcpy(&crc, hdr + 8, 8); + if (magic != TIER_MAGIC || len != ref->len || crc != ref->crc) return NULL; + out = sdsnewlen(NULL, len); + if (len && pread(tier_fd, out, len, (off_t)(ref->offset + 16)) != (ssize_t)len) { + sdsfree(out); + return NULL; + } + if (tier_crc((unsigned char *)out, len) != crc) { + sdsfree(out); + return NULL; + } + return out; +} + +void tierMarkDead(const tierRef *ref) { + if (!ref) return; + tier_dead += 16 + ref->len; +} + +void tierFreeRef(tierRef *ref) { + zfree(ref); +} + +long long tierBytesOnDisk(void) { + return (long long)tier_end; +} +long long tierSpills(void) { + return tier_spills; +} +long long tierPromotions(void) { + return tier_promotions; +} +long long tierDeadBytes(void) { + return tier_dead; +} +void tierNotePromotion(void) { + tier_promotions++; +} + +/* Rewrite live records. `live` is an array of refs still installed in keys. + * Offsets inside those refs are updated. Returns C_OK, or _exit()s first + * when tiering-debug-exit-before-rename is set. */ +int tierCompact(tierRef **live, int nlive) { + char tmp[512]; + int fd, i; + uint64_t end = 0; + if (tier_fd < 0 || nlive < 0) return C_ERR; + snprintf(tmp, sizeof(tmp), "%s.compact", server.tiering_path); + fd = open(tmp, O_RDWR | O_CREAT | O_TRUNC, 0644); + if (fd < 0) return C_ERR; + for (i = 0; i < nlive; i++) { + sds payload = tierReadRef(live[i]); + unsigned char hdr[16]; + uint32_t magic = TIER_MAGIC; + uint32_t len; + uint64_t crc; + if (!payload) { + close(fd); + unlink(tmp); + return C_ERR; + } + len = (uint32_t)sdslen(payload); + crc = tier_crc((unsigned char *)payload, len); + memcpy(hdr, &magic, 4); + memcpy(hdr + 4, &len, 4); + memcpy(hdr + 8, &crc, 8); + if (pwrite(fd, hdr, sizeof(hdr), (off_t)end) != (ssize_t)sizeof(hdr) || + (len && pwrite(fd, payload, len, (off_t)(end + 16)) != (ssize_t)len)) { + sdsfree(payload); + close(fd); + unlink(tmp); + return C_ERR; + } + live[i]->offset = end; + live[i]->crc = crc; + end += 16 + len; + sdsfree(payload); + } + if (fsync(fd) != 0) { + close(fd); + unlink(tmp); + return C_ERR; + } + if (server.tiering_debug_exit_before_rename) { + close(fd); + _exit(0); + } + close(fd); + if (rename(tmp, server.tiering_path) != 0) { + unlink(tmp); + return C_ERR; + } + close(tier_fd); + tier_fd = open(server.tiering_path, O_RDWR, 0644); + tier_end = end; + tier_dead = 0; + return tier_fd >= 0 ? C_OK : C_ERR; +} diff --git a/tests/unit/tiering.tcl b/tests/unit/tiering.tcl new file mode 100644 index 00000000000..4e17e4648c1 --- /dev/null +++ b/tests/unit/tiering.tcl @@ -0,0 +1,212 @@ +# SSD value log. Keys stay in RAM. Large cold strings move to an append-only +# log and come back on read. Default tiering is off, so the rest of the suite +# does not enter this path. + +proc tier_p99 {samples} { + set s [lsort -integer $samples] + set n [llength $s] + if {$n == 0} { return 0 } + set i [expr {($n - 1) * 99 / 100}] + return [lindex $s $i] +} + +start_server {tags {external:skip} overrides {save "" tiering-enabled yes tiering-idle-seconds 0 tiering-min-value 64 tiering-max-ram 0}} { + test {cold strings spill and promote back to the same bytes} { + set payload [string repeat x 120000] + # Spill only after the sample, so a cron tick cannot move the values + # before used_memory is recorded. + r config set tiering-enabled no + r set a $payload + r set b $payload + r set c $payload + set hot {} + for {set i 0} {$i < 40} {incr i} { + set t [clock microseconds] + r get a + lappend hot [expr {[clock microseconds] - $t}] + } + set before_spill [status r used_memory] + set spills [status r tiering_spills] + r config set tiering-enabled yes + wait_for_condition 50 100 { + [status r tiering_spills] >= $spills + 2 + } else { + fail "values were not spilled: [r info memory]" + } + set after_spill [status r used_memory] + assert {$after_spill + 100000 < $before_spill} + + set cold {} + for {set i 0} {$i < 12} {incr i} { + if {$i > 0} { + set seen [status r tiering_spills] + wait_for_condition 40 50 { + [status r tiering_spills] > $seen + } else { + fail "value was not spilled again" + } + } + set t [clock microseconds] + set got [r get b] + lappend cold [expr {[clock microseconds] - $t}] + assert_equal $got $payload + } + assert_equal [r get c] $payload + assert {[status r tiering_promotions] >= 1} + assert {[status r tiering_bytes] > 0} + assert {[tier_p99 $hot] < 500000} + assert {[tier_p99 $cold] < 500000} + } + + test {delete keeps the surviving value} { + r del a + assert_equal [string length [r get b]] 120000 + assert_equal [r exists a] 0 + } + + test {expire frees a tiered string} { + r config set tiering-enabled no + r set e [string repeat e 90000] + set spills [status r tiering_spills] + r config set tiering-enabled yes + wait_for_condition 40 50 { + [status r tiering_spills] > $spills + } else { + fail "expire key was not spilled" + } + r pexpire e 1 + wait_for_condition 50 100 { + [r exists e] == 0 + } else { + fail "tiered key did not expire" + } + assert_equal [string length [r get b]] 120000 + } + + test {mixed reads writes deletes and expires stay consistent} { + for {set i 0} {$i < 20} {incr i} { + set n [expr {$i % 5}] + if {$n == 0} { + r set mix$i [string repeat m [expr {2000 + $i * 500}]] + } elseif {$n == 1} { + r get mix$i + } elseif {$n == 2} { + r del mix$i + } elseif {$n == 3} { + r set mix$i short + r pexpire mix$i 1 + } else { + r set keep [string repeat k 80000] + assert_equal [string length [r get keep]] 80000 + } + } + r set keep [string repeat k 80000] + after 200 + assert_equal [string length [r get keep]] 80000 + } + + test {RDB reload round-trips a tiered value} { + r save + assert {[r debug reload nosave] eq {OK}} + assert_equal [string length [r get b]] 120000 + } {} {needs:debug} + + test {a torn tail does not hide earlier records} { + r config set tiering-debug-crash-after [expr {[status r tiering_spills] + 1}] + catch {r set d [string repeat y 120000]} + after 500 + if {[is_alive [srv 0 pid]]} { + exec kill -9 [srv 0 pid] + } + set pid [srv 0 pid] + set spins 0 + while {[is_alive $pid]} { + after 20 + incr spins + if {$spins > 100} break + } + restart_server 0 true false + assert_equal [string length [r get b]] 120000 + } {} {needs:debug} +} + +start_server {tags {external:skip needs:debug} overrides {save "" tiering-enabled yes tiering-idle-seconds 0 tiering-min-value 64 tiering-max-ram 0}} { + test {crash before compaction rename keeps the previous log} { + # Fresh log: three equal records. Deleting two makes dead space more + # than half the file, which is when compaction runs. The process exits + # before rename, so the previous log is the one that remains. + r config set tiering-enabled no + r set a [string repeat a 90000] + r set b [string repeat b 90000] + r set c [string repeat c 90000] + r config set tiering-enabled yes + wait_for_condition 40 50 { + [status r tiering_spills] >= 3 + } else { + fail "compaction keys were not spilled" + } + r save + set bytes [status r tiering_bytes] + r config set tiering-debug-exit-before-rename yes + r del a + r del c + after 300 + if {[is_alive [srv 0 pid]]} { + catch {r ping} + after 500 + } + set pid [srv 0 pid] + set spins 0 + while {[is_alive $pid]} { + after 20 + incr spins + if {$spins > 150} { + fail "server did not exit during compaction (dead [status r tiering_dead_bytes] bytes [status r tiering_bytes])" + } + } + restart_server 0 true false + assert_equal [string length [r get b]] 90000 + assert {[status r tiering_bytes] >= $bytes} + } +} + +start_server {tags {external:skip needs:debug} overrides {save "" appendonly yes aof-use-rdb-preamble no tiering-enabled yes tiering-idle-seconds 0 tiering-min-value 64 tiering-max-ram 0}} { + test {AOF rewrite reloads a tiered string} { + set payload [string repeat z 80000] + r config set tiering-enabled no + r set a $payload + set spills [status r tiering_spills] + r config set tiering-enabled yes + wait_for_condition 40 50 { + [status r tiering_spills] > $spills + } else { + fail "AOF key was not spilled" + } + r bgrewriteaof + waitForBgrewriteaof r + r debug loadaof + assert_equal [r get a] $payload + } +} + +start_server {tags {external:skip} overrides {save "" tiering-enabled no tiering-idle-seconds 0 tiering-min-value 1000 tiering-max-ram 0}} { + test {a dataset several times the RAM cap spills until under the cap} { + set base [status r used_memory] + set cap [expr {$base + 250000}] + set payload_len 200000 + set n [expr {int(double($cap) * 5 / $payload_len) + 1}] + set blob [string repeat q $payload_len] + r config set tiering-max-ram $cap + for {set i 0} {$i < $n} {incr i} { + r set k$i $blob + } + assert {[expr {$n * $payload_len}] > [expr {$cap * 5}]} + r config set tiering-enabled yes + wait_for_condition 100 100 { + [status r used_memory] <= $cap + } else { + fail "used_memory [status r used_memory] stayed above cap $cap (spills [status r tiering_spills])" + } + assert_equal [string length [r get k0]] $payload_len + } +} diff --git a/valkey.conf b/valkey.conf index e3b129812d7..bf156694ae0 100644 --- a/valkey.conf +++ b/valkey.conf @@ -1698,6 +1698,18 @@ disable-thp yes # # Please check https://valkey.io/topics/persistence for more information. +# SSD value log. Keys and metadata stay in RAM. String values larger than +# tiering-min-value that have not been touched for tiering-idle-seconds are +# appended to tiering-path and replaced by a pointer. A read copies the value +# back into RAM. Compaction rewrites the log and renames it into place. +# tiering-max-ram 0 means there is no RAM cap. Default is off. +# +# tiering-enabled no +# tiering-path tier.vlog +# tiering-min-value 128 +# tiering-idle-seconds 60 +# tiering-max-ram 0 + appendonly no # The base name of the append only file.