Winery-Strata / include /strata /core /conversation_cache.hpp

🍷 Winery Strata: fork of Niko1221/Strata (Winery Cuvée family, wine-red UI)

bbb6388 verified 25 days ago

9.4 kB

// CPU-only ownership and matching policy for --serve's parked conversations.
// Token equality, image identity, and steering mode are all required for reuse.
#pragma once

#include "strata/core/conversation_buffer.hpp"

#include <algorithm>
#include <array>
#include <cstddef>
#include <cstdint>
#include <deque>
#include <utility>
#include <vector>

namespace strata::core {

struct ConversationImageKey {
    int64_t start = 0;
    uint64_t hash = 0;
    bool operator==(const ConversationImageKey&) const = default;
};

struct ConversationCheckpoint {
    std::vector<int32_t> ids;
    std::vector<ConversationImageKey> imgs;
    std::vector<uint8_t> gdn, ple, tails, dead, block_pos;
    uint64_t used = 0; // upstream root-pinned/LRU checkpoint retention
    // Ordinary layer-split checkpoints retain each device's running state.
    // Whole-session parking is currently single-GPU and rejects these parts.
    std::vector<ConversationCheckpoint> stage_parts;

    size_t bytes() const {
        size_t n = ids.capacity() * sizeof(int32_t) + imgs.capacity() * sizeof(ConversationImageKey) +
               gdn.capacity() + ple.capacity() + tails.capacity() + dead.capacity() + block_pos.capacity() +
               stage_parts.capacity() * sizeof(ConversationCheckpoint);
        for (const auto& part : stage_parts) n += part.bytes();
        return n;
    }
};

// Identity-layout K/V pages and completed indexer rows. For streamed layers the
// source is the authoritative host pool, NOT the replaceable VRAM slots.
struct ConversationKv {
    int format = 0;
    int64_t cells = 0, heads = 0, head_dim = 0, page_size = 0, pooled_rows = 0, idx_dim = 0;
    ConversationBuffer k, v, k_scale, v_scale, pooled;
    size_t bytes() const {
        return k.bytes() + v.bytes() + k_scale.bytes() + v_scale.bytes() + pooled.bytes();
    }
};

struct ConversationKvReuse {
    std::vector<ConversationKv> kv;
    // Original image extent for validation, and the earliest subsequent rewrite.
    int64_t captured_tokens = 0, unchanged_tokens = 0;
    size_t bytes() const {
        size_t n = kv.capacity() * sizeof(ConversationKv);
        for (const auto& layer : kv) n += layer.bytes();
        return n;
    }
};

struct SavedConversation {
    // Runtime compatibility only; NOT a model/weights identity or disk schema.
    std::array<int64_t, 18> geometry{};
    // The session's layer carve the image was captured from ([0, n_layers) on one GPU); restore requires the same.
    int64_t layer_lo = 0, layer_hi = 0;
    ConversationCheckpoint live;
    std::vector<ConversationCheckpoint> checkpoints;
    std::vector<ConversationKv> kv; // main layers followed by the draft layer
    bool cvec = true;

    size_t bytes() const {
        size_t n = live.bytes() + checkpoints.capacity() * sizeof(ConversationCheckpoint) +
                   kv.capacity() * sizeof(ConversationKv);
        for (const auto& c : checkpoints) n += c.bytes();
        for (const auto& k : kv) n += k.bytes();
        return n;
    }
};

template<class Token>
int64_t conversation_prefix(const ConversationCheckpoint& c, const std::vector<Token>& prompt,
                            const std::vector<ConversationImageKey>& images) {
    const size_t n = c.ids.size();
    // The last prompt token always starts the next verify window.
    if (n == 0 || n >= prompt.size() || !std::equal(c.ids.begin(), c.ids.end(), prompt.begin())) return 0;
    size_t j = 0;
    for (const auto& image : images) {
        if (image.start >= (int64_t) n) continue;
        if (j == c.imgs.size() || !(c.imgs[j++] == image)) return 0;
    }
    if (j != c.imgs.size()) return 0;
    return (int64_t) n;
}

class ConversationCache {
public:
    struct Match {
        size_t index = 0;
        int64_t tokens = 0;
        bool live = false;
    };

    ConversationCache(size_t budget, size_t slots) : budget_(budget), slots_(slots) {}
    bool enabled() const { return budget_ != 0 && slots_ != 0; }
    size_t bytes() const { return bytes_ + reuse_.bytes(); }
    size_t size() const { return entries_.size(); }
    size_t evictions() const { return evictions_; }

