diff options
Diffstat (limited to '')
| -rw-r--r-- | src/CursorManager.cpp | 11 | ||||
| -rw-r--r-- | src/EphemeralRelation.cpp | 196 | ||||
| -rw-r--r-- | src/LifecycleManager.cpp | 40 | ||||
| -rw-r--r-- | src/RNT_C_API.cpp | 131 |
4 files changed, 373 insertions, 5 deletions
diff --git a/src/CursorManager.cpp b/src/CursorManager.cpp index 0d71b1c..23f826f 100644 --- a/src/CursorManager.cpp +++ b/src/CursorManager.cpp @@ -61,17 +61,18 @@ namespace nt { }; if (label == EPHEMERAL_RELATION) { - // Cursor is created exhausted — the JOIN operator writes args and - // resets the state before the first probe. The generator is copied - // off the registry-owned ephemeral_object_type so Next() never has - // to dereference the handle again. + // Cursor opens ready for a standalone scan (no bound args, offset + // zero) so a named ephemeral can be drained like a stored relation. + // A JOIN probe overwrites this state anyway: the operator writes + // args and resets the cursor before the first probe. The generator + // is copied off the registry-owned ephemeral_object_type so Next() + // never has to dereference the handle again. auto* etype = static_cast<ObjectManager::ephemeral_object_type*>( handle->object->head->type.get()); auto* c = new cursor(); c->is_ephemeral = true; c->generator = etype->generator; - c->exhausted = true; pin_snapshot(c); return c; } diff --git a/src/EphemeralRelation.cpp b/src/EphemeralRelation.cpp new file mode 100644 index 0000000..5cb2dd7 --- /dev/null +++ b/src/EphemeralRelation.cpp @@ -0,0 +1,196 @@ +#include "EphemeralRelation.h" + +#include "Types.h" + +#include <picosha2.h> + +#include <algorithm> +#include <memory> + +namespace nt::Ephemeral { + namespace { + // Object-type descriptor for an EPHEMERAL_RELATION. Ephemeral relations + // are not durable (disposable=true) and never contend (readers compose + // generators independently). The generator and cardinality live on the + // descriptor; the cursor layer copies the generator out at Open time. + std::unique_ptr<ObjectManager::object_type> + make_type(ObjectManager::ephemeral_object_type::Cardinality cardinality, + ObjectManager::ephemeral_object_type::Generator generator) { + auto t = std::make_unique<ObjectManager::ephemeral_object_type>(); + t->label = EPHEMERAL_RELATION; + t->disposable = true; + t->methods = {OPEN, CLOSE}; + t->exclusive = false; + t->cardinality = cardinality; + t->generator = std::move(generator); + return t; + } + + // Reads the merkle_root off a registry entry whose object is a stored or + // ephemeral relation. Empty string for anything else (or missing root). + std::string entry_root(ObjectManager::registry* entry) { + if (entry == nullptr || entry->object == nullptr) + return {}; + if (auto* r = dynamic_cast<ObjectManager::Relation*>(entry->object.get())) + return r->merkle_root; + if (auto* e = dynamic_cast<ObjectManager::EphemeralRelation*>(entry->object.get())) + return e->merkle_root; + return {}; + } + } // namespace + + std::vector<std::string> SplitPath(const std::string& path) { + std::vector<std::string> parts; + std::string seg; + for (char c : path) { + if (c == '/') { + if (!seg.empty()) + parts.push_back(seg); + seg.clear(); + } else { + seg.push_back(c); + } + } + if (!seg.empty()) + parts.push_back(seg); + return parts; + } + + std::string ComposeRoot(const std::string& generator_identity, + const std::vector<std::pair<std::string, std::string>>& schema, + const std::vector<std::string>& base_roots) { + // Sort schema and base roots so input order does not change the digest. + std::vector<std::pair<std::string, std::string>> sorted_schema = schema; + std::sort(sorted_schema.begin(), sorted_schema.end()); + std::vector<std::string> sorted_roots = base_roots; + std::sort(sorted_roots.begin(), sorted_roots.end()); + + std::string buf; + buf += generator_identity; + buf.push_back('\0'); + // NUL-separate each field so distinct (name, type) splits cannot + // serialize to the same bytes, then a lone NUL to close the schema. + for (const auto& [name, type] : sorted_schema) { + buf += name; + buf.push_back('\0'); + buf += type; + buf.push_back('\0'); + } + buf.push_back('\0'); + for (const auto& root : sorted_roots) { + buf += root; + buf.push_back('\n'); + } + return picosha2::hash256_hex_string(buf.begin(), buf.end()); + } + + ObjectManager::registry* Register( + ObjectManager& objects, LifecycleManager& lifecycles, const std::string& session_hash, + Tier tier, const std::string& name, + ObjectManager::ephemeral_object_type::Generator generator, + ObjectManager::ephemeral_object_type::Cardinality cardinality, + const std::string& generator_identity, + const std::vector<std::pair<std::string, std::string>>& schema, + const std::vector<std::string>& dependencies) { + if (session_hash.empty()) + return nullptr; + if (tier == Tier::Named && name.empty()) + return nullptr; + + // Resolve dependencies up front: read their current roots for the + // identity hash and keep the entries to pin after registration. An + // unresolvable dependency fails the registration; skipping it would + // store a path that was never pinned, and CascadeEphemeral would later + // Unpin a hold it never took. + std::vector<ObjectManager::registry*> dep_entries; + std::vector<std::string> base_roots; + dep_entries.reserve(dependencies.size()); + base_roots.reserve(dependencies.size()); + for (const auto& dep : dependencies) { + auto* entry = objects.Find(SplitPath(dep)); + if (entry == nullptr) + return nullptr; + dep_entries.push_back(entry); + base_roots.push_back(entry_root(entry)); + } + + const std::string root = ComposeRoot(generator_identity, schema, base_roots); + const std::string subdir = (tier == Tier::Named) ? "ephemeral" : "scratch"; + const std::string leaf = (tier == Tier::Named) ? name : root; + const std::vector<std::string> path = {"system", "sessions", session_hash, subdir, leaf}; + + // Idempotent: an identical scratch result, or a live named binding, + // reuses the existing entry without double-pinning. + if (auto* existing = objects.Find(path)) + return existing; + + auto er = std::make_unique<ObjectManager::EphemeralRelation>(); + er->merkle_root = root; + er->dependencies = dependencies; + objects.Register(path, std::move(er), make_type(cardinality, std::move(generator))); + + auto* entry = objects.Find(path); + if (entry == nullptr) + return nullptr; + + // Structural pins on the base relations: released by + // LifecycleManager::CascadeEphemeral when this entry is collected. + for (auto* dep : dep_entries) + lifecycles.Pin(dep); + + // Tier policy: a named ephemeral is owned by its session and outlives + // its cursor; a scratch intermediate is owned by nobody and dies by + // counters with its cursor. + if (tier == Tier::Named) + lifecycles.Pin(entry); + + return entry; + } + + std::size_t ReleaseSession(ObjectManager& objects, LifecycleManager& lifecycles, + const std::string& session_hash) { + const std::vector<std::string> named_prefix = {"system", "sessions", session_hash, + "ephemeral"}; + const std::vector<std::string> scratch_prefix = {"system", "sessions", session_hash, + "scratch"}; + + auto is_child = [](const std::vector<std::string>& path, + const std::vector<std::string>& prefix) { + if (path.size() != prefix.size() + 1) + return false; + for (std::size_t i = 0; i < prefix.size(); ++i) + if (path[i] != prefix[i]) + return false; + return true; + }; + + // Collect targets before mutating: Unpin / Collect splice entries out of + // the same list we are walking. + std::vector<ObjectManager::registry*> named; + std::vector<ObjectManager::registry*> scratch; + for (auto* cur = objects.entries.get(); cur != nullptr; cur = cur->next.get()) { + if (cur->head == nullptr) + continue; + const auto& p = cur->head->path; + if (is_child(p, named_prefix)) + named.push_back(cur); + else if (is_child(p, scratch_prefix)) + scratch.push_back(cur); + } + + std::size_t released = 0; + // Named: drop the session-ownership pin; collects if nothing else holds it. + for (auto* entry : named) { + const uint32_t before = entry->head->reference_count; + lifecycles.Unpin(entry); + if (before > 0) + ++released; + } + // Scratch: never session-pinned; collect any stragglers whose cursor closed. + for (auto* entry : scratch) + if (lifecycles.Collect(entry)) + ++released; + + return released; + } +} // namespace nt::Ephemeral diff --git a/src/LifecycleManager.cpp b/src/LifecycleManager.cpp index 4a12e09..10f3bb1 100644 --- a/src/LifecycleManager.cpp +++ b/src/LifecycleManager.cpp @@ -1,5 +1,6 @@ #include "LifecycleManager.h" +#include "EphemeralRelation.h" #include "Merkle.h" #include "Types.h" @@ -44,6 +45,20 @@ namespace nt { return object != nullptr; } + bool LifecycleManager::Collect(ObjectManager::registry* object) { + if (!IsEligibleForGC(object)) + return false; + // Named-lifetime types (BRANCH / SESSION) never auto-collect; TryCollect + // would no-op on them, so return false rather than report a collection. + if (object->head->type != nullptr) { + const auto label = object->head->type->label; + if (label == BRANCH || label == SESSION) + return false; + } + TryCollect(object); + return true; + } + void LifecycleManager::TryCollect(ObjectManager::registry* object) { if (!IsEligibleForGC(object)) return; @@ -66,6 +81,8 @@ namespace nt { CascadeMultigroup(object); if (label == BRANCH_TREE) CascadeBranchTree(object); + if (label == EPHEMERAL_RELATION) + CascadeEphemeral(object); objects_.Unregister(object->head->path); } @@ -140,4 +157,27 @@ namespace nt { for (auto* mg : mgs) Unpin(mg); } + + void LifecycleManager::CascadeEphemeral(ObjectManager::registry* ephemeral) { + if (ephemeral == nullptr || ephemeral->object == nullptr) + return; + auto* er = dynamic_cast<ObjectManager::EphemeralRelation*>(ephemeral->object.get()); + if (er == nullptr) + return; + + // Resolve each dependency path and collect before mutating: Unpin can + // cascade further GC, and iteration must outlive that mutation. + std::vector<ObjectManager::registry*> deps; + for (const auto& dep : er->dependencies) { + const auto parts = Ephemeral::SplitPath(dep); + if (parts.empty()) + continue; + auto* entry = objects_.Find(parts); + if (entry != nullptr) + deps.push_back(entry); + } + + for (auto* dep : deps) + Unpin(dep); + } } // namespace nt diff --git a/src/RNT_C_API.cpp b/src/RNT_C_API.cpp index a632ab9..a83ac3a 100644 --- a/src/RNT_C_API.cpp +++ b/src/RNT_C_API.cpp @@ -1,6 +1,7 @@ #include "RNT_C_API.h" #include "CursorManager.h" +#include "EphemeralRelation.h" #include "HandlerManager.h" #include "IdentityManager.h" #include "InMemoryBackend.h" @@ -751,6 +752,10 @@ int rnt_session_close(const char* session_hash) { rebind_branch_tree_pin(hash, ""); session->branch_overrides.clear(); + // Drop the session's owned ephemeral relations (named tier-1 bindings and + // any scratch stragglers), cascading their base-relation pins. + nt::Ephemeral::ReleaseSession(g_rt->objects, *g_rt->lifecycles, session_hash); + return g_rt->objects.Unregister(session_path) ? 0 : -1; } @@ -792,6 +797,132 @@ int rnt_session_set_branch(const char* session_hash, const char* branch_name, return 0; } +// --------------------------------------------------------------------------- +// Ephemeral relations +// --------------------------------------------------------------------------- + +namespace { + // Accumulates tuples emitted by a generator callback during one page + // request. Lives on the stack of the generator wrapper below; the opaque + // rnt_tuple_sink_t handed to the callback points at it. + struct TupleSink { + std::vector<nt::Tuple> tuples; + }; + + // Splits a newline-separated list into its non-empty lines. + std::vector<std::string> split_lines(const char* s) { + std::vector<std::string> lines; + if (!s) + return lines; + std::stringstream ss(s); + std::string line; + while (std::getline(ss, line)) + if (!line.empty()) + lines.push_back(line); + return lines; + } + + // Joins registry path segments back into a slash-separated logical path. + std::string join_path(const std::vector<std::string>& segs) { + std::string out; + for (const auto& seg : segs) { + if (!out.empty()) + out.push_back('/'); + out += seg; + } + return out; + } +} // namespace + +int rnt_sink_emit(rnt_tuple_sink_t sink, const char* tuple_kv) { + if (!sink || !tuple_kv) + return -1; + auto* s = static_cast<TupleSink*>(sink); + s->tuples.emplace_back(parse_kv(tuple_kv)); + return 0; +} + +int rnt_register_ephemeral_relation(const char* session_hash, int named, const char* name, + rnt_generator_fn generator, void* generator_ctx, + int cardinality, const char* generator_identity, + const char* schema_kv, const char* dependencies, + char** path_out) { + if (!g_rt || !session_hash || !generator || !generator_identity) + return -1; + using Card = nt::ObjectManager::ephemeral_object_type::Cardinality; + if (cardinality < static_cast<int>(Card::Finite) || + cardinality > static_cast<int>(Card::Continuum)) + return -1; + + // The session must exist: the entry lands under its namespace and is + // released by rnt_session_close. + auto* session = g_rt->objects.Find({"system", "sessions", session_hash}); + if (!session || !session->head || session->head->type->label != SESSION) + return -1; + + std::vector<std::pair<std::string, std::string>> schema; + for (const auto& line : split_lines(schema_kv)) { + const auto pos = line.find('='); + if (pos == std::string::npos) + return -1; + schema.emplace_back(line.substr(0, pos), line.substr(pos + 1)); + } + + // Wrap the C callback in the std::function shape the cursor layer copies + // out at Open time. The sink gives emitted tuples a single well-defined + // owner: this wrapper's stack frame, for the duration of one page request. + auto gen = [generator, generator_ctx](const std::vector<std::string>& args, + std::size_t offset, + std::size_t limit) -> std::vector<nt::Tuple> { + std::string joined; + for (const auto& a : args) { + joined += a; + joined.push_back('\n'); + } + TupleSink sink; + const int rc = generator(generator_ctx, joined.c_str(), offset, limit, &sink); + if (rc != 0) + return {}; + return std::move(sink.tuples); + }; + + // Resolve each dependency through the reference manager so callers may + // pass branch-relative paths (e.g. system/branches/<b>/multigroups/<mg>/ + // relations/<r>); the stored dependency list then holds the resolved + // snapshot paths that raw registry lookups (and the GC cascade) address. + std::vector<std::string> deps; + for (const auto& dep : split_lines(dependencies)) { + const auto resolved = g_rt->references->Resolve(nt::Ephemeral::SplitPath(dep)); + if (resolved.empty()) + return -1; + deps.push_back(join_path(resolved)); + } + + auto* entry = nt::Ephemeral::Register( + g_rt->objects, *g_rt->lifecycles, session_hash, + named ? nt::Ephemeral::Tier::Named : nt::Ephemeral::Tier::Scratch, + name ? name : "", std::move(gen), static_cast<Card>(cardinality), generator_identity, + schema, deps); + if (!entry) + return -1; + + if (path_out) + *path_out = heap_str(join_path(entry->head->path)); + return 0; +} + +int rnt_drop_ephemeral_relation(const char* session_hash, const char* name) { + if (!g_rt || !session_hash || !name || !*name) + return -1; + auto* entry = g_rt->objects.Find({"system", "sessions", session_hash, "ephemeral", name}); + if (!entry || !entry->head || entry->head->type->label != EPHEMERAL_RELATION) + return -1; + // Releases the session-ownership pin taken at registration; the entry is + // collected once no cursor or dependent ephemeral still holds it. + g_rt->lifecycles->Unpin(entry); + return 0; +} + rnt_handle_t rnt_open_handle(const char* path, const char* /*claims*/) { if (!path) return nullptr; |