    // Retain only the restored K/V buffers, not duplicate running checkpoints.
    // This optimization never evicts a parked conversation to make itself fit.
    void retain(std::vector<ConversationKv>&& kv, int64_t tokens) {
        reuse_ = {};
        ConversationKvReuse candidate{std::move(kv), tokens, tokens};
        if (enabled() && candidate.bytes() <= budget_ - bytes_) reuse_ = std::move(candidate);
    }
    void limit_reuse(int64_t first_dirty) {
        reuse_.unchanged_tokens = std::min(reuse_.unchanged_tokens, first_dirty);
        if (reuse_.unchanged_tokens <= 0) reuse_ = {};
    }
    ConversationKvReuse take_reuse() { return std::exchange(reuse_, {}); }
    size_t retained_bytes() const { return reuse_.bytes(); }
    bool can_fit(size_t incoming, size_t held = 0) const {
        return enabled() && held <= budget_ && incoming <= budget_ - held &&
               entries_.size() < slots_ && bytes() <= budget_ - held - incoming;
    }

    template<class Token>
    Match best(const std::vector<Token>& prompt, const std::vector<ConversationImageKey>& images, bool cvec) const {
        Match best;
        // Ties prefer the most recently parked branch. The caller prefers its
        // already-active state when that offers the same prefix length.
        for (size_t i = entries_.size(); i-- > 0;) {
            const auto& e = entries_[i];
            if (e.cvec != cvec) continue;
            auto consider = [&](const ConversationCheckpoint& c, bool live) {
                const int64_t n = conversation_prefix(c, prompt, images);
                if (n > best.tokens) best = {i, n, live};
            };
            consider(e.live, true);
            for (const auto& c : e.checkpoints) consider(c, false);
        }
        return best;
    }

    SavedConversation take(size_t index) {
        SavedConversation out = std::move(entries_.at(index));
        bytes_ -= out.bytes();
        entries_.erase(entries_.begin() + (std::ptrdiff_t) index);
        return out;
    }

    // Reserve before allocating a snapshot. held is an incoming image removed
    // with take() but still alive during the exchange; count it against RAM too.
    bool make_room(size_t incoming, size_t held = 0) {
        if (!enabled() || held > budget_ || incoming > budget_ - held) return false;
        if (bytes() > budget_ - held - incoming) reuse_ = {};
        while (!entries_.empty() && (entries_.size() >= slots_ || bytes_ > budget_ - held - incoming)) {
            bytes_ -= entries_.front().bytes();
            entries_.pop_front();
            ++evictions_;
        }
        return true;
    }

    // #342: drop the parked entries an outgoing conversation (its live tokens and checkpoint chain) supersedes:
    // the same conversation a turn back, whose DEEPEST checkpoint the outgoing chain still holds, so all it adds
    // is the tail the client rewrote (the reply as it was generated, before the next request re-rendered it) and
    // checkpoints older than that one.  A subagent's successive turns parked one such copy each, and make_room's
    // oldest-first eviction then pushed the parent conversation out after `slots` turns.  An entry without
    // checkpoints, or whose deepest checkpoint the outgoing chain does not hold (another conversation that only
    // shares the system prompt's root with it), is kept.  Returns how many were dropped.
    size_t drop_superseded(const std::vector<int32_t>& ids, const std::vector<ConversationImageKey>& images,
                           const std::vector<ConversationCheckpoint>& checkpoints, bool cvec) {
        auto held = [&](const ConversationCheckpoint& c) {
            if (c.ids == ids && c.imgs == images) return true;
            for (const auto& k : checkpoints)
                if (k.ids == c.ids && k.imgs == c.imgs) return true;
            return false;
        };
        size_t dropped = 0;
        for (size_t i = 0; i < entries_.size();) {
            const auto& e = entries_[i];
            const ConversationCheckpoint* deepest = nullptr;
            for (const auto& c : e.checkpoints)
                if (!deepest || c.ids.size() > deepest->ids.size()) deepest = &c;
            if (e.cvec == cvec && deepest && !deepest->ids.empty() && held(*deepest)) {
                bytes_ -= e.bytes();
                entries_.erase(entries_.begin() + (std::ptrdiff_t) i);
                ++dropped;
                continue;
            }
            ++i;
        }
        superseded_ += dropped;
        return dropped;
    }
    size_t superseded() const { return superseded_; }

    bool put(SavedConversation&& image, size_t held = 0) {
        const size_t n = image.bytes();
        if (!enabled() || held > budget_ || n > budget_ - held) return false;   // make_room's refusal, first
        drop_superseded(image.live.ids, image.live.imgs, image.checkpoints, image.cvec);
        if (!make_room(n, held)) return false;
        entries_.push_back(std::move(image));
        bytes_ += n;
        return true;
    }

private:
    size_t budget_ = 0, slots_ = 0, bytes_ = 0, evictions_ = 0, superseded_ = 0;
    std::deque<SavedConversation> entries_; // least recently active first
    ConversationKvReuse reuse_;
};

} // namespace strata::core

	// CPU-only ownership and matching policy for --serve's parked conversations.
	// Token equality, image identity, and steering mode are all required for reuse.
	#pragma once

	#include "strata/core/conversation_buffer.hpp"

	#include <algorithm>
	#include <array>
	#include <cstddef>
	#include <cstdint>
	#include <deque>
	#include <utility>
	#include <vector>

	namespace strata::core {

	struct ConversationImageKey {
	int64_t start = 0;
	uint64_t hash = 0;
	bool operator==(const ConversationImageKey&) const = default;
	};

	struct ConversationCheckpoint {
	std::vector<int32_t> ids;
	std::vector<ConversationImageKey> imgs;
	std::vector<uint8_t> gdn, ple, tails, dead, block_pos;
	uint64_t used = 0; // upstream root-pinned/LRU checkpoint retention
	// Ordinary layer-split checkpoints retain each device's running state.
	// Whole-session parking is currently single-GPU and rejects these parts.
	std::vector<ConversationCheckpoint> stage_parts;

	size_t bytes() const {
	size_t n = ids.capacity() * sizeof(int32_t) + imgs.capacity() * sizeof(ConversationImageKey) +
	gdn.capacity() + ple.capacity() + tails.capacity() + dead.capacity() + block_pos.capacity() +
	stage_parts.capacity() * sizeof(ConversationCheckpoint);
	for (const auto& part : stage_parts) n += part.bytes();
	return n;
	}
	};

	// Identity-layout K/V pages and completed indexer rows. For streamed layers the
	// source is the authoritative host pool, NOT the replaceable VRAM slots.
	struct ConversationKv {
	int format = 0;
	int64_t cells = 0, heads = 0, head_dim = 0, page_size = 0, pooled_rows = 0, idx_dim = 0;
	ConversationBuffer k, v, k_scale, v_scale, pooled;
	size_t bytes() const {
	return k.bytes() + v.bytes() + k_scale.bytes() + v_scale.bytes() + pooled.bytes();
	}
	};

	struct ConversationKvReuse {
	std::vector<ConversationKv> kv;
	// Original image extent for validation, and the earliest subsequent rewrite.
	int64_t captured_tokens = 0, unchanged_tokens = 0;
	size_t bytes() const {
	size_t n = kv.capacity() * sizeof(ConversationKv);
	for (const auto& layer : kv) n += layer.bytes();
	return n;
	}
	};

	struct SavedConversation {
	// Runtime compatibility only; NOT a model/weights identity or disk schema.
	std::array<int64_t, 18> geometry{};
	// The session's layer carve the image was captured from ([0, n_layers) on one GPU); restore requires the same.
	int64_t layer_lo = 0, layer_hi = 0;
	ConversationCheckpoint live;
	std::vector<ConversationCheckpoint> checkpoints;
	std::vector<ConversationKv> kv; // main layers followed by the draft layer
	bool cvec = true;

	size_t bytes() const {
	size_t n = live.bytes() + checkpoints.capacity() * sizeof(ConversationCheckpoint) +
	kv.capacity() * sizeof(ConversationKv);
	for (const auto& c : checkpoints) n += c.bytes();
	for (const auto& k : kv) n += k.bytes();
	return n;
	}
	};

	template<class Token>
	int64_t conversation_prefix(const ConversationCheckpoint& c, const std::vector<Token>& prompt,
	const std::vector<ConversationImageKey>& images) {
	const size_t n = c.ids.size();
	// The last prompt token always starts the next verify window.
	if (n == 0 \|\| n >= prompt.size() \|\| !std::equal(c.ids.begin(), c.ids.end(), prompt.begin())) return 0;
	size_t j = 0;
	for (const auto& image : images) {
	if (image.start >= (int64_t) n) continue;
	if (j == c.imgs.size() \|\| !(c.imgs[j++] == image)) return 0;
	}
	if (j != c.imgs.size()) return 0;
	return (int64_t) n;
	}

	class ConversationCache {
	public:
	struct Match {
	size_t index = 0;
	int64_t tokens = 0;
	bool live = false;
	};

	ConversationCache(size_t budget, size_t slots) : budget_(budget), slots_(slots) {}
	bool enabled() const { return budget_ != 0 && slots_ != 0; }
	size_t bytes() const { return bytes_ + reuse_.bytes(); }
	size_t size() const { return entries_.size(); }
	size_t evictions() const { return evictions_; }

	// Retain only the restored K/V buffers, not duplicate running checkpoints.
	// This optimization never evicts a parked conversation to make itself fit.
	void retain(std::vector<ConversationKv>&& kv, int64_t tokens) {
	reuse_ = {};
	ConversationKvReuse candidate{std::move(kv), tokens, tokens};
	if (enabled() && candidate.bytes() <= budget_ - bytes_) reuse_ = std::move(candidate);
	}
	void limit_reuse(int64_t first_dirty) {
	reuse_.unchanged_tokens = std::min(reuse_.unchanged_tokens, first_dirty);
	if (reuse_.unchanged_tokens <= 0) reuse_ = {};
	}
	ConversationKvReuse take_reuse() { return std::exchange(reuse_, {}); }
	size_t retained_bytes() const { return reuse_.bytes(); }
	bool can_fit(size_t incoming, size_t held = 0) const {
	return enabled() && held <= budget_ && incoming <= budget_ - held &&
	entries_.size() < slots_ && bytes() <= budget_ - held - incoming;
	}

	template<class Token>
	Match best(const std::vector<Token>& prompt, const std::vector<ConversationImageKey>& images, bool cvec) const {
	Match best;
	// Ties prefer the most recently parked branch. The caller prefers its
	// already-active state when that offers the same prefix length.
	for (size_t i = entries_.size(); i-- > 0;) {
	const auto& e = entries_[i];
	if (e.cvec != cvec) continue;
	auto consider = [&](const ConversationCheckpoint& c, bool live) {
	const int64_t n = conversation_prefix(c, prompt, images);
	if (n > best.tokens) best = {i, n, live};
	};
	consider(e.live, true);
	for (const auto& c : e.checkpoints) consider(c, false);
	}
	return best;
	}

	SavedConversation take(size_t index) {
	SavedConversation out = std::move(entries_.at(index));
	bytes_ -= out.bytes();
	entries_.erase(entries_.begin() + (std::ptrdiff_t) index);
	return out;
	}

	// Reserve before allocating a snapshot. held is an incoming image removed
	// with take() but still alive during the exchange; count it against RAM too.
	bool make_room(size_t incoming, size_t held = 0) {
	if (!enabled() \|\| held > budget_ \|\| incoming > budget_ - held) return false;
	if (bytes() > budget_ - held - incoming) reuse_ = {};
	while (!entries_.empty() && (entries_.size() >= slots_ \|\| bytes_ > budget_ - held - incoming)) {
	bytes_ -= entries_.front().bytes();
	entries_.pop_front();
	++evictions_;
	}
	return true;
	}

	// #342: drop the parked entries an outgoing conversation (its live tokens and checkpoint chain) supersedes:
	// the same conversation a turn back, whose DEEPEST checkpoint the outgoing chain still holds, so all it adds
	// is the tail the client rewrote (the reply as it was generated, before the next request re-rendered it) and
	// checkpoints older than that one. A subagent's successive turns parked one such copy each, and make_room's
	// oldest-first eviction then pushed the parent conversation out after `slots` turns. An entry without
	// checkpoints, or whose deepest checkpoint the outgoing chain does not hold (another conversation that only
	// shares the system prompt's root with it), is kept. Returns how many were dropped.
	size_t drop_superseded(const std::vector<int32_t>& ids, const std::vector<ConversationImageKey>& images,
	const std::vector<ConversationCheckpoint>& checkpoints, bool cvec) {
	auto held = [&](const ConversationCheckpoint& c) {
	if (c.ids == ids && c.imgs == images) return true;
	for (const auto& k : checkpoints)
	if (k.ids == c.ids && k.imgs == c.imgs) return true;
	return false;
	};
	size_t dropped = 0;
	for (size_t i = 0; i < entries_.size();) {
	const auto& e = entries_[i];
	const ConversationCheckpoint* deepest = nullptr;
	for (const auto& c : e.checkpoints)
	if (!deepest \|\| c.ids.size() > deepest->ids.size()) deepest = &c;
	if (e.cvec == cvec && deepest && !deepest->ids.empty() && held(*deepest)) {
	bytes_ -= e.bytes();
	entries_.erase(entries_.begin() + (std::ptrdiff_t) i);
	++dropped;
	continue;
	}
	++i;
	}
	superseded_ += dropped;
	return dropped;
	}
	size_t superseded() const { return superseded_; }

	bool put(SavedConversation&& image, size_t held = 0) {
	const size_t n = image.bytes();
	if (!enabled() \|\| held > budget_ \|\| n > budget_ - held) return false; // make_room's refusal, first
	drop_superseded(image.live.ids, image.live.imgs, image.checkpoints, image.cvec);
	if (!make_room(n, held)) return false;
	entries_.push_back(std::move(image));
	bytes_ += n;
	return true;
	}

	private:
	size_t budget_ = 0, slots_ = 0, bytes_ = 0, evictions_ = 0, superseded_ = 0;
	std::deque<SavedConversation> entries_; // least recently active first
	ConversationKvReuse reuse_;
	};

	} // namespace strata::core